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