¿Cómo lograr un comportamiento sin bloqueo, pero con bloqueo?

Estoy implementando una cola de consumidor único de productor único sin bloqueo para una aplicación de red intensiva. Tengo un montón de subprocesos de trabajo que reciben trabajo en sus propias colas separadas, que luego eliminan y procesan.

La eliminación de los bloqueos de estas colas ha mejorado en gran medida el rendimiento bajo carga alta, pero ya no bloquean cuando las colas están vacíaslo que a su vez hace que el uso de la CPU se dispare.

¿Cómo puedo hacer que un subproceso se bloquee de manera eficiente hasta que pueda sacar algo de la cola con éxito o se elimine/interrumpa?

Si está en Linux, considere usar un Futex. Proporciona el rendimiento de una implementación sin bloqueo mediante el uso de operaciones atómicas en lugar de llamadas al kernel como lo haría un mutex, pero si necesita establecer el proceso en inactivo debido a que alguna condición no es verdadera (es decir, bloqueo de contención), lo hará luego haga las llamadas al kernel apropiadas para poner el proceso en suspensión y reactivarlo en un evento futuro. Es básicamente como un semáforo muy rápido.

En Linux, futex se puede utilizar para bloquear un hilo. Pero ten en cuenta que Los futexes son complicados!

ACTUALIZACIÓN: las variables de condición son mucho más seguras de usar que futexes y son más portátiles. Sin embargo, una variable de condición se usa en combinación con un mutex, por lo que, estrictamente hablando, el resultado ya no estará libre de bloqueos. Sin embargo, si su objetivo principal es el rendimiento (y no la garantía del progreso global), y la parte bloqueada (es decir, una condición para verificar después de activar el subproceso) es pequeña, es posible que obtenga resultados satisfactorios sin necesidad de entrar en sutilezas de integrar futexes en el algoritmo.

Si está en Windows, no podrá usar futexes, pero Windows Vista tiene un mecanismo similar llamado Eventos con clave. Desafortunadamente, esto no es parte de la API publicada (es una API nativa de NTDLL), pero puede usarla siempre que acepte la advertencia de que podría cambiar en futuras versiones de Windows (y no es necesario que se ejecute en núcleos anteriores a Vista). Asegúrese de leer el artículo que vinculé arriba. Aquí hay un no probado boceto de cómo podría funcionar:

/* Interlocked SList queue using keyed event signaling */

struct queue {
    SLIST_HEADER slist;
    // Note: Multiple queues can (and should) share a keyed event handle
    HANDLE keyed_event;
    // Initial value: 0
    // Prior to blocking, the queue_pop function increments this to 1, then
    // rechecks the queue. If it finds an item, it attempts to compxchg back to
    // 0; if this fails, then it's racing with a push, and has to block
    LONG block_flag;
};

void init_queue(queue *qPtr) {
    NtCreateKeyedEvent(&qPtr->keyed_event, -1, NULL, 0);
    InitializeSListHead(&qPtr->slist);
    qPtr->blocking = 0;
}

void queue_push(queue *qPtr, SLIST_ENTRY *entry) {
    InterlockedPushEntrySList(&qPtr->slist, entry);

    // Transition block flag 1 -> 0. If this succeeds (block flag was 1), we
    // have committed to a keyed-event handshake
    LONG oldv = InterlockedCompareExchange(&qPtr->block_flag, 0, 1);
    if (oldv) {
        NtReleaseKeyedEvent(qPtr->keyed_event, (PVOID)qPtr, FALSE, NULL);
    }
}

SLIST_ENTRY *queue_pop(queue *qPtr) {
    SLIST_ENTRY *entry = InterlockedPopEntrySList(&qPtr->slist);
    if (entry)
        return entry; // fast path

    // Transition block flag 0 -> 1. We must recheck the queue after this point
    // in case we race with queue_push; however since ReleaseKeyedEvent
    // blocks until it is matched up with a wait, we must perform the wait if
    // queue_push sees us
    LONG oldv = InterlockedCompareExchange(&qPtr->block_flag, 1, 0);

    assert(oldv == 0);

    entry = InterlockedPopEntrySList(&qPtr->slist);
    if (entry) {
        // Try to abort
        oldv = InterlockedCompareExchange(&qPtr->block_flag, 0, 1);
        if (oldv == 1)
            return entry; // nobody saw us, we can just exit with the value
    }

    // Either we don't have an entry, or we are forced to wait because
    // queue_push saw our block flag. So do the wait
    NtWaitForKeyedEvent(qPtr->keyed_event, (PVOID)qPtr, FALSE, NULL);
    // block_flag has been reset by queue_push

    if (!entry)
        entry = InterlockedPopEntrySList(&qPtr->slist);
    assert(entry);

    return entry;
}

También podría usar un protocolo similar usando Delgado leer escribir cerraduras y Variables de condición, con una ruta rápida sin bloqueo. Estos son envoltorios sobre eventos con clave, por lo que pueden generar más gastos generales que usar eventos con clave directamente.

¿Has probado la espera condicional? Cuando la cola se vacía, simplemente comience a esperar un nuevo trabajo. El subproceso que pone trabajos en la cola debería disparar la señal. De esta manera, solo usa bloqueos cuando la cola está vacía.

https://computing.llnl.gov/tutorials/pthreads/#ConditionVariables

Puede hacer que un subproceso se duerma utilizando la función sigwait(). Puede activar el hilo con pthread_kill. Esto es mucho más rápido que las variables de condición.

Podrías agregar duerme mientras espera. Simplemente elija la mayor espera que esté dispuesto a tener, luego haga algo como esto (pseudocódigo porque no recuerdo la sintaxis de pthread):

WAIT_TIME = 100; // Set this to whatever you're happy with
while(loop_condition) {
   thing = get_from_queue()
   if(thing == null) {
       sleep(WAIT_TIME);
   } else {
       handle(thing);
   }
}

Incluso algo corto como una suspensión de 100 ms debería reducir significativamente el uso de la CPU. Sin embargo, no estoy seguro de en qué punto el cambio de contexto lo hará peor que esperar ocupado.