Merge pull request #85 from ably/3-decide-how-threading-will-work

[ECO-5377] Finalise and document threading approach

Author: lawrence-forooghian

AblyLiveObjects.xcworkspace/xcshareddata/swiftpm/Package.resolved

@@ -70,6 +70,18 @@ The `Public…` classes all follow the same pattern and are not very interesting
 
 `ObjectLifetimesTests.swift` contains tests of the behaviour described in this section.
 
+### Threading
+
+Since this is an extension of ably-cocoa, we follow the same threading approach:
+
+1. The public API can be interacted with from any thread, including synchronous methods such as getters
+2. Callbacks passed to the public API are invoked on the same queue as used by the `ARTRealtime` instance (the `dispatchQueue` client option)
+3. Synchronisation of mutable state is performed using the same internal serial dispatch queue as is used by the `ARTRealtime` instance (the `internalDispatchQueue` client option)
+
+We follow the same naming convention as in ably-cocoa whereby if a method's name contains `nosync` then it must be called on the internal dispatch queue. This allows us to avoid deadlocks that would result from attempting to call `DispatchQueue.sync { … }` when already on the internal queue.
+
+We have an extension on `DispatchQueue`, `ably_syncNoDeadlock(execute:)`, which behaves the same as `sync(execute:)` but with a runtime precondition that we are not already on the queue; favour our extension in order to avoid deadlock.
+
 ### Testing guidelines
 
 #### Attributing tests to a spec point

CONTRIBUTING.md

@@ -20,11 +20,13 @@ let package = Package(
     dependencies: [
         .package(
             url: "https://github.com/ably/ably-cocoa",
-            from: "1.2.44",
+            // TODO: Unpin before next release
+            revision: "8dde3e841aa1f861176c1341cf44e92014b95857",
         ),
         .package(
             url: "https://github.com/ably/ably-cocoa-plugin-support",
-            from: "0.1.0",
+            // TODO: Unpin before next release
+            revision: "2ce1058ed4430cc3563dcead0299e92a81d2774b",
         ),
         .package(
             url: "https://github.com/apple/swift-argument-parser",

Package.resolved

@@ -14,7 +14,7 @@ internal protocol CoreSDK: AnyObject, Sendable {
     func testsOnly_overridePublish(with newImplementation: @escaping ([OutboundObjectMessage]) async throws(InternalError) -> Void)
 
     /// Returns the current state of the Realtime channel that this wraps.
-    var channelState: _AblyPluginSupportPrivate.RealtimeChannelState { get }
+    var nosync_channelState: _AblyPluginSupportPrivate.RealtimeChannelState { get }
 }
 
 internal final class DefaultCoreSDK: CoreSDK {
@@ -81,8 +81,8 @@ internal final class DefaultCoreSDK: CoreSDK {
         }
     }
 
-    internal var channelState: _AblyPluginSupportPrivate.RealtimeChannelState {
-        pluginAPI.state(for: channel)
+    internal var nosync_channelState: _AblyPluginSupportPrivate.RealtimeChannelState {
+        pluginAPI.nosync_state(for: channel)
     }
 }
 
@@ -96,11 +96,11 @@ internal extension CoreSDK {
     ///   - invalidStates: Array of channel states that are considered invalid for the operation
     ///   - operationDescription: A description of the operation being performed, used in error messages
     /// - Throws: `ARTErrorInfo` with code 90001 and statusCode 400 if the channel is in any of the invalid states
-    func validateChannelState(
+    func nosync_validateChannelState(
         notIn invalidStates: [_AblyPluginSupportPrivate.RealtimeChannelState],
         operationDescription: String,
     ) throws(ARTErrorInfo) {
-        let currentChannelState = channelState
+        let currentChannelState = nosync_channelState
         if invalidStates.contains(currentChannelState) {
             throw LiveObjectsError.objectsOperationFailedInvalidChannelState(
                 operationDescription: operationDescription,

Package.swift

@@ -21,8 +21,8 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.
     /// Retrieves the `RealtimeObjects` for this channel.
     ///
     /// We expect this value to have been previously set by ``prepare(_:)``.
-    internal static func realtimeObjects(for channel: _AblyPluginSupportPrivate.RealtimeChannel, pluginAPI: _AblyPluginSupportPrivate.PluginAPIProtocol) -> InternalDefaultRealtimeObjects {
-        guard let pluginData = pluginAPI.pluginDataValue(forKey: pluginDataKey, channel: channel) else {
+    internal static func nosync_realtimeObjects(for channel: _AblyPluginSupportPrivate.RealtimeChannel, pluginAPI: _AblyPluginSupportPrivate.PluginAPIProtocol) -> InternalDefaultRealtimeObjects {
+        guard let pluginData = pluginAPI.nosync_pluginDataValue(forKey: pluginDataKey, channel: channel) else {
             // InternalPlugin.prepare was not called
             fatalError("To access LiveObjects functionality, you must pass the LiveObjects plugin in the client options when creating the ARTRealtime instance: `clientOptions.plugins = [.liveObjects: AblyLiveObjects.Plugin.self]`")
         }
@@ -34,25 +34,27 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.
     // MARK: - LiveObjectsInternalPluginProtocol
 
     // Populates the channel's `objects` property.
-    internal func prepare(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, client: _AblyPluginSupportPrivate.RealtimeClient) {
+    internal func nosync_prepare(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, client: _AblyPluginSupportPrivate.RealtimeClient) {
         let pluginLogger = pluginAPI.logger(for: channel)
+        let internalQueue = pluginAPI.internalQueue(for: client)
         let callbackQueue = pluginAPI.callbackQueue(for: client)
         let options = ARTClientOptions.castPluginPublicClientOptions(pluginAPI.options(for: client))
 
         let logger = DefaultLogger(pluginLogger: pluginLogger, pluginAPI: pluginAPI)
         logger.log("LiveObjects.DefaultInternalPlugin received prepare(_:)", level: .debug)
         let liveObjects = InternalDefaultRealtimeObjects(
             logger: logger,
+            internalQueue: internalQueue,
             userCallbackQueue: callbackQueue,
             clock: DefaultSimpleClock(),
             garbageCollectionOptions: options.garbageCollectionOptions ?? .init(),
         )
-        pluginAPI.setPluginDataValue(liveObjects, forKey: Self.pluginDataKey, channel: channel)
+        pluginAPI.nosync_setPluginDataValue(liveObjects, forKey: Self.pluginDataKey, channel: channel)
     }
 
     /// Retrieves the internally-typed `objects` property for the channel.
-    private func realtimeObjects(for channel: _AblyPluginSupportPrivate.RealtimeChannel) -> InternalDefaultRealtimeObjects {
-        Self.realtimeObjects(for: channel, pluginAPI: pluginAPI)
+    private func nosync_realtimeObjects(for channel: _AblyPluginSupportPrivate.RealtimeChannel) -> InternalDefaultRealtimeObjects {
+        Self.nosync_realtimeObjects(for: channel, pluginAPI: pluginAPI)
     }
 
     /// A class that wraps an object message.
@@ -102,30 +104,30 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.
         return wireObjectMessage.toWireObject.toPluginSupportDataDictionary
     }
 
-    internal func onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool) {
-        realtimeObjects(for: channel).onChannelAttached(hasObjects: hasObjects)
+    internal func nosync_onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool) {
+        nosync_realtimeObjects(for: channel).nosync_onChannelAttached(hasObjects: hasObjects)
     }
 
-    internal func handleObjectProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], channel: _AblyPluginSupportPrivate.RealtimeChannel) {
+    internal func nosync_handleObjectProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], channel: _AblyPluginSupportPrivate.RealtimeChannel) {
         guard let inboundObjectMessageBoxes = publicObjectMessages as? [ObjectMessageBox<InboundObjectMessage>] else {
             preconditionFailure("Expected to receive the same InboundObjectMessage type as we emit")
         }
 
         let objectMessages = inboundObjectMessageBoxes.map(\.objectMessage)
 
-        realtimeObjects(for: channel).handleObjectProtocolMessage(
+        nosync_realtimeObjects(for: channel).nosync_handleObjectProtocolMessage(
             objectMessages: objectMessages,
         )
     }
 
-    internal func handleObjectSyncProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], protocolMessageChannelSerial: String?, channel: _AblyPluginSupportPrivate.RealtimeChannel) {
+    internal func nosync_handleObjectSyncProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], protocolMessageChannelSerial: String?, channel: _AblyPluginSupportPrivate.RealtimeChannel) {
         guard let inboundObjectMessageBoxes = publicObjectMessages as? [ObjectMessageBox<InboundObjectMessage>] else {
             preconditionFailure("Expected to receive the same InboundObjectMessage type as we emit")
         }
 
         let objectMessages = inboundObjectMessageBoxes.map(\.objectMessage)
 
-        realtimeObjects(for: channel).handleObjectSyncProtocolMessage(
+        nosync_realtimeObjects(for: channel).nosync_handleObjectSyncProtocolMessage(
             objectMessages: objectMessages,
             protocolMessageChannelSerial: protocolMessageChannelSerial,
         )
@@ -145,10 +147,13 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.
             let internalQueue = pluginAPI.internalQueue(for: client)
 
             internalQueue.async {
-                pluginAPI.sendObject(
+                pluginAPI.nosync_sendObject(
                     withObjectMessages: objectMessageBoxes,
                     channel: channel,
                 ) { error in
+                    // We don't currently rely on this documented behaviour of `nosync_sendObject` but we may do later, so assert it to be sure it's happening.
+                    dispatchPrecondition(condition: .onQueue(internalQueue))
+
                     if let error {
                         continuation.resume(returning: .failure(error.toInternalError()))
                     } else {