Worker Pools in Standard ML

Our example demonstrates how to implement a worker pool using threads and channels in Standard ML.

structure WorkerPool = struct
  (* We'll use the Thread structure for concurrency *)
  structure T = Thread

  (* We'll simulate channels using mutable references and condition variables *)
  type 'a chan = {
    queue: 'a list ref,
    mutex: T.mutex,
    notEmpty: T.cond,
    notFull: T.cond,
    capacity: int
  }

  fun makeChan capacity = {
    queue = ref [],
    mutex = T.mutex(),
    notEmpty = T.cond(),
    notFull = T.cond(),
    capacity = capacity
  }

  fun send (ch: 'a chan) (x: 'a) =
    let
      val {queue, mutex, notEmpty, notFull, capacity} = ch
    in
      T.withMutex mutex (fn () =>
        (while (List.length (!queue) >= capacity) do
           T.waitCond (notFull, mutex);
         queue := x :: !queue;
         T.signalCond notEmpty))
    end

  fun receive (ch: 'a chan) =
    let
      val {queue, mutex, notEmpty, notFull, ...} = ch
    in
      T.withMutex mutex (fn () =>
        (while (null (!queue)) do
           T.waitCond (notEmpty, mutex);
         case !queue of
           x :: xs => (queue := xs;
                       T.signalCond notFull;
                       x)
         | [] => raise Fail "Impossible: empty queue"))
    end

  (* Worker function *)
  fun worker (id, jobs, results) =
    let
      fun loop () =
        (case receive jobs of
           NONE => ()  (* End of jobs *)
         | SOME j =>
             (print ("worker " ^ Int.toString id ^ " started  job " ^ Int.toString j ^ "\n");
              T.delay (Time.fromSeconds 1);  (* Simulate work *)
              print ("worker " ^ Int.toString id ^ " finished job " ^ Int.toString j ^ "\n");
              send results (SOME (j * 2));
              loop ()))
    in
      loop ()
    end

  (* Main function *)
  fun main () =
    let
      val numJobs = 5
      val jobs = makeChan numJobs
      val results = makeChan numJobs

      (* Start 3 worker threads *)
      val _ = List.tabulate (3, fn w =>
        T.fork (fn () => worker (w + 1, jobs, results)))

      (* Send jobs *)
      val _ = List.app (fn j => send jobs (SOME j)) (List.tabulate (numJobs, fn x => x + 1))
      val _ = List.tabulate (3, fn _ => send jobs NONE)  (* Send end signals *)

      (* Collect results *)
      val _ = List.tabulate (numJobs, fn _ => receive results)
    in
      ()
    end
end

(* Run the main function *)
val _ = WorkerPool.main ()

This Standard ML implementation simulates the behavior of the original Go example. Here’s a breakdown of the changes and adaptations:

We use the Thread structure for concurrency, which provides similar functionality to goroutines.
Channels are simulated using a custom implementation with mutable references, mutexes, and condition variables.
The worker function is similar to the Go version, but it uses our custom channel operations (send and receive).
In the main function, we create channels, start worker threads, send jobs, and collect results, mirroring the structure of the Go example.
We use Thread.delay to simulate work instead of time.Sleep.
Error handling and some concurrency patterns are slightly different due to language differences.

When you run this program, you should see output similar to the Go version, with workers starting and finishing jobs concurrently. The exact order of output may vary due to the nature of concurrent execution.

Note that Standard ML’s concurrency model is different from Go’s, so this implementation is an approximation of the original behavior using available language features.

Tickers in Standard ML

Waitgroups in Standard ML

Learn X By Example

Title here

Worker Pools in Standard ML