#import "ASRunLoopQueue.h"

#import "ASThread.h"

#import "ASLog.h"

#import <cstdlib>

#import <deque>

#define ASRunLoopQueueLoggingEnabled 0

static void runLoopSourceCallback(void *info) {

// No-op

#if ASRunLoopQueueLoggingEnabled

NSLog(@"<%@> - Called runLoopSourceCallback", 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, -init is signalled that 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);

CFRelease(timer);

[_condition lock];

[_condition signal];

// At this moment, -stop is signalled that the thread is guaranteed to be finished exiting.

[_condition unlock];

}

}

- (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];

// At this moment, the thread is guaranteed to be finished running.

[_condition unlock];

_thread = nil;

}

- (void)_stop

{

CFRunLoopStop(CFRunLoopGetCurrent());

}

- (void)dealloc

{

[self stop];

}

#pragma mark - ASRunLoopQueue

@interface ASRunLoopQueue () {

CFRunLoopRef _runLoop;

CFRunLoopSourceRef _runLoopSource;

CFRunLoopObserverRef _runLoopObserver;

std::deque<id> _internalQueue;

ASDN::RecursiveMutex _internalQueueLock;

#if ASRunLoopQueueLoggingEnabled

NSTimer *_runloopQueueLoggingTimer;

#endif

}

@property (nonatomic, copy) void (^queueConsumer)(id dequeuedItem, BOOL isQueueDrained);

@implementation ASRunLoopQueue

- (instancetype)initWithRunLoop:(CFRunLoopRef)runloop andHandler:(void(^)(id dequeuedItem, BOOL isQueueDrained))handlerBlock

{

if (self = [super init]) {

_runLoop = runloop;

_internalQueue = std::deque<id>();

_queueConsumer = handlerBlock;

_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) {

[weakSelf processQueue];

};

_runLoopObserver = CFRunLoopObserverCreateWithHandler(NULL, kCFRunLoopBeforeWaiting, true, 0, handlerBlock);

CFRunLoopAddObserver(_runLoop, _runLoopObserver, 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

CFRunLoopSourceContext *runLoopSourceContext = (CFRunLoopSourceContext *)calloc(1, sizeof(CFRunLoopSourceContext));

if (runLoopSourceContext) {

runLoopSourceContext->perform = runLoopSourceCallback;

#if ASRunLoopQueueLoggingEnabled

runLoopSourceContext->info = (__bridge void *)self;

#endif

_runLoopSource = CFRunLoopSourceCreate(NULL, 0, runLoopSourceContext);

CFRunLoopAddSource(runloop, _runLoopSource, kCFRunLoopCommonModes);

free(runLoopSourceContext);

}

#if ASRunLoopQueueLoggingEnabled

_runloopQueueLoggingTimer = [NSTimer timerWithTimeInterval:1.0 target:self selector:@selector(checkRunLoop) userInfo:nil repeats:YES];

[[NSRunLoop mainRunLoop] addTimer:_runloopQueueLoggingTimer forMode:NSRunLoopCommonModes];

#endif

}

return self;

}

- (void)dealloc

{

if (CFRunLoopContainsSource(_runLoop, _runLoopSource, kCFRunLoopCommonModes)) {

CFRunLoopRemoveSource(_runLoop, _runLoopSource, kCFRunLoopCommonModes);

}

CFRelease(_runLoopSource);

_runLoopSource = nil;

if (CFRunLoopObserverIsValid(_runLoopObserver)) {

CFRunLoopObserverInvalidate(_runLoopObserver);

}

CFRelease(_runLoopObserver);

_runLoopObserver = nil;

}

#if ASRunLoopQueueLoggingEnabled

- (void)checkRunLoop

{

NSLog(@"<%@> - Jobs: %ld", self, _internalQueue.size());

}

#endif

- (void)processQueue

{

std::deque<id> itemsToProcess = std::deque<id>();

BOOL isQueueDrained = NO;

{

ASDN::MutexLocker l(_internalQueueLock);

// Early-exit if the queue is empty.

if (_internalQueue.empty()) {

return;

}

ASProfilingSignpostStart(0, self);

// Snatch the next batch of items.

NSUInteger totalNodeCount = _internalQueue.size();

for (int i = 0; i < MIN(self.batchSize, totalNodeCount); i++) {

id node = _internalQueue[0];

itemsToProcess.push_back(node);

_internalQueue.pop_front();

}

if (_internalQueue.empty()) {

isQueueDrained = YES;

}

}

unsigned long numberOfItems = itemsToProcess.size();

for (int i = 0; i < numberOfItems; i++) {

if (isQueueDrained && i == numberOfItems - 1) {

_queueConsumer(itemsToProcess[i], YES);

} else {

_queueConsumer(itemsToProcess[i], isQueueDrained);

}

}

// If the queue is not fully drained yet force another run loop to process next batch of items

if (!isQueueDrained) {

CFRunLoopSourceSignal(_runLoopSource);

CFRunLoopWakeUp(_runLoop);

}

ASProfilingSignpostEnd(0, self);

}

- (void)enqueue:(id)object

{

if (!object) {

return;

}

ASDN::MutexLocker l(_internalQueueLock);

// Check if the object exists.

BOOL foundObject = NO;

if (_ensureExclusiveMembership) {

for (id currentObject : _internalQueue) {

if (currentObject == object) {

foundObject = YES;

break;

}

}

}

if (!foundObject) {

_internalQueue.push_back(object);

CFRunLoopSourceSignal(_runLoopSource);

CFRunLoopWakeUp(_runLoop);

}

}