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

git.iko.soy/Idris-dev

on master help sign in

Improve concurrency primitives in C RTS

0b6726fEdwin Brady12 years ago

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

compare with master

unifiedsplit

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

41msgWaiting : Process msg Bool

42msgWaiting = Lift checkMsgs

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+

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

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

45recv : Process msg msg

49recv : Process msg msg

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

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

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

48 where get : IO (Ptr, msg)

52 where get : IO (Ptr, msg)

49 get = getMsg

53 get = getMsg

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+

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

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

52recvWithSender : Process msg (ProcID msg, msg)

63recvWithSender : Process msg (ProcID msg, msg)

53recvWithSender {msg}

64recvWithSender {msg}

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

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

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

13 prim__vm dest (MkRaw val)

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

15checkMsgs : IO Bool

16checkMsgs : IO Bool

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

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

17 prim__vm

18 prim__vm

18 return (intToBool msgs)

19 return (intToBool msgs)

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+

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

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

21||| arrives.

29||| arrives.

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

31+||| a type safe wrapper.

22getMsg : IO a

32getMsg : IO a

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

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

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

34+ (Ptr -> IO Ptr) prim__vm

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

25 return x

36 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) {

137 }

138

139#ifdef HAS_PTHREAD

140+ Msg* msg;

141+

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

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

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

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

142 }

144 }

143#endif

145#endif

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

146 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,

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

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

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

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

48 vm->inbox_write = vm->inbox;

47 vm->inbox_write = vm->inbox;

50 // 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

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

1096 }

1095 }

1097

1096

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

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

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

1098+

1100−

1101− dest->inbox_write++;

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

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

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

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

1104− }

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

1105−

1102+ // or possibly make it user configurable

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

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

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

1108 exit(-1);

1104 exit(-1);

1109 }

1105 }

1110

1106

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

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

1109+ dest->inbox_write++;

1110+

1111 // Wake up the other thread

1112 pthread_mutex_lock(&dest->inbox_block);

1113 pthread_cond_signal(&dest->inbox_waiting);

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

1120}

1121

1122int idris_checkMessages(VM* vm) {

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

1123+ return idris_checkMessagesFrom(vm, NULL);

1124− return (msg != 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;

1125}

1146}

1126

1147

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

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

1128−VAL idris_recvMessage(VM* vm) {

1149+Msg* idris_recvMessage(VM* vm) {

1150+ return idris_recvMessageFrom(vm, NULL);

1151+}

1152+

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

1129− VAL msg;

1154+ Msg* msg;

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

1156+

1130 struct timespec timeout;

1157 struct timespec timeout;

1131 int status;

1158 int status;

1132

1159

1133 pthread_mutex_lock(&vm->inbox_block);

1160 pthread_mutex_lock(&vm->inbox_block);

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

1161+ msg = idris_getMessageFrom(vm, sender);

1162+

1135 while (msg == NULL) {

1163 while (msg == NULL) {

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

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

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

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

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

1141 &timeout);

1169 &timeout);

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

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

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

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

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

1172+ msg = idris_getMessageFrom(vm, sender);

1145 }

1173 }

1146 pthread_mutex_unlock(&vm->inbox_block);

1174 pthread_mutex_unlock(&vm->inbox_block);

1147

1175

1148 if (msg != NULL) {

1176 if (msg != NULL) {

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

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

1179+

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

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

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

1181+

1151− vm->inbox_ptr++;

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

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

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

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

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

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

1191+ }

1154 }

1192 }

1193+

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

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

1196+ vm->inbox_write--;

1197+

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

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

1156 } else {

1199 } else {

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

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

1158 exit(-1);

1201 exit(-1);

1159 }

1202 }

1160

1203

1161− return msg;

1204+ return ret;

1162}

1205}

1163#endif

1206#endif

1164

1207

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+

1165VAL* nullary_cons;

1220VAL* nullary_cons;

1166

1221

1167void init_nullaries() {

1222void init_nullaries() {

rts/idris_rts.h+27−8

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

80 } info;

81} Closure;

83+struct VM_t;

84+

85+struct Msg_t {

86+ struct VM_t* sender;

87+ VAL msg;

88+};

89+

83−typedef struct {

90+typedef struct Msg_t Msg;

91+

92+struct VM_t {

84 VAL* valstack;

93 VAL* valstack;

85 VAL* valstack_top;

94 VAL* valstack_top;

86 VAL* valstack_base;

95 VAL* valstack_base;

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

93 pthread_mutex_t alloc_lock;

102 pthread_mutex_t alloc_lock;

94 pthread_cond_t inbox_waiting;

103 pthread_cond_t inbox_waiting;

104

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

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

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

106+ Msg* 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

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

101

108

102 int processes; // Number of child processes

109 int processes; // Number of child processes

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

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

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

106

113

107 VAL ret;

114 VAL ret;

108 VAL reg1;

115 VAL reg1;

109−} VM;

116+};

117+

118+typedef struct VM_t VM;

110

119

111// Create a new VM

120// Create a new VM

112VM* init_vm(int stack_size, size_t heap_size,

121VM* init_vm(int stack_size, size_t heap_size,

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

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

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

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

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

281int idris_checkMessages(VM* vm);

290int idris_checkMessages(VM* vm);

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

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

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

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

284

303

285void dumpVal(VAL r);

304void dumpVal(VAL r);

286void dumpStack(VM* vm);

305void dumpStack(VM* vm);