source: project/release/4/ugarit/trunk/ugarit-backend.scm @ 25555

(module ugarit-backend
        (make-storage ; Storage records
         storage?
         storage-max-block-size
         storage-writable?
         storage-unlinkable?
         storage-put!
         storage-flush!
         storage-exists?
         storage-get
         storage-link!
         storage-unlink!
         storage-set-tag!
         storage-tag
         storage-all-tags
         storage-remove-tag!
         storage-lock-tag!
         storage-tag-locked?
         storage-unlock-tag!
         storage-close!

         export-storage! ; Export a storage via stdin/stdout
         import-storage ; Create a storage from a command line
         )

(import scheme)
(import chicken)

(use ports)
(use matchable)
(use posix)
(use srfi-4)


(define-record storage
  max-block-size  ; Integer: largest size of block we can store
  writable? ; Boolean: Can we call put!, link!, unlink!, set-tag!, lock-tag!, unlock-tag!?
  unlinkable? ; Boolean: Can we call unlink?
  put! ; Procedure: (put! key data type) - stores the data (u8vector) under the key (string) with the given type tag (symbol) and a refcount of 1. Does nothing of the key is already in use.
  flush! ; Procedure: (flush!) - all previous changes must be flushed to disk by the time the continuation is applied.
  exists? ; Procedure: (exists? key) - returns the type of the block with the given key if it exists, or #f otherwise
  get ; Procedure: (get key) - returns the contents (u8vector) of the block with the given key (string) if it exists, or #f otherwise
  link! ; Procedure: (link key) - increments the refcount of the block
  unlink! ; Procedure: (unlink key) - decrements the refcount of the block. If it's now zero, deletes it but returns its value as a u8vector. If not, returns #f.
  set-tag! ; Procedure: (set-tag! name key) - assigns the given key (string) to the given tag (named with a string). Creates a new tag if the name has not previously been used, otherwise updates an existing tag
  tag ; Procedure: (tag name) - returns the key assigned to the given tag, or #f if it does not exist.
  all-tags ; Procedure: (all-tags) - returns a list of all existing tag names
  remove-tag! ; Procedure: (remove-tag! name) - removes the named tag
  lock-tag! ; Procedure: (lock-tag! name) - locks the named tag, or blocks if already locked
  tag-locked? ; Procedure: (tag-locked? name) - returns the locker identity string if the tag is locked, #f otherwise
  unlock-tag! ; Procedure: (unlock-tag! name) - unlocks the named tag
  close!)  ; Procedure: (close!) - closes the storage engine

(define *magic* 'ugarit-backend-protocol-1)

(define (describe-exception exn)
  (list (##sys#slot exn 1) (##sys#slot exn 2)))

(define-syntax with-error-reporting
  (er-macro-transformer
   (lambda (e r c)
     (let ((body (cdr e)))
       `(,(r 'call-with-current-continuation)
         (,(r 'lambda) (,(r 'escape))
          (,(r 'with-exception-handler)
           (,(r 'lambda) (,(r 'k))
            (,(r 'write) (,(r 'list) "error" (,(r 'describe-exception) ,(r 'k))))
            (,(r 'escape) #f))
           (,(r 'lambda) ()
            ,@body))))))))

;; Given a storage object, provide the storage remote access protocol
;; via current-input-port / current-output-port until the storage is closed
;; via the protocol.
(define (export-storage! storage)
  (set-buffering-mode! (current-output-port) #:none)

  ; Write the header
  (write *magic*) (newline)
  (write (list (storage-max-block-size storage)
               (storage-writable? storage)
               (storage-unlinkable? storage)))

  ; Engage command loop
  (let loop ()
    (newline)
    (let ((command (read)))
      (if (eof-object? command)
          (begin
            (with-error-reporting
             ((storage-close! storage))
             (write "goodbye"))
            (void))
          (match
           command

           (('put! key type length)
            (let ((data (read-u8vector length)))
              (with-error-reporting
               ((storage-put! storage) key data type)
               (write #t)))
            (loop))

           (('flush!)
            (with-error-reporting
             ((storage-flush! storage))
             (write #t))
            (loop))

           (('exists? key)
            (with-error-reporting
             (write ((storage-exists? storage) key)))
            (loop))

           (('get key)
            (with-error-reporting
             (let ((data ((storage-get storage) key)))
               (if data
                   (begin
                    (write (list (u8vector-length data)))
                    (write-u8vector data))
                   (write #f))))
            (loop))

           (('link! key)
            (with-error-reporting
             ((storage-link! storage) key)
             (write #t))
            (loop))

           (('unlink! key)
            (with-error-reporting
             (let ((data ((storage-unlink! storage) key)))
               (if data
                   (begin
                     (write (list (u8vector-length data)))
                     (write-u8vector data))
                   (write #f))))
            (loop))

           (('set-tag! name key)
            (with-error-reporting
             ((storage-set-tag! storage) name key)
             (write #t))
            (loop))

           (('tag name)
            (with-error-reporting
             (write ((storage-tag storage) name)))
            (loop))

           (('all-tags)
            (with-error-reporting
             (write ((storage-all-tags storage))))
            (loop))

           (('remove-tag! name)
            (with-error-reporting
             ((storage-remove-tag! storage) name)
             (write #t))
            (loop))

           (('lock-tag! name)
            (with-error-reporting
             ((storage-lock-tag! storage) name)
             (write #t))
            (loop))

           (('tag-locked? name)
            (with-error-reporting
             (write ((storage-tag-locked? storage) name)))
            (loop))

           (('unlock-tag! name)
            (with-error-reporting
             ((storage-unlock-tag! storage) name)
             (write #t))
            (loop))

           (('close!)
            (with-error-reporting
             ((storage-close! storage))
             (write "goodbye"))
            (void))

           (else
            (write (list "error" (sprintf "Bad command ~s" command)))
            (loop)))))))

(define (read-response port)
  (let ((response (read port)))
   (match response
          (("error" err) (error "Backend protocol error" err))
          (else response))))

(define (read-response-body port)
  (let ((response (read-response port)))
    (if response
        (read-u8vector (car response) port)
        #f)))

;; Given the command line to a storage remote access protocol server,
;; activate it and return a storage object providing access to the
;; server.
(define (import-storage command-line . args)
  (let-optionals args ((debug #f))
   (let-values (((responses commands pid)
                 (process command-line)))

     #;(set-buffering-mode! commands #:none)

     (if debug (print "~a: process opened" command-line))
     (let ((magic (read responses)))
       (if debug (print "~a: read magic ~a" command-line magic))
       (if (not (equal? magic *magic*))
           (error "Invalid backend protocol header magic" magic))

       (let ((header (read responses)))
         (if debug (print "~a: read header" command-line header))
         (if (not (list? header))
             (error "Invalid backend protocol header" header))
         (if (not (= (length header) 3))
             (error "Invalid backend protocol header" header))
         (let ((max-block-size (car header))
               (writable? (cadr header))
               (unlinkable? (caddr header)))
           (make-storage
            max-block-size
            writable?
            unlinkable?

            (lambda (key data type)     ; put!
              (if debug (printf "~a: put!" command-line))
              (write `(put! ,key ,type ,(u8vector-length data)) commands)
              (write-u8vector data commands)
              (read-response responses)
              (void))

            (lambda ()                  ; flush!
              (if debug (printf "~a: flush!" command-line))
              (write `(flush!) commands)
              (read-response responses)
              (void))

            (lambda (key)               ; exists?
              (if debug (printf "~a: exists?" command-line))
              (write `(exists? ,key) commands)
              (read-response responses))

            (lambda (key)               ; get
              (if debug (printf "~a: get" command-line))
              (write `(get ,key) commands)
              (read-response-body responses))

            (lambda (key)               ; link!
              (if debug (printf "~a: link!" command-line))
              (write `(link! ,key) commands)
              (read-response responses)
              (void))

            (lambda (key)               ; unlink!
              (if debug (printf "~a: unlink! ~s" command-line key))
              (write `(unlink! ,key) commands)
              (read-response-body responses))

            (lambda (name key)          ; set-tag!
              (if debug (printf "~a: set-tag!" command-line))
              (write `(set-tag! ,name ,key) commands)
              (read-response responses)
              (void))

            (lambda (name)              ; tag
              (if debug (printf "~a: tag" command-line))
              (write `(tag ,name) commands)
              (read-response responses))

            (lambda ()                  ; all-tags
              (if debug (printf "~a: all-tags" command-line))
              (write `(all-tags) commands)
              (read-response responses))

            (lambda (name)              ; remove-tag!
              (if debug (printf "~a: remove-tag!" command-line))
              (write `(remove-tag! ,name) commands)
              (read-response responses)
              (void))

            (lambda (name)              ; lock-tag!
              (if debug (printf "~a: lock-tag!" command-line))
              (write `(lock-tag! ,name) commands)
              (read-response responses)
              (void))

            (lambda (name)              ; tag-locked?
              (if debug (printf "~a: tag-locked?" command-line))
              (write `(tag-locked? ,name) commands)
              (read-response responses))

            (lambda (name)              ; unlock-tag!
              (if debug (printf "~a: unlock-tag!" command-line))
              (write `(unlock-tag! ,name) commands)
              (read-response responses)
              (void))

            (lambda ()                  ; close!
              (if debug (printf "~a: close!!" command-line))
              (write '(close!) commands)
              (read-response responses)
              (close-input-port responses)
              (close-output-port commands)
              (void)))))))))

)