Handle incoming events from off-threads separately

On the Wine side we want to handle most events on the main UI thread.
We'll assume any events coming in from a secondary socket are safe and
can be handled directly.

src/common/communication.cpp

@@ -95,6 +95,12 @@ EventHandler::EventHandler(
 void EventHandler::connect() {
     if (acceptor) {
         acceptor->accept(socket);
+
+        // As mentioned in `acceptor's` docstring, this acceptor will be
+        // recreated in `receive()` on another context, and potentially on the
+        // other side of the connection in the case of `vst_host_callback`
+        acceptor.reset();
+        fs::remove(endpoint.path());
     } else {
         socket.connect(endpoint);
     }

src/common/communication.h

@@ -338,19 +338,17 @@ class EventHandler {
      *   this is for sending `dispatch()` events or host callbacks. Optional
      *   since it doesn't have to be done on both sides.
      * @param callback The function used to generate a response out of an event.
+     *   See the definition of `F` for more information.
      *
-     * @tparam F A function type in the form of `EventResponse(Event)`.
+     * @tparam F A function type in the form of `EventResponse(Event, bool)`.
-     *
+     *   The boolean flag is `true` when this event was received on the main
-     *         TODO: Add a boolean flag indicating that this is being called
+     *   socket, and `false` otherwise.
-     *               from an off thread
      *
      * @relates EventHandler::send
      * @relates passthrough_event
      */
     template <typename F>
     void receive(std::optional<std::pair<Logger&, bool>> logging, F callback) {
-        assert(acceptor.has_value());
-
         // As described above we'll handle incoming requests for `socket` on
         // this thread. We'll also listen for incoming connections on `endpoint`
         // on another thread. For any incoming connection we'll spawn a new
@@ -358,6 +356,9 @@ class EventHandler {
         // breaks, the listener and any still active threads will be cleaned up
         // before this function exits.
         boost::asio::io_context secondary_context{};
+        // The previous acceptor has already been shut down by
+        // `EventHandler::connect()`
+        acceptor.emplace(secondary_context, endpoint);
 
         // This works the exact same was as `active_plugins` and
         // `next_plugin_id` in `GroupBridge`
@@ -382,7 +383,7 @@ class EventHandler {
                                              event.value_payload);
                         }
 
-                        EventResult response = callback(event);
+                        EventResult response = callback(event, false);
                         if (logging) {
                             auto [logger, is_dispatch] = *logging;
                             logger.log_event_response(is_dispatch, event.opcode,
@@ -420,7 +421,7 @@ class EventHandler {
                                      event.value_payload);
                 }
 
-                EventResult response = callback(event);
+                EventResult response = callback(event, true);
                 if (logging) {
                     auto [logger, is_dispatch] = *logging;
                     logger.log_event_response(
@@ -499,8 +500,12 @@ class EventHandler {
     /**
      * This acceptor will be used once synchronously on the listening side
      * during `Sockets::connect()`. When `EventHandler::receive()` is then
-     * called, we'll asynchronously listen for new incoming socket connections
+     * called, we'll recreate the acceptor asynchronously listen for new
-     * on `endpoint` using this same acceptor.
+     * incoming socket connections on `endpoint` using this same acceptor. This
+     * is important, because on the case of `vst_host_callback` the acceptor is
+     * first accepts an initial socket on the plugin side (like all sockets),
+     * but all additional incoming connections of course have to be listened for
+     * on the plugin side.
      */
     std::optional<boost::asio::local::stream_protocol::acceptor> acceptor;
 
@@ -650,6 +655,9 @@ boost::filesystem::path generate_endpoint_base(const std::string& plugin_name);
  *
  * This is the receiving analogue of the `*DataCovnerter` objects.
  *
+ * TODO: Now that `EventHandler::receive` replaced `receive_event()`, refactor
+ *       this to just handle the event directly rather than returning a lambda
+ *
  * @param plugin The `AEffect` instance that should be passed to the callback
  *   function.
  * @param callback The function to call with the arguments received from the

src/plugin/plugin-bridge.cpp

@@ -121,10 +121,9 @@ PluginBridge::PluginBridge(audioMasterCallback host_callback)
     // instead of asynchronous IO since communication has to be handled in
     // lockstep anyway
     host_callback_handler = std::jthread([&]() {
-        // TODO: Think of a nicer way to structure this and the similar
-        //       handler in `Vst2Bridge::handle_dispatch_midi_events`
         sockets.vst_host_callback.receive(
-            std::pair<Logger&, bool>(logger, false), [&](Event& event) {
+            std::pair<Logger&, bool>(logger, false),
+            [&](Event& event, bool /*on_main_thread*/) {
                 // MIDI events sent from the plugin back to the host are a
                 // special case here. They have to sent during the
                 // `processReplacing()` function or else the host will ignore

src/wine-host/bridges/vst2.cpp

@@ -146,32 +146,43 @@ bool Vst2Bridge::should_skip_message_loop() const {
 
 void Vst2Bridge::handle_dispatch() {
     sockets.host_vst_dispatch.receive(
-        std::nullopt,
+        std::nullopt, [&](Event& event, bool on_main_thread) {
-        passthrough_event(
+            // TODO: As per the TODO in `passthrough_event`, this can use a
-            plugin,
+            //       round of refactoring now that we never use its returned
-            [&](AEffect* plugin, int opcode, int index, intptr_t value,
+            //       lambda directly anymore
-                void* data, float option) -> intptr_t {
+            return passthrough_event(
-                // Instead of running `plugin->dispatcher()` (or
+                plugin,
-                // `dispatch_wrapper()`) directly, we'll run the function within
+                [&](AEffect* plugin, int opcode, int index, intptr_t value,
-                // the IO context so all events will be executed on the same
+                    void* data, float option) -> intptr_t {
-                // thread as the one that runs the Win32 message loop
+                    // On the main thread, instead of running
-                std::promise<intptr_t> dispatch_result;
+                    // `plugin->dispatcher()` (or `dispatch_wrapper()`)
-                boost::asio::dispatch(io_context, [&]() {
+                    // directly, we'll run the function within the IO context so
-                    const intptr_t result = dispatch_wrapper(
+                    // all events will be executed on the same thread as the one
-                        plugin, opcode, index, value, data, option);
+                    // that runs the Win32 message loop. We'll assume every
+                    // event that comes in while the main thread is already
+                    // handling an event in the IO context can safely be handled
+                    // on an off thread.
+                    if (on_main_thread) {
+                        std::promise<intptr_t> dispatch_result;
+                        boost::asio::dispatch(io_context, [&]() {
+                            const intptr_t result = dispatch_wrapper(
+                                plugin, opcode, index, value, data, option);
 
-                    dispatch_result.set_value(result);
+                            dispatch_result.set_value(result);
-                });
+                        });
 
-                // The message loop and X11 event handling will be run
+                        return dispatch_result.get_future().get();
-                // separately on a timer
+                    } else {
-                return dispatch_result.get_future().get();
+                        return dispatch_wrapper(plugin, opcode, index, value,
-            }));
+                                                data, option);
+                    }
+                })(event);
+        });
 }
 
 void Vst2Bridge::handle_dispatch_midi_events() {
     sockets.host_vst_dispatch_midi_events.receive(
-        std::nullopt, [&](Event& event) {
+        std::nullopt, [&](Event& event, bool /*on_main_thread*/) {
             if (BOOST_LIKELY(event.opcode == effProcessEvents)) {
                 // For 99% of the plugins we can just call
                 // `effProcessReplacing()` and be done with it, but a select few