module Cf_gadget:Monadic composition of complex stream processors. An experimental interface for constructing interactive functional systems in a single thread of control.sig
..end
This module implements a marginally more general version of the Gadget system described in Chapter 30 of Magnus Carlsson's and Thomas Hallgren's joint Ph.D. thesis.
In the context of this module, a "gadget" is a monad that evaluates into
a Cf_flow
object, capable of alternately reading from a source of input
values and writing to a sink of output values. The continuation monad is
specialized over an abstract "work" monad type, and a scheduler handles
the calls and jumps between multiple simultaneous work units, communicating
with one another over a very lightweight message passing abstraction called
a "wire".
The abstract work monad is a kind of state-continuation monad for
operations over the internal Cf_flow
value. The operations it supports
are lifted into the gadget monad, and they are summarized as follows:
A wire is logically composed of a receiver and a transmitter, with weak mutual references between them. When either end of the wire is reclaimed by the memory allocator, the other end is automatically rendered into a null wire, i.e. receivers never get messages and transmitters put messages by discarding them.
A pair of classes are provided to represent the receiver and the
transmitter on a wire. Objects of the rx
class define a get
method for
creating a "gate" that can receive a message. Objects of the tx
class
define a put
method for transmitting a message. Both objects can be
constructed with a wire object, and a convenience operators are defined for
creating a new wire and construction a pair of associated rx
and tx
objects.
Any gadget may read from the internal input stream or write to the external output stream. Conventionally, it is often simpler to define a a reader gadget and a writer gadget to localize these effects.
Note: see Magnus Carlsson's and Thomas Hallgren's joint
Ph.D. thesis for a complete
dissertation on the nature of the system of concepts behind this module.
Types
type ('a, 'b)
work
Cf_flow
object.type ('a, 'b)
gate
guard
function.type ('a, 'b, 'c)
wire
'x
from a sender to a
a receiver in a ('i, 'o) work
continuation.type('a, 'b, 'c)
guard =(('a, 'b) gate, 'c) Cf_cmonad.t
type('a, 'b, 'c)
t =(('a, 'b) work, 'c) Cf_cmonad.t
val eval : ('a, 'b, unit) t -> ('a, 'b) Cf_flow.t
eval y
to obtain a new flow by evaluating the gadget monad y
.val start : ('a, 'b, unit) t -> ('a, 'b, unit) t
start y
to start a new gadget evaluating the gadget y
.val guard : ('a, 'b, unit) guard -> ('a, 'b, 'c) t
guard m
to receive the next message guarded by m
. The continuation
bound to the result is discarded and control passes to the scheduler.val abort : ('a, 'b, 'c) t
abort
to abort gadgeting and return to the scheduler. This is a
convenient shortcut for guard Cf_cmonad.nil
.val wire : ('a, 'b, ('c, 'a, 'b) wire) t
wire
to return a new wire for carrying messages of type 'x
.val wirepair : ('a, 'b, ('c, 'a, 'b) wire * ('d, 'a, 'b) wire)
t
wirepair
to return a pair of new wires for carrying messages of type
'x
and 'y
.val null : ('a, 'b, ('c, 'a, 'b) wire) t
null
to construct a wire that discards every message transmitted
without ever delivering it. Such wires can be useful for default arguments
to some gadget functions.val read : ('a, 'b, 'a) t
read
to get the next input value from the external stream.val write : 'a -> ('b, 'a, unit) t
write obj
to put the next output value into the
external stream.class type connector =object
..end
class[['a, 'b, 'c]]
rx :('a, 'b, 'c) wire ->
object
..end
class[['a, 'b, 'c]]
tx :('a, 'b, 'c) wire ->
object
..end
val connect : ('a, 'b, ('c, 'a, 'b) wire) t ->
('a, 'b, ('c, 'a, 'b) rx * ('c, 'a, 'b) tx) t
connect m
to construct a new matching pair of rx
and tx
objects
from the wire returned by m
.val simplex : ('a, 'b, ('c, 'a, 'b) rx * ('c, 'a, 'b) tx) t
simplex
to construct a new matching pair of rx
and tx
objects.
This is a convenient abbreviation of connect wire
.type('a, 'b, 'c, 'd)
pad =('a, 'c, 'd) rx * ('b, 'c, 'd) tx
pad
comprises a
receiver for control events and a transmitter for notification events, and
a fix
comprises the transmitter for control events and the receiver for
notification eventstype('a, 'b, 'c, 'd)
fix =('b, 'c, 'd) rx * ('a, 'c, 'd) tx
val connectpair : ('a, 'b, ('c, 'a, 'b) wire * ('d, 'a, 'b) wire)
t ->
('a, 'b, ('c, 'd, 'a, 'b) fix * ('c, 'd, 'a, 'b) pad)
t
connectpair m
to construct a new duplex communication channel,
composed with the wire pair returned by m
. A matching fix
and pad
of
the channel are returned.val duplex : ('a, 'b, ('c, 'd, 'a, 'b) fix * ('c, 'd, 'a, 'b) pad)
t
duplex
to construct a new duplex communication channel, composed of
two wires each in opposite flow. A matching fix
and pad
for each
channel are returned. This is a convenient abbreviation of
connectpair wirepair
.val wrap : ('a, 'b, 'c) #rx ->
('d, 'b, 'c) #tx ->
('a, 'd) Cf_flow.t -> ('b, 'c, unit) t
wrap rx tx w
to start a new gadget that wraps the flow w
, so that
it reads output from the flow (copying it to tx
object) and writes input
to the flow (copying it from the rx
object).class virtual[['a, 'b]]
next :object
..end
inherit ['i, 'o] next
to derive a class that implements an
intermediate state in a machine.
class virtual[['a, 'b]]
start :object
..end
inherit ['i, 'o] start
to derive a class to represent the
initial state of a machine.
val create : (('a, 'b, 'c, 'd) pad -> ('c, 'd) #start) ->
('c, 'd, ('a, 'b, 'c, 'd) fix) t
create f
to create a duplex channel, and apply f
to the resulting
pad
to obtain the initial state of a machine. The machine is started and
the corresponding fix
is returned.val createM : (('a, 'b, 'c, 'd) pad ->
('c, 'd, ('c, 'd) #start) t) ->
('c, 'd, ('a, 'b, 'c, 'd) fix) t
createM f
to create a duplex channel, and apply f
to the resulting
pad
to obtain a continuation monad that evaluates to the initial state of
a machine. The machine is started and the corresponding fix
is returned.