git.iko.soy/Idris-dev
on master help sign in
Idris-dev/test/proofsearch002/Process.idr
444 lines 18 kB history raw
1module Process
3import System.Concurrency.Raw
4import public Data.List -- public, to get proof search machinery
6%access public export
8-- Process IDs are parameterised by their interface. A request of type
9-- 'iface t' will get a response of type 't'
10data ProcID : (iface : Type -> Type) -> Type where
11 MkPID : Ptr -> ProcID iface
13data ServerID : Type where
14 MkServer : ProcID iface -> ServerID
16implicit MkServer' : ProcID iface -> ServerID
17MkServer' = MkServer
19data Replied = YesR | NoR
21data ReqHandle = MkReqH Nat
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
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.
31data ProcState : Type where
32 MkProcState : (servers : List ServerID) ->
33 (clients : Nat) ->
34 Replied ->
35 ProcState
37data 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
41dropPending : (hs : List (ReqHandle, Type)) -> Pending h hs ty ->
42 List (ReqHandle, Type)
43dropPending ((h, t) :: xs) PendingHere = xs
44dropPending ((h', t') :: xs) (PendingThere x)
45 = ((h', t') :: dropPending xs x)
47data ConnectedTo : ServerID -> ProcState -> Type where
48 IsConnectedTo : Elem p servers ->
49 ConnectedTo p (MkProcState servers c reply)
51data NoClient : ProcState -> Type where
52 IsNoClient : NoClient (MkProcState servers 0 reply)
54data OneClient : ProcState -> Type where
55 IsOneClient : OneClient (MkProcState servers (S k) reply)
57data Reply : ProcState -> Type where
58 IsReply : Reply (MkProcState s c YesR)
60data NoReply : ProcState -> Type where
61 IsNoReply : NoReply (MkProcState s c NoR)
63{-- Some useful operations on process state --}
64newClient : ProcState -> ProcState
65newClient (MkProcState servers clients r)
66 = MkProcState servers (S clients) r
68setClients : ProcState -> Nat -> ProcState
69setClients (MkProcState servers clients r) k
70 = MkProcState servers k r
72newServer : ProcID iface -> ProcState -> ProcState
73newServer p (MkProcState servers clients r)
74 = MkProcState (MkServer p :: servers) clients r
76dropServer : (pid : ProcID iface) -> (p : ProcState) ->
77 ConnectedTo (MkServer pid) p -> ProcState
78dropServer pid (MkProcState servers c r) (IsConnectedTo prf)
79 = MkProcState (dropElem servers prf) c r
81{-
82pendingReq : ProcID iface -> (h : ReqHandle) -> iface t ->
83 ProcState hs -> ProcState ((h, t) :: hs)
84pendingReq {t} p h x (MkProcState s c r) = MkProcState s c r
86doneReq : ProcState hs -> (p : Pending h hs) -> ProcState (dropPending hs p)
87doneReq (MkProcState s c r) p = MkProcState s c r
88-}
90replied : ProcState -> ProcState
91replied (MkProcState servers clients r)
92 = MkProcState servers clients YesR
94resetReplied : ProcState -> ProcState
95resetReplied (MkProcState servers clients r)
96 = MkProcState servers clients NoR
98runningServer : Nat -> ProcState
99runningServer c = MkProcState [] (S c) NoR
101doneServer : ProcState
102doneServer = MkProcState [] 0 YesR
104init : List ServerID -> ProcState
105init s = MkProcState s 0 NoR
107{-- Processes themselves.
109A process returns some type 'a', responds to requests on the interface
110'iface', and has an input and output state.
111--}
112mutual
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))
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''
126 Fork : Process () serveri [] (const []) (runningServer 1) (const Process.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 Process.doneServer)) ->
130 Process t iface hs (const hs) p (const p)
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)
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)
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))
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))
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)
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))
167 CountClients : Process Nat iface hs (const hs) p (\n => setClients p n)
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'
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)
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)
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 []))
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))
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 []))
201implicit
202Lift : IO a -> Process a iface hs (const hs) p (const p)
203Lift = Lift'
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
212TrySend : (proc : ProcID iface) -> iface ty ->
213 Process (Maybe ty) iface'
214 hs (const hs)
215 (MkProcState s c r) (const (MkProcState s c r))
216TrySend 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)
222Send : (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))
227Send pid req = do h <- Request pid req
228 GetReply h
230{--- evaluator --}
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.
236data MessageQ : (Type -> Type) -> Type where
237 ConnectMsg : MessageQ iface
238 CloseMsg : MessageQ iface
239 RequestMsg : Nat -> iface t -> MessageQ iface
241data MessageR : Type -> Type where
242 ReplyMsg : Nat -> (ans : ty) -> MessageR ty
244data Message : (Type -> Type) -> List (ReqHandle, Type) -> Type where
245 MsgQuery : MessageQ iface -> Message iface hs
246 MsgReply : (reply : MessageR ty) -> Message iface hs
248readMsg : IO (Maybe (Ptr, Message iface hs))
249readMsg {iface} {hs} =
250 do if !checkMsgs
251 then do (p, _, msg) <- getMsgWithSender {a = Message iface hs}
252 pure (Just (p, msg))
253 else pure Nothing
255readMsgTimeout : Int -> IO (Maybe (Ptr, Message iface hs))
256readMsgTimeout {iface} {hs} i =
257 do if !(checkMsgsTimeout i)
258 then do (pid, _, msg) <- getMsgWithSender {a = Message iface hs}
259 pure (Just (pid, msg))
260 else pure Nothing
262data RespEnv : List (ReqHandle, Type) -> Type where
263 Nil : RespEnv []
264 (::) : Maybe ty -> RespEnv hs -> RespEnv ((h, ty) :: hs)
266record 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
273lookup : Pending h hs ty -> RespEnv hs -> Maybe ty
274lookup PendingHere (x :: xs) = x
275lookup (PendingThere p) (x :: xs) = lookup p xs
277dropResp : (pending : Pending h hs ty) ->
278 RespEnv hs -> RespEnv (dropPending hs pending)
279dropResp PendingHere (x :: xs) = xs
280dropResp (PendingThere p) (x :: xs) = x :: dropResp p xs
282updateReplies : Pending h hs ty -> ty -> RespEnv hs -> RespEnv hs
283updateReplies PendingHere msg (x :: xs) = Just msg :: xs
284updateReplies (PendingThere p) msg (x :: xs) = x :: updateReplies p msg xs
286total
287findPending : (h : ReqHandle) -> RespEnv hs -> Maybe (ty ** Pending h hs ty)
288findPending {hs = []} k [] = Nothing
289findPending {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')
296covering
297updateQueue : EvalState iface hs -> IO (EvalState iface hs)
298updateQueue {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)
309-- Keep updating the queue with incoming messages until either we get a
310-- RequestMsg, or we reach a timeout.
311covering
312updateQueueTimeout : Int -> EvalState iface hs -> IO (EvalState iface hs)
313updateQueueTimeout {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)
327total
328removeReq : List (Ptr, MessageQ iface) -> List (Ptr, MessageQ iface) ->
329 Maybe (Ptr, (ty ** (Nat, iface ty)), List (Ptr, MessageQ iface))
330removeReq acc [] = Nothing
331removeReq acc ((pid, RequestMsg rq m) :: xs) = Just (pid, (_ ** (rq, m)), reverse acc ++ xs)
332removeReq acc (x :: xs) = removeReq (x :: acc) xs
334total
335removeConn : Nat ->
336 List (Ptr, MessageQ iface) -> List (Ptr, MessageQ iface) ->
337 (Nat, List (Ptr, MessageQ iface))
338removeConn cl acc [] = (cl, reverse acc)
339removeConn cl acc ((pid, ConnectMsg) :: xs)
340 = removeConn (cl + 1) acc xs
341removeConn cl acc ((pid, CloseMsg) :: xs)
342 = removeConn (minus cl 1) acc xs
343removeConn cl acc (x :: xs) = removeConn cl (x :: acc) xs
345-- Remove the first thing in the event list which is a request, if it
346-- exists.
347getRequest : EvalState iface hs -> Maybe (Ptr, (ty ** (Nat, iface ty)), EvalState iface hs)
348getRequest (MkEvalState queue reply clients nh)
349 = do (pid, req, queue') <- removeReq [] queue
350 pure (pid, req, MkEvalState queue' reply clients nh)
352countClients : EvalState iface hs -> (Nat, EvalState iface hs)
353countClients (MkEvalState queue reply clients nh)
354 = let (clients', queue') = removeConn clients [] queue in
355 (clients', MkEvalState queue' reply clients' nh)
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
365covering
366eval : 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
370eval st (Lift' x) k = do x' <- x
371 k x' st
372eval st (Pure x) k = k x st
373eval st (Quit x) k = k x st
374eval st (Bind x f) k = eval st x (\x', st' => eval st' (f x') k)
376eval st (Fork proc) k
377 = do ptr <- fork (eval (MkEvalState [] [] 1 0) proc (\_, _ => pure ()))
378 k (MkPID ptr) st
380eval st (Work proc cont) k
381 = do me <- getMyVM
382 ptr <- fork (eval (MkEvalState [] [] 0 0) (proc (MkPID me))
383 (\_, _ => pure ()))
384 eval (record { clients = clients st + 1 } st) cont k
386eval {hs} (MkEvalState q reqs c nh) (Request (MkPID pid) x) k
387 = do sendToThread pid 0 (MsgQuery {hs} (RequestMsg nh x))
388 k (MkReqH nh) (MkEvalState q (Nothing :: reqs) c (S nh))
390eval {p} st (GetReply {pending} h) k
391 = do -- Need to keep receiving messages until there's a response
392 -- available
393 MkEvalState q reqs c nh <- updateQueue st
394 case lookup pending reqs of
395 Nothing => do checkMsgsTimeout 1
396 eval {p} (MkEvalState q reqs c nh) (GetReply h) k
397 Just reply =>
398 k reply (MkEvalState q (dropResp pending reqs) c nh)
400eval {iface} {hs} st (TimeoutRespond timeout def f) k
401 = do st' <- updateQueueTimeout timeout st
402 case getRequest st' of
403 Nothing => k def st'
404 Just (pid, (_ ** (rq, req)), st'') => do
405 eval st'' (f req) -- (sendResponse pid k) -- ?foo
406 -- this weirdness works around an erasure bug
407 -- which causes a seg fault...
408 (\ r, st''' =>
409 case r of
410 (resp, val) => do
411 sendToThread pid 0 (MsgReply {iface} {hs} (ReplyMsg rq resp))
412 k val st''')
414eval {iface} {hs} st (Respond f) k
415 = do st' <- updateQueue st
416 case getRequest st' of
417 Nothing => eval {iface} st' (Respond f) k
418 Just (pid, (_ ** (rq, req)), st'') => do
419 eval st'' (f req) (\ (resp, val), st''' => do
420 sendToThread pid 0 (MsgReply {iface} {hs} (ReplyMsg rq resp))
421 k val st''')
423eval {hs} st (Connect {serveri} (MkPID pid)) k
424 = do me <- getMyVM
425 if pid == me then k False st else do
426 v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}
427 ConnectMsg)
428 -- TODO: Wait for ACK
429 k (v == 1) st
431eval {hs} st (Disconnect {serveri} (MkPID pid)) k
432 = do v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}
433 CloseMsg)
434 k () st
436eval st CountClients k
437 = do st' <- updateQueue st
438 let (cl, st'') = countClients st'
439 k cl st''
441eval st (Loop x) k = eval st x k
443run : Program a iface -> IO a
444run p = eval (MkEvalState [] [] 0 0) p (\res, t => pure res)