Use raise and reraise instead of fail in more places in core

src/core/lwt_pool.ml

@@ -108,7 +108,7 @@ let validate_and_return p c =
             resolver is waiting. *)
          dispose p c >>= fun () ->
          replace_disposed p;
-         Lwt.fail e)
+         Lwt.reraise e)
 
 (* Acquire a pool member. *)
 let acquire p =

src/core/lwt_seq.ml

@@ -270,7 +270,7 @@ let rec unfold f u () =
   match f u with
   | None -> return_nil
   | Some (x, u') -> Lwt.return (Cons (x, unfold f u'))
-  | exception exc when Lwt.Exception_filter.run exc -> Lwt.fail exc
+  | exception exc when Lwt.Exception_filter.run exc -> Lwt.reraise exc
 
 let rec unfold_lwt f u () =
   let* x = f u in
@@ -305,7 +305,7 @@ let rec of_seq seq () =
   | Seq.Nil -> return_nil
   | Seq.Cons (x, next) ->
     Lwt.return (Cons (x, (of_seq next)))
-  | exception exn when Lwt.Exception_filter.run exn -> Lwt.fail exn
+  | exception exn when Lwt.Exception_filter.run exn -> Lwt.reraise exn
 
 let rec of_seq_lwt (seq: 'a Lwt.t Seq.t): 'a t = fun () ->
     match seq () with
@@ -321,4 +321,4 @@ let of_seq_lwt (seq: 'a Lwt.t Seq.t): 'a t = fun () ->
        let+ x = x in
        let next = of_seq_lwt next in
        Cons (x, next)
-    | exception exc when Lwt.Exception_filter.run exc -> Lwt.fail exc
+    | exception exc when Lwt.Exception_filter.run exc -> Lwt.reraise exc

src/core/lwt_stream.ml

@@ -279,9 +279,9 @@ class ['a] bounded_push_impl (info : 'a push_bounded) wakener_cell last close =
              let waiter, wakener = Lwt.task () in
              info.pushb_push_waiter <- waiter;
              info.pushb_push_wakener <- wakener;
-             Lwt.fail exn
+             Lwt.reraise exn
            | _ ->
-             Lwt.fail exn)
+             Lwt.reraise exn)
     end else begin
       (* Push the element at the end of the queue. *)
       enqueue' (Some x) last;
@@ -367,11 +367,18 @@ let feed s =
     else begin
       (* Otherwise request a new element. *)
       let thread =
-        from.from_create () >>= fun x ->
-        (* Push the element to the end of the queue. *)
-        enqueue x s;
-        if x = None then Lwt.wakeup s.close ();
-        Lwt.return_unit
+        (* The function [from_create] can raise an exception (with
+           [raise], rather than returning a failed promise with
+           [Lwt.fail]). In this case, we have to catch the exception
+           and turn it into a safe failed promise. *)
+        Lwt.catch
+          (fun () ->
+            from.from_create () >>= fun x ->
+            (* Push the element to the end of the queue. *)
+            enqueue x s;
+            if x = None then Lwt.wakeup s.close ();
+            Lwt.return_unit)
+          Lwt.reraise
       in
       (* Allow other threads to access this thread. *)
       from.from_thread <- thread;
@@ -1070,7 +1077,7 @@ let parse s f =
     (fun () -> f s)
     (fun exn ->
        s.node <- node;
-       Lwt.fail exn)
+       Lwt.reraise exn)
 
 let hexdump stream =
   let buf = Buffer.create 80 and num = ref 0 in

test/core/test_lwt.ml

@@ -549,6 +549,15 @@ let catch_tests = suite "catch" [
     state_is (Lwt.Return Exception) p
   end;
 
+  test "rejected (raise)" begin fun () ->
+    let p =
+      Lwt.catch
+        (fun () -> raise Exception)
+        (fun exn -> Lwt.return exn)
+    in
+    state_is (Lwt.Return Exception) p
+  end;
+
   (* This is an analog of the "bind quirk," see
 
        https://github.com/ocsigen/lwt/issues/329 *)
@@ -696,7 +705,7 @@ let backtrace_catch_tests = suite "backtrace_catch" [
   test "rejected" begin fun () ->
     let p =
       Lwt.backtrace_catch add_loc
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
         (fun exn -> Lwt.return exn)
     in
     state_is (Lwt.Return Exception) p
@@ -789,7 +798,7 @@ let try_bind_tests = suite "try_bind" [
   test "rejected" begin fun () ->
     let p =
       Lwt.try_bind
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
         (fun _ -> Lwt.return Exit)
         (fun exn -> Lwt.return exn)
     in
@@ -810,7 +819,7 @@ let try_bind_tests = suite "try_bind" [
   test "rejected, h raises" begin fun () ->
     try
       ignore @@ Lwt.try_bind
-        (fun () -> Lwt.fail Exit)
+        (fun () -> raise Exit)
         (fun _ -> Lwt.return_unit)
         (fun _ -> raise Exception);
       Lwt.return_false
@@ -961,7 +970,7 @@ let backtrace_try_bind_tests = suite "backtrace_try_bind" [
   test "rejected" begin fun () ->
     let p =
       Lwt.backtrace_try_bind add_loc
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
         (fun _ -> Lwt.return Exit)
         (fun exn -> Lwt.return exn)
     in
@@ -1132,7 +1141,7 @@ let finalize_tests = suite "finalize" [
   test "rejected, f' raises" begin fun () ->
     try
       ignore @@ Lwt.finalize
-        (fun () -> Lwt.fail Exit)
+        (fun () -> raise Exit)
         (fun () -> raise Exception);
       Lwt.return_false
     with Exception ->
@@ -1169,7 +1178,7 @@ let finalize_tests = suite "finalize" [
     let p =
       Lwt.finalize
         (fun () -> p)
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
     in
     Lwt.wakeup r ();
     state_is (Lwt.Fail Exception) p
@@ -1232,7 +1241,7 @@ let finalize_tests = suite "finalize" [
     let p =
       Lwt.finalize
         (fun () -> p)
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
     in
     Lwt.wakeup_exn r Exit;
     state_is (Lwt.Fail Exception) p
@@ -1347,7 +1356,7 @@ let backtrace_finalize_tests = suite "backtrace_finalize" [
     let f'_ran = ref false in
     let p =
       Lwt.backtrace_finalize add_loc
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
         (fun () -> f'_ran := true; Lwt.return_unit)
     in
     Lwt.bind (state_is (Lwt.Fail Exception) p) (fun correct ->
@@ -1367,7 +1376,7 @@ let backtrace_finalize_tests = suite "backtrace_finalize" [
   test "rejected, f' raises" begin fun () ->
     try
       ignore @@ Lwt.backtrace_finalize add_loc
-        (fun () -> Lwt.fail Exit)
+        (fun () -> raise Exit)
         (fun () -> raise Exception);
       Lwt.return_false
     with Exception ->
@@ -1404,7 +1413,7 @@ let backtrace_finalize_tests = suite "backtrace_finalize" [
     let p =
       Lwt.backtrace_finalize add_loc
         (fun () -> p)
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
     in
     Lwt.wakeup r ();
     state_is (Lwt.Fail Exception) p
@@ -1439,7 +1448,7 @@ let backtrace_finalize_tests = suite "backtrace_finalize" [
     let p =
       Lwt.backtrace_finalize add_loc
         (fun () -> p)
-        (fun () -> Lwt.fail Exception)
+        (fun () -> raise Exception)
     in
     Lwt.wakeup_exn r Exit;
     state_is (Lwt.Fail Exception) p
@@ -1803,7 +1812,7 @@ let async_tests = suite "async" [
     let saw = ref None in
     let restore =
       set_async_exception_hook (fun exn -> saw := Some exn) in
-    Lwt.async (fun () -> Lwt.fail Exception);
+    Lwt.async (fun () -> raise Exception);
     later (fun () ->
       restore ();
       !saw = Some Exception)
@@ -1852,7 +1861,7 @@ let dont_wait_tests = suite "dont_wait" [
   test "rejected" begin fun () ->
     let saw = ref None in
     Lwt.dont_wait
-      (fun () -> Lwt.fail Exception)
+      (fun () -> raise Exception)
       (fun exn -> saw := Some exn);
     later (fun () -> !saw = Some Exception)
   end;
@@ -3371,7 +3380,7 @@ let cancel_catch_tests = suite "cancel catch" [
   test "task, pending, canceled, on_cancel, forwarded" begin fun () ->
     let on_cancel_2_ran = ref false in
     let p, _ = Lwt.task () in
-    let p' = Lwt.catch (fun () -> p) Lwt.fail in
+    let p' = Lwt.catch (fun () -> p) Lwt.reraise in
     Lwt.on_cancel p' (fun () -> on_cancel_2_ran := true);
     Lwt.cancel p';
     Lwt.return
@@ -3895,7 +3904,7 @@ let storage_tests = suite "storage" [
     Lwt.with_value key (Some 42) (fun () ->
       let p' =
         Lwt.with_value key (Some 1337) (fun () ->
-          Lwt.try_bind (fun () -> p) f Lwt.fail)
+          Lwt.try_bind (fun () -> p) f Lwt.reraise)
       in
       Lwt.wakeup r ();
       Lwt.return

test/core/test_lwt_pool.ml

@@ -16,7 +16,7 @@ let suite = suite "lwt_pool" [
   end;
 
   test "creator exception" begin fun () ->
-    let gen = fun () -> Lwt.fail Dummy_error in
+    let gen = fun () -> raise Dummy_error in
     let p = Lwt_pool.create 1 gen in
     let u = Lwt_pool.use p (fun _ -> Lwt.return 0) in
     Lwt.return (Lwt.state u = Lwt.Fail Dummy_error)
@@ -42,7 +42,7 @@ let suite = suite "lwt_pool" [
   test "validation exceptions are propagated to users" begin fun () ->
     let c = Lwt_condition.create () in
     let gen = (fun () -> let l = ref 0 in Lwt.return l) in
-    let v l = if !l = 0 then Lwt.return_true else Lwt.fail Dummy_error in
+    let v l = if !l = 0 then Lwt.return_true else raise Dummy_error in
     let p = Lwt_pool.create 1 ~validate:v gen in
     let u1 = Lwt_pool.use p (fun l -> l := 1; Lwt_condition.wait c) in
     let u2 = Lwt_pool.use p (fun l -> Lwt.return !l) in
@@ -106,7 +106,7 @@ let suite = suite "lwt_pool" [
   test "waiter are notified on replacement" begin fun () ->
     let c = Lwt_condition.create () in
     let gen = (fun () -> let l = ref 0 in Lwt.return l) in
-    let v l = if !l = 0 then Lwt.return_true else Lwt.fail Dummy_error in
+    let v l = if !l = 0 then Lwt.return_true else raise Dummy_error in
     let p = Lwt_pool.create 1 ~validate:v gen in
     let u1 = Lwt_pool.use p (fun l -> l := 1; Lwt_condition.wait c) in
     let u2 = Lwt_pool.use p (fun l -> Lwt.return !l) in
@@ -128,9 +128,9 @@ let suite = suite "lwt_pool" [
       if !k then
         let l = ref 0 in Lwt.return l
       else
-        Lwt.fail Dummy_error
+        raise Dummy_error
     in
-    let v l = if !l = 0 then Lwt.return_true else Lwt.fail Dummy_error in
+    let v l = if !l = 0 then Lwt.return_true else raise Dummy_error in
     let p = Lwt_pool.create 1 ~validate:v gen in
     let u1 = Lwt_pool.use p (fun l -> l := 1; k:= false; Lwt_condition.wait c) in
     let u2 = Lwt_pool.use p (fun l -> Lwt.return !l) in
@@ -156,7 +156,7 @@ let suite = suite "lwt_pool" [
     let cond = Lwt_condition.create() in
     let p = Lwt_pool.create 1 ~validate:v ~check:c gen in
     let _ = Lwt_pool.use p (fun l -> l := 1; Lwt_condition.wait cond) in
-    let _ = Lwt_pool.use p (fun l -> l := 2; Lwt.fail Dummy_error) in
+    let _ = Lwt_pool.use p (fun l -> l := 2; raise Dummy_error) in
     let u3 = Lwt_pool.use p (fun l -> Lwt.return !l) in
     let () = Lwt_condition.signal cond "done" in
     Lwt.bind u3 (fun v ->
@@ -169,7 +169,7 @@ let suite = suite "lwt_pool" [
     let p = Lwt_pool.create 1 gen in
     let _ = Lwt_pool.use p (fun l ->
       Lwt.bind (Lwt_condition.wait cond)
-        (fun _ -> l:= 1; Lwt.fail Dummy_error)) in
+        (fun _ -> l:= 1; raise Dummy_error)) in
     let u2 = Lwt_pool.use p (fun l -> Lwt.return !l) in
     let () = Lwt_condition.signal cond "done" in
     Lwt.bind u2 (fun v ->

test/core/test_lwt_result.ml

@@ -93,7 +93,7 @@ let suite =
 
     test "catch, error case"
       (fun () ->
-         let x () = Lwt.fail Dummy_error in
+         let x () = raise Dummy_error in
          Lwt.return (Lwt_result.catch x = Lwt_result.fail Dummy_error)
       );
 

test/core/test_lwt_stream.ml

@@ -11,7 +11,7 @@ let expect_exit f =
       Lwt.return_false)
     (function
       | Exit -> Lwt.return_true
-      | e -> Lwt.fail e)
+      | e -> Lwt.reraise e)
 
 let suite = suite "lwt_stream" [
   test "from"
@@ -351,7 +351,7 @@ let suite = suite "lwt_stream" [
                     return (Some x)
                 | (Result.Error e)::l ->
                     q := l;
-                    Lwt.fail e)
+                    raise e)
        in
        Lwt_stream.to_list (Lwt_stream.wrap_exn stream) >>= fun l' ->
        return (l = l'));
@@ -418,7 +418,7 @@ let suite = suite "lwt_stream" [
 
   test "exception passing: basic, from"
     (fun () ->
-      let stream = Lwt_stream.from (fun () -> Lwt.fail Exit) in
+      let stream = Lwt_stream.from (fun () -> raise Exit) in
       expect_exit (fun () -> Lwt_stream.get stream));
 
   test "exception passing: basic, from_direct"
@@ -428,12 +428,12 @@ let suite = suite "lwt_stream" [
 
   test "exception passing: to_list"
     (fun () ->
-      let stream = Lwt_stream.from (fun () -> Lwt.fail Exit) in
+      let stream = Lwt_stream.from (fun () -> raise Exit) in
       expect_exit (fun () -> Lwt_stream.to_list stream));
 
   test "exception passing: mapped"
     (fun () ->
-      let stream = Lwt_stream.from (fun () -> Lwt.fail Exit) in
+      let stream = Lwt_stream.from (fun () -> raise Exit) in
       let stream = Lwt_stream.map (fun v -> v) stream in
       expect_exit (fun () -> Lwt_stream.get stream));