Commit 543bad90f236960d44c632092e0b22df64cfbfd1
2017-06-25T23:58:09
rework thot-threaded
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108
diff --git a/thot-threaded.lisp b/thot-threaded.lisp
index 5666039..99bcaf3 100644
--- a/thot-threaded.lisp
+++ b/thot-threaded.lisp
@@ -1,54 +1,63 @@
(in-package :thot)
-(defvar *listen-fd*)
-
-(defun worker-thread ()
- ;(format t "~&WORKER THREAD~%")
- (loop
- (when *stop*
- (return))
- (cffi-socket:with-accept (clientfd) *listen-fd*
- (with-stream (stream (babel-io-stream (fd-io-stream clientfd)))
- (request-loop stream)))))
+(defvar *listen-fds*)
+
+(defun worker-thread (fd &optional dup)
+ (let ((listen-fd (if dup (unistd:dup fd) fd)))
+ (when dup
+ (pushnew listen-fd *listen-fds*))
+ (labels ((worker-thread-fd ()
+ ;;(format t "~&WORKER THREAD~%")
+ (unwind-protect
+ (loop
+ (when *stop*
+ (return))
+ (ignore-errors
+ (unistd:with-selected (`(,listen-fd) () () 1)
+ (readable writable errors)
+ (when readable
+ (socket:with-accept (clientfd) listen-fd
+ (with-stream (stream (babel-io-stream
+ (fd-io-stream clientfd)))
+ (request-loop stream)))))))
+ (when dup
+ (unistd:close listen-fd)))))
+ #'worker-thread-fd)))
(defparameter *init-threads* 8)
-(defvar *worker-threads*)
-(defvar *worker-sockfds*)
-
-(defun init-worker-threads (n)
- (loop
- (when (<= n (length *worker-threads*))
- (return))
- (let ((thread (bordeaux-threads:make-thread 'worker-thread
- :name "worker")))
- (push thread *worker-threads*))))
+(defun make-worker-threads (fd n)
+ (let ((threads ()))
+ (dotimes (i n)
+ (push (bordeaux-threads:make-thread
+ (worker-thread fd t)
+ :name "worker")
+ threads))
+ threads))
-(defun join-worker-threads ()
+(defun join-worker-threads (threads)
(setq *stop* t)
- (bordeaux-set:set-each (lambda (sockfd)
- (cffi-socket:shutdown sockfd t t))
- *worker-sockfds*)
- (loop
- (when (endp *worker-threads*)
- (return))
- (let ((thread (pop *worker-threads*)))
- (bordeaux-threads:join-thread thread))))
-
-(defmacro with-worker-threads (count &body body)
- `(let ((*worker-threads* ())
- (*worker-sockfds* (make-instance 'bordeaux-set:set)))
- (init-worker-threads ,count)
- (unwind-protect (progn ,@body)
- (join-worker-threads))))
+ (dolist (thread threads)
+ (bordeaux-threads:join-thread thread)))
+
+(defmacro with-worker-threads ((fd count) &body body)
+ (let ((threads (gensym "THREADS-")))
+ `(let* ((*listen-fds* ())
+ (,threads (make-worker-threads ,fd ,count)))
+ (unwind-protect (progn ,@body)
+ (join-worker-threads ,threads)))))
(defun acceptor-loop-threaded (fd)
(declare (type (unsigned-byte 31) fd))
- (setq *stop* nil
- *listen-fd* fd)
- (with-worker-threads (1- *init-threads*)
- (worker-thread)))
+ (set-nonblocking fd)
+ (with-worker-threads (fd (1- *init-threads*))
+ (funcall (worker-thread fd))))
(when bordeaux-threads:*supports-threads-p*
(setq *acceptor-loop* 'acceptor-loop-threaded))
+
+(untrace start acceptor-loop-threaded request-loop read write
+ set-nonblocking
+ socket:socket socket:bind socket:listen socket:accept
+ unistd:close unistd:select)