add nonlinear_buffered_channel

Author: Kirillog

runtime/include/verifying.h

@@ -1,6 +1,7 @@
 #pragma once
 #include <gflags/gflags.h>
 
+#include <chrono>
 #include <memory>
 #include <type_traits>
 

verifying/blocking/CMakeLists.txt

@@ -5,6 +5,7 @@ set (SOURCE_TARGET_LIST
     shared_mutexed_register.cpp
     bank.cpp
     buffered_channel.cpp
+    nonlinear_buffered_channel.cpp
     simple_deadlock.cpp
     bank_deadlock.cpp
 )
@@ -79,6 +80,10 @@ add_integration_test_blocking("buffered_channel_pct" "verify" FALSE
     buffered_channel --strategy pct --rounds 10000
 )
 
+add_integration_test_blocking("nonlinear_buffered_channel_pct" "verify" TRUE 
+    nonlinear_buffered_channel --strategy pct --rounds 10000
+)
+
 add_integration_test_blocking("buffered_channel_random" "verify" FALSE 
     buffered_channel --strategy random --rounds 10000
 )

verifying/blocking/buffered_channel.cpp

@@ -1,67 +1,13 @@
+#include "../specs/buffered_channel.h"
+
 #include <atomic>
 #include <condition_variable>
 #include <cstring>
 #include <mutex>
 
-#include "../specs/queue.h"
-#include "runtime/include/value_wrapper.h"
 #include "runtime/include/verifying.h"
 #include "verifiers/buffered_channel_verifier.h"
 
-constexpr int N = 5;
-
-namespace spec {
-struct BufferedChannel {
-  void Send(int v) { deq.push_back(v); }
-
-  int Recv() {
-    if (deq.empty()) {
-      return -1;
-    }
-    auto value = deq.front();
-    deq.pop_front();
-    return value;
-  }
-
-  using method_t = std::function<ValueWrapper(BufferedChannel *l, void *args)>;
-  static auto GetMethods() {
-    method_t send_func = [](BufferedChannel *l, void *args) {
-      auto real_args = reinterpret_cast<std::tuple<int> *>(args);
-      l->Send(std::get<0>(*real_args));
-      return void_v;
-    };
-
-    method_t recv_func = [](BufferedChannel *l, void *args) -> int {
-      return l->Recv();
-    };
-
-    return std::map<std::string, method_t>{
-        {"Send", send_func},
-        {"Recv", recv_func},
-    };
-  }
-
-  std::deque<int> deq;
-};
-
-struct BufferedChannelHash {
-  size_t operator()(const BufferedChannel &r) const {
-    int res = 0;
-    for (int elem : r.deq) {
-      res += elem;
-    }
-    return res;
-  }
-};
-
-struct BufferedChannelEquals {
-  bool operator()(const BufferedChannel &lhs,
-                  const BufferedChannel &rhs) const {
-    return lhs.deq == rhs.deq;
-  }
-};
-};  // namespace spec
-
 struct BufferedChannel {
   non_atomic void Send(int v) {
     std::unique_lock lock{mutex_};
@@ -86,7 +32,7 @@ struct BufferedChannel {
     }
     debug(stderr, "Recv\n");
     auto val = queue_[ridx_];
-    ridx_ = (ridx_ + 1) % 5;
+    ridx_ = (ridx_ + 1) % N;
     empty_ = (sidx_ == ridx_);
     full_ = false;
     send_side_cv_.notify_one();

verifying/blocking/folly_flatcombining_queue.cpp

@@ -5,7 +5,8 @@
 #include "runtime/include/verifying_macro.h"
 #include "verifying/specs/queue.h"
 
-class FlatCombiningQueue : public folly::FlatCombining<FlatCombiningQueue> {
+class FlatCombiningQueue
+    : public folly::FlatCombining<FlatCombiningQueue, ltest::mutex> {
   spec::Queue<> queue_;
 
  public:

verifying/blocking/nonlinear_buffered_channel.cpp

@@ -0,0 +1,69 @@
+#include <atomic>
+#include <condition_variable>
+#include <cstring>
+#include <mutex>
+
+#include "../specs/buffered_channel.h"
+#include "runtime/include/verifying.h"
+#include "verifiers/buffered_channel_verifier.h"
+
+struct BufferedChannel {
+  non_atomic void Send(int v) {
+    std::unique_lock lock{mutex_};
+    while (!closed_ && full_) {
+      debug(stderr, "Waiting in send...\n");
+      send_side_cv_.wait(lock);
+    }
+    debug(stderr, "Send\n");
+
+    queue_[sidx_] = v;
+    sidx_ = (sidx_ + 1) % N;
+    empty_ = false;
+    recv_side_cv_.notify_one();
+  }
+
+  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) % N;
+    empty_ = (sidx_ == ridx_);
+    full_ = false;
+    send_side_cv_.notify_one();
+    return val;
+  }
+
+  void Close() {
+    closed_.store(true);
+    send_side_cv_.notify_all();
+    recv_side_cv_.notify_all();
+  }
+
+  std::mutex mutex_;
+  std::condition_variable send_side_cv_, recv_side_cv_;
+  std::atomic<bool> closed_{false};
+
+  bool full_{false};
+  bool empty_{true};
+
+  uint32_t sidx_{0}, ridx_{0};
+
+  std::array<int, N> queue_{};
+};
+
+auto generateInt(size_t) {
+  return ltest::generators::makeSingleArg(rand() % 10 + 1);
+}
+
+using spec_t =
+    ltest::Spec<BufferedChannel, spec::BufferedChannel,
+                spec::BufferedChannelHash, spec::BufferedChannelEquals>;
+
+LTEST_ENTRYPOINT_CONSTRAINT(spec_t, spec::BufferedChannelVerifier);
+
+target_method(generateInt, void, BufferedChannel, Send, int);
+target_method(ltest::generators::genEmpty, int, BufferedChannel, Recv);

verifying/specs/buffered_channel.h

@@ -0,0 +1,60 @@
+
+#include <deque>
+#include <functional>
+#include <map>
+
+#include "runtime/include/value_wrapper.h"
+
+constexpr int N = 5;
+
+namespace spec {
+struct BufferedChannel {
+  void Send(int v) { deq.push_back(v); }
+
+  int Recv() {
+    if (deq.empty()) {
+      return -1;
+    }
+    auto value = deq.front();
+    deq.pop_front();
+    return value;
+  }
+
+  using method_t = std::function<ValueWrapper(BufferedChannel *l, void *args)>;
+  static auto GetMethods() {
+    method_t send_func = [](BufferedChannel *l, void *args) {
+      auto real_args = reinterpret_cast<std::tuple<int> *>(args);
+      l->Send(std::get<0>(*real_args));
+      return void_v;
+    };
+
+    method_t recv_func = [](BufferedChannel *l, void *args) -> int {
+      return l->Recv();
+    };
+
+    return std::map<std::string, method_t>{
+        {"Send", send_func},
+        {"Recv", recv_func},
+    };
+  }
+
+  std::deque<int> deq;
+};
+
+struct BufferedChannelHash {
+  size_t operator()(const BufferedChannel &r) const {
+    int res = 0;
+    for (int elem : r.deq) {
+      res += elem;
+    }
+    return res;
+  }
+};
+
+struct BufferedChannelEquals {
+  bool operator()(const BufferedChannel &lhs,
+                  const BufferedChannel &rhs) const {
+    return lhs.deq == rhs.deq;
+  }
+};
+};  // namespace spec