diff --git a/c_client/client_test.cc b/c_client/client_test.cc index 64ec9a80..840123f5 100644 --- a/c_client/client_test.cc +++ b/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); diff --git a/c_client/subspace.cc b/c_client/subspace.cc index f54870fe..4bd39f30 100644 --- a/c_client/subspace.cc +++ b/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; diff --git a/c_client/subspace.h b/c_client/subspace.h index 565002af..5eaed0af 100644 --- a/c_client/subspace.h +++ b/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 { diff --git a/client/client.cc b/client/client.cc index 51e84f13..e21f7a48 100644 --- a/client/client.cc +++ b/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 channel = std::make_shared( - 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 channel = std::make_shared( @@ -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( diff --git a/client/client.h b/client/client.h index 3b1d7264..e8af3e77 100644 --- a/client/client.h +++ b/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> &GetBuffers() const { return client_->GetBuffers(impl_.get()); diff --git a/client/client_channel.cc b/client/client_channel.cc index 3ae8cba4..aa3b7412 100644 --- a/client/client_channel.cc +++ b/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 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))); diff --git a/client/latency_test.cc b/client/latency_test.cc index 2a76769f..fad84efb 100644 --- a/client/latency_test.cc +++ b/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 &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 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 << ","; diff --git a/client/options.h b/client/options.h index a9619dc7..cc90b6ca 100644 --- a/client/options.h +++ b/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 diff --git a/client/publisher.cc b/client/publisher.cc index efb0c3d4..93e18d0c 100644 --- a/client/publisher.cc +++ b/client/publisher.cc @@ -76,8 +76,7 @@ absl::Status PublisherImpl::CreateOrAttachBuffers(uint64_t final_slot_size) { std::make_unique(*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, diff --git a/client/python/client.cc b/client/python/client.cc index 1fb3ad14..ed67a920 100644 --- a/client/python/client.cc +++ b/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."); diff --git a/client/python/client_test.py b/client/python/client_test.py index 8bca64b4..048608de 100644 --- a/client/python/client_test.py +++ b/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() diff --git a/proto/subspace.proto b/proto/subspace.proto index 37d25fcd..b9b05fea 100644 --- a/proto/subspace.proto +++ b/proto/subspace.proto @@ -43,6 +43,9 @@ message CreatePublisherRequest { bool split_buffers_over_bridge = 18; // Remote bridge publisher uses split buffers. // Entries in each subscriber's CCB slot queue. 0 disables the queue. int32 subscriber_queue_size = 19; + // If true, the server may use a loaded channel profile to override requested + // sizing values. If false, requested values are treated as authoritative. + bool apply_profile = 20; } message CreatePublisherResponse { @@ -60,6 +63,8 @@ message CreatePublisherResponse { int32 retirement_fd_index = 12; // My retirement fd index (read end) repeated int32 retirement_fd_indexes = 13; // Write end of all retirement fds. int32 subscriber_queue_size = 14; // Resolved capacity; 0 means disabled. + int32 slot_size = 15; // Resolved slot size. + int32 num_slots = 16; // Resolved number of slots. } // This is used both to create a new subscriber and to reload @@ -75,6 +80,9 @@ message CreateSubscriberRequest { bool for_tunnel = 7; string mux = 8; int32 vchan_id = 9; + // If true, the server may use a loaded channel profile to override requested + // sizing values. If false, requested values are treated as authoritative. + bool apply_profile = 10; } message CreateSubscriberResponse { @@ -96,6 +104,7 @@ message CreateSubscriberResponse { int32 metadata_size = 16; // Bytes reserved for user metadata (from publisher). bool use_split_buffers = 17; int32 subscriber_queue_size = 18; // Resolved capacity; 0 means disabled. + int32 max_active_messages = 19; // Resolved max active messages. } message GetTriggersRequest { string channel_name = 1; } @@ -266,6 +275,41 @@ message ChannelStatsProto { int32 num_bridge_subs = 11; // Number of subscribers that are bridges. } +message ChannelProfileObservation { + int32 slot_size = 1; + int32 num_slots = 2; + int32 subscriber_queue_size = 3; + uint32 max_message_size = 4; + uint64 total_messages = 5; + uint64 total_drops = 6; + uint32 num_resizes = 7; + int32 num_pubs = 8; + int32 num_subs = 9; + int32 num_reliable_pubs = 10; + int32 num_reliable_subs = 11; + uint64 sample_count = 12; + int32 max_active_messages = 13; +} + +message ChannelProfileRecommendation { + int32 slot_size = 1; + int32 num_slots = 2; + int32 subscriber_queue_size = 3; + int32 max_active_messages = 4; + string reason = 5; +} + +message ChannelProfile { + string channel_name = 1; + ChannelProfileObservation observed = 2; + ChannelProfileRecommendation recommended = 3; +} + +message ChannelProfileFile { + int32 version = 1; + repeated ChannelProfile channels = 2; +} + // This is published to the /subspace/Statistics channel. message Statistics { string server_id = 1; diff --git a/rust_client/src/client.rs b/rust_client/src/client.rs index 58c61ae1..5548ca2b 100644 --- a/rust_client/src/client.rs +++ b/rust_client/src/client.rs @@ -878,6 +878,7 @@ impl Client { use_split_buffers: opts.use_split_buffers, split_buffers_over_bridge: opts.split_buffers_over_bridge, subscriber_queue_size: opts.subscriber_queue_size, + apply_profile: opts.apply_profile, max_publishers: 0, publisher_id: -1, }, @@ -899,9 +900,19 @@ impl Client { return Err(SubspaceError::ServerError(pub_resp.error)); } + let resolved_num_slots = if pub_resp.num_slots > 0 { + pub_resp.num_slots + } else { + opts.num_slots + }; + let resolved_slot_size = if pub_resp.slot_size > 0 { + pub_resp.slot_size as u64 + } else { + slot_size as u64 + }; let mut pub_impl = PublisherImpl::new( channel_name.to_string(), - opts.num_slots, + resolved_num_slots, pub_resp.subscriber_queue_size, pub_resp.channel_id, pub_resp.publisher_id, @@ -937,7 +948,7 @@ impl Client { prot, )?; - pub_impl.create_or_attach_buffers(slot_size as u64)?; + pub_impl.create_or_attach_buffers(resolved_slot_size)?; register_pending_client_buffers(&mut client, &mut pub_impl.channel)?; pub_impl.trigger_fd = fds[pub_resp.pub_trigger_fd_index as usize]; @@ -1010,6 +1021,7 @@ impl Client { max_active_messages: opts.max_active_messages, mux: opts.mux.clone(), vchan_id: opts.vchan_id, + apply_profile: opts.apply_profile, }, )), }; @@ -1029,6 +1041,10 @@ impl Client { return Err(SubspaceError::ServerError(sub_resp.error)); } + let mut resolved_options = opts.clone(); + if sub_resp.max_active_messages > 0 { + resolved_options.max_active_messages = sub_resp.max_active_messages; + } let mut sub_impl = SubscriberImpl::new( channel_name.to_string(), sub_resp.num_slots, @@ -1038,7 +1054,7 @@ impl Client { sub_resp.vchan_id, client.session_id, String::from_utf8_lossy(&sub_resp.r#type).to_string(), - opts.clone(), + resolved_options, ); sub_impl.channel.use_split_buffers = sub_resp.use_split_buffers; sub_impl.channel.split_buffer_callbacks = opts.split_buffer_callbacks.clone(); @@ -1505,8 +1521,14 @@ fn reload_subscriber(client: &mut ClientInner, sub: &mut SubscriberImpl) -> Resu proto::CreateSubscriberRequest { channel_name: sub.channel.name.clone(), subscriber_id: sub.subscriber_id, + is_reliable: sub.options.reliable, + is_bridge: sub.options.bridge, + for_tunnel: sub.options.for_tunnel, + r#type: sub.options.channel_type.as_bytes().to_vec(), + max_active_messages: sub.options.max_active_messages, mux: sub.options.mux.clone(), - ..Default::default() + vchan_id: sub.options.vchan_id, + apply_profile: sub.options.apply_profile, }, )), }; @@ -1526,6 +1548,9 @@ fn reload_subscriber(client: &mut ClientInner, sub: &mut SubscriberImpl) -> Resu } sub.channel.num_slots = sub_resp.num_slots; sub.channel.subscriber_queue_size = sub_resp.subscriber_queue_size; + if sub_resp.max_active_messages > 0 { + sub.options.max_active_messages = sub_resp.max_active_messages; + } sub.channel .embargoed_slots .resize(sub_resp.num_slots as usize); diff --git a/rust_client/src/options.rs b/rust_client/src/options.rs index 4f46d78f..6b1e747d 100644 --- a/rust_client/src/options.rs +++ b/rust_client/src/options.rs @@ -14,6 +14,7 @@ pub struct PublisherOptions { pub slot_size: i32, pub num_slots: i32, pub subscriber_queue_size: i32, + pub apply_profile: bool, pub local: bool, pub reliable: bool, pub bridge: bool, @@ -38,6 +39,7 @@ impl Default for PublisherOptions { slot_size: 0, num_slots: 0, subscriber_queue_size: 0, + apply_profile: true, local: false, reliable: false, bridge: false, @@ -83,6 +85,11 @@ impl PublisherOptions { self } + pub fn set_apply_profile(mut self, v: bool) -> Self { + self.apply_profile = v; + self + } + pub fn set_local(mut self, v: bool) -> Self { self.local = v; self @@ -199,6 +206,7 @@ pub struct SubscriberOptions { pub for_tunnel: bool, pub channel_type: String, pub max_active_messages: i32, + pub apply_profile: bool, pub log_dropped_messages: bool, pub detect_dropped_messages: bool, pub pass_activation: bool, @@ -219,6 +227,7 @@ impl Default for SubscriberOptions { for_tunnel: false, channel_type: String::new(), max_active_messages: 1, + apply_profile: true, log_dropped_messages: true, detect_dropped_messages: true, pass_activation: false, @@ -258,6 +267,11 @@ impl SubscriberOptions { self } + pub fn set_apply_profile(mut self, v: bool) -> Self { + self.apply_profile = v; + self + } + pub fn set_log_dropped_messages(mut self, v: bool) -> Self { self.log_dropped_messages = v; self diff --git a/rust_client/tests/client_test.rs b/rust_client/tests/client_test.rs index e15eedac..0bf82fb4 100644 --- a/rust_client/tests/client_test.rs +++ b/rust_client/tests/client_test.rs @@ -55,6 +55,7 @@ fn publisher_options_defaults() { assert_eq!(opts.slot_size, 0); assert_eq!(opts.num_slots, 0); assert_eq!(opts.subscriber_queue_size, 0); + assert!(opts.apply_profile); assert!(!opts.local); assert!(!opts.reliable); assert!(!opts.bridge); @@ -74,6 +75,7 @@ fn publisher_options_builder_chain() { .set_slot_size(4096) .set_num_slots(16) .set_subscriber_queue_size(32) + .set_apply_profile(false) .set_reliable(true) .set_local(true) .set_fixed_size(true) @@ -88,6 +90,7 @@ fn publisher_options_builder_chain() { assert_eq!(opts.slot_size, 4096); assert_eq!(opts.num_slots, 16); assert_eq!(opts.subscriber_queue_size, 32); + assert!(!opts.apply_profile); assert!(opts.reliable); assert!(opts.local); assert!(opts.fixed_size); @@ -106,6 +109,7 @@ fn subscriber_options_defaults() { assert!(!opts.reliable); assert!(!opts.bridge); assert_eq!(opts.max_active_messages, 1); + assert!(opts.apply_profile); assert!(opts.log_dropped_messages); assert!(opts.detect_dropped_messages); assert!(!opts.pass_activation); @@ -121,6 +125,7 @@ fn subscriber_options_builder_chain() { let opts = SubscriberOptions::new() .set_reliable(true) .set_max_active_messages(8) + .set_apply_profile(false) .set_log_dropped_messages(false) .set_detect_dropped_messages(false) .set_pass_activation(true) @@ -132,6 +137,7 @@ fn subscriber_options_builder_chain() { assert!(opts.reliable); assert_eq!(opts.max_active_messages, 8); + assert!(!opts.apply_profile); assert!(!opts.log_dropped_messages); assert!(!opts.detect_dropped_messages); assert!(opts.pass_activation); diff --git a/server/BUILD.bazel b/server/BUILD.bazel index 50f0db9e..a9131d2e 100644 --- a/server/BUILD.bazel +++ b/server/BUILD.bazel @@ -29,6 +29,7 @@ cc_library( "@abseil-cpp//absl/flags:flag", "@abseil-cpp//absl/flags:parse", "@coroutines//co:co", + "@protobuf//:protobuf", ], ) diff --git a/server/client_handler.cc b/server/client_handler.cc index f48f964d..fe6f8319 100644 --- a/server/client_handler.cc +++ b/server/client_handler.cc @@ -332,19 +332,40 @@ void ClientHandler::HandleCreatePublisher( response->set_error("subscriber_queue_size must be >= 0"); return; } + int slot_size = req.slot_size(); + int num_slots = req.num_slots(); + int requested_subscriber_queue_size = req.subscriber_queue_size(); + if (req.apply_profile()) { + if (const ChannelProfileRecommendation *rec = + server_->GetProfileRecommendation(req.channel_name()); + rec != nullptr) { + if (rec->slot_size() > slot_size) { + slot_size = rec->slot_size(); + } + if (rec->num_slots() > num_slots) { + num_slots = rec->num_slots(); + } + requested_subscriber_queue_size = rec->subscriber_queue_size(); + server_->logger_.Log(toolbelt::LogLevel::kDebug, + "Applied channel profile to publisher %s: " + "slot_size=%d num_slots=%d subscriber_queue_size=%d", + req.channel_name().c_str(), slot_size, num_slots, + requested_subscriber_queue_size); + } + } const int subscriber_queue_size = - ResolveSubscriberQueueSize(req.num_slots(), req.subscriber_queue_size()); + ResolveSubscriberQueueSize(num_slots, requested_subscriber_queue_size); ServerChannel *channel = server_->FindChannel(req.channel_name()); if (channel == nullptr) { server_->logger_.Log(toolbelt::LogLevel::kDebug, "Publisher %s is creating new channel %s with size " "%d/%d and type length %zu (total of %zu channels)", client_name_.c_str(), req.channel_name().c_str(), - req.slot_size(), req.num_slots(), req.type().size(), + slot_size, num_slots, req.type().size(), server_->GetNumChannels()); absl::StatusOr ch = server_->CreateChannel( - req.channel_name(), req.slot_size(), req.num_slots(), - subscriber_queue_size, req.mux(), req.vchan_id(), req.type()); + req.channel_name(), slot_size, num_slots, subscriber_queue_size, + req.mux(), req.vchan_id(), req.type()); if (!ch.ok()) { response->set_error(ch.status().ToString()); return; @@ -355,12 +376,12 @@ void ClientHandler::HandleCreatePublisher( toolbelt::LogLevel::kDebug, "Publisher %s is remapping placeholder channel %s with size %d/%d and " "type length %zu (total of %zu channels)", - client_name_.c_str(), req.channel_name().c_str(), req.slot_size(), - req.num_slots(), req.type().size(), server_->GetNumChannels()); + client_name_.c_str(), req.channel_name().c_str(), slot_size, + num_slots, req.type().size(), server_->GetNumChannels()); // Channel exists, but it's just a placeholder. Remap the memory now // that we know the slots. absl::Status status = server_->RemapChannel( - channel, req.slot_size(), req.num_slots(), subscriber_queue_size); + channel, slot_size, num_slots, subscriber_queue_size); if (!status.ok()) { response->set_error(status.ToString()); return; @@ -457,15 +478,15 @@ void ClientHandler::HandleCreatePublisher( int current_num_slots = channel->NumSlots(); bool slot_size_changed = - channel->SlotSize() != 0 && req.slot_size() > channel->SlotSize(); - bool num_slots_changed = req.num_slots() > current_num_slots; + channel->SlotSize() != 0 && slot_size > channel->SlotSize(); + bool num_slots_changed = num_slots > current_num_slots; bool subscriber_queue_size_changed = subscriber_queue_size != channel->SubscriberQueueSize(); if (num_slots_changed) { response->set_error(absl::StrFormat( "Failed to add publisher to %s with more slots (%d) than the current " "number (%d)", - req.channel_name(), req.num_slots(), current_num_slots)); + req.channel_name(), num_slots, current_num_slots)); return; } if (subscriber_queue_size_changed) { @@ -478,21 +499,19 @@ void ClientHandler::HandleCreatePublisher( } if (slot_size_changed) { - if (slot_size_changed) { - if (channel->IsFixedSize()) { - // Fixed size channels cannot change size. - response->set_error(absl::StrFormat( - "Failed to add publisher to fixed size channel %s with different " - "slot size (%d) than the current size (%d)", - req.channel_name(), req.slot_size(), channel->SlotSize())); - return; - } + if (channel->IsFixedSize()) { + // Fixed size channels cannot change size. + response->set_error(absl::StrFormat( + "Failed to add publisher to fixed size channel %s with different " + "slot size (%d) than the current size (%d)", + req.channel_name(), slot_size, channel->SlotSize())); + return; } server_->logger_.Log( toolbelt::LogLevel::kDebug, "Publisher %s is resizing channel %s buffers from %d bytes to %d", client_name_.c_str(), channel->Name().c_str(), channel->SlotSize(), - req.slot_size()); + slot_size); } if (channel->IsLocal() != req.is_local()) { @@ -641,6 +660,8 @@ void ClientHandler::HandleCreatePublisher( response->set_vchan_id(channel->GetVirtualChannelId()); response->set_publisher_id(pub->GetId()); response->set_subscriber_queue_size(channel->SubscriberQueueSize()); + response->set_slot_size(channel->SlotSize()); + response->set_num_slots(channel->NumSlots()); const SharedMemoryFds &channel_fds = channel->GetFds(); response->set_ccb_fd_index(0); @@ -700,6 +721,23 @@ void ClientHandler::HandleCreateSubscriber( const subspace::CreateSubscriberRequest &req, subspace::CreateSubscriberResponse *response, std::vector &fds) { + int max_active_messages = req.max_active_messages(); + if (req.apply_profile()) { + if (const ChannelProfileRecommendation *rec = + server_->GetProfileRecommendation(req.channel_name()); + rec != nullptr && rec->max_active_messages() > max_active_messages) { + max_active_messages = rec->max_active_messages(); + server_->logger_.Log(toolbelt::LogLevel::kDebug, + "Applied channel profile to subscriber %s: " + "max_active_messages=%d", + req.channel_name().c_str(), max_active_messages); + } + } + if (max_active_messages < 1) { + response->set_error("max_active_messages must be >= 1"); + return; + } + ServerChannel *channel = server_->FindChannel(req.channel_name()); if (channel == nullptr) { // No channel exists, map an empty channel. @@ -778,7 +816,7 @@ void ClientHandler::HandleCreateSubscriber( } else { if (!req.is_reliable()) { absl::Status cap_ok = - channel->HasSufficientCapacity(req.max_active_messages() - 1); + channel->HasSufficientCapacity(max_active_messages - 1); if (!cap_ok.ok()) { response->set_error(absl::StrFormat( "Insufficient capacity to add a new subscriber to channel %s: %s", @@ -793,7 +831,7 @@ void ClientHandler::HandleCreateSubscriber( GetTotalVM().c_str()); absl::StatusOr subscriber = channel->AddSubscriber(this, req.is_reliable(), req.is_bridge(), - req.for_tunnel(), req.max_active_messages()); + req.for_tunnel(), max_active_messages); if (!subscriber.ok()) { response->set_error(subscriber.status().ToString()); return; @@ -801,6 +839,7 @@ void ClientHandler::HandleCreateSubscriber( channel->RecordUpdate(/*is_pub=*/false, /*add=*/true, req.is_reliable()); sub = *subscriber; } + sub->SetMaxActiveMessages(max_active_messages); if (!reclaimed) { server_->OnNewSubscriber(channel->Name(), sub->GetId()); @@ -844,6 +883,7 @@ void ClientHandler::HandleCreateSubscriber( response->set_slot_size(channel->SlotSize()); response->set_num_slots(channel->NumSlots()); response->set_subscriber_queue_size(channel->SubscriberQueueSize()); + response->set_max_active_messages(max_active_messages); response->set_checksum_size(channel->ChecksumSize()); response->set_metadata_size(channel->MetadataSize()); ServerChannel *split_response_channel = diff --git a/server/main.cc b/server/main.cc index 13a29ad0..0a4837b8 100644 --- a/server/main.cc +++ b/server/main.cc @@ -50,6 +50,9 @@ ABSL_FLAG(bool, bridge_ports_fallback_ephemeral, false, "If true, use an ephemeral TCP bridge port when --bridge_ports is " "configured but unavailable."); ABSL_FLAG(std::string, log_level, "info", "Log level"); +ABSL_FLAG(std::string, profile_file, "", + "Optional channel profile textproto file. Empty disables profile " + "loading, writing, and sizing overrides."); ABSL_FLAG(std::string, interface, "", "Discovery network interface"); ABSL_FLAG(bool, local, false, "Use local computer only"); ABSL_FLAG(bool, tcp_discovery, false, @@ -197,6 +200,7 @@ int main(int argc, char **argv) { } server->SetLogLevel(absl::GetFlag(FLAGS_log_level)); + server->SetProfileFile(absl::GetFlag(FLAGS_profile_file)); int bridge_first_port = 0; int bridge_last_port = 0; if (!ParseBridgePorts(absl::GetFlag(FLAGS_bridge_ports), &bridge_first_port, diff --git a/server/python/server.cc b/server/python/server.cc index 4bb9af2f..cba4ba12 100644 --- a/server/python/server.cc +++ b/server/python/server.cc @@ -19,8 +19,9 @@ namespace py = pybind11; // any of those details. class ServerWrapper { public: - ServerWrapper(std::string socket_name, bool local) - : socket_name_(std::move(socket_name)), local_(local) {} + ServerWrapper(std::string socket_name, bool local, std::string profile_file) + : socket_name_(std::move(socket_name)), local_(local), + profile_file_(std::move(profile_file)) {} ~ServerWrapper() { if (running_) { @@ -44,6 +45,7 @@ class ServerWrapper { /*disc_port=*/0, /*peer_port=*/0, local_, /*notify_fd=*/pipe_fds[1]); + server_->SetProfileFile(profile_file_); server_thread_ = std::thread([this]() { absl::Status s = server_->Run(); @@ -93,6 +95,7 @@ class ServerWrapper { std::string socket_name_; bool local_; + std::string profile_file_; co::CoroutineScheduler scheduler_; std::unique_ptr server_; std::thread server_thread_; @@ -119,13 +122,15 @@ Or as a context manager: with subspace_server.Server("/tmp/my_socket", local=True) as server: # ... run tests against server.socket_name ... )doc") - .def(py::init(), + .def(py::init(), R"doc(Create a server bound to the given Unix-domain socket path. Args: socket_name: Path for the Unix-domain socket. - local: If True the server will not attempt network discovery.)doc", - py::arg("socket_name"), py::arg("local") = true) + local: If True the server will not attempt network discovery. + profile_file: Optional channel profile textproto path.)doc", + py::arg("socket_name"), py::arg("local") = true, + py::arg("profile_file") = "") .def("start", &ServerWrapper::Start, "Start the server on a background thread. Blocks until the " "server is ready to accept connections.") diff --git a/server/server.cc b/server/server.cc index 280a8a44..4b3ca8b3 100644 --- a/server/server.cc +++ b/server/server.cc @@ -13,7 +13,11 @@ #include "toolbelt/clock.h" #include "toolbelt/hexdump.h" #include "toolbelt/sockets.h" +#include "google/protobuf/text_format.h" +#include #include +#include +#include #include #include #include @@ -27,6 +31,118 @@ #include namespace subspace { +namespace { + +constexpr int kChannelProfileVersion = 1; + +int Align64(int value) { + if (value <= 0) { + return 0; + } + return (value + 63) & ~63; +} + +ChannelProfile *FindMutableProfile(ChannelProfileFile *profile, + const std::string &channel_name) { + for (auto &channel : *profile->mutable_channels()) { + if (channel.channel_name() == channel_name) { + return &channel; + } + } + ChannelProfile *channel = profile->add_channels(); + channel->set_channel_name(channel_name); + return channel; +} + +const ChannelProfile *FindProfile(const ChannelProfileFile &profile, + const std::string &channel_name) { + for (const auto &channel : profile.channels()) { + if (channel.channel_name() == channel_name) { + return &channel; + } + } + return nullptr; +} + +std::string BuildRecommendationReason(const ChannelProfileObservation &obs) { + std::vector reasons; + if (obs.total_drops() > 0) { + reasons.push_back("drops"); + } + if (obs.num_resizes() > 0) { + reasons.push_back("resizes"); + } + if (obs.max_message_size() > static_cast(obs.slot_size())) { + reasons.push_back("message-size"); + } + if (reasons.empty()) { + return "observed"; + } + std::string reason = reasons.front(); + for (size_t i = 1; i < reasons.size(); ++i) { + reason += ","; + reason += reasons[i]; + } + return reason; +} + +void ComputeRecommendation(const ChannelProfileObservation &obs, + ChannelProfileRecommendation *rec) { + const int current_slot_size = std::max(0, obs.slot_size()); + int recommended_slot_size = current_slot_size; + if (obs.max_message_size() > 0) { + const int with_headroom = + Align64(static_cast(obs.max_message_size()) + + std::max(64, static_cast(obs.max_message_size() / 8))); + recommended_slot_size = std::max(recommended_slot_size, with_headroom); + } + if (rec->slot_size() > 0) { + recommended_slot_size = std::max(recommended_slot_size, rec->slot_size()); + } + + const int current_max_active = std::max(1, obs.max_active_messages()); + int recommended_num_slots = std::max(0, obs.num_slots()); + const int capacity_floor = + std::max(recommended_num_slots, + obs.num_pubs() + obs.num_subs() + current_max_active + 8); + recommended_num_slots = std::max(recommended_num_slots, capacity_floor); + if (obs.total_drops() > 0) { + recommended_num_slots = + std::max(recommended_num_slots, std::max(1, obs.num_slots()) * 3 / 2 + 8); + } + if (rec->num_slots() > 0) { + recommended_num_slots = std::max(recommended_num_slots, rec->num_slots()); + } + + int recommended_queue_size = std::max(0, obs.subscriber_queue_size()); + if (obs.total_drops() > 0 && obs.num_reliable_subs() == 0) { + if (recommended_queue_size == 0) { + recommended_queue_size = 4; + } else { + recommended_queue_size = std::min(std::max(recommended_queue_size * 2, 1), + std::max(recommended_num_slots, 1)); + } + recommended_queue_size = std::min(recommended_queue_size, 1024); + } + if (rec->subscriber_queue_size() > 0) { + recommended_queue_size = + std::max(recommended_queue_size, rec->subscriber_queue_size()); + } + + int recommended_max_active = current_max_active; + if (rec->max_active_messages() > 0) { + recommended_max_active = + std::max(recommended_max_active, rec->max_active_messages()); + } + + rec->set_slot_size(recommended_slot_size); + rec->set_num_slots(recommended_num_slots); + rec->set_subscriber_queue_size(recommended_queue_size); + rec->set_max_active_messages(recommended_max_active); + rec->set_reason(BuildRecommendationReason(obs)); +} + +} // namespace // In multithreaded tests we can't dlclose the plugins because the dynamic // linker doesn't play well with threads. @@ -473,6 +589,159 @@ Server::~Server() { } } +void Server::SetProfileFile(std::string path) { + profile_file_ = std::move(path); +} + +absl::Status Server::LoadProfileFile() { + profile_.Clear(); + profile_.set_version(kChannelProfileVersion); + profile_loaded_ = false; + + if (profile_file_.empty()) { + return absl::OkStatus(); + } + if (!std::filesystem::exists(profile_file_)) { + profile_loaded_ = true; + return absl::OkStatus(); + } + + std::ifstream input(profile_file_); + if (!input.is_open()) { + return absl::InternalError( + absl::StrFormat("Failed to open profile file %s", profile_file_)); + } + std::string text((std::istreambuf_iterator(input)), + std::istreambuf_iterator()); + if (text.empty()) { + profile_loaded_ = true; + return absl::OkStatus(); + } + ChannelProfileFile loaded; + if (!google::protobuf::TextFormat::ParseFromString(text, &loaded)) { + return absl::InvalidArgumentError( + absl::StrFormat("Failed to parse profile file %s", profile_file_)); + } + if (loaded.version() != kChannelProfileVersion) { + logger_.Log(toolbelt::LogLevel::kWarning, + "Ignoring channel profile %s with unsupported version %d", + profile_file_.c_str(), loaded.version()); + profile_.Clear(); + profile_.set_version(kChannelProfileVersion); + return absl::OkStatus(); + } + + profile_ = std::move(loaded); + profile_loaded_ = true; + return absl::OkStatus(); +} + +absl::Status Server::SaveProfileFile() { + if (profile_file_.empty()) { + return absl::OkStatus(); + } + + std::string text; + profile_.set_version(kChannelProfileVersion); + if (!google::protobuf::TextFormat::PrintToString(profile_, &text)) { + return absl::InternalError("Failed to serialize channel profile"); + } + + const std::string tmp_file = profile_file_ + ".tmp"; + { + std::ofstream output(tmp_file, std::ios::out | std::ios::trunc); + if (!output.is_open()) { + return absl::InternalError( + absl::StrFormat("Failed to open temporary profile file %s", + tmp_file.c_str())); + } + output << text; + output.flush(); + if (!output.good()) { + return absl::InternalError( + absl::StrFormat("Failed to write profile file %s", tmp_file.c_str())); + } + } + if (std::rename(tmp_file.c_str(), profile_file_.c_str()) != 0) { + return absl::InternalError( + absl::StrFormat("Failed to rename %s to %s: %s", tmp_file.c_str(), + profile_file_.c_str(), strerror(errno))); + } + return absl::OkStatus(); +} + +void Server::SampleChannelProfiles() { + profile_.set_version(kChannelProfileVersion); + for (const auto &[name, channel] : channels_) { + if (channel == nullptr || channel->IsVirtual()) { + continue; + } + + auto *profile = FindMutableProfile(&profile_, name); + auto *obs = profile->mutable_observed(); + obs->set_slot_size(channel->SlotSize()); + obs->set_num_slots(channel->NumSlots()); + obs->set_subscriber_queue_size(channel->SubscriberQueueSize()); + + uint64_t total_bytes = 0; + uint64_t total_messages = 0; + uint32_t max_message_size = 0; + uint32_t total_drops = 0; + channel->GetStatsCounters(total_bytes, total_messages, max_message_size, + total_drops); + obs->set_total_messages(total_messages); + obs->set_total_drops(total_drops); + obs->set_max_message_size(max_message_size); + + const ChannelCounters &counters = scb_->counters[channel->GetChannelId()]; + obs->set_num_resizes(counters.num_resizes); + obs->set_num_pubs(counters.num_pubs); + obs->set_num_subs(counters.num_subs); + obs->set_num_reliable_pubs(counters.num_reliable_pubs); + obs->set_num_reliable_subs(counters.num_reliable_subs); + obs->set_sample_count(obs->sample_count() + 1); + + int max_active_messages = obs->max_active_messages(); + for (const auto &[id, user] : channel->GetUsers()) { + (void)id; + if (user->IsSubscriber()) { + max_active_messages = + std::max(max_active_messages, + static_cast(user.get()) + ->MaxActiveMessages()); + } + } + obs->set_max_active_messages(std::max(max_active_messages, 1)); + + ComputeRecommendation(*obs, profile->mutable_recommended()); + } +} + +void Server::ProfileCoroutine(async::Context ctx) { + constexpr int kProfilePeriodSecs = 2; + while (!shutting_down_) { + async::Sleep(ctx, kProfilePeriodSecs); + SampleChannelProfiles(); + if (absl::Status status = SaveProfileFile(); !status.ok()) { + logger_.Log(toolbelt::LogLevel::kError, + "Failed to save channel profile: %s", + status.ToString().c_str()); + } + } +} + +const ChannelProfileRecommendation * +Server::GetProfileRecommendation(const std::string &channel_name) const { + if (profile_file_.empty() || !profile_loaded_) { + return nullptr; + } + const ChannelProfile *profile = FindProfile(profile_, channel_name); + if (profile == nullptr || !profile->has_recommended()) { + return nullptr; + } + return &profile->recommended(); +} + void Server::Stop(bool force) { if (shutting_down_) { return; @@ -843,6 +1112,17 @@ absl::Status Server::Run(int num_asio_threads) { scb_ = *scb; } + if (!profile_file_.empty()) { + if (absl::Status profile_status = LoadProfileFile(); !profile_status.ok()) { + logger_.Log(toolbelt::LogLevel::kWarning, + "Ignoring channel profile %s: %s", profile_file_.c_str(), + profile_status.ToString().c_str()); + profile_.Clear(); + profile_.set_version(kChannelProfileVersion); + profile_loaded_ = true; + } + } + // Connect any shadow that wasn't tried during recovery so it can // receive the re-replication. if (!primary_connected && primary_shadow_replicator_ != nullptr) { @@ -1023,6 +1303,13 @@ absl::Status Server::Run(int num_asio_threads) { .interrupt_fd = shutdown_trigger_fd_.GetPollFd().Fd()}); } + if (!profile_file_.empty()) { + runtime_.SpawnOnStrand( + [this](async::Context ctx) { ProfileCoroutine(ctx); }, + {.name = "Channel profile", + .interrupt_fd = shutdown_trigger_fd_.GetPollFd().Fd()}); + } + if (!local_) { if (tcp_discovery_) { // TCP discovery: a server with a peer address dials it; otherwise we diff --git a/server/server.h b/server/server.h index 252b057e..74c93dac 100644 --- a/server/server.h +++ b/server/server.h @@ -29,6 +29,7 @@ #include "toolbelt/triggerfd.h" #include #include +#include #include namespace subspace { @@ -140,6 +141,7 @@ class Server { bool fallback_to_ephemeral = false); void SetCleanupFilesystem(bool v) { cleanup_filesystem_ = v; } + void SetProfileFile(std::string path); // Use a TCP connection (instead of UDP broadcast/unicast) for discovery. // This is useful when the two servers cannot exchange UDP datagrams @@ -265,6 +267,12 @@ class Server { void ChannelDirectoryCoroutine(async::Context ctx); void SendChannelDirectory(); void StatisticsCoroutine(async::Context ctx); + void ProfileCoroutine(async::Context ctx); + absl::Status LoadProfileFile(); + absl::Status SaveProfileFile(); + void SampleChannelProfiles(); + const ChannelProfileRecommendation * + GetProfileRecommendation(const std::string &channel_name) const; void DiscoveryReceiverCoroutine(async::Context ctx); void DiscoveryListenerCoroutine(async::Context ctx); void DiscoveryConnectorCoroutine(async::Context ctx); @@ -418,6 +426,9 @@ class Server { bool publish_server_channels_ = true; BridgePortRange bridge_port_range_; bool bridge_ports_fallback_to_ephemeral_ = false; + std::string profile_file_; + ChannelProfileFile profile_; + bool profile_loaded_ = false; std::unique_ptr primary_shadow_replicator_; std::unique_ptr secondary_shadow_replicator_; diff --git a/server/server_channel.h b/server/server_channel.h index 51b69677..341f6b43 100644 --- a/server/server_channel.h +++ b/server/server_channel.h @@ -88,6 +88,7 @@ class SubscriberUser : public User { max_active_messages_(max_active_messages) {} bool IsSubscriber() const override { return true; } int MaxActiveMessages() const { return max_active_messages_; } + void SetMaxActiveMessages(int n) { max_active_messages_ = n; } private: int max_active_messages_; diff --git a/server/server_test.cc b/server/server_test.cc index ac22e71e..85bf9dfe 100644 --- a/server/server_test.cc +++ b/server/server_test.cc @@ -13,7 +13,11 @@ #include "proto/subspace.pb.h" #include "toolbelt/fd.h" #include "toolbelt/sockets.h" +#include +#include +#include #include +#include // Helper to send raw Request protos and receive Response protos + FDs, // using the same wire format as the real client (4-byte length prefix, @@ -73,7 +77,7 @@ class RawConnection { int vchan_id = 0, bool for_tunnel = false, bool notify_retirement = false, int checksum_size = 0, int metadata_size = 0, int max_publishers = 0, - int subscriber_queue_size = 0) { + int subscriber_queue_size = 0, bool apply_profile = true) { subspace::Request req; auto *cmd = req.mutable_create_publisher(); cmd->set_channel_name(channel); @@ -91,6 +95,7 @@ class RawConnection { cmd->set_metadata_size(metadata_size); cmd->set_max_publishers(max_publishers); cmd->set_subscriber_queue_size(subscriber_queue_size); + cmd->set_apply_profile(apply_profile); cmd->set_publisher_id(-1); auto result = Send(req); return std::move(*result); @@ -101,7 +106,8 @@ class RawConnection { CreateSubscriber(const std::string &channel, const std::string &type = "", bool reliable = false, int max_active_messages = 4, const std::string &mux = "", - int vchan_id = 0, bool for_tunnel = false) { + int vchan_id = 0, bool for_tunnel = false, + bool apply_profile = true) { subspace::Request req; auto *cmd = req.mutable_create_subscriber(); cmd->set_channel_name(channel); @@ -112,6 +118,7 @@ class RawConnection { cmd->set_mux(mux); cmd->set_vchan_id(vchan_id); cmd->set_for_tunnel(for_tunnel); + cmd->set_apply_profile(apply_profile); auto result = Send(req); return std::move(*result); } @@ -122,6 +129,102 @@ class RawConnection { class ServerTest : public SubspaceTestBase {}; +class ProfileServerHarness { +public: + explicit ProfileServerHarness(std::string profile_file) + : profile_file_(std::move(profile_file)) { +#if defined(__ANDROID__) + char socket_name_template[] = "/data/local/tmp/subspace_profileXXXXXX"; // NOLINT +#else + char socket_name_template[] = "/tmp/subspace_profileXXXXXX"; // NOLINT +#endif + int fd = mkstemp(&socket_name_template[0]); + if (fd >= 0) { + ::close(fd); + } + socket_ = &socket_name_template[0]; + (void)remove(socket_.c_str()); + (void)pipe(server_pipe_); + server_ = std::make_unique( + engine_, socket_, "", 0, 0, /*local=*/true, server_pipe_[1], + /*initial_ordinal=*/1, /*wait_for_clients=*/false); + server_->SetProfileFile(profile_file_); + server_thread_ = std::thread([this]() { +#if SUBSPACE_CORO_BACKEND == SUBSPACE_CORO_BACKEND_ASIO + absl::Status s = server_->Run(/*num_asio_threads=*/1); +#else + absl::Status s = server_->Run(); +#endif + if (!s.ok()) { + fprintf(stderr, "Profile server error: %s\n", s.ToString().c_str()); + } + }); + char buf[8]; + (void)::read(server_pipe_[0], buf, 8); + } + + ~ProfileServerHarness() { Stop(); } + + void Stop() { + if (stopped_) { + return; + } + server_->Stop(); + char buf[8]; + (void)::read(server_pipe_[0], buf, 8); + server_thread_.join(); + server_->CleanupAfterSession(); + (void)remove(socket_.c_str()); + (void)::close(server_pipe_[0]); + (void)::close(server_pipe_[1]); + stopped_ = true; + } + + const std::string &Socket() const { return socket_; } + +private: + subspace::async::RuntimeEngine engine_; + std::string socket_; + std::string profile_file_; + int server_pipe_[2] = {-1, -1}; + std::unique_ptr server_; + std::thread server_thread_; + bool stopped_ = false; +}; + +std::string TempPath(const char *prefix) { +#if defined(__ANDROID__) + std::string pattern = std::string("/data/local/tmp/") + prefix + "XXXXXX"; +#else + std::string pattern = std::string("/tmp/") + prefix + "XXXXXX"; +#endif + std::vector buffer(pattern.begin(), pattern.end()); + buffer.push_back('\0'); + int fd = mkstemp(buffer.data()); + if (fd >= 0) { + ::close(fd); + } + std::string path(buffer.data()); + (void)remove(path.c_str()); + return path; +} + +void WriteProfile(const std::string &path, const std::string &channel, + int slot_size, int num_slots, int subscriber_queue_size, + int max_active_messages, int version = 1) { + std::ofstream out(path); + out << "version: " << version << "\n" + << "channels {\n" + << " channel_name: \"" << channel << "\"\n" + << " recommended {\n" + << " slot_size: " << slot_size << "\n" + << " num_slots: " << num_slots << "\n" + << " subscriber_queue_size: " << subscriber_queue_size << "\n" + << " max_active_messages: " << max_active_messages << "\n" + << " }\n" + << "}\n"; +} + // --------------------------------------------------------------------------- // Protocol-level tests // --------------------------------------------------------------------------- @@ -164,6 +267,201 @@ TEST_F(ServerTest, CreateSubscriberSuccess) { ASSERT_GE(static_cast(fds.size()), 4); } +TEST(ServerProfileTest, PublisherProfileOverrideReturnsResolvedValues) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_pub_override", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8); + ProfileServerHarness server(profile_file); + + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreatePublisher("profile_pub_override", /*slot_size=*/64, + /*num_slots=*/4, "", false, true, false, "", 0, + false, false, 0, 0, 0, + /*subscriber_queue_size=*/0, + /*apply_profile=*/true); + + ASSERT_TRUE(resp.create_publisher().error().empty()); + EXPECT_EQ(resp.create_publisher().slot_size(), 256); + EXPECT_EQ(resp.create_publisher().num_slots(), 12); + EXPECT_EQ(resp.create_publisher().subscriber_queue_size(), 4); + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, PublisherProfileOptOutPreservesRequestedValues) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_pub_optout", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8); + ProfileServerHarness server(profile_file); + + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreatePublisher("profile_pub_optout", /*slot_size=*/64, + /*num_slots=*/4, "", false, true, false, "", 0, + false, false, 0, 0, 0, + /*subscriber_queue_size=*/0, + /*apply_profile=*/false); + + ASSERT_TRUE(resp.create_publisher().error().empty()); + EXPECT_EQ(resp.create_publisher().slot_size(), 64); + EXPECT_EQ(resp.create_publisher().num_slots(), 4); + EXPECT_EQ(resp.create_publisher().subscriber_queue_size(), 0); + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, SubscriberProfileOverrideReturnsResolvedMaxActive) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_sub_override", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8); + ProfileServerHarness server(profile_file); + + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreateSubscriber("profile_sub_override", "", false, + /*max_active_messages=*/2, "", 0, false, + /*apply_profile=*/true); + + ASSERT_TRUE(resp.create_subscriber().error().empty()); + EXPECT_EQ(resp.create_subscriber().max_active_messages(), 8); + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, CppClientHandlesReportResolvedProfileValues) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_cpp_client", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8); + ProfileServerHarness server(profile_file); + + { + subspace::Client client; + ASSERT_OK(client.Init(server.Socket())); + auto pub = client.CreatePublisher( + "profile_cpp_client", + subspace::PublisherOptions() + .SetSlotSize(64) + .SetNumSlots(4) + .SetSubscriberQueueSize(0) + .SetApplyProfile(true)); + ASSERT_OK(pub); + EXPECT_EQ(pub->SlotSize(), 256); + EXPECT_EQ(pub->NumSlots(), 12); + EXPECT_EQ(pub->SubscriberQueueSize(), 4); + + auto sub = client.CreateSubscriber( + "profile_cpp_client", + subspace::SubscriberOptions() + .SetMaxActiveMessages(2) + .SetApplyProfile(true)); + ASSERT_OK(sub); + EXPECT_EQ(sub->NumSlots(), 12); + EXPECT_EQ(sub->SubscriberQueueSize(), 4); + EXPECT_EQ(sub->MaxActiveMessages(), 8); + } + + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, SubscriberProfileOptOutPreservesRequestedMaxActive) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_sub_optout", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8); + ProfileServerHarness server(profile_file); + + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreateSubscriber("profile_sub_optout", "", false, + /*max_active_messages=*/2, "", 0, false, + /*apply_profile=*/false); + + ASSERT_TRUE(resp.create_subscriber().error().empty()); + EXPECT_EQ(resp.create_subscriber().max_active_messages(), 2); + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, ProfileVersionMismatchIsIgnored) { + std::string profile_file = TempPath("subspace_profile_"); + WriteProfile(profile_file, "profile_version_mismatch", /*slot_size=*/256, + /*num_slots=*/12, /*subscriber_queue_size=*/4, + /*max_active_messages=*/8, /*version=*/99); + ProfileServerHarness server(profile_file); + + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreatePublisher("profile_version_mismatch", /*slot_size=*/64, + /*num_slots=*/4, "", false, true, false, "", 0, + false, false, 0, 0, 0, + /*subscriber_queue_size=*/0, + /*apply_profile=*/true); + + ASSERT_TRUE(resp.create_publisher().error().empty()); + EXPECT_EQ(resp.create_publisher().slot_size(), 64); + EXPECT_EQ(resp.create_publisher().num_slots(), 4); + EXPECT_EQ(resp.create_publisher().subscriber_queue_size(), 0); + server.Stop(); + (void)remove(profile_file.c_str()); +} + +TEST(ServerProfileTest, ProfilePersistsAndAppliesAfterRestart) { + std::string profile_file = TempPath("subspace_profile_"); + const char *channel = "profile_restart"; + { + ProfileServerHarness server(profile_file); + { + subspace::Client client; + ASSERT_OK(client.Init(server.Socket())); + auto pub = client.CreatePublisher( + channel, subspace::PublisherOptions() + .SetSlotSize(64) + .SetNumSlots(4) + .SetApplyProfile(true)); + ASSERT_OK(pub); + auto buffer = pub->GetMessageBuffer(32); + ASSERT_OK(buffer); + memset(*buffer, 0xAB, 32); + auto msg = pub->PublishMessage(32); + ASSERT_OK(msg); + std::this_thread::sleep_for(std::chrono::milliseconds(2500)); + } + server.Stop(); + } + + { + ProfileServerHarness server(profile_file); + RawConnection conn; + ASSERT_OK(conn.Connect(server.Socket())); + ASSERT_OK(conn.Init()); + auto [resp, fds] = + conn.CreatePublisher(channel, /*slot_size=*/64, /*num_slots=*/4, "", + false, true, false, "", 0, false, false, 0, 0, 0, + /*subscriber_queue_size=*/0, + /*apply_profile=*/true); + ASSERT_TRUE(resp.create_publisher().error().empty()); + EXPECT_GE(resp.create_publisher().num_slots(), 10); + server.Stop(); + } + (void)remove(profile_file.c_str()); +} + // --------------------------------------------------------------------------- // CreatePublisher error paths // ---------------------------------------------------------------------------