Parallel map

[thedeemon]

Не раз уже видел восторженные отчеты об успешном применении Эрланга в задаче скачивания большого числа файлов. На мое замечание, что распараллеливание кучи одинаковых независимых заданий - невелико достижение, мне ответили, что для обычных императивных языков это вполне достижение. Сегодня я сам столкнулся с такой задачей, и на совершенно императивном языке Руби задача решилась очень просто.


Недавно узнал о сайте с комиксами abstrusegoose.com, местами попадаются неплохие, но листать по одному задалбывает. Решил сгенерить себе страничку со всеми сериями сразу.

Функция, которая скачивает страничку очередной серии, выдирает из нее нужный кусочек и приводит его к нужному виду, выглядит так:

require 'open-uri'

def fetch(n, tries=5)
  print n, "..\n"
  begin
    data = open("http://abstrusegoose.com/#{n}").read
  rescue
    if tries > 0 then return fetch(n, tries-1)
    else data = "" end
  end
  r = %r{<p><img class="aligncenter".*?</p>}m
  "#{n}<br> #{r.match(data).to_s} <hr>\n"
end

Получить последовательно все кусочки и собрать в одну строку можно так:

all = ""
for n in 1..267
  all << fetch(n)
end

Но это не слишком быстро, большая часть времени проходит в ожидании прихода данных. Зато такое ожидание - как раз одна из немногих вещей, которые в Руби поддаются эффективному распараллеливанию. :) Добавим в класс массивов операцию параллельного отображения (map):

class Array
 def par_map(nt, &f)
  slices = Array.new(nt) {|i| self[i*length/nt...(i+1)*length/nt] }
  threads = slices.map {|xs| Thread.new { xs.map &f } }
  threads.inject([]) {|r, t| r + t.value}
 end
end

Параметр nt - число потоков, на которое распараллеливать.
В первой строчке исходный массив делится на примерно равные части, количество которых равно числу потоков.
Во второй строчке для каждой части создается поток, который применяет переданную в метод операцию ко всем элементам своей части. Все созданные потоки сохраняются в массиве threads. При покидании второй строчки потоки созданы, некоторые, возможно, уже начали работу.
В третьей строчке результаты работы потоков объединяются в один массив. Для получения результата работы потока t идет обращение к свойству t.value, которое дожидается окончания его работы, прежде чем вернуть результат. inject - это обычный fold так в Руби называется.
Вот и все, три строчки - все распараллеливание.

Запустить скачивание в 10 потоков и собрать результаты в одну строку можно так:

all = (1..267).to_a.par_map(10) {|n| fetch(n)}.join

Тут range 1..267 превращается в массив со значениями от 1 до 267, у него вызывается только что описанный метод параллельного отображения, которому передается число потоков и операция для элемента. Результирующий массив строк собирается в одну методом join. Задача решена! Ускорение в данном случае близко к линейному.

Из-за GIL распараллелить таким образом вычисления на Руби не выйдет, зато благодаря GIL нет никаких проблем с синхронизацией.

Comments

[epicmonkey.livejournal.com]

Two days ago I found a pretty similar article in reddit about parallelization with Ruby (here's the link: http://t-a-w.blogspot.com/2010/05/very-simple-parallelization-with-ruby.html). In that article Taw (the author of the article) also extended Enumerable module, in your case Array class, with "in_parallel" method. However he added "todo" query that prevented Ruby from creating to many threats instead of splitting array to smaller pieces and processing them in separate threads (btw, he also extended Exception class to ignore exception and to print error messages). Actually I found it interesting and hope it can help you too.

[thedeemon.livejournal.com]

Funny synchronism, didn't see that article.

There are two problems I see there:
1. Results of computations are ignored, that's not always good.
2. The start filling the queue after running threads, and each thread terminates when the queue is empty, so they can have some or all threads terminated before they fill the queue, that's a big error.

[thedeemon.livejournal.com]

Oops, my bad, ignore #2. Threads are terminated when meet a nil value, not on empty queue. So they won't terminate before starting the work, but the whole thing will behave badly if there are some nils in enumeration.