6688d0e · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Disable windows raw fd IO code

6688d0eHarendra Kumar7 years ago

src/Streamly/FileSystem/FD.hs+18−10

compare with master

unifiedsplit

@@ -139,10 +139,14 @@ import qualified Streamly.FileSystem.FDIO as RawIO

139

140import Streamly.Mem.Array.Types (Array(..))

141import Streamly.Streams.Serial (SerialT)

142+import Streamly.Streams.StreamK.Type (IsStream, mkStream)

143+import Streamly.Mem.Array.Types (byteLength, defaultChunkSize)

144+

145+#if !defined(mingw32_HOST_OS)

146+import Streamly.Mem.Array.Types (groupIOVecsOf)

142import Streamly.Streams.StreamD (toStreamD)

147import Streamly.Streams.StreamD (toStreamD)

143import Streamly.Streams.StreamD.Type (fromStreamD)

148import Streamly.Streams.StreamD.Type (fromStreamD)

144−import Streamly.Streams.StreamK.Type (IsStream, mkStream)

149+#endif

145−import Streamly.Mem.Array.Types (byteLength, defaultChunkSize, groupIOVecsOf)

146-- import Streamly.Fold (Fold)

150-- import Streamly.Fold (Fold)

147-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

151-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

148

152

@@ -249,6 +253,7 @@ writeArray (Handle fd) arr = withForeignPtr (aStart arr) $ \p -> do

249 where

253 where

250 aLen = byteLength arr

254 aLen = byteLength arr

251

255

256+#if !defined(mingw32_HOST_OS)

252-- | Write an array of 'IOVec' to a file handle.

257-- | Write an array of 'IOVec' to a file handle.

253--

258--

254-- @since 0.7.0

259-- @since 0.7.0

@@ -258,6 +263,7 @@ writeIOVec _ iov | A.length iov == 0 = return ()

258writeIOVec (Handle fd) iov =

263writeIOVec (Handle fd) iov =

259 withForeignPtr (aStart iov) $ \p ->

264 withForeignPtr (aStart iov) $ \p ->

260 RawIO.writevAll fd p (A.length iov)

265 RawIO.writevAll fd p (A.length iov)

266+#endif

261

267

262-------------------------------------------------------------------------------

268-------------------------------------------------------------------------------

263-- Stream of Arrays IO

269-- Stream of Arrays IO

@@ -344,14 +350,6 @@ read = A.flattenArrays . readArrays

344writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()

350writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()

345writeArrays h m = S.mapM_ (liftIO . writeArray h) m

351writeArrays h m = S.mapM_ (liftIO . writeArray h) m

346

352

347−-- XXX this is incomplete

348−-- | Write a stream of 'IOVec' arrays to a handle.

349−--

350−-- @since 0.7.0

351−{-# INLINE writev #-}

352−writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()

353−writev h m = S.mapM_ (liftIO . writeIOVec h) m

354−

355-- | Write a stream of arrays to a handle after coalescing them in chunks of

353-- | Write a stream of arrays to a handle after coalescing them in chunks of

356-- specified size. The chunk size is only a maximum and the actual writes could

354-- specified size. The chunk size is only a maximum and the actual writes could

357-- be smaller than that as we do not split the arrays to fit them to the

355-- be smaller than that as we do not split the arrays to fit them to the

@@ -363,6 +361,15 @@ writeArraysPackedUpto :: (MonadIO m, Storable a)

363 => Int -> Handle -> SerialT m (Array a) -> m ()

361 => Int -> Handle -> SerialT m (Array a) -> m ()

364writeArraysPackedUpto n h xs = writeArrays h $ A.packArraysChunksOf n xs

362writeArraysPackedUpto n h xs = writeArrays h $ A.packArraysChunksOf n xs

365

363

364+#if !defined(mingw32_HOST_OS)

365+-- XXX this is incomplete

366+-- | Write a stream of 'IOVec' arrays to a handle.

367+--

368+-- @since 0.7.0

369+{-# INLINE writev #-}

370+writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()

371+writev h m = S.mapM_ (liftIO . writeIOVec h) m

372+

366-- XXX this is incomplete

373-- XXX this is incomplete

367-- | Write a stream of arrays to a handle after grouping them in 'IOVec' arrays

374-- | Write a stream of arrays to a handle after grouping them in 'IOVec' arrays

368-- of up to a maximum total size. Writes are performed using gather IO via

375-- of up to a maximum total size. Writes are performed using gather IO via

@@ -375,6 +382,7 @@ _writevArraysPackedUpto :: MonadIO m

375 => Int -> Handle -> SerialT m (Array a) -> m ()

382 => Int -> Handle -> SerialT m (Array a) -> m ()

376_writevArraysPackedUpto n h xs =

383_writevArraysPackedUpto n h xs =

377 writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)

384 writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)

385+#endif

378

386

379-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

387-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

380--

388--

src/Streamly/FileSystem/FDIO.hs+29−6

@@ -28,21 +28,25 @@ module Streamly.FileSystem.FDIO

28 )

29where

31+#if !defined(mingw32_HOST_OS)

31import Control.Concurrent (threadWaitWrite)

32import Control.Concurrent (threadWaitWrite)

32import Control.Monad (when)

33import Control.Monad (when)

33import Data.Int (Int64)

34import Data.Int (Int64)

34−import Data.Word (Word8)

35import Foreign.C.Error (throwErrnoIfMinus1RetryMayBlock)

36−import Foreign.C.Types (CSize(..), CInt(..))

37#if __GLASGOW_HASKELL__ >= 802

36#if __GLASGOW_HASKELL__ >= 802

38import Foreign.C.Types (CBool(..))

37import Foreign.C.Types (CBool(..))

39#endif

38#endif

40−import Foreign.Ptr (plusPtr, Ptr)

41import System.Posix.Internals (c_write, c_safe_write)

39import System.Posix.Internals (c_write, c_safe_write)

40+import Streamly.FileSystem.IOVec (c_writev, c_safe_writev)

41+#endif

42+

43+import Foreign.C.Types (CSize(..), CInt(..))

44+import Data.Word (Word8)

45+import Foreign.Ptr (plusPtr, Ptr)

43import GHC.IO.FD (FD(..))

47import GHC.IO.FD (FD(..))

45−import Streamly.FileSystem.IOVec (IOVec(..), c_writev, c_safe_writev)

49+import Streamly.FileSystem.IOVec (IOVec(..))

47-------------------------------------------------------------------------------

51-------------------------------------------------------------------------------

48-- IO Routines

52-- IO Routines

@@ -54,13 +58,13 @@ import Streamly.FileSystem.IOVec (IOVec(..), c_writev, c_safe_writev)

54-- Write without blocking the underlying OS thread

58-- Write without blocking the underlying OS thread

55-------------------------------------------------------------------------------

59-------------------------------------------------------------------------------

61+#if !defined(mingw32_HOST_OS)

62+

57foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool

63foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool

59isNonBlocking :: FD -> Bool

65isNonBlocking :: FD -> Bool

60isNonBlocking fd = fdIsNonBlocking fd /= 0

66isNonBlocking fd = fdIsNonBlocking fd /= 0

62−#if !defined(mingw32_HOST_OS)

63−

64-- "poll"s the fd for data to become available or timeout

68-- "poll"s the fd for data to become available or timeout

65-- See cbits/inputReady.c in base package

69-- See cbits/inputReady.c in base package

66#if __GLASGOW_HASKELL__ >= 804

70#if __GLASGOW_HASKELL__ >= 804

@@ -110,6 +114,25 @@ writevNonBlocking loc !fd !iov !cnt

110 safe_write = do_write (c_safe_writev (fdFD fd) iov (fromIntegral cnt))

114 safe_write = do_write (c_safe_writev (fdFD fd) iov (fromIntegral cnt))

111

115

112#else

116#else

117+writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt

118+writeNonBlocking = undefined

119+

120+writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt

121+writevNonBlocking = undefined

122+#endif

123+

124+-- Windows code is disabled for now

125+#if 0

126+

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

113

136

114foreign import WINDOWS_CCONV safe "recv"

137foreign import WINDOWS_CCONV safe "recv"

115 c_safe_recv :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt

138 c_safe_recv :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt

src/Streamly/FileSystem/IOVec.hsc+6−1

@@ -25,8 +25,10 @@ where

25import Data.Word (Word8, Word64)

26import Foreign.C.Types (CInt(..))

27import Foreign.Ptr (Ptr)

28−import Foreign.Storable (Storable(..))

29import System.Posix.Types (CSsize(..))

28import System.Posix.Types (CSsize(..))

29+#if !defined(mingw32_HOST_OS)

30+import Foreign.Storable (Storable(..))

31+#endif

31-------------------------------------------------------------------------------

33-------------------------------------------------------------------------------

32-- IOVec

34-- IOVec

@@ -65,6 +67,9 @@ foreign import capi safe "sys/uio.h writev"

65 c_safe_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize

67 c_safe_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize

67#else

69#else

70+c_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize

68c_writev = error "writev not implemented for windows"

71c_writev = error "writev not implemented for windows"

72+

73+c_safe_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize

69c_safe_writev = error "writev not implemented for windows"

74c_safe_writev = error "writev not implemented for windows"

70#endif

75#endif

src/Streamly/Mem/Array/Types.hs+7

@@ -39,7 +39,9 @@ module Streamly.Mem.Array.Types

39 , flattenArrays

40 , flattenArraysRev

41 , packArraysChunksOf

42+#if !defined(mingw32_HOST_OS)

42 , groupIOVecsOf

43 , groupIOVecsOf

44+#endif

43 , splitOn

45 , splitOn

45 -- * Elimination

47 -- * Elimination

@@ -97,7 +99,10 @@ import GHC.Ptr (Ptr(..))

98import Streamly.Fold.Types (Fold(..))

100import Streamly.Fold.Types (Fold(..))

99import Streamly.SVar (adaptState)

101import Streamly.SVar (adaptState)

102+

103+#if !defined(mingw32_HOST_OS)

100import Streamly.FileSystem.FDIO (IOVec(..))

104import Streamly.FileSystem.FDIO (IOVec(..))

105+#endif

101

106

102import qualified Streamly.Streams.StreamD.Type as D

107import qualified Streamly.Streams.StreamD.Type as D

103import qualified Streamly.Streams.StreamK as K

108import qualified Streamly.Streams.StreamK as K

@@ -923,6 +928,7 @@ packArraysChunksOf n (D.Stream step state) =

923

928

924 step' _ (SpliceYielding arr next) = return $ D.Yield arr next

929 step' _ (SpliceYielding arr next) = return $ D.Yield arr next

925

930

931+#if !defined(mingw32_HOST_OS)

926data GatherState s arr

932data GatherState s arr

927 = GatherInitial s

933 = GatherInitial s

928 | GatherBuffering s arr Int

934 | GatherBuffering s arr Int

@@ -990,6 +996,7 @@ groupIOVecsOf n maxIOVLen (D.Stream step state) =

990 step' _ GatherFinish = return D.Stop

996 step' _ GatherFinish = return D.Stop

991

997

992 step' _ (GatherYielding iov next) = return $ D.Yield iov next

998 step' _ (GatherYielding iov next) = return $ D.Yield iov next

999+#endif

993

1000

994-- | Create two slices of an array without copying the original array. The

1001-- | Create two slices of an array without copying the original array. The

995-- specified index @i@ is the first index of the second slice.

1002-- specified index @i@ is the first index of the second slice.