The Fudgets implementation of Gadgets is purely functional,
written in Haskell, which means that all primitives described
above are defined within Haskell. The ``runtime'' system (the
process scheduler) is also written in Haskell, except that it
uses a
(not defined in ordinary Haskell) at one
place, as we will see.
In the functional version, processes cannot have the simple
function type s -> s any more, since we
must be explicit about the effects that processes can
have. Instead, we will define the process type in steps, where
we start with a stream-processor type that handles messages
related to the keyboard, mouse and screen. On top of the
stream-processor type, we define a state monad (SPms)
with operations for manipulating a state in addition to the I/O
operations of the stream processor. The state is used by the
scheduler, and is used to define a simple process type Process0, which amounts to the Gadget processes except that
they do not have any local state. Having done this, we define
the full Gadget processes on top. The steps are summarised in
the following table.
We use the state stream-processor monad to implement the
stateless processes, called
Process0. The state of the
stream processor is used by the scheduler for bookkeeping.
type Process0 i o = SPms i o (SchedulerState i o) ()
data SchedulerState i o = SS{ freeWire :: Wno
, messageQs :: MessageQueues
, ready :: [Process0 i o]
, guarded :: [Guarded0 i o]
, input :: [i -> Process0 i o]
}
Just as in the Gofer implementation, we use integers to
identify wire ends, except that we call the integers
wire
numbers (
Wno).
newtype Wno = Wno Int
newtype In a = In Wno
newtype Out a = Out Wno
What follows are definitions of the primitives for creating
wires and processes, and communication over wires. We suffix
the primitives with a
0 to indicate that they operate
on processes without local state.
A new wire is allocated with primWire0, which increments
the field freeWire in the state, and hands a fresh wire
to the continuation.
primWire0 :: (Wire a -> Process0 i o) -> Process0 i o
primWire0 c =
do ps@(SS{ freeWire = w@(Wno i) }) <- loadSPms
storeSPms ps{ freeWire = Wno (i+1) }
c (In w, Out w)
The second component in the scheduler state (
messageQs)
is a mapping from wire numbers to queues of not yet delivered
messages.
type MessageQueues = IntMap (Queue Msg)
The types
IntMap and
Queue implement integer
maps and Okasaki's queues [
Oka95] come from HBC's
library, and have the following signatures:
module Queue where
empty :: Queue a
snoc :: a -> Queue a -> Queue a
tail :: Queue a -> Queue a
head :: Queue a -> a
null :: Queue a -> Bool
module IntMap where
empty :: IntMap a
modify :: (a -> a) -> a -> Int -> IntMap a -> IntMap a
delete :: Int -> IntMap a -> IntMap a
lookup :: Int -> IntMap a -> Maybe a
The operations are standard, except
modify, which
deserves an explanation. The expression
modify f
a i m applies the function
f
to the entry
i in
m if it exists. Otherwise, it
inserts the value
a at
i.
Each message is paired with the wire number. Since different
wires can have different type, messages can also be of
different type. We use an existential type (an extension to
Haskell provided by HBC) to hide the message type when putting
messages in the queue.
data Msg = Msg ?a
Constructing values of type
Msg is easy, but when
de-constructing them, we cannot assume anything about the type of
the argument. We return to this problem later.
Sending a value on a wire amounts to queueing the wire number
together with the value.
primTx0 :: Out a -> a -> Process0 i o -> Process0 i o
primTx0 (Out wno) msg p =
if wno == ncWno then p
else
do ps@(SS{ messageQs, ready }) <- loadSPms
storeSPms ps{ messageQs = addMsg wno (Msg msg) messageQs
, ready = p:ready}
scheduler
addMsg :: Wno -> Msg -> MessageQueues -> MessageQueues
addMsg wno m = modify (snoc m) (snoc m Queue.empty) wno
The field
ready holds a list of processes that are ready
to run. When spawning off a new process, we put it on the ready
list.
primSpawn0 :: Process0 i o -> Process0 i o -> Process0 i o
primSpawn0 p' p =
do ps@(SS{ ready }) <- loadSPms
storeSPms ps{ ready = p':ready }
p
There is also a list of processes waiting for messages, stored
in the field
guarded. The elements are lists of stateless
guarded processes (
AGuarded0 i o).
data AGuarded0 i o = AGuarded0 Wno (?a -> Process0 i o)
A guarded process is a wire number and a function which takes a
message as a parameter. The actual type of the message is hidden
in
AGuarded0, so that we can form a list of guarded
processes regardless of what message type they are waiting
for.
type Guarded0 i o = [AGuarded0 i o]
A guarded stateless process is formed with
primFrom0.
primFrom0 :: In m -> (m -> Process0 i o) -> AGuarded0 i o
primFrom0 (In wno) f = AGuarded0 wno f
The function
primRx0 will wait for a message to arrive to
any of the guarded processes in the first parameter. It adds the
guarded processes to the state, and then jump to the
scheduler to find another process to execute.
primRx0 :: Guarded0 i o -> Process0 i o -> Process0 i o
primRx0 g def =
do ps@(SS{ guarded }) <- loadSPms
storeSPms ps{ guarded = g:guarded }
scheduler
The scheduler's (
Figure 83) job is to apply guarded
processes to matching messages, move them to the ready list,
and pick one from the ready list to run. In case the ready
list is empty, the
input list is investigated. This list
contains processes waiting for input from the outside of the
stream processor. If this list is also empty, then the gadget
program is finished. Otherwise, we do stream-processor input
and give the message to all processes in the input list.
scheduler :: Process0 i o
scheduler =
do ps@(SS{ freeWire, messageQs, ready, guarded, input }) <- loadSPms
let (messageQs',guarded',moreReady) = match messageQs guarded
let run p ready' input' =
do storeSPms ps{ messageQs = messageQs'
, ready = ready'
, guarded = guarded'
, input = input'
}
p
case (moreReady++ready) of
[] -> if null input
then nullSPms
else do i <- getSPms
case [ih i | ih <- input] of
p:ready' -> run p ready' []
p:ready' -> run p ready' input
match :: MessageQueues -> [Guarded0 i o]
-> (MessageQueues,[Guarded0 i o],[Process0 i o])
match m [] = (m,[],[])
match m (g:f) = case match1 m g of
Nothing -> (m',g:f',r) where (m',f',r) = match m f
Just (m1,p) -> (m2,f',p:r) where (m2,f',r) = match m1 f
match1 :: MessageQueues -> Guarded0 i o
-> Maybe (MessageQueues,Process0 i o)
match1 m [] = Nothing
match1 m ((AGuarded0 (Wno w) f):gs) =
case IntMap.lookup w m of
Nothing -> match1 m gs
Just mq -> case Queue.head mq of
Msg msg -> Just (m',cast f msg) -- ! type cast !
where mq' = Queue.tail mq
m' = if Queue.null mq'
then delete w m
else modify Queue.tail undefined w m
cast :: a -> b -- Not defined in Haskell. |
Figure 83. The scheduler.
The function match applies all guarded processes for
which there are matching messages. It returns the remaining
unmatched messages and guarded processes, together with a list
of new ready processes.
Recall that each element in the field guarded is itself
a list, which comes from a call to primRx. The function
match1 looks for a matching message for one of the
elements in such a list, possibly returning a new message
queue and a ready process. A matching message must have the
same wire number as the guarded process. It seems like this
cannot be expressed in the type system, so we are forced to
use a type cast (see the function match1 in
Figure 83).
The stateless processes can do stream-processor input/output
by means of get0 and put0. The output part is easy:
put0 :: o -> Process0 i o -> Process0 i o
put0 o p =
do putSPms (Right o)
p
When it comes to input, the process does not directly call
getSPms, since that would block other threads as
well. Instead, the continuation is put on the input list in
the scheduler state, and jump to the scheduler. Note that more
than one process may call
get0. As we have already
seen, the scheduler will ensure that all of them will receive
the next message that the stream processor inputs.
get0 :: (i -> Process0 i o) -> Process0 i o
get0 i =
do ps@(SS{ input }) <- loadSPms
storeSPms ps{ input = i:input }
scheduler
If a process terminates, we need to schedule some other
process for execution if possible. Therefore,
primTerminate0 simply jumps to the scheduler.
primTerminate0 :: Process0 i o
primTerminate0 = scheduler
To launch a process, the process state must be
initialised. This is done in
primLaunch0.
primLaunch0 :: Process0 i o -> Process0 i o
primLaunch0 p =
do storeSPms SS{ freeWire = startWno
, messageQs = IntMap.empty
, ready = []
, guarded = []
, input = []
}
p
So far, we have been quite general about the type of messages
that our stateless processes will speak. To implement gadget
processes, we will use the stream-processor I/O to simulate
the keyboard, mouse and screen, as discussed in
Section 31.1.1. We will call stateless gadget
processes
GProcess0.
type GProcess0 = Process0 GEvent GCommand
The types
GEvent and
GCommand will be defined in
Section 31.2.4.
Now, we have defined most of the necessary primitive
operations required for Gadget processes, except for the ones
that manipulate a local state. It turns out to be
straightforward to add state to
GProcess0:
newtype Process s = P ((s -> GProcess0) -> s -> GProcess0)
A stateful process is a process-valued function which takes a
stateless process continuation (parameterised over its input
state), and an input state as parameters. It can modify the
state before applying it to the continuation, and also use the
stateless process primitives.
The state parameter is accessed by setState and readState.
unp (P p) = p
setState :: s -> Process s -> Process s
setState a p = P $ \c s -> unp p c a
readState :: (s -> Process s) -> Process s
readState p = P $ \c s -> unp (p s) c s
We now need to lift the primitive operations of type
GProcess0 to
Process. We use two auxiliary functions,
depending on whether the continuation takes an argument or
not. (This ``duplication of code'' is a price we pay for not
working with monads: in monadic style, all operations return a
value, which might be
() if it is uninteresting. In
CPS, operations without a result take continuations without an
argument, which can be seen as a slight optimisation, but adds
to the complexity of CPS programming.)
liftP0arg :: ((a -> GProcess0) -> GProcess0)
-> (a -> Process s) -> Process s
liftP0arg p0 p = P $ \c s -> p0 (\a->unp (p a) c s)
liftP0c :: (GProcess0 -> GProcess0)
-> Process s -> Process s
liftP0c p0 p = P $ \c s -> p0 (unp p c s)
We also need to lift stateless processes into stateful ones:
liftP0 :: GProcess0 -> Process s
liftP0 p0 = P $ \c s -> p0
The operations for creating a wire and transmitting a message
are straightforward to lift.
primWire :: (Wire a -> Process s) -> Process s
primWire = liftP0arg primWire0
primTx :: Out o -> o -> Process s -> Process s
primTx o m = liftP0c $ primTx0 o m
We will also need an auxiliary function to ``downgrade'' a
stateful process to a function from state to a stateless process.
down :: Process s -> (s -> GProcess0)
down (P p) s = p (\s' -> primTerminate0) s
When lifting
primFrom, we must ensure that the
guarded processes get access to the state. Guarded stateful processes
are therefore guarded stateless processes parameterised over the state.
type AGuarded s = s -> AGuarded0 GEvent GCommand
type Guarded s = [AGuarded s]
primFrom :: In m -> (m -> Process s) -> AGuarded s
primFrom i p = \s -> primFrom0 i (\m -> down (p m) s)
In
primRx, we apply the state to each guarded process,
revealing the stateless guarded processes that
primRx0
accepts.
primRx :: Guarded s -> Process s
primRx gs = P $ \c s -> primRx0 [g s | g <- gs]
The remaining primitive operations are straightforward to
lift.
primTerminate :: Process s
primTerminate = P $ \c s -> primTerminate0
primSpawn :: Process a -> a -> Process s -> Process s
primSpawn p' s p = liftP0c (primSpawn0 (down p' s)) p
primLaunch :: Process s -> s -> GProcess0
primLaunch p s = primLaunch0 (down p s)
To complete the functional implementation of the Gadget
primitives, we still must define
mouse,
screen and
keyboard. We use the stream-processor input/output to
mediate the events and commands from/to the mouse, keyboard
and screen.
data GEvent = ME MouseEvnt | KE KeyboardEvnt | SE ScreenEvnt
data GCommand = MC MouseCmnd | KC KeyboardCmnd | SC [ScreenCmnd]
Each device is controlled by two processes: the
output
handler, which injects commands received on a wire into the
type
GCommand and outputs them, and the
input
handler, which inputs events, extracts those specific for the
device and transmit them on a wire. These two handlers run in
parallel. This is captured with the
deviceHandler.
deviceHandler :: (c -> GCommand) -> (GEvent -> Maybe e)
-> In c -> Out e -> Process s
deviceHandler inj extract cw ew =
liftP0 $ primSpawn0 ohandler ihandler
where ohandler = primRx0 [primFrom0 cw $ \cmd ->
put0 (inj cmd) ohandler]
ihandler = get0 $ \i ->
case extract i of
Just evt -> primTx0 ew evt $
ihandler
Nothing -> ihandler
We can now form our devices.
keyboard :: In KeyboardCmnd -> Out KeyboardEvnt -> Process s
keyboard = deviceHandler KC (\i -> case i of KE e -> Just e
_ -> Nothing)
mouse :: In MouseCmnd -> Out MouseEvnt -> Process s
mouse = deviceHandler MC (\i -> case i of ME e -> Just e
_ -> Nothing)
screen :: In [ScreenCmnd] -> Out ScreenEvnt -> Process s
screen = deviceHandler SC (\i -> case i of SE e -> Just e
_ -> Nothing)
Outside the Gadget stream processor, the screen commands are
transformed into corresponding Fudget drawing commands,
whereas the keyboard and mouse control commands are
ignored. Conversely, Fudget keyboard presses, mouse clicks and
screen exposure events are transformed into
GEvent
messages. This is done in the fudget
gadgetF, of type
gadgetF :: Gadget -> F a b
Note that the high-level streams of
gadgetF are
unused. It would be nice to use them for communication between
gadget processes and the rest of the fudget program, but this
is not possible in a type-safe way. The reason is that such a
communication could be used to exchange wires between
different instances of
gadgetF. Each
gadgetF has
its own scheduler, and mixing wires between schedulers is not
type safe.
