Trouble using async_parallel to distribute computation

[Alex M]

Hi all,

I'm trying to use Async_parallel to distribute my computation across multiple threads/cores, following the tutorial found here: https://blogs.janestreet.com/async-parallel.  However, I don't think I'm doing it right, because even though two child processes are created, they don't appear to be running in parallel (e.g., I would expect the "Got Ping" messages to be interleaved and use different client IDs).  Please see the below code (or https://gist.github.com/amaloz/7cd0bbde736b0a06db4d).  Any ideas of what I'm doing wrong?  Thanks for the help!

Alex

P.S. Is this the best place to post ocaml-core questions?

Code:

open Core.Std
open Async.Std
open Async_parallel.Std

let workers = ["localhost"; "localhost"]

let worker h =
  Pipe.iter_without_pushback (Hub.listen_simple h) ~f:(fun (id, `Ping) ->
      printf "%s Got Ping\n%!" (Hub.Client_id.sexp_of_t id |> Sexp.to_string_hum);
      Hub.send h id `Pong)
  >>| fun () -> `Done
 
let main () =
  printf "Start\n%!";
  Deferred.List.iter [1; 2; 3; 4] ~f:(fun i ->
      Parallel.spawn ~where:Parallel.round_robin worker
      >>= fun (c, res) ->
      printf "Value = %d\n%!" i;
      let rec loop = function
        | 0 -> Channel.close c
        | i ->
        Channel.write c `Ping;
        Channel.read c
        >>= fun `Pong ->
        Clock.after (sec 1.)
        >>= fun _ -> loop (i - 1)
      in
      loop 2
    )
  >>> fun _ -> Shutdown.shutdown 0

let _ =
  Parallel.init ~cluster:{ Cluster.master_machine = Unix.gethostname ();
                           worker_machines = workers } ();
  main ();
  never_returns (Scheduler.go ())

[Ben Millwood]

I would expect the client ID you're printing there to be the client ID of the process that sent the `Ping, i.e. the main process, which would explain why it's always the same. Try printing out . I can't explain the interleaving, but it's not 100% compelling evidence either. I don't know how a Hub works, but it's entirely plausible that it iterates over connected machines in some fixed order.

I would also recommend that you consider using Rpc_parallel in preference to Async_parallel, but unfortunately it looks like we don't have any generated docs for it. If you're feeling bold you can go ahead and dive into the mli's, and I hope we'll have a fixed documentation generator soon.

--
You received this message because you are subscribed to the Google Groups "ocaml-core" group.
To unsubscribe from this group and stop receiving emails from it, send an email to ocaml-core+...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Valentin Gatien-Baron]

The lack of interleaving may be because you use [Deferred.List.iter] instead of [Deferred.List.iter ~how:`Parallel].

[Alex Malozemoff]

Yes, that did the trick. Thanks!

[Ben Millwood]

On 12 January 2015 at 16:04, Ben Millwood <bmil...@janestreet.com> wrote:

Try printing out .

Oh, I clearly left a space here to go look up what you should be printing again, and then forgot to do so. After digging about a bit I'm going to stick to my recommendation of Rpc_parallel over Async_parallel. In the long run it'll be less painful :)

[Yaron Minsky]

I'd also encourage you to look at Rpc_parallel, which we now tend to
prefer over Async_parallel. There's a blog post describing this one
too.

https://blogs.janestreet.com/what-the-interns-have-wrought-rpc_parallel-and-core_profiler/

It's a bit easier to think about because it uses more explicit
serialization techniques, rather than OCaml's built in marshal, which
can be hard to think about in the presence of non-functional values
and values like closures that can be backed by all sorts of hard to
track data.

y

[Alex Malozemoff]

Thanks, I'll check it out. Although right now I can't get it to
install using opam or building from github. I get the following
error:

File "lib_core/parallel.ml", line 335, characters 68-74:
Error: This expression has type
([< Async_extra.Import.Socket.Address.t ] as 'a) -> unit
but an expression was expected of type
'a -> Async.Std.Rpc.Connection.t -> unit
Type unit is not compatible with type
Async.Std.Rpc.Connection.t -> unit

I don't have time right away to investigate this, and this is probably
not the right place to report such errors anyways, but if anyone knows
a quick-fix to the above that'd be great! Otherwise, I'll look into
it later today. (I'm using all the latest packages on opam and ocaml
v4.02.1)

[Ben Millwood]

Hmm. I don't see any errors when I build from opam. The right place to report is to https://github.com/janestreet/async_parallel/issues ; could you give as much detail as you can there?