serialize.lisp

(in-package :graph-db)

;;(declaim (optimize (speed 3)))

(defgeneric serialize (object))
(defgeneric deserialize-help (become object))
(defgeneric deserialize-help-mmap (become object length header-length))
(defgeneric make-serialized-key (object))
(defgeneric deserialize (object))
(defgeneric serialized-equal (x y))

(let ((length-table #+sbcl (make-hash-table :synchronized t)
                    #+lispworks (make-hash-table :single-thread nil)
                    #+ccl (make-hash-table :shared t)))
  (defun encode-length (int)
    (declare (type integer int))
    (or (gethash int length-table)
	(let* ((n-bytes (ceiling (integer-length int) 8))
	       (original-int int)
	       (vec (make-array (+ 1 n-bytes) :element-type '(unsigned-byte 8))))
	  (setf (aref vec 0) n-bytes)
	  (dotimes (i n-bytes)
	    (setf (aref vec (+ 1 i)) (ldb (byte 8 (* i 8)) int)))
	  (setf (gethash original-int length-table) vec)))))

(defun decode-length (bytes)
  (declare (type (array (unsigned-byte 8)) bytes))
  (let ((int 0) (n-bytes (length bytes)))
    (dotimes (i n-bytes)
      (setq int (dpb (aref bytes i) (byte 8 (* i 8)) int)))
    int))

(defun extract-length (a)
  (declare (type (array (unsigned-byte 8)) a))
  (let ((id-byte (aref a 0)))
    (cond ((or (= id-byte +uuid+) ;; These are all fixed length
               (= id-byte +bit-vector+)
	       (= id-byte +positive-integer+)
	       (= id-byte +negative-integer+)
	       (= id-byte +character+)
	       (= id-byte +single-float+)
	       (= id-byte +double-float+))
           (values (aref a 1) 2))
          ((= id-byte +timestamp+)
           (values 24 1))
          ((= id-byte +mpointer+)
           (values 8 1))
          ((or (= id-byte +vertex+)
               (= id-byte +edge+))
           (values -1 1))
	  ((or (= id-byte +t+) (= id-byte +null+))
	   (values 1 0))
	  (t ;; strings, lists, vectors, blobs, nodes, triples have variable bytes
	   (let ((header-length (+ 2 (aref a 1))))
	     (values (decode-length (subseq a 2 header-length)) header-length))))))

(defun decode-length-mmap (mf n-bytes offset)
  (let ((int 0))
    (dotimes (i n-bytes)
      (let ((byte (get-byte mf offset)))
        (setq int (dpb byte (byte 8 (* i 8)) int))
        (incf offset)))
    (values int offset)))

(defun extract-length-mmap (p &optional id-byte)
  (let ((mf (mpointer-mmap p)) (offset (mpointer-loc p)))
    (let ((id-byte (or id-byte (get-byte mf offset))))
      (cond ((or (= id-byte +uuid+) ;; These are all fixed length
                 (= id-byte +positive-integer+)
                 (= id-byte +negative-integer+)
                 (= id-byte +character+)
                 (= id-byte +single-float+)
                 (= id-byte +double-float+))
             (values (get-byte mf (incf offset)) 2 offset))
            ((= id-byte +timestamp+)
             (values 24 1))
            ((or (= id-byte +t+) (= id-byte +null+))
             (values 1 0 offset))
            ((or (= id-byte +vertex+)
                 (= id-byte +edge+))
             (values -1 1))
            (t
             ;; strings, lists, vectors, blobs, nodes, triples have variable bytes
             (let ((l-of-l (get-byte mf (incf offset))))
               (multiple-value-bind (length new-offset)
                   (decode-length-mmap mf l-of-l (incf offset))
                 (values length (+ 2 l-of-l) new-offset))))))))

(defun extract-all-subseqs (a)
  (declare (type (array (unsigned-byte 8)) a))
  (cond
    ((= 0 (length a))
     nil)
    (t
     (multiple-value-bind (data-length header-length) (extract-length a)
       (cons (subseq a 0 (+ header-length data-length))
             (extract-all-subseqs (subseq a (+ header-length data-length))))))))

(defun read-bytes (mpointer)
  (multiple-value-bind (len h-len) (extract-length-mmap mpointer)
    (get-bytes (mpointer-mmap mpointer) (mpointer-loc mpointer) (+ h-len len))))

(defmethod deserialize ((a array))
  (declare (type (array (unsigned-byte 8)) a))
  (multiple-value-bind (data-length header-length) (extract-length a)
    (values
     (deserialize-help (aref a 0)
                       (if (> data-length 0)
                           (subseq a header-length (+ header-length data-length))
                           (subseq a header-length)))
     (+ header-length data-length))))

(defmethod deserialize ((p mpointer))
  (let ((type-byte (get-byte (mpointer-mmap p) (mpointer-loc p))))
    (multiple-value-bind (data-length header-length)
        (extract-length-mmap p type-byte)
      (incf (mpointer-loc p) header-length)
      (values (deserialize-help-mmap type-byte p data-length header-length)
              (+ data-length (mpointer-loc p))))))

(defmethod serialized-equal ((x array) (y array))
  (declare (type (array (unsigned-byte 8)) x y))
  (let ((length (length x)))
    (and (= length (length y))
         (dotimes (i length t)
           (let ((x-el (aref x i))
                 (y-el (aref y i)))
             (unless (= x-el y-el)
               (return nil)))))))

(defmethod serialized-equal ((x mpointer) (y mpointer))
  (let ((type-x (get-byte (mpointer-mmap x) (mpointer-loc x)))
        (type-y (get-byte (mpointer-mmap y) (mpointer-loc y))))
    (and (= type-x type-y)
         (multiple-value-bind (data-length-x header-length-x)
             (extract-length-mmap y)
           (multiple-value-bind (data-length-y header-length-y)
               (extract-length-mmap y)
             (= data-length-x data-length-y)
             (= header-length-x header-length-y)
             (dotimes (i data-length-x t)
               (let ((x-el (get-byte (mpointer-mmap x)
                                     (+ i header-length-x (mpointer-loc x))))
                     (y-el (get-byte (mpointer-mmap y)
                                     (+ i header-length-y (mpointer-loc y)))))
                 (unless (= x-el y-el)
                   (return nil)))))))))

(defmethod deserialize :around (object)
  (handler-case
      (call-next-method)
    (error (condition)
      (error 'deserialization-error :instance object :reason condition))))

(defmethod serialize :around (object)
  (handler-case
      (call-next-method)
    (error (condition)
      (error 'serialization-error :instance object :reason condition))))

(defun serialize-multiple (type-specifier &rest slots)
  (declare (type integer type-specifier))
  (let* ((serialized-slots (mapcar #'serialize slots))
	 (serialized-slot-lengths (mapcar #'length serialized-slots))
	 (total-length (apply #'+ serialized-slot-lengths))
	 (encoded-length (encode-length total-length))
	 (length-of-encoded-length (length encoded-length)))
    (declare (type fixnum total-length length-of-encoded-length))
    (declare (type (array (unsigned-byte 8)) encoded-length))
    (let ((a (make-array (+ 1 length-of-encoded-length total-length)
			 :element-type '(unsigned-byte 8))))
      (setf (aref a 0) type-specifier)
      (dotimes (i length-of-encoded-length)
	(setf (aref a (1+ i)) (aref encoded-length i)))
      (dotimes (i (length serialized-slots))
	(dotimes (j (nth i serialized-slot-lengths))
	  (setf (aref a (+ 1
			   length-of-encoded-length
			   j
			   (apply #'+ (subseq serialized-slot-lengths 0 i))))
		(aref (nth i serialized-slots) j))))
      a)))

(defmethod make-slot-key (id slot-name)
  (if (symbolp slot-name) (setq slot-name (symbol-name slot-name)))
  (let* ((serialized-id (serialize id))
	 (serialized-slot-name (serialize slot-name))
	 (total-length (+ (length serialized-id) (length serialized-slot-name))))
    (declare (type fixnum total-length))
    (let ((a (make-array (+ 1 total-length) :element-type '(unsigned-byte 8))))
      (setf (aref a 0) +slot-key+)
      (dotimes (i (length serialized-id))
	(setf (aref a (1+ i)) (aref serialized-id i)))
      (dotimes (i (length serialized-slot-name))
	(setf (aref a (+ 1 (length serialized-id) i))
              (aref serialized-slot-name i)))
      a)))

(defmethod serialize ((ts timestamp))
  (let ((v (make-byte-vector 25)) (offset 0))
    (setf (aref v 0) +timestamp+)
    (dotimes (i 8)
      (setf (aref v (incf offset)) (ldb (byte 8 (* i 8)) (day-of ts))))
    (dotimes (i 8)
      (setf (aref v (incf offset)) (ldb (byte 8 (* i 8)) (sec-of ts))))
    (dotimes (i 8)
      (setf (aref v (incf offset)) (ldb (byte 8 (* i 8)) (nsec-of ts))))
    v))

(defmethod deserialize-help ((become (eql +timestamp+)) (bytes array))
  (make-timestamp :day (deserialize-uint64 bytes 0)
                  :sec (deserialize-uint64 bytes 8)
                  :nsec (deserialize-uint64 bytes 16)))

(defmethod deserialize-help ((become (eql +vector+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type integer become))
  (map 'vector #'deserialize (extract-all-subseqs bytes)))

(defmethod serialize ((v vector))
  (if (equal (array-element-type v) '(unsigned-byte 8))
      v
      (let* ((serialized-items (map 'list #'serialize v))
	     (total-length (reduce #'+ serialized-items :key 'length))
	     (encoded-length (encode-length total-length))
	     (length-of-encoded-length (length encoded-length))
	     (vec (make-array 0 :fill-pointer t :adjustable t
                              :element-type '(unsigned-byte 8))))
	(declare (type fixnum total-length length-of-encoded-length))
	(declare (type (array (unsigned-byte 8)) encoded-length))
	(vector-push-extend +vector+ vec)
	(dotimes (i length-of-encoded-length)
	  (vector-push-extend (aref encoded-length i) vec))
	(dolist (item serialized-items)
	  (dotimes (i (length item))
	    (vector-push-extend (aref item i) vec)))
	vec)))

(defun bit-vector->integer (bit-vector)
  "Create a positive integer from a bit-vector."
  (reduce #'(lambda (first-bit second-bit)
              (+ (* first-bit 2) second-bit))
          bit-vector))

(defun integer->bit-vector (integer)
  "Create a bit-vector from a positive integer."
  (labels ((integer->bit-list (int &optional accum)
             (cond ((> int 0)
                    (multiple-value-bind (i r) (truncate int 2)
                      (integer->bit-list i (push r accum))))
                   ((null accum) (push 0 accum))
                   (t accum))))
    (coerce (integer->bit-list integer) 'bit-vector)))

;; FIXME: write methods for bignum serialization, and use them here for larger
;; bit vectors
(defmethod serialize ((v bit-vector))
  (let ((vec (serialize (bit-vector->integer v))))
    (setf (aref vec 0) +bit-vector+)
    vec))

(defmethod deserialize-help ((become (eql +bit-vector+)) (bytes array))
  (integer->bit-vector (deserialize-help +positive-integer+ bytes)))

(defmethod deserialize-help ((become (eql +uuid+)) (bytes array))
  "Decode a UUID."
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type integer become))
  (uuid:byte-array-to-uuid bytes))

(defmethod serialize ((uuid uuid:uuid))
  "Encode a UUID."
  (uuid:uuid-to-byte-array uuid +uuid+))

(defmethod deserialize-help ((become (eql +positive-integer+)) (bytes array))
  "Decode a positive integer."
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type integer become))
  (let ((int 0) (n-bytes (length bytes)))
    (dotimes (i n-bytes)
      (setq int (dpb (elt bytes i) (byte 8 (* i 8)) int)))
    int))

(defmethod deserialize-help-mmap ((become (eql +positive-integer+)) (p mpointer)
                                  n-bytes header-length)
  "Decode a positive integer."
  (declare (ignore header-length))
  (declare (type integer become))
  (let ((int 0))
    (dotimes (i n-bytes)
      (setq int (dpb (get-byte (mpointer-mmap p) (+ i (mpointer-loc p)))
                     (byte 8 (* i 8)) int)))
    int))

(defmethod deserialize-help ((become (eql +negative-integer+)) (bytes array))
  "Decode a negative integer."
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type integer become))
  (- (deserialize-help +positive-integer+ bytes)))

(defmethod serialize ((int integer))
  "Encodes integers between (- (1- (expt 2 (* 8 255)))) and
 (1- (expt 2 (* 8 255)))"
  (if (= 0 int)
      (make-array 3
                  :element-type '(unsigned-byte 8)
                  :initial-contents `(,+positive-integer+ 1 0))
      (let* ((n-bytes (ceiling (integer-length int) 8))
             (vec (make-array (+ 2 n-bytes) :element-type '(unsigned-byte 8))))
        (if (minusp int)
            (progn
              (setf (aref vec 0) +negative-integer+)
              (setq int (abs int)))
            (setf (aref vec 0) +positive-integer+))
        (setf (aref vec 1) n-bytes)
        (dotimes (i n-bytes)
          (setf (aref vec (+ 2 i)) (ldb (byte 8 (* i 8)) int)))
        vec)))

(defmethod deserialize-help ((become (eql +single-float+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type integer become))
  (ieee-floats:decode-float32 (deserialize-help +positive-integer+ bytes)))

(defmethod serialize ((float single-float))
  (let ((vec (serialize (ieee-floats:encode-float32 float))))
    (setf (aref vec 0) +single-float+)
    vec))

(defmethod deserialize-help ((become (eql +double-float+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (ieee-floats:decode-float64 (deserialize-help +positive-integer+ bytes)))

(defmethod serialize ((float double-float))
  (let ((vec (serialize (ieee-floats:encode-float64 float))))
    (setf (aref vec 0) +double-float+)
    vec))

(defmethod deserialize-help ((become (eql +character+)) (bytes array))
  "Decode a Unicode-encoded byte sequence."
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (let ((int 0) (n-bytes (length bytes)))
    (dotimes (i n-bytes)
      (setq int (dpb (elt bytes i) (byte 8 (* i 8)) int)))
    (code-char int)))

(defmethod serialize ((char character))
  "Encode a Unicode character."
  (let* ((code (char-code char))
	 (total-bytes (ceiling (integer-length code) 8))
	 (vec (make-array (+ 2 total-bytes) :element-type '(unsigned-byte 8))))
    (setf (aref vec 0) +character+)
    (setf (aref vec 1) total-bytes)
    (dotimes (i total-bytes)
      (setf (aref vec (+ 2 i)) (ldb (byte 8 (* i 8)) code)))
    vec))

(defmethod deserialize-help ((become (eql +string+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (babel:octets-to-string bytes))

(defmethod serialize ((string string))
  "Unicode aware string encoding. Not as efficient as it could be: creates 2
arrays: one to get lisp's internal byte representation of the string, and then
another for prepending our code and the length of the object."
  (let* ((unicode (babel:string-to-octets string))
	 (vector-length (length unicode))
	 (encoded-length (encode-length vector-length))
	 (length-of-encoded-length (length encoded-length))
	 (vec (make-array (+ 1 length-of-encoded-length vector-length)
			  :element-type '(unsigned-byte 8))))
    (setf (aref vec 0) +string+)
    (dotimes (i length-of-encoded-length)
      (setf (aref vec (1+ i)) (aref encoded-length i)))
    (dotimes (i vector-length)
      (setf (aref vec (+ 1 length-of-encoded-length i)) (aref unicode i)))
    vec))

(defmethod deserialize-help ((become (eql +t+)) (bytes array))
  t)

(defmethod deserialize-help ((become (eql +null+)) (bytes array))
  nil)

(defmethod deserialize-help ((become (eql +symbol+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (destructuring-bind (symbol package) (extract-all-subseqs bytes)
    (intern (deserialize symbol) (find-package (deserialize package)))))

(defmethod deserialize-help ((become (eql +keyword+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (intern (deserialize-help +string+ bytes) :keyword))

(defmethod serialize ((symbol symbol))
  (or (and (null symbol) #(#.+null+))
      (and (eq symbol t) #(#.+t+))
      (let ((symbol-name (serialize (symbol-name symbol))))
        (if (keywordp symbol)
            (progn
              (setf (aref symbol-name 0) +keyword+)
              symbol-name)
            (let* ((package-name (serialize (package-name
                                             (symbol-package symbol))))
                   (size (+ (length symbol-name) (length package-name)))
                   (encoded-length (encode-length size))
                   (vec (make-byte-vector (+ 1 (length encoded-length) size))))
              (setf (aref vec 0) +symbol+)
              (dotimes (i (length encoded-length))
                (setf (aref vec (1+ i)) (aref encoded-length i)))
              (dotimes (i (length symbol-name))
                (setf (aref vec (+ 1 (length encoded-length) i))
                      (aref symbol-name i)))
              (dotimes (i (length package-name))
                (setf (aref vec (+ 1 (length encoded-length) (length symbol-name)
                                   i))
                      (aref package-name i)))
              vec)))))

(defmethod deserialize-help ((become (eql +list+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (mapcar #'deserialize (extract-all-subseqs bytes)))

(defmethod deserialize-help ((become (eql +dotted-list+)) (bytes array))
  (declare (type (array (unsigned-byte 8)) bytes))
  (declare (type (integer 0 255) become))
  (let* ((items (extract-all-subseqs bytes))
	 (result nil))
    (loop for i downfrom (- (length items) 2) to 0 do
      (push (deserialize (nth i items)) result))
    (nconc result (deserialize (car (last items))))))

(defmethod serialize ((list list))
  (if (proper-listp list)
      (let* ((serialized-items (mapcar #'serialize list))
	     (total-length (reduce #'+ serialized-items :key 'length))
	     (encoded-length (encode-length total-length))
	     (length-of-encoded-length (length encoded-length))
	     (vec (make-array 0 :fill-pointer t :adjustable t
                              :element-type '(unsigned-byte 8))))
	(vector-push-extend +list+ vec)
	(dotimes (i length-of-encoded-length)
	  (vector-push-extend (aref encoded-length i) vec))
	(dolist (item serialized-items)
	  (dotimes (i (length item))
	    (vector-push-extend (aref item i) vec)))
	vec)
      (let ((vec (make-array 0 :fill-pointer t :adjustable t
                             :element-type '(unsigned-byte 8)))
	    (serialized-items nil))
	(loop for elt on list do
	     (push (serialize (car elt)) serialized-items)
	     (when (atom (cdr elt))
	       ;; The last element
	       (push (serialize (cdr elt)) serialized-items)))
	(let* ((total-length (reduce #'+ serialized-items :key 'length))
	       (encoded-length (encode-length total-length))
	       (length-of-encoded-length (length encoded-length)))
	  (vector-push-extend +dotted-list+ vec)
	  (dotimes (i length-of-encoded-length)
	    (vector-push-extend (aref encoded-length i) vec))
	  (dolist (item (reverse serialized-items))
	    (dotimes (i (length item))
	      (vector-push-extend (aref item i) vec)))
	  vec))))