git.iko.soy/Idris-dev
on master help sign in
Add System.Concurrency.Channels
2c83d47Edwin Brady
libs/base/System/Concurrency/Channels.idrnew file+63
compare with master
unifiedsplit
@@ -0,0 +1,63 @@
1+module System.Concurrency.Channels
2+
3+import System.Concurrency.Raw
4+
5+||| A Channel is a connection between two processes. Channels can be created
6+||| either using 'listen' to wait for an incoming connection, or 'connect'
7+||| which initiates a connection to another process.
8+||| Channels cannot be passed between processes.
9+export
10+data Channel : Type where
11+ MkConc : (pid : Ptr) -> (ch_id : Int) -> Channel
12+
13+||| A PID is a process identifier, as returned by `spawn`
14+export
15+data PID : Type where
16+ MkPID : (pid : Ptr) -> PID
17+
18+||| Spawn a process in a new thread, returning the process ID
19+export
20+spawn : IO () -> IO PID
21+spawn proc = do pid <- fork proc
22+ pure (MkPID pid)
23+
24+||| Create a channel which connects this process to another process
25+export
26+connect : (proc : PID) -> IO (Maybe Channel)
27+connect (MkPID pid) = do ch_id <- sendToThread pid 0 ()
28+ if (ch_id /= 0)
29+ then pure (Just (MkConc pid ch_id))
30+ else pure Nothing
31+
32+||| Listen for incoming connections. If another process has initiated a
33+||| communication with this process, returns a channel
34+export
35+listen : (timeout : Int) -> IO (Maybe Channel)
36+listen timeout = do checkMsgsTimeout timeout
37+ Just (client, ch_id) <- listenMsgs
38+ | Nothing => pure Nothing
39+ getMsgFrom {a = ()} client ch_id -- remove init message
40+ pure (Just (MkConc client ch_id))
41+
42+||| Send a message on a channel. Returns whether the message was successfully
43+||| sent to the process at the other end of the channel. This will fail if
44+||| the process on the channel is no longer running.
45+||| This is unsafe because there is no type checking, so there must be
46+||| a protocol (externally checked) which ensures that the message send
47+||| is of type expected by the receiver.
48+export
49+unsafeSend : Channel -> (val : a) -> IO Bool
50+unsafeSend (MkConc pid ch_id) val
51+ = do ok <- sendToThread pid ch_id val
52+ pure (ok /= 0)
53+
54+||| Receive a message on a channel, with an explicit type.
55+||| Blocks if there is nothing to receive. Returns `Nothing` if the
56+||| process on the channel is no longer running.
57+||| This is unsafe because there is no type checking, so there must be
58+||| a protocol (externally checked) which ensures that the message received
59+||| is of the type given by the sender.
60+export
61+unsafeRecv : (a : Type) -> Channel -> IO (Maybe a)
62+unsafeRecv a (MkConc pid ch_id) = getMsgFrom {a} pid ch_id
63+
libs/base/System/Concurrency/Raw.idr+50−28
@@ -9,11 +9,14 @@ import System
9%access export
9%access export
10
10
11||| Send a message of any type to the thread with the given thread id
11||| Send a message of any type to the thread with the given thread id
12−||| Returns 1 if the message was sent successfully, 0 otherwise
12+||| Returns channel ID if the message was sent successfully, 0 otherwise
13+|||
14+||| @channel an ID of a specific channel to send the message on. If 0,
15+||| the receiver will create a new channel ID
13−sendToThread : (thread_id : Ptr) -> a -> IO Int
16+sendToThread : (thread_id : Ptr) -> (channel : Int) -> a -> IO Int
14−sendToThread {a} dest val
17+sendToThread {a} dest channel val
15− = foreign FFI_C "idris_sendMessage" (Ptr -> Ptr -> Raw a -> IO Int)
18+ = foreign FFI_C "idris_sendMessage" (Ptr -> Int -> Ptr -> Raw a -> IO Int)
16− prim__vm dest (MkRaw val)
19+ prim__vm channel dest (MkRaw val)
17
20
18||| Check for messages in the process inbox
21||| Check for messages in the process inbox
19checkMsgs : IO Bool
22checkMsgs : IO Bool
@@ -31,21 +34,32 @@ checkMsgsTimeout timeout
31 null <- nullPtr msgs
34 null <- nullPtr msgs
32 return (not null)
35 return (not null)
33
36
37+private
38+sender : Ptr -> IO Ptr
39+sender msg = foreign FFI_C "idris_getSender" (Ptr -> IO Ptr) msg
40+
41+private
42+channel_id : Ptr -> IO Int
43+channel_id msg = foreign FFI_C "idris_getChannel" (Ptr -> IO Int) msg
44+
34−||| Check for messages in the process inbox.
45+||| Check for messages initiating a conversation in the process inbox.
35−||| Returns either 'Nothing', if none, or 'Just pid' as pid of sender.
46+||| Returns either 'Nothing', if none, or 'Just (pid, channel)' as pid
47+||| of sender and new channel id.
36−listenMsgs : IO (Maybe Ptr)
48+listenMsgs : IO (Maybe (Ptr, Int))
37−listenMsgs = do sender <- foreign FFI_C "idris_checkMessages" (Ptr -> IO Ptr)
49+listenMsgs = do msg <- foreign FFI_C "idris_checkInitMessages" (Ptr -> IO Ptr)
38 prim__vm
50 prim__vm
39− null <- nullPtr sender
51+ null <- nullPtr msg
40− return (if null
41− then Nothing
52+ if null then pure Nothing
42− else Just sender)
53+ else do s_id <- sender msg
54+ c_id <- channel_id msg
55+ pure (Just (s_id, c_id))
43
56
44−||| Check for messages from a specific sender in the process inbox
57+||| Check for messages from a specific sender/channel in the process inbox
58+||| If channel is '0', accept on any channel.
45−checkMsgsFrom : Ptr -> IO Bool
59+checkMsgsFrom : Ptr -> (channel : Int) -> IO Bool
46−checkMsgsFrom sender
60+checkMsgsFrom sender channel
47− = do msgs <- foreign FFI_C "idris_checkMessagesFrom" (Ptr -> Ptr -> IO Ptr)
61+ = do msgs <- foreign FFI_C "idris_checkMessagesFrom" (Ptr -> Int -> Ptr -> IO Ptr)
48− prim__vm sender
62+ prim__vm channel sender
49 null <- nullPtr msgs
63 null <- nullPtr msgs
50 return (not null)
64 return (not null)
51
65
@@ -60,23 +74,31 @@ getMsg {a} = do m <- foreign FFI_C "idris_recvMessage"
60 return x
74 return x
61
75
62||| Check inbox for messages. If there are none, blocks until a message
76||| Check inbox for messages. If there are none, blocks until a message
63−||| arrives. Return pair of sender's ID and the message.
77+||| arrives. Return triple of sender's ID, channel ID, and the message.
64||| Note that this is not at all type safe! It is intended to be used in
78||| Note that this is not at all type safe! It is intended to be used in
65||| a type safe wrapper.
79||| a type safe wrapper.
66−getMsgWithSender : IO (Ptr, a)
80+getMsgWithSender : IO (Ptr, Int, a)
67getMsgWithSender {a}
81getMsgWithSender {a}
68 = do m <- foreign FFI_C "idris_recvMessage"
82 = do m <- foreign FFI_C "idris_recvMessage"
69 (Ptr -> IO Ptr) prim__vm
83 (Ptr -> IO Ptr) prim__vm
70 MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m
84 MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m
71− vm <- foreign FFI_C "idris_getSender" (Ptr -> IO Ptr) m
85+ vm <- sender m
86+ chan <- channel_id m
72 foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m
87 foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m
73− return (vm, x)
88+ return (vm, chan, x)
74
89
90+||| Check inbox for messages on a particular channel. If there are none,
91+||| blocks until a message arrives. Returns `Nothing` if the sender isn't
92+||| alive
75−getMsgFrom : Ptr -> IO a
93+getMsgFrom : Ptr -> (channel : Int) -> IO (Maybe a)
76−getMsgFrom {a} sender
94+getMsgFrom {a} sender channel
77 = do m <- foreign FFI_C "idris_recvMessageFrom"
95 = do m <- foreign FFI_C "idris_recvMessageFrom"
78− (Ptr -> Ptr -> IO Ptr) prim__vm sender
96+ (Ptr -> Int -> Ptr -> IO Ptr) prim__vm channel sender
97+ null <- nullPtr m
98+ if null
99+ then pure Nothing
100+ else do
79− MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m
101+ MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m
80− foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m
102+ foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m
81− return x
103+ pure (Just x)
82
104
libs/base/base.ipkg+1−1
@@ -31,5 +31,5 @@ modules = System,
31 Control.Category, Control.Arrow,
31 Control.Category, Control.Arrow,
32 Control.Catchable, Control.IOExcept,
32 Control.Catchable, Control.IOExcept,
33
33
34− System.Concurrency.Raw
34+ System.Concurrency.Raw, System.Concurrency.Channels
35
35
libs/contrib/System/Concurrency/Process.idr+10−8
@@ -38,8 +38,8 @@ myID = Lift (return (MkPID prim__vm))
38||| Returns whether the send was unsuccessful.
38||| Returns whether the send was unsuccessful.
39export
39export
40send : ProcID msg -> msg -> Process msg Bool
40send : ProcID msg -> msg -> Process msg Bool
41−send (MkPID p) m = Lift (do x <- sendToThread p (prim__vm, m)
41+send (MkPID p) m = Lift (do x <- sendToThread p 0 (prim__vm, m)
42− return (x == 1))
42+ return (x /= 0))
43
43
44||| Return whether a message is waiting in the queue
44||| Return whether a message is waiting in the queue
45export
45export
@@ -49,7 +49,7 @@ msgWaiting = Lift checkMsgs
49||| Return whether a message is waiting in the queue from a specific sender
49||| Return whether a message is waiting in the queue from a specific sender
50export
50export
51msgWaitingFrom : ProcID msg -> Process msg Bool
51msgWaitingFrom : ProcID msg -> Process msg Bool
52−msgWaitingFrom (MkPID p) = Lift (checkMsgsFrom p)
52+msgWaitingFrom (MkPID p) = Lift (checkMsgsFrom p 0)
53
53
54||| Receive a message - blocks if there is no message waiting
54||| Receive a message - blocks if there is no message waiting
55export
55export
@@ -60,12 +60,14 @@ recv {msg} = do (senderid, m) <- Lift get
60 get = getMsg
60 get = getMsg
61
61
62||| Receive a message from specific sender - blocks if there is no message waiting
62||| Receive a message from specific sender - blocks if there is no message waiting
63+||| Fails if the sender is no longer running
63export
64export
64−recvFrom : ProcID msg -> Process msg msg
65+recvFrom : ProcID msg -> Process msg (Maybe msg)
65−recvFrom (MkPID p) {msg} = do (senderid, m) <- Lift get
66+recvFrom (MkPID p) {msg} = do Just (senderid, m) <- Lift get
66− return m
67+ | pure Nothing
68+ pure (Just m)
67− where get : IO (Ptr, msg)
69+ where get : IO (Maybe (Ptr, msg))
68− get = getMsgFrom p
70+ get = getMsgFrom p 0
69
71
70||| receive a message, and return with the sender's process ID.
72||| receive a message, and return with the sender's process ID.
71export
73export
rts/idris_rts.c+43−13
@@ -44,6 +44,7 @@ VM* init_vm(int stack_size, size_t heap_size,
44 memset(vm->inbox, 0, 1024*sizeof(VAL));
44 memset(vm->inbox, 0, 1024*sizeof(VAL));
45 vm->inbox_end = vm->inbox + 1024;
45 vm->inbox_end = vm->inbox + 1024;
46 vm->inbox_write = vm->inbox;
46 vm->inbox_write = vm->inbox;
47+ vm->inbox_nextid = 1;
47
48
48 // The allocation lock must be reentrant. The lock exists to ensure that
49 // The allocation lock must be reentrant. The lock exists to ensure that
49 // no memory is allocated during the message sending process, but we also
50 // no memory is allocated during the message sending process, but we also
@@ -825,7 +826,7 @@ VAL copyTo(VM* vm, VAL x) {
825}
826}
826
827
827// Add a message to another VM's message queue
828// Add a message to another VM's message queue
828−int idris_sendMessage(VM* sender, VM* dest, VAL msg) {
829+int idris_sendMessage(VM* sender, int channel_id, VM* dest, VAL msg) {
829 // FIXME: If GC kicks in in the middle of the copy, we're in trouble.
830 // FIXME: If GC kicks in in the middle of the copy, we're in trouble.
830 // Probably best check there is enough room in advance. (How?)
831 // Probably best check there is enough room in advance. (How?)
831
832
@@ -861,6 +862,14 @@ int idris_sendMessage(VM* sender, VM* dest, VAL msg) {
861 }
862 }
862
863
863 dest->inbox_write->msg = dmsg;
864 dest->inbox_write->msg = dmsg;
865+ if (channel_id == 0) {
866+ // Set lowest bit to indicate this message is initiating a channel
867+ channel_id = 1 + ((dest->inbox_nextid++) << 1);
868+ } else {
869+ channel_id = channel_id << 1;
870+ }
871+ dest->inbox_write->channel_id = channel_id;
872+
864 dest->inbox_write->sender = sender;
873 dest->inbox_write->sender = sender;
865 dest->inbox_write++;
874 dest->inbox_write++;
866
875
@@ -873,26 +882,39 @@ int idris_sendMessage(VM* sender, VM* dest, VAL msg) {
873
882
874 pthread_mutex_unlock(&(dest->inbox_lock));
883 pthread_mutex_unlock(&(dest->inbox_lock));
875// printf("Sending [unlock]...\n");
884// printf("Sending [unlock]...\n");
876− return 1;
885+ return channel_id >> 1;
877}
886}
878
887
879VM* idris_checkMessages(VM* vm) {
888VM* idris_checkMessages(VM* vm) {
880− return idris_checkMessagesFrom(vm, NULL);
889+ return idris_checkMessagesFrom(vm, 0, NULL);
881}
890}
882
891
892+Msg* idris_checkInitMessages(VM* vm) {
893+ Msg* msg;
894+
895+ for (msg = vm->inbox; msg < vm->inbox_end && msg->msg != NULL; ++msg) {
896+ if (msg->channel_id && 1 == 1) { // init bit set
897+ return msg;
898+ }
899+ }
900+ return 0;
901+}
902+
883−VM* idris_checkMessagesFrom(VM* vm, VM* sender) {
903+VM* idris_checkMessagesFrom(VM* vm, int channel_id, VM* sender) {
884 Msg* msg;
904 Msg* msg;
885
905
886 for (msg = vm->inbox; msg < vm->inbox_end && msg->msg != NULL; ++msg) {
906 for (msg = vm->inbox; msg < vm->inbox_end && msg->msg != NULL; ++msg) {
887 if (sender == NULL || msg->sender == sender) {
907 if (sender == NULL || msg->sender == sender) {
908+ if (channel_id == 0 || channel_id == msg->channel_id >> 1) {
888− return msg->sender;
909+ return msg->sender;
910+ }
889 }
911 }
890 }
912 }
891 return 0;
913 return 0;
892}
914}
893
915
894VM* idris_checkMessagesTimeout(VM* vm, int delay) {
916VM* idris_checkMessagesTimeout(VM* vm, int delay) {
895− VM* sender = idris_checkMessagesFrom(vm, NULL);
917+ VM* sender = idris_checkMessagesFrom(vm, 0, NULL);
896 if (sender != NULL) {
918 if (sender != NULL) {
897 return sender;
919 return sender;
898 }
920 }
@@ -910,16 +932,18 @@ VM* idris_checkMessagesTimeout(VM* vm, int delay) {
910 (void)(status); //don't emit 'unused' warning
932 (void)(status); //don't emit 'unused' warning
911 pthread_mutex_unlock(&vm->inbox_block);
933 pthread_mutex_unlock(&vm->inbox_block);
912
934
913− return idris_checkMessagesFrom(vm, NULL);
935+ return idris_checkMessagesFrom(vm, 0, NULL);
914}
936}
915
937
916
938
917−Msg* idris_getMessageFrom(VM* vm, VM* sender) {
939+Msg* idris_getMessageFrom(VM* vm, int channel_id, VM* sender) {
918 Msg* msg;
940 Msg* msg;
919
941
920 for (msg = vm->inbox; msg < vm->inbox_write; ++msg) {
942 for (msg = vm->inbox; msg < vm->inbox_write; ++msg) {
921 if (sender == NULL || msg->sender == sender) {
943 if (sender == NULL || msg->sender == sender) {
944+ if (channel_id == 0 || channel_id == msg->channel_id >> 1) {
922− return msg;
945+ return msg;
946+ }
923 }
947 }
924 }
948 }
925 return NULL;
949 return NULL;
@@ -927,18 +951,20 @@ Msg* idris_getMessageFrom(VM* vm, VM* sender) {
927
951
928// block until there is a message in the queue
952// block until there is a message in the queue
929Msg* idris_recvMessage(VM* vm) {
953Msg* idris_recvMessage(VM* vm) {
930− return idris_recvMessageFrom(vm, NULL);
954+ return idris_recvMessageFrom(vm, 0, NULL);
931}
955}
932
956
933−Msg* idris_recvMessageFrom(VM* vm, VM* sender) {
957+Msg* idris_recvMessageFrom(VM* vm, int channel_id, VM* sender) {
934 Msg* msg;
958 Msg* msg;
935 Msg* ret = malloc(sizeof(Msg));
959 Msg* ret = malloc(sizeof(Msg));
936
960
937 struct timespec timeout;
961 struct timespec timeout;
938 int status;
962 int status;
963+
964+ if (sender && sender->active == 0) { return NULL; } // No VM to receive from
939
965
940 pthread_mutex_lock(&vm->inbox_block);
966 pthread_mutex_lock(&vm->inbox_block);
941− msg = idris_getMessageFrom(vm, sender);
967+ msg = idris_getMessageFrom(vm, channel_id, sender);
942
968
943 while (msg == NULL) {
969 while (msg == NULL) {
944// printf("No message yet\n");
970// printf("No message yet\n");
@@ -949,7 +975,7 @@ Msg* idris_recvMessageFrom(VM* vm, VM* sender) {
949 &timeout);
975 &timeout);
950 (void)(status); //don't emit 'unused' warning
976 (void)(status); //don't emit 'unused' warning
951// printf("Waiting [unlock]... %d\n", status);
977// printf("Waiting [unlock]... %d\n", status);
952− msg = idris_getMessageFrom(vm, sender);
978+ msg = idris_getMessageFrom(vm, channel_id, sender);
953 }
979 }
954 pthread_mutex_unlock(&vm->inbox_block);
980 pthread_mutex_unlock(&vm->inbox_block);
955
981
@@ -993,6 +1019,10 @@ VM* idris_getSender(Msg* msg) {
993 return msg->sender;
1019 return msg->sender;
994}
1020}
995
1021
1022+int idris_getChannel(Msg* msg) {
1023+ return msg->channel_id >> 1;
1024+}
1025+
996void idris_freeMsg(Msg* msg) {
1026void idris_freeMsg(Msg* msg) {
997 free(msg);
1027 free(msg);
998}
1028}
rts/idris_rts.h+11−3
@@ -76,6 +76,9 @@ struct VM;
76
76
77struct Msg_t {
77struct Msg_t {
78 struct VM* sender;
78 struct VM* sender;
79+ // An identifier to say which conversation this message is part of.
80+ // Lowest bit is set if the id is the first message in a conversation.
81+ int channel_id;
79 VAL msg;
82 VAL msg;
80};
83};
81
84
@@ -101,6 +104,7 @@ struct VM {
101
104
102 Msg* inbox; // Block of memory for storing messages
105 Msg* inbox; // Block of memory for storing messages
103 Msg* inbox_end; // End of block of memory
106 Msg* inbox_end; // End of block of memory
107+ int inbox_nextid; // Next channel id
104 Msg* inbox_write; // Location of next message to write
108 Msg* inbox_write; // Location of next message to write
105
109
106 int processes; // Number of child processes
110 int processes; // Number of child processes
@@ -310,23 +314,27 @@ void* vmThread(VM* callvm, func f, VAL arg);
310VAL copyTo(VM* newVM, VAL x);
314VAL copyTo(VM* newVM, VAL x);
311
315
312// Add a message to another VM's message queue
316// Add a message to another VM's message queue
313−int idris_sendMessage(VM* sender, VM* dest, VAL msg);
317+int idris_sendMessage(VM* sender, int channel_id, VM* dest, VAL msg);
314// Check whether there are any messages in the queue and return PID of
318// Check whether there are any messages in the queue and return PID of
315// sender if so (null if not)
319// sender if so (null if not)
316VM* idris_checkMessages(VM* vm);
320VM* idris_checkMessages(VM* vm);
321+// Check whether there are any messages which are initiating a conversation
322+// in the queue and return the message if so (without removing it)
323+Msg* idris_checkInitMessages(VM* vm);
317// Check whether there are any messages in the queue
324// Check whether there are any messages in the queue
318−VM* idris_checkMessagesFrom(VM* vm, VM* sender);
325+VM* idris_checkMessagesFrom(VM* vm, int channel_id, VM* sender);
319// Check whether there are any messages in the queue, and wait if not
326// Check whether there are any messages in the queue, and wait if not
320VM* idris_checkMessagesTimeout(VM* vm, int timeout);
327VM* idris_checkMessagesTimeout(VM* vm, int timeout);
321// block until there is a message in the queue
328// block until there is a message in the queue
322Msg* idris_recvMessage(VM* vm);
329Msg* idris_recvMessage(VM* vm);
323// block until there is a message in the queue
330// block until there is a message in the queue
324−Msg* idris_recvMessageFrom(VM* vm, VM* sender);
331+Msg* idris_recvMessageFrom(VM* vm, int channel_id, VM* sender);
325
332
326// Query/free structure used to return message data (recvMessage will malloc,
333// Query/free structure used to return message data (recvMessage will malloc,
327// so needs an explicit free)
334// so needs an explicit free)
328VAL idris_getMsg(Msg* msg);
335VAL idris_getMsg(Msg* msg);
329VM* idris_getSender(Msg* msg);
336VM* idris_getSender(Msg* msg);
337+int idris_getChannel(Msg* msg);
330void idris_freeMsg(Msg* msg);
338void idris_freeMsg(Msg* msg);
331
339
332void dumpVal(VAL r);
340void dumpVal(VAL r);
test/io003/test018.idr+2−2
@@ -10,11 +10,11 @@ pong : IO ()
10pong = do (sender, x) <- recvMsg
10pong = do (sender, x) <- recvMsg
11 putStrLn x
11 putStrLn x
12 putStrLn "Received"
12 putStrLn "Received"
13− sendToThread sender "Hello to you too!"
13+ sendToThread sender 0 "Hello to you too!"
14 return ()
14 return ()
15
15
16ping : Ptr -> IO ()
16ping : Ptr -> IO ()
17−ping thread = do sendToThread thread (prim__vm, "Hello!")
17+ping thread = do sendToThread thread 0 (prim__vm, "Hello!")
18 return ()
18 return ()
19
19
20pingpong : IO ()
20pingpong : IO ()
test/proofsearch002/Process.idr+11−11
@@ -248,15 +248,15 @@ data Message : (Type -> Type) -> List (ReqHandle, Type) -> Type where
248readMsg : IO (Maybe (Ptr, Message iface hs))
248readMsg : IO (Maybe (Ptr, Message iface hs))
249readMsg {iface} {hs} =
249readMsg {iface} {hs} =
250 do if !checkMsgs
250 do if !checkMsgs
251− then do msg <- getMsgWithSender {a = Message iface hs}
251+ then do (p, _, msg) <- getMsgWithSender {a = Message iface hs}
252− pure (Just msg)
252+ pure (Just (p, msg))
253 else pure Nothing
253 else pure Nothing
254
254
255readMsgTimeout : Int -> IO (Maybe (Ptr, Message iface hs))
255readMsgTimeout : Int -> IO (Maybe (Ptr, Message iface hs))
256readMsgTimeout {iface} {hs} i =
256readMsgTimeout {iface} {hs} i =
257 do if !(checkMsgsTimeout i)
257 do if !(checkMsgsTimeout i)
258− then do msg <- getMsgWithSender {a = Message iface hs}
258+ then do (pid, _, msg) <- getMsgWithSender {a = Message iface hs}
259− pure (Just msg)
259+ pure (Just (pid, msg))
260 else pure Nothing
260 else pure Nothing
261
261
262data RespEnv : List (ReqHandle, Type) -> Type where
262data RespEnv : List (ReqHandle, Type) -> Type where
@@ -383,7 +383,7 @@ eval st (Work proc cont) k
383 eval (record { clients = clients st + 1 } st) cont k
383 eval (record { clients = clients st + 1 } st) cont k
384
384
385eval {hs} (MkEvalState q reqs c nh) (Request (MkPID pid) x) k
385eval {hs} (MkEvalState q reqs c nh) (Request (MkPID pid) x) k
386− = do sendToThread pid (MsgQuery {hs} (RequestMsg nh x))
386+ = do sendToThread pid 0 (MsgQuery {hs} (RequestMsg nh x))
387 k (MkReqH nh) (MkEvalState q (Nothing :: reqs) c (S nh))
387 k (MkReqH nh) (MkEvalState q (Nothing :: reqs) c (S nh))
388
388
389eval {p} st (GetReply {pending} h) k
389eval {p} st (GetReply {pending} h) k
@@ -407,7 +407,7 @@ eval {iface} {hs} st (TimeoutRespond timeout def f) k
407 (\ r, st''' =>
407 (\ r, st''' =>
408 case r of
408 case r of
409 (resp, val) => do
409 (resp, val) => do
410− sendToThread pid (MsgReply {iface} {hs} (ReplyMsg rq resp))
410+ sendToThread pid 0 (MsgReply {iface} {hs} (ReplyMsg rq resp))
411 k val st''')
411 k val st''')
412
412
413eval {iface} {hs} st (Respond f) k
413eval {iface} {hs} st (Respond f) k
@@ -416,19 +416,19 @@ eval {iface} {hs} st (Respond f) k
416 Nothing => eval {iface} st' (Respond f) k
416 Nothing => eval {iface} st' (Respond f) k
417 Just (pid, (_ ** (rq, req)), st'') => do
417 Just (pid, (_ ** (rq, req)), st'') => do
418 eval st'' (f req) (\ (resp, val), st''' => do
418 eval st'' (f req) (\ (resp, val), st''' => do
419− sendToThread pid (MsgReply {iface} {hs} (ReplyMsg rq resp))
419+ sendToThread pid 0 (MsgReply {iface} {hs} (ReplyMsg rq resp))
420 k val st''')
420 k val st''')
421
421
422eval {hs} st (Connect {serveri} (MkPID pid)) k
422eval {hs} st (Connect {serveri} (MkPID pid)) k
423 = if pid == prim__vm then k False st else do
423 = if pid == prim__vm then k False st else do
424− v <- sendToThread pid (MsgQuery {iface=serveri} {hs}
424+ v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}
425− ConnectMsg)
425+ ConnectMsg)
426 -- TODO: Wait for ACK
426 -- TODO: Wait for ACK
427 k (v == 1) st
427 k (v == 1) st
428
428
429eval {hs} st (Disconnect {serveri} (MkPID pid)) k
429eval {hs} st (Disconnect {serveri} (MkPID pid)) k
430− = do v <- sendToThread pid (MsgQuery {iface=serveri} {hs}
430+ = do v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}
431− CloseMsg)
431+ CloseMsg)
432 k () st
432 k () st
433
433
434eval st CountClients k
434eval st CountClients k