Threading in MzScheme

This week I have been playing around with threading in MzScheme… I like it!

(thread (lambda () …))

is enough to create a thread and start it running.

You can use channels to pass data between threads safely with (make-channel), (channel-get <channel>) and (channel-put <channel> <data>).

Just to prove you can write Java in any language… say I wanted to have a connection manager. The API is pretty simple – I want to add connections and remove connections. Also, it might be useful to be able to start and stop the connection manager and display the number of connections. There are a number of ways I could implement this, e.g. have a global variable and synchronise access to put and get connections. Alternatively, I could have a thread that looks after connections and send messages to that thread via a channel. Just for fun, let me choose the second option.

The functionality is encapsulated and available through the following API.

(connection-manager-start)
(connection-manager-stop)
(connection-manager-clear-connections)
(connection-manager-show-connections)
(connection-add <input-port> <output-port>)
(connection-remove <input-port>)

If this was production code, I should put the code in a module and only make the above functions available using (provide…). Outside a module, we can of course by-pass the API and put messages in the channel directly.

;; Support function LOG

(define (pad n) (substring (number->string (+ n 100)) 1 3))

(define (now)
  (let* ((date (seconds->date (current-seconds)))
         (secs (date-second date))
         (mins (date-minute date))
         (hour (date-hour date)))
    (format "~a:~a:~a" (pad hour) (pad mins) (pad secs))))

(define (LOG . args)
  (printf "[ ~a ] : " (now))
  (for-each display args)
  (newline)
  (flush-output))

;; First of all, we define a struct for the connection which has
;; an input and an output port

(define-struct connection (input output))

;; Globals for the channel, and the thread - it is a singleton obviously

(define *connections-channel* (make-channel))
(define *connection-manager-thread* #f)

;; The actual manager function
;;   this constantly loops and accepts messages via *connections-channel*
;;   possible messages are:
;;     'stop, 'add-connection, 'remove-connection,
;;     'show-connection, 'clear-connections

(define (connection-manager)
  (letrec
      ((show-command
        (lambda (command)
          (LOG "Received command [" (symbol->string command) "]")))

       (show-connections
        (lambda (connections)
          (let* ((num (length connections))
                 (plural (not (= num 1))))
            (LOG "There "
                 (if plural "are " "is ")
                 num
                 " connection"
                 (if plural "s" "")))))

       (loop
        (lambda (connections)
          (let* ((command (channel-get *connections-channel*))
                 (head (car command)))
            (cond
             ((eq? head 'stop)
              (LOG "Connection manager stopping..."))
             ((eq? head 'add-connection)
              (loop (cons (cadr command) connections)))
             ((eq? head 'remove-connection)
              (let ((c (cadr command)))
                (LOG "Removing connection [" c "]")
                (loop (remove (lambda (e) (eqv? (connection-input e) c))
                              connections))))
             ((eq? head 'show-connections)
              (show-connections connections)
              (loop connections))
             ((eq? head 'clear-connections)
              (loop '()))
             (else
              (LOG "Error: unrecognised-command [" head "]")
              (loop connections)))))))

    (LOG "Connection manager starting...")
    (loop '())))

;; connection-manager-running? checks if the connection-manager
;; is currently running - it is a belt and braces effort

(define (connection-manager-running?)
  (and (thread? *connection-manager-thread*)
       (thread-running? *connection-manager-thread*)))

;; We always check if the connection-manager is running before
;; trying to send a message as (channel-put ...) waits until there
;; is a corresponding (channel-get ...) on another thread.  When
;; testing at the REPL, this can cause problems.  We could use an
;; asynch stream which wouldn't have this problem but I prefer
;; the explicit check

(define (connection-manager-send-message message)
  (if (connection-manager-running?)
      (channel-put *connections-channel* message)
      (begin
        (LOG "WARN - Connection manager is not currently running")
        #f)))

;; The actual API functions.  Most of these simply check if the
;; connection manager is running, and if so, put the appropriate
;; message in the channel.

(define (connection-manager-start)
  (if (connection-manager-running?)
      (LOG "WARN - The connection manager is already running")
      (set! *connection-manager-thread* (thread connection-manager))))

(define (connection-manager-stop)
  (connection-manager-send-message (list 'stop))
  (set! *connection-manager-thread* #f))

(define (connection-manager-clear-connections)
  (connection-manager-send-message (list 'clear-connections)))

(define (connection-manager-show-connections)
  (connection-manager-send-message (list 'show-connections)))

(define (connection-add input-port output-port)
  (connection-manager-send-message
   (list 'add-connection (make-connection input-port output-port))))

(define (connection-remove input-port)
  (connection-manager-send-message (list 'remove-connection input-port)))

;;; Connection Manager Test

;; The test...

(define (test-connection-manager)
  (connection-manager-start)
  (connection-manager-clear-connections)
  (connection-manager-show-connections)
  (connection-add 1 2)
  (connection-manager-show-connections)
  (connection-remove 1)
  (connection-manager-show-connections)

  (time
   (begin
     (do ((i 0 (+ i 1)))
         ((= i 100000) #t)
       (connection-add 1 2))
     (connection-manager-show-connections)))

  (connection-manager-stop))

;; (test-connection-manager)

About these ads