Move IPC threading decision from receiver to sender side (picking channel implies intended threading)

Author: sgretscher

IPC/ARAIPCAudioUnit_v3.mm

@@ -222,28 +222,13 @@ void sendMessage (MessageID messageID, MessageEncoder * encoder) override
     ~AUProxyPlugIn () override
     {
 #if !__has_feature(objc_arc)
-        dispatch_release (_readAudioQueue);
         [_initAU release];
 #endif
     }
 
-protected:
-    DispatchTarget getDispatchTargetForIncomingTransaction (MessageID messageID) override
-    {
-        // AUMessageChannel cannot be called back from the same thread it receives the message,
-        // so we dispatch audio requests to a dedicated read samples queue and let the inherited
-        // dispatch all other calls to the main thread.
-        if (messageID == ARA_IPC_HOST_METHOD_ID (ARAAudioAccessControllerInterface, readAudioSamples))
-            return _readAudioQueue;
-        return ProxyPlugIn::getDispatchTargetForIncomingTransaction (messageID);
-    }
-
 private:
     AUProxyPlugIn (NSObject<AUMessageChannel> * _Nonnull mainChannel, NSObject<AUMessageChannel> * _Nonnull otherChannel, AUAudioUnit * _Nonnull initAU)
     : ProxyPlugIn { this },
-      // \todo maybe we should make this configurable, so hosts can set this queue if it already has an appropriate one?
-      // \todo there's also QOS_CLASS_USER_INTERACTIVE which seems more appropriate but is undocumented...
-      _readAudioQueue { dispatch_queue_create ("ARA read audio samples", dispatch_queue_attr_make_with_qos_class(DISPATCH_QUEUE_SERIAL, QOS_CLASS_USER_INITIATED, -1)) },
       _initAU { initAU }
     {
         setMainThreadChannel (new ProxyPlugInMessageChannel { mainChannel });
@@ -255,7 +240,6 @@ DispatchTarget getDispatchTargetForIncomingTransaction (MessageID messageID) ove
     }
 
 private:
-    const dispatch_queue_t _readAudioQueue;
     const AUAudioUnit * __strong _initAU;   // workaround for macOS 14: keep the AU that vends the message channels alive, otherwise the channels will eventually stop working
                                             // \todo once this is fixed in macOS, we only need to store this on older macOS versions
 };

IPC/ARAIPCConnection.cpp

@@ -80,22 +80,20 @@
 //
 // Another challenge when injecting IPC into the ARA communication is that for each
 // message that comes in, an appropriate thread has to be selected to process it.
-// For the main thread channel, this is trivial.
+// When a new transaction is initiated, the implementation checks whether this is happening
+// on the main thread or on any other thread, and chooses the appropriate channel accordingly.
+// On the receiving side, calls coming in on the main thread channel are forwarded to the main
+// thread (unless the receive code already runs there), and for the other threads the code is
+// executed directly on the receive thread.
 //
-// For the channel that handles all communication on other threads, the implementation
-// adds a token to each call message that identifies the sending thread. The reply message
-// or any stacked call the receiver might make in response to the message will pass back
-// this thread token, enabling proper routing back to the original sending thread.
+// Replies or callbacks are routed back to the originating thread in the sender. This is done
+// by adding a token when sending a message that identifies the sending thread, and replies and
+// callbacks pass this token back to allow for proper dispatching from the receive thread to the
+// thread that initiated the transaction.
 // Since the actual ARA code is agnostic to IPC being used, the receiving side uses
 // thread local storage to make the sender's thread token available for all stacked calls
 // that the ARA code might make in response to the message.
 //
-// If a message is received that does not contain the token, this message indicates
-// the start of a transaction that was initiated on the other side. For this new
-// transaction, a proper thread has to be selected. This is done based on the message ID -
-// the current implementation dispatches audio reading to a dedicated dispatch queue and
-// handles the remaining calls directly on the IPC receive thread.
-//
 // Note that there is a crucial difference between the dispatch of a new transaction and
 // the dispatch of any follow-up messages in the transaction: the initial message is dispatched
 // to a thread that is potentially executing other code as well in some form of run loop,
@@ -298,15 +296,16 @@ void MessageDispatcher::routeReceivedMessage (MessageID messageID, const Message
     else
     {
         ARA_INTERNAL_ASSERT (messageID != 0);
-        if (const auto dispatchTarget { _connection->getMessageHandler ()->getDispatchTargetForIncomingTransaction (messageID) })
+        if (!_sendLock &&
+            !_connection->wasCreatedOnCurrentThread ())
         {
             ARA_IPC_LOG ("dispatches received message with ID %i (new transaction)", messageID);
 #if defined (_WIN32)
             auto params { new APCProcessReceivedMessageParams { this, messageID, decoder } };
-            const auto result { ::QueueUserAPC (APCRouteNewTransactionFunc, dispatchTarget, reinterpret_cast<ULONG_PTR> (params)) };
+            const auto result { ::QueueUserAPC (APCRouteNewTransactionFunc, _connection->getCreationThreadDispatchTarget (), reinterpret_cast<ULONG_PTR> (params)) };
             ARA_INTERNAL_ASSERT (result != 0);
 #elif defined (__APPLE__)
-            dispatch_async (dispatchTarget,
+            dispatch_async (_connection->getCreationThreadDispatchTarget (),
                 ^{
                     _handleReceivedMessage (messageID, decoder);
                 });
@@ -381,7 +380,7 @@ HANDLE _GetRealCurrentThread ()
 #endif
 
 
-MessageHandler::MessageHandler ()
+Connection::Connection ()
 : _creationThreadID { std::this_thread::get_id () },
 #if defined (_WIN32)
   _creationThreadDispatchTarget { _GetRealCurrentThread () }
@@ -399,16 +398,6 @@ MessageHandler::MessageHandler ()
 #endif
 }
 
-MessageHandler::DispatchTarget MessageHandler::getDispatchTargetForIncomingTransaction (MessageID /*messageID*/)
-{
-    return (std::this_thread::get_id () == _creationThreadID) ? nullptr : _creationThreadDispatchTarget;
-}
-
-
-Connection::Connection ()
-: _creationThreadID { std::this_thread::get_id () }
-{}
-
 Connection::~Connection ()
 {
     delete _otherDispatcher;
@@ -436,7 +425,7 @@ void Connection::setMessageHandler (MessageHandler* messageHandler)
 void Connection::sendMessage (MessageID messageID, MessageEncoder* encoder, ReplyHandler replyHandler, void* replyHandlerUserData)
 {
     ARA_INTERNAL_ASSERT ((_mainDispatcher != nullptr) && (_otherDispatcher != nullptr) && (_messageHandler != nullptr));
-    if (std::this_thread::get_id () == _creationThreadID)
+    if (wasCreatedOnCurrentThread ())
         _mainDispatcher->sendMessage (messageID, encoder, replyHandler, replyHandlerUserData);
     else
         _otherDispatcher->sendMessage (messageID, encoder, replyHandler, replyHandlerUserData);

IPC/ARAIPCConnection.h

@@ -51,35 +51,12 @@ class MessageDispatcher;
 class MessageHandler
 {
 public:
-    MessageHandler ();
     virtual ~MessageHandler () = default;
 
-    //! type returned from getDispatchTargetForIncomingTransaction()
-#if defined (_WIN32)
-    using DispatchTarget = HANDLE;
-    friend void APCProcessReceivedMessageFunc (ULONG_PTR parameter);
-#elif defined (__APPLE__)
-    using DispatchTarget = dispatch_queue_t;
-#else
-    #error "not yet implemented on this platform"
-#endif
-
-    //! IPC connections will call this method to determine which thread should be
-    //! used for handling an incoming transaction. Returning nullptr results in the
-    //! current thread being used, otherwise the call will be forwarded to the
-    //! returned target thread.
-    //! The default implementation dispatches to the thread where the MessageHandler
-    //! was created, which can be utilized by overrides of this method.
-    virtual DispatchTarget getDispatchTargetForIncomingTransaction (MessageID messageID);
-
     //! IPC connections will call this method for incoming messages after
     //! after filtering replies and routing them to the correct thread.
     virtual void handleReceivedMessage (const MessageID messageID, const MessageDecoder* const decoder,
                                         MessageEncoder* const replyEncoder) = 0;
-
-private:
-    std::thread::id const _creationThreadID;
-    DispatchTarget const _creationThreadDispatchTarget;
 };
 
 
@@ -132,15 +109,25 @@ class Connection
 
     MessageHandler* getMessageHandler () { return _messageHandler; }
 
-#if ARA_ENABLE_INTERNAL_ASSERTS
     bool wasCreatedOnCurrentThread () const { return std::this_thread::get_id () == _creationThreadID; }
+
+    //! type returned from getDispatchTargetForIncomingTransaction()
+#if defined (_WIN32)
+    using DispatchTarget = HANDLE;
+    friend void APCProcessReceivedMessageFunc (ULONG_PTR parameter);
+#elif defined (__APPLE__)
+    using DispatchTarget = dispatch_queue_t;
+#else
+    #error "not yet implemented on this platform"
 #endif
+    DispatchTarget getCreationThreadDispatchTarget () { return _creationThreadDispatchTarget; }
 
 private:
     MessageDispatcher* _mainDispatcher {};
     MessageDispatcher* _otherDispatcher {};
     MessageHandler* _messageHandler {};
     std::thread::id const _creationThreadID;
+    DispatchTarget const _creationThreadDispatchTarget;
 };
 //! @}
 

IPC/ARAIPCProxyHost.cpp

@@ -661,16 +661,6 @@ ProxyHost::ProxyHost (Connection* connection)
 : RemoteCaller (connection)
 {}
 
-ProxyHost::DispatchTarget ProxyHost::getDispatchTargetForIncomingTransaction (MessageID messageID)
-{
-    // getPlaybackRegionHeadAndTailTime() is valid on any thread, so we can directly serve it from
-    // the current IPC thread. For all other calls, we call the inherited default implementation to
-    // dispatch to the creation thread.
-    if (messageID == ARA_IPC_PLUGIN_METHOD_ID (ARADocumentControllerInterface, getPlaybackRegionHeadAndTailTime))
-        return nullptr;
-    return MessageHandler::getDispatchTargetForIncomingTransaction (messageID);
-}
-
 void ProxyHost::handleReceivedMessage (const MessageID messageID, const MessageDecoder* const decoder,
                                        MessageEncoder* const replyEncoder)
 {

IPC/ARAIPCProxyHost.h

@@ -45,8 +45,6 @@ class ProxyHost : public MessageHandler, public RemoteCaller
     explicit ProxyHost (Connection* connection);
 
 public:
-    DispatchTarget getDispatchTargetForIncomingTransaction (MessageID messageID) override;
-
     void handleReceivedMessage (const MessageID messageID, const MessageDecoder* const decoder,
                                 MessageEncoder* const replyEncoder) override;
 };

IPC/ARAIPCProxyPlugIn.cpp

@@ -1431,23 +1431,6 @@ void ARAIPCProxyPlugInUninitializeARA (ARAIPCConnectionRef connectionRef, const
 
 /*******************************************************************************/
 
-ProxyPlugIn::DispatchTarget ProxyPlugIn::getDispatchTargetForIncomingTransaction (MessageID messageID)
-{
-    // The archiving controller interface must never be called without the host first starting
-    // the transaction via some (un)archiving call.
-    ARA_INTERNAL_ASSERT (!MethodID::isMessageToHostInterface<ARAArchivingControllerInterface> (messageID));
-
-    // The model update controller interface must never be called without the host first starting
-    // the transaction via notifyModelUpdates().
-    ARA_INTERNAL_ASSERT (!MethodID::isMessageToHostInterface<ARAModelUpdateControllerInterface> (messageID));
-
-    // readAudioSamples() is valid on any thread, so we can directly serve it from the current IPC thread.
-    // For all other calls, we call the inherited default implementation to dispatch to the creation thread.
-    if (messageID == ARA_IPC_HOST_METHOD_ID (ARAAudioAccessControllerInterface, readAudioSamples))
-        return nullptr;
-    return MessageHandler::getDispatchTargetForIncomingTransaction (messageID);
-}
-
 void ProxyPlugIn::handleReceivedMessage (const MessageID messageID, const MessageDecoder* const decoder,
                                          MessageEncoder* const replyEncoder)
 {

IPC/ARAIPCProxyPlugIn.h

@@ -45,8 +45,6 @@ class ProxyPlugIn : public MessageHandler, public RemoteCaller
 public:
     using RemoteCaller::RemoteCaller;
 
-    DispatchTarget getDispatchTargetForIncomingTransaction (MessageID messageID) override;
-
     void handleReceivedMessage (const MessageID messageID, const MessageDecoder* const decoder,
                                 MessageEncoder* const replyEncoder) override;
 };