18 Concurrency and Synchronization
Racket provides concurrency in the form of threads, and it provides a general sync function that can be used to synchronize both threads and other implicit forms of concurrency, such as ports.
Threads run concurrently in the sense that one thread can preempt another without its cooperation, but threads do not run in parallel in the sense of using multiple hardware processors. See Parallelism for information on parallelism in Racket.
18.1 Threads
To execute a procedure concurrrently, use thread. The following example creates two new threads from the main thread:
(displayln "This is the original thread") (thread (lambda () (displayln "This is a new thread."))) (thread (lambda () (displayln "This is another new thread.")))
The next example creates a new thread that would otherwise loop forever, but the main thread uses sleep to pause itself for 2.5 seconds, then uses kill-thread to terminate the worker thread:
(define worker (thread (lambda () (let loop () (displayln "Working...") (sleep 0.2) (loop))))) (sleep 2.5) (kill-thread worker)
In DrRacket, the main thread keeps going until the Stop button is clicked, so in DrRacket the thread-wait is not necessary.
If the main thread finishes or is killed, the application exits, even if other threads are still running. A thread can use thread-wait to wait for another thread to finish. Here, the main thread uses thread-wait to make sure the worker thread finishes before the main thread exits:
(define worker (thread (lambda () (for ([i 100]) (printf "Working hard... ~a~n" i))))) (thread-wait worker) (displayln "Worker finished")
18.2 Thread Mailboxes
Each thread has a mailbox for receiving messages. The thread-send function asynchronously sends a message to another thread’s mailbox, while thread-receive returns the oldest message from the current thread’s mailbox, blocking to wait for a message if necessary. In the following example, the main thread sends data to the worker thread to be processed, then sends a 'done message when there is no more data and waits for the worker thread to finish.
(define worker-thread (thread (lambda () (let loop () (match (thread-receive) [(? number? num) (printf "Processing ~a~n" num) (loop)] ['done (printf "Done~n")]))))) (for ([i 20]) (thread-send worker-thread i)) (thread-send worker-thread 'done) (thread-wait worker-thread)
In the next example, the main thread delegates work to multiple arithmetic threads, then waits to receive the results. The arithmetic threads process work items then send the results to the main thread.
(define (make-arithmetic-thread operation) (thread (lambda () (let loop () (match (thread-receive) [(list oper1 oper2 result-thread) (thread-send result-thread (format "~a + ~a = ~a" oper1 oper2 (operation oper1 oper2))) (loop)]))))) (define addition-thread (make-arithmetic-thread +)) (define subtraction-thread (make-arithmetic-thread -)) (define worklist '((+ 1 1) (+ 2 2) (- 3 2) (- 4 1))) (for ([item worklist]) (match item [(list '+ o1 o2) (thread-send addition-thread (list o1 o2 (current-thread)))] [(list '- o1 o2) (thread-send subtraction-thread (list o1 o2 (current-thread)))])) (for ([i (length worklist)]) (displayln (thread-receive)))
18.3 Semaphores
Semaphores facilitate synchronized access to an arbitrary shared resource. Use semaphores when multiple threads must perform non-atomic operations on a single resource.
In the following example, multiple threads print to standard output concurrently. Without synchronization, a line printed by one thread might appear in the middle of a line printed by another thread. By using a semaphore initialized with a count of 1, only one thread will print at a time. The semaphore-wait function blocks until the semaphore’s internal counter is non-zero, then decrements the counter and returns. The semaphore-post function increments the counter so that another thread can unblock and then print.
(define output-semaphore (make-semaphore 1)) (define (make-thread name) (thread (lambda () (for [(i 10)] (semaphore-wait output-semaphore) (printf "thread ~a: ~a~n" name i) (semaphore-post output-semaphore))))) (define threads (map make-thread '(A B C))) (for-each thread-wait threads)
The pattern of waiting on a semaphore, working, and posting to the semaphore can also be expressed using call-with-semaphore,which has the advantage of posting to the semaphore if control escapes (e.g., due to an exception):
(define output-semaphore (make-semaphore 1)) (define (make-thread name) (thread (lambda () (for [(i 10)] (call-with-semaphore output-semaphore (lambda () (printf "thread ~a: ~a~n" name i))))))) (define threads (map make-thread '(A B C))) (for-each thread-wait threads)
Semaphores are a low-level technique. Often, a better solution is to restrict resource access to a single thread. For example, synchronizing access to standard output might be better accomplished by having a dedicated thread for printing output.
18.4 Channels
Channels synchronize two threads while a value is passed from one thread to the other. Unlike a thread mailbox, multiple threads can get items from a single channel, so channels should be used when multiple threads need to consume items from a single work queue.
In the following example, the main thread adds items to a channel using channel-put, while multiple worker threads consume those items using channel-get. Each call to either procedure blocks until another thread calls the other procedure with the same channel. The workers process the items and then pass their results to the result thread via the result-channel.
(define result-channel (make-channel)) (define result-thread (thread (lambda () (let loop () (displayln (channel-get result-channel)) (loop))))) (define work-channel (make-channel)) (define (make-worker thread-id) (thread (lambda () (let loop () (define item (channel-get work-channel)) (case item [(DONE) (channel-put result-channel (format "Thread ~a done" thread-id))] [else (channel-put result-channel (format "Thread ~a processed ~a" thread-id item)) (loop)]))))) (define work-threads (map make-worker '(1 2))) (for ([item '(A B C D E F G H DONE DONE)]) (channel-put work-channel item)) (for-each thread-wait work-threads)
18.5 Buffered Asynchronous Channels
Buffered asynchronous channels are similar to the channels described above, but
the “put” operation of asynchronous channels does not block—
In the following example, the main thread adds items to the work channel, which holds a maximum of three items at a time. The worker threads process items from this channel and then send results to the print thread.
(require racket/async-channel) (define print-thread (thread (lambda () (let loop () (displayln (thread-receive)) (loop))))) (define (safer-printf . items) (thread-send print-thread (apply format items))) (define work-channel (make-async-channel 3)) (define (make-worker-thread thread-id) (thread (lambda () (let loop () (define item (async-channel-get work-channel)) (safer-printf "Thread ~a processing item: ~a" thread-id item) (loop))))) (for-each make-worker-thread '(1 2 3)) (for ([item '(a b c d e f g h i j k l m)]) (async-channel-put work-channel item))
Note the above example lacks any synchronization to verify that all items were processed. If the main thread were to exit without such synchronization, it is possible that the worker threads will not finish processing some items or the print thread will not print all items.
18.6 Synchronizable Events and sync
There are other ways to synchronize threads. The sync function allows threads to coordinate via synchronizable events. Many values double as events, allowing a uniform way to synchronize threads using different types. Examples of events include channels, ports, threads, and alarms.
In the next example, a channel and an alarm are used as synchronizable events. The workers sync on both so that they can process channel items until the alarm is activated. The channel items are processed, and then results are sent back to the main thread.
(define main-thread (current-thread)) (define alarm (alarm-evt (+ 3000 (current-inexact-milliseconds)))) (define channel (make-channel)) (define (make-worker-thread thread-id) (thread (lambda () (define evt (sync channel alarm)) (cond [(equal? evt alarm) (thread-send main-thread 'alarm)] [else (thread-send main-thread (format "Thread ~a received ~a" thread-id evt))])))) (make-worker-thread 1) (make-worker-thread 2) (make-worker-thread 3) (channel-put channel 'A) (channel-put channel 'B) (let loop () (match (thread-receive) ['alarm (displayln "Done")] [result (displayln result) (loop)]))
The next example shows a function for use in a simple TCP echo server. The function uses sync/timeout to synchronize on input from the given port or a message in the thread’s mailbox. The first argument to sync/timeout specifies the maximum number of seconds it should wait on the given events. The read-line-evt function returns an event that is ready when a line of input is available in the given input port. The result of thread-receive-evt is ready when thread-receive would not block. In a real application, the messages received in the thread mailbox could be used for control messages, etc.
(define (serve in-port out-port) (let loop [] (define evt (sync/timeout 2 (read-line-evt in-port 'any) (thread-receive-evt))) (cond [(not evt) (displayln "Timed out, exiting") (tcp-abandon-port in-port) (tcp-abandon-port out-port)] [(string? evt) (fprintf out-port "~a~n" evt) (flush-output out-port) (loop)] [else (printf "Received a message in mailbox: ~a~n" (thread-receive)) (loop)])))
The serve function is used in the following example, which starts a server thread and a client thread that communicate over TCP. The client prints three lines to the server, which echoes them back. The client’s copy-port call blocks until EOF is received. The server times out after two seconds, closing the ports, which allows copy-port to finish and the client to exit. The main thread uses thread-wait to wait for the client thread to exit (since, without thread-wait, the main thread might exit before the other threads are finished).
(define port-num 4321) (define (start-server) (define listener (tcp-listen port-num)) (thread (lambda () (define-values [in-port out-port] (tcp-accept listener)) (serve in-port out-port)))) (start-server) (define client-thread (thread (lambda () (define-values [in-port out-port] (tcp-connect "localhost" port-num)) (display "first\nsecond\nthird\n" out-port) (flush-output out-port) ; copy-port will block until EOF is read from in-port (copy-port in-port (current-output-port))))) (thread-wait client-thread)
Sometimes, you want to attach result behavior directly to the event passed to sync. In the following example, the worker thread synchronizes on three channels, but each channel must be handled differently. Using handle-evt associates a callback with the given event. When sync selects the given event, it calls the callback to generate the synchronization result, rather than using the event’s normal synchronization result. Since the event is handled in the callback, there is no need to dispatch on the return value of sync.
(define add-channel (make-channel)) (define multiply-channel (make-channel)) (define append-channel (make-channel)) (define (work) (let loop () (sync (handle-evt add-channel (lambda (list-of-numbers) (printf "Sum of ~a is ~a~n" list-of-numbers (apply + list-of-numbers)))) (handle-evt multiply-channel (lambda (list-of-numbers) (printf "Product of ~a is ~a~n" list-of-numbers (apply * list-of-numbers)))) (handle-evt append-channel (lambda (list-of-strings) (printf "Concatenation of ~s is ~s~n" list-of-strings (apply string-append list-of-strings))))) (loop))) (define worker (thread work)) (channel-put add-channel '(1 2)) (channel-put multiply-channel '(3 4)) (channel-put multiply-channel '(5 6)) (channel-put add-channel '(7 8)) (channel-put append-channel '("a" "b"))
The result of handle-evt invokes its callback in tail position with respect to sync, so it is safe to use recursion as in the following example.
(define control-channel (make-channel)) (define add-channel (make-channel)) (define subtract-channel (make-channel)) (define (work state) (printf "Current state: ~a~n" state) (sync (handle-evt add-channel (lambda (number) (printf "Adding: ~a~n" number) (work (+ state number)))) (handle-evt subtract-channel (lambda (number) (printf "Subtracting: ~a~n" number) (work (- state number)))) (handle-evt control-channel (lambda (kill-message) (printf "Done~n"))))) (define worker (thread (lambda () (work 0)))) (channel-put add-channel 2) (channel-put subtract-channel 3) (channel-put add-channel 4) (channel-put add-channel 5) (channel-put subtract-channel 1) (channel-put control-channel 'done) (thread-wait worker)
The wrap-evt function is like handle-evt, except that its handler is not called in tail position with respect to sync. At the same time, wrap-evt disables break exceptions during its handler’s invocation.