Made IPC MessageDispatcher and its subclasses private details of Connection implementation

Author: sgretscher

IPC/ARAIPCAudioUnit_v3.mm

@@ -82,12 +82,12 @@
 #endif
 
 public:
-    void routeReceivedMessage (NSDictionary * _Nonnull message)
+    void routeReceivedDictionary (NSDictionary * _Nonnull message)
     {
         ARA_INTERNAL_ASSERT (![NSThread isMainThread]);
         const MessageID messageID { [(NSNumber *) [message objectForKey:_messageIDKey] intValue] };
         auto decoder { std::make_unique<CFMessageDecoder> ((__bridge CFDictionaryRef) message) };
-        getMessageDispatcher ()->routeReceivedMessage (messageID, std::move (decoder));
+        routeReceivedMessage (messageID, std::move (decoder));
     }
 
     void sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encoder) override
@@ -150,7 +150,7 @@ void sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encode
         _audioUnitChannel.callHostBlock =
             ^NSDictionary * _Nullable (NSDictionary * _Nonnull message)
             {
-                routeReceivedMessage (message);
+                routeReceivedDictionary (message);
                 return [NSDictionary dictionary];   // \todo it would yield better performance if the callHostBlock would allow nil as return value
             };
 
@@ -265,7 +265,7 @@ ARAIPCProxyPlugInRef ARA_CALL ARAIPCAUProxyPlugInInitialize (AUAudioUnit * _Nonn
 
 void ARA_CALL ARAIPCAUProxyPlugInPerformPendingMainThreadTasks (ARAIPCProxyPlugInRef _Nonnull proxyPlugInRef)
 {
-    fromIPCRef (proxyPlugInRef)->getConnection ()->getMainThreadDispatcher ()->processPendingMessageIfNeeded ();
+    fromIPCRef (proxyPlugInRef)->getConnection ()->processPendingMessageOnCreationThreadIfNeeded ();
 }
 
 const ARAPlugInExtensionInstance * _Nonnull ARA_CALL ARAIPCAUProxyPlugInBindToDocumentController (AUAudioUnit * _Nonnull audioUnit,
@@ -346,7 +346,7 @@ ARAIPCMessageChannelRef _Nullable ARA_CALL ARAIPCAUProxyHostInitializeMessageCha
 NSDictionary * _Nonnull ARA_CALL ARAIPCAUProxyHostCommandHandler (ARAIPCMessageChannelRef _Nonnull messageChannelRef, NSDictionary * _Nonnull message)
 {
     if ([message count])    // \todo filter dummy message sent in ProxyPlugInMessageChannel(), see there
-        static_cast<ProxyHostMessageChannel *> (fromIPCRef (messageChannelRef))->routeReceivedMessage (message);
+        static_cast<ProxyHostMessageChannel *> (fromIPCRef (messageChannelRef))->routeReceivedDictionary (message);
     return [NSDictionary dictionary];   // \todo it would yield better performance if -callAudioUnit: would allow nil as return value
 }
 

IPC/ARAIPCConnection.cpp

@@ -136,6 +136,50 @@ namespace IPC {
 
 static bool _debugAsHost {};
 
+//------------------------------------------------------------------------------
+
+class MessageDispatcher
+{
+public:     // needs to be public for thread-local variables (which cannot be class members)
+#if defined (_WIN32)
+    using ThreadRef = int32_t;
+#elif defined (__APPLE__)
+    using ThreadRef = size_t;
+#else
+    #error "not yet implemented on this platform"
+#endif
+    static constexpr ThreadRef _invalidThread { 0 };
+
+public:
+    explicit MessageDispatcher (Connection* connection, std::unique_ptr<MessageChannel> && messageChannel);
+    virtual ~MessageDispatcher () = default;
+
+// protected: this does not work with thread_local...
+    struct PendingReplyHandler
+    {
+        ReplyHandler* const _replyHandler;
+        const PendingReplyHandler* _prevPendingReplyHandler;
+    };
+
+    static bool isReply (MessageID messageID) { return messageID == 0; }
+
+    static ThreadRef getCurrentThread ();
+
+protected:
+    Connection* getConnection () const { return  _connection; }
+    MessageChannel* getMessageChannel () const { return  _messageChannel.get (); }
+
+    void _sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encoder, bool isNewTransaction);
+    bool _waitForMessage ();
+    void _handleReply (std::unique_ptr<const MessageDecoder> && decoder, ReplyHandler && replyHandler);
+    std::unique_ptr<MessageEncoder> _handleReceivedMessage (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder);
+
+private:
+    Connection* const _connection;
+    const std::unique_ptr<MessageChannel> _messageChannel;
+};
+
+
 MessageDispatcher::MessageDispatcher (Connection* connection, std::unique_ptr<MessageChannel> && messageChannel)
 : _connection { connection },
   _messageChannel { std::move (messageChannel) }
@@ -144,8 +188,6 @@ MessageDispatcher::MessageDispatcher (Connection* connection, std::unique_ptr<Me
 // unfortunately at least in clang std::thread::id's c'tor isn't constexpr
 //  static_assert (std::thread::id {} == *reinterpret_cast<const std::thread::id*>(&_invalidThread), "the current implementation relies on invalid thread IDs being 0");
     ARA_INTERNAL_ASSERT (std::thread::id {} == *reinterpret_cast<const std::thread::id*>(&_invalidThread));
-    
-    _messageChannel->setMessageDispatcher (this);
 }
 
 MessageDispatcher::ThreadRef MessageDispatcher::getCurrentThread ()
@@ -189,6 +231,44 @@ std::unique_ptr<MessageEncoder> MessageDispatcher::_handleReceivedMessage (Messa
     return replyEncoder;
 }
 
+//------------------------------------------------------------------------------
+
+// single-threaded variant for main thread communication only
+class MainThreadMessageDispatcher : public MessageDispatcher
+{
+public:
+    MainThreadMessageDispatcher (Connection* connection, std::unique_ptr<MessageChannel> && messageChannel)
+    : MessageDispatcher { connection, std::move (messageChannel) }
+    {
+        getMessageChannel ()->_receivedMessageRouter = [this] (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder)
+                                                              { routeReceivedMessage (messageID, std::move (decoder)); };
+    }
+
+    void sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encoder, ReplyHandler && replyHandler);
+
+    void routeReceivedMessage (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder);
+
+    void processPendingMessageIfNeeded ();
+
+private:
+    // key to indicate whether an outgoing call is made in response to a currently handled incoming call
+    // or a new call, which is necessary to deal with the decoupled main threads concurrency
+    // to optimize for performance, it is only added when the call is a response, being a new call is implicit
+    // (it is also never added to replies because they always are a response anyways)
+    static constexpr MessageArgumentKey kIsResponseKey { -1 };
+
+private:
+    int32_t _processingMessagesCount { 0 };
+
+// \todo this is not allowed for some reason, so we must cast at every use of _noPendingMessageDecoder...
+//  static constexpr auto _noPendingMessageDecoder { reinterpret_cast<const MessageDecoder*> (static_cast<intptr_t> (-1)) };
+    static constexpr auto _noPendingMessageDecoder { static_cast<intptr_t> (-1) };
+    MessageID _pendingMessageID { 0 };  // read/write _pendingMessageDecoder with proper barrier before/after reading/writing this
+    std::atomic<const MessageDecoder*> _pendingMessageDecoder { reinterpret_cast<const MessageDecoder*> (_noPendingMessageDecoder) };
+
+    const PendingReplyHandler* _pendingReplyHandler { nullptr };
+};
+
 
 void MainThreadMessageDispatcher::sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encoder,
                                                ReplyHandler && replyHandler)
@@ -281,6 +361,46 @@ void MainThreadMessageDispatcher::routeReceivedMessage (MessageID messageID, std
     }
 }
 
+//------------------------------------------------------------------------------
+
+// multi-threaded variant for all non-main thread communication
+class OtherThreadsMessageDispatcher : public MessageDispatcher
+{
+public:
+    OtherThreadsMessageDispatcher (Connection* connection, std::unique_ptr<MessageChannel> && messageChannel)
+    : MessageDispatcher { connection, std::move (messageChannel) }
+    {
+        getMessageChannel ()->_receivedMessageRouter = [this] (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder)
+                                                              { routeReceivedMessage (messageID, std::move (decoder)); };
+    }
+
+    void sendMessage (MessageID messageID, std::unique_ptr<MessageEncoder> && encoder, ReplyHandler && replyHandler);
+
+    void routeReceivedMessage (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder);
+
+private:
+    // keys to store the threading information in the IPC messages
+    static constexpr MessageArgumentKey kSendThreadKey { -1 };
+    static constexpr MessageArgumentKey kReceiveThreadKey { -2 };
+
+    struct RoutedMessage
+    {
+        MessageID _messageID { 0 };
+        std::unique_ptr<const MessageDecoder> _decoder;
+        ThreadRef _targetThread { _invalidThread };
+    };
+    RoutedMessage* _getRoutedMessageForThread (ThreadRef thread);
+
+    void _processReceivedMessage (MessageID messageID, std::unique_ptr<const MessageDecoder> && decoder);
+
+private:
+    // incoming data is stored in _routedMessages by the receive handler for the
+    // sending threads waiting to pick it up (signalled via _routeReceiveCondition)
+    std::condition_variable _routeReceiveCondition;
+    std::vector<RoutedMessage> _routedMessages { 12 };  // we shouldn't use more than a handful of threads concurrently for the IPC
+    std::mutex _sendLock;
+    std::mutex _routeLock;
+};
 
 // actually "static" members of OtherThreadsMessageDispatcher, but for some reason C++ doesn't allow this...
 thread_local OtherThreadsMessageDispatcher::ThreadRef _remoteTargetThread { 0 };
@@ -416,6 +536,7 @@ void OtherThreadsMessageDispatcher::_processReceivedMessage (MessageID messageID
     _remoteTargetThread = previousRemoteTargetThread;
 }
 
+//------------------------------------------------------------------------------
 
 #if defined (_WIN32)
 
@@ -572,6 +693,11 @@ bool Connection::waitForMessageOnCreationThread ()
     return false;
 }
 
+void Connection::processPendingMessageOnCreationThreadIfNeeded ()
+{
+    _mainThreadDispatcher->processPendingMessageIfNeeded ();
+}
+
 void Connection::signalMesssageReceived ()
 {
 #if __cplusplus >= 202002L