Move Unix I/O multiplexers to io.unix.multiplexers, new run loop-based multiplexer integrates a kqueue with a CFRunLoop on Mac OS X

basis/core-foundation/core-foundation.factor

@@ -1,8 +1,8 @@
 ! Copyright (C) 2006, 2008 Slava Pestov
 ! See http://factorcode.org/license.txt for BSD license.
 USING: alien alien.c-types alien.strings alien.syntax kernel
-math sequences io.encodings.utf8 destructors accessors
-combinators byte-arrays ;
+math math.bitwise sequences io.encodings.utf8 destructors
+accessors combinators byte-arrays ;
 IN: core-foundation
 
 TYPEDEF: void* CFAllocatorRef
@@ -195,11 +195,22 @@ FUNCTION: CFFileDescriptorRef CFFileDescriptorCreate (
     CFFileDescriptorContext* context
 ) ;
 
+: kCFFileDescriptorReadCallBack 1 ; inline
+: kCFFileDescriptorWriteCallBack 2 ; inline
+   
 FUNCTION: void CFFileDescriptorEnableCallBacks (
     CFFileDescriptorRef f,
     CFOptionFlags callBackTypes
 ) ;
 
+: enable-all-callbacks ( fd -- )
+    { kCFFileDescriptorReadCallBack kCFFileDescriptorWriteCallBack } flags
+    CFFileDescriptorEnableCallBacks ;
+
+: <CFFileDescriptor> ( fd callback -- handle )
+    [ f swap ] [ t swap ] bi* f CFFileDescriptorCreate
+    [ "CFFileDescriptorCreate failed" throw ] unless* ;
+
 : load-framework ( name -- )
     dup <CFBundle> [
         CFBundleLoadExecutable drop

basis/core-foundation/fsevents/fsevents.factor

@@ -3,10 +3,10 @@
 USING: alien alien.c-types alien.strings alien.syntax kernel
 math sequences namespaces make assocs init accessors
 continuations combinators core-foundation
-core-foundation.run-loop core-foundation.run-loop.thread
-io.encodings.utf8 destructors locals arrays
-specialized-arrays.direct.alien specialized-arrays.direct.int
-specialized-arrays.direct.longlong ;
+core-foundation.run-loop io.encodings.utf8 destructors locals
+arrays specialized-arrays.direct.alien
+specialized-arrays.direct.int specialized-arrays.direct.longlong
+;
 IN: core-foundation.fsevents
 
 : kFSEventStreamCreateFlagUseCFTypes 2 ; inline

basis/core-foundation/run-loop/run-loop.factor

@@ -32,6 +32,12 @@ FUNCTION: void CFRunLoopAddSource (
    CFStringRef mode
 ) ;
 
+FUNCTION: void CFRunLoopRemoveSource (
+   CFRunLoopRef rl,
+   CFRunLoopSourceRef source,
+   CFStringRef mode
+) ;
+
 : CFRunLoopDefaultMode ( -- alien )
     #! Ugly, but we don't have static NSStrings
     \ CFRunLoopDefaultMode get-global dup expired? [

basis/core-foundation/run-loop/thread/summary.txt

@@ -1 +0,0 @@
-Vocabulary with init hook for running CoreFoundation event loop

basis/core-foundation/run-loop/thread/thread.factor

@@ -1,16 +0,0 @@
-! Copyright (C) 2008 Slava Pestov
-! See http://factorcode.org/license.txt for BSD license.
-USING: calendar core-foundation.run-loop init kernel threads ;
-IN: core-foundation.run-loop.thread
-
-! Load this vocabulary if you need a run loop running.
-
-: run-loop-thread ( -- )
-    CFRunLoopDefaultMode 0 f CFRunLoopRunInMode
-    kCFRunLoopRunHandledSource = [ 1 seconds sleep ] unless
-    run-loop-thread ;
-
-: start-run-loop-thread ( -- )
-    [ run-loop-thread t ] "CFRunLoop dispatcher" spawn-server drop ;
-
-[ start-run-loop-thread ] "core-foundation.run-loop.thread" add-init-hook

basis/io/unix/backend/backend.factor

@@ -5,7 +5,7 @@ kernel.private math io.ports sequences strings sbufs threads
 unix vectors io.buffers io.backend io.encodings math.parser
 continuations system libc qualified namespaces make io.timeouts
 io.encodings.utf8 destructors accessors summary combinators
-locals unix.time fry ;
+locals unix.time fry io.unix.multiplexers ;
 QUALIFIED: io
 IN: io.unix.backend
 
@@ -37,38 +37,6 @@ M: fd dispose
 
 M: fd handle-fd dup check-disposed fd>> ;
 
-! I/O multiplexers
-TUPLE: mx fd reads writes ;
-
-: new-mx ( class -- obj )
-    new
-        H{ } clone >>reads
-        H{ } clone >>writes ; inline
-
-GENERIC: add-input-callback ( thread fd mx -- )
-
-M: mx add-input-callback reads>> push-at ;
-
-GENERIC: add-output-callback ( thread fd mx -- )
-
-M: mx add-output-callback writes>> push-at ;
-
-GENERIC: remove-input-callbacks ( fd mx -- callbacks )
-
-M: mx remove-input-callbacks reads>> delete-at* drop ;
-
-GENERIC: remove-output-callbacks ( fd mx -- callbacks )
-
-M: mx remove-output-callbacks writes>> delete-at* drop ;
-
-GENERIC: wait-for-events ( ms mx -- )
-
-: input-available ( fd mx -- )
-    reads>> delete-at* drop [ resume ] each ;
-
-: output-available ( fd mx -- )
-    writes>> delete-at* drop [ resume ] each ;
-
 M: fd cancel-operation ( fd -- )
     dup disposed>> [ drop ] [
         fd>>

basis/io/unix/bsd/bsd.factor

@@ -1,7 +1,8 @@
 ! Copyright (C) 2008 Slava Pestov.
 ! See http://factorcode.org/license.txt for BSD license.
 USING: namespaces system kernel accessors assocs continuations
-unix io.backend io.unix.backend io.unix.kqueue ;
+unix io.backend io.unix.backend io.unix.multiplexers
+io.unix.multiplexers.kqueue ;
 IN: io.unix.bsd
 
 M: bsd init-io ( -- )

basis/io/unix/linux/linux.factor

@@ -1,7 +1,7 @@
 ! Copyright (C) 2008 Slava Pestov.
 ! See http://factorcode.org/license.txt for BSD license.
-USING: kernel io.backend io.monitors io.unix.backend
-io.unix.epoll io.unix.linux.monitors system namespaces ;
+USING: kernel system namespaces io.backend io.unix.backend
+io.unix.multiplexers io.unix.multiplexers.epoll ;
 IN: io.unix.linux
 
 M: linux init-io ( -- )

basis/io/unix/linux/monitors/monitors.factor

@@ -2,10 +2,10 @@
 ! See http://factorcode.org/license.txt for BSD license.
 USING: kernel io.backend io.monitors io.monitors.recursive
 io.files io.buffers io.monitors io.ports io.timeouts
-io.unix.backend io.unix.select io.encodings.utf8
-unix.linux.inotify assocs namespaces make threads continuations
-init math math.bitwise sets alien alien.strings alien.c-types
-vocabs.loader accessors system hashtables destructors unix ;
+io.unix.backend io.encodings.utf8 unix.linux.inotify assocs
+namespaces make threads continuations init math math.bitwise
+sets alien alien.strings alien.c-types vocabs.loader accessors
+system hashtables destructors unix ;
 IN: io.unix.linux.monitors
 
 SYMBOL: watches

basis/io/unix/macosx/macosx.factor

@@ -1,7 +1,10 @@
 ! Copyright (C) 2008 Slava Pestov.
 ! See http://factorcode.org/license.txt for BSD license.
+USING: io.backend system namespaces io.unix.multiplexers
+io.unix.multiplexers.run-loop ;
 IN: io.unix.macosx
-USING: io.unix.backend io.unix.bsd io.backend
-namespaces system ;
+
+M: macosx init-io ( -- )
+    <run-loop-mx> mx set-global ;
 
 macosx set-io-backend

basis/io/unix/multiplexers/epoll/epoll.factor

@@ -0,0 +1,66 @@
+! Copyright (C) 2008 Slava Pestov.
+! See http://factorcode.org/license.txt for BSD license.
+USING: accessors alien.c-types kernel destructors bit-arrays
+sequences assocs struct-arrays math namespaces locals fry unix
+unix.linux.epoll unix.time io.ports io.unix.backend
+io.unix.multiplexers ;
+IN: io.unix.multiplexers.epoll
+
+TUPLE: epoll-mx < mx events ;
+
+: max-events ( -- n )
+    #! We read up to 256 events at a time. This is an arbitrary
+    #! constant...
+    256 ; inline
+
+: <epoll-mx> ( -- mx )
+    epoll-mx new-mx
+        max-events epoll_create dup io-error >>fd
+        max-events "epoll-event" <struct-array> >>events ;
+
+M: epoll-mx dispose fd>> close-file ;
+
+: make-event ( fd events -- event )
+    "epoll-event" <c-object>
+    [ set-epoll-event-events ] keep
+    [ set-epoll-event-fd ] keep ;
+
+:: do-epoll-ctl ( fd mx what events -- )
+    mx fd>> what fd fd events make-event epoll_ctl io-error ;
+
+: do-epoll-add ( fd mx events -- )
+    EPOLL_CTL_ADD swap EPOLLONESHOT bitor do-epoll-ctl ;
+
+: do-epoll-del ( fd mx events -- )
+    EPOLL_CTL_DEL swap do-epoll-ctl ;
+
+M: epoll-mx add-input-callback ( thread fd mx -- )
+    [ EPOLLIN do-epoll-add ] [ call-next-method ] 2bi ;
+
+M: epoll-mx add-output-callback ( thread fd mx -- )
+    [ EPOLLOUT do-epoll-add ] [ call-next-method ] 2bi ;
+
+M: epoll-mx remove-input-callbacks ( fd mx -- seq )
+    2dup reads>> key? [
+        [ call-next-method ] [ EPOLLIN do-epoll-del ] 2bi
+    ] [ 2drop f ] if ;
+
+M: epoll-mx remove-output-callbacks ( fd mx -- seq )
+    2dup writes>> key? [
+        [ EPOLLOUT do-epoll-del ] [ call-next-method ] 2bi
+    ] [ 2drop f ] if ;
+
+: wait-event ( mx us -- n )
+    [ [ fd>> ] [ events>> ] bi [ underlying>> ] [ length ] bi ] [ 1000 /i ] bi*
+    epoll_wait multiplexer-error ;
+
+: handle-event ( event mx -- )
+    [ epoll-event-fd ] dip
+    [ EPOLLIN EPOLLOUT bitor do-epoll-del ]
+    [ input-available ] [ output-available ] 2tri ;
+
+: handle-events ( mx n -- )
+    [ dup events>> ] dip head-slice swap '[ _ handle-event ] each ;
+
+M: epoll-mx wait-for-events ( us mx -- )
+    swap 60000000 or dupd wait-event handle-events ;

basis/io/unix/multiplexers/kqueue/authors.txt

@@ -0,0 +1 @@
+Slava Pestov

basis/io/unix/multiplexers/kqueue/kqueue.factor

@@ -0,0 +1,76 @@
+! Copyright (C) 2008 Slava Pestov.
+! See http://factorcode.org/license.txt for BSD license.
+USING: accessors alien.c-types combinators destructors
+io.unix.backend kernel math.bitwise sequences struct-arrays unix
+unix.kqueue unix.time assocs io.unix.multiplexers ;
+IN: io.unix.multiplexers.kqueue
+
+TUPLE: kqueue-mx < mx events ;
+
+: max-events ( -- n )
+    #! We read up to 256 events at a time. This is an arbitrary
+    #! constant...
+    256 ; inline
+
+: <kqueue-mx> ( -- mx )
+    kqueue-mx new-mx
+        kqueue dup io-error >>fd
+        max-events "kevent" <struct-array> >>events ;
+
+M: kqueue-mx dispose fd>> close-file ;
+
+: make-kevent ( fd filter flags -- event )
+    "kevent" <c-object>
+    [ set-kevent-flags ] keep
+    [ set-kevent-filter ] keep
+    [ set-kevent-ident ] keep ;
+
+: register-kevent ( kevent mx -- )
+    fd>> swap 1 f 0 f kevent io-error ;
+
+M: kqueue-mx add-input-callback ( thread fd mx -- )
+    [ call-next-method ] [
+        [ EVFILT_READ { EV_ADD EV_ONESHOT } flags make-kevent ] dip
+        register-kevent
+    ] 2bi ;
+
+M: kqueue-mx add-output-callback ( thread fd mx -- )
+    [ call-next-method ] [
+        [ EVFILT_WRITE { EV_ADD EV_ONESHOT } flags make-kevent ] dip
+        register-kevent
+    ] 2bi ;
+
+M: kqueue-mx remove-input-callbacks ( fd mx -- seq )
+    2dup reads>> key? [
+        [ call-next-method ] [
+            [ EVFILT_READ EV_DELETE make-kevent ] dip
+            register-kevent
+        ] 2bi
+    ] [ 2drop f ] if ;
+
+M: kqueue-mx remove-output-callbacks ( fd mx -- seq )
+    2dup writes>> key? [
+        [
+            [ EVFILT_WRITE EV_DELETE make-kevent ] dip
+            register-kevent
+        ] [ call-next-method ] 2bi
+    ] [ 2drop f ] if ;
+
+: wait-kevent ( mx timespec -- n )
+    [
+        [ fd>> f 0 ]
+        [ events>> [ underlying>> ] [ length ] bi ] bi
+    ] dip kevent multiplexer-error ;
+
+: handle-kevent ( mx kevent -- )
+    [ kevent-ident swap ] [ kevent-filter ] bi {
+        { EVFILT_READ [ input-available ] }
+        { EVFILT_WRITE [ output-available ] }
+    } case ;
+
+: handle-kevents ( mx n -- )
+    [ dup events>> ] dip head-slice [ handle-kevent ] with each ;
+
+M: kqueue-mx wait-for-events ( us mx -- )
+    swap dup [ make-timespec ] when
+    dupd wait-kevent handle-kevents ;

basis/io/unix/multiplexers/kqueue/tags.txt

@@ -0,0 +1 @@
+unportable

basis/io/unix/multiplexers/multiplexers.factor

@@ -0,0 +1,35 @@
+! Copyright (C) 2008 Slava Pestov.
+! See http://factorcode.org/license.txt for BSD license.
+USING: kernel accessors assocs sequences threads ;
+IN: io.unix.multiplexers
+
+TUPLE: mx fd reads writes ;
+
+: new-mx ( class -- obj )
+    new
+        H{ } clone >>reads
+        H{ } clone >>writes ; inline
+
+GENERIC: add-input-callback ( thread fd mx -- )
+
+M: mx add-input-callback reads>> push-at ;
+
+GENERIC: add-output-callback ( thread fd mx -- )
+
+M: mx add-output-callback writes>> push-at ;
+
+GENERIC: remove-input-callbacks ( fd mx -- callbacks )
+
+M: mx remove-input-callbacks reads>> delete-at* drop ;
+
+GENERIC: remove-output-callbacks ( fd mx -- callbacks )
+
+M: mx remove-output-callbacks writes>> delete-at* drop ;
+
+GENERIC: wait-for-events ( ms mx -- )
+
+: input-available ( fd mx -- )
+    reads>> delete-at* drop [ resume ] each ;
+
+: output-available ( fd mx -- )
+    writes>> delete-at* drop [ resume ] each ;

basis/io/unix/multiplexers/run-loop/run-loop.factor

@@ -0,0 +1,57 @@
+! Copyright (C) 2008 Slava Pestov.
+! See http://factorcode.org/license.txt for BSD license.
+USING: kernel namespaces math accessors threads alien locals
+destructors combinators core-foundation core-foundation.run-loop
+io.unix.multiplexers io.unix.multiplexers.kqueue ;
+IN: io.unix.multiplexers.run-loop
+
+TUPLE: run-loop-mx kqueue-mx fd source ;
+
+: kqueue-callback ( -- callback )
+    "void" { "CFFileDescriptorRef" "CFOptionFlags" "void*" }
+    "cdecl" [
+        3drop
+        0 mx get kqueue-mx>> wait-for-events
+        mx get fd>> enable-all-callbacks
+        yield
+    ]
+    alien-callback ;
+
+SYMBOL: kqueue-run-loop-source
+
+: create-kqueue-source ( fd -- source )
+    f swap 0 CFFileDescriptorCreateRunLoopSource ;
+
+: add-kqueue-to-run-loop ( mx -- )
+    CFRunLoopGetMain swap source>> CFRunLoopDefaultMode CFRunLoopAddSource ;
+
+: remove-kqueue-from-run-loop ( source -- )
+    CFRunLoopGetMain swap source>> CFRunLoopDefaultMode CFRunLoopRemoveSource ;
+
+: <run-loop-mx> ( -- mx )
+    [
+        <kqueue-mx> |dispose
+        dup fd>> kqueue-callback <CFFileDescriptor> |dispose
+        dup create-kqueue-source run-loop-mx boa
+        dup add-kqueue-to-run-loop
+    ] with-destructors ;
+
+M: run-loop-mx dispose
+    [
+        {
+            [ fd>> &dispose drop ]
+            [ source>> &dispose drop ]
+            [ remove-kqueue-from-run-loop ]
+            [ kqueue-mx>> &dispose drop ]
+        } cleave
+    ] with-destructors ;
+
+M: run-loop-mx add-input-callback kqueue-mx>> add-input-callback ;
+M: run-loop-mx add-output-callback kqueue-mx>> add-output-callback ;
+M: run-loop-mx remove-input-callbacks kqueue-mx>> remove-input-callbacks ;
+M: run-loop-mx remove-output-callbacks kqueue-mx>> remove-output-callbacks ;
+
+M:: run-loop-mx wait-for-events ( us mx -- )
+    mx fd>> enable-all-callbacks
+    CFRunLoopDefaultMode us [ 1000000 /f ] [ 60 ] if* t CFRunLoopRunInMode
+    kCFRunLoopRunHandledSource = [ 0 mx wait-for-events ] when ;

basis/io/unix/multiplexers/run-loop/tags.txt

@@ -0,0 +1 @@
+unportable

basis/io/unix/multiplexers/select/authors.txt

@@ -0,0 +1 @@
+Slava Pestov

basis/io/unix/multiplexers/select/select.factor

@@ -0,0 +1,56 @@
+! Copyright (C) 2004, 2008 Slava Pestov.
+! See http://factorcode.org/license.txt for BSD license.
+USING: alien.c-types kernel bit-arrays sequences assocs unix
+math namespaces accessors math.order locals unix.time fry
+io.ports io.unix.backend io.unix.multiplexers ;
+IN: io.unix.multiplexers.select
+
+TUPLE: select-mx < mx read-fdset write-fdset ;
+
+! Factor's bit-arrays are an array of bytes, OS X expects
+! FD_SET to be an array of cells, so we have to account for
+! byte order differences on big endian platforms
+: munge ( i -- i' )
+    little-endian? [ BIN: 11000 bitxor ] unless ; inline
+
+: <select-mx> ( -- mx )
+    select-mx new-mx
+        FD_SETSIZE 8 * <bit-array> >>read-fdset
+        FD_SETSIZE 8 * <bit-array> >>write-fdset ;
+
+: clear-nth ( n seq -- ? )
+    [ nth ] [ [ f ] 2dip set-nth ] 2bi ;
+
+:: check-fd ( fd fdset mx quot -- )
+    fd munge fdset clear-nth [ fd mx quot call ] when ; inline
+
+: check-fdset ( fds fdset mx quot -- )
+    [ check-fd ] 3curry each ; inline
+
+: init-fdset ( fds fdset -- )
+    '[ t swap munge _ set-nth ] each ;
+
+: read-fdset/tasks ( mx -- seq fdset )
+    [ reads>> keys ] [ read-fdset>> ] bi ;
+
+: write-fdset/tasks ( mx -- seq fdset )
+    [ writes>> keys ] [ write-fdset>> ] bi ;
+
+: max-fd ( assoc -- n )
+    dup assoc-empty? [ drop 0 ] [ keys supremum ] if ;
+
+: num-fds ( mx -- n )
+    [ reads>> max-fd ] [ writes>> max-fd ] bi max 1+ ;
+
+: init-fdsets ( mx -- nfds read write except )
+    [ num-fds ]
+    [ read-fdset/tasks [ init-fdset ] [ underlying>> ] bi ]
+    [ write-fdset/tasks [ init-fdset ] [ underlying>> ] bi ] tri
+    f ;
+
+M:: select-mx wait-for-events ( us mx -- )
+    mx
+    [ init-fdsets us dup [ make-timeval ] when select multiplexer-error drop ]
+    [ [ read-fdset/tasks ] keep [ input-available ] check-fdset ]
+    [ [ write-fdset/tasks ] keep [ output-available ] check-fdset ]
+    tri ;

basis/io/unix/multiplexers/select/tags.txt

@@ -0,0 +1 @@
+unportable