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