Implementazione del pattern produttore-consumatore in Ruby

Question

Diciamo che ho 200 costose chiamate al metodo da fare (ognuna con argomenti diversi). Per qualche ragione, posso eseguire 5 di queste chiamate in parallelo, ma non di più. I potrebbe eseguirli uno alla volta, ma fare 5 alla volta è 5 volte più veloce.

Voglio eseguire sempre cinque cose. Non voglio fare la fila cinque, aspettare fino a quando tutti e cinque sono finiti, e poi fare la coda altri cinque. Se faccio la coda A, B, C, D, E e C finiamo per primi, voglio immediatamente sostituirlo con F, anche se A e B non sono ancora finiti.

Ho letto questo problema, poiché è qualcosa che posso immaginare di accadere regolarmente. La soluzione sembra essere il modello produttore-consumatore e Ruby ha alcune strutture incorporate nella sua libreria standard da utilizzare con quel modello (Queue e SizedQueue). Ho giocato con esempi di codice, ho letto parte della documentazione e penso di averne una comprensione grossolana. Ma ho alcune domande che non sono sicuro della mia soluzione, e l'intera area del multithreading è una novità per me, quindi ho pensato di chiedere qui per assicurarmi di non essere, sai, totalmente sbagliato e solo fortunato .

Quindi, ecco un programma di test che ho scritto:

q = Queue.new 
q << 'balloon' 
q << 'sandwich' 
q << 'clown' 
q << 'fairy floss' 
q << 'ferris wheel' 
q << 'magician' 
q << 'cake' 
q << 'present' 
q << 'chip' 
q << 'game' 
q << 'animal' 

consumer_1 = Thread.new do 
    until q.empty? 
    sleep rand(0..10) 
    print "#{q.pop}\n" 
    end 
end 

# consumer 2 and 3 are identical to consumer 1 

[consumer_1, consumer_2, consumer_3].map(&:join) 

La coda contiene una lista delle cose che abbiamo bisogno per una festa di compleanno. 3 thread di consumatori funzionano attraverso l'elenco.

Questo funziona, le mie domande sono:

Se è il numero di consumatori che determina il numero di elementi sono in corso di elaborazione in parallelo, qual è il punto di avere un di dimensioni coda?

Una coda di dimensioni può essere utile solo in situazioni in cui le attività sono infinite, sconosciute o di numero elevato e si desidera interrompere prima di riempire la coda?

Non sono riuscito a implementare correttamente il problema? Creare più thread a mano e quindi chiamare join su di essi sembra leggermente disordinato. C'è una soluzione migliore?

Answer 1

A SizedQueue impedisce al produttore di aggiungere elementi più velocemente di quanto i consumatori possano consumare.

SizedQueue#push

Se non v'è più spazio nella coda, aspetta fino a quando lo spazio diventa disponibili

Queue#pop/SizedQueue#pop

Se la coda è vuoto , il thread chiamante viene sospeso finché i dati non sono inseriti nella coda.

SizedQueue vs coda

require 'thread' 
require 'logger' 

queue = SizedQueue.new(3) 
#queue = Queue.new # goes berzerk 
logger = Logger.new(STDOUT) 

Thread.new do 
    item = 0 
    loop do 
    item += 1 
    queue << item 
    logger.info "#{item} produced" 
    end 
end 

consumers = 2.times.map do |i| 
    Thread.new do 
    loop do 
     item = queue.pop 
     logger.info "consumed #{item}" 
     sleep item 
    end 
    end 
end 

consumers.each(&:join) 

Come fermare i thread di lavoro

require 'thread' 
require 'logger' 

queue = Queue.new 
logger = Logger.new(STDOUT) 
consumers_count = 5 

end_object = BasicObject.new 

consumers = consumers_count.times.map do 
    Thread.new do 
    until (item = queue.pop) == end_object 
     logger.info "consumed #{item}" 
    end 
    end 
end 

1000.times.each { |item| queue << item } 
consumers_count.times { queue << end_object } 

consumers.each(&:join) 

Ulteriori approfondimenti:

• a writeup about Queue and SizedQueue