Convert native I/O to use inheritance

db4
Slava Pestov 2008-04-11 16:08:40 -05:00
parent 1625511df7
commit e794466efb
13 changed files with 100 additions and 115 deletions

View File

@ -129,9 +129,6 @@ HELP: <process>
{ $values { "process" process } }
{ $description "Creates a new, empty process. It must be filled in before being passed to " { $link run-process } "." } ;
HELP: process-stream
{ $class-description "A bidirectional stream for interacting with a running process. Instances are created by calling " { $link <process-stream> } ". The " { $link process-stream-process } " slot holds a " { $link process } " instance." } ;
HELP: <process-stream>
{ $values
{ "desc" "a launch descriptor" }
@ -144,7 +141,7 @@ HELP: with-process-stream
{ "desc" "a launch descriptor" }
{ "quot" quotation }
{ "status" "an exit code" } }
{ $description "Calls " { $snippet "quot" } " in a dynamic scope where " { $link stdio } " is rebound to a " { $link process-stream } ". After the quotation returns, waits for the process to end and outputs the exit code." } ;
{ $description "Calls " { $snippet "quot" } " in a dynamic scope where " { $link stdio } " is rebound to a process stream. After the quotation returns, waits for the process to end and outputs the exit code." } ;
HELP: wait-for-process
{ $values { "process" process } { "status" integer } }

View File

@ -150,18 +150,18 @@ M: process timed-out kill-process ;
HOOK: (process-stream) io-backend ( process -- handle in out )
TUPLE: process-stream process ;
: <process-stream*> ( desc encoding -- stream process )
>r >process dup dup (process-stream) <reader&writer>
r> <encoder-duplex> -roll
process-started ;
: <process-stream> ( desc encoding -- stream )
>r >process dup dup (process-stream)
>r >r process-started process-stream construct-boa
r> r> <reader&writer> r> <encoder-duplex>
over set-delegate ;
<process-stream*> drop ; inline
: with-process-stream ( desc quot -- status )
swap <process-stream>
swap <process-stream*> >r
[ swap with-stream ] keep
process>> wait-for-process ; inline
r> wait-for-process ; inline
: notify-exit ( process status -- )
>>status

View File

@ -36,10 +36,10 @@ HELP: port
$nl
"Ports have the following slots:"
{ $list
{ { $link port-handle } " - a native handle identifying the underlying native resource used by the port" }
{ { $link port-error } " - the most recent I/O error, if any. This error is thrown to the waiting thread when " { $link pending-error } " is called by stream operations" }
{ { $link port-type } " - a symbol identifying the port's intended purpose" }
{ { $link port-eof? } " - a flag indicating if the port has reached the end of file while reading" }
{ { $snippet "handle" } " - a native handle identifying the underlying native resource used by the port" }
{ { $snippet "error" } " - the most recent I/O error, if any. This error is thrown to the waiting thread when " { $link pending-error } " is called by stream operations" }
{ { $snippet "type" } " - a symbol identifying the port's intended purpose" }
{ { $snippet "eof" } " - a flag indicating if the port has reached the end of file while reading" }
} } ;
HELP: input-port

View File

@ -1,46 +1,39 @@
! Copyright (C) 2005, 2008 Slava Pestov, Doug Coleman
! See http://factorcode.org/license.txt for BSD license.
IN: io.nonblocking
USING: math kernel io sequences io.buffers io.timeouts generic
byte-vectors system io.streams.duplex io.encodings
io.backend continuations debugger classes byte-arrays namespaces
splitting dlists assocs io.encodings.binary accessors ;
IN: io.nonblocking
SYMBOL: default-buffer-size
64 1024 * default-buffer-size set-global
! Common delegate of native stream readers and writers
TUPLE: port
handle
buffer
error
timeout
type eof ;
TUPLE: port handle buffer error timeout closed eof ;
M: port timeout port-timeout ;
M: port timeout timeout>> ;
M: port set-timeout set-port-timeout ;
SYMBOL: closed
PREDICATE: input-port < port port-type input-port eq? ;
PREDICATE: output-port < port port-type output-port eq? ;
M: port set-timeout (>>timeout) ;
GENERIC: init-handle ( handle -- )
GENERIC: close-handle ( handle -- )
: <port> ( handle type -- port )
port construct-empty
swap >>type
swap dup init-handle >>handle ;
: <port> ( handle class -- port )
construct-empty
swap dup init-handle >>handle ; inline
: <buffered-port> ( handle type -- port )
: <buffered-port> ( handle class -- port )
<port>
default-buffer-size get <buffer> >>buffer ;
default-buffer-size get <buffer> >>buffer ; inline
TUPLE: input-port < port ;
: <reader> ( handle -- input-port )
input-port <buffered-port> ;
TUPLE: output-port < port ;
: <writer> ( handle -- output-port )
output-port <buffered-port> ;
@ -50,6 +43,9 @@ GENERIC: close-handle ( handle -- )
: pending-error ( port -- )
[ f ] change-error drop [ throw ] when* ;
: check-closed ( port -- port )
dup closed>> [ "Port closed" throw ] when ;
HOOK: cancel-io io-backend ( port -- )
M: object cancel-io drop ;
@ -69,6 +65,7 @@ GENERIC: (wait-to-read) ( port -- )
[ f >>eof drop f ] r> if ; inline
M: input-port stream-read1
check-closed
dup wait-to-read1 [ buffer>> buffer-pop ] unless-eof ;
: read-step ( count port -- byte-array/f )
@ -87,6 +84,7 @@ M: input-port stream-read1
] if ;
M: input-port stream-read
check-closed
>r 0 max >fixnum r>
2dup read-step dup [
pick over length > [
@ -98,6 +96,7 @@ M: input-port stream-read
] [ 2nip ] if ;
M: input-port stream-read-partial ( max stream -- byte-array/f )
check-closed
>r 0 max >fixnum r> read-step ;
: can-write? ( len buffer -- ? )
@ -107,10 +106,12 @@ M: input-port stream-read-partial ( max stream -- byte-array/f )
tuck buffer>> can-write? [ drop ] [ stream-flush ] if ;
M: output-port stream-write1
check-closed
1 over wait-to-write
buffer>> byte>buffer ;
M: output-port stream-write
check-closed
over length over buffer>> buffer-size > [
[ buffer>> buffer-size <groups> ]
[ [ stream-write ] curry ] bi
@ -123,41 +124,43 @@ M: output-port stream-write
GENERIC: port-flush ( port -- )
M: output-port stream-flush ( port -- )
check-closed
[ port-flush ] [ pending-error ] bi ;
: close-port ( port type -- )
output-port eq? [ dup port-flush ] when
GENERIC: close-port ( port -- )
M: output-port close-port
[ port-flush ] [ call-next-method ] bi ;
M: port close-port
dup cancel-io
dup handle>> close-handle
[ [ buffer-free ] when* f ] change-buffer drop ;
M: port dispose
dup type>> closed eq?
[ drop ]
[ [ closed ] change-type swap close-port ]
if ;
dup closed>> [ drop ] [ t >>closed close-port ] if ;
TUPLE: server-port addr client client-addr encoding ;
TUPLE: server-port < port addr client client-addr encoding ;
: <server-port> ( handle addr encoding -- server )
rot server-port <port>
{ set-server-port-addr set-server-port-encoding set-delegate }
server-port construct ;
swap >>encoding
swap >>addr ;
: check-server-port ( port -- )
port-type server-port assert= ;
: check-server-port ( port -- port )
dup server-port? [ "Not a server port" throw ] unless ; inline
TUPLE: datagram-port addr packet packet-addr ;
TUPLE: datagram-port < port addr packet packet-addr ;
: <datagram-port> ( handle addr -- datagram )
>r datagram-port <port> r>
{ set-delegate set-datagram-port-addr }
datagram-port construct ;
swap datagram-port <port>
swap >>addr ;
: check-datagram-port ( port -- )
port-type datagram-port assert= ;
: check-datagram-port ( port -- port )
check-closed
dup datagram-port? [ "Not a datagram port" throw ] unless ; inline
: check-datagram-send ( packet addrspec port -- )
dup check-datagram-port
datagram-port-addr [ class ] bi@ assert=
class byte-array assert= ;
: check-datagram-send ( packet addrspec port -- packet addrspec port )
check-datagram-port
2dup addr>> [ class ] bi@ assert=
pick class byte-array assert= ;

View File

@ -12,17 +12,17 @@ SYMBOL: servers
LOG: accepted-connection NOTICE
: with-client ( client quot -- )
: with-client ( client addrspec quot -- )
[
over client-stream-addr accepted-connection
swap accepted-connection
with-stream*
] curry with-disposal ; inline
] 2curry with-disposal ; inline
\ with-client DEBUG add-error-logging
: accept-loop ( server quot -- )
[
>r accept r> [ with-client ] 2curry "Client" spawn drop
>r accept r> [ with-client ] 3curry "Client" spawn drop
] 2keep accept-loop ; inline
: server-loop ( addrspec encoding quot -- )

View File

@ -17,8 +17,6 @@ ARTICLE: "network-connection" "Connection-oriented networking"
"Connection-oriented network servers are implemented by first opening a server socket, then waiting for connections:"
{ $subsection <server> }
{ $subsection accept }
"The stream returned by " { $link accept } " holds the address specifier of the remote client:"
{ $subsection client-stream-addr }
"Server sockets are closed by calling " { $link dispose } "."
$nl
"Address specifiers have the following interpretation with connection-oriented networking words:"
@ -118,10 +116,8 @@ HELP: <server>
{ $errors "Throws an error if the address is already in use, or if it if the system forbids access." } ;
HELP: accept
{ $values { "server" "a handle" } { "client" "a bidirectional stream" } }
{ $description "Waits for a connection to a server socket created by " { $link <server> } ", and outputs a bidirectional stream when the connection has been established. The encoding of this stream is the one that was passed to the server constructor."
$nl
"The returned client stream responds to the " { $link client-stream-addr } " word with the address of the incoming connection." }
{ $values { "server" "a handle" } { "client" "a bidirectional stream" } { "addrspec" "an address specifier" } }
{ $description "Waits for a connection to a server socket created by " { $link <server> } ", and outputs a bidirectional stream when the connection has been established. The encoding of this stream is the one that was passed to the server constructor." }
{ $errors "Throws an error if the server socket is closed or otherwise is unavailable." } ;
HELP: <datagram>

View File

@ -1,7 +1,7 @@
! Copyright (C) 2007, 2008 Slava Pestov, Daniel Ehrenberg.
! See http://factorcode.org/license.txt for BSD license.
USING: generic kernel io.backend namespaces continuations
sequences arrays io.encodings io.nonblocking ;
sequences arrays io.encodings io.nonblocking accessors ;
IN: io.sockets
TUPLE: local path ;
@ -21,20 +21,14 @@ TUPLE: inet host port ;
C: <inet> inet
TUPLE: client-stream addr ;
HOOK: ((client)) io-backend ( addrspec -- client-in client-out )
: <client-stream> ( addrspec delegate -- stream )
{ set-client-stream-addr set-delegate }
client-stream construct ;
HOOK: (client) io-backend ( addrspec -- client-in client-out )
GENERIC: client* ( addrspec -- client-in client-out )
M: array client* [ (client) 2array ] attempt-all first2 ;
M: object client* (client) ;
GENERIC: (client) ( addrspec -- client-in client-out )
M: array (client) [ ((client)) 2array ] attempt-all first2 ;
M: object (client) ((client)) ;
: <client> ( addrspec encoding -- stream )
>r client* r> <encoder-duplex> ;
>r (client) r> <encoder-duplex> ;
HOOK: (server) io-backend ( addrspec -- handle )
@ -43,10 +37,9 @@ HOOK: (server) io-backend ( addrspec -- handle )
HOOK: (accept) io-backend ( server -- addrspec handle )
: accept ( server -- client )
[ (accept) dup <reader&writer> ] keep
server-port-encoding <encoder-duplex>
<client-stream> ;
: accept ( server -- client addrspec )
[ (accept) dup <reader&writer> ] [ encoding>> ] bi
<encoder-duplex> swap ;
HOOK: <datagram> io-backend ( addrspec -- datagram )
@ -58,7 +51,8 @@ HOOK: resolve-host io-backend ( host serv passive? -- seq )
HOOK: host-name io-backend ( -- string )
M: inet client*
dup inet-host swap inet-port f resolve-host
dup empty? [ "Host name lookup failed" throw ] when
client* ;
M: inet (client)
[ host>> ] [ port>> ] bi f resolve-host
[ empty? [ "Host name lookup failed" throw ] when ]
[ (client) ]
bi ;

View File

@ -178,7 +178,7 @@ M: write-task do-io-task
: (wait-to-write) ( port -- )
[ add-write-io-task ] with-port-continuation drop ;
M: port port-flush ( port -- )
M: output-port port-flush ( port -- )
dup buffer>> buffer-empty? [ drop ] [ (wait-to-write) ] if ;
M: unix io-multiplex ( ms/f -- )
@ -190,11 +190,10 @@ M: unix (init-stdio) ( -- )
2 <writer> ;
! mx io-task for embedding an fd-based mx inside another mx
TUPLE: mx-port mx ;
TUPLE: mx-port < port mx ;
: <mx-port> ( mx -- port )
dup fd>> mx-port <port>
{ set-mx-port-mx set-delegate } mx-port construct ;
dup fd>> mx-port <port> swap >>mx ;
TUPLE: mx-task < io-task ;

View File

@ -7,7 +7,7 @@ USING: alien alien.c-types generic io kernel math namespaces
io.nonblocking parser threads unix sequences
byte-arrays io.sockets io.binary io.unix.backend
io.streams.duplex io.sockets.impl math.parser continuations libc
combinators io.backend io.files io.files.private system ;
combinators io.backend io.files io.files.private system accessors ;
IN: io.unix.sockets
: pending-init-error ( port -- )
@ -42,7 +42,7 @@ M: connect-task do-io-task
: wait-to-connect ( port -- )
[ <connect-task> add-io-task ] with-port-continuation drop ;
M: unix (client) ( addrspec -- client-in client-out )
M: unix ((client)) ( addrspec -- client-in client-out )
dup make-sockaddr/size >r >r
protocol-family SOCK_STREAM socket-fd
dup r> r> connect
@ -97,11 +97,10 @@ M: unix (server) ( addrspec -- handle )
M: unix (accept) ( server -- addrspec handle )
#! Wait for a client connection.
dup check-server-port
dup wait-to-accept
dup pending-error
dup server-port-client-addr
swap server-port-client ;
check-server-port
[ wait-to-accept ]
[ pending-error ]
[ [ client-addr>> ] [ client>> ] bi ] tri ;
! Datagram sockets - UDP and Unix domain
M: unix <datagram>
@ -148,11 +147,10 @@ M: receive-task do-io-task
[ <receive-task> add-io-task ] with-port-continuation drop ;
M: unix receive ( datagram -- packet addrspec )
dup check-datagram-port
dup wait-receive
dup pending-error
dup datagram-port-packet
swap datagram-port-packet-addr ;
check-datagram-port
[ wait-receive ]
[ pending-error ]
[ [ packet>> ] [ packet-addr>> ] bi ] tri ;
: do-send ( socket data sockaddr len -- n )
>r >r dup length 0 r> r> sendto ;
@ -180,7 +178,7 @@ M: send-task do-io-task
2drop 2drop ;
M: unix send ( packet addrspec datagram -- )
3dup check-datagram-send
check-datagram-send
[ >r make-sockaddr/size r> wait-send ] keep
pending-error ;

View File

@ -11,7 +11,7 @@ IN: io.unix.tests
socket-server <local>
ascii <server> [
accept [
accept drop [
"Hello world" print flush
readln "XYZ" = "FOO" "BAR" ? print flush
] with-stream

View File

@ -122,7 +122,7 @@ TUPLE: AcceptEx-args port
M: winnt (accept) ( server -- addrspec handle )
[
[
dup check-server-port
check-server-port
\ AcceptEx-args construct-empty
[ init-accept ] keep
[ ((accept)) ] keep
@ -159,7 +159,7 @@ TUPLE: WSARecvFrom-args port
: init-WSARecvFrom ( datagram WSARecvFrom -- )
[ set-WSARecvFrom-args-port ] 2keep
[
>r delegate port-handle delegate win32-file-handle r>
>r handle>> handle>> r>
set-WSARecvFrom-args-s*
] 2keep [
>r datagram-port-addr sockaddr-type heap-size r>
@ -192,7 +192,7 @@ TUPLE: WSARecvFrom-args port
M: winnt receive ( datagram -- packet addrspec )
[
dup check-datagram-port
check-datagram-port
\ WSARecvFrom-args construct-empty
[ init-WSARecvFrom ] keep
[ call-WSARecvFrom ] keep
@ -244,7 +244,7 @@ USE: io.sockets
M: winnt send ( packet addrspec datagram -- )
[
3dup check-datagram-send
check-datagram-send
\ WSASendTo-args construct-empty
[ init-WSASendTo ] keep
[ call-WSASendTo ] keep

View File

@ -152,11 +152,10 @@ M: windows delete-directory ( path -- )
HOOK: WSASocket-flags io-backend ( -- DWORD )
TUPLE: win32-socket ;
TUPLE: win32-socket < win32-file ;
: <win32-socket> ( handle -- win32-socket )
f <win32-file>
\ win32-socket construct-delegate ;
f win32-file construct-boa ;
: open-socket ( family type -- socket )
0 f 0 WSASocket-flags WSASocket dup socket-error ;

View File

@ -22,9 +22,8 @@ IN: tools.deploy.backend
+stdout+ >>stderr
+closed+ >>stdin
+low-priority+ >>priority
utf8 <process-stream>
dup copy-lines
process>> wait-for-process zero? [
utf8 <process-stream*>
>r copy-lines r> wait-for-process zero? [
"Deployment failed" throw
] unless ;