MLton MLTONWIKIVERSION

MLtonThread

Home Index

    signature MLTON_THREAD =
       sig
          structure AtomicState:
             sig
                datatype t = NonAtomic | Atomic of int
             end

          val atomically: (unit -> 'a) -> 'a
          val atomicBegin: unit -> unit
          val atomicEnd: unit -> unit
          val atomicState: unit -> AtomicState.t

          structure Runnable:
             sig
                type t
             end

          type 'a t

          val atomicSwitch: ('a t -> Runnable.t) -> 'a
          val new: ('a -> unit) -> 'a t
          val prepend: 'a t * ('b -> 'a) -> 'b t
          val prepare: 'a t * 'a -> Runnable.t
          val switch: ('a t -> Runnable.t) -> 'a
       end

`MLton.Thread` provides access to MLton's user-level thread implementation (i.e. not OS-level threads). Threads are lightweight data structures that represent a paused computation. Runnable threads are threads that will begin or continue computing when `switch`ed to. `MLton.Thread` does not include a default scheduling mechanism, but it can be used to implement both preemptive and non-preemptive threads.

- `type AtomicState.t`
  the type of atomic states.

- `atomically f`
  runs `f` in a critical section.

- `atomicBegin ()`
  begins a critical section.

- `atomicEnd ()`
  ends a critical section.

- `atomicState ()`
  returns the current atomic state.

- `type Runnable.t`
  the type of threads that can be resumed.

- `type 'a t`
  the type of threads that expect a value of type `'a`.

- `atomicSwitch f`
  like `switch`, but assumes an atomic calling context. Upon `switch`ing back to the current thread, an implicit `atomicEnd` is performed.

- `new f`
  creates a new thread that, when run, applies `f` to the value given to the thread. `f` must terminate by `switch`ing to another thread or exiting the process.

- `prepend (t, f)`
  creates a new thread (destroying `t` in the process) that first applies `f` to the value given to the thread and then continues with `t`. This is a constant time operation.

- `prepare (t, v)`
  prepares a new runnable thread (destroying `t` in the process) that will evaluate `t` on `v`.

- `switch f`
  applies `f` to the current thread to get `rt`, and then start running thread `rt`. It is an error for `f` to perform another `switch`. `f` is guaranteed to run atomically.

## Example of non-preemptive threads

    structure Queue:
       sig
          type 'a t

          val new: unit -> 'a t
          val enque: 'a t * 'a -> unit
          val deque: 'a t -> 'a option
       end =
       struct
          datatype 'a t = T of {front: 'a list ref, back: 'a list ref}

          fun new() = T{front = ref [], back = ref []}

          fun enque(T{back, ...}, x) = back := x :: !back

          fun deque(T{front, back}) =
             case !front of
                [] => (case !back of
                          [] => NONE
                        | l => let val l = rev l
                               in case l of
                                  [] => raise Fail "deque"
                                | x :: l => (back := []; front := l; SOME x)
                               end)
              | x :: l => (front := l; SOME x)
       end

    structure Thread:
       sig
          val exit: unit -> 'a
          val run: unit -> unit
          val spawn: (unit -> unit) -> unit
          val yield: unit -> unit
       end =
       struct
          open MLton
          open Thread

          val topLevel: Thread.Runnable.t option ref = ref NONE

          local
             val threads: Thread.Runnable.t Queue.t = Queue.new()
          in
             fun ready (t: Thread.Runnable.t) : unit =
                Queue.enque(threads, t)
             fun next () : Thread.Runnable.t =
                case Queue.deque threads of
                   NONE => valOf(!topLevel)
                 | SOME t => t
          end

          fun 'a exit(): 'a = switch(fn _ => next())

          fun new(f: unit -> unit): Thread.Runnable.t =
             Thread.prepare
             (Thread.new (fn () => ((f() handle _ => exit())
                                    ; exit())),
              ())

          fun schedule t = (ready t; next())

          fun yield(): unit = switch(fn t => schedule (Thread.prepare (t, ())))

          val spawn = ready o new

          fun run(): unit =
             (switch(fn t =>
                     (topLevel := SOME (Thread.prepare (t, ()))
                      ; next()))
              ; topLevel := NONE)
       end

    val rec loop =
       fn 0 => ()
        | n => (print(concat[Int.toString n, "\n"])
                ; Thread.yield()
                ; loop(n - 1))

    val rec loop' =
       fn 0 => ()
        | n => (Thread.spawn(fn () => loop n); loop'(n - 2))

    val _ = Thread.spawn(fn () => loop' 10)

    val _ = Thread.run()

    val _ = print "success\n"

## Example of preemptive threads

    structure Queue:
       sig
          type 'a t

          val new: unit -> 'a t
          val enque: 'a t * 'a -> unit
          val deque: 'a t -> 'a option
       end =
       struct
          datatype 'a t = T of {front: 'a list ref, back: 'a list ref}

          fun new () = T {front = ref [], back = ref []}

          fun enque (T {back, ...}, x) = back := x :: !back

          fun deque (T {front, back}) =
             case !front of
                [] => (case !back of
                          [] => NONE
                        | l => let val l = rev l
                               in case l of
                                  [] => raise Fail "deque"
                                | x :: l => (back := []; front := l; SOME x)
                               end)
              | x :: l => (front := l; SOME x)
       end

    structure Thread:
       sig
          val exit: unit -> 'a
          val run: unit -> unit
          val spawn: (unit -> unit) -> unit
          val yield: unit -> unit
       end =
       struct
          open Posix.Signal
          open MLton
          open Itimer Signal Thread

          val topLevel: Thread.Runnable.t option ref = ref NONE

          local
             val threads: Thread.Runnable.t Queue.t = Queue.new ()
          in
             fun ready t = Queue.enque (threads, t)
             fun next () =
                case Queue.deque threads of
                   NONE => valOf (!topLevel)
                 | SOME t => t
          end

          fun 'a exit (): 'a = switch (fn _ => next ())

          fun new (f: unit -> unit): Thread.Runnable.t =
             Thread.prepare
             (Thread.new (fn () => ((f () handle _ => exit ())
                                    ; exit ())),
              ())

          fun schedule t = (ready t; next ())

          fun yield (): unit = switch (fn t => schedule (Thread.prepare (t, ())))

          val spawn = ready o new

          fun setItimer t =
             Itimer.set (Itimer.Real,
                         {value = t,
                          interval = t})

          fun run (): unit =
             (switch (fn t =>
                      (topLevel := SOME (Thread.prepare (t, ()))
                       ; new (fn () => (setHandler (alrm, Handler.handler schedule)
                                        ; setItimer (Time.fromMilliseconds 20)))))
              ; setItimer Time.zeroTime
              ; ignore alrm
              ; topLevel := NONE)
       end

    val rec delay =
       fn 0 => ()
        | n => delay (n - 1)

    val rec loop =
       fn 0 => ()
        | n => (delay 500000; loop (n - 1))

    val rec loop' =
       fn 0 => ()
        | n => (Thread.spawn (fn () => loop n); loop' (n - 1))

    val _ = Thread.spawn (fn () => loop' 10)

    val _ = Thread.run ()

    val _ = print "success\n"

Last edited on 2007-08-23 03:45:45 by MatthewFluet.