Merge branch 'master' of https://github.com/Sudha247/lwt-multicore

Author: Sudha247

src/core/dune

@@ -26,7 +26,7 @@ let () = Jbuild_plugin.V1.send @@ {|
  (synopsis "Monadic promises and concurrent I/O")
  (wrapped false)
  |} ^ preprocess ^ {|
- (libraries bytes result seq)
+ (libraries bytes result seq domainslib)
  (flags (:standard -w +A-29)))
 
 (documentation

src/core/lwt_lock.ml

@@ -0,0 +1,32 @@
+(** Locks for multicore*)
+module C = Domainslib.Chan 
+
+type t = int C.t  
+
+let create_lock () = 
+  let c = C.make_bounded 1 in 
+  C.send c 0;
+  c
+
+let acquire_lock lock = 
+  C.recv lock |> ignore 
+
+let release_lock lock = 
+  C.send lock 0
+
+let is_locked lock = 
+  let c = C.recv_poll lock in 
+  match c with 
+  | Some v ->
+      C.send lock v;
+      false
+  | None ->
+      true
+
+let try_acquire lock =
+  match C.recv_poll lock with 
+  | Some _ -> true 
+  | None -> false 
+
+let try_release lock = 
+  C.send_poll lock 0

src/core/lwt_mutex.ml

@@ -18,18 +18,25 @@ type t = { mutable locked : bool; mutable waiters : unit Lwt.u Lwt_sequence.t  }
 
 let create () = { locked = false; waiters = Lwt_sequence.create () }
 
+let l = Lwt_lock.create_lock ()
+
 let lock m =
   if m.locked then
     (Lwt.add_task_r [@ocaml.warning "-3"]) m.waiters
   else begin
+    Lwt_lock.acquire_lock l;
     m.locked <- true;
+    Lwt_lock.release_lock l;
     Lwt.return_unit
   end
 
 let unlock m =
   if m.locked then begin
-    if Lwt_sequence.is_empty m.waiters then
-      m.locked <- false
+    if Lwt_sequence.is_empty m.waiters then begin 
+      Lwt_lock.acquire_lock l;
+      m.locked <- false;
+      Lwt_lock.release_lock l
+    end
     else
       (* We do not use [Lwt.wakeup] here to avoid a stack overflow
          when unlocking a lot of threads. *)

src/core/lwt_mvar.ml

@@ -49,6 +49,8 @@ type 'a t = {
   (* Threads waiting for a value *)
 }
 
+let lock = Lwt_lock.create_lock ()
+
 let create_empty () =
   { mvar_contents = None;
     writers = Lwt_sequence.create ();
@@ -60,6 +62,7 @@ let create v =
     readers = Lwt_sequence.create () }
 
 let put mvar v =
+  Lwt_lock.acquire_lock lock;
   match mvar.mvar_contents with
   | None ->
     begin match Lwt_sequence.take_opt_l mvar.readers with
@@ -68,11 +71,13 @@ let put mvar v =
       | Some w ->
         Lwt.wakeup_later w v
     end;
+    Lwt_lock.release_lock lock;
     Lwt.return_unit
   | Some _ ->
     let (res, w) = Lwt.task () in
     let node = Lwt_sequence.add_r (v, w) mvar.writers in
     Lwt.on_cancel res (fun _ -> Lwt_sequence.remove node);
+    Lwt_lock.release_lock lock;
     res
 
 let next_writer mvar =
@@ -86,7 +91,9 @@ let next_writer mvar =
 let take_available mvar =
   match mvar.mvar_contents with
   | Some v ->
+    Lwt_lock.acquire_lock lock;
     next_writer mvar;
+    Lwt_lock.release_lock lock;
     Some v
   | None ->
     None

src/core/lwt_sequence.ml

@@ -20,6 +20,7 @@ type 'a node = {
 external seq_of_node : 'a node -> 'a t = "%identity"
 external node_of_seq : 'a t -> 'a node = "%identity"
 
+let lock = Lwt_lock.create_lock ()
 (* +-----------------------------------------------------------------+
    | Operations on nodes                                             |
    +-----------------------------------------------------------------+ *)
@@ -28,14 +29,20 @@ let get node =
   node.node_data
 
 let set node data =
-  node.node_data <- data
+  Lwt_lock.acquire_lock lock;
+  node.node_data <- data;
+  Lwt_lock.release_lock lock
 
 let remove node =
   if node.node_active then begin
+    let is_locked = Lwt_lock.try_acquire lock in 
     node.node_active <- false;
     let seq = seq_of_node node in
     seq.prev.next <- seq.next;
-    seq.next.prev <- seq.prev
+    seq.next.prev <- seq.prev;
+    match is_locked with
+    | true -> Lwt_lock.release_lock lock
+    | false -> ()
   end
 
 (* +-----------------------------------------------------------------+
@@ -62,32 +69,40 @@ let length seq =
   loop seq.next 0
 
 let add_l data seq =
+  Lwt_lock.acquire_lock lock;
   let node = { node_prev = seq; node_next = seq.next; node_data = data; node_active = true } in
   seq.next.prev <- seq_of_node node;
   seq.next <- seq_of_node node;
+  Lwt_lock.release_lock lock;
   node
 
 let add_r data seq =
+  Lwt_lock.acquire_lock lock;
   let node = { node_prev = seq.prev; node_next = seq; node_data = data; node_active = true } in
   seq.prev.next <- seq_of_node node;
   seq.prev <- seq_of_node node;
+  Lwt_lock.release_lock lock;
   node
 
 let take_l seq =
   if is_empty seq then
     raise Empty
   else begin
+    Lwt_lock.acquire_lock lock;
     let node = node_of_seq seq.next in
     remove node;
+    Lwt_lock.release_lock lock;
     node.node_data
   end
 
 let take_r seq =
   if is_empty seq then
     raise Empty
   else begin
+    Lwt_lock.acquire_lock lock;
     let node = node_of_seq seq.prev in
     remove node;
+    Lwt_lock.release_lock lock;
     node.node_data
   end
 
@@ -110,20 +125,24 @@ let take_opt_r seq =
   end
 
 let transfer_l s1 s2 =
+  Lwt_lock.acquire_lock lock;
   s2.next.prev <- s1.prev;
   s1.prev.next <- s2.next;
   s2.next <- s1.next;
   s1.next.prev <- s2;
   s1.prev <- s1;
-  s1.next <- s1
+  s1.next <- s1;
+  Lwt_lock.release_lock lock
 
 let transfer_r s1 s2 =
+  Lwt_lock.acquire_lock lock;
   s2.prev.next <- s1.next;
   s1.next.prev <- s2.prev;
   s2.prev <- s1.prev;
   s1.prev.next <- s2;
   s1.prev <- s1;
-  s1.next <- s1
+  s1.next <- s1;
+  Lwt_lock.release_lock lock
 
 let iter_l f seq =
   let rec loop curr =

src/unix/dune

@@ -42,7 +42,7 @@ let () = Jbuild_plugin.V1.send @@ {|
  (synopsis "Unix support for Lwt")
  (optional)
  (wrapped false)
- (libraries bigarray lwt mmap ocplib-endian.bigstring threads unix)
+ (libraries bigarray lwt mmap ocplib-endian.bigstring threads unix domainslib)
  |} ^ preprocess ^ {|
  (c_names
   lwt_unix_stubs