package rpc_parallel

You can search for identifiers within the package.

in-package search v0.2.0

On This Page

Map functions
Map-reduce

rpc_parallel
- Rpc_parallel
  - Executable_location
  - Fd_redirection
  - Map_reduce
    
    Config
    
    Make_map_function
    
    Input
    
    Output
    
    Param
    
    S
    
    Input
    
    Output
    
    Worker
    
    Make_map_function_with_init
    
    Input
    
    Output
    
    Param
    
    S
    
    Input
    
    Output
    
    Param
    
    Worker
    
    Make_map_reduce_function
    
    Accum
    
    Input
    
    Param
    
    S
    
    Accum
    
    Input
    
    Worker
    
    Make_map_reduce_function_with_init
    
    Accum
    
    Input
    
    Param
    
    S
    
    Accum
    
    Input
    
    Param
    
    Worker
    
    Map_function
    
    Input
    
    Output
    
    Param
    
    Worker
    
    Map_function_spec
    
    Input
    
    Output
    
    Map_function_with_init_spec
    
    Input
    
    Output
    
    Param
    
    Map_reduce_function
    
    Accum
    
    Input
    
    Param
    
    Worker
    
    Map_reduce_function_spec
    
    Accum
    
    Input
    
    Map_reduce_function_with_init_spec
    
    Accum
    
    Input
    
    Param
    
    Worker
  - Parallel
    
    Creator
    
    Expert
    
    Worker_env
    
    For_testing
    
    Function
    
    Direct_pipe
    
    Functions
    
    Make
    
    Connection
    
    Deprecated
    
    For_internal_testing
    
    Spawn_in_foreground_result
    
    Id
    
    Hash_queue
    
    Hash_set
    
    Provide_bin_io
    
    X
    
    Provide_of_sexp
    
    X
    
    Map
    
    Key
    
    Provide_bin_io
    
    Key
    
    Provide_hash
    
    Key
    
    Provide_of_sexp
    
    Key
    
    Tree
    
    Provide_of_sexp
    
    K
    
    Replace_polymorphic_compare
    
    Set
    
    Elt
    
    Named
    
    Provide_bin_io
    
    Elt
    
    Provide_hash
    
    Elt
    
    Provide_of_sexp
    
    Elt
    
    Tree
    
    Named
    
    Provide_of_sexp
    
    Elt
    
    Table
    
    Provide_bin_io
    
    Key
    
    Provide_of_sexp
    
    Key
    
    S
    
    Connection_state
    
    Functions
    
    C
    
    Worker_state
    
    Shutdown_on
    
    M
    
    Spawn_in_foreground_exn_result
    
    Spawn_in_foreground_result
    
    State
    
    Worker
    
    Connection
    
    Deprecated
    
    For_internal_testing
    
    Spawn_in_foreground_result
    
    Id
    
    Hash_queue
    
    Hash_set
    
    Provide_bin_io
    
    X
    
    Provide_of_sexp
    
    X
    
    Map
    
    Key
    
    Provide_bin_io
    
    Key
    
    Provide_hash
    
    Key
    
    Provide_of_sexp
    
    Key
    
    Tree
    
    Provide_of_sexp
    
    K
    
    Replace_polymorphic_compare
    
    Set
    
    Elt
    
    Named
    
    Provide_bin_io
    
    Elt
    
    Provide_hash
    
    Elt
    
    Provide_of_sexp
    
    Elt
    
    Tree
    
    Named
    
    Provide_of_sexp
    
    Elt
    
    Table
    
    Provide_bin_io
    
    Key
    
    Provide_of_sexp
    
    Key
    
    Shutdown_on
    
    M
    
    Spawn_in_foreground_exn_result
    
    Spawn_in_foreground_result
    
    Worker_spec
    
    Connection_state
    
    Functions
    
    C
    
    Worker_state
  - Parallel_managed
    
    Make
    
    Id
    
    Hash_queue
    
    Hash_set
    
    Provide_bin_io
    
    X
    
    Provide_of_sexp
    
    X
    
    Map
    
    Key
    
    Provide_bin_io
    
    Key
    
    Provide_hash
    
    Key
    
    Provide_of_sexp
    
    Key
    
    Tree
    
    Provide_of_sexp
    
    K
    
    Replace_polymorphic_compare
    
    Set
    
    Elt
    
    Named
    
    Provide_bin_io
    
    Elt
    
    Provide_hash
    
    Elt
    
    Provide_of_sexp
    
    Elt
    
    Tree
    
    Named
    
    Provide_of_sexp
    
    Elt
    
    Table
    
    Provide_bin_io
    
    Key
    
    Provide_of_sexp
    
    Key
    
    S
    
    Connection_state
    
    Functions
    
    C
    
    Worker_state
    
    Worker
    
    Id
    
    Hash_queue
    
    Hash_set
    
    Provide_bin_io
    
    X
    
    Provide_of_sexp
    
    X
    
    Map
    
    Key
    
    Provide_bin_io
    
    Key
    
    Provide_hash
    
    Key
    
    Provide_of_sexp
    
    Key
    
    Tree
    
    Provide_of_sexp
    
    K
    
    Replace_polymorphic_compare
    
    Set
    
    Elt
    
    Named
    
    Provide_bin_io
    
    Elt
    
    Provide_hash
    
    Elt
    
    Provide_of_sexp
    
    Elt
    
    Tree
    
    Named
    
    Provide_of_sexp
    
    Elt
    
    Table
    
    Provide_bin_io
    
    Key
    
    Provide_of_sexp
    
    Key
  - Remote_executable
  - Std

Legend:
Library
Module
Module type
Parameter
Class
Class type

A parallel map/reduce library. See examples/add_numbers.ml and examples/number_stats.ml for examples.

module Config : sig ... end

module type Worker = sig ... end

Map functions

Map_function modules must be created using the Make_map_function or Make_map_function_with_init functors. The init variety allows you to specify an init function that takes a "param" argument. The non-init variety is equivalent to the init variety with init equal to return and a unit "param" argument. Similarly, Map_reduce_function modules must be created using the Make_map_combine_function or Make_map_combine_function_with_init functors.

module type Map_function = sig ... end

module type Map_function_with_init_spec = sig ... end

module Make_map_function_with_init
  (S : Map_function_with_init_spec) : 
  Map_function
    with type Param.t = S.Param.t
     and type Input.t = S.Input.t
     and type Output.t = S.Output.t

module type Map_function_spec = sig ... end

module Make_map_function
  (S : Map_function_spec) : 
  Map_function
    with type Param.t = unit
     and type Input.t = S.Input.t
     and type Output.t = S.Output.t

val map_unordered : 
  Config.t ->
  'a Async.Pipe.Reader.t ->
  m:
    (module Map_function
      with type Input.t = 'a
       and type Output.t = 'b
       and type Param.t = 'param) ->
  param:'param ->
  ('b * int) Async.Pipe.Reader.t Async.Deferred.t

The map_unordered operation takes 'a Pipe.Reader.t along with a Map_function and sends the 'a values to workers for mapping. Each pair in the resulting ('b * int) Pipe.Reader.t contains the mapped value and the index of the value in the input pipe.

val map : 
  Config.t ->
  'a Async.Pipe.Reader.t ->
  m:
    (module Map_function
      with type Input.t = 'a
       and type Output.t = 'b
       and type Param.t = 'param) ->
  param:'param ->
  'b Async.Pipe.Reader.t Async.Deferred.t

The map operation is similar to map_unordered, but the result is a 'b Pipe.Reader.t where the mapped values are guaranteed to be in the same order as the input values.

val find_map : 
  Config.t ->
  'a Async.Pipe.Reader.t ->
  m:
    (module Map_function
      with type Input.t = 'a
       and type Output.t = 'b option
       and type Param.t = 'param) ->
  param:'param ->
  'b option Async.Deferred.t

The find_map operation takes 'a Pipe.Reader.t along with a Map_function that returns 'b option values. As soon as map returns Some value, all workers are stopped and Some value is returned. If map never returns Some value then None is returned. If more than one worker returns Some value, one value is chosen arbitrarily and returned.

Map-reduce

functions

module type Map_reduce_function = sig ... end

module type Map_reduce_function_with_init_spec = sig ... end

module Make_map_reduce_function_with_init
  (S : Map_reduce_function_with_init_spec) : 
  Map_reduce_function
    with type Param.t = S.Param.t
     and type Accum.t = S.Accum.t
     and type Input.t = S.Input.t

module type Map_reduce_function_spec = sig ... end

module Make_map_reduce_function
  (S : Map_reduce_function_spec) : 
  Map_reduce_function
    with type Param.t = unit
     and type Accum.t = S.Accum.t
     and type Input.t = S.Input.t

val map_reduce_commutative : 
  Config.t ->
  'a Async.Pipe.Reader.t ->
  m:
    (module Map_reduce_function
      with type Accum.t = 'accum
       and type Input.t = 'a
       and type Param.t = 'param) ->
  param:'param ->
  'accum option Async.Deferred.t

The map_reduce_commutative operation takes 'a Pipe.Reader.t along with a Map_reduce_function and applies the map function to 'a values (in an unspecified order), resulting in 'accum values. The combine function is then called to combine the 'accum values (in an unspecified order) into a single 'accum value. Commutative map-reduce assumes that combine is associative and commutative.

val map_reduce : 
  Config.t ->
  'a Async.Pipe.Reader.t ->
  m:
    (module Map_reduce_function
      with type Accum.t = 'accum
       and type Input.t = 'a
       and type Param.t = 'param) ->
  param:'param ->
  'accum option Async.Deferred.t

The map_reduce operation makes strong guarantees about the order in which the values are processed by combine. For a list a_0, a_1, a_2, ..., a_n of 'a values, the noncommutative map-reduce operation applies the map function to produce acc_{i,i+1} from each a_i. The combine function is used to compute combine acc_{i,j} acc_{j,k} for i<j<k, producing acc_{i,k}. The map and combine functions are called repeatedly until the entire list is reduced to a single acc_{0,n+1} value. Noncommutative map-reduce assumes that combine is associative.