Add channel profile-based sizing overrides

c_client/client_test.cc

@@ -277,6 +277,19 @@ SubspaceSubscriberOptions CSubscriberOptionsDefault() {
   return options;
 }
 
+TEST_F(ClientTest, ProfileOptionDefaults) {
+  SubspacePublisherOptions pub_opts =
+      subspace_publisher_options_default(128, 4);
+  EXPECT_TRUE(pub_opts.apply_profile);
+  pub_opts.apply_profile = false;
+  EXPECT_FALSE(pub_opts.apply_profile);
+
+  SubspaceSubscriberOptions sub_opts = subspace_subscriber_options_default();
+  EXPECT_TRUE(sub_opts.apply_profile);
+  sub_opts.apply_profile = false;
+  EXPECT_FALSE(sub_opts.apply_profile);
+}
+
 TEST_F(ClientTest, CreatePublisherThenSubscriber) {
   SubspaceClient client = subspace_create_client_with_socket(Socket().c_str());
   ASSERT_NE(nullptr, client.client);

c_client/subspace.cc

@@ -490,6 +490,7 @@ SubspaceSubscriberOptions subspace_subscriber_options_default(void) {
   options.max_active_messages = 1;
   options.detect_dropped_messages = true;
   options.vchan_id = -1;
+  options.apply_profile = true;
   return options;
 }
 
@@ -518,6 +519,7 @@ SubspacePublisherOptions subspace_publisher_options_default(int32_t slot_size,
       false,
       0,
       SubspaceSplitBufferCallbacks{},
+      true,
   };
   return options;
 }
@@ -539,6 +541,7 @@ subspace_create_subscriber(SubspaceClient client, const char *channel_name,
       .SetPassChecksumErrors(options.pass_checksum_errors)
       .SetKeepActiveMessage(options.keep_active_message)
       .SetDetectDroppedMessages(options.detect_dropped_messages)
+      .SetApplyProfile(options.apply_profile)
       .SetSplitBufferCallbacks(ToCppSplitCallbacks(options.split_callbacks));
   subspace_options.SetLogDroppedMessages(options.log_dropped_messages);
   subspace_clear_error();
@@ -584,6 +587,7 @@ SubspacePublisher subspace_create_publisher(SubspaceClient client,
       .SetMaxPublishers(options.max_publishers)
       .SetUseSplitBuffers(options.use_split_buffers)
       .SetSplitBuffersOverBridge(options.split_buffers_over_bridge)
+      .SetApplyProfile(options.apply_profile)
       .SetSplitBufferCallbacks(ToCppSplitCallbacks(options.split_callbacks));
   subspace_clear_error();
   SubspacePublisher publisher;

c_client/subspace.h

@@ -254,6 +254,8 @@ typedef struct {
   bool split_buffers_over_bridge;
   int32_t max_publishers;
   SubspaceSplitBufferCallbacks split_callbacks;
+  // If true, a server with a profile file may override sizing values.
+  bool apply_profile;
 } SubspacePublisherOptions;
 
 typedef struct {
@@ -277,6 +279,8 @@ typedef struct {
   // Optional callbacks used when the server reports that the publisher
   // created split payload buffers.
   SubspaceSplitBufferCallbacks split_callbacks;
+  // If true, a server with a profile file may override sizing values.
+  bool apply_profile;
 } SubspaceSubscriberOptions;
 
 typedef enum {

client/client.cc

@@ -429,8 +429,12 @@ ClientImpl::CreatePublisher(const std::string &channel_name,
     (void)SendRequestReceiveResponse(remove_req, remove_resp, remove_fds);
   };
 
+  const int resolved_num_slots =
+      pub_resp.num_slots() > 0 ? pub_resp.num_slots() : opts.num_slots;
+  const int resolved_slot_size =
+      pub_resp.slot_size() > 0 ? pub_resp.slot_size() : Aligned(opts.slot_size);
   std::shared_ptr<PublisherImpl> channel = std::make_shared<PublisherImpl>(
-      channel_name, opts.num_slots, pub_resp.subscriber_queue_size(),
+      channel_name, resolved_num_slots, pub_resp.subscriber_queue_size(),
       pub_resp.channel_id(), pub_resp.publisher_id(), pub_resp.vchan_id(),
       session_id_, pub_resp.type(), opts,
       [this](Channel *c) {
@@ -481,7 +485,7 @@ ClientImpl::CreatePublisher(const std::string &channel_name,
   }
 
   if (absl::Status status =
-          channel->CreateOrAttachBuffers(Aligned(opts.slot_size));
+          channel->CreateOrAttachBuffers(Aligned(resolved_slot_size));
       !status.ok()) {
     remove_server_publisher();
     return status;
@@ -562,6 +566,9 @@ ClientImpl::CreateSubscriber(const std::string &channel_name,
   }
 
   SubscriberOptions subscriber_options = opts;
+  if (sub_resp.max_active_messages() > 0) {
+    subscriber_options.max_active_messages = sub_resp.max_active_messages();
+  }
   subscriber_options.use_split_buffers = sub_resp.use_split_buffers();
 
   std::shared_ptr<SubscriberImpl> channel = std::make_shared<SubscriberImpl>(
@@ -1385,7 +1392,14 @@ absl::Status ClientImpl::ReloadSubscriber(SubscriberImpl *subscriber) {
   auto *cmd = req.mutable_create_subscriber();
   cmd->set_channel_name(subscriber->Name());
   cmd->set_subscriber_id(subscriber->GetSubscriberId());
-  cmd->set_mux(subscriber->options_.mux);
+  cmd->set_is_reliable(subscriber->options_.IsReliable());
+  cmd->set_is_bridge(subscriber->options_.IsBridge());
+  cmd->set_for_tunnel(subscriber->options_.ForTunnel());
+  cmd->set_type(subscriber->options_.Type());
+  cmd->set_max_active_messages(subscriber->options_.MaxActiveMessages());
+  cmd->set_mux(subscriber->options_.Mux());
+  cmd->set_vchan_id(subscriber->options_.VchanId());
+  cmd->set_apply_profile(subscriber->options_.ApplyProfile());
 
   // Send request to server and wait for response.
   Response resp;
@@ -1407,6 +1421,9 @@ absl::Status ClientImpl::ReloadSubscriber(SubscriberImpl *subscriber) {
   if (!sub_resp.type().empty()) {
     subscriber->SetType(sub_resp.type());
   }
+  if (sub_resp.max_active_messages() > 0) {
+    subscriber->options_.max_active_messages = sub_resp.max_active_messages();
+  }
   subscriber->options_.use_split_buffers = sub_resp.use_split_buffers();
   subscriber->SetNumSlots(sub_resp.num_slots());
   subscriber->SetSubscriberQueueSize(sub_resp.subscriber_queue_size());
@@ -1866,6 +1883,7 @@ void ClientImpl::FillCreatePublisherRequest(CreatePublisherRequest *cmd,
   cmd->set_use_split_buffers(opts.UseSplitBuffers());
   cmd->set_split_buffers_over_bridge(opts.SplitBuffersOverBridge());
   cmd->set_subscriber_queue_size(opts.SubscriberQueueSize());
+  cmd->set_apply_profile(opts.ApplyProfile());
 }
 
 void ClientImpl::ApplyPublisherResponseFds(
@@ -1905,6 +1923,7 @@ void ClientImpl::FillCreateSubscriberRequest(CreateSubscriberRequest *cmd,
   cmd->set_max_active_messages(opts.MaxActiveMessages());
   cmd->set_mux(opts.Mux());
   cmd->set_vchan_id(opts.VchanId());
+  cmd->set_apply_profile(opts.ApplyProfile());
 }
 
 void ClientImpl::ApplySubscriberResponseFds(

client/client.h

@@ -1406,6 +1406,7 @@ class Subscriber {
   int32_t SlotSize() const { return impl_->SlotSize(); }
   int32_t NumSlots() const { return impl_->NumSlots(); }
   int32_t SubscriberQueueSize() const { return impl_->SubscriberQueueSize(); }
+  int32_t MaxActiveMessages() const { return impl_->MaxActiveMessages(); }
 
   const std::vector<std::unique_ptr<details::BufferSet>> &GetBuffers() const {
     return client_->GetBuffers(impl_.get());

client/client_channel.cc

@@ -81,8 +81,11 @@ ClientChannel::CreatePosixSharedMemoryFile(const std::string &filename,
   // Create a file in /tmp and make it the same size as the shared memory.  This
   // will not actually allocate any disk space.
   auto &shim = GetSyscallShim();
-  int fd = shim.open_fn(filename.c_str(), O_RDWR | O_CREAT, 0666);
+  int fd = shim.open_fn(filename.c_str(), O_RDWR | O_CREAT | O_EXCL, 0666);
   if (fd < 0) {
+    if (errno == EEXIST) {
+      return PosixSharedMemoryName(filename);
+    }
     return absl::InternalError(
         absl::StrFormat("Failed to open shadow file %s: %s", filename.c_str(),
                         strerror(errno)));
@@ -537,14 +540,13 @@ ClientChannel::CreatePosixBuffer(const std::string &filename, size_t size) {
   // On Posix we need to create a shadow file that has the same size as the
   // shared memory file.  This is because the fstat of the shm "file" returns a
   // page aligned size, which is not what we want.  The shadow file is used
-  // to determine the size of the shared memory segment.
+  // to determine the size of the shared memory segment. If the shadow already
+  // exists, do not truncate it because it records the existing generation size.
   absl::StatusOr<std::string> shm_name =
       CreatePosixSharedMemoryFile(filename, off_t(size));
   if (!shm_name.ok()) {
     return shm_name.status();
   }
-
-  // shm_name is the name of the shared memory.
   auto shm_fd = OpenSharedMemoryFile(*shm_name, O_RDWR | O_CREAT | O_EXCL);
   if (!shm_fd.ok()) {
     return shm_fd.status();
@@ -557,7 +559,7 @@ ClientChannel::CreatePosixBuffer(const std::string &filename, size_t size) {
   // Make it the appropriate size.
   int e = shim.ftruncate_fn(shm_fd->Fd(), off_t(size));
   if (e == -1) {
-    (void)shim.shm_unlink_fn(filename.c_str());
+    (void)shim.shm_unlink_fn(shm_name->c_str());
     return absl::InternalError(
         absl::StrFormat("Failed to set length of shared memory %s: %s",
                         filename, strerror(errno)));

client/latency_test.cc

@@ -824,6 +824,9 @@ TEST_F(LatencyTest, PublisherLatencyHistogram) {
   ASSERT_OK(pub_client.Init(Socket()));
   ASSERT_OK(sub_client.Init(Socket()));
 
+  const int subscriber_queue_size = std::atoi(
+      LatencyEnvOrDefault("SUBSPACE_SUBSCRIBER_QUEUE_SIZE", "0"));
+  std::cerr << "subscriber_queue_size: " << subscriber_queue_size << "\n";
   std::cerr << "num_slots,min,median,p99,max,average\n";
   auto show_latencies = [](std::vector<uint64_t> &latencies,
                            const std::string &test, const std::string &series,
@@ -860,7 +863,10 @@ TEST_F(LatencyTest, PublisherLatencyHistogram) {
        num_slots < LatencyValueForSplitBuffers(100000, 20000, 3000);
        num_slots = (num_slots)*15 / 10) {
     absl::StatusOr<Publisher> pub = pub_client.CreatePublisher(
-        "publat", 256, num_slots, subspace::PublisherOptions().SetReliable(false));
+        "publat", 256, num_slots,
+        subspace::PublisherOptions()
+            .SetReliable(false)
+            .SetSubscriberQueueSize(subscriber_queue_size));
     ASSERT_OK(pub);
 
     std::cerr << num_slots << ",";

client/options.h

@@ -39,6 +39,7 @@ struct PublisherOptions {
   int32_t SlotSize() const { return slot_size; }
   int32_t NumSlots() const { return num_slots; }
   int32_t SubscriberQueueSize() const { return subscriber_queue_size; }
+  bool ApplyProfile() const { return apply_profile; }
   PublisherOptions &SetSlotSize(int32_t size) {
     slot_size = size;
     return *this;
@@ -60,6 +61,12 @@ struct PublisherOptions {
     subscriber_queue_size = size;
     return *this;
   }
+  // Allow the server to use channel profile data to override sizing values
+  // when a profile file is configured. Set false to force the requested sizes.
+  PublisherOptions &SetApplyProfile(bool v) {
+    apply_profile = v;
+    return *this;
+  }
 
   // A public publisher's messages will be seen outside of the
   // publishing computer.
@@ -234,6 +241,7 @@ struct PublisherOptions {
   int32_t slot_size = 0;
   int32_t num_slots = 0;
   int32_t subscriber_queue_size = 0;
+  bool apply_profile = true;
 
   bool local = false;
   bool reliable = false;
@@ -290,6 +298,13 @@ struct SubscriberOptions {
   const std::string &Type() const { return type; }
   int MaxSharedPtrs() const { return max_active_messages - 1; }
   int MaxActiveMessages() const { return max_active_messages; }
+  bool ApplyProfile() const { return apply_profile; }
+  // Allow the server to use channel profile data to override sizing values
+  // when a profile file is configured. Set false to force the requested values.
+  SubscriberOptions &SetApplyProfile(bool v) {
+    apply_profile = v;
+    return *this;
+  }
   bool LogDroppedMessages() const { return log_dropped_messages; }
   void SetLogDroppedMessages(bool v) { log_dropped_messages = v; }
   bool DetectDroppedMessages() const { return detect_dropped_messages; }
@@ -381,6 +396,7 @@ struct SubscriberOptions {
   bool for_tunnel = false;
   std::string type;
   int max_active_messages = 1;
+  bool apply_profile = true;
   bool log_dropped_messages = true;
   bool detect_dropped_messages = true;
   bool pass_activation = false; // If true, the subscriber will pass activation

client/publisher.cc

@@ -76,8 +76,7 @@ absl::Status PublisherImpl::CreateOrAttachBuffers(uint64_t final_slot_size) {
             std::make_unique<BufferSet>(*size, current_slot_size, *addr);
         buffer_set->fd = std::move(*shm_fd);
         buffers_.push_back(std::move(buffer_set));
-        bcb_->sizes[buffers_.size()].store(final_buffer_size,
-                                           std::memory_order_relaxed);
+        bcb_->sizes[buffer_index].store(*size, std::memory_order_relaxed);
       } else {
         // We successfully created the /dev/shm file.
         bcb_->sizes[buffers_.size()].store(final_buffer_size,

client/python/client.cc

@@ -116,6 +116,10 @@ PYBIND11_MODULE(subspace, m) {
            "Set each subscriber queue's capacity. 0 disables the queue.")
       .def("subscriber_queue_size", &PublisherOptions::SubscriberQueueSize,
            "Get each subscriber queue's configured capacity.")
+      .def("set_apply_profile", &PublisherOptions::SetApplyProfile,
+           "Set whether the server may apply channel profile sizing.")
+      .def("apply_profile", &PublisherOptions::ApplyProfile,
+           "Get whether the server may apply channel profile sizing.")
       .def("set_notify_retirement", &PublisherOptions::SetNotifyRetirement,
            "Set whether the publisher notifies on message retirement.")
       .def("notify_retirement", &PublisherOptions::NotifyRetirement,
@@ -158,6 +162,10 @@ PYBIND11_MODULE(subspace, m) {
            "Set the maximum number of active messages for the subscriber.")
       .def("max_active_messages", &SubscriberOptions::MaxActiveMessages,
            "Get the maximum number of active messages for the subscriber.")
+      .def("set_apply_profile", &SubscriberOptions::SetApplyProfile,
+           "Set whether the server may apply channel profile sizing.")
+      .def("apply_profile", &SubscriberOptions::ApplyProfile,
+           "Get whether the server may apply channel profile sizing.")
       .def("set_log_dropped_messages",
            &SubscriberOptions::SetLogDroppedMessages,
            "Sets whether the subscriber logs dropped messages.")
@@ -586,6 +594,9 @@ checksum_error).  Use as a context manager to auto-release the slot:
                        &Subscriber::SubscriberQueueSize,
                        "Get each subscriber queue's resolved capacity.");
 
+  subscriber_class.def("max_active_messages", &Subscriber::MaxActiveMessages,
+                       "Get the resolved max active message count.");
+
   subscriber_class.def("get_current_ordinal", &Subscriber::GetCurrentOrdinal,
                        "Get the most recently received ordinal.");
 

client/python/client_test.py

@@ -114,6 +114,9 @@ def test_publisher_accessors(self):
         opts.set_num_slots(8)
         opts.set_type("my_type")
         opts.set_subscriber_queue_size(11)
+        self.assertTrue(opts.apply_profile())
+        opts.set_apply_profile(False)
+        self.assertFalse(opts.apply_profile())
         pub = client.create_publisher(channel_name="ch_pub_acc",
                                       options=opts)
         self.assertEqual(pub.type(), "my_type")
@@ -138,6 +141,10 @@ def test_subscriber_accessors(self):
                                       options=opts)
         sub = client.create_subscriber(channel_name="ch_sub_acc",
                                        type="sub_type")
+        sub_opts = subspace.SubscriberOptions()
+        self.assertTrue(sub_opts.apply_profile())
+        sub_opts.set_apply_profile(False)
+        self.assertFalse(sub_opts.apply_profile())
 
         pub.publish_message(b"probe")
         sub.wait()