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

我在Racket小组中找到了一个关于创建性能的主题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,会出现错误

致命错误:异常Sys_error("Thread.create:资源暂时不可用")

我用这个命令编译并运行

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

ivg.. 9

OCaml Thread模块使用真实系统(内核)线程.线程总数受内核限制:

 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库中的协作轻量级线程.这将是非常有效和好的.

这个实现紧跟着博客文章中的Go实现,但我认为我们可以在不使用邮箱的情况下更有效地在OCaml中实现这一点(但我不确定它是否符合基准规则).

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

上面的代码是原始Go版本的完全诚实的重新实现.但是它如此之快的原因之一是OCaml和Lwt非常聪明,虽然它创建了100_000线程和100_001通道,但是没有任何线程可以产生到后台,因为每次whisper被调用时,通道都已包含数据,所以线程处于就绪状态.因此,这只是一个有效的循环,可以创建线程和通道.它可以在50毫秒内创建数百万个线程.

所以这是一种习惯性和正确的做事方式.但为了真实比较mimick Go行为.以下实现将首先急切地在堆100_001通道和100_000个线程中创建,等待从左到右通道传输数据.然后它才会在最左边的通道中放置一个值来激发一连串的反应.这基本上可以模仿引擎盖下发生的事情.

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倍(我使用了100万个线程来比较它们),但它仍然比Go快10倍.

OCaml Thread模块使用真实系统(内核)线程.线程总数受内核限制:

 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库中的协作轻量级线程.这将是非常有效和好的.

这个实现紧跟着博客文章中的Go实现,但我认为我们可以在不使用邮箱的情况下更有效地在OCaml中实现这一点(但我不确定它是否符合基准规则).

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

上面的代码是原始Go版本的完全诚实的重新实现.但是它如此之快的原因之一是OCaml和Lwt非常聪明,虽然它创建了100_000线程和100_001通道,但是没有任何线程可以产生到后台,因为每次whisper被调用时,通道都已包含数据,所以线程处于就绪状态.因此,这只是一个有效的循环,可以创建线程和通道.它可以在50毫秒内创建数百万个线程.

所以这是一种习惯性和正确的做事方式.但为了真实比较mimick Go行为.以下实现将首先急切地在堆100_001通道和100_000个线程中创建,等待从左到右通道传输数据.然后它才会在最左边的通道中放置一个值来激发一连串的反应.这基本上可以模仿引擎盖下发生的事情.

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倍(我使用了100万个线程来比较它们),但它仍然比Go快10倍.