git.iko.soy/streamly
on ef2ea11 help sign in
streamly/src/Streamly/Internal/FileSystem/FDIO.hs
221 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)
30import Foreign.C.Types (CBool(..))
31import System.Posix.Internals (c_write, c_safe_write)
32import Streamly.Internal.System.IOVec.Type (c_writev, c_safe_writev)
33#endif
35import Foreign.C.Types (CSize(..), CInt(..))
36import Data.Word (Word8)
37import Foreign.Ptr (plusPtr, Ptr)
39import GHC.IO.FD (FD(..))
41-------------------------------------------------------------------------------
42-- IO Routines
43-------------------------------------------------------------------------------
45-- See System.POSIX.Internals in GHC base package
47-------------------------------------------------------------------------------
48-- Write without blocking the underlying OS thread
49-------------------------------------------------------------------------------
51#if !defined(mingw32_HOST_OS)
53foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
55isNonBlocking :: FD -> Bool
56isNonBlocking fd = fdIsNonBlocking fd /= 0
58-- "poll"s the fd for data to become available or timeout
59-- See cbits/inputReady.c in base package
60#if __GLASGOW_HASKELL__ >= 804
61foreign import ccall unsafe "fdReady"
62 unsafe_fdReady :: CInt -> CBool -> Int64 -> CBool -> IO CInt
63#else
64foreign import ccall safe "fdReady"
65 unsafe_fdReady :: CInt -> CInt -> CInt -> CInt -> IO CInt
66#endif
68writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
69writeNonBlocking loc !fd !buf !off !len
70 | isNonBlocking fd = unsafe_write -- unsafe is ok, it can't block
71 | otherwise = do
72 let isWrite = 1
73 isSocket = 0
74 msecs = 0
75 r <- unsafe_fdReady (fdFD fd) isWrite msecs isSocket
76 when (r == 0) $ threadWaitWrite (fromIntegral (fdFD fd))
77 if threaded then safe_write else unsafe_write
79 where
81 do_write call = fromIntegral `fmap`
82 throwErrnoIfMinus1RetryMayBlock loc call
83 (threadWaitWrite (fromIntegral (fdFD fd)))
84 unsafe_write = do_write (c_write (fdFD fd) (buf `plusPtr` off) len)
85 safe_write = do_write (c_safe_write (fdFD fd) (buf `plusPtr` off) len)
87writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt
88writevNonBlocking loc !fd !iov !cnt
89 | isNonBlocking fd = unsafe_write -- unsafe is ok, it can't block
90 | otherwise = do
91 let isWrite = 1
92 isSocket = 0
93 msecs = 0
94 r <- unsafe_fdReady (fdFD fd) isWrite msecs isSocket
95 when (r == 0) $ threadWaitWrite (fromIntegral (fdFD fd))
96 if threaded then safe_write else unsafe_write
98 where
100 do_write call = fromIntegral `fmap`
101 throwErrnoIfMinus1RetryMayBlock loc call
102 (threadWaitWrite (fromIntegral (fdFD fd)))
103 unsafe_write = do_write (c_writev (fdFD fd) iov (fromIntegral cnt))
104 safe_write = do_write (c_safe_writev (fdFD fd) iov (fromIntegral cnt))
106#else
107writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
108writeNonBlocking = undefined
110writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt
111writevNonBlocking = undefined
112#endif
114-- Windows code is disabled for now
115#if 0
117#if defined(mingw32_HOST_OS)
118# if defined(i386_HOST_ARCH)
119# define WINDOWS_CCONV stdcall
120# elif defined(x86_64_HOST_ARCH)
121# define WINDOWS_CCONV ccall
122# else
123# error Unknown mingw32 arch
124# endif
125#endif
127foreign import WINDOWS_CCONV safe "recv"
128 c_safe_recv :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
130foreign import WINDOWS_CCONV safe "send"
131 c_safe_send :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
133blockingWriteRawBufferPtr :: String -> FD -> Ptr Word8-> Int -> CSize -> IO CInt
134blockingWriteRawBufferPtr loc !fd !buf !off !len
135 = throwErrnoIfMinus1Retry loc $ do
136 let start_ptr = buf `plusPtr` off
137 send_ret = c_safe_send (fdFD fd) start_ptr (fromIntegral len) 0
138 write_ret = c_safe_write (fdFD fd) start_ptr (fromIntegral len)
139 r <- bool write_ret send_ret (fdIsSocket fd)
140 when (r == -1) c_maperrno
141 return r
142 -- We don't trust write() to give us the correct errno, and
143 -- instead do the errno conversion from GetLastError()
144 -- ourselves. The main reason is that we treat ERROR_NO_DATA
145 -- (pipe is closing) as EPIPE, whereas write() returns EINVAL
146 -- for this case. We need to detect EPIPE correctly, because it
147 -- shouldn't be reported as an error when it happens on stdout.
148 -- As for send()'s case, Winsock functions don't do errno
149 -- conversion in any case so we have to do it ourselves.
150 -- That means we're doing the errno conversion no matter if the
151 -- fd is from a socket or not.
153-- NOTE: "safe" versions of the read/write calls for use by the threaded RTS.
154-- These calls may block, but that's ok.
156asyncWriteRawBufferPtr :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
157asyncWriteRawBufferPtr loc !fd !buf !off !len = do
158 (l, rc) <- asyncWrite (fromIntegral (fdFD fd)) (fdIsSocket_ fd)
159 (fromIntegral len) (buf `plusPtr` off)
160 if l == (-1)
161 then let sock_errno = c_maperrno_func (fromIntegral rc)
162 non_sock_errno = Errno (fromIntegral rc)
163 errno = bool non_sock_errno sock_errno (fdIsSocket fd)
164 in ioError (errnoToIOError loc errno Nothing Nothing)
165 else return (fromIntegral l)
167writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
168writeNonBlocking loc !fd !buf !off !len
169 | threaded = blockingWriteRawBufferPtr loc fd buf off len
170 | otherwise = asyncWriteRawBufferPtr loc fd buf off len
172#endif
174-- | @write FD buffer offset length@ tries to write data on the given
175-- filesystem fd (cannot be a socket) up to sepcified length starting from the
176-- given offset in the buffer. The write will not block the OS thread, it may
177-- suspend the Haskell thread until write can proceed. Returns the actual
178-- amount of data written.
179write :: FD -> Ptr Word8 -> Int -> CSize -> IO CInt
180write = writeNonBlocking "Streamly.Internal.FileSystem.FDIO"
182-- XXX sendAll for sockets has a similar code, we can deduplicate the two.
183-- XXX we need to check the errno to determine if the loop should continue. For
184-- example, write may return without writing all data if the process file-size
185-- limit has reached, in that case keep writing in a loop is fruitless.
186--
187-- | Keep writing in a loop until all data in the buffer has been written.
188writeAll :: FD -> Ptr Word8 -> Int -> IO ()
189writeAll fd ptr bytes = do
190 res <- write fd ptr 0 (fromIntegral bytes)
191 let res' = fromIntegral res
192 when (res' < bytes) $
193 writeAll fd (ptr `plusPtr` res') (bytes - res')
195-------------------------------------------------------------------------------
196-- Vector IO
197-------------------------------------------------------------------------------
199-- | @write FD iovec count@ tries to write data on the given filesystem fd
200-- (cannot be a socket) from an iovec with specified number of entries. The
201-- write will not block the OS thread, it may suspend the Haskell thread until
202-- write can proceed. Returns the actual amount of data written.
203writev :: FD -> Ptr IOVec -> Int -> IO CInt
204writev = writevNonBlocking "Streamly.Internal.FileSystem.FDIO"
206-- XXX incomplete
207-- | Keep writing an iovec in a loop until all the iovec entries are written.
208writevAll :: FD -> Ptr IOVec -> Int -> IO ()
209writevAll fd iovec count = do
210 _res <- writev fd iovec count
211 {-
212 let res' = fromIntegral res
213 totalBytes = countIOVecBytes
214 if res' < totalBytes
215 then do
216 let iovec' = createModifiedIOVec
217 count' = ...
218 writeAll fd iovec' count'
219 else return ()
220 -}
221 return ()