extend buffered channel test

Author: Kirillog

Dockerfile

@@ -1,10 +1,10 @@
 FROM silkeh/clang:19 AS ltest
 
-RUN apt update && apt install -y git ninja-build valgrind libboost-context-dev libgflags-dev libstdc++-11-dev libclang-19-dev
+RUN apt update && apt install -y git ninja-build valgrind libgflags-dev libstdc++-11-dev libclang-19-dev
 RUN mv /usr/lib/gcc/x86_64-linux-gnu/12 /usr/lib/gcc/x86_64-linux-gnu/_12
 
 FROM ltest as blocking
-RUN apt install -y pkg-config libcapstone-dev && \
+RUN apt install -y pkg-config libcapstone-dev libboost-context-dev && \
     git clone https://github.com/Kirillog/syscall_intercept.git &&  \
     cmake syscall_intercept -G Ninja -B syscall_intercept/build -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_BUILD_TYPE=Release -DCMAKE_C_COMPILER=clang && \
     cmake --build syscall_intercept/build --target install
@@ -15,6 +15,12 @@ RUN apt install -y libboost-filesystem-dev libboost-program-options-dev libboost
     git clone https://github.com/Kirillog/folly.git && \
     cmake folly -G Ninja -B folly/build_dir -DCMAKE_BUILD_TYPE=Release
     # cmake --build folly/build_dir --target install
+FROM ltest as userver-blocking
+# userver conflicts with default libboost-context-dev (1.74) version, 1.81 required
+RUN apt install -y python3-dev python3-venv \
+        libboost-context1.81-dev libboost-filesystem1.81-dev libboost-program-options1.81-dev libboost-regex1.81-dev 
+        libboost-stacktrace1.81-dev libboost-locale1.81-dev \
+        libzstd-dev libyaml-cpp-dev libcrypto++-dev libnghttp2-dev libev-dev
 RUN sh -c "$(wget -O- https://github.com/deluan/zsh-in-docker/releases/download/v1.2.1/zsh-in-docker.sh)" -- \
        -p git
 CMD [ "zsh" ]

runtime/CMakeLists.txt

@@ -13,7 +13,7 @@ set (SOURCE_FILES
 )
 
 add_library(runtime SHARED ${SOURCE_FILES})
-find_package(Boost 1.74.0 REQUIRED COMPONENTS context)
+find_package(Boost REQUIRED COMPONENTS context)
 target_include_directories(runtime PRIVATE include ${Boost_INCLUDE_DIRS})
 target_link_libraries(runtime PRIVATE gflags ${Boost_LIBRARIES})
 target_link_options(runtime PRIVATE ${CMAKE_ASAN_FLAGS})

runtime/include/pct_strategy.h

@@ -17,14 +17,7 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
       : BaseStrategyWithThreads<TargetObj, Verifier>(threads_count, ctrs),
         current_depth(1),
         current_schedule_length(0) {
-    // We have information about potential number of resumes
-    // but because of the implementation, it's only available in the task.
-    // In fact, it doesn't depend on the task, it only depends on the
-    // constructor
-    size_t avg_k = 0;
-    avg_k = avg_k / this->constructors.size();
-
-    PrepareForDepth(current_depth, avg_k);
+    PrepareForDepth(current_depth, 1);
   }
 
   std::optional<size_t> NextThreadId() override {
@@ -36,9 +29,9 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
       // Ignore waiting tasks
       // debug(stderr, "prior: %d, number %d\n", priorities[i], i);
       if (!threads[i].empty() && threads[i].back()->IsBlocked()) {
-        // debug(stderr, "blocked on %p val %d\n",
-        // threads[i].back()->GetBlockState().addr,
-        // threads[i].back()->GetBlockState().value);
+        debug(stderr, "blocked on %p val %d\n",
+              threads[i].back()->GetBlockState().addr,
+              threads[i].back()->GetBlockState().value);
         // dual waiting if request finished, but follow up isn't
         // skip dual tasks that already have finished the request
         // section(follow-up will be executed in another task, so we can't

runtime/include/scheduler.h

@@ -259,7 +259,6 @@ struct BaseStrategyWithThreads : public Strategy {
                 constructor.Build(this->state.get(), thread_index, task_index);
             auto verified = this->sched_checker.Verify(
                 std::string(task->GetName()), thread_index);
-            assert(verified && "wrong release task at termination");
             thread.emplace_back(task);
           }
         }

verifying/blocking/buffered_channel.cpp

@@ -16,7 +16,7 @@ struct BufferedChannel {
     return 0;
   }
 
-  int TryRecv() {
+  int Recv() {
     if (deq.empty()) {
       return -1;
     }
@@ -32,13 +32,13 @@ struct BufferedChannel {
       return l->Send(std::get<0>(*real_args));
     };
 
-    method_t try_recv_func = [](BufferedChannel *l, void *args) -> int {
-      return l->TryRecv();
+    method_t recv_func = [](BufferedChannel *l, void *args) -> int {
+      return l->Recv();
     };
 
     return std::map<std::string, method_t>{
         {"Send", send_func},
-        {"TryRecv", try_recv_func},
+        {"Recv", recv_func},
     };
   }
 
@@ -67,7 +67,7 @@ struct BufferedChannel {
   non_atomic int Send(int v) {
     std::unique_lock lock{mutex_};
     while (!closed_ && full_) {
-      debug(stderr, "Waiting...\n");
+      debug(stderr, "Waiting in send...\n");
       send_side_cv_.wait(lock);
     }
     debug(stderr, "Send\n");
@@ -76,15 +76,17 @@ struct BufferedChannel {
     sidx_ = (sidx_ + 1) % N;
     full_ = (sidx_ == ridx_);
     empty_ = false;
+    recv_side_cv_.notify_one();
     return 0;
   }
 
-  // TryRecv is not blocking otherwise it is dual structure
-  non_atomic int TryRecv() {
-    std::lock_guard lock{mutex_};
-    if (closed_ || empty_) {
-      return -1;
+  non_atomic int Recv() {
+    std::unique_lock lock{mutex_};
+    while (!closed_ && empty_) {
+      debug(stderr, "Waiting in recv...\n");
+      recv_side_cv_.wait(lock);
     }
+    debug(stderr, "Recv\n");
     auto val = queue_[ridx_];
     ridx_ = (ridx_ + 1) % 5;
     empty_ = (sidx_ == ridx_);
@@ -93,15 +95,16 @@ struct BufferedChannel {
     return val;
   }
 
-  non_atomic int Close() {
+  int Close() {
     closed_.store(true);
     send_side_cv_.notify_all();
+    recv_side_cv_.notify_all();
     return 0;
   }
 
   std::mutex mutex_;
-  std::condition_variable send_side_cv_;
-  std::atomic_bool closed_{false};
+  std::condition_variable send_side_cv_, recv_side_cv_;
+  std::atomic<bool> closed_{false};
 
   bool full_{false};
   bool empty_{true};
@@ -122,5 +125,4 @@ using spec_t =
 LTEST_ENTRYPOINT_CONSTRAINT(spec_t, spec::BufferedChannelVerifier);
 
 target_method(generateInt, void, BufferedChannel, Send, int);
-target_method(ltest::generators::genEmpty, int, BufferedChannel, TryRecv);
-// target_method(ltest::generators::genEmpty, int, BufferedChannel, Close);
+target_method(ltest::generators::genEmpty, int, BufferedChannel, Recv);

verifying/blocking/verifiers/buffered_channel_verifier.h

@@ -12,11 +12,15 @@ struct BufferedChannelVerifier {
         return true;
       }
       return false;
-    } else if (task_name == "TryRecv") {
-      if (size_ > 0) {
-        --size_;
+    } else if (task_name == "Recv") {
+      if (receivers_ == 0) {
+        ++receivers_;
+        if (size_ > 0) {
+          --size_;
+        }
+        return true;
       }
-      return true;
+      return false;
     } else {
       assert(false);
     }
@@ -27,21 +31,23 @@ struct BufferedChannelVerifier {
     if (task_name == "Send") {
       --senders_;
       return;
-    } else if (task_name == "TryRecv") {
+    } else if (task_name == "Recv") {
+      --receivers_;
       return;
     } else {
       assert(false);
     }
   }
 
   std::optional<std::string> ReleaseTask(size_t thread_id) {
-    if (size_ > 0) {
-      return {"TryRecv"};
+    if (senders_ > 0) {
+      return {"Recv"};
     }
     return std::nullopt;
   }
 
   size_t senders_;
+  size_t receivers_;
   size_t size_;
 };
 }  // namespace spec