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

[mr-aleph.livejournal.com]

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

фанаты, что с них взять =)

Задача конечно игрушечная

[fenikso.livejournal.com]

Интересно, сколько памяти/процессора съест руби если распараллелить на 267 потоков :)

Re: Задача конечно игрушечная

[thedeemon.livejournal.com]

Много. Но, как говорится, "Вы не должны этого хотеть". Столько потоков делать не нужно.

[fenikso.livejournal.com]

>Столько потоков делать не нужно.
Идея в том, что если нужно - то уже не Ruby нужен. Поэтому я и упомянул что задача изначально игрушечная.

Попадалась ли Вам неплохая статья (http://www.metabrew.com/article/a-million-user-comet-application-with-mochiweb-part-1) про миллион клиентских подключений? :) Хороший пример для Эрланга.

[thedeemon.livejournal.com]

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

Re: Задача конечно игрушечная

[thedeemon.livejournal.com]

Хотя нет, вру. Попробовал создать 100 потоков, делающих sleep, память не съелась - весь процесс занял 5 мегов.

[volodymir-k.livejournal.com]

GIL вроде есть только для настоящего руби, а я jruby вроде нету?

[epicmonkey.livejournal.com]

Exactly. 100%.

[zeux.livejournal.com]

Поправьте меня, если я ошибаюсь, но...
- в данном примере наличие или отсутствие GIL ничего не изменило бы
- в общем случае GIL совсем не отменяет синхронизации из-за сложных состояний, меняющихся не атомарно (больше чем за 1 такт с т.з. GIL).

[thedeemon.livejournal.com]

1. Да, здесь бы не изменило. Но если бы не было GIL, можно было бы использовать этот метод для паралеллизации вычислений.
2. Да, сложные изменения надо синхронизировать.

[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.