X-Git-Url: https://thelambdalab.xyz/gitweb/index.cgi?a=blobdiff_plain;f=sam.scm;h=1d23712a9baf50af8e6502ddaaae708d3be6f53b;hb=5d2d076345efb45ced333e832d1cbf766bcd5e8a;hp=c819ca52a5d7d99868ddd7f1a1a66bb143dc2a01;hpb=369b3f9b8089c1a6045d1ae8b99295f97c7056dd;p=sam.git diff --git a/sam.scm b/sam.scm index c819ca5..1d23712 100644 --- a/sam.scm +++ b/sam.scm @@ -1,10 +1,17 @@ ;; Simple Actor Machine ;; -;; Houses a population of actors which can communicate using messages -;; with actors on the same machine or other machines via the network. +;; 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 (chicken io) +(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 @@ -13,45 +20,65 @@ udp fifo) +;; Global variables + +(define trace #f) + +(define sam-host "localhost") +(define sam-port 8000) + +(define sam-version "0.1") + ;; Actors -(define this-host "localhost") -(define this-port 1234) (define (make-address host port id) - (make-uri #:scheme "actor" - #:host host - #:port port - #:path (list '/ id))) - -(define next-local-address - (let ((mutex (make-mutex "actor address mutex"))) - (lambda () - (let ((res #f)) - (mutex-lock! mutex) - (set! res (make-address this-host this-port (uuid))) - (mutex-unlock! mutex) - res)))) - -(define (address-id address) (cadr (uri-path address))) + (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? (uri-host address) this-host) - (equal? (uri-port address) this-port))) + (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 (next-local-address)) + (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 " address) + (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) @@ -78,46 +105,61 @@ (define (send-network-message address . message) (let ((s (udp-open-socket)) - (machine (address-machine address))) + (packet (with-output-to-string + (lambda () + (write (cons address message)))))) (udp-bind! s #f 0) (udp-connect! s - (machine-host machine) - (machine-port machine)) - (udp-send s message) + (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! 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 loop () - (apply deliver-message (next-local-message)) - (loop))))) + (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)))))) -;; Console +;; 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 console - (make-actor (lambda (self . message) - (mutex-lock! reader-queue-mutex) - (fifo-push reader-queue (car message)) - (mutex-unlock! reader-available-mutex) - (mutex-unlock! reader-queue-mutex) - 'sleep))) - (define (next-reader) (let ((res #f)) (mutex-lock! reader-available-mutex #f #f) @@ -129,44 +171,75 @@ res)) (define (start-console) - (let loop () - (let ((reader (next-reader))) - (##sys#thread-block-for-i/o! (current-thread) 0 #t) - (thread-yield!) - (send-message reader (read-line))) - (loop))) - + (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 interface +;; 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)))) -(define system - (make-actor (lambda (self . message) - (match message - (('shutdown) - (print "## System actor received shutdown message.") - (exit 0) - 'done) - (('println strings ...) - (apply print strings) - 'sleep))))) - -;; Testing - - -(send-message system 'println "Hello, what is your name?") -(send-message console - (make-actor (lambda (self . message) - (match message - ((name) - (send-message system 'println "Hello, " name "!") - 'done))))) - -(thread-start! - (lambda () - (thread-sleep! 10) - (send-message system 'shutdown))) - -(start-scheduler) -(start-console) - -;; (thread-join! scheduler-thread)