如何在OCaml中创建大量的线程？

Question

我在球拍组中发现了topic关于创建channel的性能。

我想写一个OCaml的版本来测试。

let post (c,x) = Event.sync (Event.send c x);; 

let accept c = Event.sync (Event.receive c);; 

let get_chan c = let n = accept c in print_int n;print_newline();; 

let chan_trans (old_chan, new_chan) = 
    let s = accept old_chan in 
    post (new_chan,(s+1));; 

let rec whisper count init_val = 
    let rec aux n chan = 
    if n >= count then chan 
    else 
     let new_chan = Event.new_channel() 
     in Thread.create chan_trans (chan, new_chan); 
     aux (n+1) new_chan 
    in let leftest_chan = Event.new_channel() 
    in let t0 = Thread.create post (leftest_chan, init_val) 
    in let rightest_chan = aux 0 leftest_chan 
    in get_chan rightest_chan;; 

whisper 10000 1;; 

的问题是，当我测试whisper 1000 1，它产生1001按预期方式。然而，当我试图测试whisper 10000 1，有作为

Fatal error: exception Sys_error("Thread.create: Resource temporarily unavailable")

我用这个命令来编译和运行

ocamlc -thread unix.cma threads.cma -o prog whisper.ml&&./prog -I +threads

Answer 1

OCaml的主题模块使用真正的系统（内核）线程错误。线程的总量是由内核为界：

cat /proc/sys/kernel/threads-max 
251422 

您可以增加此当然

echo 100000 > /proc/sys/kernel/threads-max 

，但更好的方法是治疗线程作为一种资源和管理相应。

let rec whisper count init_val = 
    let rec aux n t chan = 
    if n >= count then chan 
    else 
     let new_chan = Event.new_channel() in 
     let t' = Thread.create chan_trans (chan, new_chan) in 
     Thread.join t; 
     aux (n+1) t' new_chan in 
    let leftest_chan = Event.new_channel() in 
    let t = Thread.create post (leftest_chan, init_val) in 
    let rightest_chan = aux 0 t leftest_chan in 
    get_chan rightest_chan 

在这种情况下，它将以任意大小的管道运行。例如：

$ ocamlbuild -use-ocamlfind -tag thread -pkg threads ev.native 
$ time ./ev.native 
100001 

real 0m1.581s 

但是这种中文语言的实现是非常粗糙和低效的。你不应该为此使用重量级本地线程（并且都不会使用它们）。相反，您应该使用Lwt或Async库中的轻量级协作线程。这将是非常有效的，很好。

实现与LWT

此实现密切关注从blog post转到实现，但我认为我们可以做到这一点OCaml中更加高效和conscise不使用的邮箱（但我不知道它是否会符合到基准的规则）。

open Lwt.Infix 

let whispers n = 
    let rec whisper i p = 
    if i < n then 
     Lwt_mvar.take p >>= fun x -> 
     whisper (i+1) (Lwt_mvar.create (x+1)) 
    else Lwt_mvar.take p in 
    whisper 0 (Lwt_mvar.create 1) 

let() = print_int @@ Lwt_main.run (whispers 100000) 

的结果是：

$ ocamlbuild -use-ocamlfind -tag thread -pkg lwt.unix lev.native -- 
$ time ./lev.native 
100001 
real 0m0.007s 

要与我的机器上的Go执行比较：

$ go build whispers.go 
$ time ./whispers 
100001 

real 0m0.952s 

“慢” 的实施

上面的代码是完全地诚实reimplemtation原来的Go版本。但是它的速度如此之快的原因之一就是OCaml和Lwt非常聪明，虽然它创建了100_000线程和100_001通道，但是没有线程被放弃到后台，因为每次调用whisper时，通道已经包含数据，所以线程处于就绪状态。因此，这只是一个有效的循环，可以创建线程和通道。它可以在50毫秒内创建一百万个线程。

所以这是一种习惯和正确的做事方式。但让我们为了真正的比较模仿Go行为。以下实现将 首先在堆100_001中首先热切地创建通道，并且100_000个线程等待从左向右通道传输数据。只有在此之后，它才会为最左边的通道添加一个值，以引发一连串的反应。这基本上可以模仿Go下面发生的事情。

let whispers n = 
    let rec loop i p = 
    if i < n then 
     let p' = Lwt_mvar.create_empty() in 
     let _t = 
     Lwt_mvar.take p >>= fun x -> 
     Lwt_mvar.put p' (x+1) in 
     loop (i+1) p' 
    else Lwt_mvar.take p in 
    let p0 = Lwt_mvar.create_empty() in 
    let t = loop 1 p0 in 
    Lwt_mvar.put p0 1 >>= fun() -> t 

$ time ./lev.native 
100001 
real 0m0.111s 

所以这是稍微慢一些，实际上它是比以前慢执行20次（我用百万线程对它们进行比较），但它比走得快还是10倍。

Answer 2

阅读链接的帖子，似乎你可能想要使用lwt这是一个“OCaml合作线程库”。其结果将是这个样子：

let whisper left right = 
    let%lwt n = Lwt_mvar.take right in 
    Lwt_mvar.put left (n+1) 

let main() = 
    let n = 100_000 in 
    let%lwt() = Lwt_io.printf "With %d mvars!\n" n in 
    let leftmost = Lwt_mvar.create_empty() in 
    let rec setup_whispers left i = 
    if i >= n 
    then left 
    else let right = Lwt_mvar.create_empty() in 
     let() = Lwt.async (fun() -> whisper left right) in 
     setup_whispers right (i+1) in 
    let rightmost = setup_whispers leftmost 0 in 
    let%lwt() = Lwt_mvar.put rightmost 1 in 
    let%lwt res = Lwt_mvar.take leftmost in 
    Lwt_io.printf "%d\n" res 

let() = Lwt_main.run (main()) 

然后编译并运行它

$ ocamlbuild -use-ocamlfind -pkg lwt,lwt.ppx,lwt.unix whisper.native 
$ time ./whisper.native 
With 100000 mvars! 
100001 

real 0m0.169s 
user 0m0.156s 
sys 0m0.008s