Any way to ensure GCD tasks finish in order without serial queues?

Question

I'm using GCD to do some heavy lifting - image manipulation and so on - often with 3 or 4 tasks running concurrently.

Some of these tasks complete more quickly than others. How do I ensure that the callbacks are fired in the correct, original order - without using a serial queue?

For example:

• Task one takes 1 second
• Task two takes 5 seconds
• Task three takes 2 seconds

How do I ensure the final callback order of one, two, three - despite the varied computation time?

// self.queue = dispatch_queue_create("com.example.queue", DISPATCH_QUEUE_CONCURRENT);

dispatch_async(self.queue, ^{
    // Long-running code here of varying complexity

    dispatch_async(dispatch_get_main_queue(), ^{
        // Callback here
    });
});

Edit:

As per the comments, the first notification should go out as soon as Task One completes, even if the remaining tasks are processing. When Task Three completes, it should hold until Task Two is complete, then first off notifications for Two and Three in rapid succession.

I'm thinking some kind of mutable array for pushing and shifting tasks could work. Is there a cleaner way though?

Answer 1

Each of your completion blocks (except the very first) has two dependencies: the heavy-lifting job and the completion block of the prior heavy-lifting job.

It will be much simpler to meet your requirements using NSOperationQueue and NSBlockOperation instead of using GCD directly. (NSOperationQueue is built on top of GCD.)

You need an operation queue and a reference to the prior completion operation:

@property (nonatomic, strong) NSOperationQueue *queue;
@property (nonatomic, strong) NSOperation *priorCompletionOperation;

Initialize the queue to an NSOperationQueue. Leave priorCompletionOperation nil until you get the first job.

Then it's just a matter of setting up your dependencies before submitting the operations to the queues:

    NSBlockOperation *heavyLifting = [NSBlockOperation blockOperationWithBlock:^{
        // long-running code here of varying complexity
    }];

    NSBlockOperation *completion = [NSBlockOperation blockOperationWithBlock:^{
        // Callback here
    }];

    [completion addDependency:heavyLifting];
    if (self.priorCompletionOperation) {
        [completion addDependency:self.priorCompletionOperation];
    }

    [self.queue addOperation:heavyLifting];
    [[NSOperationQueue mainQueue] addOperation:completion];

    self.priorCompletionOperation = completion;

Note that you should make sure this job-queuing code only runs from a single thread at a time. If you only enqueue jobs from the main thread (or main queue) that will happen automatically.

Answer 2

So the final solution was simple enough.

I love the simplicity of NSOperation but I didn't like the idea of having to keep a reference to the previous iteration's queue, or creating and destroying NSOperations on such a tight loop.

I've finally opted for a variation to the GCD/Semaphore model:

// self.serialQueue = dispatch_queue_create("com.example.queue", NULL);
// self.conQueue = dispatch_queue_create("com.example.queue", DISPATCH_QUEUE_CONCURRENT);

// Create a variable that both blocks can access, so we can pass data between them
__block id *message = nil;

// Create semaphore which will be used to lock the serial/callback queue
dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);

// This is effectively the 'callback' that gets fired when the semaphore is released
dispatch_async(self.serialQueue, ^{
    dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);
    if (message) {
        NSLog(@"Callback message: %@", message);
    }
});

// Do the heavy lifting, then release the semaphore which allows the callback to fire
dispatch_async(self.conQueue, ^{

    // ... Heavy lifting here

   message = [NSString stringWithFormat:@"Success!"];
   dispatch_semaphore_signal(semaphore);
});

Answer 3

I admire rob mayoff's solution. There is however also an easy way to accomplish this purely in GCD ;)

Here's a complete sample:

#import <Foundation/Foundation.h>
#include <dispatch/dispatch.h>

typedef void (^completion_block_t)(id result);
typedef void (^operation_t)(completion_block_t completionHandler);

static void enqueueOperation(dispatch_queue_t process_queue,
                             dispatch_queue_t sync_queue,
                             operation_t operation)
{
    __block id blockResult = nil;
    dispatch_semaphore_t sem = dispatch_semaphore_create(0);

    dispatch_async(sync_queue, ^{
        dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
        NSLog(@"%@", blockResult);
    });

    dispatch_async(process_queue, ^{
        operation(^(id result) {
            blockResult = result;
            dispatch_semaphore_signal(sem);
            // release semaphore
        });
    });
}

int main(int argc, const char * argv[])
{
    @autoreleasepool {

        NSLog(@"Start");

        dispatch_queue_t process_queue = dispatch_get_global_queue(0, 0);
        dispatch_queue_t sync_queue = dispatch_queue_create("sync_queue", NULL);

        enqueueOperation(process_queue, sync_queue, ^(completion_block_t completionHandler) {
            sleep(1);
            completionHandler(@"#1");
        });

        enqueueOperation(process_queue, sync_queue, ^(completion_block_t completionHandler) {
            sleep(5);
            completionHandler(@"#2");
        });

        enqueueOperation(process_queue, sync_queue, ^(completion_block_t completionHandler) {
            sleep(2);
            completionHandler(@"#3");
        });

        sleep(6);
        NSLog(@"Finished");
    }
    return 0;
}



2013-07-16 14:16:53.461 test[14140:303] Start
2013-07-16 14:16:54.466 test[14140:1a03] #1
2013-07-16 14:16:58.467 test[14140:1a03] #2
2013-07-16 14:16:58.468 test[14140:1a03] #3
2013-07-16 14:16:59.467 test[14140:303] Finished

(by the way, this is/was part of my network library for processing multipart/x-replace messages in parallel)

Answer 4

As mentioned in the comments the simplest solution is to run the callback only when the final task has completed. You can then call the callback code for all the tasks sequentially. I.e. add the objects into an array if needed and then when a incremental counter in the callback reaches the number of objects in the array, you perform the callback code.

If the callback is object independent, then you can simply use counters and run code after completion. Remember to run any counter manipulation also in the main thread or with a '@sync()' directive to avoid race conditions.

dispatch_async(dispatch_get_main_queue(), ^(void){ /* code*/ });

Edit: using the same array technique in the completion handler you set flag of the object as ready to be sent. Then traverse the array as far as possible and send all the objects that are ready sequentially. Otherwise stop and wait for the next completion handler call. You can use a counter to keep track of position or remove items from the array as they are done, but be sure to do it on the main thread or with a sync block.

@interface MyImage : NSImage
@property (assign) BOOL ready;
@end


@implementation MyImage

@synthesize ready;

- (void)send {
    //Send image, make sure it's threaded send, NSConnection should be okay
}
@end


 NSArray *imagesNeededToSend = [NSMutableArray arrayWithObjects:image1, image2, image3, nil];

dispatch_async(self.queue, ^{
    // Long-running code here of varying complexity

    dispatch_async(dispatch_get_main_queue(), ^{
        self.ready = YES;
        dispatch_async(dispatch_get_main_queue(), ^(void){ [self sendImages] });
    });
});

...

- (void)sendImages {
    MyImage *firstImage = [imagesNeededToSend firstObject]; //Xcode Beta, but you should have a a category for firstObject, very useful.
   if (firstImage.ready) {
       [firstImage send];
       [imagesNeededToSend removeObjectAtIndex:0];
       [self sendImages];
   }
}

Answer 5

You have 2 different things, a concurrent set of image processing and a serial set of upload processing. You could create NSOperations to handle the upload processing and set dependencies between them such that they must run in order. The operations would also wait after they are started (semaphore) until they receive the data to upload. That data would be provided by the concurrent GCD tasks.

Answer 6

How do I ensure that the callbacks are fired in the correct, original order - without using a serial queue?

You don't - the callbacks tell you when the task is finished, and if they finish in an order that differs from the enqueued order, then the callbacks will fire in their completion order.

However, the difference is how you handle receiving the completion notifications. You can just treat it much like you would a threading problem, where you are doing a join on three worker threads. Wait for the first, then the second, then the third to complete. For this, you could use a semaphore or a mutex (or even an atomic counter) which is raised when each completes.

This way, your handler code doesn't really care about the order of completion; it simply waits until each task is completed in order. If a task has already finished, there will be no waiting required. In between waiting, you can always do something else too.