how to do the concurrent version of List.map in ocaml?

Question

I am trying to implement the concurrent version of List.map function in Ocaml. I know there is a similar function in module Async.Std.Deferred.List, but here I am just trying to attack this myself.

Here is what I have:

let deferred_listmap (f : 'a -> 'b Deferred.t) (l : 'a list) =
let f' acc elem =
    (f elem) >>= fun s ->
    return (acc@[s]) in
List.fold_left f' [] l

And obviously there is a type matching error here, since fold_left requires a function f with type ('b list -> 'a -> 'b list), while the f' I have here is of type ('b list -> 'a -> 'b list Deferred.t). But for this I cannot help since I am using a bind(>>=) here which expects the return type of its function argument to be a deferred type. So I am getting kind of a contradiction here, and I cannot think of other uses on bind or folds or other functions in List module to solve this. Any suggestions?

if you are using libraries coming separately from the Ocaml compiler, perhaps you should indicate that? — didierc
Now, if you are using Async utilities to reimplement this function, I suppose that Async.Std.Deferred.all would be handy (and is most certainly what the real function uses). That function in turn may be implemented using a fold of both (in the same module) on the list. — didierc
Obviously it depends what do you mean with concurrent. Should every map be done on another thread? (Like the example from @Oleg) Or should it be done in chunks of n items? In order to answer more details are needed — denis631

Oleg Oleg · Accepted Answer · 2017-04-01T20:53:13

I guess you could try with the Thread and Event Modules .

open Thread
open Event

let tapply f ls = 
    let res = 
        List.map (fun e ->  
        let c = new_channel () in 
        let _ = create (fun c -> send c (f e) |> sync) c in
        receive c) ls in
        List.map sync res

example :

# tapply (fun e -> delay 3.0; e*e) [17;573;2;6;3];;
- : int list = [289; 328329; 4; 36; 9] (*waiting time only 3 sec, not 15*)

how to do the concurrent version of List.map in ocaml?

1 Answers