@@ -0,0 +1,443 @@
1+module Process
2+
3+import System.Concurrency.Raw
4+import public Data.List -- public, to get proof search machinery
5+
6+%access public
7+
8+-- Process IDs are parameterised by their interface. A request of type
9+-- 'iface t' will get a response of type 't'
10+data ProcID : (iface : Type -> Type) -> Type where
11+ MkPID : Ptr -> ProcID iface
12+
13+data ServerID : Type where
14+ MkServer : ProcID iface -> ServerID
15+
16+implicit MkServer' : ProcID iface -> ServerID
17+MkServer' = MkServer
18+
19+data Replied = YesR | NoR
20+
21+data ReqHandle = MkReqH Nat
22+
23+-- Current state of a process includes:
24+-- * the servers it currently has an open connection to
25+-- * the number of clients it currently has connected
26+-- * whether it has responded to a request yet
27+
28+-- Therefore, we can write process types which make clear that a process
29+-- cannot quit while it is talking to a server, or while it still has clients
30+-- expecting to communicate with it, or if it has not serviced any requests.
31+data ProcState : Type where
32+ MkProcState : (servers : List ServerID) ->
33+ (clients : Nat) ->
34+ Replied ->
35+ ProcState
36+
37+data Pending : ReqHandle -> List (ReqHandle, Type) -> Type -> Type where
38+ PendingHere : Pending h ((h, t) :: hs) t
39+ PendingThere : Pending h hs t -> Pending h ((h', t') :: hs) t
40+
41+dropPending : (hs : List (ReqHandle, Type)) -> Pending h hs ty ->
42+ List (ReqHandle, Type)
43+dropPending ((h, t) :: xs) PendingHere = xs
44+dropPending ((h', t') :: xs) (PendingThere x)
45+ = ((h', t') :: dropPending xs x)
46+
47+data ConnectedTo : ServerID -> ProcState -> Type where
48+ IsConnectedTo : Elem p servers ->
49+ ConnectedTo p (MkProcState servers c reply)
50+
51+data NoClient : ProcState -> Type where
52+ IsNoClient : NoClient (MkProcState servers 0 reply)
53+
54+data OneClient : ProcState -> Type where
55+ IsOneClient : OneClient (MkProcState servers (S k) reply)
56+
57+data Reply : ProcState -> Type where
58+ IsReply : Reply (MkProcState s c YesR)
59+
60+data NoReply : ProcState -> Type where
61+ IsNoReply : NoReply (MkProcState s c NoR)
62+
63+{-- Some useful operations on process state --}
64+newClient : ProcState -> ProcState
65+newClient (MkProcState servers clients r)
66+ = MkProcState servers (S clients) r
67+
68+setClients : ProcState -> Nat -> ProcState
69+setClients (MkProcState servers clients r) k
70+ = MkProcState servers k r
71+
72+newServer : ProcID iface -> ProcState -> ProcState
73+newServer p (MkProcState servers clients r)
74+ = MkProcState (MkServer p :: servers) clients r
75+
76+dropServer : (pid : ProcID iface) -> (p : ProcState) ->
77+ ConnectedTo (MkServer pid) p -> ProcState
78+dropServer pid (MkProcState servers c r) (IsConnectedTo prf)
79+ = MkProcState (dropElem servers prf) c r
80+
81+{-
82+pendingReq : ProcID iface -> (h : ReqHandle) -> iface t ->
83+ ProcState hs -> ProcState ((h, t) :: hs)
84+pendingReq {t} p h x (MkProcState s c r) = MkProcState s c r
85+
86+doneReq : ProcState hs -> (p : Pending h hs) -> ProcState (dropPending hs p)
87+doneReq (MkProcState s c r) p = MkProcState s c r
88+-}
89+
90+replied : ProcState -> ProcState
91+replied (MkProcState servers clients r)
92+ = MkProcState servers clients YesR
93+
94+resetReplied : ProcState -> ProcState
95+resetReplied (MkProcState servers clients r)
96+ = MkProcState servers clients NoR
97+
98+runningServer : Nat -> ProcState
99+runningServer c = MkProcState [] (S c) NoR
100+
101+doneServer : ProcState
102+doneServer = MkProcState [] 0 YesR
103+
104+init : List ServerID -> ProcState
105+init s = MkProcState s 0 NoR
106+
107+{-- Processes themselves.
108+
109+A process returns some type 'a', responds to requests on the interface
110+'iface', and has an input and output state.
111+--}
112+mutual
113+ data Process : (a : Type) -> (iface : Type -> Type) ->
114+ List (ReqHandle, Type) -> (a -> List (ReqHandle, Type)) ->
115+ ProcState -> (a -> ProcState) ->
116+ Type where
117+ -- Some plumbing
118+ Lift' : IO a -> Process a iface hs (const hs) p (const p)
119+ Pure : a -> Process a iface hs (const hs) p (const p)
120+ Quit : a -> Process a iface hs (const hs) p (const (resetReplied p))
121+
122+ bind : Process a iface hs hs' p p' ->
123+ ((x : a) -> Process b iface (hs' x) hs'' (p' x) p'') ->
124+ Process b iface hs hs'' p p''
125+
126+ Fork : Process () serveri [] (const []) (runningServer 1) (const doneServer) ->
127+ Process (ProcID serveri) iface hs (const hs) p (\res => (newServer res p))
128+ Work : (worker : (pid : ProcID iface) -> Worker [pid] ()) ->
129+ (waiter : Process t iface hs (const hs) (runningServer 1) (const doneServer)) ->
130+ Process t iface hs (const hs) p (const p)
131+
132+ Request : (r : ProcID serveri) -> (x : serveri ty) ->
133+ {auto connected : ConnectedTo (MkServer r) p} ->
134+ Process ReqHandle iface
135+ hs (\h => (h, ty) :: hs)
136+ p (const p)
137+
138+ GetReply : (h : ReqHandle) ->
139+ {auto pending : Pending h hs ty} ->
140+ Process ty iface
141+ hs (const (dropPending hs pending))
142+ p (const p)
143+
144+ TimeoutRespond : (timeout : Int) ->
145+ (def : res) ->
146+ ({t : Type} -> (x : iface t) ->
147+ Response (t, res) iface hs p) ->
148+ Process res iface hs (const hs) p (const (replied p))
149+
150+ Respond : ({t : Type} -> (x : iface t) ->
151+ Response (t, res) iface hs p) ->
152+ Process res iface hs (const hs) p (const (replied p))
153+
154+ Connect : (r : ProcID serveri) ->
155+ Process Bool iface
156+ hs (const hs)
157+ p (\ok => case ok of
158+ True => newServer r p
159+ False => p)
160+
161+ Disconnect : (r : ProcID serveri) ->
162+ {auto connected : ConnectedTo (MkServer r) p} ->
163+ Process () iface
164+ hs (const hs)
165+ p (const (dropServer r p connected))
166+
167+ CountClients : Process Nat iface hs (const hs) p (\n => setClients p n)
168+
169+ -- FIXME: The process had better be guaranteed to change the system
170+ -- state (e.g. finish with a YesR since it starts with a NoR) because
171+ -- then it can't be used in a Respond, so responding can't loop.
172+ Loop : Inf (Process t iface hs hs' (resetReplied p) p') ->
173+ Process t iface hs hs' p p'
174+
175+ -- 'Running a iface' is the type of a process which is currently
176+ -- responding to requests (i.e. knows it has at least one client connected)
177+ -- and will not exit unless there are no clients connected
178+ Running : Type -> (iface : Type -> Type) -> Type
179+ Running a iface = {k : Nat} -> Process a iface [] (const []) (runningServer k) (const doneServer)
180+
181+ Response : Type -> (iface : Type -> Type) -> List (ReqHandle, Type) ->
182+ ProcState -> Type
183+ Response a iface hs p = Process a iface hs (const hs) p (const p)
184+
185+ -- 'Program a' is the type of a process which does not respond to any requests
186+ -- and begins and ends with no connections to any server open.
187+ Program : Type -> (iface : Type -> Type) -> Type
188+ Program a iface = Process a iface [] (const []) (init []) (const (init []))
189+
190+ -- 'Connected s a' is the type of a process which does not respond to any
191+ -- requests and begins and ends with connections to a given server list.
192+ Connected : List ServerID -> Type -> Type
193+ Connected s a = Process a (const Void) [] (const []) (init s) (const (init s))
194+
195+ -- 'Worker s a' is the type of a process which does not respond to any
196+ -- requests, and begins with a connection to a server it is to send a
197+ -- notification to.
198+ Worker : List ServerID -> Type -> Type
199+ Worker s a = Process a (const Void) [] (const []) (init s) (const (init []))
200+
201+implicit
202+Lift : IO a -> Process a iface hs (const hs) p (const p)
203+Lift = Lift'
204+
205+%no_implicit -- helps error messages, and speeds things up a bit
206+%inline -- so that the productivity checker treats 'bind' as a constructor!
207+(>>=) : Process a iface hs hs' p p' ->
208+ ((x : a) -> Process b iface (hs' x) hs'' (p' x) p'') ->
209+ Process b iface hs hs'' p p''
210+(>>=) = bind
211+
212+TrySend : (proc : ProcID iface) -> iface ty ->
213+ Process (Maybe ty) iface'
214+ hs (const hs)
215+ (MkProcState s c r) (const (MkProcState s c r))
216+TrySend pid req = do True <- Connect pid | False => Pure Nothing
217+ h <- Request pid req
218+ resp <- GetReply h
219+ Disconnect pid
220+ Pure (Just resp)
221+
222+Send : (proc : ProcID iface) -> iface ty ->
223+ {auto prf : Elem (MkServer proc) s} ->
224+ Process ty iface'
225+ hs (const hs)
226+ (MkProcState s c r) (const (MkProcState s c r))
227+Send pid req = do h <- Request pid req
228+ GetReply h
229+
230+{--- evaluator --}
231+
232+-- The evaluator keeps track of the number of client connections open,
233+-- and manages Connect/Disconnect requests by managing them whenever a
234+-- 'Respond' or 'TimeoutRespond' is encountered.
235+
236+data MessageQ : (Type -> Type) -> Type where
237+ ConnectMsg : MessageQ iface
238+ CloseMsg : MessageQ iface
239+ RequestMsg : Nat -> iface t -> MessageQ iface
240+
241+data MessageR : Type -> Type where
242+ ReplyMsg : Nat -> (ans : ty) -> MessageR ty
243+
244+data Message : (Type -> Type) -> List (ReqHandle, Type) -> Type where
245+ MsgQuery : MessageQ iface -> Message iface hs
246+ MsgReply : (reply : MessageR ty) -> Message iface hs
247+
248+readMsg : IO (Maybe (Ptr, Message iface hs))
249+readMsg {iface} {hs} =
250+ do if !checkMsgs
251+ then do msg <- getMsgWithSender {a = Message iface hs}
252+ pure (Just msg)
253+ else pure Nothing
254+
255+readMsgTimeout : Int -> IO (Maybe (Ptr, Message iface hs))
256+readMsgTimeout {iface} {hs} i =
257+ do if !(checkMsgsTimeout i)
258+ then do msg <- getMsgWithSender {a = Message iface hs}
259+ pure (Just msg)
260+ else pure Nothing
261+
262+data RespEnv : List (ReqHandle, Type) -> Type where
263+ Nil : RespEnv []
264+ (::) : Maybe ty -> RespEnv hs -> RespEnv ((h, ty) :: hs)
265+
266+record EvalState (iface : Type -> Type) (hs : List (ReqHandle, Type)) where
267+ constructor MkEvalState
268+ queue : List (Ptr, MessageQ iface)
269+ reply : RespEnv hs
270+ clients : Nat
271+ nexthandle : Nat
272+
273+lookup : Pending h hs ty -> RespEnv hs -> Maybe ty
274+lookup PendingHere (x :: xs) = x
275+lookup (PendingThere p) (x :: xs) = lookup p xs
276+
277+dropResp : (pending : Pending h hs ty) ->
278+ RespEnv hs -> RespEnv (dropPending hs pending)
279+dropResp PendingHere (x :: xs) = xs
280+dropResp (PendingThere p) (x :: xs) = x :: dropResp p xs
281+
282+updateReplies : Pending h hs ty -> ty -> RespEnv hs -> RespEnv hs
283+updateReplies PendingHere msg (x :: xs) = Just msg :: xs
284+updateReplies (PendingThere p) msg (x :: xs) = x :: updateReplies p msg xs
285+
286+total
287+findPending : (h : ReqHandle) -> RespEnv hs -> Maybe (ty ** Pending h hs ty)
288+findPending {hs = []} k [] = Nothing
289+findPending {hs = ((MkReqH h, t) :: hs)} (MkReqH k) (x :: xs) with (decEq h k)
290+ findPending {hs = ((MkReqH h, t) :: hs)} (MkReqH h) (x :: xs) | (Yes Refl)
291+ = Just (t ** PendingHere)
292+ findPending {hs = ((MkReqH h, t) :: hs)} (MkReqH k) (x :: xs) | (No contra)
293+ = do (ty' ** p') <- findPending (MkReqH k) xs
294+ Just (ty' ** PendingThere p')
295+
296+covering
297+updateQueue : EvalState iface hs -> IO (EvalState iface hs)
298+updateQueue {iface} {hs} st
299+ = case !(readMsg {iface} {hs}) of
300+ Nothing => pure st
301+ Just (pid, MsgQuery msg) =>
302+ updateQueue (record { queue = queue st ++ [(pid, msg)] } st)
303+ Just (pid, MsgReply (ReplyMsg rq ans)) =>
304+ case findPending (MkReqH rq) (reply st) of
305+ Nothing => updateQueue {iface} {hs} st -- can't happen!
306+ Just (ty ** pend) =>
307+ updateQueue (record { reply = updateReplies pend (believe_me ans) (reply st) } st)
308+
309+-- Keep updating the queue with incoming messages until either we get a
310+-- RequestMsg, or we reach a timeout.
311+covering
312+updateQueueTimeout : Int -> EvalState iface hs -> IO (EvalState iface hs)
313+updateQueueTimeout {iface} {hs} i st
314+ = case !(readMsgTimeout {iface} {hs} i) of
315+ Nothing => pure st
316+ Just (pid, MsgQuery (RequestMsg rq msg)) =>
317+ pure (record { queue = queue st ++ [(pid, RequestMsg rq msg)] } st)
318+ Just (pid, MsgQuery msg) => do
319+ -- TODO: Why not update client count here too?
320+ updateQueueTimeout i (record { queue = queue st ++ [(pid, msg)] } st)
321+ Just (pid, MsgReply (ReplyMsg rq ans)) =>
322+ case findPending (MkReqH rq) (reply st) of
323+ Nothing => updateQueueTimeout i {iface} {hs} st -- can't happen!
324+ Just (ty ** pend) =>
325+ updateQueueTimeout i (record { reply = updateReplies pend (believe_me ans) (reply st) } st)
326+
327+total
328+removeReq : List (Ptr, MessageQ iface) -> List (Ptr, MessageQ iface) ->
329+ Maybe (Ptr, (ty ** (Nat, iface ty)), List (Ptr, MessageQ iface))
330+removeReq acc [] = Nothing
331+removeReq acc ((pid, RequestMsg rq m) :: xs) = Just (pid, (_ ** (rq, m)), reverse acc ++ xs)
332+removeReq acc (x :: xs) = removeReq (x :: acc) xs
333+
334+total
335+removeConn : Nat ->
336+ List (Ptr, MessageQ iface) -> List (Ptr, MessageQ iface) ->
337+ (Nat, List (Ptr, MessageQ iface))
338+removeConn cl acc [] = (cl, reverse acc)
339+removeConn cl acc ((pid, ConnectMsg) :: xs)
340+ = removeConn (cl + 1) acc xs
341+removeConn cl acc ((pid, CloseMsg) :: xs)
342+ = removeConn (cl - 1) acc xs
343+removeConn cl acc (x :: xs) = removeConn cl (x :: acc) xs
344+
345+-- Remove the first thing in the event list which is a request, if it
346+-- exists.
347+getRequest : EvalState iface hs -> Maybe (Ptr, (ty ** (Nat, iface ty)), EvalState iface hs)
348+getRequest (MkEvalState queue reply clients nh)
349+ = do (pid, req, queue') <- removeReq [] queue
350+ return (pid, req, MkEvalState queue' reply clients nh)
351+
352+countClients : EvalState iface hs -> (Nat, EvalState iface hs)
353+countClients (MkEvalState queue reply clients nh)
354+ = let (clients', queue') = removeConn clients [] queue in
355+ (clients', MkEvalState queue' reply clients' nh)
356+
357+-- sendResponse : Ptr ->
358+-- (ty -> EvalState iface hs -> IO a) ->
359+-- (x, ty) -> EvalState iface hs ->
360+-- IO a
361+-- sendResponse {iface} {hs} pid k (resp, val) st = do
362+-- sendToThread pid (MsgReply {hs} {iface} ?prf (ReplyMsg resp))
363+-- k val st
364+
365+covering
366+eval : EvalState iface hs ->
367+ {p : ProcState} -> {p' : ty -> ProcState} ->
368+ Process ty iface hs hs' p p' ->
369+ ((res : ty) -> EvalState iface (hs' res) -> IO a) -> IO a
370+eval st (Lift' x) k = do x' <- x
371+ k x' st
372+eval st (Pure x) k = k x st
373+eval st (Quit x) k = k x st
374+eval st (bind x f) k = eval st x (\x', st' => eval st' (f x') k)
375+
376+eval st (Fork proc) k
377+ = do ptr <- fork (eval (MkEvalState [] [] 1 0) proc (\_, _ => pure ()))
378+ k (MkPID ptr) st
379+
380+eval st (Work proc cont) k
381+ = do ptr <- fork (eval (MkEvalState [] [] 0 0) (proc (MkPID prim__vm))
382+ (\_, _ => pure ()))
383+ eval (record { clients = clients st + 1 } st) cont k
384+
385+eval {hs} (MkEvalState q reqs c nh) (Request (MkPID pid) x) k
386+ = do sendToThread pid (MsgQuery {hs} (RequestMsg nh x))
387+ k (MkReqH nh) (MkEvalState q (Nothing :: reqs) c (S nh))
388+
389+eval {p} st (GetReply {pending} h) k
390+ = do -- Need to keep receiving messages until there's a response
391+ -- available
392+ MkEvalState q reqs c nh <- updateQueue st
393+ case lookup pending reqs of
394+ Nothing => do checkMsgsTimeout 1
395+ eval {p} (MkEvalState q reqs c nh) (GetReply h) k
396+ Just reply =>
397+ k reply (MkEvalState q (dropResp pending reqs) c nh)
398+
399+eval {iface} {hs} st (TimeoutRespond timeout def f) k
400+ = do st' <- updateQueueTimeout timeout st
401+ case getRequest st' of
402+ Nothing => k def st'
403+ Just (pid, (_ ** (rq, req)), st'') => do
404+ eval st'' (f req) -- (sendResponse pid k) -- ?foo
405+ -- this weirdness works around an erasure bug
406+ -- which causes a seg fault...
407+ (\ r, st''' =>
408+ case r of
409+ (resp, val) => do
410+ sendToThread pid (MsgReply {iface} {hs} (ReplyMsg rq resp))
411+ k val st''')
412+
413+eval {iface} {hs} st (Respond f) k
414+ = do st' <- updateQueue st
415+ case getRequest st' of
416+ Nothing => eval {iface} st' (Respond f) k
417+ Just (pid, (_ ** (rq, req)), st'') => do
418+ eval st'' (f req) (\ (resp, val), st''' => do
419+ sendToThread pid (MsgReply {iface} {hs} (ReplyMsg rq resp))
420+ k val st''')
421+
422+eval {hs} st (Connect {serveri} (MkPID pid)) k
423+ = if pid == prim__vm then k False st else do
424+ v <- sendToThread pid (MsgQuery {iface=serveri} {hs}
425+ ConnectMsg)
426+ -- TODO: Wait for ACK
427+ k (v == 1) st
428+
429+eval {hs} st (Disconnect {serveri} (MkPID pid)) k
430+ = do v <- sendToThread pid (MsgQuery {iface=serveri} {hs}
431+ CloseMsg)
432+ k () st
433+
434+eval st CountClients k
435+ = do st' <- updateQueue st
436+ let (cl, st'') = countClients st'
437+ k cl st''
438+
439+eval st (Loop x) k = eval st x k
440+
441+run : Program a iface -> IO a
442+run p = eval (MkEvalState [] [] 0 0) p (\res, t => pure res)
443+