2c83d47 · Idris-dev · git.iko.soy

libs/base/System/Concurrency/Channels.idrnew file+63

@@ -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

99%access export

1010

1111||| 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

13−sendToThread : (thread_id : Ptr) -> a -> IO Int

14−sendToThread {a} dest val

15− = foreign FFI_C "idris_sendMessage" (Ptr -> Ptr -> Raw a -> IO Int)

16− prim__vm dest (MkRaw val)

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

16+sendToThread : (thread_id : Ptr) -> (channel : Int) -> a -> IO Int

17+sendToThread {a} dest channel val

18+ = foreign FFI_C "idris_sendMessage" (Ptr -> Int -> Ptr -> Raw a -> IO Int)

19+ prim__vm channel dest (MkRaw val)

1720

1821||| Check for messages in the process inbox

1922checkMsgs : IO Bool

@@ -31,21 +34,32 @@ checkMsgsTimeout timeout

3134 null <- nullPtr msgs

3235 return (not null)

3336

34−||| Check for messages in the process inbox.

35−||| Returns either 'Nothing', if none, or 'Just pid' as pid of sender.

36−listenMsgs : IO (Maybe Ptr)

37−listenMsgs = do sender <- foreign FFI_C "idris_checkMessages" (Ptr -> IO Ptr)

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+

45+||| Check for messages initiating a conversation in the process inbox.

46+||| Returns either 'Nothing', if none, or 'Just (pid, channel)' as pid

47+||| of sender and new channel id.

48+listenMsgs : IO (Maybe (Ptr, Int))

49+listenMsgs = do msg <- foreign FFI_C "idris_checkInitMessages" (Ptr -> IO Ptr)

3850 prim__vm

39− null <- nullPtr sender

40− return (if null

41− then Nothing

42− else Just sender)

51+ null <- nullPtr msg

52+ if null then pure Nothing

53+ else do s_id <- sender msg

54+ c_id <- channel_id msg

55+ pure (Just (s_id, c_id))

4356

44−||| Check for messages from a specific sender in the process inbox

45−checkMsgsFrom : Ptr -> IO Bool

46−checkMsgsFrom sender

47− = do msgs <- foreign FFI_C "idris_checkMessagesFrom" (Ptr -> Ptr -> IO Ptr)

48− prim__vm sender

57+||| Check for messages from a specific sender/channel in the process inbox

58+||| If channel is '0', accept on any channel.

59+checkMsgsFrom : Ptr -> (channel : Int) -> IO Bool

60+checkMsgsFrom sender channel

61+ = do msgs <- foreign FFI_C "idris_checkMessagesFrom" (Ptr -> Int -> Ptr -> IO Ptr)

62+ prim__vm channel sender

4963 null <- nullPtr msgs

5064 return (not null)

5165

@@ -60,23 +74,31 @@ getMsg {a} = do m <- foreign FFI_C "idris_recvMessage"

6074 return x

6175

6276||| 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.

6478||| Note that this is not at all type safe! It is intended to be used in

6579||| a type safe wrapper.

66−getMsgWithSender : IO (Ptr, a)

80+getMsgWithSender : IO (Ptr, Int, a)

6781getMsgWithSender {a}

6882 = do m <- foreign FFI_C "idris_recvMessage"

6983 (Ptr -> IO Ptr) prim__vm

7084 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

7287 foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m

73− return (vm, x)

88+ return (vm, chan, x)

7489

75−getMsgFrom : Ptr -> IO a

76−getMsgFrom {a} sender

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

93+getMsgFrom : Ptr -> (channel : Int) -> IO (Maybe a)

94+getMsgFrom {a} sender channel

7795 = do m <- foreign FFI_C "idris_recvMessageFrom"

78− (Ptr -> Ptr -> IO Ptr) prim__vm sender

79− MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m

80− foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m

81− return x

96+ (Ptr -> Int -> Ptr -> IO Ptr) prim__vm channel sender

97+ null <- nullPtr m

98+ if null

99+ then pure Nothing

100+ else do

101+ MkRaw x <- foreign FFI_C "idris_getMsg" (Ptr -> IO (Raw a)) m

102+ foreign FFI_C "idris_freeMsg" (Ptr -> IO ()) m

103+ pure (Just x)

82104

libs/base/base.ipkg+1−1

@@ -31,5 +31,5 @@ modules = System,

3131 Control.Category, Control.Arrow,

3232 Control.Catchable, Control.IOExcept,

3333

34− System.Concurrency.Raw

34+ System.Concurrency.Raw, System.Concurrency.Channels

3535

libs/contrib/System/Concurrency/Process.idr+10−8

@@ -38,8 +38,8 @@ myID = Lift (return (MkPID prim__vm))

3838||| Returns whether the send was unsuccessful.

3939export

4040send : ProcID msg -> msg -> Process msg Bool

41−send (MkPID p) m = Lift (do x <- sendToThread p (prim__vm, m)

42− return (x == 1))

41+send (MkPID p) m = Lift (do x <- sendToThread p 0 (prim__vm, m)

42+ return (x /= 0))

4343

4444||| Return whether a message is waiting in the queue

4545export

@@ -49,7 +49,7 @@ msgWaiting = Lift checkMsgs

4949||| Return whether a message is waiting in the queue from a specific sender

5050export

5151msgWaitingFrom : ProcID msg -> Process msg Bool

52−msgWaitingFrom (MkPID p) = Lift (checkMsgsFrom p)

52+msgWaitingFrom (MkPID p) = Lift (checkMsgsFrom p 0)

5353

5454||| Receive a message - blocks if there is no message waiting

5555export

@@ -60,12 +60,14 @@ recv {msg} = do (senderid, m) <- Lift get

6060 get = getMsg

6161

6262||| Receive a message from specific sender - blocks if there is no message waiting

63+||| Fails if the sender is no longer running

6364export

64−recvFrom : ProcID msg -> Process msg msg

65−recvFrom (MkPID p) {msg} = do (senderid, m) <- Lift get

66− return m

67− where get : IO (Ptr, msg)

68− get = getMsgFrom p

65+recvFrom : ProcID msg -> Process msg (Maybe msg)

66+recvFrom (MkPID p) {msg} = do Just (senderid, m) <- Lift get

67+ | pure Nothing

68+ pure (Just m)

69+ where get : IO (Maybe (Ptr, msg))

70+ get = getMsgFrom p 0

6971

7072||| receive a message, and return with the sender's process ID.

7173export

rts/idris_rts.c+43−13

@@ -44,6 +44,7 @@ VM* init_vm(int stack_size, size_t heap_size,

4444 memset(vm->inbox, 0, 1024*sizeof(VAL));

4545 vm->inbox_end = vm->inbox + 1024;

4646 vm->inbox_write = vm->inbox;

47+ vm->inbox_nextid = 1;

4748

4849 // The allocation lock must be reentrant. The lock exists to ensure that

4950 // no memory is allocated during the message sending process, but we also

@@ -825,7 +826,7 @@ VAL copyTo(VM* vm, VAL x) {

825826}

826827

827828// 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) {

829830 // FIXME: If GC kicks in in the middle of the copy, we're in trouble.

830831 // Probably best check there is enough room in advance. (How?)

831832

@@ -861,6 +862,14 @@ int idris_sendMessage(VM* sender, VM* dest, VAL msg) {

861862 }

862863

863864 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+

864873 dest->inbox_write->sender = sender;

865874 dest->inbox_write++;

866875

@@ -873,26 +882,39 @@ int idris_sendMessage(VM* sender, VM* dest, VAL msg) {

873882

874883 pthread_mutex_unlock(&(dest->inbox_lock));

875884// printf("Sending [unlock]...\n");

876− return 1;

885+ return channel_id >> 1;

877886}

878887

879888VM* idris_checkMessages(VM* vm) {

880− return idris_checkMessagesFrom(vm, NULL);

889+ return idris_checkMessagesFrom(vm, 0, NULL);

881890}

882891

883−VM* idris_checkMessagesFrom(VM* vm, VM* sender) {

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+

903+VM* idris_checkMessagesFrom(VM* vm, int channel_id, VM* sender) {

884904 Msg* msg;

885905

886906 for (msg = vm->inbox; msg < vm->inbox_end && msg->msg != NULL; ++msg) {

887907 if (sender == NULL || msg->sender == sender) {

888− return msg->sender;

908+ if (channel_id == 0 || channel_id == msg->channel_id >> 1) {

909+ return msg->sender;

910+ }

889911 }

890912 }

891913 return 0;

892914}

893915

894916VM* idris_checkMessagesTimeout(VM* vm, int delay) {

895− VM* sender = idris_checkMessagesFrom(vm, NULL);

917+ VM* sender = idris_checkMessagesFrom(vm, 0, NULL);

896918 if (sender != NULL) {

897919 return sender;

898920 }

@@ -910,16 +932,18 @@ VM* idris_checkMessagesTimeout(VM* vm, int delay) {

910932 (void)(status); //don't emit 'unused' warning

911933 pthread_mutex_unlock(&vm->inbox_block);

912934

913− return idris_checkMessagesFrom(vm, NULL);

935+ return idris_checkMessagesFrom(vm, 0, NULL);

914936}

915937

916938

917−Msg* idris_getMessageFrom(VM* vm, VM* sender) {

939+Msg* idris_getMessageFrom(VM* vm, int channel_id, VM* sender) {

918940 Msg* msg;

919941

920942 for (msg = vm->inbox; msg < vm->inbox_write; ++msg) {

921943 if (sender == NULL || msg->sender == sender) {

922− return msg;

944+ if (channel_id == 0 || channel_id == msg->channel_id >> 1) {

945+ return msg;

946+ }

923947 }

924948 }

925949 return NULL;

@@ -927,18 +951,20 @@ Msg* idris_getMessageFrom(VM* vm, VM* sender) {

927951

928952// block until there is a message in the queue

929953Msg* idris_recvMessage(VM* vm) {

930− return idris_recvMessageFrom(vm, NULL);

954+ return idris_recvMessageFrom(vm, 0, NULL);

931955}

932956

933−Msg* idris_recvMessageFrom(VM* vm, VM* sender) {

957+Msg* idris_recvMessageFrom(VM* vm, int channel_id, VM* sender) {

934958 Msg* msg;

935959 Msg* ret = malloc(sizeof(Msg));

936960

937961 struct timespec timeout;

938962 int status;

963+

964+ if (sender && sender->active == 0) { return NULL; } // No VM to receive from

939965

940966 pthread_mutex_lock(&vm->inbox_block);

941− msg = idris_getMessageFrom(vm, sender);

967+ msg = idris_getMessageFrom(vm, channel_id, sender);

942968

943969 while (msg == NULL) {

944970// printf("No message yet\n");

@@ -949,7 +975,7 @@ Msg* idris_recvMessageFrom(VM* vm, VM* sender) {

949975 &timeout);

950976 (void)(status); //don't emit 'unused' warning

951977// printf("Waiting [unlock]... %d\n", status);

952− msg = idris_getMessageFrom(vm, sender);

978+ msg = idris_getMessageFrom(vm, channel_id, sender);

953979 }

954980 pthread_mutex_unlock(&vm->inbox_block);

955981

@@ -993,6 +1019,10 @@ VM* idris_getSender(Msg* msg) {

9931019 return msg->sender;

9941020}

9951021

1022+int idris_getChannel(Msg* msg) {

1023+ return msg->channel_id >> 1;

1024+}

1025+

9961026void idris_freeMsg(Msg* msg) {

9971027 free(msg);

9981028}

rts/idris_rts.h+11−3

@@ -76,6 +76,9 @@ struct VM;

7676

7777struct Msg_t {

7878 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;

7982 VAL msg;

8083};

8184

@@ -101,6 +104,7 @@ struct VM {

101104

102105 Msg* inbox; // Block of memory for storing messages

103106 Msg* inbox_end; // End of block of memory

107+ int inbox_nextid; // Next channel id

104108 Msg* inbox_write; // Location of next message to write

105109

106110 int processes; // Number of child processes

@@ -310,23 +314,27 @@ void* vmThread(VM* callvm, func f, VAL arg);

310314VAL copyTo(VM* newVM, VAL x);

311315

312316// 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);

314318// Check whether there are any messages in the queue and return PID of

315319// sender if so (null if not)

316320VM* 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);

317324// 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);

319326// Check whether there are any messages in the queue, and wait if not

320327VM* idris_checkMessagesTimeout(VM* vm, int timeout);

321328// block until there is a message in the queue

322329Msg* idris_recvMessage(VM* vm);

323330// 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);

325332

326333// Query/free structure used to return message data (recvMessage will malloc,

327334// so needs an explicit free)

328335VAL idris_getMsg(Msg* msg);

329336VM* idris_getSender(Msg* msg);

337+int idris_getChannel(Msg* msg);

330338void idris_freeMsg(Msg* msg);

331339

332340void dumpVal(VAL r);

test/io003/test018.idr+2−2

@@ -10,11 +10,11 @@ pong : IO ()

1010pong = do (sender, x) <- recvMsg

1111 putStrLn x

1212 putStrLn "Received"

13− sendToThread sender "Hello to you too!"

13+ sendToThread sender 0 "Hello to you too!"

1414 return ()

1515

1616ping : Ptr -> IO ()

17−ping thread = do sendToThread thread (prim__vm, "Hello!")

17+ping thread = do sendToThread thread 0 (prim__vm, "Hello!")

1818 return ()

1919

2020pingpong : IO ()

test/proofsearch002/Process.idr+11−11

@@ -248,15 +248,15 @@ data Message : (Type -> Type) -> List (ReqHandle, Type) -> Type where

248248readMsg : IO (Maybe (Ptr, Message iface hs))

249249readMsg {iface} {hs} =

250250 do if !checkMsgs

251− then do msg <- getMsgWithSender {a = Message iface hs}

252− pure (Just msg)

251+ then do (p, _, msg) <- getMsgWithSender {a = Message iface hs}

252+ pure (Just (p, msg))

253253 else pure Nothing

254254

255255readMsgTimeout : Int -> IO (Maybe (Ptr, Message iface hs))

256256readMsgTimeout {iface} {hs} i =

257257 do if !(checkMsgsTimeout i)

258− then do msg <- getMsgWithSender {a = Message iface hs}

259− pure (Just msg)

258+ then do (pid, _, msg) <- getMsgWithSender {a = Message iface hs}

259+ pure (Just (pid, msg))

260260 else pure Nothing

261261

262262data RespEnv : List (ReqHandle, Type) -> Type where

@@ -383,7 +383,7 @@ eval st (Work proc cont) k

383383 eval (record { clients = clients st + 1 } st) cont k

384384

385385eval {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))

387387 k (MkReqH nh) (MkEvalState q (Nothing :: reqs) c (S nh))

388388

389389eval {p} st (GetReply {pending} h) k

@@ -407,7 +407,7 @@ eval {iface} {hs} st (TimeoutRespond timeout def f) k

407407 (\ r, st''' =>

408408 case r of

409409 (resp, val) => do

410− sendToThread pid (MsgReply {iface} {hs} (ReplyMsg rq resp))

410+ sendToThread pid 0 (MsgReply {iface} {hs} (ReplyMsg rq resp))

411411 k val st''')

412412

413413eval {iface} {hs} st (Respond f) k

@@ -416,19 +416,19 @@ eval {iface} {hs} st (Respond f) k

416416 Nothing => eval {iface} st' (Respond f) k

417417 Just (pid, (_ ** (rq, req)), st'') => do

418418 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))

420420 k val st''')

421421

422422eval {hs} st (Connect {serveri} (MkPID pid)) k

423423 = if pid == prim__vm then k False st else do

424− v <- sendToThread pid (MsgQuery {iface=serveri} {hs}

425− ConnectMsg)

424+ v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}

425+ ConnectMsg)

426426 -- TODO: Wait for ACK

427427 k (v == 1) st

428428

429429eval {hs} st (Disconnect {serveri} (MkPID pid)) k

430− = do v <- sendToThread pid (MsgQuery {iface=serveri} {hs}

431− CloseMsg)

430+ = do v <- sendToThread pid 0 (MsgQuery {iface=serveri} {hs}

431+ CloseMsg)

432432 k () st

433433

434434eval st CountClients k