[ASRunLoopQueue - Performance] Add ASDeallocQueue for efficient object teardown. (#2399)

* [ASRunLoopQueue - Performance] Add ASDeallocQueue for efficient object teardown.

This measurably reduces block overhead and context switching.  In the layout benchmark,
it increases ops/s while actually reducing CPU utilization.  This suggests that we are
now at a lock-bounded local maximum, at least for tri-core devices.

* [ASDeallocQueue] Update convenience helper method and adopt in ASImageNode etc.

* [ASDeallocQueue] Reimplement the queue using a timer-based runloop.

* [Debugging] Re-enable ASDisplayNode Event Log.

* [ASDeallocQueue] Final refinements, comments, code minimization.

* [ASDeallocQueue] Fix for lock release needed in early return (refactoring typo from last commit)

AsyncDisplayKit/ASRunLoopQueue.mm

@@ -12,6 +12,7 @@
 
 #import "ASRunLoopQueue.h"
 #import "ASThread.h"
+#import "ASLog.h"
 
 #import <cstdlib>
 #import <deque>
@@ -25,10 +26,118 @@ static void runLoopSourceCallback(void *info) {
 #endif
 }
 
+#pragma mark - ASDeallocQueue
+
+@implementation ASDeallocQueue {
+  NSThread *_thread;
+  NSCondition *_condition;
+  std::deque<id> _queue;
+  ASDN::RecursiveMutex _queueLock;
+}
+
++ (instancetype)sharedDeallocationQueue
+{
+  static ASDeallocQueue *deallocQueue = nil;
+  static dispatch_once_t onceToken;
+  dispatch_once(&onceToken, ^{
+    deallocQueue = [[ASDeallocQueue alloc] init];
+  });
+  return deallocQueue;
+}
+
+- (void)releaseObjectInBackground:(id)object
+{
+  _queueLock.lock();
+  _queue.push_back(object);
+  _queueLock.unlock();
+}
+
+- (void)threadMain
+{
+  @autoreleasepool {
+    __unsafe_unretained __typeof__(self) weakSelf = self;
+    // 100ms timer.  No resources are wasted in between, as the thread sleeps, and each check is fast.
+    // This time is fast enough for most use cases without excessive churn.
+    CFRunLoopTimerRef timer = CFRunLoopTimerCreateWithHandler(NULL, -1, 0.1, 0, 0, ^(CFRunLoopTimerRef timer) {
+#if ASRunLoopQueueLoggingEnabled
+      NSLog(@"ASDeallocQueue Processing: %d objects destroyed", weakSelf->_queue.size());
+#endif
+      weakSelf->_queueLock.lock();
+      std::deque<id> currentQueue = weakSelf->_queue;
+      if (currentQueue.size() == 0) {
+        weakSelf->_queueLock.unlock();
+        return;
+      }
+      // Sometimes we release 10,000 objects at a time.  Don't hold the lock while releasing.
+      weakSelf->_queue = std::deque<id>();
+      weakSelf->_queueLock.unlock();
+      currentQueue.clear();
+    });
+    
+    CFRunLoopRef runloop = CFRunLoopGetCurrent();
+    CFRunLoopAddTimer(runloop, timer, kCFRunLoopCommonModes);
+    
+    [_condition lock];
+    [_condition signal];
+    // At this moment, the thread is guaranteed to be finished starting.
+    [_condition unlock];
+    
+    // Keep processing events until the runloop is stopped.
+    CFRunLoopRun();
+    
+    CFRunLoopTimerInvalidate(timer);
+    CFRunLoopRemoveTimer(runloop, timer, kCFRunLoopCommonModes);
+  }
+}
+
+- (instancetype)init
+{
+  if ((self = [super init])) {
+    _condition = [[NSCondition alloc] init];
+    
+    _thread = [[NSThread alloc] initWithTarget:self selector:@selector(threadMain) object:nil];
+    _thread.name = @"ASDeallocQueue";
+    
+    // Use condition to ensure NSThread has finished starting.
+    [_condition lock];
+    [_thread start];
+    [_condition wait];
+    [_condition unlock];
+  }
+  return self;
+}
+
+- (void)stop
+{
+  if (!_thread) {
+    return;
+  }
+  
+  [_condition lock];
+  [self performSelector:@selector(_stop) onThread:_thread withObject:nil waitUntilDone:NO];
+  [_condition wait];
+  [_condition unlock];
+  _thread = nil;
+}
+
+- (void)_stop
+{
+  CFRunLoopStop(CFRunLoopGetCurrent());
+}
+
+- (void)dealloc
+{
+  [self stop];
+}
+
+@end
+
+#pragma mark - ASRunLoopQueue
+
 @interface ASRunLoopQueue () {
   CFRunLoopRef _runLoop;
-  CFRunLoopObserverRef _runLoopObserver;
   CFRunLoopSourceRef _runLoopSource;
+  CFRunLoopObserverRef _runLoopObserver;
   std::deque<id> _internalQueue;
   ASDN::RecursiveMutex _internalQueueLock;
   
@@ -50,8 +159,13 @@ static void runLoopSourceCallback(void *info) {
     _internalQueue = std::deque<id>();
     _queueConsumer = [handlerBlock copy];
     _batchSize = 1;
+    _ensureExclusiveMembership = YES;
+    
+    // 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) {
-      [self processQueue];
+      [weakSelf processQueue];
     };
     _runLoopObserver = CFRunLoopObserverCreateWithHandler(NULL, kCFRunLoopBeforeWaiting, true, 0, handlerBlock);
     CFRunLoopAddObserver(_runLoop, _runLoopObserver,  kCFRunLoopCommonModes);
@@ -101,7 +215,7 @@ static void runLoopSourceCallback(void *info) {
 
 - (void)processQueue
 {
-  std::deque<id> itemsToProcess = std::deque<id>();
+  std::deque<id> itemsToProcess = std::deque<id>();  
   BOOL isQueueDrained = NO;
   {
     ASDN::MutexLocker l(_internalQueueLock);
@@ -129,9 +243,9 @@ static void runLoopSourceCallback(void *info) {
   unsigned long numberOfItems = itemsToProcess.size();
   for (int i = 0; i < numberOfItems; i++) {
     if (isQueueDrained && i == numberOfItems - 1) {
-      self.queueConsumer(itemsToProcess[i], YES);
+      _queueConsumer(itemsToProcess[i], YES);
     } else {
-      self.queueConsumer(itemsToProcess[i], isQueueDrained);
+      _queueConsumer(itemsToProcess[i], isQueueDrained);
     }
   }
 
@@ -139,7 +253,7 @@ static void runLoopSourceCallback(void *info) {
   if (!isQueueDrained) {
     CFRunLoopSourceSignal(_runLoopSource);
     CFRunLoopWakeUp(_runLoop);
-   }
+  }
   
   ASProfilingSignpostEnd(0, self);
 }
@@ -154,10 +268,13 @@ static void runLoopSourceCallback(void *info) {
 
   // Check if the object exists.
   BOOL foundObject = NO;
-  for (id currentObject : _internalQueue) {
-    if (currentObject == object) {
-      foundObject = YES;
-      break;
+    
+  if (_ensureExclusiveMembership) {
+    for (id currentObject : _internalQueue) {
+      if (currentObject == object) {
+        foundObject = YES;
+        break;
+      }
     }
   }