Commit b12de5a5 authored by  Hayley Patton's avatar Hayley Patton 🐢

make parser generic, redo event handler, more stuff

parent f5ad2c65
Pipeline #684 passed with stage
in 2 minutes and 18 seconds
......@@ -9,6 +9,7 @@
:components ((:module src
:components ((:file "package")
(:file "macros")
(:file "protocol")
(:file "network")
(:file "client")
(:file "channels")
......
......@@ -43,11 +43,6 @@
(let ((split (split-sequence #\Space string)))
(unless (= (length split) 0)
split)))
(defun data-to-string (data)
(format nil "~{~a~^~%~}" data))
(defun string-to-data (string)
(split-sequence #\Newline string))
(defun create-item (name data version channels)
(make-dao 'decentralise-row
:name name
......@@ -63,8 +58,7 @@
(defmethod put-item ((system psql-system) name data version channels)
(declare (ignore system))
(let ((channels (channel-list-to-string channels))
(data (data-to-string data)))
(let ((channels (channel-list-to-string channels)))
(handler-case
(get-dao-row name)
(block-not-found ()
......@@ -75,7 +69,7 @@
(defmethod get-item ((system psql-system) name)
(declare (ignore system))
(let ((dao (get-dao-row name)))
(values (string-to-data (row-data dao))
(values (row-data dao)
(row-version dao)
(channel-string-to-list (row-channels dao)))))
......
......@@ -3,48 +3,73 @@
(in-package :decentralise)
(defun write-block (connection name values &key
(version 0)
(channels nil))
(defmethod write-block ((connection connection) name values &key
(version 0)
(channels nil))
"Send a new block."
(format (connection-conn connection)
"BLOCK ~a ~d~{ ~a~}~%~{~a~%~}~a~%"
"BLOCK ~a ~d~{ ~a~}~%"
name
version
channels
values
+eof-marker+)
channels)
(etypecase values
(string (write-string values
(connection-conn connection))
(terpri (connection-conn connection)))
(list
(dolist (line values)
(write-line line (connection-conn connection)))))
(write-line +eof-marker+ (connection-conn connection))
(finish-output (connection-conn connection)))
(defun get-block (connection name)
(format (connection-conn connection)
"GET ~a~%"
name)
(finish-output (connection-conn connection)))
(defun get-blocks (connection names)
(defmethod get-blocks ((connection connection) names)
(format (connection-conn connection)
"GET ~{~a~^ ~}~%"
names)
(finish-output (connection-conn connection)))
(defun write-error (connection name &optional (reason ""))
(defun get-block (connection name)
(get-blocks connection (list name)))
(defmethod write-error ((connection connection) name &optional (reason ""))
(format (connection-conn connection)
"ERROR ~a ~a~%" name reason)
(finish-output (connection-conn connection)))
(defun write-ok (connection name)
(defmethod write-ok ((connection connection) name)
(format (connection-conn connection)
"OK ~a~%" name)
(finish-output (connection-conn connection)))
(defun subscribe-channels (connection channels)
(defmethod subscribe-channels ((connection connection) channels)
(format (connection-conn connection)
"SUBSCRIBE ~{~a~^ ~}~%" channels)
(finish-output (connection-conn connection)))
(defun announce (connection &optional (announcep t))
(defmethod announce ((connection connection) &optional (announcep t))
(format (connection-conn connection)
"ANNOUNCE ~a~%"
(if announcep "yes" "no"))
(finish-output (connection-conn connection)))
(defmethod read-event ((connection connection))
(let ((status (read-status-line connection)))
(macrolet ((possible-event-types (&body types)
`(cond
,@(loop for type in types collect
(destructuring-bind (name binds &body body) type
`((string= (first status) ,name)
(destructuring-bind ,binds (rest status) ,@body)))))))
(possible-event-types
("get" (&rest names) `(:get ,names))
("block" (name version-string &rest channels)
`(:block ,name ,(parse-integer version-string)
,channels
,(read-rest-of-block connection)))
("ok" (name) `(:ok ,name))
("error" (name &rest error-bits)
`(:error ,name ,(apply #'concatenate 'string error-bits)))
("subscribe" (&rest channels)
`(:subscribe ,channels))
("announce" (string)
`(:announce ,(string= string "yes")))))))
(in-package :decentralise)
(defun make-event-handler (line data event)
(ematch event
((list* :get-blocks (list block-list) body)
`(when (string= (first ,line) "get")
(let ((,block-list (rest ,line)))
,@body)))
((list* :new-block (list new-name new-version new-channels new-data) body)
(let ((new-version-str (gensym "version-str"))) ; required to hold unparsed integer
`(when (string= (first ,line) "block")
(destructuring-bind (,new-name ,new-version-str &rest ,new-channels)
(rest ,line)
(let ((,new-version (parse-integer ,new-version-str))
(,new-data ,data))
,@body)))))
((list* (list :new-block channels) (list new-name new-version new-channels new-data) body)
(let ((new-version-str (gensym "version-str")) ; required to hold unparsed integer
(channel-handler `(should-send ,channels ,new-channels)))
`(when (string= (first ,line) "block")
(destructuring-bind (,new-name ,new-version-str &rest ,new-channels)
(rest ,line)
(when ,channel-handler
(let ((,new-version (parse-integer ,new-version-str))
(,new-data ,data))
,@body))))))
((list* :error (list name problem) body)
`(when (string= (first ,line) "error")
(destructuring-bind (,name ,problem)
(rest ,line)
,@body)))
((list* :ok (list name) body)
`(when (string= (first ,line) "ok")
(let ((,name (second ,line)))
,@body)))
((list* :subscribe (list channels) body)
`(when (string= (first ,line) "subscribe")
(let ((,channels (rest ,line)))
,@body)))
((list* :announce (list announcep) body)
`(when (string= (first ,line) "announce")
(let ((,announcep (string= (second ,line) "yes")))
,@body)))
((list* :bad-syntax (list tokens) body)
`(unless (member (first ,line) '("get" "block" "error" "ok" "subscribe" "announce")
:test #'string=)
(let ((,tokens ,line))
,@body)))))
(eval-when (:compile-toplevel :load-toplevel :execute)
(defun make-event-handler (event)
(ematch event
((list* :get-blocks (list names) body)
`((list :get ,names) ,@body))
((list* :new-block (list name version channels data) body)
`((list :block ,name ,version ,channels ,data) ,@body))
((list* (list :new-block channel-filter) (list name version channels data) body)
`((list :block ,name ,version
(and ,channels
(satisfies (lambda (c) (should-send ,channel-filter c))))
,data)
,@body))
((list* :error (list name problem) body)
`((list :error ,name ,problem) ,@body))
((list* :ok (list name) body)
`((list :ok ,name) ,@body))
((list* :subscribe (list channels) body)
`((list :subscribe ,channels) ,@body))
((list* :announce (list announcep) body)
`((list :announce ,announcep) ,@body))
((list* :bad-syntax (list) body)
`(nil ,@body)))))
(defmacro with-event-handlers (connection &body events)
(let ((forced-stream (gensym "FORCED-STREAM"))
(line (gensym "LINE"))
(block-data (gensym "DATA")))
`(let ((,forced-stream ,connection))
(let* ((,line (read-status-line ,forced-stream))
(,block-data (when (string= (first ,line) "block")
(read-rest-of-block ,forced-stream))))
(progn
,@(loop for event in events
collect (make-event-handler line block-data event)))))))
`(match (read-event ,connection)
,@(loop for event in events
collect (make-event-handler event))))
;;;; network.lisp: metadata for generic connections and streams.
;;;; network.lisp: helper code for reading and writing blocks and text
;;; This file is a part of cl-decentralised.
(in-package :decentralise)
(defvar +eof-marker+ "--EOF--")
(defstruct connection
(ip "")
(port 1892 :type fixnum)
(conn nil))
(defun split-by-spaces (line)
(split-sequence #\Space line))
(defun connect-to-ip (ip &optional (port 1892))
(let ((connection (make-connection :ip ip :port port)))
(let ((connection (make-instance 'connection :ip ip :port port)))
(setf (connection-conn connection)
(usocket:socket-stream
(usocket:socket-connect ip port)))
......@@ -32,17 +27,21 @@
(defun read-block (connection)
(let ((status-line (read-status-line connection)))
(if (string/= (first status-line) "BLOCK")
(if (string/= (first status-line) "block")
;; Errors don't have data.
(error (format nil "~{~a~^ ~}" status-line))
(values (read-rest-of-block connection)
(second status-line)))))
(defun read-rest-of-block (connection)
(loop for line = (read-conn-line connection)
until (string= line "--EOF--")
collect line into lines
finally (return lines)))
(with-output-to-string (output-stream)
(loop
for line = (read-conn-line connection)
for first = nil then t
until (string= line "--EOF--")
when first
do (terpri output-stream)
do (write-string line output-stream))))
(defun write-text (connection text)
(princ text (connection-conn connection))
......
......@@ -11,6 +11,11 @@
:accept
:connect-to-ip
;; network
:connection
:connection-ip
:connection-port
:connection-conn
;; client
:write-block
:get-block
:get-blocks
......
;;;; protocol.lisp: structures & gfs cl-decentralise uses. not to be confused with network or client
;;; This file is a part of cl-decentralise.
(in-package :decentralise)
(defclass system ()
((connection-list :initform nil)
(connection :initform nil)
(port :initarg :port
:initform 1892
:reader system-port)
(sync-limit :initarg :sync-limit
:initform 70
:accessor system-sync-limit))
(:documentation "A decentralise system, a representation of (part of) a
distributed program and its networking.
:PORT can be used to change the port from the default of 1892, and
:SYNC-LIMIT can be used to change the amount of nodes cl-decentralise will
track and connect to."))
(defclass connection ()
((ip :initform "" :initarg :ip :reader connection-ip)
(port :initform 1892 :initarg :port :reader connection-port)
(conn :initform nil :initarg :connection :accessor connection-conn))
(:documentation "A representation of a connection to a decentralise instance.
This representation assumes that the client has an IP address and port; but
these may usually remain at their defaults or possibly unbound with no detriment
to the core of cl-decentralise.
:IP sets the IP address of the client, and similarly
:PORT sets the port.
:CONN is the underlying stream the connection relies on."))
(defstruct connection-meta
"The internal state of a connection, including the connection itself,
what blocks should be sent in the future (STACK), what channels the connection
subscribed to (CHANNELS), a lock to prevent threading-related issues (LOCK),
and a flag set if the node should be published in the NODES block (ANNOUNCEP)."
socket
(stack nil)
(channels nil)
(lock (bt:make-lock "connection-meta lock"))
(announcep nil :type boolean))
(defgeneric listing-generator (system))
(defgeneric get-item (system name))
(defgeneric put-item (system name text version channels))
(defgeneric stop-put-p (system name text version channels))
(defmethod stop-put-p ((system system) name text version channels)
(declare (ignore system name text version channels))
nil)
(defvar *ip-blacklist* '("localhost" "127.0.0.1" "::1"))
;;; Synchronisation functions.
(defgeneric bind (system &key host))
(defgeneric write-object (system stream name))
(defgeneric handle-others-listing (system data meta))
(defgeneric handle-others-nodes (system text))
(defgeneric handle-new-block (name version text system connection meta channels))
(defgeneric handle-lines-from (connection system meta))
(defgeneric repl (connection system))
(defgeneric accept (system &key log))
(defgeneric connect-to (system host &key port timeout announcep))
;;; Network functions.
(defgeneric get-blocks (connection names))
(defgeneric write-block (connection name values &key version channels))
(defgeneric write-error (connection name &optional reason))
(defgeneric write-ok (connection name))
(defgeneric subscribe-channels (connection channels))
(defgeneric announce (connection &optional announcep))
(defgeneric read-event (connection))
......@@ -3,25 +3,6 @@
(in-package :decentralise)
(defclass system ()
((connection-list :initform nil)
(connection :initform nil)
(port :initarg :port
:initform 1892
:reader system-port)
(sync-limit :initarg :sync-limit
:initform 70
:accessor system-sync-limit)))
(defgeneric listing-generator (system))
(defgeneric get-item (system name))
(defgeneric put-item (system name text version channels))
(defgeneric stop-put-p (system name text version channels))
(defmethod stop-put-p ((system system) name text version channels)
(declare (ignore system name text version channels))
nil)
(defvar *ip-blacklist* '("localhost" "127.0.0.1" "::1"))
(defmethod print-object ((system system) out)
(print-unreadable-object (system out :type t)
(with-slots (connection-list connection port) system
......@@ -29,13 +10,6 @@
~t:Connection ~a
~t:Port ~a" connection-list connection port))))
(defstruct connection-meta
socket
(stack nil)
(channels nil)
(announcep nil :type boolean))
(defgeneric bind (system &key host))
(defmethod bind ((system system) &key (host "0.0.0.0"))
(setf (slot-value system 'connection)
(socket-listen host
......@@ -52,7 +26,6 @@
(return-from list-collecting-loop table))
(setf (gethash name table) version))))
(defgeneric write-object (system stream name))
(defmethod write-object ((system system) stream name)
(cond ((string= name "nodes")
(write-block stream
......@@ -77,7 +50,7 @@
(multiple-value-setq (data version channels)
(get-item system name))
(condition (c)
(write-error stream
(write-error stream name
(princ-to-string c))
(return-from write-object)))
(write-block stream
......@@ -86,15 +59,16 @@
:version version
:channels channels)))))
(defgeneric handle-others-listing (system data meta lock))
(defmethod handle-others-listing ((system system) data meta lock)
(let* ((connection (connection-meta-socket system))
(defmethod handle-others-listing ((system system) data meta)
(let* ((lock (connection-meta-lock meta))
(connection (connection-meta-socket system))
(our-data (get-every-list (listing-generator system)))
(their-data
(loop
with table = (make-hash-table :test 'equal
:size (hash-table-size our-data))
for item in data
for item in (split-sequence #\Newline data)
do (destructuring-bind (name version)
(split-sequence #\Space item)
(setf (gethash name table)
......@@ -115,11 +89,9 @@
(> their-version our-version))
collect name)))
(push-all give (connection-meta-stack meta))
(bt:with-lock-held (lock)
(write-text connection
(format nil "GET ~{~a~^ ~}" take)))))
(bt:with-lock-held ((connection-meta-lock lock))
(get-blocks connection take))))
(defgeneric handle-others-nodes (system text))
(defmethod handle-others-nodes (system text)
(let* ((our-nodes
(remove-duplicates
......@@ -144,34 +116,36 @@
:announcep t)))
:name "Discovered node"))))
(defgeneric handle-new-block (name version text system connection meta channels))
(defmethod handle-new-block (name version text (system system) connection meta channels)
(let ((err (stop-put-p system name text version channels)))
(cond
(err
(write-error connection name err))
(t
(let ((our-version
(ignore-errors
(nth-value 1 (get-item system name)))))
(cond
((or
(null our-version)
(> version our-version))
;; if we have a new version: add it
(put-item system name text version channels)
(write-ok connection name)
;; tell the world about it
(send-to-all-interested system meta name))
(t
;; tell them they're wrong.
(write-error connection name "too old")
(when (should-send (connection-meta-channels meta)
channels)
(push name (connection-meta-stack meta))))))))))
(cond
((string= name "list")
(handle-others-listing system text meta))
(t
(let ((err (stop-put-p system name text version channels)))
(cond
(err
(write-error connection name err))
(t
(let ((our-version
(ignore-errors
(nth-value 1 (get-item system name)))))
(cond
((or
(null our-version)
(> version our-version))
;; if we have a new version: add it
(put-item system name text version channels)
(write-ok connection name)
;; tell the world about it
(send-to-all-interested system meta name))
(t
;; tell them they're wrong.
(write-error connection name "too old")
(when (should-send (connection-meta-channels meta)
channels)
(push name (connection-meta-stack meta))))))))))))
(defgeneric handle-lines-from (connection system meta write-lock))
(defmethod handle-lines-from (connection (system system) meta write-lock)
(defmethod handle-lines-from (connection (system system) meta)
(with-event-handlers connection
(:new-block (name version-number channels data)
;; BLOCK (NAME) (VERSION) (CHANNELS*)
......@@ -186,33 +160,34 @@
(:subscribe (subscriptions)
;; SUBSCRIBE (CHANNELS*)
(setf (connection-meta-channels meta) subscriptions))
(:bad-syntax (tokens)
(with-lock-held (write-lock)
(write-text connection (format nil "bad syntax: ~{~a~^ ~}" tokens))))))
(:bad-syntax ()
(with-lock-held ((connection-meta-lock meta))
(write-error connection "bad-syntax")))))
(defgeneric repl (connection system))
(defmethod repl (connection (system system))
(let ((meta (make-connection-meta :socket connection
:stack nil))
(lock (bt:make-lock "Write lock")))
(let* ((lock (bt:make-lock "Write lock"))
(meta (make-connection-meta :socket connection
:stack nil
:lock lock)))
(push meta
(slot-value system 'connection-list))
(handler-case
(progn
(with-thread (:name "Connection listener")
(loop
(loop
for val = (pop
(connection-meta-stack meta))
while val
do (with-lock-held (lock)
(write-object system connection val)))
(unless (null (connection-meta-stack meta))
(with-lock-held (lock)
(loop
for val = (pop
(connection-meta-stack meta))
while val
do (write-object system connection val))))
(sleep 0.1)))
(loop
(handle-lines-from connection system meta lock)))
(handle-lines-from connection system meta)))
(end-of-file ()
(format t "Byebye, ~a.~%" connection))
#-debug
#-debug
(condition (c)
(format t "Killing ~a because ~a~%" connection c)
(close (connection-conn connection))
......@@ -221,18 +196,18 @@
(slot-value system 'connection-list)
:test #'eq))))))
(defgeneric accept (system &key log))
(defmethod accept ((system system) &key log)
(defmethod accept ((system system) &key log (connection-type 'connection))
(make-thread
(lambda ()
(loop
for connection = (socket-accept (slot-value system 'connection))
do (make-thread
(lambda ()
(repl (make-connection :ip (stringify-address
(get-peer-address connection))
:port (get-peer-port connection)
:conn (socket-stream connection))
(repl (make-instance connection-type
:ip (stringify-address
(get-peer-address connection))
:port (get-peer-port connection)
:connection (socket-stream connection))
system)
(socket-close connection)))
when log
......@@ -240,17 +215,17 @@
(get-peer-address connection)
(get-peer-port connection))))))
(defgeneric connect-to (system host &key port timeout announcep))
(defmethod connect-to ((system system) host &key (port 1892) (timeout 5) (announcep t))
(let* ((socket (socket-connect host port :timeout timeout))
(raw-conn (socket-stream socket))
(conn (make-connection :ip host
:port port
:conn raw-conn)))
(conn (make-instance 'connection
:ip host
:port port
:connection raw-conn)))
;; ask for a file listing and all nodes
(format raw-conn "list nodes~%")
(get-blocks conn '("list" "nodes"))
(when announcep
(format raw-conn "ANNOUNCE~%"))
(announce conn t))
(finish-output raw-conn)
(unwind-protect
(repl conn system)
......
......@@ -12,25 +12,24 @@
;; get the string from OUTPUT-STREAM
(defmacro test-output (system input &optional expected-output)
(alexandria:with-gensyms (input-stream output-stream two-way-stream
connection-meta connection write-lock)
connection-meta connection)
`(ignore-errors
(let* (;; Streams
(,input-stream (make-string-input-stream ,input))
(,output-stream (make-string-output-stream))
(,two-way-stream (make-two-way-stream ,input-stream ,output-stream))
;; Internal decentralise values.
(,write-lock (bt:make-lock "This lock does nothing!"))
(,connection (decentralise::make-connection :ip "This doesn't matter!"
:port 12345
:conn ,two-way-stream))
(,connection (make-instance 'decentralise::connection
:ip "This doesn't matter!"
:port 12345
:connection ,two-way-stream))
(,connection-meta (decentralise::make-connection-meta)))
;; Loop until there's nothing to read.
(loop
while (peek-char nil ,two-way-stream nil nil)
do (decentralise::handle-lines-from ,connection
,system
,connection-meta
,write-lock))
,connection-meta))
;; Write every object out.
(loop
for value = (pop (decentralise::connection-meta-stack ,connection-meta))
......
......@@ -32,7 +32,7 @@
(defmethod stop-put-p ((system my-system) name text version channels)
(declare (ignore version channels name))
(when (some (lambda (line) (search "boop" line)) text)
(when (search "boop" text)
"No boops allowed!"))
(defvar *system* (make-instance 'my-system :port 12345))
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment