Problema productor-consumidor

En informática , el problema del productor-consumidor (también conocido como el problema del búfer limitado ) es una familia de problemas descritos por Edsger W. Dijkstra desde 1965.

Dijkstra encontró la solución al problema productor-consumidor mientras trabajaba como consultor para las computadoras Electrologica X1 y X8: "El primer uso del concepto productor-consumidor fue en parte software y en parte hardware: el componente que se encargaba del transporte de información entre la memoria y el periférico se denominaba 'canal'... La sincronización se controlaba mediante dos semáforos de conteo en lo que ahora conocemos como la configuración productor/consumidor: un semáforo, que indicaba la longitud de la cola, era incrementado (en V) por la CPU y decrementado (en P) por el canal; el otro, que contaba el número de finalizaciones no confirmadas, era incrementado por el canal y decrementado por la CPU. [Si el segundo semáforo era positivo, se activaba el indicador de interrupción correspondiente.]" ^{[ 1 ]}

Dijkstra escribió sobre el caso del búfer ilimitado: «Consideramos dos procesos, denominados respectivamente "productor" y "consumidor". El productor es un proceso cíclico y, en cada ciclo, produce una cierta porción de información que debe ser procesada por el consumidor. El consumidor también es un proceso cíclico y, en cada ciclo, puede procesar la siguiente porción de información producida por el productor... Suponemos que ambos procesos están conectados mediante un búfer de capacidad ilimitada». ^{[ 2 ]}

Escribió sobre el caso del búfer limitado: "Hemos estudiado un productor y un consumidor acoplados a través de un búfer con capacidad ilimitada... La relación se vuelve simétrica si los dos están acoplados a través de un búfer de tamaño finito, digamos N porciones" ^{[ 3 ].}

Y sobre el caso de múltiples productores-consumidores: "Consideramos varios pares productor/consumidor, donde el par _i está acoplado mediante un flujo de información que contiene n _i porciones. Suponemos... que el búfer finito que debe contener todas las porciones de todos los flujos tiene una capacidad de 'total' porciones." ^{[ 4 ]}

Per Brinch Hansen y Niklaus Wirth pronto vieron el problema de los semáforos: «He llegado a la misma conclusión con respecto a los semáforos, a saber, que no son adecuados para lenguajes de alto nivel. En cambio, los eventos de sincronización naturales son los intercambios de mensajes ». ^{[ 5 ]}

La solución original de búfer delimitado por semáforos se escribió en estilo ALGOL . El búfer puede almacenar N porciones o elementos. El semáforo "número de porciones en cola" cuenta las posiciones llenas en el búfer, el semáforo "número de posiciones vacías" cuenta las posiciones vacías en el búfer y el semáforo "manipulación del búfer" funciona como mutex para las operaciones de inserción y obtención del búfer. Si el búfer está lleno, es decir, el número de posiciones vacías es cero, el hilo productor esperará en la operación P(número de posiciones vacías). Si el búfer está vacío, es decir, el número de porciones en cola es cero, el hilo consumidor esperará en la operación P(número de porciones en cola). Las operaciones V() liberan los semáforos. Como efecto secundario, un hilo puede moverse de la cola de espera a la cola de listos. La operación P() disminuye el valor del semáforo a cero. La operación V() aumenta el valor del semáforo. ^{[ 6 ]}

begin integer number of queueing parts , number of empty positions , buffer manipulation ; number of queueing parts := 0 ; number of empty positions := N ; buffer manipulation := 1 ; parbegin producer : begin again 1 : produce next piece ; P ( number of empty positions ) ; P ( buffer manipulation ) ; add piece to buffer ; V ( buffer manipulation ) ; V ( number of queueing parts ) ; goto again 1 end ; consumer : begin again 2 : P ( number of queueing parts ) ; P ( buffer manipulation ) ; take piece from buffer ; V ( buffer manipulation ) ; V ( number of empty positions ) ; process piece took ; goto again 2 end parend end

A partir de C++ 20, los semáforos forman parte del lenguaje. La solución de Dijkstra se puede escribir fácilmente en C++ moderno. La variable buffer_manipulation es un mutex. No se necesita la función de adquisición de semáforos en un hilo y liberación en otro. La instrucción lock_guard() en lugar de un par lock() y unlock() cumple con la regla RAII de C++ . El destructor lock_guard garantiza la liberación del bloqueo en caso de excepción. Esta solución puede manejar múltiples hilos consumidores y/o múltiples hilos productores.

#include <hilo> #include <mutex> #include <semaphore>std :: counting_semaphore <N> number_of_queueing_portions { 0 } ; std :: counting_semaphore <N> number_of_empty_positions { N } ; std :: mutex buffer_manipulation ;​​void producer () { for (;;) { Portion porción = produce_next_portion (); number_of_empty_positions . acquire (); { std :: lock_guard < std :: mutex > g ( buffer_manipulation ); add_portion_to_buffer ( porción ); } number_of_queueing_portions . release (); } }void consumidor () { for ( ;; ) { number_of_queueing_portions.acquire (); Portion porción ; { std :: lock_guard < std :: mutex > g ( manipulación_buffer ) ; porción = tomar_porción_del_buffer (); } number_of_empty_positions.release ( ) ; procesar_porción_tomada ( porción ); } }int main ( ) { std :: thread t1 ( productor ); std :: thread t2 ( consumidor ); t1.join ( ); t2.join ( ) ; }

Per Brinch Hansen definió el monitor: Usaré el término monitor para referirme a una variable compartida y al conjunto de operaciones significativas sobre ella. El propósito de un monitor es controlar la programación de recursos entre procesos individuales de acuerdo con una política determinada. ^{[ 7 ]} Tony Hoare sentó las bases teóricas para el monitor. ^{[ 8 ]}

buffer acotado : monitor inicio buffer : matriz 0 .. N - 1 de porción ; cabeza , cola : 0 .. N - 1 ; contador : 0 .. N ; no vacío , no lleno : condición ; procedimiento agregar ( x : porción ) ; inicio si contador = N entonces no lleno . esperar ; nota 0 <= contador < N ; buffer [ cola ] := x ; cola := cola ( + ) 1 ; contador := contador + 1 ; no vacío . señal fin agregar ; procedimiento eliminar ( resultado x : porción ) ; inicio si contador = 0 entonces no vacío . esperar ; nota 0 < contador <= N ; x := buffer [ cabeza ] ; cabeza := cabeza ( + ) 1 ; contador := contador - 1 ; no lleno . señal fin eliminar ; cabeza := 0 ; cola := 0 ; recuento := 0 ; fin del búfer delimitado ;

El monitor es un objeto que contiene las variables buffer, head, taily countpara realizar un búfer circular , las variables de condición nonemptyy nonfullpara la sincronización y los métodos appendy removepara acceder al búfer limitado. La operación wait del monitor corresponde a la operación de semáforo P o acquire, signal corresponde a V o release. Las operaciones circulares (+) se toman módulo N. El pseudocódigo presentado en estilo Pascal muestra un monitor Hoare. Un monitor Mesa usa while counten lugar de if count. Una versión en lenguaje de programación C++ es:

template < size_t N > class Bounded_buffer { Portion buffer [ N ]; // 0..N-1 size_t head = 0 , tail = 0 ; // 0..N-1 size_t size = 0 ; // 0..N std :: condition_variable non_empty , non_full ; std :: mutex mtx ;public : void append ( Portion porción ) { std :: unique_lock lck ( mtx ); non_full.wait ( lck , [ & ]{ return size != N ; } ); assert ( size < N ); buffer [ tail ++ ] = std :: move ( porción ); tail % = N ; ++ size ; non_empty.notify_one ( ) ; }Portion remove () { std :: unique_lock lck ( mtx ); non_empty . wait ( lck , [ & ]{ return size != 0 ; }); assert ( size <= N ); Portion portion = std :: move ( buffer [ head ++ ]); head %= N ; -- size ; non_full . notify_one (); return portion ; } };

La versión en C++ necesita un mutex adicional por razones técnicas. Utiliza assert para garantizar las condiciones previas para las operaciones de adición y eliminación de búferes.

La primera solución productor-consumidor en las computadoras Electrologica utilizó "canales". Hoare definió los canales : una alternativa a la denominación explícita de origen y destino sería nombrar un puerto a través del cual se realizaría la comunicación. Los nombres de los puertos serían locales a los procesos, y la forma en que los pares de puertos se conectarían mediante canales podría declararse en la cabecera de un comando paralelo. ^{[ 9 ]} Brinch Hansen implementó canales en los lenguajes de programación Joyce y Super Pascal . El lenguaje de programación del sistema operativo Plan 9, Alef , y el lenguaje de programación del sistema operativo Inferno, Limbo, tienen canales. El siguiente código fuente en C compila en Plan 9 desde el espacio de usuario :

#incluir "uh" #incluir "libc.h" #incluir "thread.h"enum { PILA = 8192 };void productor ( void * v ) { Canal * ch = v ; para ( uint i = 1 ; ; ++ i ) { dormir ( 400 ); imprimir ( "p %d \n " , i ); enviar ( ch , i ); } } void consumidor ( void * v ) { Canal * ch = v ; para (;;) { uint p = recvul ( ch ); imprimir ( " \t\t c %d \n " , p ); dormir ( 200 + nrand ( 600 )); } } void hiloprincipal ( int argc , char ** argv ) { int ( * mk )( void ( * fn )( void * ), void * arg , uint pila ); mk = hilocrear ; Canal * ch = crear canal ( sizeof ( ulong ), 1 ); mk ( productor , ch , PILA ); mk ( consumidor , ch , STACK ); recvp ( chancreate ( sizeof ( void * ), 0 )); threadexitsall ( 0 ); }

El punto de entrada del programa se encuentra en la función threadmain. La llamada ch = chancreate(sizeof(ulong), 1)a la función crea el canal, sendul(ch, i)envía un valor al canal y p = recvul(ch)recibe un valor del canal. El lenguaje de programación Go también tiene canales. Un ejemplo en Go:

paquete principalimport ( "fmt" "math/rand" "time" )var sendMsg = 0func produceMessage () int { time.Sleep ( 400 * time.Millisecond ) sendMsg ++ fmt.Printf ( " sendMsg = %v\n" , sendMsg ) return sendMsg } func consumeMessage ( recvMsg int ) { fmt.Printf ( " \ t \ trecvMsg = % v\n" , recvMsg ) time.Sleep ( time.Duration ( 200 + rand.Intn ( 600 ) ) * time.Millisecond ) } func main ( ) { ch : = make ( chan int , 3 ) go func ( ) { for { ch < - produceMessage () } } ( ) for recvMsg : = range ch { consumeMessage ( recvMsg ) } }

La solución productor-consumidor de Go utiliza la rutina principal de Go para el consumidor y crea una nueva rutina de Go sin nombre para el productor. Ambas rutinas están conectadas mediante el canal `ch`. Este canal puede almacenar hasta tres valores enteros en cola. La instrucción crea el canal, la instrucción envía un valor al canal y la instrucción recibe un valor del canal. ^{[ 10 ]} La asignación de recursos de memoria, la asignación de recursos de procesamiento y la sincronización de recursos se realizan automáticamente mediante el lenguaje de programación.ch := make(chan int, 3)ch <- produceMessage()recvMsg := range ch

Leslie Lamport documentó una solución productor-consumidor de búfer limitado para un productor y un consumidor: Suponemos que el búfer puede contener como máximo b mensajes, b >= 1. En nuestra solución, hacemos que k sea una constante mayor que b, y que s y r sean variables enteras que toman valores entre 0 y k-1. Suponemos que inicialmente s=r y que el búfer está vacío. Al elegir k como un múltiplo de b, el búfer se puede implementar como un arreglo B [0: b - 1]. El productor simplemente coloca cada nuevo mensaje en B[s mod b], y el consumidor toma cada mensaje de B[r mod b]. ^{[ 11 ]} El algoritmo se muestra a continuación, generalizado para k infinito.

Productor : L : si ( s - r ) mod k = b entonces ir a L fi ; poner mensaje en el búfer ; s := ( s + 1 ) mod k ; ir a L ; Consumidor : L : si ( s - r ) mod k = 0 entonces ir a L fi ; tomar mensaje del búfer ; r := ( r + 1 ) mod k ; ir a L ;

La solución de Lamport utiliza espera activa en el hilo en lugar de esperar en el planificador. Esta solución ignora el impacto del cambio de hilo del planificador en momentos inconvenientes. Si el primer hilo ha leído el valor de una variable de la memoria, el planificador cambia al segundo hilo que cambia el valor de la variable, y el planificador vuelve al primer hilo, entonces el primer hilo utiliza el valor antiguo de la variable, no el valor actual. La lectura-modificación-escritura atómica resuelve este problema. C++ moderno ofrece atomicvariables y operaciones para la programación multihilo. La siguiente solución de espera activa en C++11 para un productor y un consumidor utiliza operaciones de lectura-modificación-escritura atómicas fetch_addy fetch_suben la variable atómica count.

enum { N = 4 }; Búfer de mensajes [ N ]; std :: atomic < unsigned > count { 0 }; void productor () { unsigned tail { 0 }; for (;;) { Mensaje mensaje = produceMensaje (); while ( N == count ) ; // ocupado esperando buffer [ tail ++ ] = mensaje ; tail %= N ; count . fetch_add ( 1 , std :: memory_order_relaxed ); } } void consumidor () { unsigned head { 0 }; for (;;) { while ( 0 == count ) ; // ocupado esperando Mensaje mensaje = buffer [ head ++ ]; head %= N ; count . fetch_sub ( 1 , std :: memory_order_relaxed ); consumeMensaje ( mensaje ); } } int main () { std :: thread t1 ( productor ); std :: thread t2 ( consumidor ); t1 . unir ( ); t2.unir ( ); }

Las variables de índice del búfer circular headson taillocales al hilo y, por lo tanto, no son relevantes para la consistencia de la memoria. La variable countcontrola la espera activa del hilo productor y del hilo consumidor.

• Operación atómica
• Patrón de diseño
• FIFO
• Tubería
• Canal
• Implementación en Java: Servicio de Mensajería Java

• Mark Grand Patterns in Java, Volumen 1, Un catálogo de patrones de diseño reutilizables ilustrados con UML
• La revista C/C++ Users Journal (Dr. Dobb's) de enero de 2004 publicó "A C++ Producer-Consumer Concurrency Template Library" , de Ted Yuan, una biblioteca de plantillas de C++ lista para usar. El código fuente y los ejemplos de esta pequeña biblioteca de plantillas se pueden encontrar aquí.
• Ioan Tinca, La evolución del problema productor-consumidor en Java