ice-9 async-queue

ice-9 async-queue

[Andy Wingo]

[-- Attachment #1: Type: text/plain, Size: 185 bytes --]

Hi all,

I was thinking of adding the following to Guile, to eventually help make
the web server a little less terrible.  What do you think?  I haven't
tested it properly yet.  

Andy


[-- Attachment #2: (ice-9 async-queue) --]
[-- Type: text/plain, Size: 4045 bytes --]

;;; Asynchronous queues

;; Copyright (C)  2012 Free Software Foundation, Inc.

;; This library is free software; you can redistribute it and/or
;; modify it under the terms of the GNU Lesser General Public
;; License as published by the Free Software Foundation; either
;; version 3 of the License, or (at your option) any later version.
;;
;; This library is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;; Lesser General Public License for more details.
;;
;; You should have received a copy of the GNU Lesser General Public
;; License along with this library; if not, write to the Free Software
;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
;; 02110-1301 USA

;;; Commentary:
;;;
;;; An implementation of thread-safe asynchronous queues, with both
;;; blocking and nonblocking interfaces.
;;;
;;; Code:

(define-module (ice-9 async-queue)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-9 gnu)
  #:use-module (ice-9 format)
  #:use-module (ice-9 threads)
  #:export (make-async-queue
            async-queue-length async-queue-capacity
            async-queue-push!
            async-queue-pop! async-queue-try-pop!))

;; One thing that we should be careful about is to avoid exposing
;; information about the way this queue is implemented.
;;
;; Currently we use an array, but it's easy to imagine a functional
;; implementation facilitated by compare-and-swap operations, with
;; perhaps the option to disable the blocking interfaces (and thereby
;; remove the need for the mutex and cond var).
;;

(define-record-type <async-queue>
  (make-aq mutex condvar buf length capacity read-idx)
  async-queue?
  (mutex aq-mutex)
  (condvar aq-condvar)
  (buf aq-buf)
  (capacity aq-capacity)
  (length aq-length set-aq-length!)
  (read-idx aq-read-idx set-aq-read-idx!))

(set-record-type-printer!
 <async-queue>
 (lambda (aq port)
   (format port "<async-queue ~x ~a/~a>" (object-address aq)
           (aq-length aq) (aq-capacity aq))))

(define (aq-inc! aq)
  (set-aq-length! aq (1+ (aq-length aq)))
  (signal-condition-variable (aq-condvar aq)))

(define (aq-dec! aq)
  (set-aq-length! aq (1- (aq-length aq)))
  (signal-condition-variable (aq-condvar aq)))

(define (aq-idx aq idx)
  (modulo idx (aq-capacity aq)))

(define (aq-wait aq time)
  (if time
      (wait-condition-variable (aq-condvar aq) (aq-mutex aq) time)
      (wait-condition-variable (aq-condvar aq) (aq-mutex aq))))

(define* (make-async-queue #:key (capacity 10))
  (make-aq (make-mutex)
           (make-condition-variable)
           (make-vector capacity #f)
           capacity
           0
           0))

(define (async-queue-length aq)
  (with-mutex (aq-mutex aq)
    (aq-length aq)))

(define (async-queue-capacity aq)
  (aq-capacity aq))

(define* (async-queue-push! aq item #:optional time)
  (with-mutex (aq-mutex aq)
    (let lp ()
      (if (< (aq-length aq) (aq-capacity aq))
          (let ((idx (aq-idx aq (+ (aq-read-idx aq) (aq-length aq)))))
            (vector-set! (aq-buf aq) idx item)
            (aq-inc! aq)
            #t)
          (and (aq-wait aq time) (lp))))))

(define* (async-queue-pop! aq #:optional time)
  (with-mutex (aq-mutex aq)
    (let lp ()
      (if (zero? (aq-length aq))
          (if (aq-wait aq time)
              (lp)
              (values #f #f))
          (let* ((idx (aq-read-idx aq))
                 (item (vector-ref (aq-buf aq) idx)))
            (vector-set! (aq-buf aq) idx #f)
            (set-aq-read-idx! aq (aq-idx aq (1+ idx)))
            (aq-dec! aq)
            (values item #t))))))

(define* (async-queue-try-pop! aq)
  (with-mutex (aq-mutex aq)
    (if (zero? (aq-length aq))
        (values #f #f)
        (let* ((idx (aq-read-idx aq))
               (item (vector-ref (aq-buf aq) idx)))
          (vector-set! (aq-buf aq) idx #f)
          (set-aq-read-idx! aq (aq-idx aq (1+ idx)))
          (aq-dec! aq)
          (values item #t)))))


[-- Attachment #3: Type: text/plain, Size: 26 bytes --]


-- 
http://wingolog.org/

Re: ice-9 async-queue

[Mike Gran]

> From: Andy Wingo <wingo@pobox.com>
>Subject: ice-9 async-queue
>;;; Asynchronous queues

Hey Andy,
 
FYI, there is also an (ice-9 q).  I haven't really looked
at it, but, maybe either (ice-9 q) or (ice-9 async-queue)
could become a generalized version and the other could
become a specific version or the same codebase.
 
-Mike

Re: ice-9 async-queue

[Ludovic Courtès]

Hi!

Andy Wingo <wingo@pobox.com> skribis:

> I was thinking of adding the following to Guile, to eventually help make
> the web server a little less terrible.  What do you think?

An “asynchronous queue” is a queue of tasks, right?

What kind of tasks would it be: I/O?  Computation?

How does it fit with the web server?

Thanks,
Ludo’.

Re: ice-9 async-queue

[Andy Wingo]

Hi Mike,

On Mon 06 Feb 2012 23:09, Mike Gran <spk121@yahoo.com> writes:

>> From: Andy Wingo <wingo@pobox.com>
>>Subject: ice-9 async-queue
>>;;; Asynchronous queues
>
> FYI, there is also an (ice-9 q).  I haven't really looked
> at it, but, maybe either (ice-9 q) or (ice-9 async-queue)
> could become a generalized version and the other could
> become a specific version or the same codebase.

The implementations have to be totally different, I think.  For example
(ice-9 q) has O(N) access to the length of the queue, which is something
that is accessed all the time in a blocking queue.  Given that we don't
really have interfaces, and we couldn't for (ice-9 q) given that it
isn't an abstract data type, I'm OK with a second implementation.

Andy
-- 
http://wingolog.org/

Re: ice-9 async-queue

[Andy Wingo]

On Mon 06 Feb 2012 23:09, Mike Gran <spk121@yahoo.com> writes:

> maybe either (ice-9 q) or (ice-9 async-queue) could become a
> generalized version and the other could become a specific version or
> the same codebase.

I forgot to mention the other way: ice-9 q can't be a specific version
of anything else, because its implementation is exposed as its
interface:

       A queue is implemented as a cons cell, the `car' containing a list
    of queued elements, and the `cdr' being the last cell in that list (for
    ease of enqueuing).

         (LIST . LAST-CELL)

    If the queue is empty, LIST is the empty list and LAST-CELL is `#f'.

       An application can directly access the queue list if desired, for
    instance to search the elements or to insert at a specific point.

Cheers,

Andy
-- 
http://wingolog.org/

Re: ice-9 async-queue

[Andy Wingo]

On Mon 06 Feb 2012 23:57, ludo@gnu.org (Ludovic Courtès) writes:

> Andy Wingo <wingo@pobox.com> skribis:
>
>> I was thinking of adding the following to Guile, to eventually help make
>> the web server a little less terrible.  What do you think?
>
> An “asynchronous queue” is a queue of tasks, right?

It's a message queue for use in threads organized as producers and
consumers.

> What kind of tasks would it be: I/O?  Computation?

Depends on what you want to do.

> How does it fit with the web server?

The web server is single-threaded and uses blocking IO (though it does
poll(2) for keepalive).  As such, any slow writer or slow reader can
block the process.  Using non-blocking I/O is too difficult, for now.
So, threads.

I'd like to create a pool of threads for I/O.  Some threads would pop
ports off of the "to-read" queue, read request headers and bodies, then
push the requests onto a "to-process" queue.  Something (currently the
main thread) would process requests, and push them onto the "to-write"
queue.  IO threads would pop data (or closures) off of the to-write
queue, and write them to clients, possibly pushing the ports back on a
"to-keepalive" queue, which the poll loop would notice and add those fds
back to the poll set.

I'd also like to consider creating a separate pool of threads for
computation.  Obviously the size of these thread pools would be
different.  We could use futures for that, I suppose, but I'd like to
also be able to stop those threads, forcefully if needed, when the web
server stops.

Andy
-- 
http://wingolog.org/

Re: ice-9 async-queue

[Daniel Hartwig]

>            async-queue-push!
>            async-queue-pop! async-queue-try-pop!))

Note that these verbs conflict with those defined in (ice-9 q) where
"push" adds elements to the front, not the rear, of the queue.


This is a very tidy implementation.  I presume the final version will
include a type and empty-ness predicate also?


Thanks

Re: ice-9 async-queue

[Andy Wingo]

On Tue 07 Feb 2012 11:02, Daniel Hartwig <mandyke@gmail.com> writes:

> Note that these verbs conflict with those defined in (ice-9 q) where
> "push" adds elements to the front, not the rear, of the queue.

Hum, indeed.  I guess (ice-9 q) is actually a deque...

> I presume the final version will include a type and empty-ness
> predicate also?

A type predicate, yes.  And some other things; see
wip-threaded-web-server.

Would you find an emptiness predicate useful?  It sounds like one of
those things that introduces time-of-check-to-time-of-use bugs.

Andy
-- 
http://wingolog.org/

Re: ice-9 async-queue

[Daniel Hartwig]

On 8 February 2012 05:46, Andy Wingo <wingo@pobox.com> wrote:
>> I presume the final version will include a type and empty-ness
>> predicate also?
>
> A type predicate, yes.  And some other things; see
> wip-threaded-web-server.
>
> Would you find an emptiness predicate useful?  It sounds like one of
> those things that introduces time-of-check-to-time-of-use bugs.

I guess most situations would be served better by try-pop.

Re: ice-9 async-queue

[Ludovic Courtès]

Hi!

Andy Wingo <wingo@pobox.com> skribis:

> The web server is single-threaded and uses blocking IO (though it does
> poll(2) for keepalive).  As such, any slow writer or slow reader can
> block the process.  Using non-blocking I/O is too difficult, for now.
> So, threads.
>
> I'd like to create a pool of threads for I/O.  Some threads would pop
> ports off of the "to-read" queue, read request headers and bodies, then
> push the requests onto a "to-process" queue.  Something (currently the
> main thread) would process requests, and push them onto the "to-write"
> queue.  IO threads would pop data (or closures) off of the to-write
> queue, and write them to clients, possibly pushing the ports back on a
> "to-keepalive" queue, which the poll loop would notice and add those fds
> back to the poll set.

OK, I see.

> I'd also like to consider creating a separate pool of threads for
> computation.  Obviously the size of these thread pools would be
> different.  We could use futures for that, I suppose, but I'd like to
> also be able to stop those threads, forcefully if needed, when the web
> server stops.

What do you think of adding a ‘cancel’ primitive to futures?

Thanks,
Ludo’.

Re: ice-9 async-queue

[Andy Wingo]

Hi Ludo :)

Picking up some loose ends...

On Wed 08 Feb 2012 14:44, ludo@gnu.org (Ludovic Courtès) writes:

> What do you think of adding a ‘cancel’ primitive to futures?

It sounds good, but tricky to implement.  I'm also not sure it's exactly
the right interface -- for example, Java seems to have switched entirely
away from the "cancel" API to the "interrupt" API.  But you would know
more about that than I would.  Would you use cancel-thread or would you
have the thread raise an exception via an async, or what?

Curiously yours,

Andy
-- 
http://wingolog.org/

Re: ice-9 async-queue

[Ludovic Courtès]

Hi!

Andy Wingo <wingo@pobox.com> skribis:

> On Wed 08 Feb 2012 14:44, ludo@gnu.org (Ludovic Courtès) writes:
>
>> What do you think of adding a ‘cancel’ primitive to futures?
>
> It sounds good, but tricky to implement.  I'm also not sure it's exactly
> the right interface -- for example, Java seems to have switched entirely
> away from the "cancel" API to the "interrupt" API.  But you would know
> more about that than I would.  Would you use cancel-thread or would you
> have the thread raise an exception via an async, or what?

I don’t know.

Actually, what kind of computation would the web server start that could
possibly need to be canceled?  (Pure computation, that is.)

Thanks,
Ludo’.