1 ! Copyright (C) 2004, 2008 Slava Pestov.
2 ! See http://factorcode.org/license.txt for BSD license.
3 USING: alien alien.c-types alien.data alien.syntax generic
4 assocs kernel kernel.private math io.ports sequences strings
5 sbufs threads unix unix.ffi unix.stat vectors io.buffers io.backend
6 io.encodings math.parser continuations system libc namespaces
7 make io.timeouts io.encodings.utf8 destructors
8 destructors.private accessors summary combinators locals
9 unix.time unix.types fry io.backend.unix.multiplexers
10 classes.struct hints ;
14 GENERIC: handle-fd ( handle -- fd )
16 TUPLE: fd < disposable fd ;
18 : init-fd ( fd -- fd )
21 dup fd>> F_SETFL O_NONBLOCK [ fcntl ] unix-system-call drop
22 dup fd>> F_SETFD FD_CLOEXEC [ fcntl ] unix-system-call drop
26 fd new-disposable swap >>fd ;
33 [ unregister-disposable ]
38 M: fd handle-fd dup check-disposed fd>> ;
40 M: fd cancel-operation ( fd -- )
44 [ remove-input-callbacks [ t swap resume-with ] each ]
45 [ remove-output-callbacks [ t swap resume-with ] each ]
49 M: unix tell-handle ( handle -- n )
50 fd>> 0 SEEK_CUR [ lseek ] unix-system-call [ io-error ] [ ] bi ;
52 M: unix seek-handle ( n seek-type handle -- )
54 { io:seek-absolute [ SEEK_SET ] }
55 { io:seek-relative [ SEEK_CUR ] }
56 { io:seek-end [ SEEK_END ] }
59 [ fd>> swap ] dip [ lseek ] unix-system-call drop ;
61 M: unix can-seek-handle? ( handle -- ? )
62 fd>> SEEK_CUR 0 lseek -1 = not ;
64 M: unix handle-length ( handle -- n/f )
65 fd>> \ stat <struct> [ fstat -1 = not ] keep
66 swap [ st_size>> ] [ drop f ] if ;
68 SYMBOL: +retry+ ! just try the operation again without blocking
74 M: io-timeout summary drop "I/O operation timed out" ;
76 : wait-for-fd ( handle event -- )
77 dup +retry+ eq? [ 2drop ] [
78 [ [ self ] dip handle-fd mx get-global ] dip {
79 { +input+ [ add-input-callback ] }
80 { +output+ [ add-output-callback ] }
82 "I/O" suspend [ io-timeout ] when
85 : wait-for-port ( port event -- )
86 '[ handle>> _ wait-for-fd ] with-timeout ;
89 CONSTANT: file-mode 0o0666
92 : (refill) ( port -- n )
94 [ buffer>> buffer-end ]
95 [ buffer>> buffer-capacity ] tri read ;
97 ! Returns an event to wait for which will ensure completion of
99 GENERIC: refill ( port handle -- event/f )
102 fd>> over buffer>> [ buffer-end ] [ buffer-capacity ] bi read
104 { [ dup 0 >= ] [ swap buffer>> n>buffer f ] }
105 { [ errno EINTR = ] [ 2drop +retry+ ] }
106 { [ errno EAGAIN = ] [ 2drop +input+ ] }
111 { buffered-port fd } ;
113 M: unix (wait-to-read) ( port -- )
115 dup handle>> dup check-disposed refill dup
116 [ dupd wait-for-port (wait-to-read) ] [ 2drop ] if ;
119 GENERIC: drain ( port handle -- event/f )
122 fd>> over buffer>> [ buffer@ ] [ buffer-length ] bi write
125 over buffer>> buffer-consume
126 buffer>> buffer-empty? f +output+ ?
128 { [ errno EINTR = ] [ 2drop +retry+ ] }
129 { [ errno EAGAIN = ] [ 2drop +output+ ] }
133 M: unix (wait-to-write) ( port -- )
135 dup handle>> dup check-disposed drain
136 dup [ wait-for-port ] [ 2drop ] if ;
138 M: unix io-multiplex ( ms/f -- )
139 mx get-global wait-for-events ;
141 ! On Unix, you're not supposed to set stdin to non-blocking
142 ! because the fd might be shared with another process (either
143 ! parent or child). So what we do is have the VM start a thread
144 ! which pumps data from the real stdin to a pipe. We set the
145 ! pipe to non-blocking, and read from it instead of the real
146 ! stdin. Very crufty, but it will suffice until we get native
147 ! threading support at the language level.
148 TUPLE: stdin < disposable control size data ;
152 [ control>> &dispose drop ]
153 [ size>> &dispose drop ]
154 [ data>> &dispose drop ]
158 : wait-for-stdin ( stdin -- size )
159 [ control>> CHAR: X over io:stream-write1 io:stream-flush ]
160 [ size>> ssize_t heap-size swap io:stream-read ssize_t deref ]
163 :: refill-stdin ( buffer stdin size -- )
164 stdin data>> handle-fd buffer buffer-end size read
167 errno EINTR = [ buffer stdin size refill-stdin ] [ (io-error) ] if
169 size = [ "Error reading stdin pipe" throw ] unless
175 buffer>> _ dup wait-for-stdin refill-stdin f
178 M: stdin cancel-operation
179 [ size>> ] [ control>> ] bi [ cancel-operation ] bi@ ;
181 : control-write-fd ( -- fd ) &: control_write uint deref ;
183 : size-read-fd ( -- fd ) &: size_read uint deref ;
185 : data-read-fd ( -- fd ) &: stdin_read uint deref ;
187 : <stdin> ( -- stdin )
189 control-write-fd <fd> <output-port> >>control
190 size-read-fd <fd> init-fd <input-port> >>size
191 data-read-fd <fd> >>data ;
193 SYMBOL: dispatch-signal-hook
195 dispatch-signal-hook [ [ drop ] ] initialize
197 : signal-pipe-fd ( -- n )
198 OBJ-SIGNAL-PIPE special-object ; inline
200 : signal-pipe-loop ( port -- )
202 int heap-size _ io:stream-read
203 dup [ int deref dispatch-signal-hook get call( x -- ) ] when*
206 : start-signal-pipe-thread ( -- )
208 <fd> init-fd <input-port>
209 '[ _ signal-pipe-loop ] "Signals" spawn drop
218 ! mx io-task for embedding an fd-based mx inside another mx
219 TUPLE: mx-port < port mx ;
221 : <mx-port> ( mx -- port )
222 dup fd>> mx-port <port> swap >>mx ;
224 : multiplexer-error ( n -- n )
226 errno [ EAGAIN = ] [ EINTR = ] bi or
227 [ drop 0 ] [ (io-error) ] if
230 :: ?flag ( n mask symbol -- n )
231 n mask bitand 0 > [ symbol , ] when n ;