Blocking receive support

src/apps/msg_gen/main.cc

@@ -12,6 +12,7 @@
 #include <chrono>
 #include <csignal>
 #include <cstdint>
+#include <deque>
 #include <numeric>
 #include <sstream>
 #include <thread>
@@ -32,6 +33,8 @@ DEFINE_bool(active_generator, false,
             "When 'true' this host is generating the traffic, otherwise it is "
             "bouncing.");
 DEFINE_bool(verify, false, "Verify payload of received messages.");
+DEFINE_uint32(blocking, 0, "Blocking receive");
+DEFINE_uint32(load, 100, "kRPS load");
 
 static volatile int g_keep_running = 1;
 
@@ -102,6 +105,7 @@ class ThreadCtx {
   hdr_histogram *latency_hist;
   size_t num_request_latency_samples;
   std::vector<msg_latency_info_t> msg_latency_info_vec;
+  std::chrono::microseconds time_limit;
 
   struct {
     stats_t current;
@@ -171,7 +175,7 @@ void ServerLoop(void *channel_ctx) {
     MachnetFlow_t rx_flow;
     const ssize_t rx_size =
         machnet_recv(channel_ctx, thread_ctx.rx_message.data(),
-                     thread_ctx.rx_message.size(), &rx_flow);
+                     thread_ctx.rx_message.size(), &rx_flow, FLAGS_blocking);
     if (rx_size <= 0) continue;
     stats_cur.rx_count++;
     stats_cur.rx_bytes += rx_size;
@@ -235,75 +239,88 @@ void ClientSendOne(ThreadCtx *thread_ctx, uint64_t window_slot) {
 }
 
 // Return the window slot for which a response was received
-uint64_t ClientRecvOneBlocking(ThreadCtx *thread_ctx) {
+int64_t ClientRecvOne(ThreadCtx *thread_ctx) {
   const auto *channel_ctx = thread_ctx->channel_ctx;
 
-  while (true) {
-    if (g_keep_running == 0) {
-      LOG(INFO) << "ClientRecvOneBlocking: Exiting.";
-      return 0;
-    }
+  MachnetFlow_t rx_flow;
+  const ssize_t rx_size =
+      machnet_recv(channel_ctx, thread_ctx->rx_message.data(),
+                   thread_ctx->rx_message.size(), &rx_flow, FLAGS_blocking);
+  if (rx_size <= 0) return -1;
 
-    MachnetFlow_t rx_flow;
-    const ssize_t rx_size =
-        machnet_recv(channel_ctx, thread_ctx->rx_message.data(),
-                     thread_ctx->rx_message.size(), &rx_flow);
-    if (rx_size <= 0) continue;
+  thread_ctx->stats.current.rx_count++;
+  thread_ctx->stats.current.rx_bytes += rx_size;
 
-    thread_ctx->stats.current.rx_count++;
-    thread_ctx->stats.current.rx_bytes += rx_size;
-
-    const auto *msg_hdr =
-        reinterpret_cast<msg_hdr_t *>(thread_ctx->rx_message.data());
-    if (msg_hdr->window_slot >= FLAGS_msg_window) {
-      LOG(ERROR) << "Received invalid window slot: " << msg_hdr->window_slot;
-      continue;
-    }
+  const auto *msg_hdr =
+      reinterpret_cast<msg_hdr_t *>(thread_ctx->rx_message.data());
+  if (msg_hdr->window_slot > FLAGS_msg_window) {
+    LOG(ERROR) << "Received invalid window slot: " << msg_hdr->window_slot;
+    abort();
+  }
 
-    const size_t latency_us =
-        thread_ctx->RecordRequestEnd(msg_hdr->window_slot);
-    VLOG(1) << "Client: Received message for window slot "
-            << msg_hdr->window_slot << " in " << latency_us << " us";
-
-    if (FLAGS_verify) {
-      for (uint32_t i = sizeof(msg_hdr_t); i < rx_size; i++) {
-        if (thread_ctx->rx_message[i] != thread_ctx->message_gold[i]) {
-          LOG(ERROR) << "Message data mismatch at index " << i << std::hex
-                     << " " << static_cast<uint32_t>(thread_ctx->rx_message[i])
-                     << " "
-                     << static_cast<uint32_t>(thread_ctx->message_gold[i]);
-          break;
-        }
+  const size_t latency_us = thread_ctx->RecordRequestEnd(msg_hdr->window_slot);
+  VLOG(1) << "Client: Received message for window slot " << msg_hdr->window_slot
+          << " in " << latency_us << " us";
+
+  if (FLAGS_verify) {
+    for (uint32_t i = sizeof(msg_hdr_t); i < rx_size; i++) {
+      if (thread_ctx->rx_message[i] != thread_ctx->message_gold[i]) {
+        LOG(ERROR) << "Message data mismatch at index " << i << std::hex << " "
+                   << static_cast<uint32_t>(thread_ctx->rx_message[i]) << " "
+                   << static_cast<uint32_t>(thread_ctx->message_gold[i]);
+        break;
       }
     }
-
-    return msg_hdr->window_slot;
   }
-
-  LOG(FATAL) << "Should not reach here";
-  return 0;
+  return msg_hdr->window_slot;
 }
 
 void ClientLoop(void *channel_ctx, MachnetFlow *flow) {
   ThreadCtx thread_ctx(channel_ctx, flow);
+  thread_ctx.time_limit = std::chrono::duration_cast<std::chrono::microseconds>(
+      std::chrono::microseconds(1000) / (FLAGS_load));
   LOG(INFO) << "Client Loop: Starting.";
+  LOG(INFO) << "Time limit is: " << thread_ctx.time_limit.count();
 
   // Send a full window of messages
   for (uint32_t i = 0; i < FLAGS_msg_window; i++) {
     ClientSendOne(&thread_ctx, i /* window slot */);
   }
 
-  while (true) {
-    if (g_keep_running == 0) {
-      LOG(INFO) << "MsgGenLoop: Exiting.";
-      break;
+  auto next = std::chrono::steady_clock::now() + thread_ctx.time_limit;
+  std::deque<uint32_t> backlog;
+
+  while (g_keep_running) {
+    auto rx_window_slot = ClientRecvOne(&thread_ctx);
+
+    if (rx_window_slot <= 0) {
+      // Inner loop to handle the case where no message is received
+      while (g_keep_running) {
+        rx_window_slot = ClientRecvOne(&thread_ctx);
+        if (rx_window_slot > 0) break;
+
+        if (std::chrono::steady_clock::now() > next) {
+          // Handle timeout scenario
+          next = std::chrono::steady_clock::now() + thread_ctx.time_limit;
+          auto next_window = ++FLAGS_msg_window;
+          thread_ctx.msg_latency_info_vec.resize(next_window);
+          backlog.push_back(next_window);
+          ClientSendOne(&thread_ctx, backlog.front());
+          backlog.pop_front();
+        }
+      }
     }
-
-    const uint64_t rx_window_slot = ClientRecvOneBlocking(&thread_ctx);
-    ClientSendOne(&thread_ctx, rx_window_slot);
-
+    if (g_keep_running == 0) break;
+    // Check if the time limit has passed and a message is received
+    if (std::chrono::steady_clock::now() > next) {
+      ClientSendOne(&thread_ctx, backlog.front());
+      backlog.pop_front();
+      next = std::chrono::steady_clock::now() + thread_ctx.time_limit;
+    }
+    backlog.push_back(rx_window_slot);
     ReportStats(&thread_ctx);
   }
+  LOG(INFO) << "MsgGenLoop: Exiting.";
 
   auto &stats_cur = thread_ctx.stats.current;
   LOG(INFO) << "Application Statistics (TOTAL) - [TX] Sent: "

src/core/drivers/shm/channel.cc

@@ -21,7 +21,8 @@ ShmChannel::ShmChannel(const std::string channel_name,
       channel_fd_(channel_fd),
       cached_buf_indices(),
       cached_bufs(),
-      cached_buf_count(0) {}
+      cached_buf_count(0),
+      posted(0) {}
 
 ShmChannel::~ShmChannel() {
   __machnet_channel_destroy(

src/core/drivers/shm/channel_bench.cc

@@ -4,13 +4,15 @@
 #include <glog/logging.h>
 #include <machnet.h>
 #include <machnet_common.h>
-#include <signal.h>
-#include <ttime.h>
 #include <unistd.h>
 #include <utils.h>
 
+#include <csignal>
 #include <numeric>
 #include <thread>
+#include <utility>
+
+DEFINE_uint32(blocking, 0, "Block on receive");
 
 static constexpr uint8_t kStackCpuCoreId = 3;
 static constexpr uint8_t kAppCpuCoreId = 5;
@@ -39,7 +41,7 @@ struct thread_conf {
   thread_conf(std::shared_ptr<ShmChannel> ch, uint8_t core_id,
               uint64_t messages_to_send, uint64_t tx_message_size,
               uint64_t messages_to_receive)
-      : channel(ch),
+      : channel(std::move(ch)),
         cpu_core(core_id),
         messages_to_send(messages_to_send),
         tx_message_size(tx_message_size),
@@ -107,7 +109,6 @@ void stack_loop(thread_conf *conf) {
       }
       continue;
     }
-
     buf = channel->MsgBufAlloc();
     if (buf == nullptr) {
       continue;
@@ -126,7 +127,6 @@ void stack_loop(thread_conf *conf) {
     // Send the message.
     ret = channel->EnqueueMessages(&buf, 1);
     if (ret != 1) {
-      LOG(ERROR) << "Couldn't enqueue message. ret: " << ret;
       channel->MsgBufFree(buf);
     }
     conf->messages_sent += ret;
@@ -159,15 +159,14 @@ void application_loop(thread_conf *conf) {
   while (!g_start.load()) {
     __asm__ volatile("pause" ::: "memory");
   }
-
   // Now start receiving messages.
   auto start = std::chrono::high_resolution_clock::now();
   while (!g_should_stop.load()) {
     // RX.
     MachnetFlow_t flow;
 
-    auto nbytes =
-        machnet_recv(channel->ctx(), rx_buffer.data(), rx_buffer.size(), &flow);
+    auto nbytes = machnet_recv(channel->ctx(), rx_buffer.data(),
+                               rx_buffer.size(), &flow, FLAGS_blocking);
     if (nbytes > 0) {
       conf->messages_received++;
       CHECK_EQ(nbytes, conf->tx_message_size);
@@ -284,11 +283,18 @@ void print_results(const thread_conf &stack_conf, const thread_conf &app_conf) {
                                                   1e9))
                        : 0.0)
             << std::endl;
+  if (FLAGS_blocking) {
+    std::cout << juggler::utils::Format(
+        "Stack notified Application side %d times", channel->GetPosted());
+    channel->ResetPosted();
+  }
   std::cout << std::endl;
 }
 
-int main() {
+int main(int argc, char *argv[]) {
   google::InitGoogleLogging("channel_bench");
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+
   FLAGS_logtostderr = 1;
   signal(SIGINT, [](int) { g_should_stop.store(true); });
 
@@ -306,7 +312,8 @@ int main() {
   std::vector<std::pair<uint64_t, uint64_t>> exp_config_vec;
 
   exp_config_vec.emplace_back(kMessagesToSend, 0);  // Stack -> app only
-  exp_config_vec.emplace_back(0, kMessagesToSend);  // App -> stack only
+  if (!FLAGS_blocking)
+    exp_config_vec.emplace_back(0, kMessagesToSend);  // App -> stack only
   exp_config_vec.emplace_back(kMessagesToSend, kMessagesToSend);  // Bi-dir
 
   LOG(INFO) << "Running channel_bench";

src/core/drivers/shm/channel_test.cc

@@ -212,7 +212,7 @@ TEST(BasicChannelTest, ChannelEnqueue) {
   rx_msghdr.msg_iov = &rx_iov;
   rx_msghdr.msg_iovlen = 1;
 
-  EXPECT_EQ(machnet_recvmsg(channel->ctx(), &rx_msghdr), 1);
+  EXPECT_EQ(machnet_recvmsg(channel->ctx(), &rx_msghdr, NON_BLOCKING), 1);
   EXPECT_EQ(rx_msghdr.msg_size, kMessageSize);
   EXPECT_EQ(rx_msg, tx_msg);
 }
@@ -396,7 +396,7 @@ TEST(ChannelFullDuplex, SendRecvMsg) {
       msghdr.flags = 0;
       msghdr.flow_info = {
           .src_ip = 0, .dst_ip = 0, .src_port = 0, .dst_port = 0};
-      auto ret = machnet_recvmsg(ctx, &msghdr);
+      auto ret = machnet_recvmsg(ctx, &msghdr, NON_BLOCKING);
       if (ret == 1) msg_rx++;
 
       // If already sent the amount of messages needed skip.

src/core/flow_test.cc

@@ -124,6 +124,8 @@ class FlowTest : public ::testing::Test {
       num_msgbufs -= msgbuf_nr;
       batch.Clear();
     }
+    CHECK_NOTNULL(head);
+    CHECK_NOTNULL(tail);
     head->set_msg_length(data.size());
     head->set_last(tail->index());
     head->mark_first();
@@ -317,7 +319,7 @@ TEST_F(FlowTest, RXQueue_Push) {
     rx_msghdr.flow_info = {0, 0, 0, 0};
     rx_msghdr.msg_iov = &rx_iov;
     rx_msghdr.msg_iovlen = 1;
-    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr);
+    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr, NON_BLOCKING);
     EXPECT_EQ(ret, 1) << "Failed to deliver message to application";
     EXPECT_EQ(tx_message, rx_message);
     EXPECT_EQ(channel_->GetFreeBufCount(), channel_->GetTotalBufCount());
@@ -397,7 +399,7 @@ TEST_F(FlowTest, RXQueue_Push_OutOfOrder1) {
     rx_msghdr.flow_info = {0, 0, 0, 0};
     rx_msghdr.msg_iov = &rx_iov;
     rx_msghdr.msg_iovlen = 1;
-    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr);
+    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr, NON_BLOCKING);
     EXPECT_EQ(ret, 1) << "Failed to deliver message to application";
     EXPECT_EQ(tx_message, rx_message);
     EXPECT_EQ(channel_->GetFreeBufCount(), channel_->GetTotalBufCount());
@@ -496,7 +498,7 @@ TEST_F(FlowTest, RXQueue_Push_OutOfOrder2) {
     rx_msghdr.flow_info = {0, 0, 0, 0};
     rx_msghdr.msg_iov = &rx_iov;
     rx_msghdr.msg_iovlen = 1;
-    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr);
+    auto ret = machnet_recvmsg(channel_->ctx(), &rx_msghdr, NON_BLOCKING);
     EXPECT_EQ(ret, 1) << "Failed to deliver message to application";
     EXPECT_EQ(tx_message, rx_message);
     EXPECT_EQ(channel_->GetFreeBufCount(), channel_->GetTotalBufCount());

src/ext/Makefile

@@ -1,6 +1,6 @@
 CC = gcc
 CFLAGS = -Wall -fPIC
-LDFLAGS = -shared
+LDFLAGS = -shared -pthread
 LIBS = -luuid
 TARGET = libmachnet_shim.so
 SRCS = machnet.c

src/ext/machnet.c

@@ -11,6 +11,7 @@
 #include <arpa/inet.h>
 #include <errno.h>
 #include <netinet/in.h>
+#include <semaphore.h>
 #include <sys/mman.h>
 #include <sys/socket.h>
 #include <sys/types.h>
@@ -639,22 +640,23 @@ int machnet_sendmmsg(const void *channel_ctx,
 }
 
 ssize_t machnet_recv(const void *channel_ctx, void *buf, size_t len,
-                     MachnetFlow_t *flow) {
+                     MachnetFlow_t *flow, uint32_t blocking) {
   MachnetMsgHdr_t msghdr;
   MachnetIovec_t iov;
   iov.base = buf;
   iov.len = len;
   msghdr.msg_iov = &iov;
   msghdr.msg_iovlen = 1;
 
-  const int ret = machnet_recvmsg(channel_ctx, &msghdr);
+  const int ret = machnet_recvmsg(channel_ctx, &msghdr, blocking);
   if (ret <= 0) return ret;  // No message available, or error code
 
   *flow = msghdr.flow_info;
   return msghdr.msg_size;
 }
 
-int machnet_recvmsg(const void *channel_ctx, MachnetMsgHdr_t *msghdr) {
+int machnet_recvmsg(const void *channel_ctx, MachnetMsgHdr_t *msghdr,
+                    uint32_t blocking) {
   assert(channel_ctx != NULL);
   assert(msghdr != NULL);
   MachnetChannelCtx_t *ctx = (MachnetChannelCtx_t *)channel_ctx;
@@ -664,8 +666,17 @@ int machnet_recvmsg(const void *channel_ctx, MachnetMsgHdr_t *msghdr) {
   // Deque a message from the ring.
   MachnetRingSlot_t buffer_index;
   uint32_t n = __machnet_channel_machnet_ring_dequeue(ctx, 1, &buffer_index);
-  if (n != 1) return 0;  // No message available.
-
+  while (n == 0) {
+    if (!blocking) return 0;
+    __atomic_store_n(&ctx->receiver_active, 0, __ATOMIC_SEQ_CST);
+    //    n = __machnet_channel_machnet_ring_dequeue(ctx, 1, &buffer_index);
+    //    if (n == 0) {
+    sem_wait(&ctx->sem);
+    n = __machnet_channel_machnet_ring_dequeue(ctx, 1, &buffer_index);
+    //    assert(n == 1);
+    //    }
+  }
+  assert(n == 1);
   MachnetMsgBuf_t *buffer;
   buffer = __machnet_channel_buf(ctx, buffer_index);
   MachnetFlow_t flow_info = buffer->flow;