More raise less fail

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
@@ -299,7 +299,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
@@ -315,4 +315,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

src/ppx/ppx_lwt.ml

@@ -23,7 +23,7 @@ let add_wildcard_case cases =
   if not has_wildcard
   then cases
        @ (let loc = Location.none in
-          [case ~lhs:[%pat? exn] ~guard:None ~rhs:[%expr Lwt.fail exn]])
+          [case ~lhs:[%pat? exn] ~guard:None ~rhs:[%expr Lwt.reraise exn]])
   else cases
 
 (** {3 Internal names} *)
@@ -154,11 +154,11 @@ let lwt_expression mapper exp attributes ext_loc =
     Some (mapper#expression { new_exp with pexp_attributes })
 
   (* [assert%lwt $e$] ≡
-     [try Lwt.return (assert $e$) with exn -> Lwt.fail exn] *)
+     [try Lwt.return (assert $e$) with exn -> Lwt.reraise exn] *)
   | Pexp_assert e ->
     let new_exp =
       let loc = !default_loc in
-      [%expr try Lwt.return (assert [%e e]) with exn -> Lwt.fail exn]
+      [%expr try Lwt.return (assert [%e e]) with exn -> Lwt.reraise exn]
     in
     Some (mapper#expression { new_exp with pexp_attributes })
 

src/ppx/ppx_lwt.mli

@@ -80,10 +80,10 @@ catch (fun () -> f x)
         prerr_endline msg;
         return ()
     | exn ->
-        Lwt.fail exn)
+        Lwt.reraise exn)
    ]}
 
-   Note that the [exn -> Lwt.fail exn] branch is automatically added
+   Note that the [exn -> Lwt.reraise exn] branch is automatically added
    when needed.
 
    - finalizer:

src/unix/lwt_io.ml

@@ -231,7 +231,7 @@ let perform_io : type mode. mode _channel -> int Lwt.t = fun ch ->
             (function
               | Unix.Unix_error (Unix.EPIPE, _, _) ->
                 Lwt.return 0
-              | exn -> Lwt.fail exn) [@ocaml.warning "-4"]
+              | exn -> Lwt.reraise exn) [@ocaml.warning "-4"]
         else
           perform ch.buffer ptr len
       in
@@ -525,7 +525,7 @@ let make :
     max = (match mode with
       | Input -> 0
       | Output -> size);
-    close = lazy(Lwt.catch close Lwt.fail);
+    close = lazy(Lwt.catch close Lwt.reraise);
     abort_waiter = abort_waiter;
     abort_wakener = abort_wakener;
     main = wrapper;
@@ -537,7 +537,7 @@ let make :
         perform_io,
         fun pos cmd ->
           try seek pos cmd
-          with e when Lwt.Exception_filter.run e -> Lwt.fail e
+          with e when Lwt.Exception_filter.run e -> Lwt.reraise e
     );
   } and wrapper = {
     state = Idle;
@@ -678,7 +678,7 @@ struct
     let ptr = ic.ptr in
     if ptr = ic.max then
       refill ic >>= function
-      | 0 -> Lwt.fail End_of_file
+      | 0 -> raise End_of_file
       | _ -> read_char ic
     else begin
       ic.ptr <- ptr + 1;
@@ -690,7 +690,7 @@ struct
       (fun () -> read_char ic >|= fun ch -> Some ch)
       (function
         | End_of_file -> Lwt.return_none
-        | exn -> Lwt.fail exn)
+        | exn -> Lwt.reraise exn)
 
   let read_line ic =
     let buf = Buffer.create 128 in
@@ -711,7 +711,7 @@ struct
             if cr_read then Buffer.add_char buf '\r';
             Lwt.return(Buffer.contents buf)
           | exn ->
-            Lwt.fail exn)
+            Lwt.reraise exn)
     in
     read_char ic >>= function
     | '\r' -> loop true
@@ -723,7 +723,7 @@ struct
       (fun () -> read_line ic >|= fun ch -> Some ch)
       (function
         | End_of_file -> Lwt.return_none
-        | exn -> Lwt.fail exn)
+        | exn -> Lwt.reraise exn)
 
   let unsafe_read_into' ic blit buf ofs len =
     let avail = ic.max - ic.ptr in
@@ -771,7 +771,7 @@ struct
     let rec loop ic buf ofs len =
       read_into ic buf ofs len >>= function
       | 0 ->
-          Lwt.fail End_of_file
+          raise End_of_file
       | n ->
           let len = len - n in
           if len = 0 then
@@ -985,7 +985,7 @@ struct
     if ic.max - ic.ptr < size then
       refill ic >>= function
       | 0 ->
-        Lwt.fail End_of_file
+        raise End_of_file
       | _ ->
         read_block_unsafe ic size f
     else begin
@@ -1440,7 +1440,7 @@ let open_temp_file ?buffer ?flags ?perm ?temp_dir ?prefix ?(suffix = "") () =
         Lwt.return (fname, chan))
       (function
         | Unix.Unix_error _ when n < 1000 -> attempt (n + 1)
-        | exn -> Lwt.fail exn)
+        | exn -> Lwt.reraise exn)
   in
   attempt 0
 
@@ -1468,7 +1468,7 @@ let create_temp_dir
         Lwt.return name)
       (function
       | Unix.Unix_error (Unix.EEXIST, _, _) when n < 1000 -> attempt (n + 1)
-      | exn -> Lwt.fail exn)
+      | exn -> Lwt.reraise exn)
   in
   attempt 0
 
@@ -1489,10 +1489,10 @@ let win32_unlink fn =
                 (* If everything succeeded but the final removal still failed,
                    restore original permissions *)
                 Lwt_unix.chmod fn st_perm >>= fun () ->
-                Lwt.fail exn)
+                Lwt.reraise exn)
          )
-         (fun _ -> Lwt.fail exn)
-     | exn -> Lwt.fail exn)
+         (fun _ -> Lwt.reraise exn)
+     | exn -> Lwt.reraise exn)
 
 let unlink =
   if Sys.win32 then
@@ -1549,7 +1549,7 @@ let close_socket fd =
          (function
            (* Occurs if the peer closes the connection first. *)
            | Unix.Unix_error (Unix.ENOTCONN, _, _) -> Lwt.return_unit
-           | exn -> Lwt.fail exn) [@ocaml.warning "-4"])
+           | exn -> Lwt.reraise exn) [@ocaml.warning "-4"])
     (fun () ->
        Lwt_unix.close fd)
 
@@ -1574,7 +1574,7 @@ let open_connection ?fd ?in_buffer ?out_buffer sockaddr =
                      ~mode:output (Lwt_bytes.write fd)))
     (fun exn ->
        Lwt_unix.close fd >>= fun () ->
-       Lwt.fail exn)
+       Lwt.reraise exn)
 
 let with_close_connection f (ic, oc) =
   (* If the user already tried to close the socket and got an exception, we
@@ -1639,7 +1639,7 @@ let establish_server_generic
         (function
           | Unix.Unix_error (Unix.ECONNABORTED, _, _) ->
             Lwt.return `Try_again
-          | e -> Lwt.fail e)
+          | e -> Lwt.reraise e)
     in
 
     Lwt.pick [try_to_accept; should_stop] >>= function

src/unix/lwt_process.cppo.ml

@@ -378,7 +378,7 @@ let read_opt read ic =
     (function
       | Unix.Unix_error (Unix.EPIPE, _, _) | End_of_file ->
         Lwt.return_none
-      | exn -> Lwt.fail exn) [@ocaml.warning "-4"]
+      | exn -> Lwt.reraise exn) [@ocaml.warning "-4"]
 
 let recv_chars pr =
   let ic = pr#stdout in
@@ -512,8 +512,8 @@ let pmap ?timeout ?env ?cwd ?stderr cmd text =
       | Lwt.Canceled as exn ->
         (* Cancel the getter if the sender was canceled. *)
         Lwt.cancel getter;
-        Lwt.fail exn
-      | exn -> Lwt.fail exn)
+        Lwt.reraise exn
+      | exn -> Lwt.reraise exn)
 
 let pmap_chars ?timeout ?env ?cwd ?stderr cmd chars =
   let pr = open_process ?timeout ?env ?cwd ?stderr cmd in
@@ -534,8 +534,8 @@ let pmap_line ?timeout ?env ?cwd ?stderr cmd line =
       | Lwt.Canceled as exn ->
         (* Cancel the getter if the sender was canceled. *)
         Lwt.cancel getter;
-        Lwt.fail exn
-      | exn -> Lwt.fail exn)
+        Lwt.reraise exn
+      | exn -> Lwt.reraise exn)
 
 let pmap_lines ?timeout ?env ?cwd ?stderr cmd lines =
   let pr = open_process ?timeout ?env ?cwd ?stderr cmd in

src/unix/lwt_unix.cppo.ml

@@ -147,7 +147,7 @@ let auto_pause timeout =
 
 exception Timeout
 
-let timeout d = sleep d >>= fun () -> Lwt.fail Timeout
+let timeout d = sleep d >>= fun () -> raise Timeout
 
 let with_timeout d f = Lwt.pick [timeout d; Lwt.apply f ()]
 
@@ -582,7 +582,7 @@ let wrap_syscall event ch action =
   | Retry_write ->
     register_action Write ch action
   | e when Lwt.Exception_filter.run e ->
-    Lwt.fail e
+    Lwt.reraise e
 
 (* +-----------------------------------------------------------------+
    | Basic file input/output                                         |
@@ -636,7 +636,7 @@ let wait_read ch =
          Lwt.return_unit
        else
          register_action Read ch ignore)
-    Lwt.fail
+    Lwt.reraise
 
 external stub_read : Unix.file_descr -> Bytes.t -> int -> int -> int = "lwt_unix_read"
 external read_job : Unix.file_descr -> Bytes.t -> int -> int -> int job = "lwt_unix_read_job"
@@ -694,7 +694,7 @@ let wait_write ch =
          Lwt.return_unit
        else
          register_action Write ch ignore)
-    Lwt.fail
+    Lwt.reraise
 
 external stub_write : Unix.file_descr -> Bytes.t -> int -> int -> int = "lwt_unix_write"
 external write_job : Unix.file_descr -> Bytes.t -> int -> int -> int job = "lwt_unix_write_job"
@@ -1034,7 +1034,7 @@ let file_exists name =
     (fun e ->
        match e with
        | Unix.Unix_error _ -> Lwt.return_false
-       | _ -> Lwt.fail e) [@ocaml.warning "-4"]
+       | _ -> Lwt.reraise e) [@ocaml.warning "-4"]
 
 external utimes_job : string -> float -> float -> unit job =
   "lwt_unix_utimes_job"
@@ -1140,7 +1140,7 @@ struct
       (fun e ->
          match e with
          | Unix.Unix_error _ -> Lwt.return_false
-         | _ -> Lwt.fail e) [@ocaml.warning "-4"]
+         | _ -> Lwt.reraise e) [@ocaml.warning "-4"]
 
 end
 
@@ -1408,7 +1408,7 @@ let files_of_directory path =
               (fun () -> readdir_n handle chunk_size)
               (fun exn ->
                  closedir handle >>= fun () ->
-                 Lwt.fail exn) >>= fun entries ->
+                 Lwt.reraise exn) >>= fun entries ->
             if Array.length entries < chunk_size then begin
               state := LDS_done;
               closedir handle >>= fun () ->
@@ -1423,7 +1423,7 @@ let files_of_directory path =
               (fun () -> readdir_n handle chunk_size)
               (fun exn ->
                  closedir handle >>= fun () ->
-                 Lwt.fail exn) >>= fun entries ->
+                 Lwt.reraise exn) >>= fun entries ->
             if Array.length entries < chunk_size then begin
               state := LDS_done;
               closedir handle >>= fun () ->
@@ -2395,7 +2395,7 @@ let () =
 let _waitpid flags pid =
   Lwt.catch
     (fun () -> Lwt.return (Unix.waitpid flags pid))
-    Lwt.fail
+    Lwt.reraise
 
 let waitpid =
   if Sys.win32 then