git.iko.soy/streamly
on a05abfc help sign in
streamly/src/Streamly/Internal/FileSystem/FDIO.hs
223 lines 8.4 kB history raw
1#include "inline.hs"
3-- |
4-- Module : Streamly.Internal.FileSystem.FDIO
5-- Copyright : (c) 2019 Composewell Technologies
6-- Copyright : (c) 1994-2008 The University of Glasgow
7--
8-- License : BSD3
9-- Maintainer : streamly@composewell.com
10-- Stability : experimental
11-- Portability : GHC
12--
13-- Low level IO routines interfacing the operating system.
14--
16module Streamly.Internal.FileSystem.FDIO
17 ( write
18 , writeAll
19 , writev
20 , writevAll
21 )
22where
24import Control.Monad (when)
25import Streamly.Internal.System.IOVec.Type (IOVec)
26#if !defined(mingw32_HOST_OS)
27import Control.Concurrent (threadWaitWrite)
28import Data.Int (Int64)
29import Foreign.C.Error (throwErrnoIfMinus1RetryMayBlock)
30#if __GLASGOW_HASKELL__ >= 802
31import Foreign.C.Types (CBool(..))
32#endif
33import System.Posix.Internals (c_write, c_safe_write)
34import Streamly.Internal.System.IOVec.Type (c_writev, c_safe_writev)
35#endif
37import Foreign.C.Types (CSize(..), CInt(..))
38import Data.Word (Word8)
39import Foreign.Ptr (plusPtr, Ptr)
41import GHC.IO.FD (FD(..))
43-------------------------------------------------------------------------------
44-- IO Routines
45-------------------------------------------------------------------------------
47-- See System.POSIX.Internals in GHC base package
49-------------------------------------------------------------------------------
50-- Write without blocking the underlying OS thread
51-------------------------------------------------------------------------------
53#if !defined(mingw32_HOST_OS)
55foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
57isNonBlocking :: FD -> Bool
58isNonBlocking fd = fdIsNonBlocking fd /= 0
60-- "poll"s the fd for data to become available or timeout
61-- See cbits/inputReady.c in base package
62#if __GLASGOW_HASKELL__ >= 804
63foreign import ccall unsafe "fdReady"
64 unsafe_fdReady :: CInt -> CBool -> Int64 -> CBool -> IO CInt
65#else
66foreign import ccall safe "fdReady"
67 unsafe_fdReady :: CInt -> CInt -> CInt -> CInt -> IO CInt
68#endif
70writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
71writeNonBlocking loc !fd !buf !off !len
72 | isNonBlocking fd = unsafe_write -- unsafe is ok, it can't block
73 | otherwise = do
74 let isWrite = 1
75 isSocket = 0
76 msecs = 0
77 r <- unsafe_fdReady (fdFD fd) isWrite msecs isSocket
78 when (r == 0) $ threadWaitWrite (fromIntegral (fdFD fd))
79 if threaded then safe_write else unsafe_write
81 where
83 do_write call = fromIntegral `fmap`
84 throwErrnoIfMinus1RetryMayBlock loc call
85 (threadWaitWrite (fromIntegral (fdFD fd)))
86 unsafe_write = do_write (c_write (fdFD fd) (buf `plusPtr` off) len)
87 safe_write = do_write (c_safe_write (fdFD fd) (buf `plusPtr` off) len)
89writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt
90writevNonBlocking loc !fd !iov !cnt
91 | isNonBlocking fd = unsafe_write -- unsafe is ok, it can't block
92 | otherwise = do
93 let isWrite = 1
94 isSocket = 0
95 msecs = 0
96 r <- unsafe_fdReady (fdFD fd) isWrite msecs isSocket
97 when (r == 0) $ threadWaitWrite (fromIntegral (fdFD fd))
98 if threaded then safe_write else unsafe_write
100 where
102 do_write call = fromIntegral `fmap`
103 throwErrnoIfMinus1RetryMayBlock loc call
104 (threadWaitWrite (fromIntegral (fdFD fd)))
105 unsafe_write = do_write (c_writev (fdFD fd) iov (fromIntegral cnt))
106 safe_write = do_write (c_safe_writev (fdFD fd) iov (fromIntegral cnt))
108#else
109writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
110writeNonBlocking = undefined
112writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt
113writevNonBlocking = undefined
114#endif
116-- Windows code is disabled for now
117#if 0
119#if defined(mingw32_HOST_OS)
120# if defined(i386_HOST_ARCH)
121# define WINDOWS_CCONV stdcall
122# elif defined(x86_64_HOST_ARCH)
123# define WINDOWS_CCONV ccall
124# else
125# error Unknown mingw32 arch
126# endif
127#endif
129foreign import WINDOWS_CCONV safe "recv"
130 c_safe_recv :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
132foreign import WINDOWS_CCONV safe "send"
133 c_safe_send :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
135blockingWriteRawBufferPtr :: String -> FD -> Ptr Word8-> Int -> CSize -> IO CInt
136blockingWriteRawBufferPtr loc !fd !buf !off !len
137 = throwErrnoIfMinus1Retry loc $ do
138 let start_ptr = buf `plusPtr` off
139 send_ret = c_safe_send (fdFD fd) start_ptr (fromIntegral len) 0
140 write_ret = c_safe_write (fdFD fd) start_ptr (fromIntegral len)
141 r <- bool write_ret send_ret (fdIsSocket fd)
142 when (r == -1) c_maperrno
143 return r
144 -- We don't trust write() to give us the correct errno, and
145 -- instead do the errno conversion from GetLastError()
146 -- ourselves. The main reason is that we treat ERROR_NO_DATA
147 -- (pipe is closing) as EPIPE, whereas write() returns EINVAL
148 -- for this case. We need to detect EPIPE correctly, because it
149 -- shouldn't be reported as an error when it happens on stdout.
150 -- As for send()'s case, Winsock functions don't do errno
151 -- conversion in any case so we have to do it ourselves.
152 -- That means we're doing the errno conversion no matter if the
153 -- fd is from a socket or not.
155-- NOTE: "safe" versions of the read/write calls for use by the threaded RTS.
156-- These calls may block, but that's ok.
158asyncWriteRawBufferPtr :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
159asyncWriteRawBufferPtr loc !fd !buf !off !len = do
160 (l, rc) <- asyncWrite (fromIntegral (fdFD fd)) (fdIsSocket_ fd)
161 (fromIntegral len) (buf `plusPtr` off)
162 if l == (-1)
163 then let sock_errno = c_maperrno_func (fromIntegral rc)
164 non_sock_errno = Errno (fromIntegral rc)
165 errno = bool non_sock_errno sock_errno (fdIsSocket fd)
166 in ioError (errnoToIOError loc errno Nothing Nothing)
167 else return (fromIntegral l)
169writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
170writeNonBlocking loc !fd !buf !off !len
171 | threaded = blockingWriteRawBufferPtr loc fd buf off len
172 | otherwise = asyncWriteRawBufferPtr loc fd buf off len
174#endif
176-- | @write FD buffer offset length@ tries to write data on the given
177-- filesystem fd (cannot be a socket) up to sepcified length starting from the
178-- given offset in the buffer. The write will not block the OS thread, it may
179-- suspend the Haskell thread until write can proceed. Returns the actual
180-- amount of data written.
181write :: FD -> Ptr Word8 -> Int -> CSize -> IO CInt
182write = writeNonBlocking "Streamly.Internal.FileSystem.FDIO"
184-- XXX sendAll for sockets has a similar code, we can deduplicate the two.
185-- XXX we need to check the errno to determine if the loop should continue. For
186-- example, write may return without writing all data if the process file-size
187-- limit has reached, in that case keep writing in a loop is fruitless.
188--
189-- | Keep writing in a loop until all data in the buffer has been written.
190writeAll :: FD -> Ptr Word8 -> Int -> IO ()
191writeAll fd ptr bytes = do
192 res <- write fd ptr 0 (fromIntegral bytes)
193 let res' = fromIntegral res
194 when (res' < bytes) $
195 writeAll fd (ptr `plusPtr` res') (bytes - res')
197-------------------------------------------------------------------------------
198-- Vector IO
199-------------------------------------------------------------------------------
201-- | @write FD iovec count@ tries to write data on the given filesystem fd
202-- (cannot be a socket) from an iovec with specified number of entries. The
203-- write will not block the OS thread, it may suspend the Haskell thread until
204-- write can proceed. Returns the actual amount of data written.
205writev :: FD -> Ptr IOVec -> Int -> IO CInt
206writev = writevNonBlocking "Streamly.Internal.FileSystem.FDIO"
208-- XXX incomplete
209-- | Keep writing an iovec in a loop until all the iovec entries are written.
210writevAll :: FD -> Ptr IOVec -> Int -> IO ()
211writevAll fd iovec count = do
212 _res <- writev fd iovec count
213 {-
214 let res' = fromIntegral res
215 totalBytes = countIOVecBytes
216 if res' < totalBytes
217 then do
218 let iovec' = createModifiedIOVec
219 count' = ...
220 writeAll fd iovec' count'
221 else return ()
222 -}
223 return ()