Playing with Concurrent Haskell - Декларативное программирование

Изучая многопоточность в Хаскелле, сделал простенькую программку, которая создаёт некоторое количество потоков, каждый из которых всего лишь увеличивает на единицу определённое количесвто раз мутабельную переменную (var). При создании каждого потока также на единицу увеличивается другая переменная (cnt); при завершении работы каждый поток уменьшает на единицу эту переменную, а основная программа после создания кучи потоков просто ждёт, пока значение переменной cnt не станет равным нулю...

import IO
import Time
import System
import Control.Concurrent
import Control.Concurrent.MVar

n_threads', n_incs' :: Int
n_threads' = 1000
n_incs'    = 1000

-- Командная строка: имя.exe +RTS -K100M -RTS кол-во_потоков кол-во_инкрементов

main = do
    arg <- getArgs
    let (n_threads, n_incs) = case ((map read arg)::[Int]) of
            nt:ni:_ -> (nt,         ni)
            _       -> (n_threads', n_incs')

    t1  <- Time.getClockTime

    hSetBuffering stdout NoBuffering
    putStr $ "Config: " ++ show n_threads ++ " by " ++ show n_incs ++ "\nStarting... "

    var <- newMVar 0    -- Общий счётчик
    cnt <- newMVar 0    -- Число активных потоков

    let foo 0 = cnt -:= 1
        foo n = var +:= 1 >> foo (n-1)

        makeThreads 0 = return ()
        makeThreads n = cnt +:= 1 >> forkIO (foo n_incs) >> makeThreads (n-1)

        waitFor0 = do cnt' <- readMVar cnt
                      if cnt' == 0 then return ()
                                   else threadDelay 40000 >> waitFor0
    makeThreads n_threads

    putStr "Finishing... "
    waitFor0

    val <- takeMVar var
    putStrLn $ "Done: " ++ show val
    t2  <- Time.getClockTime; printTimeDif t2 t1
  where
    (+:=), (-:=) :: MVar Int -> Int -> IO ()
    var -:= n = var +:= (-n)

    var +:= n = do  prev <- takeMVar var
                    var `putMVar` (prev+n)

printTimeDif ft st = putStrLn $ "Time = " ++ show(Time.tdHour td) ++ ":" ++ show(Time.tdMin td) ++ ":"
                                          ++ show nsecs           ++ "." ++ snms
  where
    td           = ft `Time.diffClockTimes` st
    secs         = Time.tdSec     td
    ms           = Time.tdPicosec td `div` 1000000000
    (nsecs, nms) = if ms < 0 then (secs - 1, ms + 1000) else (secs, ms)
    snms         = reverse $ take 3 $ (reverse $ show nms) ++ "000"

Интересный тест получился (Celeron 1.8, RAM DualDDR266 512MB, Win2k3, GHC 6.6).
Столбцы под forkIO — время работы лёгких потоков, создаваемых по forkIO, а forkOS — тяжёлых потоков OS (по forkOS).

n_threads n_incs       время  (разброс времени)      время (разброс времени)
                             forkIO                        forkOS
    1    1000000     0.7 sec                      0.9 sec
    10    100000     1.1 sec                      1.4 sec
    100    10000     1.0 sec                      0.9 sec (0.9 .. 1.8 sec)
    1000    1000     1.1 sec                      1.2 sec
    5000     200     1.2 sec                      2.5 sec
    10000    100     1.1 sec                      4.1 sec
    20000     50     0.9 sec (0.6 .. 0.9 sec)     7.5 sec
    25000     40     1.0 sec (0.7 .. 1.0 sec)     9.3 sec
    30303     33     0.7 sec (0.7 .. 1.2 sec)    11.2 sec
    40000     25     1.4 sec (0.8 .. 2.6 sec)    14.4 sec
    50000     20     3.2 sec (1.1 .. 3.3 sec)    17.6 sec
    100000    10     4.6 sec (0.9 .. 7.7 sec)    34.7 sec 
    166667     6    14   sec (1.1 .. 18  sec)    57   sec (56.8 .. 57.6 sec)
    200000     5    12   sec (3   .. 24  sec)    Упс! Cannot create OS thread
    250000     4     7   sec (1.3 .. 13  sec)    ----
    333333     3    40   sec (2.4 .. 48  sec)    ----
    500000     2     2.1 sec                     ----
    1000000    1     4.2 sec                     ----

Иногда (очень редко) шедулер распределяет потоки так, что они вообще не конфликтуют, и время выполнения минимально (0.7 сек), а иногда (тоже редко) начинается своппинг и задачу приходится снимать...
Когда количество потоков больше ста тысяч, то начинаются проблемы на моём железе — большой разброс времени работы. Ни о каком soft real time явно речи идти не может.

С процессами forkOS картинка немного другая — с одной стороны, очень повторяемые результаты, а с другой стороны — большие накладные расходы при количестве процессов от нескольких тысяч и более. Больше 170 тыс. процессов вообще создать не удалось...

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

Вот интересно, как такая программа выглядела бы на Эрланге и Немерле (на макросах), и какие характеристики у этих вариантов...

для erlang на celeron 1.5, 256mb, erlang-1:10.b.7-1, в своп не уходил.
но сравнение с хаскелем для примера ниже не вполне корректно — без глобальных переменных передача тика идет через сообщение, что при 1млн потоков и мгновенном завершении каждого из них создает огромную очередь в ожидании приема сообщений.
41> rsdn:test(1000000,1).
{ok,30.2363}
42> rsdn:test(1000000,1).
{ok,28.1009}
49> rsdn:test(1,1000000).
{ok,4.58340e-2}
50> rsdn:test(10,100000).
{ok,4.74110e-2}
51> rsdn:test(100,10000).
{ok,4.81870e-2}
53> rsdn:test(1000,1000).
{ok,0.515728}
54> rsdn:test(1000,1000).
{ok,5.49930e-2}
56> rsdn:test(5000,200).
{ok,9.18940e-2}
57> rsdn:test(10000,100).
{ok,0.164270}
58> rsdn:test(10000,100).
{ok,0.140982}
59> rsdn:test(10000,100).
{ok,0.154148}
60> rsdn:test(100000,10).
{ok,1.17681}
61> rsdn:test(100000,10).

%% запуск для 100 тысяч потоков и 10 тиков: rsdn:test(100000,10).
-module(rsdn).
-compile(export_all).

test(NThreads, NTicks) ->
    T1 = now(),
    go(NThreads, NTicks),
    Var = wait(NThreads),
    T2 = now(),
    {Var, timer:now_diff(T2,T1)/1000000}.

wait(0) ->
    ok;
wait(NThreads) ->
    receive 
        {tick, _Var} ->
%%            io:format("~p~n",[_Var]),
            wait(NThreads-1)
    end.
        

go(0, _Ticks) ->
    ok;
go(NThreads, Ticks) ->
    spawn(?MODULE, tick, [self(), 0, Ticks]),
    go(NThreads-1, Ticks).
    
    
tick(Pid, Var, 0) ->
    Pid ! {tick, Var};
tick(Pid, Var, Ticks) ->
    tick(Pid, Var + 1, Ticks-1).

кстати, если запустить процедуру ожидания в отдельный поток — можно добиться ускорения работы в 3-4 раза на большом количестве потоков. но все равно разница между 1000000, 1 и 1,1000000 — 2-3 порядка.
68> rsdn:test(1000000,1).
{ok,8.79680}
69> rsdn:test(1000000,1).
{ok,8.77959}
70> rsdn:test(1,1000000).
{ok,4.59500e-2}
71> rsdn:test(1,1000000).
{ok,4.96880e-2}
исправленный код:

-module(rsdn).
-compile(export_all).
test(NThreads, NTicks) ->
    T1 = now(),
    Pid = spawn(?MODULE, wait, [NThreads, self()]),
    go(NThreads, NTicks, Pid),
    receive {wait, ok} -> ok
    end,
    T2 = now(),
    {ok, timer:now_diff(T2,T1)/1000000}.
wait(0, Pid) ->
    Pid ! {wait, ok};
wait(NThreads, Pid) ->
    receive 
        {tick, _Var} ->
            wait(NThreads-1, Pid)
    end.
go(0, _Ticks, _Pid) ->
    ok;
go(NThreads, Ticks, Pid) ->
    spawn(?MODULE, tick, [Pid, 0, Ticks]),
    go(NThreads-1, Ticks, Pid).
tick(Pid, Var, 0) ->
    Pid ! {tick, Var};
tick(Pid, Var, Ticks) ->
    tick(Pid, Var + 1, Ticks-1).

и еще — время в основном тратится на создание потоков, а не на вычисление. т.е. на реальных задачах большое число "легких" потоков таки оправдано

Здравствуйте, geniepro, Вы писали:

Да, разброс с параметрами 333333 3 — 1..40 сек.
Но есть еще интересная опция линковки GHC -threaded. После компиляции с ее применением в RTS появляются дополнительные опции:

-N<n>     Use <n> OS threads (default: 1)
-qw       Migrate a thread to the current CPU when it is woken up

Если запускать так: +RTS -N2 -qw -RTS, то стабильно получается 2.5 сек. Это всё для версии с forkIO.

Здравствуйте, Аноним, Вы писали:

А>кстати, если запустить процедуру ожидания в отдельный поток — можно добиться ускорения работы в 3-4 раза на большом количестве потоков. но все равно разница между 1000000, 1 и 1,1000000 — 2-3 порядка.

Да, у Эрланга, похоже, действительно очень лёгкие процессы (по сравнению с Хаскеллем)...

Здравствуйте, geniepro, Вы писали:

G>Да, у Эрланга, похоже, действительно очень лёгкие процессы (по сравнению с Хаскеллем)...

Он ради этого разрабатывался.

... << RSDN@Home 1.2.0 alpha rev. 637>>

Здравствуйте, Аноним, Вы писали:

A> сравнение с хаскелем для примера ниже не вполне корректно — без глобальных переменных передача тика идет через сообщение, что при 1млн потоков и мгновенном завершении каждого из них создает огромную очередь в ожидании приема сообщений

Окей, я переписал программу так, что вместо мутабельных переменных (локальных, кстати) используется обмен сообщениями через три канала Chan: ticker, finish и result. Если я правильно понял программу на Эрланге — принцип примерно такой же.

Программа создаёт три канала и соответственно процессы для работы с этими каналами:

1) Процесс tick считывает с канала ticker сообщения типа Ticker. В случае сообщения Tick происходит увеличение на единицу значения параметра k функции tick; сообщение Done сигнализирует об окончании работы — пора выдавать результат через канал result.

2) Процесс wait считывает с канала finish сигнал о том, что очередная копия процесса foo завершила свою работу. Тип сигнала неважен — принимаем тип () — что-то типа unit из F#/OCaml...
Как только количество полученных сигналов finish станет равно количеству запущенных копий процесса foo (что означает завершение всех копий foo) — в канал ticker будет передано сообщение Done.

3) Функция makeThreads запускает нужное количество (n_threads) копий процесса foo.

4) Процессы foo просто посылают в канал ticker сообщения Tick в количестве n_incs штук, а затем — сигнал () в канал finish.

5) Ну и наконец, после создания всех нужных процессов, основной процесс (функция main просто ждёт, когда ему кто-нибудь пришлёт результат в канал result, который затем и распечатает... :о)

import IO
import Time
import System
import Control.Concurrent
import Control.Concurrent.Chan

data Ticker = Tick | Done

n_threads',          n_incs' :: Int
n_threads'  = 1000;  n_incs' = 1000

-- Командная строка: имя.exe +RTS -K100M -RTS кол-во_потоков кол-во_инкрементов

main = do
    arg <- getArgs
    let (n_threads, n_incs) = case (map read arg) of
            nt:ni:_ -> (nt,         ni)
            _       -> (n_threads', n_incs')

    t1  <- Time.getClockTime
    hSetBuffering stdout NoBuffering
    putStr $ "Config: " ++ show n_threads ++ " by " ++ show n_incs ++ "  Starting... "

    ticker <- newChan
    finish <- newChan
    result <- newChan

    let foo 0 = finish `writeChan` ()
        foo n = ticker `writeChan` Tick >> foo (n-1)

        makeThreads 0 = return ()
        makeThreads n = forkIO (foo n_incs) >> makeThreads (n-1)

        tick :: Int -> IO ()
        tick k = do msg <- readChan ticker
                    case msg of
                        Tick -> tick (k+1)
                        Done -> result `writeChan` k

        wait 0 = ticker `writeChan` Done
        wait n = readChan finish >> wait (n-1)

    forkIO     (tick 0)
    forkIO     (wait n_threads)
    makeThreads n_threads

    putStr "Finishing... "
    val <- readChan result
    putStr $ "Done: " ++ show val
    t2  <- Time.getClockTime; printTimeDif t2 t1

printTimeDif ft st = putStrLn $ " Time = " ++ show(Time.tdHour td) ++ ":" ++ show(Time.tdMin td) ++ ":"
                                           ++ show nsecs           ++ "." ++ snms
  where
    td           = ft `Time.diffClockTimes` st
    secs         = Time.tdSec     td
    ms           = Time.tdPicosec td `div` 1000000000
    (nsecs, nms) = if ms < 0 then (secs - 1, ms + 1000) else (secs, ms)
    snms         = reverse $ take 3 $ (reverse $ show nms) ++ "000"

Результаты, честно говоря, удручающие. Результаты привожу только для версии с forkIO, потому что с forkOS результаты вроде не лучше. Запускал с параметрами +RTS -K100M -RTS кол-во_потоков кол-во_инкрементов

n_threads n_incs      время  
    1    1000000     1.2 sec 
    10    100000     2.3 sec 
    100    10000     2.6 sec 
    1000    1000     3.1 sec 
    5000     200     8.1 sec    
    10000    100    10.6 sec    
    25000     40    18.7 sec 
    50000     20    24.7 sec 
    100000    10    40.4 sec 
    250000     4  ~200   sec (своппинг)

следовательно, вывод — на большом количестве потоков erlang рвет своих конкурентов. что и требовалось доказать

. а haskell, похоже, скоро станет чем-то вроде common lisp — есть всё, что можно себе представить, пусть и не в лучшей реализации

Здравствуйте, Аноним, Вы писали:

А>следовательно, вывод — на большом количестве потоков erlang рвет своих конкурентов. что и требовалось доказать

. а haskell, похоже, скоро станет чем-то вроде common lisp — есть всё, что можно себе представить, пусть и не в лучшей реализации

Да в принципе, конкуренты Эрланга представлены-то и не были. Хаскелл — не конкурент в этих делах, да и не пытается вроде им стать (пока) :о)
Вот любопытно, что даст Хаскеллу Nested Data Parallelism, когда будет реализован... Но опять же, это параллельная обработка данных, а не конкурентная...

Однако... http://shootout.alioth.debian.org/gp4/benchmark.php?test=message&lang=all

    x     Program & Logs     Full CPU Time s   Memory Use KB   GZip Bytes

    1.0   Mozart/Oz                 3.02           4,312          360
    1.7   Erlang HiPE #2            5.19           5,088          263
    2.0   Haskell GHC               5.95           3,596          236
    2.3   Forth bigForth            6.84           1,076          328
    3.3   C gcc #2                  9.87           1,048          525
    4.1   SML MLton                12.25           3,656          422
    4.4   Python                   13.41           2,656          160
    6.6   Scala #2                 19.94         146,060          380
    6.8   D Digital Mars #2        20.58           2,772          423
    7.3   Lua #2                   22.13           1,476          229
    8.9   SML MLton #2             26.99           5,892          319
    9.7   Pascal Free Pascal       29.33           2,872          718
   14     Smalltalk VisualWorks    40.84          18,368          487
   18     Ada 95 GNAT              52.95           7,540          452
   18     C gcc                    53.32           2,512          519
   18     C++ g++ #2               54.09           2,964          894
   22     Ruby #2                  66.70           3,956          149
   37     Java JDK -server #2     112.25          25,176          527
   59     Scheme MzScheme         178.75          10,524          279
   72     Pike                    217.77          10,540          226
   78     Tcl #2                  236.59          56,212          227
   98     Scala                   295.39          46,160          400
1,237     Ruby                  3,739.18          10,032          177

Хаскелл на третьем месте — не так уж и плохо, лишь чуть-чуть хуже, чем у Эрланга, хотя оба они слили Моцарт/Озу... :о))

Всё таки я не смог успокоиться и сделал третий вариант программы, заменив первоначальные MVar Int на IORef Int. А что бы конкурентные потоки не внесли хаос одновременным изменением IORef-переменной, операции над ними атомарными.

import IO
import Time
import System
import Data.IORef
import Control.Concurrent

n_threads',         n_incs' :: Int
n_threads' = 1000;  n_incs' = 1000

-- Командная строка: имя.exe +RTS -K100M -RTS кол-во_потоков кол-во_инкрементов

main = do
    arg <- getArgs
    let (n_threads, n_incs) = case (map read arg) of
            nt:ni:_ -> (nt,         ni)
            _       -> (n_threads', n_incs')

    t1  <- Time.getClockTime
    hSetBuffering stdout NoBuffering
    putStr $ "Config: " ++ show n_threads ++ " by " ++ show n_incs ++ " Starting... "

    var <- newIORef 0    -- Общий счётчик
    cnt <- newIORef 0    -- Число активных потоков

    let foo 0 = cnt -:= 1
        foo n = var +:= 1 >> foo (n-1)

        makeThreads 0 = return ()
        makeThreads n = cnt +:= 1 >> forkIO (foo n_incs) >> makeThreads (n-1)

        waitFor0 = do cnt' <- readIORef cnt
                      if cnt' == 0 then return ()
                                   else threadDelay 40000 >> waitFor0
    makeThreads n_threads

    putStr "Finishing... "

    waitFor0

    val <- readIORef var
    putStr $ " Done: " ++ show val
    t2  <- Time.getClockTime; printTimeDif t2 t1
  where
    (+:=), (-:=) :: IORef Int -> Int -> IO ()
    var -:= n = var +:= (-n)

    var +:= n = var `atomicModifyIORef` (\x -> (x+n, ()))

printTimeDif ft st = putStrLn $ "  Time = " ++ show(Time.tdHour td) ++ ":" ++ show(Time.tdMin td) ++ ":"
                                            ++ show nsecs           ++ "." ++ snms
  where
    td           = ft `Time.diffClockTimes` st
    secs         = Time.tdSec     td
    ms           = Time.tdPicosec td `div` 1000000000
    (nsecs, nms) = if ms < 0 then (secs - 1, ms + 1000) else (secs, ms)
    snms         = reverse $ take 3 $ (reverse $ show nms) ++ "000"

Что порадовало — результаты стали очень стабильными и в целом — неплохими... :о)
Похоже, в многопоточных программах не стоит игнорировать IORef'ы, если это позволяет задача...

n_threads  n_incs      время  
  1       1000000     2.4 sec 
  10       100000     2.4 sec 
  100       10000     2.4 sec 
  1000       1000     2.5 sec 
  5000        200     2.6 sec    
  10000       100     2.6 sec    
  25000        40     2.7 sec 
  50000        20     2.7 sec 
  100000       10     3.0 sec
  250000        4     4.0 sec
  333333        3     4.4 sec
  500000        2     5.7 sec
  1000000       1    10.7 sec

Здравствуйте, geniepro, Вы писали:

G>Всё таки я не смог успокоиться и сделал третий вариант программы, заменив первоначальные MVar Int на IORef Int. А что бы конкурентные потоки не внесли хаос одновременным изменением IORef-переменной, операции над ними атомарными.

G>Что порадовало — результаты стали очень стабильными и в целом — неплохими... :о)
G>Похоже, в многопоточных программах не стоит игнорировать IORef'ы, если это позволяет задача...

Мне не совсем понятно, что именно измеряет твой бенчмарк. Все-таки тысячи потоков, непрерывно модифицирующие одну глобальную переменную — довольно странный юз-кейс, понятно, что все время будет тратиться на синхронизацию. Надо бы придумать какой-то бенчмарк пореалистичнее. А пока, с вот такой модификацией:

foo n = do when (n `mod` 20==0) $ var +:= 1 
           foo (n-1)

получается такая картинка:

Config: 1 by 1000000 Starting... Finishing...  Done: 50000  Time = 0:0:0.046
Config: 10 by 100000 Starting... Finishing...  Done: 50000  Time = 0:0:0.047
Config: 100 by 10000 Starting... Finishing...  Done: 50000  Time = 0:0:0.078
Config: 1000 by 1000 Starting... Finishing...  Done: 50000  Time = 0:0:0.078
Config: 5000 by 200 Starting... Finishing...  Done: 50000  Time = 0:0:0.062
Config: 10000 by 100 Starting... Finishing...  Done: 50000  Time = 0:0:0.109
Config: 25000 by 40 Starting... Finishing...  Done: 50000  Time = 0:0:0.110
Config: 50000 by 20 Starting... Finishing...  Done: 50000  Time = 0:0:0.141
Config: 100000 by 10 Starting... Finishing...  Done: 0  Time = 0:0:0.218
Config: 250000 by 4 Starting... Finishing...  Done: 0  Time = 0:0:0.594
Config: 333333 by 3 Starting... Finishing...  Done: 0  Time = 0:0:0.875
Config: 500000 by 2 Starting... Finishing...  Done: 0  Time = 0:0:1.501
Config: 1000000 by 1 Starting... Finishing...  Done: 0  Time = 0:0:4.250

Здравствуйте, palm mute, Вы писали:

PM> Мне не совсем понятно, что именно измеряет твой бенчмарк. Все-таки тысячи потоков, непрерывно модифицирующие одну глобальную переменную — довольно странный юз-кейс, понятно, что все время будет тратиться на синхронизацию.

Да меня, вобщем-то, именно накладные расходы на синхронизацию и интересовали, поэтому никакой сложной логики я и не закладывал — просто граничный случай...
А расходы на изменение самих IORef'ов и MVar'ов (в однопоточном режиме) вроде ненамного больше, чем на изменение аргументов функций: IORef — примерно в два раза медленнее, MVar — в 2.7 раза дольше...

PM> А пока, с вот такой модификацией:
PM> foo n = do when (n `mod` 20==0) $ var +:= 1
PM> foo (n-1)
PM> получается такая картинка:

Так тут уже гораздо (в 20 раз) меньше операций над переменной var получается — понятно, что меньше конфликтов, и программа быстрее работает...

Кстати, что за when такой? Я порылся по библиотеке — не нашёл... Симитировал так:

when c a = if c then a else return ()

От:	geniepro	http://geniepro.livejournal.com/
Дата:	12.05.07 23:51
Оценка:	38 (5)

	От:	Аноним
	Дата:	13.05.07 07:51
	Оценка:

	От:	Аноним
	Дата:	13.05.07 08:30
	Оценка:

	От:	Аноним
	Дата:	13.05.07 08:38
	Оценка:

	От:	R.K.
	Дата:	13.05.07 09:09
	Оценка:	27 (4)

От:	VladD2	www.nemerle.org
Дата:	14.05.07 01:06
Оценка:

От:	lomeo	http://lomeo.livejournal.com/
Дата:	16.05.07 14:30
Оценка: