Optimize subscriber slot delivery

c_client/client_test.cc

@@ -282,6 +282,7 @@ TEST_F(ClientTest, CreatePublisherThenSubscriber) {
   ASSERT_NE(nullptr, client.client);
 
   SubspacePublisherOptions pub_opts = CPublisherOptionsDefault(256, 10);
+  ASSERT_EQ(64'000, pub_opts.subscriber_queue_arena_size);
   pub_opts.type.type = "foo";
   pub_opts.type.type_length = strlen(pub_opts.type.type);
   SubspacePublisher pub = subspace_create_publisher(client, "dave1", pub_opts);
@@ -300,6 +301,9 @@ TEST_F(ClientTest, CreatePublisherThenSubscriber) {
       subspace_create_subscriber(client, "dave1", CSubscriberOptionsDefault());
   ASSERT_NE(nullptr, sub.subscriber);
   ASSERT_FALSE(subspace_has_error());
+  ASSERT_EQ(16, subspace_get_publisher_queue_size(pub));
+  ASSERT_EQ(64'000, subspace_get_publisher_queue_arena_size(pub));
+  ASSERT_EQ(16, subspace_get_subscriber_queue_size(sub));
 
   ASSERT_TRUE(subspace_remove_subscriber(&sub));
   ASSERT_TRUE(subspace_remove_publisher(&pub));
@@ -781,11 +785,13 @@ TEST_F(ClientTest, ClientPublisherSubscriberIntrospection) {
   pub_opts.mux = mux;
   pub_opts.mux_length = strlen(mux);
   pub_opts.metadata_size = 8;
+  pub_opts.subscriber_queue_arena_size = 12'000;
   SubspacePublisher pub =
       subspace_create_publisher(client, "c_introspection", pub_opts);
   ASSERT_NE(nullptr, pub.publisher);
 
   SubspaceSubscriberOptions sub_opts = CSubscriberOptionsDefault();
+  sub_opts.subscriber_queue_size = 4;
   sub_opts.type.type = type;
   sub_opts.type.type_length = strlen(type);
   sub_opts.mux = mux;
@@ -836,6 +842,8 @@ TEST_F(ClientTest, ClientPublisherSubscriberIntrospection) {
   ASSERT_FALSE(subspace_is_publisher_for_tunnel(pub));
   ASSERT_EQ(192, subspace_get_publisher_slot_size(pub));
   ASSERT_EQ(6, subspace_get_publisher_num_slots(pub));
+  ASSERT_EQ(16, subspace_get_publisher_queue_size(pub));
+  ASSERT_EQ(12'000, subspace_get_publisher_queue_arena_size(pub));
   ASSERT_TRUE(SubspaceStringEquals(subspace_get_publisher_name(pub),
                                    "c_introspection"));
   ASSERT_TRUE(SubspaceStringEquals(subspace_get_publisher_type(pub), type));
@@ -860,6 +868,7 @@ TEST_F(ClientTest, ClientPublisherSubscriberIntrospection) {
   ASSERT_EQ(0, subspace_get_subscriber_num_active_messages(sub));
   ASSERT_EQ(8, subspace_get_subscriber_metadata_size(sub));
   ASSERT_EQ(4, subspace_get_subscriber_checksum_size(sub));
+  ASSERT_EQ(4, subspace_get_subscriber_queue_size(sub));
   ASSERT_GE(subspace_get_subscriber_prefix_size(sub), 64);
   ASSERT_GE(subspace_get_subscriber_virtual_memory_usage(sub), 0U);
 
@@ -1398,8 +1407,11 @@ TEST_F(ClientTest, InvalidArgumentsReportErrors) {
   ASSERT_EQ(-1, subspace_get_publisher_retirement_fd(invalid_publisher));
   ASSERT_EQ(0, subspace_get_subscriber_slot_size(invalid_subscriber));
   ASSERT_EQ(0, subspace_get_subscriber_num_slots(invalid_subscriber));
+  ASSERT_EQ(0, subspace_get_subscriber_queue_size(invalid_subscriber));
   ASSERT_EQ(0, subspace_get_publisher_slot_size(invalid_publisher));
   ASSERT_EQ(0, subspace_get_publisher_num_slots(invalid_publisher));
+  ASSERT_EQ(0, subspace_get_publisher_queue_size(invalid_publisher));
+  ASSERT_EQ(0, subspace_get_publisher_queue_arena_size(invalid_publisher));
   ASSERT_EQ(0, subspace_get_publisher_metadata_size(invalid_publisher));
   ASSERT_EQ(0, subspace_get_subscriber_metadata_size(invalid_subscriber));
   ASSERT_EQ(0, subspace_get_publisher_prefix_size(invalid_publisher));

c_client/subspace.cc

@@ -127,6 +127,9 @@ SubspaceChannelInfo ToCChannelInfo(const subspace::ChannelInfo &info,
           .type = ToCString(type),
           .slot_size = info.slot_size,
           .num_slots = info.num_slots,
+          .subscriber_queue_size = info.subscriber_queue_size,
+          .subscriber_queue_arena_size =
+              info.subscriber_queue_arena_size,
           .reliable = info.reliable};
 }
 
@@ -487,6 +490,7 @@ bool subspace_get_all_channel_stats(SubspaceClient client,
 SubspaceSubscriberOptions subspace_subscriber_options_default(void) {
   SubspaceSubscriberOptions options = {};
   options.max_active_messages = 1;
+  options.detect_dropped_messages = true;
   options.vchan_id = -1;
   return options;
 }
@@ -496,6 +500,7 @@ SubspacePublisherOptions subspace_publisher_options_default(int32_t slot_size,
   SubspacePublisherOptions options = {
       slot_size,
       num_slots,
+      subspace::kDefaultSubscriberQueueArenaSize,
       false,
       false,
       false,
@@ -524,6 +529,7 @@ subspace_create_subscriber(SubspaceClient client, const char *channel_name,
                            SubspaceSubscriberOptions options) {
   subspace::SubscriberOptions subspace_options;
   subspace_options.SetReliable(options.reliable)
+      .SetSubscriberQueueSize(options.subscriber_queue_size)
       .SetBridge(options.bridge)
       .SetForTunnel(options.for_tunnel)
       .SetType(StringFromPointer(options.type.type, options.type.type_length))
@@ -535,6 +541,7 @@ subspace_create_subscriber(SubspaceClient client, const char *channel_name,
       .SetChecksum(options.checksum)
       .SetPassChecksumErrors(options.pass_checksum_errors)
       .SetKeepActiveMessage(options.keep_active_message)
+      .SetDetectDroppedMessages(options.detect_dropped_messages)
       .SetSplitBufferCallbacks(ToCppSplitCallbacks(options.split_callbacks));
   subspace_options.SetLogDroppedMessages(options.log_dropped_messages);
   subspace_clear_error();
@@ -575,6 +582,7 @@ SubspacePublisher subspace_create_publisher(SubspaceClient client,
       .SetChecksum(options.checksum)
       .SetChecksumSize(options.checksum_size)
       .SetMetadataSize(options.metadata_size)
+      .SetSubscriberQueueArenaSize(options.subscriber_queue_arena_size)
       .SetPreferRetiredSlots(options.prefer_retired_slots)
       .SetMaxPublishers(options.max_publishers)
       .SetUseSplitBuffers(options.use_split_buffers)
@@ -1442,6 +1450,21 @@ int32_t subspace_get_publisher_num_slots(SubspacePublisher publisher) {
   return (*PublisherPtr(publisher))->NumSlots();
 }
 
+int32_t subspace_get_publisher_queue_size(SubspacePublisher publisher) {
+  if (publisher.publisher == nullptr) {
+    return 0;
+  }
+  return (*PublisherPtr(publisher))->SubscriberQueueSize();
+}
+
+uint64_t
+subspace_get_publisher_queue_arena_size(SubspacePublisher publisher) {
+  if (publisher.publisher == nullptr) {
+    return 0;
+  }
+  return (*PublisherPtr(publisher))->SubscriberQueueArenaSize();
+}
+
 SubspaceString subspace_get_publisher_name(SubspacePublisher publisher) {
   if (publisher.publisher == nullptr) {
     return {};
@@ -1596,6 +1619,14 @@ int subspace_get_subscriber_num_slots(SubspaceSubscriber subscriber) {
   return (*sub_ptr)->NumSlots();
 }
 
+int32_t
+subspace_get_subscriber_queue_size(SubspaceSubscriber subscriber) {
+  if (subscriber.subscriber == nullptr) {
+    return 0;
+  }
+  return (*SubscriberPtr(subscriber))->SubscriberQueueSize();
+}
+
 int64_t subspace_get_subscriber_current_ordinal(SubspaceSubscriber subscriber) {
   if (subscriber.subscriber == nullptr) {
     return -1;
@@ -1844,13 +1875,19 @@ bool subspace_snapshot_message_slot(SubspaceMessageSlot slot,
   }
   auto *message_slot = reinterpret_cast<subspace::MessageSlot *>(slot.slot);
   *snapshot = {.id = message_slot->id,
-               .ordinal = message_slot->ordinal,
-               .message_size = message_slot->message_size,
-               .buffer_index = message_slot->buffer_index,
-               .vchan_id = message_slot->vchan_id,
-               .timestamp = message_slot->timestamp,
-               .flags = message_slot->flags,
-               .bridged_slot_id = message_slot->bridged_slot_id};
+               .ordinal =
+                   message_slot->ordinal.load(std::memory_order_relaxed),
+               .message_size =
+                   message_slot->message_size.load(std::memory_order_relaxed),
+               .buffer_index =
+                   message_slot->buffer_index.load(std::memory_order_relaxed),
+               .vchan_id =
+                   message_slot->vchan_id.load(std::memory_order_relaxed),
+               .timestamp =
+                   message_slot->timestamp.load(std::memory_order_relaxed),
+               .flags = message_slot->flags.load(std::memory_order_relaxed),
+               .bridged_slot_id = message_slot->bridged_slot_id.load(
+                   std::memory_order_relaxed)};
   return true;
 }
 

c_client/subspace.h

@@ -92,6 +92,8 @@ typedef struct {
   SubspaceString type;
   uint64_t slot_size;
   int num_slots;
+  int subscriber_queue_size;
+  uint64_t subscriber_queue_arena_size;
   bool reliable;
 } SubspaceChannelInfo;
 
@@ -211,6 +213,10 @@ typedef struct {
 typedef struct {
   const int32_t slot_size; // Initial size of slots (might be resized).
   const int num_slots;     // Number of slots (never changes)
+  // Total bytes reserved for packed per-subscriber queues in the CCB. The
+  // options factory selects 64,000 bytes; zero selects the bitset path by
+  // default.
+  uint64_t subscriber_queue_arena_size;
   bool local;              // If true, messages stay local to this machine.
   bool reliable;           // Reliable publisher.
   bool bridge;             // This publisher is for the bridge.
@@ -253,12 +259,16 @@ typedef struct {
 
 typedef struct {
   bool reliable;           // Reliable subscriber.
+  // Capacity of this subscriber's CCB slot queue. 0 uses the publisher
+  // default.
+  int32_t subscriber_queue_size;
   bool bridge;             // This subscriber is for the bridge.
   bool for_tunnel;         // Mark subscriptions for external tunnels.
   SubspaceTypeInfo type;   // Type of the message.  This is an opaque string.
   int max_active_messages; // Max number of message that can be active at once.
   bool pass_activation;    // Pass activation message in read.
   bool log_dropped_messages; // Log dropped messages to stderr.
+  bool detect_dropped_messages; // Detect and count ordinal gaps internally.
   bool read_write;           // Map buffers writable for this subscriber.
   const char *mux;           // Optional mux channel name for virtual channels.
   size_t mux_length;
@@ -398,6 +408,8 @@ int subspace_get_subscriber_fd(SubspaceSubscriber subscriber);
 // is received.
 int32_t subspace_get_subscriber_slot_size(SubspaceSubscriber subscriber);
 int subspace_get_subscriber_num_slots(SubspaceSubscriber subscriber);
+int32_t
+subspace_get_subscriber_queue_size(SubspaceSubscriber subscriber);
 
 // This is a shortcut to wait for a message to be available.  It will block
 // until a message is available.
@@ -532,6 +544,9 @@ bool subspace_is_publisher_for_tunnel(SubspacePublisher publisher);
 bool subspace_publisher_uses_split_buffers(SubspacePublisher publisher);
 int32_t subspace_get_publisher_slot_size(SubspacePublisher publisher);
 int32_t subspace_get_publisher_num_slots(SubspacePublisher publisher);
+int32_t subspace_get_publisher_queue_size(SubspacePublisher publisher);
+uint64_t
+subspace_get_publisher_queue_arena_size(SubspacePublisher publisher);
 SubspaceString subspace_get_publisher_name(SubspacePublisher publisher);
 SubspaceString subspace_get_publisher_type(SubspacePublisher publisher);
 SubspaceString subspace_get_publisher_mux(SubspacePublisher publisher);