Source file CCSemaphore.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
(** {1 Semaphores} *)

type t = {
  mutable n : int;
  mutex : Mutex.t;
  cond : Condition.t;
}

let create n =
  if n <= 0 then invalid_arg "Semaphore.create";
  { n;
    mutex=Mutex.create();
    cond=Condition.create();
  }

let get t = t.n

(* assume [t.mutex] locked, try to acquire [t] *)
let acquire_once_locked_ m t =
  while t.n < m do
    Condition.wait t.cond t.mutex;
  done;
  assert (t.n >= m);
  t.n <- t.n - m;
  Condition.broadcast t.cond;
  Mutex.unlock t.mutex

let acquire m t =
  Mutex.lock t.mutex;
  acquire_once_locked_ m t

(* assume [t.mutex] locked, try to release [t] *)
let release_once_locked_ m t =
  t.n <- t.n + m;
  Condition.broadcast t.cond;
  Mutex.unlock t.mutex

let release m t =
  Mutex.lock t.mutex;
  release_once_locked_ m t;
  ()

(*$R
  let s = create 1 in
  let r = CCLock.create false in
  let _ = Thread.create (fun s -> acquire 5 s; CCLock.set r true) s in
  Thread.yield ();
  assert_equal false (CCLock.get r);
  release 4 s;
  Thread.delay 0.2;
  assert_equal true (CCLock.get r);
  assert_equal 0 (get s)
*)

let with_acquire ~n t ~f =
  acquire n t;
  try
    let x = f() in
    release n t;
    x
  with e ->
    release n t;
    raise e

(*$R
  let s = create 5 in
  let n = CCLock.create 0 in
  let a = Array.init 100 (fun i ->
    Thread.create (fun _ ->
      for _i = 1 to 100 do
        with_acquire ~n:(1 + (i mod 5)) s
          ~f:(fun () -> Thread.yield(); CCLock.incr n)
      done)
    ())
  in
  Array.iter Thread.join a;
  assert_equal ~printer:CCInt.to_string 5 (get s);
  assert_equal ~printer:CCInt.to_string 10_000 (CCLock.get n)
*)

let wait_until_at_least ~n t ~f =
  Mutex.lock t.mutex;
  while t.n < n do
    Condition.wait t.cond t.mutex;
  done;
  assert (t.n >= n);
  Mutex.unlock t.mutex;
  f ()

(*$R
  let output s = () in
  let s = create 2 in
  let res = CCLock.create false in
  let id = Thread.create
    (fun () ->
      output "start";
      wait_until_at_least ~n:5 s
        ~f:(fun () ->
          assert (get s >= 5);
          output "modify now";
          CCLock.set res true)
    ) ()
  in
  output "launched thread";
  Thread.yield();
  assert_bool "start" (not (CCLock.get res));
  output "release 2";
  release 2 s;
  Thread.yield();
  assert_bool "after release 2" (not (CCLock.get res));
  output "release 1";
  release 1 s;
  (* should work now *)
  Thread.delay 0.2;
  Thread.join id;
  output "check";
  assert_bool "after release 1" (CCLock.get res)
*)