Merge pull request #9 from issuu/mku/no-deferred-in-thread

No deferred in thread

Author: Leonidas-from-XIV

CHANGES.md

@@ -1,3 +1,6 @@
+* Remove `Deferred.t` from function running in the pool. Running async
+  operations outside the main thread is not safe.
+
 0.11.0
 ======
 

src/thread_pool_async.ml

@@ -28,7 +28,7 @@ let init ~name ~(threads: int) ~(create: unit -> 'state) ~(destroy: 'state -> un
 type 'result computation = [`Ok of 'result | `Attempt_retry]
 
 let with'
-  : 'state t -> ?retries:int -> ('state -> 'result computation Deferred.t) -> 'result Deferred.Or_error.t
+  : 'state t -> ?retries:int -> ('state -> 'result computation) -> 'result Deferred.Or_error.t
   = fun { pool = (reader, writer); create; destroy; _ } ?(retries=0) f ->
   assert (retries >= 0);
 
@@ -40,7 +40,7 @@ let with'
   let run_once
     : 'thread -> 'state -> ('result computation Or_error.t * 'state) Deferred.t
     = fun thread state ->
-    let%bind result = In_thread.run ~thread (fun () -> Deferred.Or_error.try_with (fun () -> f state)) |> Deferred.join in
+    let%bind result = In_thread.run ~thread (fun () -> Or_error.try_with (fun () -> f state)) in
     match result with
     | Ok `Ok res -> return (Ok (`Ok res), state)
     | Ok `Attempt_retry ->

test/dune

@@ -1,7 +1,7 @@
 (executable
  (name test)
  (preprocess (pps ppx_let))
- (libraries alcotest alcotest-async thread_pool_async core async_kernel async_unix))
+ (libraries alcotest alcotest-async thread_pool_async core async_kernel))
 
 (alias
  (name runtest)

test/test.ml

@@ -1,6 +1,5 @@
 open Core
 open Async_kernel
-open Async_unix
 
 module Thread_pool = Thread_pool_async
 
@@ -16,8 +15,8 @@ let test_simple () =
   let create () = Int.incr states in
   let destroy () = Int.decr states in
   let worker () =
-    let%bind () = after (Time.Span.of_ms 5.) in
-    return @@ `Ok (Int.incr jobs)
+    ignore @@ Unix.nanosleep 0.005;
+    `Ok (Int.incr jobs)
   in
 
   let%bind pool = Thread_pool.init ~name:"unittest" ~threads ~create ~destroy in
@@ -46,8 +45,8 @@ let test_error () =
       Int.incr errors;
       failwith "Stop"
     | _ ->
-      let%bind () = after (Time.Span.of_ms 5.) in
-      return @@ `Ok (Int.incr jobs)
+      ignore @@ Unix.nanosleep 0.005;
+      `Ok (Int.incr jobs)
   in
 
   let%bind pool = Thread_pool.init ~name:"unittest" ~threads ~create ~destroy:ignore in
@@ -73,7 +72,7 @@ let test_retry () =
       | false -> `Ok (Int.incr work_done)
     in
     should_fail := false;
-    return result
+    result
   in
   let%bind pool = Thread_pool.init ~name:"unittest" ~threads ~create ~destroy in
   let%bind _result = List.init work_expected ~f:ignore |> Deferred.Or_error.List.iter ~how:`Parallel ~f:(fun () -> Thread_pool.with' ~retries:1 pool (worker (ref true))) in
@@ -92,7 +91,7 @@ let test_retry_fail () =
   let destroy = ignore in
   let worker () =
     Int.incr tries;
-    return `Attempt_retry
+    `Attempt_retry
   in
   let%bind pool = Thread_pool.init ~name:"unittest" ~threads ~create ~destroy in
   let%bind result = List.init work ~f:ignore |> Deferred.Or_error.List.iter ~how:`Parallel ~f:(fun () -> Thread_pool.with' ~retries:0 pool worker) in