7831e10 · Idris-dev · git.iko.soy

git.iko.soy/Idris-dev

Check messages are passed to active VMs

As currently implemented, this will leak a few bytes per VM after they terminate, but it is at least better than segfaulting when sending messages to VMs which have terminated.

commit7831e10f1cd0632303462f6f660314d2660f24e0 authorEdwin Brady <ecb10@st-andrews.ac.uk> dateSat 11 Jul 2015, 23:11 (11 years ago) parent386ac94

6 files changed+28−8

compare with master

unifiedsplit

libs/base/System/Concurrency/Raw.idr+3−2 libs/contrib/System/Concurrency/Process.idr+4−2 rts/idris_rts.c+11−2 rts/idris_rts.h+5−1 test/io003/test018.idr+3−1 test/io003/test018a.idr+2

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

@@ -7,9 +7,10 @@ module System.Concurrency.Raw

77import System

99||| Send a message of any type to the thread with the given thread id

10−sendToThread : (thread_id : Ptr) -> a -> IO ()

10+||| Returns 1 if the message was sent successfully, 0 otherwise

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

1112sendToThread {a} dest val

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

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

1314 prim__vm dest (MkRaw val)

1415

1516||| Check for messages in the process inbox

libs/contrib/System/Concurrency/Process.idr+4−2

@@ -34,8 +34,10 @@ myID : Process msg (ProcID msg)

3434myID = Lift (return (MkPID prim__vm))

3535

3636||| Send a message to another process

37−send : ProcID msg -> msg -> Process msg ()

38−send (MkPID p) m = Lift (sendToThread p (prim__vm, m))

37+||| Returns whether the send was unsuccessful.

38+send : ProcID msg -> msg -> Process msg Bool

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

40+ return (x == 1))

3941

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

4143msgWaiting : Process msg Bool

rts/idris_rts.c+11−2

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

2626

2727 VAL* valstack = malloc(stack_size * sizeof(VAL));

2828

29+ vm->active = 1;

2930 vm->valstack = valstack;

3031 vm->valstack_top = valstack;

3132 vm->valstack_base = valstack;

@@ -104,7 +105,12 @@ Stats terminate(VM* vm) {

104105 pthread_mutex_destroy(&(vm -> inbox_block));

105106 pthread_cond_destroy(&(vm -> inbox_waiting));

106107#endif

107− free(vm);

108+ // free(vm);

109+ // Set the VM as inactive, so that if any message gets sent to it

110+ // it will not get there, rather than crash the entire system.

111+ // (TODO: We really need to be cleverer than this if we're going to

112+ // write programs than create lots of threads...)

113+ vm->active = 0;

108114

109115 STATS_LEAVE_EXIT(stats)

110116 return stats;

@@ -738,7 +744,7 @@ VAL copyTo(VM* vm, VAL x) {

738744}

739745

740746// Add a message to another VM's message queue

741−void idris_sendMessage(VM* sender, VM* dest, VAL msg) {

747+int idris_sendMessage(VM* sender, VM* dest, VAL msg) {

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

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

744750

@@ -749,6 +755,8 @@ void idris_sendMessage(VM* sender, VM* dest, VAL msg) {

749755 // So: we try to copy, if a collection happens, we do the copy again

750756 // under the assumption there's enough space this time.

751757

758+ if (dest->active == 0) { return 0; } // No VM to send to

759+

752760 int gcs = dest->stats.collections;

753761 pthread_mutex_lock(&dest->alloc_lock);

754762 VAL dmsg = copyTo(dest, msg);

@@ -784,6 +792,7 @@ void idris_sendMessage(VM* sender, VM* dest, VAL msg) {

784792

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

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

795+ return 1;

787796}

788797

789798VM* idris_checkMessages(VM* vm) {

rts/idris_rts.h+5−1

@@ -78,6 +78,10 @@ struct Msg_t {

7878typedef struct Msg_t Msg;

7979

8080struct VM_t {

81+ int active; // 0 if no longer running; keep for message passing

82+ // TODO: If we're going to have lots of concurrent threads,

83+ // we really need to be cleverer than this!

84+

8185 VAL* valstack;

8286 VAL* valstack_top;

8387 VAL* valstack_base;

@@ -253,7 +257,7 @@ void* vmThread(VM* callvm, func f, VAL arg);

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

254258

255259// Add a message to another VM's message queue

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

260+int idris_sendMessage(VM* sender, VM* dest, VAL msg);

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

258262// sender if so (null if not)

259263VM* idris_checkMessages(VM* vm);

test/io003/test018.idr+3−1

@@ -11,9 +11,11 @@ pong = do (sender, x) <- recvMsg

1111 putStrLn x

1212 putStrLn "Received"

1313 sendToThread sender "Hello to you too!"

14+ return ()

1415

1516ping : Ptr -> IO ()

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

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

18+ return ()

1719

1820pingpong : IO ()

1921pingpong

test/io003/test018a.idr+2

@@ -11,12 +11,14 @@ ping main proc

1111 msg <- recv

1212 Lift (putStrLn ("Reply: " ++ show msg))

1313 send main "Done"

14+ return ()

1415

1516pong : Process String ()

1617pong = do -- Lift (putStrLn "Waiting for message")

1718 (sender, m) <- recvWithSender

1819 Lift $ putStrLn ("Received " ++ m)

1920 send sender ("Hello back!")

21+ return ()

2022

2123mainProc : Process String ()

2224mainProc = do mainID <- myID