ringbuf

Buffer de anillo sin bloqueo

El búfer de anillo de un solo consumidor (MPSC) sin bloqueo que admite operaciones de rango contiguas y que puede usarse convenientemente para pasar mensajes. La implementación se escribe en C11 y se distribuye bajo la licencia BSD de 2 cláusula.

API

• int ringbuf_setup(ringbuf_t *rbuf, unsigned nworkers, size_t length)

  • Configure un nuevo búfer de anillo de una longitud dada. El rbuf es un puntero al objeto de búfer de anillo opaco; La persona que llama es responsable de asignar el espacio para este objeto. Por lo general, el objeto se asignaría dinámicamente si use subprocesos o reservado en una memoria compartida bloqueada si usa procesos. El tamaño de asignación para el objeto se obtendrá utilizando la función ringbuf_get_sizes . Devuelve 0 en el éxito y -1 en la falla.

• void ringbuf_get_sizes(unsigned nworkers, size_t *ringbuf_obj_size, size_t *ringbuf_worker_size)

  • Devuelve el tamaño del opaco ringbuf_t y, opcionalmente, ringbuf_worker_t estructuras. El tamaño de la estructura ringbuf_t depende del número de trabajadores, especificado por el parámetro nworkers .

• ringbuf_worker_t *ringbuf_register(ringbuf_t *rbuf, unsigned i)

  • Registre al trabajador actual (hilo o proceso) como productor. Cada productor debe registrarse. El i es un número de trabajador, comenzando desde cero (es decir, será que nworkers utilizados en la configuración). En el éxito, devuelve un puntero a un opaco ringbuf_worker_t estructurado, que es parte del bloque de memoria ringbuf_t . Al fallar, devuelve NULL .

• void ringbuf_unregister(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Un registro del trabajador especificado de la lista de productores.

• ssize_t ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *worker, size_t len)

  • Solicite un espacio de una longitud dada en el búfer de anillo. Devuelve el desplazamiento en el que el espacio está disponible o -1 en la falla. Una vez que los datos están listos (por lo general, cuando se completa el búfer de anillo), se debe llamar a la función ringbuf_produce para indicar eso. No se permiten llamadas de adquisición anidada.

• void ringbuf_produce(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Indique que el rango adquirido en el amortiguador se produce y está listo para ser consumido.

• size_t ringbuf_consume(ringbuf_t *rbuf, size_t *offset)

  • Obtenga un rango contiguo que esté listo para consumirse. Devuelve cero si no hay datos disponibles para el consumo. Una vez que se consumen los datos (generalmente, cuando se completa la lectura del búfer de anillo), se debe llamar a la función ringbuf_release para indicar eso.

• void ringbuf_release(ringbuf_t *rbuf, size_t nbytes)

  • Indique que la gama consumida ahora puede ser lanzada y ahora puede ser reutilizada por los productores.

Notas

El consumidor devolverá un bloque contiguo de rangos producidos, es decir, la llamada ringbuf_consume no devolverá rangos parciales. Si piensa en el rango producido como un mensaje, el consumidor devolverá un bloque de mensajes, siempre terminando en el límite del mensaje. Tal comportamiento nos permite usar esta implementación de búfer de anillo como cola de mensajes.

La implementación se probó ampliamente en una máquina X86 de 24 núcleos, consulte la prueba de estrés para obtener los detalles sobre la técnica. También proporciona un ejemplo de cómo se puede utilizar el mecanismo para el paso de mensajes.

Advertencias

Esta implementación del búfer de anillo siempre proporciona una gama contigua de espacio para el productor. Se logra mediante una envoltura temprana si el rango solicitado no puede caber al final. La implicación de esto es que la llamada ringbuf_acquire puede fallar si el rango solicitado es mayor que la mitad del tamaño del búfer. Por lo tanto, puede ser necesario asegurarse de que el tamaño del búfer de anillo sea al menos el doble que el tamaño máximo de la unidad de producción.

También se debe tener en cuenta que una de las compensaciones de dicho diseño es que el consumidor actualmente realiza un escaneo O (n) en la lista de productores.

Ejemplo

Productores:

 if (( w = ringbuf_register ( r , worker_id )) == NULL )
	err ( EXIT_FAILURE , "ringbuf_register" )

...

if (( off = ringbuf_acquire ( r , w , len )) != -1 ) {
	memcpy ( & buf [ off ], payload , len );
	ringbuf_produce ( r , tls );
}

Consumidor:

 if (( len = ringbuf_consume ( r , & off )) != 0 ) {
	process ( & buf [ off ], len );
	ringbuf_release ( r , len );
}