CXXCBC-285: write to sockets from IO threads (#348)

The writing procedure should be invoked from the IO thread and bound to
IO executor. Otherwise when main thread invokes ASIO write directly, the
OpenSSL structures are not guaranteed to be thread-safe.

Author: avsej

core/io/http_session.hxx

@@ -293,7 +293,7 @@ class http_session : public std::enable_shared_from_this<http_session>
         if (stopped_) {
             return;
         }
-        do_write();
+        asio::post(asio::bind_executor(ctx_, [self = shared_from_this()]() { self->do_write(); }));
     }
 
     template<typename Handler>

core/io/mcbp_session.cxx

@@ -851,7 +851,7 @@ class mcbp_session_impl
         if (stopped_) {
             return;
         }
-        do_write();
+        asio::post(asio::bind_executor(ctx_, [self = shared_from_this()]() { self->do_write(); }));
     }
 
     void write_and_flush(std::vector<std::byte>&& buf)
@@ -1397,8 +1397,10 @@ class mcbp_session_impl
                 std::scoped_lock inner_lock(self->writing_buffer_mutex_);
                 self->writing_buffer_.clear();
             }
-            self->do_write();
-            self->do_read();
+            asio::post(asio::bind_executor(self->ctx_, [self]() {
+                self->do_write();
+                self->do_read();
+            }));
         });
     }
 

test/tools/tool_kv_loader.cxx

@@ -76,15 +76,15 @@ enum class operation {
 
 namespace
 {
-volatile std::sig_atomic_t running{ 1 };
+std::atomic_flag running{ 1 };
 std::size_t operations_limit{ 0 };
 
 } // namespace
 
 void
 sigint_handler(int /* signal */)
 {
-    running = 0;
+    running.test_and_set();
 }
 
 static void
@@ -207,7 +207,7 @@ main()
     std::vector<std::thread> io_pool{};
     io_pool.reserve(number_of_io_threads);
     for (std::size_t i = 0; i < number_of_io_threads; ++i) {
-        io_pool.emplace_back(std::thread([&io]() { io.run(); }));
+        io_pool.emplace_back([&io]() { io.run(); });
     }
 
     test::utils::open_cluster(cluster, origin);
@@ -251,7 +251,7 @@ main()
 
     asio::steady_timer stats_timer(io);
     dump_stats(stats_timer, start_time, total);
-    while (running != 0) {
+    while (running.test_and_set()) {
         auto opcode = dist(gen) <= chance_of_get ? operation::get : operation::upsert;
         if (opcode == operation::get && known_keys.empty()) {
             opcode = operation::upsert;
@@ -284,7 +284,7 @@ main()
                         ++errors[resp.ctx.ec()];
                     }
                     if (operations_limit > 0 && total >= operations_limit) {
-                        running = 0;
+                        running.clear();
                     }
                 });
             } break;
@@ -297,7 +297,7 @@ main()
                         ++errors[resp.ctx.ec()];
                     }
                     if (operations_limit > 0 && total >= operations_limit) {
-                        running = 0;
+                        running.clear();
                     }
                 });
             } break;
@@ -312,7 +312,7 @@ main()
                     ++errors[resp.ctx.ec];
                 }
                 if (operations_limit > 0 && total > operations_limit) {
-                    running = 0;
+                    running.clear();
                 }
             });
         }