0b6726f · Idris-dev · git.iko.soy

git.iko.soy/Idris-dev

Improve concurrency primitives in C RTS

Now allow recv to only receive from a specific sender. This way a server (for example) can carry on a conversation with a specific process without having to mess about with the mailbox in various error prone ways...

commit0b6726ff49dce6ba594d9092c65de9e8aa0d6b78 authorEdwin Brady <ecb10@st-andrews.ac.uk> dateSun 15 Feb 2015, 21:46 (12 years ago) parent0fdbc0b

5 files changed+152−32

compare with master

unifiedsplit

libs/base/System/Concurrency/Process.idr+11 libs/base/System/Concurrency/Raw.idr+35−2 rts/idris_gc.c+4−2 rts/idris_rts.c+75−20 rts/idris_rts.h+27−8

libs/base/System/Concurrency/Process.idr+11

@@ -41,6 +41,10 @@ send (MkPID p) m = Lift (sendToThread p (prim__vm, m))

4141msgWaiting : Process msg Bool

4242msgWaiting = Lift checkMsgs

4343

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

45+msgWaitingFrom : ProcID msg -> Process msg Bool

46+msgWaitingFrom (MkPID p) = Lift (checkMsgsFrom p)

47+

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

4549recv : Process msg msg

4650recv {msg} = do (senderid, m) <- Lift get

@@ -48,6 +52,13 @@ recv {msg} = do (senderid, m) <- Lift get

4852 where get : IO (Ptr, msg)

4953 get = getMsg

5054

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

56+recvFrom : ProcID msg -> Process msg msg

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

58+ return m

59+ where get : IO (Ptr, msg)

60+ get = getMsgFrom p

61+

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

5263recvWithSender : Process msg (ProcID msg, msg)

5364recvWithSender {msg}

libs/base/System/Concurrency/Raw.idr+35−2

@@ -12,14 +12,47 @@ sendToThread {a} dest val

1212 = foreign FFI_C "idris_sendMessage" (Ptr -> Ptr -> Raw a -> IO ())

1313 prim__vm dest (MkRaw val)

1414

15+||| Check for messages in the process inbox

1516checkMsgs : IO Bool

1617checkMsgs = do msgs <- foreign FFI_C "idris_checkMessages" (Ptr -> IO Int)

1718 prim__vm

1819 return (intToBool msgs)

1920

21+||| Check for messages from a specific sender in the process inbox

22+checkMsgsFrom : Ptr -> IO Bool

23+checkMsgsFrom sender

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

25+ prim__vm sender

26+ return (intToBool msgs)

27+

2028||| Check inbox for messages. If there are none, blocks until a message

2129||| arrives.

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

31+||| a type safe wrapper.

2232getMsg : IO a

23−getMsg {a} = do MkRaw x <- foreign FFI_C "idris_recvMessage"

24− (Ptr -> IO (Raw a)) prim__vm

33+getMsg {a} = do m <- foreign FFI_C "idris_recvMessage"

34+ (Ptr -> IO Ptr) prim__vm

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

2536 return x

37+

38+||| Check inbox for messages. If there are none, blocks until a message

39+||| arrives. Return pair of sender's ID and the message.

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

41+||| a type safe wrapper.

42+getMsgWithSender : IO (Ptr, a)

43+getMsgWithSender {a}

44+ = do m <- foreign FFI_C "idris_recvMessage"

45+ (Ptr -> IO Ptr) prim__vm

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

47+ vm <- foreign FFI_C "idris_getSender" (Ptr -> IO Ptr) m

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

49+ return (vm, x)

50+

51+getMsgFrom : Ptr -> IO a

52+getMsgFrom {a} sender

53+ = do m <- foreign FFI_C "idris_recvMessageFrom"

54+ (Ptr -> Ptr -> IO Ptr) prim__vm sender

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

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

57+ return x

58+

rts/idris_gc.c+4−2

@@ -137,8 +137,10 @@ void idris_gc(VM* vm) {

137137 }

138138

139139#ifdef HAS_PTHREAD

140− for(root = vm->inbox_ptr; root < vm->inbox_write; ++root) {

141− *root = copy(vm, *root);

140+ Msg* msg;

141+

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

143+ msg->msg = copy(vm, msg->msg);

142144 }

143145#endif

144146 vm->ret = copy(vm, vm->ret);

rts/idris_rts.c+75−20

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

4444 vm->inbox = malloc(1024*sizeof(VAL));

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

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

47− vm->inbox_ptr = vm->inbox;

4847 vm->inbox_write = vm->inbox;

4948

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

@@ -1096,18 +1095,19 @@ void idris_sendMessage(VM* sender, VM* dest, VAL msg) {

10961095 }

10971096

10981097 pthread_mutex_lock(&(dest->inbox_lock));

1099− *(dest->inbox_write) = dmsg;

1100−

1101− dest->inbox_write++;

1098+

11021099 if (dest->inbox_write >= dest->inbox_end) {

1103− dest->inbox_write = dest->inbox;

1104− }

1105−

1106− if (dest->inbox_write == dest->inbox_ptr) {

1107− fprintf(stderr, "Inbox full"); // Maybe grow it instead...

1100+ // FIXME: This is obviously bad in the long run. This should

1101+ // either block, make the inbox bigger, or return an error code,

1102+ // or possibly make it user configurable

1103+ fprintf(stderr, "Inbox full");

11081104 exit(-1);

11091105 }

11101106

1107+ dest->inbox_write->msg = dmsg;

1108+ dest->inbox_write->sender = sender;

1109+ dest->inbox_write++;

1110+

11111111 // Wake up the other thread

11121112 pthread_mutex_lock(&dest->inbox_block);

11131113 pthread_cond_signal(&dest->inbox_waiting);

@@ -1120,18 +1120,46 @@ void idris_sendMessage(VM* sender, VM* dest, VAL msg) {

11201120}

11211121

11221122int idris_checkMessages(VM* vm) {

1123− VAL msg = *(vm->inbox_ptr);

1124− return (msg != NULL);

1123+ return idris_checkMessagesFrom(vm, NULL);

1124+}

1125+

1126+int idris_checkMessagesFrom(VM* vm, VM* sender) {

1127+ Msg* msg;

1128+

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

1130+ if (sender == NULL || msg->sender == sender) {

1131+ return 1;

1132+ }

1133+ }

1134+ return 0;

1135+}

1136+

1137+Msg* idris_getMessageFrom(VM* vm, VM* sender) {

1138+ Msg* msg;

1139+

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

1141+ if (sender == NULL || msg->sender == sender) {

1142+ return msg;

1143+ }

1144+ }

1145+ return NULL;

11251146}

11261147

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

1128−VAL idris_recvMessage(VM* vm) {

1129− VAL msg;

1149+Msg* idris_recvMessage(VM* vm) {

1150+ return idris_recvMessageFrom(vm, NULL);

1151+}

1152+

1153+Msg* idris_recvMessageFrom(VM* vm, VM* sender) {

1154+ Msg* msg;

1155+ Msg* ret = malloc(sizeof(Msg));

1156+

11301157 struct timespec timeout;

11311158 int status;

11321159

11331160 pthread_mutex_lock(&vm->inbox_block);

1134− msg = *(vm->inbox_ptr);

1161+ msg = idris_getMessageFrom(vm, sender);

1162+

11351163 while (msg == NULL) {

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

11371165// printf("Waiting [lock]...\n");

@@ -1141,27 +1169,54 @@ VAL idris_recvMessage(VM* vm) {

11411169 &timeout);

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

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

1144− msg = *(vm->inbox_ptr);

1172+ msg = idris_getMessageFrom(vm, sender);

11451173 }

11461174 pthread_mutex_unlock(&vm->inbox_block);

11471175

11481176 if (msg != NULL) {

1177+ ret->msg = msg->msg;

1178+ ret->sender = msg->sender;

1179+

11491180 pthread_mutex_lock(&(vm->inbox_lock));

1150− *(vm->inbox_ptr) = NULL;

1151− vm->inbox_ptr++;

1152− if (vm->inbox_ptr >= vm->inbox_end) {

1153− vm->inbox_ptr = vm->inbox;

1181+

1182+ // Slide everything down after the message in the inbox,

1183+ // Move the inbox_write pointer down, and clear the value at the

1184+ // end - O(n) but it's easier since the message from a specific

1185+ // sender could be anywhere in the inbox

1186+

1187+ for(;msg < vm->inbox_write; ++msg) {

1188+ if (msg+1 != vm->inbox_end) {

1189+ msg->sender = (msg + 1)->sender;

1190+ msg->msg = (msg + 1)->msg;

1191+ }

11541192 }

1193+

1194+ vm->inbox_write->msg = NULL;

1195+ vm->inbox_write->sender = NULL;

1196+ vm->inbox_write--;

1197+

11551198 pthread_mutex_unlock(&(vm->inbox_lock));

11561199 } else {

11571200 fprintf(stderr, "No messages waiting");

11581201 exit(-1);

11591202 }

11601203

1161− return msg;

1204+ return ret;

11621205}

11631206#endif

11641207

1208+VAL idris_getMsg(Msg* msg) {

1209+ return msg->msg;

1210+}

1211+

1212+VM* idris_getSender(Msg* msg) {

1213+ return msg->sender;

1214+}

1215+

1216+void idris_freeMsg(Msg* msg) {

1217+ free(msg);

1218+}

1219+

11651220VAL* nullary_cons;

11661221

11671222void init_nullaries() {

rts/idris_rts.h+27−8

@@ -80,7 +80,16 @@ typedef struct Closure {

8080 } info;

8181} Closure;

8282

83−typedef struct {

83+struct VM_t;

84+

85+struct Msg_t {

86+ struct VM_t* sender;

87+ VAL msg;

88+};

89+

90+typedef struct Msg_t Msg;

91+

92+struct VM_t {

8493 VAL* valstack;

8594 VAL* valstack_top;

8695 VAL* valstack_base;

@@ -93,11 +102,9 @@ typedef struct {

93102 pthread_mutex_t alloc_lock;

94103 pthread_cond_t inbox_waiting;

95104

96− VAL* inbox; // Block of memory for storing messages

97− VAL* inbox_end; // End of block of memory

98−

99− VAL* inbox_ptr; // Next message to read

100− VAL* inbox_write; // Location of next message to write

105+ Msg* inbox; // Block of memory for storing messages

106+ Msg* inbox_end; // End of block of memory

107+ Msg* inbox_write; // Location of next message to write

101108

102109 int processes; // Number of child processes

103110 int max_threads; // maximum number of threads to run in parallel

@@ -106,7 +113,9 @@ typedef struct {

106113

107114 VAL ret;

108115 VAL reg1;

109−} VM;

116+};

117+

118+typedef struct VM_t VM;

110119

111120// Create a new VM

112121VM* init_vm(int stack_size, size_t heap_size,

@@ -279,8 +288,18 @@ VAL copyTo(VM* newVM, VAL x);

279288void idris_sendMessage(VM* sender, VM* dest, VAL msg);

280289// Check whether there are any messages in the queue

281290int idris_checkMessages(VM* vm);

291+// Check whether there are any messages in the queue

292+int idris_checkMessagesFrom(VM* vm, VM* sender);

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

283−VAL idris_recvMessage(VM* vm);

294+Msg* idris_recvMessage(VM* vm);

295+// block until there is a message in the queue

296+Msg* idris_recvMessageFrom(VM* vm, VM* sender);

297+

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

299+// so needs an explicit free)

300+VAL idris_getMsg(Msg* msg);

301+VM* idris_getSender(Msg* msg);

302+void idris_freeMsg(Msg* msg);

284303

285304void dumpVal(VAL r);

286305void dumpStack(VM* vm);