En informática , el problema productor-consumidor (también conocido como problema del buffer acotado ) 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 los ordenadores Electrologica X1 y X8: "El primer uso de productor-consumidor fue en parte software, en parte hardware: el componente que se encargaba del transporte de información entre la tienda y El periférico se llamaba 'un canal'... La sincronización estaba controlada por dos semáforos de conteo en lo que ahora conocemos como la disposición productor/consumidor: el semáforo que indicaba la longitud de la cola, era incrementado (en una V) por la CPU y "[ 1]
Dijkstra escribió sobre el caso del buffer ilimitado: "Consideramos dos procesos, que se llaman 'productor' y 'consumidor' respectivamente. El productor es un proceso cíclico y cada vez que pasa por su ciclo produce una cierta porción de información, que tiene que ser procesado por el consumidor. El consumidor también es un proceso cíclico y cada vez que pasa por su ciclo, puede procesar la siguiente porción de información, tal como ha sido producida por el productor... Suponemos que los dos procesos conectarse para ello a través de un buffer con capacidad ilimitada." [2]
Escribió sobre el caso del buffer acotado: "Hemos estudiado un productor y un consumidor acoplados a través de un buffer con capacidad ilimitada... La relación se vuelve simétrica, si los dos se acoplan a través de un buffer de tamaño finito, digamos N porciones" [3 ]
Y sobre el caso de múltiples productores-consumidores: "Consideramos una serie de pares de productores/consumidores, donde el par i está acoplado a través de un flujo de información que contiene n i porciones. Suponemos... el buffer finito que debería contener todas las porciones de todos los flujos tener una capacidad de porciones 'pequeñas'." [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, es decir, que no son adecuados para lenguajes de nivel superior. En cambio, los eventos de sincronización naturales son intercambios de mensajes ". [5]
La solución tampón delimitada por semáforo original se escribió en estilo ALGOL . El buffer puede almacenar N porciones o elementos. El semáforo "número de porciones de cola" cuenta las ubicaciones llenas en el búfer, el semáforo "número de posiciones vacías" cuenta las ubicaciones vacías en el búfer y el semáforo "manipulación del búfer" funciona como mutex para las operaciones de colocació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 subproceso 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 de la cola es cero, el subproceso del consumidor esperará en la operación P (número de porciones de la cola). Las operaciones V() liberan los semáforos. Como efecto secundario, un hilo puede pasar de la cola de espera a la cola de listo. La operación P() reduce el valor del semáforo hasta cero. La operación V() aumenta el valor del semáforo. [6]
comenzar número entero de partes de la cola , número de posiciones vacías , manipulación del buffer ; número de porciones en cola : = 0 ; número de posiciones vacías := N ; manipulación del búfer : = 1 ; productor parbegin : comenzar de nuevo 1 : producir la siguiente porción ; P ( número de posiciones vacías ) ; P ( manipulación del buffer ) ; agregar porción al búfer ; V ( manipulación del buffer ) ; V ( número de porciones de cola ) ; ir de nuevo 1 final ; consumidor : empezar de nuevo 2 : P ( número de porciones en cola ) ; P ( manipulación del buffer ) ; tomar parte del buffer ; V ( manipulación del buffer ) ; V ( número de posiciones vacías ) ; porción del proceso tomada ; ir de nuevo 2 fin parend fin A partir de C++ 20, los semáforos son parte del lenguaje. La solución de Dijkstra se puede escribir fácilmente en C++ moderno. La variable buffer_manipulation es un mutex. La característica del semáforo de adquirir en un hilo y liberar en otro hilo no es necesaria. La instrucción lock_guard() en lugar de un par lock() y unlock() es C++ RAII . El destructor lock_guard garantiza la liberación del bloqueo en caso de una excepción. Esta solución puede manejar múltiples subprocesos de consumidores y/o múltiples subprocesos de productores.
#include <hilo> #include <mutex> #include <semáforo> std :: contando_semaphore < N > número_de_porciones_en_cola { 0 }; std :: contando_semaphore < N > número_de_posiciones_empty { N }; std :: manipulación_búfer mutex ; productor vacío () { para (;;) { porción porción = producir_next_portion (); número_de_posiciones_vacías . adquirir (); { std :: lock_guard < std :: mutex > g ( manipulación_búfer ); add_portion_to_buffer ( porción ); } número_de_porciones_en_cola . liberar (); } } consumidor vacío () { para (;;) { número_de_porciones_queueing . adquirir (); Porción porción ; { std :: lock_guard < std :: mutex > g ( buffer_manipulación ); porción = tomar_porción_de_buffer (); } número_de_posiciones_vacías . liberar (); proceso_porción_tomada ( porción ); } } int main () { std :: hilo t1 ( productor ); std :: hilo t2 ( consumidor ); t1 . unirse (); t2 . unirse (); } [¿ síntesis inadecuada? ]
Per Brinch Hansen definió el monitor: Usaré el término monitor para denotar una variable compartida y el conjunto de operaciones significativas en 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]
búfer delimitado : monitor de inicio del búfer : matriz 0 .. N - 1 de la porción ; cabeza , cola : 0 .. N - 1 ; contar : 0 .. N ; no vacío , no lleno : condición ; procedimiento anexar ( x : porción ) ; comenzar si cuenta = N entonces no completo . esperar ; nota 0 <= recuento < N ; buffer [ cola ] := x ; cola := cola ( + ) 1 ; contar := contar + 1 ; no vacío . agregar fin de señal ; procedimiento eliminar ( resultado x : porción ) ; comience si cuenta = 0 entonces no está vacío . esperar ; nota 0 < cuenta <= N ; x := búfer [ cabeza ] ; cabeza := cabeza ( + ) 1 ; contar : = contar - 1 ; no lleno . eliminación del final de la señal ; cabeza := 0 ; cola := 0 ; contar : = 0 ; final del búfer acotado ; El monitor es un objeto que contiene variables buffer, heady para realizar un búfer circular , las variables de condición y para la sincronización y los métodos y para acceder al búfer delimitado tail. La espera de operación del monitor corresponde a la operación del semáforo P o adquisición, la señal corresponde a V o liberación. La operación marcada con un círculo (+) se toma en módulo N. El pseudocódigo de estilo Pascal presentado muestra un monitor Hoare. Un monitor Mesa utiliza en lugar de . Una versión del lenguaje de programación C++ es:countnonemptynonfullappendremovewhile countif count
class Bounded_buffer { Búfer de porción [ N ]; // 0..N-1 cabeza , cola sin firmar ; // 0..N-1 recuento sin firmar ; // 0..N std :: variable_condición no vacía , no llena ; std :: exclusión mutua mtx ; público : void append ( Porción x ) { std :: Unique_lock < std :: mutex > lck ( mtx ); no lleno . esperar ( lck , [ & ]{ regresar ! ( N == contar ); }); afirmar ( 0 <= recuento && recuento < N ); búfer [ cola ++ ] = x ; cola %= N ; ++ contar ; no vacío . notificar_uno (); } Porción remove () { std :: Unique_lock < std :: mutex > lck ( mtx ); no vacío . esperar ( lck , [ & ]{ regresar ! ( 0 == contar ); }); afirmar ( 0 < recuento && recuento <= N ); Porción x = buffer [ head ++ ]; cabeza %= N ; -- contar ; no lleno . notificar_uno (); devolver x ; } Bounded_buffer () { cabeza = 0 ; cola = 0 ; contar = 0 ; } }; La versión C++ necesita un mutex adicional por razones técnicas. Utiliza afirmar para hacer cumplir las condiciones previas para las operaciones de agregar y eliminar el búfer.
La primera solución productor-consumidor en las computadoras de Electrologica utilizó "canales". Canales definidos por Hoare : una alternativa a la denominación explícita de origen y destino sería nombrar un puerto a través del cual se realizará la comunicación. Los nombres de los puertos serían locales para los procesos, y la forma en que los pares de puertos se conectarán mediante canales podría declararse en el encabezado 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 , el lenguaje de programación del sistema operativo Inferno, Limbo, tienen canales. El siguiente código fuente C se compila en Plan 9 desde User Space :
#incluye "uh" #incluye "libc.h" #incluye "thread.h" enumeración { PILA = 8192 }; productor vacío ( void * v ) { Canal * ch = v ; para ( uint i = 1 ; ; ++ i ) { dormir ( 400 ); imprimir ( "p %d \n " , i ); sendul ( ch , i ); } } consumidor vacío ( void * v ) { Canal * ch = v ; para (;;) { uint p = recvul ( ch ); imprimir ( " \t\t c %d \n " , p ); dormir ( 200 + nrand ( 600 )); } } void threadmain ( int argc , char ** argv ) { int ( * mk ) ( void ( * fn ) ( void * ), void * arg , uint pila ); mk = crear hilo ; Canal * ch = chancreate ( sizeof ( ulong ), 1 ); mk ( productor , canal , PILA ); mk ( consumidor , canal , PILA ); recvp ( chancreate ( tamaño de ( vacío * ), 0 )); hilo sale todo ( 0 ); } El punto de entrada del programa está en la función threadmain. La llamada a la función ch = chancreate(sizeof(ulong), 1)crea el canal, la llamada a la función sendul(ch, i)envía un valor al canal y la llamada a la función p = recvul(ch)recibe un valor del canal. El lenguaje de programación Go también tiene canales. Un ejemplo de Ir:
paquete principal importar ( "fmt" "matemáticas/rand" "tiempo" ) var enviar mensaje = 0 func producirMessage () int { tiempo . Dormir ( 400 * tiempo . Milisegundos ) sendMsg ++ fmt . Printf ( "enviarMsg = %v\n" , enviarMsg ) return enviarMsg } func consumirMessage ( recvMsg int ) { fmt . Printf ( "\t\trecvMsg = %v\n" , recvMsg ) tiempo . Dormir ( tiempo . Duración ( 200 + rand . Intn ( 600 )) * tiempo . Milisegundo ) } func main () { ch := make ( chan int , 3 ) go func () { for { ch <- produceMessage () } }() para recvMsg := rango ch { consumirMensaje ( recvMsg ) } } La solución Go productor-consumidor utiliza la rutina Go principal para el consumidor y crea una nueva rutina Go sin nombre para el productor. Las dos rutinas Go están conectadas con el canal ch. Este canal puede poner en cola hasta tres valores int. La declaración ch := make(chan int, 3)crea el canal, la declaración ch <- produceMessage()envía un valor al canal y la declaración recvMsg := range chrecibe 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.
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, dejamos que k sea una constante mayor que by s y r deben ser variables enteras que asuman valores entre 0 y k-1. Suponemos que inicialmente s=r y el buffer está vacío. Al elegir que k sea un múltiplo de b, el búfer se puede implementar como una matriz B [0: b - 1]. El productor simplemente coloca cada mensaje nuevo 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 vaya a L fi ; poner el mensaje en el buffer ; s := ( s + 1 ) mod k ; ir a L ; Consumidor : L : si ( s - r ) mod k = 0 entonces vaya a L fi ; tomar mensaje del buffer ; r := ( r + 1 ) mod k ; ir a L ; La solución Lamport utiliza la espera ocupada en el hilo en lugar de esperar en el programador. Esta solución ignora el impacto del cambio de subproceso del programador en momentos inconvenientes. Si el primer subproceso ha leído un valor de variable de la memoria, el programador cambia al segundo subproceso que cambia el valor de la variable, y el programador vuelve al primer subproceso, entonces el primer subproceso usa el valor anterior de la variable, no el valor actual. . La lectura-modificación-escritura atómica resuelve este problema. El C++ moderno ofrece atomicvariables y operaciones para programación multiproceso. La siguiente solución C++ 11 de espera ocupada para un productor y un consumidor utiliza operaciones atómicas de lectura, modificación y escritura fetch_addy fetch_suben la variable atómica count.
enumeración { norte = 4 }; Búfer de mensajes [ N ]; std :: atómico < sin firmar > contar {0} ; productor vacío () { cola sin firmar {0} ; for (;;) { Mensaje mensaje = producirMessage (); mientras ( N == contar ) ; // buffer de espera ocupado [ tail ++ ] = mensaje ; cola %= N ; contar . fetch_add ( 1 , std :: memoria_order_relaxed ); } } consumidor vacío () { cabeza sin firmar { 0 }; para (;;) { mientras ( 0 == contar ) ; // ocupado esperando Mensaje mensaje = buffer [ head ++ ]; cabeza %= N ; contar . fetch_sub ( 1 , std :: memoria_order_relaxed ); consumirMensaje ( mensaje ); } } int main () { std :: hilo t1 ( productor ); std :: hilo t2 ( consumidor ); t1 . unirse (); t2 . unirse (); } Las variables de índice del búfer circular heady tailson locales de subproceso y, por lo tanto, no son relevantes para la coherencia de la memoria. La variable countcontrola la espera ocupada del hilo productor y consumidor.
