[Caml-list] semaphore puzzle

[Caml-list] semaphore puzzle

[Matej Kosik]

[-- Attachment #1: Type: text/plain, Size: 1000 bytes --]

Hi,

I am trying to find the most simple and, if possible, clean way to synchronize my threads.

In this particular case, semaphores would be ideal
(alternatively, unbounded channels)

  (* Create a new semaphore with a given initial value. *)
  val create : int -> t

  (* V *)
  val up : t -> unit

  (* P *)
  val down : t -> unit

  val try_down : t -> bool

The "down" or "try_down" are almost the solutions, but not quite.

"down" may block forever ---> there is no timeout at all
                              (I need some)

"try_down" --->  the timeout is zero
                 (I prefer a non-zero timeout)

Obviously, I could use "try_down" in a loop, checking the system time myself and then either give up or actually manage to decrement the semaphore.

Is there a better solution than this?

(There is a Unix module, there are signals, but I am not sure whether it is safe to use them in multithreaded program.
 At least, I did not have a luck.)

Thanks in advance for patience &| help.

[-- Attachment #2: semaphore.ml --]
[-- Type: text/plain, Size: 1075 bytes --]

exception Timeout

type t = {mutex : Mutex.t;
          condition_lock : Condition.t;
          mutable value : int}

let create initial_value =
  assert (0 <= initial_value);
  {mutex = Mutex.create ();
   condition_lock = Condition.create ();
   value = initial_value}

let up semaphore =
  Mutex.lock semaphore.mutex;
  semaphore.value <- succ semaphore.value;
  Mutex.unlock semaphore.mutex;
  Condition.signal semaphore.condition_lock

let down semaphore =
  Mutex.lock semaphore.mutex;
  assert (0 <= semaphore.value);
  while semaphore.value = 0 do
    Condition.wait semaphore.condition_lock semaphore.mutex
  done;
  semaphore.value <- pred semaphore.value;
  Mutex.unlock semaphore.mutex

(* If it is possible to decrement the semaphore, do it and return "true".
   Otherwise return "false". *)
let try_down semaphore =
  Mutex.lock semaphore.mutex;
  if 0 < semaphore.value then
    begin
      semaphore.value <- pred semaphore.value;
      Mutex.unlock semaphore.mutex;
      true
    end
  else
    begin
      Mutex.unlock semaphore.mutex;
      false
    end

Re: [Caml-list] semaphore puzzle

[Gerd Stolpmann]

[-- Attachment #1: Type: text/plain, Size: 2797 bytes --]

Hi,

Ocamlnet includes support for semaphores (at least for Unix platforms). 

Am Montag, den 30.09.2013, 19:30 +0100 schrieb Matej Kosik:
> Hi,
> 
> I am trying to find the most simple and, if possible, clean way to synchronize my threads.
> 
> In this particular case, semaphores would be ideal
> (alternatively, unbounded channels)
> 
>   (* Create a new semaphore with a given initial value. *)
>   val create : int -> t

Netsys_posix.sem_create (or sem_open or sem_init).

> 
>   (* V *)
>   val up : t -> unit

Netsys_posix.sem_post

>   (* P *)
>   val down : t -> unit

Netsys_posix.sem_wait with option SEM_WAIT_BLOCK.


>   val try_down : t -> bool

Netsys_posix.sem_wait with option SEM_WAIT_NONBLOCK.

> The "down" or "try_down" are almost the solutions, but not quite.
> 
> "down" may block forever ---> there is no timeout at all
>                               (I need some)
> 
> "try_down" --->  the timeout is zero
>                  (I prefer a non-zero timeout)
> 
> Obviously, I could use "try_down" in a loop, checking the system time myself and then either give up or actually manage to decrement the semaphore.
> 
> Is there a better solution than this?

The POSIX API specifies sem_timedwait, but I haven't included it in my
bindings, as it is optional in POSIX. What you could do as workaround is
to start a helper thread which will sem_post after the timeout. I admit
that it is difficult to find a clean solution here without race
conditions.

> (There is a Unix module, there are signals, but I am not sure whether it is safe to use them in multithreaded program.
>  At least, I did not have a luck.)

The signal handling as provided by Unix (or better by the OCaml runtime)
is not sufficient for this kind of programming.

An alternative might be to use a pipe as returned by Unix.pipe. The
number of bytes in the pipe buffer is the semaphore counter:

post = add a byte to the buffer
wait = read a byte from the buffer

You can use Unix.select to specify a timeout when waiting. The solution
works as long as the pipe buffer doesn't fill up (at least 4096 bytes
are guaranteed). You can totally avoid that if you combine the pipe with
an additional counter (i.e. semaphore counter = pipe buffer + additional
counter), and you use the pipe only when the additional counter is 0.

Gerd


> 
> Thanks in advance for patience &| help.
> 

-- 
------------------------------------------------------------
Gerd Stolpmann, Darmstadt, Germany    gerd@gerd-stolpmann.de
My OCaml site:          http://www.camlcity.org
Contact details:        http://www.camlcity.org/contact.html
Company homepage:       http://www.gerd-stolpmann.de
------------------------------------------------------------

[-- Attachment #2: This is a digitally signed message part --]
[-- Type: application/pgp-signature, Size: 490 bytes --]