;; Simple Actor Machine
;;
;; A virtual machine which houses a population of actors which can
;; communicate using messages with actors on the same machine or other
;; machines via the network.

(import scheme
        (chicken base)
        (chicken io)
        (chicken string)
        (chicken port)
        (chicken process-context)
        (chicken file)
        (chicken condition)
        matchable
        srfi-18 ; threads
        srfi-69 ; hash-table
        uuid ; ids for actors
        uri-generic
        udp
        fifo)

;; Global variables

(define trace #f)

(define sam-host "localhost")
(define sam-port 8000)

(define sam-version "0.1")

;; Actors


(define (make-address host port id)
  (list id host port))

(define (make-local-address . args)
  (make-address sam-host
                sam-port
                (if (null? args)
                    (uuid)
                    (car args))))

(define (address-id address)
  (car address))
(define (address-host address)
  (cadr address))
(define (address-port address)
  (caddr address))
(define (address->string address)
  (uri->string
   (make-uri #:scheme "actor"
             #:host (address-host address)
             #:port (address-port address)
             #:path (list '/ (address-id address)))))
(define (string->address str)
  (let ((uri (uri-reference str)))
    (make-address (uri-host uri)
                  (uri-port uri)
                  (cadr (uri-path uri)))))

(define (address-local? address)
  (and (equal? (address-host address) sam-host)
       (equal? (address-port address) sam-port)))

(define actor-table (make-hash-table))

(define (make-actor beh)
  (let* ((address (make-local-address))
         (id (address-id address)))
    (hash-table-set! actor-table id beh)
    address))

(define (deliver-message address . message)
  (if trace (print "Delivering to " address ": " message))
  (let ((id (address-id address)))
    (let ((behaviour (hash-table-ref/default actor-table id '())))
      (if (null? behaviour)
          (print "Warning: discarded message " message
                 " to unknown actor id " id)
          (match (apply (hash-table-ref actor-table id) (cons address message))
            ('done (hash-table-delete! actor-table id))
            ('sleep 'do-nothing)
            (new-beh (hash-table-set! actor-table id new-beh)))))))

;; Scheduler

(define local-queue-mutex (make-mutex "message queue"))
(define message-available-mutex (make-mutex "message available"))
(mutex-lock! message-available-mutex #f #f)
(define local-queue (make-fifo))

(define (send-message address . message)
  (apply (if (address-local? address)
             send-local-message
             send-network-message)
         (cons address message)))

(define (send-local-message address . message)
  (mutex-lock! local-queue-mutex)
  (fifo-push local-queue (cons address message))
  (mutex-unlock! message-available-mutex)
  (mutex-unlock! local-queue-mutex))

(define (send-network-message address . message)
  (let ((s (udp-open-socket))
        (packet (with-output-to-string
                  (lambda ()
                    (write (cons address message))))))
    (udp-bind! s #f 0)
    (udp-connect! s
                  (address-host address)
                  (address-port address))
    (udp-send s packet)
    (udp-close-socket s)))

(define (send-message-later address time . message)
  (thread-start!
   (lambda ()
     (thread-sleep! time)
     (apply send-message (cons address message)))))

(define (next-local-message)
  (let ((res #f))
    (mutex-lock! message-available-mutex #f #f)
    (mutex-lock! local-queue-mutex)
    (set! res (fifo-pop local-queue))
    (if (not (fifo-empty? local-queue))
        (mutex-unlock! message-available-mutex))
    (mutex-unlock! local-queue-mutex)
    res))

(define (start-scheduler)
  (let loop ()
    (apply deliver-message (next-local-message))
    (loop)))


;; Network

(define (start-network-listener)
  (thread-start!
   (lambda ()
     (let ((s (udp-open-socket*)))
       (udp-bind! s #f sam-port)
       (let loop ()
         (let-values (((n str) (udp-recv s 65536)))
           (match (with-input-from-string str read)
             ((address message ...)
              (apply send-message (cons address message)))
             (else
              (print "Warning: received badly formatted message string '" str "'"))))
         (loop))))))

;; System interface

(define reader-queue-mutex (make-mutex "reader queue"))
(define reader-available-mutex (make-mutex "reader available"))
(mutex-lock! reader-available-mutex #f #f)
(define reader-queue (make-fifo))

(define (next-reader)
  (let ((res #f))
    (mutex-lock! reader-available-mutex #f #f)
    (mutex-lock! reader-queue-mutex)
    (set! res (fifo-pop reader-queue))
    (if (not (fifo-empty? reader-queue))
        (mutex-unlock! reader-available-mutex))
    (mutex-unlock! reader-queue-mutex)
    res))

(define (start-console)
  (thread-start!
   (lambda ()
     (let loop ()
       (let ((reader (next-reader)))
         (##sys#thread-block-for-i/o! (current-thread) 0 #t)
         (thread-yield!)
         (send-message reader (read-line)))
       (loop)))))

;; System initialization

(define (system-beh self . message)
  (match message

    (('shutdown)
     (print "## System actor received shutdown message.")
     (exit 0)
     'done)

    (('print strings ...)
     (apply print strings)
     'sleep)

    (('read reader)
     (mutex-lock! reader-queue-mutex)
     (fifo-push reader-queue reader)
     (mutex-unlock! reader-available-mutex)
     (mutex-unlock! reader-queue-mutex)
     'sleep)))

(define (boot-sam)
  (start-console)
  (start-network-listener)
  (let ((system (make-actor system-beh))
        (main #f))
    (condition-case
        (begin
          (set! main (make-actor main-beh)))
      ((exn)
       (print "Error starting main actor. Is main-beh defined?")
       (exit 1)))
    (send-message main system))
  (start-scheduler))

(define (print-usage)
  (print "Simple Actor Machine v" sam-version "\n"
         "\n"
         "Usage: sam -h|--help\n"
         "       sam [-n hostname] [-p port] source-file-1 [source-file-2 [...]] "))


(let loop ((args (cdr (argv))))
  (match args
    (((or "-h" "--help"))
     (print-usage))
    (((or "-p" "--port") pstr rest ...)
     (set! sam-port (string->number pstr))
     (loop rest))
    (((or "-n" "--hostname") hstr rest ...)
     (set! sam-host hstr)
     (loop rest))
    (((? file-exists? filename) rest ...)
     (print* "Loading " filename "...")
     (load filename)
     (print " done.")
     (loop rest))
    (()
     (boot-sam))
    (else
     (print "Unrecognised argument '" (car args) "'.\n")
     (print-usage))))