lib/cml/cml-lib/multicast.sml

   1 (* multicast.sml
   2  * 2004 Matthew Fluet (mfluet@acm.org)
   3  *  Ported to MLton threads.
   4  *)
   5
   6 (* multicast.sml
   7  *
   8  * COPYRIGHT (c) 1994 AT&T Bell Laboratories.
   9  *
  10  * Asynchronous multicast (one-to-many) channels.  This implementation
  11  * is based on a condition variable implementation of multicast channels.
  12  * See Chapter 5 of "Concurrent Programming in ML" for details.
  13  *)
  14
  15 structure Multicast : MULTICAST =
  16    struct
  17
  18       structure SV = SyncVar
  19
  20       type 'a event = 'a CML.event
  21
  22       datatype 'a request =
  23          Message of 'a
  24        | NewPort
  25       datatype 'a mc_state = MCState of ('a * 'a mc_state SV.ivar)
  26       datatype 'a port =
  27          Port of (('a * 'a mc_state SV.ivar) CML.chan * 'a mc_state SV.ivar SV.mvar)
  28       datatype 'a mchan =
  29          MChan of ('a request CML.chan * 'a port CML.chan)
  30
  31     fun mkPort cv =
  32        let
  33           val outCh = CML.channel()
  34           val stateVar = SV.mVarInit cv
  35           fun tee cv =
  36              let
  37                 val (MCState(v, nextCV)) = SV.iGet cv
  38              in
  39                 CML.send (outCh, (v, nextCV))
  40                 ; tee nextCV
  41              end
  42           val _ = CML.spawn (fn () => tee cv)
  43        in
  44           Port(outCh, stateVar)
  45        end
  46
  47     fun mChannel () =
  48        let
  49           val reqCh = CML.channel()
  50           and replyCh = CML.channel()
  51           fun server cv =
  52              case (CML.recv reqCh) of
  53                 NewPort =>
  54                    (CML.send (replyCh, mkPort cv)
  55                     ; server cv)
  56               | (Message m) =>
  57                    let
  58                       val nextCV = SV.iVar()
  59                    in
  60                       SV.iPut (cv, MCState(m, nextCV))
  61                       ; server nextCV
  62                    end
  63           val _ = CML.spawn (fn () => server (SV.iVar()))
  64        in
  65           MChan(reqCh, replyCh)
  66        end
  67
  68     fun multicast (MChan(ch, _), m) = CML.send (ch, Message m)
  69
  70     fun port (MChan(reqCh, replyCh)) =
  71        (CML.send (reqCh, NewPort)
  72         ; CML.recv replyCh)
  73
  74     fun copy (Port(_, stateV)) = mkPort(SV.mGet stateV)
  75
  76     fun recvMsg stateV (v, nextCV) =
  77        let val _ = SV.mSwap (stateV, nextCV)
  78        in v
  79        end
  80
  81     fun recv (Port(ch, stateV)) = recvMsg stateV (CML.recv ch)
  82     fun recvEvt (Port(ch, stateV)) = CML.wrap(CML.recvEvt ch, recvMsg stateV)
  83    end
  84