Starting in version 1.6.0, Janet has the ability to do true, non-cooperative multithreading with the
functions. Janet threads correspond to native threads on the host operating system, which may be either pthreads on
posix systems or Windows threads. Each thread has it's own Janet heap, which means threads behave more like processes
that communicate by message passing. However, this does not prevent native code from sharing memory across these threads.
Without native extensions, however, the only for two Janet threads to communicate directly is through message passing.
Use `(thread/new func &opt capacity)` to create a new thread. This thread will start and wait for a message containing a function
that it will run as the main body. This function must also be able to take 1 parameter, the parent thread.
Each thread has it's own mailbox, which can asynchronously receive messages. These messages are added to a queue, which has
a configurable maximum capacity according to the second optional argument passed to
thread/new. If unspecified, the mailbox
will have an initial capacity of 10.
(defn worker [parent] (print "New thread started!")) # New thread's mailbox has capacity for 32 messages. (def thread (new/thread worker 32))
Sending and Receiving Messages
Since threads do not share memory a heap, the way threads must communicate is through message passing. Use
to send a message to a thread, and
thread/receive to get messages sent to the current thread.
(defn worker [parent] (print "waiting for message...") (def msg (thread/receive)) (print "got message: " msg)) (def thread (thread/new worker)) # Thread objects support the :send method as an alias for thread/send. (:send thread "Hello!")
Limitations of Messages
Since threads do not share Janet heaps, all values sent as messages are first marshalled to a byte sequence by the sending thread, and then
unmarshalled by the receiving thread. For marshalling and unmarshalling, threads use the two tables
load-image-dict. This means you can send over core bindings, even if the underlying value cannot be marshalled.
Semantically, messages sent with
(thread/send to msg) are first converted to a buffer
(marshal msg make-image-dict mailbox-buf),
and then unmarshalled by the receiving thread via
(unmarshal mailbox-buf load-image-dict).
Values that cannot be marshalled, including thread values, cannot be sent as messages to other threads, or even as part of a message to another thread. For example, the following will not work because open file handles cannot be marshalled.
(def file (file/open "myfile.txt" :w)) (defn worker [parent] (with-dyns [:out file] (print "Writing to file."))) # Will throw an error, as worker contains a reference to file, which cannot be marshalled. (thread/new worker)
The fix here is to move the file creation inside the worker function, since every worker thread is going to have it's own copy of the file handle.
(defn worker [parent] (def file (file/open "myfile.txt" :w)) (with-dyns [:out file] (print "Writing to file."))) # Will throw an error, as worker contains a reference to file, which cannot be marshalled. (thread/new worker)
Blocking and Non-blocking Sends and Receives
All sends and receives can be blocking or non-blocking by providing and optional timeout value. A timeout
of 0 makes
(thread/send to msg &opt timeout) and
(thread/receive &opt timeout) non-blocking, while a positive timeout value
indicates and blocking send or receive, that will resume the current thread by throwing an error after
timeout seconds. Timeouts are limited to 30 days, with exception the using
math/inf for a timeout means that
a timeout will never occur.
(defn worker [parent] (for i 0 10 (os/sleep 0.5) (:send parent :ping)) # sleep a long while to make the parent timeout (os/sleep 2) (:send parent :done)) # Will print the first 10 pings, but timeout and throw an error before :done (try (let [t (thread/new worker)] (for i 0 11 (print "got message: " (thread/receive 1)))) ([err] (print "error: " err))) # Flush :done message (thread/receive math/inf) # Will not error and work successfully (let [t (thread/new worker)] (for i 0 11 (print "got message: " (thread/receive math/inf))))