Various build fixes

edwardbr · edwardbr · commit a8f1eb6b0dd8 · 2026-02-08T21:33:52.000Z
diff --git a/demos/comprehensive/src/transport/benchmark.cpp b/demos/comprehensive/src/transport/benchmark.cpp
@@ -567,7 +567,6 @@ int main()
     {
         for (size_t blob_size : blob_sizes)
         {
-            fmt::print("blob_size {}", blob_size);
             auto result = run_spsc_benchmark(enc.enc, blob_size);
             if (result.error != rpc::error::OK())
             {
@@ -578,21 +577,21 @@ int main()
         }
     }
 
-    fmt::print("run_tcp_benchmark\n");
-    uint16_t tcp_port = 18900;
-    for (const auto& enc : encodings)
-    {
-        for (size_t blob_size : blob_sizes)
-        {
-            auto result = run_tcp_benchmark(enc.enc, blob_size, tcp_port++);
-            if (result.error != rpc::error::OK())
-            {
-                fmt::print("tcp    | {:>18} | {:>9} | error {}\n", enc.name, blob_size, result.error);
-                continue;
-            }
-            print_stats("tcp", enc.name, blob_size, result.stats);
-        }
-    }
+    // fmt::print("run_tcp_benchmark\n");
+    // uint16_t tcp_port = 18900;
+    // for (const auto& enc : encodings)
+    // {
+    //     for (size_t blob_size : blob_sizes)
+    //     {
+    //         auto result = run_tcp_benchmark(enc.enc, blob_size, tcp_port++);
+    //         if (result.error != rpc::error::OK())
+    //         {
+    //             fmt::print("tcp    | {:>18} | {:>9} | error {}\n", enc.name, blob_size, result.error);
+    //             continue;
+    //         }
+    //         print_stats("tcp", enc.name, blob_size, result.stats);
+    //     }
+    // }
 
     print_footer();
     return 0;
diff --git a/rpc/include/rpc/internal/remote_pointer.h b/rpc/include/rpc/internal/remote_pointer.h
@@ -752,7 +752,7 @@ namespace rpc
                     std::allocator_traits<decltype(actual_cb_allocator)>::deallocate(actual_cb_allocator, this, 1);
                 }
                 void* get_deleter_ptr(const std::type_info&) noexcept override { return nullptr; }
-                ~control_block_make_shared() DEFAULT_DESTRUCTOR
+                ~control_block_make_shared() { }
             };
         } // namespace __shared_ptr_control_block
 
diff --git a/telemetry/include/rpc/telemetry/multiplexing_telemetry_service.h b/telemetry/include/rpc/telemetry/multiplexing_telemetry_service.h
@@ -74,7 +74,7 @@ namespace rpc
         explicit multiplexing_telemetry_service(std::vector<std::shared_ptr<i_telemetry_service>>&& child_services);
 
         virtual ~multiplexing_telemetry_service() DEFAULT_DESTRUCTOR;
-        ;
+
         multiplexing_telemetry_service(const multiplexing_telemetry_service&) = delete;
         multiplexing_telemetry_service& operator=(const multiplexing_telemetry_service&) = delete;
 
diff --git a/transports/spsc/include/transports/spsc/transport.h b/transports/spsc/include/transports/spsc/transport.h
@@ -55,6 +55,7 @@ namespace rpc::spsc
         stdex::member_ptr<spsc_transport> keep_alive_;
 
         std::atomic<bool> peer_requested_disconnection_ = false;
+        std::atomic<bool> pumps_started_ = false;
 
         struct activity_tracker
         {
@@ -170,7 +171,7 @@ namespace rpc::spsc
             // Check if the operation was cancelled during shutdown
             if (res_payload.error_code != rpc::error::OK())
             {
-                RPC_DEBUG("call_peer returning cancelled error for zone: {} sequence_number: {}",
+                RPC_ERROR("call_peer returning cancelled error for zone: {} sequence_number: {}",
                     get_service()->get_zone_id().get_val(),
                     sequence_number);
                 CO_RETURN res_payload.error_code;
diff --git a/transports/spsc/src/transport.cpp b/transports/spsc/src/transport.cpp
@@ -359,6 +359,14 @@ namespace rpc::spsc
         auto self = shared_from_this();
         RPC_DEBUG("pump_send_and_receive zone={}", get_zone_id().get_val());
 
+        // Guard against multiple calls
+        bool expected = false;
+        if (!pumps_started_.compare_exchange_strong(expected, true))
+        {
+            RPC_ERROR("pump_send_and_receive called MULTIPLE TIMES on zone {} - BUG!", get_zone_id().get_val());
+            return;
+        }
+
         // Schedule both producer and consumer tasks
         auto svc = get_service();
         auto scheduler = svc->get_scheduler();
@@ -429,7 +437,7 @@ namespace rpc::spsc
                                 stop_loop = true;
                                 break;
                             }
-                            // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));
+                            // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));
                             CO_AWAIT svc->get_scheduler()->schedule();
                         }
                         break;
@@ -457,7 +465,7 @@ namespace rpc::spsc
                     auto str_err = rpc::from_yas_binary(rpc::span(prefix_buf), prefix);
                     if (!str_err.empty())
                     {
-                        RPC_ERROR("failed invalid prefix");
+                        RPC_ERROR("Instance #{} deserialization FAILED: {}", my_instance, str_err);
                         break;
                     }
                     assert(prefix.direction);
@@ -484,9 +492,14 @@ namespace rpc::spsc
                     message_blob blob;
                     if (!receive_spsc_queue_->pop(blob))
                     {
+                        // if (get_status() == rpc::transport_status::DISCONNECTING)
+                        // {
+                        //     stop_loop = true;
+                        //     break;
+                        // }
                         if (!received_any)
                         {
-                            // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));
+                            // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));
                             CO_AWAIT svc->get_scheduler()->schedule();
                         }
                         break;
@@ -506,6 +519,9 @@ namespace rpc::spsc
                     }
                 }
 
+                if (stop_loop)
+                    break;
+
                 if (receive_data.empty())
                 {
                     envelope_payload payload;
@@ -682,7 +698,7 @@ namespace rpc::spsc
             auto status = push_message(send_data);
             if (status == send_queue_status::SEND_QUEUE_EMPTY || status == send_queue_status::SPSC_QUEUE_FULL)
             {
-                // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));
+                // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));
                 CO_AWAIT svc->get_scheduler()->schedule();
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -752,7 +752,7 @@ namespace rpc`
`752`	`752`	`std::allocator_traits<decltype(actual_cb_allocator)>::deallocate(actual_cb_allocator, this, 1);`
`753`	`753`	`}`
`754`	`754`	`void* get_deleter_ptr(const std::type_info&) noexcept override { return nullptr; }`
`755`		`- ~control_block_make_shared() DEFAULT_DESTRUCTOR`
	`755`	`+ ~control_block_make_shared() { }`
`756`	`756`	`};`
`757`	`757`	`} // namespace __shared_ptr_control_block`
`758`	`758`
Original file line number	Diff line number	Diff line change
`@@ -55,6 +55,7 @@ namespace rpc::spsc`
`55`	`55`	`stdex::member_ptr<spsc_transport> keep_alive_;`
`56`	`56`
`57`	`57`	`std::atomic<bool> peer_requested_disconnection_ = false;`
	`58`	`+ std::atomic<bool> pumps_started_ = false;`
`58`	`59`
`59`	`60`	`struct activity_tracker`
`60`	`61`	`{`
`@@ -170,7 +171,7 @@ namespace rpc::spsc`
`170`	`171`	`// Check if the operation was cancelled during shutdown`
`171`	`172`	`if (res_payload.error_code != rpc::error::OK())`
`172`	`173`	`{`
`173`		`- RPC_DEBUG("call_peer returning cancelled error for zone: {} sequence_number: {}",`
	`174`	`+ RPC_ERROR("call_peer returning cancelled error for zone: {} sequence_number: {}",`
`174`	`175`	`get_service()->get_zone_id().get_val(),`
`175`	`176`	`sequence_number);`
`176`	`177`	`CO_RETURN res_payload.error_code;`
Original file line number	Diff line number	Diff line change
`@@ -359,6 +359,14 @@ namespace rpc::spsc`
`359`	`359`	`auto self = shared_from_this();`
`360`	`360`	`RPC_DEBUG("pump_send_and_receive zone={}", get_zone_id().get_val());`
`361`	`361`
	`362`	`+ // Guard against multiple calls`
	`363`	`+ bool expected = false;`
	`364`	`+ if (!pumps_started_.compare_exchange_strong(expected, true))`
	`365`	`+ {`
	`366`	`+ RPC_ERROR("pump_send_and_receive called MULTIPLE TIMES on zone {} - BUG!", get_zone_id().get_val());`
	`367`	`+ return;`
	`368`	`+ }`
	`369`	`+`
`362`	`370`	`// Schedule both producer and consumer tasks`
`363`	`371`	`auto svc = get_service();`
`364`	`372`	`auto scheduler = svc->get_scheduler();`
`@@ -429,7 +437,7 @@ namespace rpc::spsc`
`429`	`437`	`stop_loop = true;`
`430`	`438`	`break;`
`431`	`439`	`}`
`432`		`- // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));`
	`440`	`+ // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));`
`433`	`441`	`CO_AWAIT svc->get_scheduler()->schedule();`
`434`	`442`	`}`
`435`	`443`	`break;`
`@@ -457,7 +465,7 @@ namespace rpc::spsc`
`457`	`465`	`auto str_err = rpc::from_yas_binary(rpc::span(prefix_buf), prefix);`
`458`	`466`	`if (!str_err.empty())`
`459`	`467`	`{`
`460`		`- RPC_ERROR("failed invalid prefix");`
	`468`	`+ RPC_ERROR("Instance #{} deserialization FAILED: {}", my_instance, str_err);`
`461`	`469`	`break;`
`462`	`470`	`}`
`463`	`471`	`assert(prefix.direction);`
`@@ -484,9 +492,14 @@ namespace rpc::spsc`
`484`	`492`	`message_blob blob;`
`485`	`493`	`if (!receive_spsc_queue_->pop(blob))`
`486`	`494`	`{`
	`495`	`+ // if (get_status() == rpc::transport_status::DISCONNECTING)`
	`496`	`+ // {`
	`497`	`+ // stop_loop = true;`
	`498`	`+ // break;`
	`499`	`+ // }`
`487`	`500`	`if (!received_any)`
`488`	`501`	`{`
`489`		`- // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));`
	`502`	`+ // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));`
`490`	`503`	`CO_AWAIT svc->get_scheduler()->schedule();`
`491`	`504`	`}`
`492`	`505`	`break;`
`@@ -506,6 +519,9 @@ namespace rpc::spsc`
`506`	`519`	`}`
`507`	`520`	`}`
`508`	`521`
	`522`	`+ if (stop_loop)`
	`523`	`+ break;`
	`524`	`+`
`509`	`525`	`if (receive_data.empty())`
`510`	`526`	`{`
`511`	`527`	`envelope_payload payload;`
`@@ -682,7 +698,7 @@ namespace rpc::spsc`
`682`	`698`	`auto status = push_message(send_data);`
`683`	`699`	`if (status == send_queue_status::SEND_QUEUE_EMPTY \|\| status == send_queue_status::SPSC_QUEUE_FULL)`
`684`	`700`	`{`
`685`		`- // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(100));`
	`701`	`+ // CO_AWAIT svc->get_scheduler()->yield_for(std::chrono::milliseconds(1));`
`686`	`702`	`CO_AWAIT svc->get_scheduler()->schedule();`
`687`	`703`	`}`
`688`	`704`	`}`