small-cl-src

Download

small-cl-src@common-lisp.net

August 2004

2 participants
2 discussions

[Small-cl-src] ipc.lisp
by Helmut Eller 28 Aug '04

28 Aug '04

;; ipc.lisp -- IPC with SYS V message queues. ;; ;; Written by Helmut Eller 2004 ;; ;; This package implements some Erlang-style concurrency primitives ;; with Unix processes and SYS V message queues. The following ;; primitives are available: ;; ;; spawn: forks a new process and calls a function in the new process. ;; ;; snd: sends a message to another process. The message must be ;; PRINTable and READable. ;; ;; rcv: receives a message. A test predicate can be used to ;; emulate `selective receive'. This operation blocks if there's no ;; message in the mailbox, but a timeout value can be specified. ;; ;; Some simple usage examples are at the end of this file. ;; ;; Caution: Erlangish link/unlink is not supported, so it is quite ;; difficult to handle errors in other processes. Currently a child ;; process prints a backtrace and exits instead of entering a ;; debugger. There's also no way to discover or kill other processes. ;; You'll have to use ps etc. for this. The SYS V message queues are ;; not reliably removed. Again you'll have to use ipcs and ipcrm to ;; do it manually. ;; ;; This code is written for CMUCL. (defpackage :ipc (:use :cl :ext :unix :alien :c-call) (:export :spawn :snd :rcv :self)) (in-package :ipc) (defvar *ipc-debug* t) (defun dbg (fstring &rest args) (when *ipc-debug* (apply #'format *debug-io* fstring args) (finish-output *debug-io*))) (defmacro ccall ((name rettype &rest argtypes) &rest args) "Call the C function NAME with ARGS." `(alien:alien-funcall (alien:extern-alien ,name (function ,rettype ,@argtypes)) ,@args)) (defmacro let-errno ((var errno value) &body body) "Execute VALUE; bind the result to VAR, bind ERRNO, and execute BODY." `(let ((,var ,value)) (let ((,errno ;; unix-errno is suddenly a function in recent cmucls. ,(cond ((fboundp 'unix:unix-errno) '(unix:unix-errno)) (t `(progn (unix::unix-get-errno) unix:unix-errno))))) ,@body))) (def-alien-type sap system-area-pointer) (def-alien-type uint unsigned) (defun unix-error (errno) (error "~A" (get-unix-error-msg errno))) (defconstant +ipc_nowait+ #x800) (defconstant +ipc_creat+ #x200) (defconstant +ipc_excl+ #x400) (defconstant +ipc_set+ #x1) (defconstant +ipc_rmid+ #x0) (defun %msgget (key flags) (ccall (msgget int uint uint) key flags)) (defun make-mq () "Create a new SYS V message queue. Return the (per process) queue handle and the public key." (let ((key (random #xffff))) (let-errno (mq errno (%msgget key (logior #o600 +ipc_creat+ +ipc_excl+))) (cond ((/= mq -1) (values mq key)) ((= errno unix:eexist) (make-mq)) ; retry with new key (t (unix-error errno)))))) (defun get-mq (key) "Get the (existing) message queue with id KEY." (let-errno (mq errno (%msgget key #o600)) (cond ((= mq -1) (unix-error errno)) (t mq)))) (defun delete-mq (mq) (dbg "; Closing: ~A (pid ~A)~%" mq (unix-getpid)) (let-errno (code errno (ccall (msgctl int int int sap) mq +ipc_rmid+ (sys:int-sap 0))) (cond ((zerop code)) (t (unix-error errno))))) (defun %msgsnd (mq string) (ccall (msgsnd int int sap uint uint) mq (sys:vector-sap string) (length string) +ipc_nowait+)) (defun msgsnd (mq string) "Write STRING to message queue MQ." (let-errno (code errno (%msgsnd mq string)) (cond ((/= code -1) nil) (t (unix-error errno))))) (defun %msgrcv (mq buffer) (ccall (msgrcv int int sap uint uint uint) mq (sys:vector-sap buffer) (length buffer) 0 0)) (defun msgrcv (mq) "Return the first message in MQ." (let ((buffer (make-string 8192))) (let-errno (code errno (%msgrcv mq buffer)) (cond ((/= code -1) (subseq buffer 0 code)) ((= errno unix:eintr) (msgrcv mq)) (t (unix-error errno)))))) (defun decode (string) (with-standard-io-syntax (read-from-string string))) (defun encode (message) (with-standard-io-syntax (prin1-to-string message))) ;;; Process structure (pendant to Erlangs PID) (defstruct (process (:conc-name process.) (:predicate process?)) ;; Unix PID (pid (required-argument) :type fixnum :read-only t) ;; The key of the SYSV message queue (qid (required-argument) :type fixnum :read-only t)) (defvar *queues* (make-hash-table) "Maps pids to SYSV IPC message queues.") (defun process-queue (process) (or (gethash (process.pid process) *queues*) (setf (process-queue process) (get-mq (process.qid process))))) (defun (setf process-queue) (queue process) (setf (gethash (process.pid process) *queues*) queue)) (defvar *self* nil) (defvar *parent* nil) (defvar *linked-processes* '()) (defvar *trap-exit* nil) (defun init-self () "Allocate a new message queue and initialize *SELF*." (multiple-value-bind (mq qid) (make-mq) (let ((p (make-process :pid (unix-getpid) :qid qid))) (setf (process-queue p) mq) (setf *self* p)))) (defun self () (or *self* (init-self))) (defun parent () *parent*) (defun self? (process) "Is PROCESS this unix process?" (equalp process (self))) (defun alive? (process) (unix-kill (process.pid process) :check)) (defvar *mailbox* (list)) (defun snd (message process) "Send MESSAGE to PROCESS." (cond ((self? process) (setf *mailbox* (nconc *mailbox* (list message)))) (t (msgsnd (process-queue process) (encode message))))) ;;; SYS V message queues have no support for receive with timeouts. ;;; We use the itimer to implement timeouts. (defun assert-timer-unused () "Raise an error if the reatime is already in use." (multiple-value-bind (ok isec iusec vsec vusec) (unix-getitimer :real) (declare (ignore ok isec iusec)) (assert (and (zerop vsec) (zerop vusec)) () "itimer already used"))) (defun decode-timeout (timeout) "Return the timevalue TIMEOUT as to two values integers SECS USECS. TIMEOUT the number of seconds." (multiple-value-bind (secs rem) (truncate timeout) (let ((usecs (round (* rem 1000000)))) (assert (or (plusp secs) (and (zerop secs) (plusp usecs)))) (values secs usecs)))) (defun with-timeout (timeout fn timeout-fn) "Call FN. If executing FN completes withing TIMEOUT seconds return the result. Otherwise abort, unwind the stack and call TIMEOUT-FN instead. TIMEOUT is in seconds (can be a float). Return either the result of executing FN or the result of calling TIMEOUT-FN." (if (minusp timeout) (funcall timeout-fn) (%with-timeout timeout fn timeout-fn))) (defun %with-timeout (timeout fn timeout-fn) (block abort (return-from %with-timeout (labels ((handler (signal code scp) (declare (ignore signal code scp)) (return-from abort nil))) (multiple-value-bind (secs usecs) (decode-timeout timeout) (assert-timer-unused) (sys:with-enabled-interrupts ((unix:sigalrm #'handler)) (unix-setitimer :real 0 0 secs usecs) (unwind-protect (funcall fn) (unix-setitimer :real 0 0 0 0))))))) (funcall timeout-fn)) (defun endtime (timeout) (+ (get-internal-real-time) (/ timeout (coerce internal-time-units-per-second 'double-float)))) (defun seconds-until (endtime) (* (- endtime (get-internal-real-time)) internal-time-units-per-second)) (defun rcv-from-mq (mq test endtime) "Return the first message from MQ satisfying the predicate TEST. The second return value is non-nil when timeout ENDTIME expires." (let ((msg (if endtime (with-timeout (seconds-until endtime) (lambda () (decode (msgrcv mq))) (lambda () (return-from rcv-from-mq (values nil t)))) (decode (msgrcv mq))))) (cond ((funcall test msg) (values msg nil)) (t (setf *mailbox* (nconc *mailbox* (list msg))) (rcv-from-mq mq test endtime))))) (defun rcv (&key (test (constantly t)) timeout) "Receive the next message which satisfies TEST. Return the message and a flag indicating whether the timeout expired." (let ((tail (member-if test *mailbox*))) (cond (tail (setf *mailbox* (nconc (ldiff *mailbox* tail) (cdr tail))) (values (car tail) nil)) (t (rcv-from-mq (process-queue (self)) test (if timeout (endtime timeout))))))) (in-package :vm) (eval-when (:compile-toplevel :load-toplevel :execute) (defknown %reset-csp (function) nil) ;; Reset the control stack and call FN with the empty stack. ;; Unwind-protect blocks are reset and NOT executed. (define-vop (reset-csp) (:policy :fast-safe) (:args (fn :scs (descriptor-reg control-stack))) (:save-p t) (:translate %reset-csp) (:generator 0 (move eax-tn fn) (load-foreign-data-symbol esp-tn "control_stack_end") (inst mov esp-tn (make-ea :dword :base esp-tn)) (store-symbol-value 0 lisp::*current-catch-block*) (store-symbol-value 0 lisp::*current-unwind-protect-block*) (inst push 0) (inst mov ebp-tn esp-tn) (inst mov esi-tn esp-tn) (inst jmp (make-fixup 'tail-call-variable :assembly-routine))))) (in-package :ipc) (defun call-with-empty-stacks (fn) (mp::unbind-binding-stack) (setf x86::*alien-stack* (kernel:make-lisp-obj mp::*alien-stack-top*)) (setf kernel:*eval-stack-top* 0) (setf (kernel:binding-stack-pointer-sap) (alien:extern-alien "binding_stack" system-area-pointer)) (vm::%reset-csp fn)) (defun child-debugger-hook (condition hook) "Print a backtrace and exit." (declare (ignore hook)) (ignore-errors (format *debug-io* "~2&~A~% [Condition of type ~S]~2&" (debug::safe-condition-message condition) (type-of condition))) (ignore-errors (debug:backtrace)) (throw '%exit-child nil)) (defun start-child (stream mq fn args) (let ((*standard-output* stream) (*debug-io* stream) (*error-output* stream) (*query-io* stream) (*standard-input* stream)) (catch '%exit-child (unwind-protect (let ((*debugger-hook* (lambda (c h) (child-debugger-hook c h)))) (apply fn args)) (ignore-errors (delete-mq mq) (dbg "; Child exited: ~S~%" (self)) (lisp::finish-standard-output-streams)) (unix:unix-exit))))) (defun initialize-child (parent mq qid) (setf *parent* parent) (setf *self* (make-process :pid (unix-getpid) :qid qid)) (setf (process-queue *self*) mq) (setf *mailbox* (list))) (defun %spawn (fn args) (multiple-value-bind (mq qid) (make-mq) (let ((parent (self))) (mapc #'force-output (list *standard-output* *debug-io* *query-io*)) ;; CMUCL's SIGCHLD handler (sys:enable-interrupt unix:sigchld #'ext::sigchld-handler) (multiple-value-bind (pid errno) (unix-fork) (cond ((and pid (zerop pid)) (initialize-child parent mq qid) (let ((io *debug-io*)) (call-with-empty-stacks (lambda () (start-child io mq fn args))))) (pid (dbg "; Forked: ~D ~D (parent ~D)~%" pid qid (unix-getpid)) (let ((child (make-process :pid pid :qid qid))) (setf (process-queue child) mq) child)) (t (error "fork: ~A" (get-unix-error-msg errno)))))))) (defun spawn (fn &rest args) "Create a new process and apply FN to ARGS in the new process." (%spawn fn args)) #| ;; The counter process maintains the current value. Operations on the ;; counter are invoked via messges. The messages are as follows: ;; :up incerments the counter ;; :quit termintates the counter process ;; <a process> sends the current value to the process. (defun counter (value) (dbg "value: ~S~%" value) (let ((msg (rcv))) (dbg "msg: ~S~%" msg) (cond ((eq msg :up) (counter (1+ value))) ((eq msg :down) (counter (1- value))) ((eq msg :quit) ;; return ) ((process? msg) (snd value msg) (counter value))))) (defun test-counter () (let ((p (spawn (lambda () (counter 0))))) (snd :up p) (snd (self) p) (format t "~&value: ~A~&" (rcv)) (snd :up p) (snd (self) p) (format t "~&value: ~A~&" (rcv)) (snd :quit p))) ;; This expamle counts the nodes in a cons tree by spawing a new ;; process for each subtree. (defun count-nodes-aux (tree parent) "Count the number of nodes in TREE an send it to PARENT." (cond ((atom tree) (snd 1 parent)) (t (spawn #'count-nodes-aux (car tree) (self)) (spawn #'count-nodes-aux (cdr tree) (self)) (snd (+ 1 (rcv) (rcv)) parent)))) (defun count-nodes (tree) (count-nodes-aux tree (self)) (rcv)) (defun test-count-nodes () (time (count-nodes '(a ((b c)) d e )))) |# ;; Local Variables: ;; eval: (put 'let-errno 'common-lisp-indent-function 1) ;; End:

1 0

[Small-cl-src] bit I/O
by Zach Beane 27 Aug '04

27 Aug '04

;;;; The following stuff is taken, verbatim and somewhat unpolished, ;;;; from something I'm currently working on. ;;;; ;;;; Bit-level I/O. The Flash spec has various structures packed into ;;;; bit fields. ;;;; ;;;; $Id: bitio.lisp,v 1.5 2004/07/08 02:47:20 xach Exp $ (in-package :flash) (defclass bitstream () ((current-byte :initarg :current-byte :accessor current-byte) (bits-left :initarg :bits-left :accessor bits-left) (stream :initarg :stream :reader stream)) (:documentation "A class for writing or reading bits to or from an (UNSIGNED-BYTE 8) stream.")) ;;; Input (defun make-input-bitstream (stream &key (current-byte 0) (bits-left 0)) "Return a bitstream suitable for reading. The CURRENT-BYTE and BITS-LEFT, if specified, initialize the internal byte buffer and count." (assert (input-stream-p stream)) (assert (subtypep (stream-element-type stream) '(unsigned-byte 8))) (make-instance 'bitstream :stream stream :current-byte current-byte :bits-left bits-left)) (defmethod read-bits (width (bitstream bitstream) &optional (signedp nil)) (with-accessors ((bits-left bits-left) (input stream) (current-byte current-byte)) bitstream (loop with result = 0 with bits-needed = width while (plusp bits-needed) if (> bits-needed bits-left) do (setf result (logior (ash result bits-left) (ldb (byte bits-left 0) current-byte)) bits-needed (- bits-needed bits-left) current-byte (read-byte input) bits-left 8) else do (setf result (logior (ash result bits-needed) (ldb (byte bits-needed (- bits-left bits-needed)) current-byte)) bits-left (- bits-left bits-needed) bits-needed 0) finally (if (and signedp (logbitp (1- width) result)) (return (1- (- (logandc2 (1- (ash 1 width)) result)))) (return result))))) (defmethod read-boolean ((bitstream bitstream)) "Read and return a single bit from BITSTREAM as a boolean." (logbitp 0 (read-bits 1 bitstream))) (defmacro with-input-bitstream ((bitstream stream &key (current-byte 0) (bits-left 0)) &body body) "Evaluate BODY with a newly-created input bitstream bound to BITSTREAM." (let ((-current-byte- (gensym)) (-bits-left- (gensym))) `(let* ((,-current-byte- ,current-byte) (,-bits-left- ,bits-left) (,bitstream (make-input-bitstream ,stream :current-byte ,-current-byte- :bits-left ,-bits-left-))) ,@body))) ;;; Output (defun make-output-bitstream (stream &key (current-byte 0) (bits-left 8)) "Return a bitstream suitable for writing. The CURRENT-BYTE and BITS-LEFT, if specified, initialize the internal byte buffer and count." (assert (output-stream-p stream)) (assert (subtypep (stream-element-type stream) '(unsigned-byte 8))) (make-instance 'bitstream :stream stream :current-byte current-byte :bits-left bits-left)) (defmethod write-bits (integer width (bitstream bitstream)) "Write the rightmost WIDTH bits from INTEGER to the bitstream" (with-accessors ((bits-left bits-left) (stream stream) (current-byte current-byte)) bitstream (when (minusp integer) (setf integer (ldb (byte width 0) integer))) (loop with bits-needed = width while (plusp bits-needed) if (> bits-needed bits-left) do (setf (ldb (byte bits-left 0) current-byte) (ldb (byte bits-left (- bits-needed bits-left)) integer) bits-needed (- bits-needed bits-left)) (write-byte current-byte stream) (setf current-byte 0 bits-left 8) else do (setf (ldb (byte bits-needed (- bits-left bits-needed)) current-byte) (ldb (byte bits-needed 0) integer) bits-left (- bits-left bits-needed) bits-needed 0)))) (defmethod write-bit (bit (bitstream bitstream)) "Write a single bit to the bitstream." (write-bits bit 1 bitstream)) (defmethod write-boolean (boolean (bitstream bitstream)) "Write BOOLEAN as a bit to the bitstream." (if boolean (write-bit 1 bitstream) (write-bit 0 bitstream))) (defmethod finish-write ((bitstream bitstream)) "Flush the output buffer byte of BITSTREAM. Uninitialized bits are written as 0." (when (< (bits-left bitstream) 8) (write-byte (current-byte bitstream) (stream bitstream)) (setf (current-byte bitstream) 0) (setf (bits-left bitstream) 8))) (defmacro with-output-bitstream ((bitstream stream &key (current-byte 0) (bits-left 8)) &body body) "Evaluate BODY with a newly-created output bitstream bound to BITSTREAM. Pending output is flushed at the end of evaluation." (let ((-current-byte- (gensym)) (-bits-left- (gensym))) `(let* ((,-current-byte- ,current-byte) (,-bits-left- ,bits-left) (,bitstream (make-output-bitstream ,stream :current-byte ,-current-byte- :bits-left ,-bits-left-))) (unwind-protect (progn ,@body) (finish-write ,bitstream))))) (defun test-bitstreams (file width integer) (with-open-file (output file :direction :output :element-type '(unsigned-byte 8) :if-exists :supersede :if-does-not-exist :create) (with-output-bitstream (bitstream output) (write-bits integer width bitstream))) (with-open-file (input file :direction :input :element-type '(unsigned-byte 8)) (with-input-bitstream (bitstream input) (list (read-bits width bitstream t) integer)))) ;;; Utility (defun sufficient-bits (&rest numbers) "Return the number of bits necessary to represent the largest number in NUMBERS." (loop for number in numbers maximizing (1+ (integer-length number))))

1 1