basis/io/unix/backend/backend.factor

   1 ! Copyright (C) 2004, 2008 Slava Pestov.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: alien alien.c-types generic assocs kernel kernel.private
   4 math io.ports sequences strings structs sbufs threads unix
   5 vectors io.buffers io.backend io.encodings math.parser
   6 continuations system libc qualified namespaces make io.timeouts
   7 io.encodings.utf8 destructors accessors summary combinators
   8 locals ;
   9 QUALIFIED: io
  10 IN: io.unix.backend
  11
  12 GENERIC: handle-fd ( handle -- fd )
  13
  14 TUPLE: fd fd disposed ;
  15
  16 : init-fd ( fd -- fd )
  17     [
  18         |dispose
  19         dup fd>> F_SETFL O_NONBLOCK fcntl io-error
  20         dup fd>> F_SETFD FD_CLOEXEC fcntl io-error
  21     ] with-destructors ;
  22
  23 : <fd> ( n -- fd )
  24     #! We drop the error code rather than calling io-error,
  25     #! since on OS X 10.3, this operation fails from init-io
  26     #! when running the Factor.app (presumably because fd 0 and
  27     #! 1 are closed).
  28     f fd boa ;
  29
  30 M: fd dispose
  31     dup disposed>> [ drop ] [
  32         [ cancel-operation ]
  33         [ t >>disposed drop ]
  34         [ fd>> close-file ]
  35         tri
  36     ] if ;
  37
  38 M: fd handle-fd dup check-disposed fd>> ;
  39
  40 ! I/O multiplexers
  41 TUPLE: mx fd reads writes ;
  42
  43 : new-mx ( class -- obj )
  44     new
  45         H{ } clone >>reads
  46         H{ } clone >>writes ; inline
  47
  48 GENERIC: add-input-callback ( thread fd mx -- )
  49
  50 M: mx add-input-callback reads>> push-at ;
  51
  52 GENERIC: add-output-callback ( thread fd mx -- )
  53
  54 M: mx add-output-callback writes>> push-at ;
  55
  56 GENERIC: remove-input-callbacks ( fd mx -- callbacks )
  57
  58 M: mx remove-input-callbacks reads>> delete-at* drop ;
  59
  60 GENERIC: remove-output-callbacks ( fd mx -- callbacks )
  61
  62 M: mx remove-output-callbacks writes>> delete-at* drop ;
  63
  64 GENERIC: wait-for-events ( ms mx -- )
  65
  66 : input-available ( fd mx -- )
  67     remove-input-callbacks [ resume ] each ;
  68
  69 : output-available ( fd mx -- )
  70     remove-output-callbacks [ resume ] each ;
  71
  72 M: fd cancel-operation ( fd -- )
  73     dup disposed>> [ drop ] [
  74         fd>>
  75         mx get-global
  76         [ remove-input-callbacks [ t swap resume-with ] each ]
  77         [ remove-output-callbacks [ t swap resume-with ] each ]
  78         2bi
  79     ] if ;
  80
  81 SYMBOL: +retry+ ! just try the operation again without blocking
  82 SYMBOL: +input+
  83 SYMBOL: +output+
  84
  85 ERROR: io-timeout ;
  86
  87 M: io-timeout summary drop "I/O operation timed out" ;
  88
  89 : wait-for-fd ( handle event -- )
  90     dup +retry+ eq? [ 2drop ] [
  91         [
  92             >r
  93             swap handle-fd
  94             mx get-global
  95             r> {
  96                 { +input+ [ add-input-callback ] }
  97                 { +output+ [ add-output-callback ] }
  98             } case
  99         ] curry "I/O" suspend nip [ io-timeout ] when
 100     ] if ;
 101
 102 : wait-for-port ( port event -- )
 103     [ >r handle>> r> wait-for-fd ] curry with-timeout ;
 104
 105 ! Some general stuff
 106 : file-mode OCT: 0666 ;
 107
 108 ! Readers
 109 : (refill) ( port -- n )
 110     [ handle>> ]
 111     [ buffer>> buffer-end ]
 112     [ buffer>> buffer-capacity ] tri read ;
 113
 114 ! Returns an event to wait for which will ensure completion of
 115 ! this request
 116 GENERIC: refill ( port handle -- event/f )
 117
 118 M: fd refill
 119     fd>> over buffer>> [ buffer-end ] [ buffer-capacity ] bi read
 120     {
 121         { [ dup 0 >= ] [ swap buffer>> n>buffer f ] }
 122         { [ err_no EINTR = ] [ 2drop +retry+ ] }
 123         { [ err_no EAGAIN = ] [ 2drop +input+ ] }
 124         [ (io-error) ]
 125     } cond ;
 126
 127 M: unix (wait-to-read) ( port -- )
 128     dup
 129     dup handle>> dup check-disposed refill dup
 130     [ dupd wait-for-port (wait-to-read) ] [ 2drop ] if ;
 131
 132 ! Writers
 133 GENERIC: drain ( port handle -- event/f )
 134
 135 M: fd drain
 136     fd>> over buffer>> [ buffer@ ] [ buffer-length ] bi write
 137     {
 138         { [ dup 0 >= ] [
 139             over buffer>> buffer-consume
 140             buffer>> buffer-empty? f +output+ ?
 141         ] }
 142         { [ err_no EINTR = ] [ 2drop +retry+ ] }
 143         { [ err_no EAGAIN = ] [ 2drop +output+ ] }
 144         [ (io-error) ]
 145     } cond ;
 146
 147 M: unix (wait-to-write) ( port -- )
 148     dup
 149     dup handle>> dup check-disposed drain
 150     dup [ wait-for-port ] [ 2drop ] if ;
 151
 152 M: unix io-multiplex ( ms/f -- )
 153     mx get-global wait-for-events ;
 154
 155 ! On Unix, you're not supposed to set stdin to non-blocking
 156 ! because the fd might be shared with another process (either
 157 ! parent or child). So what we do is have the VM start a thread
 158 ! which pumps data from the real stdin to a pipe. We set the
 159 ! pipe to non-blocking, and read from it instead of the real
 160 ! stdin. Very crufty, but it will suffice until we get native
 161 ! threading support at the language level.
 162 TUPLE: stdin control size data disposed ;
 163
 164 M: stdin dispose*
 165     [
 166         [ control>> &dispose drop ]
 167         [ size>> &dispose drop ]
 168         [ data>> &dispose drop ]
 169         tri
 170     ] with-destructors ;
 171
 172 : wait-for-stdin ( stdin -- n )
 173     [ control>> CHAR: X over io:stream-write1 io:stream-flush ]
 174     [ size>> "ssize_t" heap-size swap io:stream-read *int ]
 175     bi ;
 176
 177 :: refill-stdin ( buffer stdin size -- )
 178     stdin data>> handle-fd buffer buffer-end size read
 179     dup 0 < [
 180         drop
 181         err_no EINTR = [ buffer stdin size refill-stdin ] [ (io-error) ] if
 182     ] [
 183         size = [ "Error reading stdin pipe" throw ] unless
 184         size buffer n>buffer
 185     ] if ;
 186
 187 M: stdin refill
 188     [ buffer>> ] [ dup wait-for-stdin ] bi* refill-stdin f ;
 189
 190 : control-write-fd ( -- fd ) "control_write" f dlsym *uint ;
 191
 192 : size-read-fd ( -- fd ) "size_read" f dlsym *uint ;
 193
 194 : data-read-fd ( -- fd ) "stdin_read" f dlsym *uint ;
 195
 196 : <stdin> ( -- stdin )
 197     stdin new
 198         control-write-fd <fd> <output-port> >>control
 199         size-read-fd <fd> init-fd <input-port> >>size
 200         data-read-fd <fd> >>data ;
 201
 202 M: unix (init-stdio) ( -- )
 203     <stdin> <input-port>
 204     1 <fd> <output-port>
 205     2 <fd> <output-port> ;
 206
 207 ! mx io-task for embedding an fd-based mx inside another mx
 208 TUPLE: mx-port < port mx ;
 209
 210 : <mx-port> ( mx -- port )
 211     dup fd>> mx-port <port> swap >>mx ;
 212
 213 : multiplexer-error ( n -- )
 214     0 < [
 215         err_no [ EAGAIN = ] [ EINTR = ] bi or
 216         [ (io-error) ] unless
 217     ] when ;
 218
 219 : ?flag ( n mask symbol -- n )
 220     pick rot bitand 0 > [ , ] [ drop ] if ;