Ejecutando de forma asíncrona pero escribiendo sincrónicamente

Tengo esta situación: un video debe procesarse fotogtwig a fotogtwig, pero a medida que los fotogtwigs se procesan, la salida debe escribirse en un file en secuencia.

Quiero disparar bloques asíncronos utilizando dispatch_async a una queue simultánea para acelerar el process, pero como esta queue es asíncrona, no veo cómo coordinaré para escribir los cuadros en serie en la salida.

Suponga que esta situación: los cuadros 1, 2, 3, 4 y 5 se envían a las queues simultáneas para su procesamiento. Debido a que cualquier bloque puede terminar en cualquier momento, el marco 4 puede ser el primero en terminar, seguido de 5, 3, 1, 2. Entonces, ¿cómo puedo escribir los cuadros en order secuencial a la salida?

Tengo un código como este:

 dispatch_queue_t aQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0); while (true) { video >> frame; // read a frame from the video dispatch_async(aQueue, ^{ processVideo(frame, outputFrame); writeToVideo(outputFrame); // this is here just to show what needs to be done }); // bla bla } 

¿alguna pista?

Gracias

    Utilizaría una combinación de una queue de envío en serie y una NSCondition . La queue de serie se asegura de que ninguna de las NSCondition ocurra simultáneamente, mientras que la NSCondition se asegura de que sucedan en el order correcto.

    De la NSCondition docs:

    Un object de condición actúa como un locking y un punto de control en un hilo determinado. El locking protege su código mientras testing la condición y realiza la tarea desencadenada por la condición. El comportamiento del punto de control requiere que la condición sea verdadera antes de que el hilo continúe con su tarea. Si bien la condición no es verdadera, el hilo bloquea.

    En su situación específica haría algo como esto …

    En su ciclo primero declara un BOOL (inicialmente configurado en NO ) que indica si su marco ha sido procesado o no, y una NSCondition . Luego, dispatch_async a la queue de background para procesar el cuadro y a la queue de serie para escribir los datos.

    Cuando se ejecuta el bloque en la queue de serie, bloquee la NSCondition y luego verifique BOOL para ver si el marco se ha procesado. Si lo tiene, continúe con la escritura. Si no lo ha hecho, wait a NSCondition una signal de NSCondition y verifique nuevamente cuando lo reciba. Cuando haya terminado, unlock la NSCondition .

    Cuando se ejecuta el bloque en la queue de background, bloquee la NSCondition y procese el marco. Cuando se procesa la ttwig, configure BOOL para indicar que se procesa la ttwig. Luego NSCondition y unlock la NSCondition .

    Nota: es importante que solo acceda al BOOL que indica que el marco se procesa y su outputFrame dentro del NSCondition de NSCondition ; el locking se está asegurando de que se mantengan sincronizados entre los hilos.

     // Create the background and serial queues dispatch_queue_t backgroundQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0); dispatch_queue_t writeQueue = dispatch_queue_create("writeQueue", DISPATCH_QUEUE_SERIAL); while (true) { // I'm assuming you have some way to break out of this... NSCondition *condition = [[NSCondition alloc] init]; // These need the __block attribute so they can be changed inside the blocks __block BOOL frameProcessed = NO; __block FrameType outputFrame = nil; // video >> frame; // read a frame from the video // dispatch the frame for processing dispatch_async(backgroundQueue, ^{ [condition lock]; processVideo(frame, outputFrame); frameProcessed = YES; [condition signal]; [condition unlock]; }); // dispatch the write dispatch_async(writeQueue, ^{ [condition lock]; while (!frameProcessed) { [condition wait]; // this will block the current thread until it gets a signal } writeToVideo(outputFrame); [condition unlock]; }); } 

    Nota: Hay un truco semi-sutil con BOOL frameProcessed en el código anterior también. Dado que está declarado dentro del ciclo en lugar de fuera, cada bloque capturará el asociado con su marco.


    Actualización: agregando una NSCondition para leer también.

    Debido a que la escritura en video es lenta en comparación con la ejecución en paralelo, se asignan millones de frameworks y se sitúan en la memory hasta que se guardan en el disco.

    Me encargaría de esto al limitar las lecturas utilizando otra NSCondition que bloquea tus lecturas si hay demasiados fotogtwigs esperando a ser escritos en tu writeQueue . El concepto es casi idéntico al NSCondition que agregamos antes, es solo una condición diferente; en este lanzamiento sería un int que indica cuántos cuadros están esperando ser escritos.

    Antes de su ciclo, defina una readCondition , writeQueueSize y maxWriteQueueSize . Dentro del bucle, primero lock la readCondition , verifique si writeQueueSize >= maxWriteQueueSize . Si no es así, continúe leyendo un marco y haciendo queue para el procesamiento y la escritura. Justo antes de enviar a writeQueue , incremente writeQueueSize . Luego, unlock readCondition .

    Luego, dentro del bloque enviado a writeQueue , una vez que se completa la escritura, lock readCondition , decremento writeQueueSize , y signal y unlock readCondition .

    Eso debería garantizar que nunca haya más que bloques maxWriteQueueSize esperando en writeQueue . Si hay tantos bloques esperando, pausará efectivamente la lectura de los cuadros del video hasta que el writeQueue esté listo para más.

     // Create the background and serial queues dispatch_queue_t backgroundQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0); dispatch_queue_t writeQueue = dispatch_queue_create("writeQueue", DISPATCH_QUEUE_SERIAL); NSCondition *readCondition = [[NSCondition alloc] init]; __block int writeQueueSize = 0; const int maxWriteQueueSize = 10; while (true) { // I'm assuming you have some way to break out of this... NSCondition *writeCondition = [[NSCondition alloc] init]; // These need the __block attribute so they can be changed inside the blocks __block BOOL frameProcessed = NO; __block FrameType outputFrame = nil; [readCondition lock]; while (writeQueueSize >= maxWriteQueueSize) { [readCondition wait]; } // video >> frame; // read a frame from the video // dispatch the frame for processing dispatch_async(backgroundQueue, ^{ [writeCondition lock]; processVideo(frame, outputFrame); frameProcessed = YES; [writeCondition signal]; [writeCondition unlock]; }); // dispatch the write writeQueueSize++; // Increment the write queue size here, before the actual dispatch dispatch_async(writeQueue, ^{ [writeCondition lock]; while (!frameProcessed) { [writeCondition wait]; // this will block the current thread until it gets a signal } writeToVideo(outputFrame); [writeCondition unlock]; // Decrement the write queue size and signal the readCondition that it changed [readCondition lock]; writeQueueSize--; [readCondition signal]; [readCondition unlock]; }); [readCondition unlock]; } 

    Puede hacer esto dando a cada cuadro su propia queue de resultados y encadenando todas las queues en order. Suspendemos todas las queues, excepto la primera. Luego, cuando cada marco finaliza, reanuda la siguiente queue de resultados. Esto forzará a las queues a entregar sus resultados en el order que queramos, independientemente de cuándo completen su trabajo.

    Aquí hay un ejemplo que simplemente usa el sleep para simular una cantidad aleatoria de trabajo e imprime los resultados en el order correcto. El dispatch_group se utiliza aquí para evitar que el progtwig salga demasiado pronto. Puede que no lo necesites en tu caso.

     int main(int argc, const char * argv[]) { @autoreleasepool { dispatch_queue_t mainQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_CONCURRENT); dispatch_group_t group = dispatch_group_create(); dispatch_queue_t myQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL); for (unsigned x = 1; x <= 5; x++ ) { // Chain the queues together in order; suspend all but the first. dispatch_queue_t subQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL); dispatch_set_target_queue(subQueue, myQueue); dispatch_suspend(subQueue); dispatch_group_async(group, mainQueue,^{ // Perform a random amount of work u_int32_t sleepTime = arc4random_uniform(10); NSLog(@"Sleeping for thread %d (%d)", x, sleepTime); sleep(sleepTime); // OK, done with our work, queue our printing, and tell the next guy he can print dispatch_sync(myQueue, ^{ printf("%d ", x); dispatch_resume(subQueue); }); }); myQueue = subQueue; } // Wait for the whole group to finish before terminating dispatch_group_wait(group, DISPATCH_TIME_FOREVER); } return 0; }