basis/io/backend/unix/unix.factor

   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 init ;
  11 QUALIFIED: io
  12 IN: io.backend.unix
  13
  14 GENERIC: handle-fd ( handle -- fd )
  15
  16 TUPLE: fd < disposable fd ;
  17
  18 : init-fd ( fd -- fd )
  19     [
  20         |dispose
  21         dup fd>> F_SETFL O_NONBLOCK [ fcntl ] unix-system-call drop
  22         dup fd>> F_SETFD FD_CLOEXEC [ fcntl ] unix-system-call drop
  23     ] with-destructors ;
  24
  25 : <fd> ( n -- fd )
  26     fd new-disposable swap >>fd ;
  27
  28 M: fd dispose
  29     [
  30         {
  31             [ cancel-operation ]
  32             [ t >>disposed drop ]
  33             [ unregister-disposable ]
  34             [ fd>> close-file ]
  35         } cleave
  36     ] unless-disposed ;
  37
  38 M: fd handle-fd dup check-disposed fd>> ;
  39
  40 M: fd cancel-operation ( fd -- )
  41     [
  42         fd>>
  43         mx get-global
  44         [ remove-input-callbacks [ t swap resume-with ] each ]
  45         [ remove-output-callbacks [ t swap resume-with ] each ]
  46         2bi
  47     ] unless-disposed ;
  48
  49 M: unix tell-handle ( handle -- n )
  50     fd>> 0 SEEK_CUR [ lseek ] unix-system-call [ io-error ] [ ] bi ;
  51
  52 M: unix seek-handle ( n seek-type handle -- )
  53     swap {
  54         { io:seek-absolute [ SEEK_SET ] }
  55         { io:seek-relative [ SEEK_CUR ] }
  56         { io:seek-end [ SEEK_END ] }
  57         [ io:bad-seek-type ]
  58     } case
  59     [ fd>> swap ] dip [ lseek ] unix-system-call drop ;
  60
  61 M: unix can-seek-handle? ( handle -- ? )
  62     fd>> SEEK_CUR 0 lseek -1 = not ;
  63
  64 M: unix handle-length ( handle -- n/f )
  65     fd>> \ stat <struct> [ fstat -1 = not ] keep
  66     swap [ st_size>> ] [ drop f ] if ;
  67
  68 SYMBOL: +retry+ ! just try the operation again without blocking
  69 SYMBOL: +input+
  70 SYMBOL: +output+
  71
  72 ERROR: io-timeout ;
  73
  74 M: io-timeout summary drop "I/O operation timed out" ;
  75
  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 ] }
  81         } case
  82         "I/O" suspend [ io-timeout ] when
  83     ] if ;
  84
  85 : wait-for-port ( port event -- )
  86     '[ handle>> _ wait-for-fd ] with-timeout ;
  87
  88 ! Some general stuff
  89 CONSTANT: file-mode OCT: 0666
  90
  91 ! Readers
  92 : (refill) ( port -- n )
  93     [ handle>> ]
  94     [ buffer>> buffer-end ]
  95     [ buffer>> buffer-capacity ] tri read ;
  96
  97 ! Returns an event to wait for which will ensure completion of
  98 ! this request
  99 GENERIC: refill ( port handle -- event/f )
 100
 101 M: fd refill
 102     fd>> over buffer>> [ buffer-end ] [ buffer-capacity ] bi read
 103     {
 104         { [ dup 0 >= ] [ swap buffer>> n>buffer f ] }
 105         { [ errno EINTR = ] [ 2drop +retry+ ] }
 106         { [ errno EAGAIN = ] [ 2drop +input+ ] }
 107         [ (io-error) ]
 108     } cond ;
 109
 110 M: unix (wait-to-read) ( port -- )
 111     dup
 112     dup handle>> dup check-disposed refill dup
 113     [ dupd wait-for-port (wait-to-read) ] [ 2drop ] if ;
 114
 115 ! Writers
 116 GENERIC: drain ( port handle -- event/f )
 117
 118 M: fd drain
 119     fd>> over buffer>> [ buffer@ ] [ buffer-length ] bi write
 120     {
 121         { [ dup 0 >= ] [
 122             over buffer>> buffer-consume
 123             buffer>> buffer-empty? f +output+ ?
 124         ] }
 125         { [ errno EINTR = ] [ 2drop +retry+ ] }
 126         { [ errno EAGAIN = ] [ 2drop +output+ ] }
 127         [ (io-error) ]
 128     } cond ;
 129
 130 M: unix (wait-to-write) ( port -- )
 131     dup
 132     dup handle>> dup check-disposed drain
 133     dup [ wait-for-port ] [ 2drop ] if ;
 134
 135 M: unix io-multiplex ( ms/f -- )
 136     mx get-global wait-for-events ;
 137
 138 ! On Unix, you're not supposed to set stdin to non-blocking
 139 ! because the fd might be shared with another process (either
 140 ! parent or child). So what we do is have the VM start a thread
 141 ! which pumps data from the real stdin to a pipe. We set the
 142 ! pipe to non-blocking, and read from it instead of the real
 143 ! stdin. Very crufty, but it will suffice until we get native
 144 ! threading support at the language level.
 145 TUPLE: stdin < disposable control size data ;
 146
 147 M: stdin dispose*
 148     [
 149         [ control>> &dispose drop ]
 150         [ size>> &dispose drop ]
 151         [ data>> &dispose drop ]
 152         tri
 153     ] with-destructors ;
 154
 155 : wait-for-stdin ( stdin -- size )
 156     [ control>> CHAR: X over io:stream-write1 io:stream-flush ]
 157     [ size>> ssize_t heap-size swap io:stream-read ssize_t deref ]
 158     bi ;
 159
 160 :: refill-stdin ( buffer stdin size -- )
 161     stdin data>> handle-fd buffer buffer-end size read
 162     dup 0 < [
 163         drop
 164         errno EINTR = [ buffer stdin size refill-stdin ] [ (io-error) ] if
 165     ] [
 166         size = [ "Error reading stdin pipe" throw ] unless
 167         size buffer n>buffer
 168     ] if ;
 169
 170 M: stdin refill
 171     '[
 172         buffer>> _ dup wait-for-stdin refill-stdin f
 173     ] with-timeout ;
 174
 175 M: stdin cancel-operation
 176     [ size>> ] [ control>> ] bi [ cancel-operation ] bi@ ;
 177
 178 : control-write-fd ( -- fd ) &: control_write uint deref ;
 179
 180 : size-read-fd ( -- fd ) &: size_read uint deref ;
 181
 182 : data-read-fd ( -- fd ) &: stdin_read uint deref ;
 183
 184 : <stdin> ( -- stdin )
 185     stdin new-disposable
 186         control-write-fd <fd> <output-port> >>control
 187         size-read-fd <fd> init-fd <input-port> >>size
 188         data-read-fd <fd> >>data ;
 189
 190 SYMBOL: dispatch-signal-hook
 191
 192 dispatch-signal-hook [ [ drop ] ] initialize
 193
 194 : signal-pipe-fd ( -- n )
 195     OBJ-SIGNAL-PIPE special-object ; inline
 196
 197 : signal-pipe-loop ( port -- )
 198     '[
 199         int heap-size _ io:stream-read
 200         dup [ int deref dispatch-signal-hook get call( x -- ) ] when*
 201     ] loop ;
 202
 203 : start-signal-pipe-thread ( -- )
 204     signal-pipe-fd [
 205         <fd> init-fd <input-port>
 206         '[ _ signal-pipe-loop ] "Signals" spawn drop
 207     ] when* ;
 208
 209 M: unix init-stdio
 210     <stdin> <input-port>
 211     1 <fd> <output-port>
 212     2 <fd> <output-port>
 213     set-stdio ;
 214
 215 ! mx io-task for embedding an fd-based mx inside another mx
 216 TUPLE: mx-port < port mx ;
 217
 218 : <mx-port> ( mx -- port )
 219     dup fd>> mx-port <port> swap >>mx ;
 220
 221 : multiplexer-error ( n -- n )
 222     dup 0 < [
 223         errno [ EAGAIN = ] [ EINTR = ] bi or
 224         [ drop 0 ] [ (io-error) ] if
 225     ] when ;
 226
 227 :: ?flag ( n mask symbol -- n )
 228     n mask bitand 0 > [ symbol , ] when n ;
 229
 230 [ start-signal-pipe-thread ] "io.backend.unix:signal-pipe-thread" add-startup-hook