Filetage en Ruby avec une limite

Question

J'ai une tâche à effectuer, do_stuff(opts), qui prendra ~ 1s chacun, même si 1 à 10 d'entre eux fonctionnent en parallèle. J'ai besoin de collecter un tableau des résultats pour chaque opération à la fin. Si j'ai 30 choses à faire, comment pourrais-je utiliser threading efficacement pour mettre en file d'attente les opérations do_stuff(opts) donc pas plus de 10 sont en cours d'exécution, mais le tableau de résultats n'est pas donné/imprimé/etc jusqu'à ce que tous (30) les tâches ont été accomplies?

J'ai généralement au moins du code pour essayer d'illustrer ce que je veux dire, mais avec le filetage, je suis un peu perdu! Merci à l'avance

Answer 1

Vous devez implémenter this pattern
This question explique comment cela peut être fait en Ruby

Answer 2

Aussi, jetez un oeil à this tutoriel si vous êtes nouveau fils Ruby.

Answer 3

Si vous êtes vraiment après la performance, vous pouvez également regarder jruby.
Il utilise des threads OS réels et non les fils verts les autres implémentations de Ruby utilisent

Answer 4

Je ne sais pas comment ça va marcher pour une application plus complexe, mais je l'ai trouvé quelque chose comme cela fonctionne bien pour un scénario simple de filetage avec macruby.

thread_limit = 4 

threads = [] 
things_to_process.each do |thing| 
    until threads.map { |t| t.status }.count("run") < thread_limit do sleep 5 end 
    threads << Thread.new { the_task(thing) } 
end 
output = threads.map { |t| t.value } 

la boucle jusqu'à ce que attend jusqu'à ce qu'il n'y autour de moins que le nombre spécifié de threads créés en cours d'exécution avant d'autoriser l'exécution du fil conducteur de continuer à démarrer le fil suivant.

la variable de sortie recevra un tableau des valeurs renvoyées par the_task avec un ordre correspondant au tableau d'entrée things_to_process. Le thread principal bloquera jusqu'à ce que chaque thread créé renvoie une valeur.

Answer 5

Cette solution rassemble les résultats dans le tableau $ results. Il permet de créer des threads 'thread_limit', puis les attend pour se terminer avant d'en créer d'autres.

$results = [] 

    def do_stuff(opts={}) 
    'done' 
    end 

    def thread_wait(threads) 
    threads.each{|t| t.join} 
    threads.each {|t| $results << t } 
    threads.delete_if {|t| t.status == false} 
    threads.delete_if {|t| t.status.nil? } 
    end 

    opts = {} 
    thread_limit = 20 
    threads = [] 
    records.each do |r| 
    thread_wait(threads) while threads.length >= thread_limit 
    t = Thread.new { do_stuff(opts) } 
    t.abort_on_exception = true 
    threads << t 
    end 
    # Ensure remaining threads complete 
    threads.each{|t| t.join} 

Answer 6

J'utilise parals et paralsmap:

def parals(objects, n: 50) 
    objects.shuffle.each_slice(n).map do |g| 
    print '{' 
    threads = [] 
    g.map { |i| threads << Thread.new(i) { |i| yield(i) } } 
    threads.each(&:join) 
    print '}' 
    end 
end 

def paralsmap(objects, n: 50) 
    res = [] 

    objects.each_slice(n).map do |g| 
    print '{' 
    threads = [] 
    g.map { |i| threads << Thread.new(i, res) { |i| res << yield(i) } } 
    threads.each(&:join) 
    print '}' 
    end 

    res 
end 

.: par exemple

parals((0..100).to_a) { |i| puts i } 
urls = parals((0..100).to_a) { |i| "https://google.com/?q=#{i}" } 

Vous pouvez utiliser le paramètre n pour limiter le nombre de threads.