regression/prodcons.sml

   1 open Posix.Signal MLton.Signal
   2
   3 (* Translated from prodcons.ocaml. *)
   4 fun for (start, stop, f) =
   5    let
   6       fun loop i =
   7          if i > stop
   8             then ()
   9          else (f i; loop (i + 1))
  10    in
  11       loop start
  12    end
  13
  14 fun print s = ()
  15
  16 structure Queue:
  17    sig
  18       type 'a t
  19
  20       val new: unit -> 'a t
  21       val enque: 'a t * 'a -> unit
  22       val deque: 'a t -> 'a option
  23    end =
  24    struct
  25       datatype 'a t = T of {front: 'a list ref, back: 'a list ref}
  26
  27       fun new () = T {front = ref [], back = ref []}
  28
  29       fun enque (T {back, ...}, x) = back := x :: !back
  30
  31       fun deque (T {front, back}) =
  32          case !front of
  33             [] => (case !back of
  34                       [] => NONE
  35                     | l => let val l = rev l
  36                            in case l of
  37                               [] => raise Fail "deque"
  38                             | x :: l => (back := []; front := l; SOME x)
  39                            end)
  40           | x :: l => (front := l; SOME x)
  41    end
  42
  43 structure Thread:
  44    sig
  45       val exit: unit -> 'a
  46       val run: unit -> unit
  47       val spawn: (unit -> unit) -> unit
  48       val yield: unit -> unit
  49       structure Mutex:
  50          sig
  51             type t
  52
  53             val new: unit -> t
  54             val lock: t * string -> unit
  55             val unlock: t -> unit
  56          end
  57       structure Condition:
  58          sig
  59             type t
  60
  61             val new: unit -> t
  62             val signal: t -> unit
  63             val wait: t * Mutex.t -> unit
  64          end
  65    end =
  66    struct
  67       open MLton
  68       open Itimer Signal Thread
  69
  70       val topLevel: Thread.Runnable.t option ref = ref NONE
  71
  72       local
  73          val threads: Thread.Runnable.t Queue.t = Queue.new ()
  74       in
  75          fun ready t: unit = Queue.enque (threads, t)
  76          fun next () : Thread.Runnable.t =
  77             case Queue.deque threads of
  78                NONE => (print "switching to toplevel\n"
  79                         ; valOf (!topLevel))
  80              | SOME t => t
  81       end
  82
  83       fun 'a exit (): 'a = switch (fn _ => next ())
  84
  85       fun new (f: unit -> unit): Thread.Runnable.t =
  86          Thread.prepare
  87          (Thread.new (fn () => ((f () handle _ => exit ())
  88                                 ; exit ())),
  89           ())
  90
  91       fun schedule t =
  92          (print "scheduling\n"
  93           ; ready t
  94           ; next ())
  95
  96       fun yield (): unit = switch (fn t => schedule (Thread.prepare (t, ())))
  97
  98       val spawn = ready o new
  99
 100       fun setItimer t =
 101          Itimer.set (Itimer.Real,
 102                      {value = t,
 103                       interval = t})
 104
 105       fun run (): unit =
 106          (switch (fn t =>
 107                   (topLevel := SOME (Thread.prepare (t, ()))
 108                    ; new (fn () => (setHandler (alrm, Handler.handler schedule)
 109                                     ; setItimer (Time.fromMilliseconds 20)))))
 110           ; setItimer Time.zeroTime
 111           ; ignore alrm
 112           ; topLevel := NONE)
 113
 114       structure Mutex =
 115          struct
 116             datatype t = T of {locked: bool ref,
 117                                waiting: unit Thread.t Queue.t}
 118
 119             fun new () =
 120                T {locked = ref false,
 121                   waiting = Queue.new ()}
 122
 123             fun lock (T {locked, waiting, ...}, name) =
 124                let
 125                   fun loop () =
 126                      (print (concat [name, " lock looping\n"])
 127                       ; Thread.atomicBegin ()
 128                       ; if !locked
 129                            then (print "mutex is locked\n"
 130                                  ; switch (fn t =>
 131                                            (Thread.atomicEnd ()
 132                                             ; Queue.enque (waiting, t)
 133                                             ; next ()))
 134                                  ; loop ())
 135                         else (print "mutex is not locked\n"
 136                               ; locked := true
 137                               ; Thread.atomicEnd ()))
 138                in loop ()
 139                end
 140
 141             fun safeUnlock (T {locked, waiting, ...}) =
 142                (locked := false
 143                 ; (case Queue.deque waiting of
 144                       NONE => ()
 145                     | SOME t => (print "unlock found waiting thread\n"
 146                                  ; ready (Thread.prepare (t, ())))))
 147
 148             fun unlock (m: t) =
 149                (print "unlock atomicBegin\n"
 150                 ; Thread.atomicBegin ()
 151                 ; safeUnlock m
 152                 ; Thread.atomicEnd ())
 153          end
 154
 155       structure Condition =
 156          struct
 157             datatype t = T of {waiting: unit Thread.t Queue.t}
 158
 159             fun new () = T {waiting = Queue.new ()}
 160
 161             fun wait (T {waiting, ...}, m) =
 162                (switch (fn t =>
 163                         (Mutex.safeUnlock m
 164                          ; print "wait unlocked mutex\n"
 165                          ; Queue.enque (waiting, t)
 166                          ; next ()))
 167                 ; Mutex.lock (m, "wait"))
 168
 169             fun signal (T {waiting, ...}) =
 170                case Queue.deque waiting of
 171                   NONE => ()
 172                 | SOME t => ready (Thread.prepare (t, ()))
 173          end
 174
 175    end
 176
 177 structure Mutex = Thread.Mutex
 178 structure Condition = Thread.Condition
 179
 180 val count = ref 0
 181 val data = ref 0
 182 val produced = ref 0
 183 val consumed = ref 0
 184 val m = Mutex.new ()
 185 val c = Condition.new ()
 186
 187 fun producer n =
 188    for (1, n, fn i =>
 189         (print (concat ["producer acquiring lock ", Int.toString i, "\n"])
 190          ; Mutex.lock (m, "producer")
 191          ; print "producer acquired lock\n"
 192          ; while !count = 1 do Condition.wait (c, m)
 193          ; print "producer passed condition\n"
 194          ; data := i
 195          ; count := 1
 196          ; Condition.signal c
 197          ; print "producer releasing lock\n"
 198          ; Mutex.unlock m
 199          ; print "producer released lock\n"
 200          ; produced := !produced + 1))
 201
 202 fun consumer n =
 203    let val i = ref 0
 204    in
 205       while !i <> n do
 206          (print (concat ["consumer acquiring lock ", Int.toString (!i), "\n"])
 207           ; Mutex.lock (m, "consumer")
 208           ; print "consumer acquired lock\n"
 209           ; while !count = 0 do Condition.wait (c, m)
 210           ; i := !data
 211           ; count := 0
 212           ; Condition.signal c
 213           ; print "consumer releasing lock\n"
 214           ; Mutex.unlock m
 215           ; print "consumer released lock\n"
 216           ; consumed := !consumed + 1)
 217    end
 218
 219 fun atoi s = case Int.fromString s of SOME num => num | NONE => 0
 220 fun printl [] = TextIO.print "\n" | printl (h::t) = ( TextIO.print h ; printl t )
 221
 222 fun main (name, args) =
 223    let
 224       val n = atoi (hd (args @ ["1"]))
 225       val p = Thread.spawn (fn () => producer n)
 226       val c = Thread.spawn (fn () => consumer n)
 227       val _ = Thread.run ()
 228       val _ = Posix.Process.sleep (Time.fromSeconds 1)
 229       val _ = printl [Int.toString (!produced),
 230                       " ",
 231                       Int.toString (!consumed)]
 232    in
 233       ()
 234    end
 235
 236 val _ = main ( "prodcons", ["100000"] )