Re-applied "Rewrote AEMainThreadEndpoint to use dispatch semaphore, r…

…ather than polling" We have the final word on the realtime-thread safety of Mach semaphores, and they *are* fine. The internal implementation has diverged from the open-source one, and includes protections against priority inversion.
cjhanson · Jun 14, 2016 · 8cfec37 · 8cfec37
1 parent d77c2fb
commit 8cfec37
Show file tree

Hide file tree

Showing 5 changed files with 50 additions and 40 deletions.
diff --git a/Tests/AECrossThreadMessagingTests.m b/Tests/AECrossThreadMessagingTests.m
@@ -32,7 +32,7 @@ - (void)testMainThreadEndpointMessaging {
     }];
 
     AEMainThreadEndpointSend(endpoint, NULL, 0);
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:endpoint.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertEqualObjects(messages, (@[[NSData dataWithBytes:NULL length:0]]));
     [messages removeAllObjects];
@@ -43,13 +43,13 @@ - (void)testMainThreadEndpointMessaging {
     AEMainThreadEndpointSend(endpoint, &value1, sizeof(value1));
     AEMainThreadEndpointSend(endpoint, &value2, sizeof(value2));
     AEMainThreadEndpointSend(endpoint, &value3, sizeof(value3));
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:endpoint.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertEqualObjects(messages, (@[[NSData dataWithBytes:&value1 length:sizeof(value1)], [NSData dataWithBytes:&value2 length:sizeof(value2)], [NSData dataWithBytes:&value3 length:sizeof(value3)]]));
     [messages removeAllObjects];
 
     AEMainThreadEndpointSend(endpoint, &value1, sizeof(value1));
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:endpoint.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertEqualObjects(messages, (@[[NSData dataWithBytes:&value1 length:sizeof(value1)]]));
     [messages removeAllObjects];
@@ -130,7 +130,7 @@ - (void)testMessageQueueAudioThreadMessaging {
     XCTAssertNotNil(weakObject);
 
     AEMessageQueuePoll(queue);
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:queue.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertTrue(hitBlock);
     XCTAssertTrue(hitCompletionBlock);
@@ -147,7 +147,7 @@ - (void)testMessageQueueMainThreadMessaging {
     AEMessageQueue * queue = [AEMessageQueue new];
 
     [self sendMainThreadMessage:queue];
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:queue.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertEqual(self.mainThreadMessageValue1, 1);
     XCTAssertEqual(self.mainThreadMessageValue2, self);
@@ -159,7 +159,7 @@ - (void)testMessageQueueMainThreadMessaging {
 
     AEMessageQueuePerformSelectorOnMainThread(queue, self, @selector(mainThreadMessageTestWithNoArguments), AEArgumentNone);
 
-    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:queue.pollInterval]];
+    [[NSRunLoop currentRunLoop] runUntilDate:[NSDate dateWithTimeIntervalSinceNow:0.01]];
 
     XCTAssertEqual(self.mainThreadMessageValue1, 3);
 }

diff --git a/TheAmazingAudioEngine/Utilities/AEMainThreadEndpoint.h b/TheAmazingAudioEngine/Utilities/AEMainThreadEndpoint.h
@@ -107,10 +107,6 @@ void * _Nullable AEMainThreadEndpointCreateMessage(__unsafe_unretained AEMainThr
  */
 void AEMainThreadEndpointDispatchMessage(__unsafe_unretained AEMainThreadEndpoint * _Nonnull endpoint);
 
-
-//! The poll interval (default is 10ms)
-@property (nonatomic) AESeconds pollInterval;
-
 @end
 
 #ifdef __cplusplus

diff --git a/TheAmazingAudioEngine/Utilities/AEMainThreadEndpoint.m b/TheAmazingAudioEngine/Utilities/AEMainThreadEndpoint.m
@@ -26,13 +26,21 @@
 
 #import "AEMainThreadEndpoint.h"
 #import "TPCircularBuffer.h"
-#import "AEWeakRetainingProxy.h"
+#import <dispatch/semaphore.h>
+
+@class AEMainThreadEndpointThread;
 
 @interface AEMainThreadEndpoint () {
     TPCircularBuffer _buffer;
+    BOOL _hasPendingMainThreadMessages;
 }
 @property (nonatomic, copy) AEMainThreadEndpointHandler handler;
-@property (nonatomic, strong) NSTimer * timer;
+@property (nonatomic) dispatch_semaphore_t semaphore;
+@property (nonatomic, strong) AEMainThreadEndpointThread * thread;
+@end
+
+@interface AEMainThreadEndpointThread : NSThread
+@property (nonatomic, unsafe_unretained) AEMainThreadEndpoint * endpoint;
 @end
 
 @implementation AEMainThreadEndpoint
@@ -44,31 +52,29 @@ - (instancetype)initWithHandler:(AEMainThreadEndpointHandler)handler {
 - (instancetype)initWithHandler:(AEMainThreadEndpointHandler)handler bufferCapacity:(size_t)bufferCapacity {
     if ( !(self = [super init]) ) return nil;
 
-    _pollInterval = 0.01;
     self.handler = handler;
 
     if ( !TPCircularBufferInit(&_buffer, (int32_t)bufferCapacity) ) {
         return nil;
     }
 
-    [self startTimer];
+    self.semaphore = dispatch_semaphore_create(0);
+
+    self.thread = [AEMainThreadEndpointThread new];
+    self.thread.endpoint = self;
+    [self.thread start];
 
     return self;
 }
 
 - (void)dealloc {
-    [self.timer invalidate];
+    @synchronized ( self.thread ) {
+        [self.thread cancel];
+        dispatch_semaphore_signal(_semaphore);
+    }
     TPCircularBufferCleanup(&_buffer);
 }
 
-- (void)setPollInterval:(AESeconds)pollInterval {
-    _pollInterval = pollInterval;
-
-    // Restart timer
-    [self.timer invalidate];
-    [self startTimer];
-}
-
 BOOL AEMainThreadEndpointSend(__unsafe_unretained AEMainThreadEndpoint * THIS, const void * data, size_t length) {
 
     // Prepare message
@@ -114,14 +120,10 @@ void AEMainThreadEndpointDispatchMessage(__unsafe_unretained AEMainThreadEndpoin
 
     // Mark as ready to read
     TPCircularBufferProduce(&THIS->_buffer, (int32_t)size);
+    dispatch_semaphore_signal(THIS->_semaphore);
 }
 
-- (void)startTimer {
-    self.timer = [NSTimer scheduledTimerWithTimeInterval:self.pollInterval target:[AEWeakRetainingProxy proxyWithTarget:self]
-                                                selector:@selector(poll) userInfo:nil repeats:YES];
-}
-
-- (void)poll {
+- (void)serviceMessages {
     while ( 1 ) {
         // Get pointer to readable bytes
         int32_t availableBytes;
@@ -133,11 +135,33 @@ - (void)poll {
         void * data = length > 0 ? (tail + sizeof(size_t)) : NULL;
 
         // Run handler
-        self.handler(data, length);
+        dispatch_sync(dispatch_get_main_queue(), ^{
+            self.handler(data, length);
+        });
 
         // Mark as read
         TPCircularBufferConsume(&_buffer, (int32_t)(sizeof(size_t) + length));
     }
 }
 
 @end
+
+@implementation AEMainThreadEndpointThread
+
+- (void)main {
+    dispatch_semaphore_t semaphore = self.endpoint.semaphore;
+
+    while ( 1 ) {
+        @synchronized ( self ) {
+            if ( self.cancelled ) {
+                break;
+            }
+            @autoreleasepool {
+                [self.endpoint serviceMessages];
+            }
+        }
+        dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);
+    }
+}
+
+@end
diff --git a/TheAmazingAudioEngine/Utilities/AEMessageQueue.h b/TheAmazingAudioEngine/Utilities/AEMessageQueue.h
@@ -193,9 +193,6 @@ BOOL AEMessageQueuePerformSelectorOnMainThread(__unsafe_unretained AEMessageQueu
  */
 void AEMessageQueuePoll(__unsafe_unretained AEMessageQueue * _Nonnull THIS);
 
-//! The poll interval (default is 10ms)
-@property (nonatomic) AESeconds pollInterval;
-
 @end
 
 #ifdef __cplusplus

diff --git a/TheAmazingAudioEngine/Utilities/AEMessageQueue.m b/TheAmazingAudioEngine/Utilities/AEMessageQueue.m
@@ -133,8 +133,6 @@ - (instancetype)initWithBufferCapacity:(size_t)bufferCapacity {
         AEMainThreadEndpointSend(mainThread, data, length);
     } bufferCapacity:bufferCapacity];
 
-    _pollInterval = self.mainThreadEndpoint.pollInterval;
-
     return self;
 }
 
@@ -233,9 +231,4 @@ void AEMessageQueuePoll(__unsafe_unretained AEMessageQueue * _Nonnull THIS) {
     AEAudioThreadEndpointPoll(THIS->_audioThreadEndpoint);
 }
 
-- (void)setPollInterval:(AESeconds)pollInterval {
-    _pollInterval = pollInterval;
-    self.mainThreadEndpoint.pollInterval = pollInterval;
-}
-
 @end