Optimize ASCATransactionQueue (#1350)

* Optimize ASCATransactionQueue. This queue is very busy, and it runs on the main thread so it's important for it to be fast.

Avoid waking up the run loop for every single node.
Avoid a ton of NSPointerArray overhead that we don't need.
Avoid retain/release traffic on the singleton by using an inline function. I confirmed that in release mode, the static __strong is correctly inlined and no ARC traffic is incurred.

* Comment

* Unlock right

* Remove magic number

Source/ASDisplayNode.mm

@@ -2840,7 +2840,7 @@ - (void)willEnterHierarchy
 
   if (![self supportsRangeManagedInterfaceState]) {
     self.interfaceState = ASInterfaceStateInHierarchy;
-  } else if (ASCATransactionQueue.sharedQueue.isEnabled) {
+  } else if (ASCATransactionQueueGet().enabled) {
     __instanceLock__.lock();
     ASInterfaceState state = _preExitingInterfaceState;
     _preExitingInterfaceState = ASInterfaceStateNone;
@@ -2900,7 +2900,7 @@ - (void)didExitHierarchy
       }
     };
 
-    if (!ASCATransactionQueue.sharedQueue.enabled) {
+    if (!ASCATransactionQueueGet().enabled) {
       dispatch_async(dispatch_get_main_queue(), exitVisibleInterfaceState);
     } else {
       exitVisibleInterfaceState();
@@ -2965,13 +2965,13 @@ - (ASInterfaceState)interfaceState
 
 - (void)setInterfaceState:(ASInterfaceState)newState
 {
-  if (!ASCATransactionQueue.sharedQueue.enabled) {
+  if (!ASCATransactionQueueGet().enabled) {
     [self applyPendingInterfaceState:newState];
   } else {
     ASDN::MutexLocker l(__instanceLock__);
     if (_pendingInterfaceState != newState) {
       _pendingInterfaceState = newState;
-      [[ASCATransactionQueue sharedQueue] enqueue:self];
+      [ASCATransactionQueueGet() enqueue:self];
     }
   }
 }
@@ -2997,7 +2997,7 @@ - (void)applyPendingInterfaceState:(ASInterfaceState)newPendingState
     ASDN::MutexLocker l(__instanceLock__);
     // newPendingState will not be used when ASCATransactionQueue is enabled
     // and use _pendingInterfaceState instead for interfaceState update.
-    if (!ASCATransactionQueue.sharedQueue.enabled) {
+    if (!ASCATransactionQueueGet().enabled) {
       _pendingInterfaceState = newPendingState;
     }
     oldState = _interfaceState;

Source/ASRunLoopQueue.h

@@ -48,12 +48,7 @@ AS_SUBCLASSING_RESTRICTED
 
 @end
 
-AS_SUBCLASSING_RESTRICTED
-@interface ASCATransactionQueue : ASAbstractRunLoopQueue
-
-@property (readonly) BOOL isEmpty;
 
-@property (readonly, getter=isEnabled) BOOL enabled;
 
 /**
  * The queue to run on main run loop before CATransaction commit.
@@ -62,13 +57,26 @@ AS_SUBCLASSING_RESTRICTED
  * to get last chance of updating/coalesce info like interface state.
  * Each node will only be called once per transaction commit to reflect interface change.
  */
-@property (class, readonly) ASCATransactionQueue *sharedQueue;
-+ (ASCATransactionQueue *)sharedQueue NS_RETURNS_RETAINED;
+AS_SUBCLASSING_RESTRICTED
+@interface ASCATransactionQueue : ASAbstractRunLoopQueue
+
+@property (readonly) BOOL isEmpty;
+
+@property (readonly, getter=isEnabled) BOOL enabled;
 
 - (void)enqueue:(id<ASCATransactionQueueObserving>)object;
 
 @end
 
+NS_INLINE ASCATransactionQueue *ASCATransactionQueueGet(void) {
+  static ASCATransactionQueue *q;
+  static dispatch_once_t onceToken;
+  dispatch_once(&onceToken, ^{
+    q = [[ASCATransactionQueue alloc] init];
+  });
+  return q;
+}
+
 @interface ASDeallocQueue : NSObject
 
 @property (class, readonly) ASDeallocQueue *sharedDeallocationQueue;

Source/ASRunLoopQueue.mm

@@ -301,9 +301,10 @@ - (void)enqueue:(id)object
 
   if (!foundObject) {
     [_internalQueue addPointer:(__bridge void *)object];
-
-    CFRunLoopSourceSignal(_runLoopSource);
-    CFRunLoopWakeUp(_runLoop);
+    if (_internalQueue.count == 1) {
+      CFRunLoopSourceSignal(_runLoopSource);
+      CFRunLoopWakeUp(_runLoop);
+    }
   }
 }
 
@@ -320,11 +321,20 @@ - (BOOL)isEmpty
 #pragma mark - ASCATransactionQueue
 
 @interface ASCATransactionQueue () {
-  CFRunLoopRef _runLoop;
   CFRunLoopSourceRef _runLoopSource;
   CFRunLoopObserverRef _preTransactionObserver;
-  NSPointerArray *_internalQueue;
-  ASDN::RecursiveMutex _internalQueueLock;
+
+  // Current buffer for new entries, only accessed from within its mutex.
+  std::vector<id<ASCATransactionQueueObserving>> _internalQueue;
+
+  // No retain, no release, pointer hash, pointer equality.
+  // Enforce uniqueness in our queue. std::unordered_set does a heap allocation for each entry – not good.
+  CFMutableSetRef _internalQueueHashSet;
+
+  // Temporary buffer, only accessed from the main thread in -process.
+  std::vector<id<ASCATransactionQueueObserving>> _batchBuffer;
+
+  ASDN::Mutex _internalQueueLock;
 
   // In order to not pollute the top-level activities, each queue has 1 root activity.
   os_activity_t _rootActivity;
@@ -342,22 +352,16 @@ @implementation ASCATransactionQueue
 // but after most other scheduled work on the runloop has processed.
 static int const kASASCATransactionQueueOrder = 1000000;
 
-+ (ASCATransactionQueue *)sharedQueue NS_RETURNS_RETAINED
-{
-  static dispatch_once_t onceToken;
-  static ASCATransactionQueue *sharedQueue;
-  dispatch_once(&onceToken, ^{
-    sharedQueue = [[ASCATransactionQueue alloc] init];
-  });
-  return sharedQueue;
-}
-
 - (instancetype)init
 {
   if (self = [super init]) {
-    _runLoop = CFRunLoopGetMain();
-    NSPointerFunctionsOptions options = NSPointerFunctionsStrongMemory;
-    _internalQueue = [[NSPointerArray alloc] initWithOptions:options];
+    _internalQueueHashSet = CFSetCreateMutable(NULL, 0, NULL);
+
+    // This is going to be a very busy queue – every node in the preload range will enter this queue.
+    // Save some time on first render by reserving space up front.
+    static constexpr int kInternalQueueInitialCapacity = 64;
+    _internalQueue.reserve(kInternalQueueInitialCapacity);
+    _batchBuffer.reserve(kInternalQueueInitialCapacity);
 
     // We don't want to pollute the top-level app activities with run loop batches, so we create one top-level
     // activity per queue, and each batch activity joins that one instead.
@@ -371,14 +375,13 @@ - (instancetype)init
     // Self is guaranteed to outlive the observer.  Without the high cost of a weak pointer,
     // __unsafe_unretained allows us to avoid flagging the memory cycle detector.
     __unsafe_unretained __typeof__(self) weakSelf = self;
-    void (^handlerBlock) (CFRunLoopObserverRef observer, CFRunLoopActivity activity) = ^(CFRunLoopObserverRef observer, CFRunLoopActivity activity) {
-      while (weakSelf->_internalQueue.count > 0) {
-      [weakSelf processQueue];
+    _preTransactionObserver = CFRunLoopObserverCreateWithHandler(NULL, kCFRunLoopBeforeWaiting, true, kASASCATransactionQueueOrder, ^(CFRunLoopObserverRef observer, CFRunLoopActivity activity) {
+      while (!weakSelf->_internalQueue.empty()) {
+        [weakSelf processQueue];
       }
-    };
-    _preTransactionObserver = CFRunLoopObserverCreateWithHandler(NULL, kCFRunLoopBeforeWaiting, true, kASASCATransactionQueueOrder, handlerBlock);
+    });
 
-    CFRunLoopAddObserver(_runLoop, _preTransactionObserver,  kCFRunLoopCommonModes);
+    CFRunLoopAddObserver(CFRunLoopGetMain(), _preTransactionObserver, kCFRunLoopCommonModes);
 
     // It is not guaranteed that the runloop will turn if it has no scheduled work, and this causes processing of
     // the queue to stop. Attaching a custom loop source to the run loop and signal it if new work needs to be done
@@ -388,7 +391,7 @@ - (instancetype)init
     sourceContext.info = (__bridge void *)self;
 #endif
     _runLoopSource = CFRunLoopSourceCreate(NULL, 0, &sourceContext);
-    CFRunLoopAddSource(_runLoop, _runLoopSource, kCFRunLoopCommonModes);
+    CFRunLoopAddSource(CFRunLoopGetMain(), _runLoopSource, kCFRunLoopCommonModes);
 
 #if ASRunLoopQueueLoggingEnabled
     _runloopQueueLoggingTimer = [NSTimer timerWithTimeInterval:1.0 target:self selector:@selector(checkRunLoop) userInfo:nil repeats:YES];
@@ -400,7 +403,10 @@ - (instancetype)init
 
 - (void)dealloc
 {
-  CFRunLoopRemoveSource(_runLoop, _runLoopSource, kCFRunLoopCommonModes);
+  ASDisplayNodeAssertMainThread();
+
+  CFRelease(_internalQueueHashSet);
+  CFRunLoopRemoveSource(CFRunLoopGetMain(), _runLoopSource, kCFRunLoopCommonModes);
   CFRelease(_runLoopSource);
   _runLoopSource = nil;
 
@@ -420,60 +426,30 @@ - (void)checkRunLoop
 
 - (void)processQueue
 {
-  // If we have an execution block, this vector will be populated, otherwise remains empty.
-  // This is to avoid needlessly retaining/releasing the objects if we don't have a block.
-  std::vector<id> itemsToProcess;
-
-  {
-    ASDN::MutexLocker l(_internalQueueLock);
-
-    NSInteger internalQueueCount = _internalQueue.count;
-    // Early-exit if the queue is empty.
-    if (internalQueueCount == 0) {
-      return;
-    }
-
-    ASSignpostStart(ASSignpostRunLoopQueueBatch);
+  ASDisplayNodeAssertMainThread();
 
-    /**
-     * For each item in the next batch, if it's non-nil then NULL it out
-     * and if we have an execution block then add it in.
-     * This could be written a bunch of different ways but
-     * this particular one nicely balances readability, safety, and efficiency.
-     */
-    NSInteger foundItemCount = 0;
-    for (NSInteger i = 0; i < internalQueueCount && foundItemCount < internalQueueCount; i++) {
-      /**
-       * It is safe to use unsafe_unretained here. If the queue is weak, the
-       * object will be added to the autorelease pool. If the queue is strong,
-       * it will retain the object until we transfer it (retain it) in itemsToProcess.
-       */
-      __unsafe_unretained id ptr = (__bridge id)[_internalQueue pointerAtIndex:i];
-      if (ptr != nil) {
-        foundItemCount++;
-        itemsToProcess.push_back(ptr);
-        [_internalQueue replacePointerAtIndex:i withPointer:NULL];
-      }
-    }
-
-    [_internalQueue compact];
+  ASDN::UniqueLock l(_internalQueueLock);
+  NSInteger count = _internalQueue.size();
+  // Early-exit if the queue is empty.
+  if (count == 0) {
+    return;
   }
-
-  // itemsToProcess will be empty if _queueConsumer == nil so no need to check again.
-  const auto count = itemsToProcess.size();
-  if (count > 0) {
-    as_activity_scope_verbose(as_activity_create("Process run loop queue batch", _rootActivity, OS_ACTIVITY_FLAG_DEFAULT));
-    const auto itemsEnd = itemsToProcess.cend();
-    for (auto iterator = itemsToProcess.begin(); iterator < itemsEnd; iterator++) {
-      __unsafe_unretained id value = *iterator;
-      [value prepareForCATransactionCommit];
-      as_log_verbose(ASDisplayLog(), "processed %@", value);
-    }
-    if (count > 1) {
-      as_log_verbose(ASDisplayLog(), "processed %lu items", (unsigned long)count);
-    }
+  as_activity_scope_verbose(as_activity_create("Process run loop queue batch", _rootActivity, OS_ACTIVITY_FLAG_DEFAULT));
+  ASSignpostStart(ASSignpostRunLoopQueueBatch);
+
+  // Swap buffers, clear our hash table.
+  _internalQueue.swap(_batchBuffer);
+  CFSetRemoveAllValues(_internalQueueHashSet);
+
+  // Unlock early. We are done with internal queue, and batch buffer is main-thread-only so no lock.
+  l.unlock();
+
+  for (const id<ASCATransactionQueueObserving> &value : _batchBuffer) {
+    [value prepareForCATransactionCommit];
+    as_log_verbose(ASDisplayLog(), "processed %@", value);
   }
-
+  _batchBuffer.clear();
+  as_log_verbose(ASDisplayLog(), "processed %lu items", (unsigned long)count);
   ASSignpostEnd(ASSignpostRunLoopQueueBatch);
 }
 
@@ -489,29 +465,21 @@ - (void)enqueue:(id<ASCATransactionQueueObserving>)object
   }
 
   ASDN::MutexLocker l(_internalQueueLock);
-
-  // Check if the object exists.
-  BOOL foundObject = NO;
-
-  for (id currentObject in _internalQueue) {
-    if (currentObject == object) {
-      foundObject = YES;
-      break;
-    }
+  if (CFSetContainsValue(_internalQueueHashSet, (__bridge void *)object)) {
+    return;
   }
-
-  if (!foundObject) {
-    [_internalQueue addPointer:(__bridge void *)object];
-
+  CFSetAddValue(_internalQueueHashSet, (__bridge void *)object);
+  _internalQueue.emplace_back(object);
+  if (_internalQueue.size() == 1) {
     CFRunLoopSourceSignal(_runLoopSource);
-    CFRunLoopWakeUp(_runLoop);
+    CFRunLoopWakeUp(CFRunLoopGetMain());
   }
 }
 
 - (BOOL)isEmpty
 {
   ASDN::MutexLocker l(_internalQueueLock);
-  return _internalQueue.count == 0;
+  return _internalQueue.empty();
 }
 
 - (BOOL)isEnabled

Tests/ASDisplayNodeTestsHelper.mm

@@ -58,7 +58,7 @@ void ASDisplayNodeSizeToFitSizeRange(ASDisplayNode *node, ASSizeRange sizeRange)
 
 void ASCATransactionQueueWait(ASCATransactionQueue *q)
 {
-  if (!q) { q = ASCATransactionQueue.sharedQueue; }
+  if (!q) { q = ASCATransactionQueueGet(); }
   NSDate *date = [NSDate dateWithTimeIntervalSinceNow:1];
   BOOL whileResult = YES;
   while ([date timeIntervalSinceNow] > 0 &&