add Fut.{reify_error,bind_reify_error}

This commit is contained in:
Simon Cruanes 2023-07-26 16:22:42 -04:00
parent d18e88a772
commit aa0dea3e34
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
3 changed files with 26 additions and 1 deletions

View file

@ -9,8 +9,10 @@ let work_ _i q : unit =
(* A domain level worker. It should not do too much except for starting (* A domain level worker. It should not do too much except for starting
new threads for pools. *) new threads for pools. *)
(* TODO: take [Run of (unit -> unit) | Incr | Decr] to handle refcounting? *)
type worker = { q: (unit -> unit) Bb_queue.t } [@@unboxed] type worker = { q: (unit -> unit) Bb_queue.t } [@@unboxed]
(* TODO: use [worker option array], with a mechanism to start/stop them *)
let domains_ : worker array lazy_t = let domains_ : worker array lazy_t =
lazy lazy
((* number of domains we spawn. Note that we spawn n-1 domains ((* number of domains we spawn. Note that we spawn n-1 domains

View file

@ -100,6 +100,14 @@ let spawn ~on f : _ t =
Pool.run_async on task; Pool.run_async on task;
fut fut
let reify_error (f : 'a t) : 'a or_error t =
match peek f with
| Some res -> return res
| None ->
let fut, promise = make () in
on_result f (fun r -> fulfill promise (Ok r));
fut
let map ?on ~f fut : _ t = let map ?on ~f fut : _ t =
let map_res r = let map_res r =
match r with match r with
@ -161,6 +169,7 @@ let bind ?on ~f fut : _ t =
fut2 fut2
let bind_reify_error ?on ~f fut : _ t = bind ?on ~f (reify_error fut)
let join ?on fut = bind ?on fut ~f:(fun x -> x) let join ?on fut = bind ?on fut ~f:(fun x -> x)
let update_ (st : 'a A.t) f : 'a = let update_ (st : 'a A.t) f : 'a =

View file

@ -85,6 +85,12 @@ val spawn : on:Runner.t -> (unit -> 'a) -> 'a t
(** [spaw ~on f] runs [f()] on the given runner [on], and return a future that will (** [spaw ~on f] runs [f()] on the given runner [on], and return a future that will
hold its result. *) hold its result. *)
val reify_error : 'a t -> 'a or_error t
(** [reify_error fut] turns a failing future into a non-failing
one that contain [Error (exn, bt)]. A non-failing future
returning [x] is turned into [Ok x]
@since NEXT_RELEASE *)
val map : ?on:Runner.t -> f:('a -> 'b) -> 'a t -> 'b t val map : ?on:Runner.t -> f:('a -> 'b) -> 'a t -> 'b t
(** [map ?on ~f fut] returns a new future [fut2] that resolves (** [map ?on ~f fut] returns a new future [fut2] that resolves
with [f x] if [fut] resolved with [x]; with [f x] if [fut] resolved with [x];
@ -92,11 +98,19 @@ val map : ?on:Runner.t -> f:('a -> 'b) -> 'a t -> 'b t
@param on if provided, [f] runs on the given runner *) @param on if provided, [f] runs on the given runner *)
val bind : ?on:Runner.t -> f:('a -> 'b t) -> 'a t -> 'b t val bind : ?on:Runner.t -> f:('a -> 'b t) -> 'a t -> 'b t
(** [map ?on ~f fut] returns a new future [fut2] that resolves (** [bind ?on ~f fut] returns a new future [fut2] that resolves
like the future [f x] if [fut] resolved with [x]; like the future [f x] if [fut] resolved with [x];
and fails with [e] if [fut] fails with [e] or [f x] raises [e]. and fails with [e] if [fut] fails with [e] or [f x] raises [e].
@param on if provided, [f] runs on the given runner *) @param on if provided, [f] runs on the given runner *)
val bind_reify_error : ?on:Runner.t -> f:('a or_error -> 'b t) -> 'a t -> 'b t
(** [bind_reify_error ?on ~f fut] returns a new future [fut2] that resolves
like the future [f (Ok x)] if [fut] resolved with [x];
and resolves like the future [f (Error (exn, bt))]
if [fut] fails with [exn] and backtrace [bt].
@param on if provided, [f] runs on the given runner
@since NEXT_RELEASE *)
val join : ?on:Runner.t -> 'a t t -> 'a t val join : ?on:Runner.t -> 'a t t -> 'a t
(** [join fut] is [fut >>= Fun.id]. It joins the inner layer of the future. (** [join fut] is [fut >>= Fun.id]. It joins the inner layer of the future.
@since 0.2 *) @since 0.2 *)