git.iko.soy/streamly
on master help sign in
Raw IO w/o buffering, add some array APIs
694eab4Harendra Kumar
benchmark/FileIO.hs+3−3
compare with master
unifiedsplit
@@ -18,12 +18,12 @@ import Gauge
18import qualified Streamly.FileSystem.Handle as FH
18import qualified Streamly.FileSystem.Handle as FH
19import qualified Streamly.Mem.Array as A
19import qualified Streamly.Mem.Array as A
20import qualified Streamly.Prelude as S
20import qualified Streamly.Prelude as S
21−import qualified Streamly.String as SS
22
21
23#ifdef DEVBUILD
22#ifdef DEVBUILD
24import Data.Char (ord, chr)
23import Data.Char (ord, chr)
25import System.IO (hSeek, SeekMode(..))
24import System.IO (hSeek, SeekMode(..))
26import qualified Streamly.Fold as FL
25import qualified Streamly.Fold as FL
26+import qualified Streamly.String as SS
27#endif
27#endif
28
28
29-- Input and output file handles
29-- Input and output file handles
@@ -94,8 +94,8 @@ main = do
94 Handles inh _ <- readIORef href
94 Handles inh _ <- readIORef href
95 let s = FH.readArrays inh
95 let s = FH.readArrays inh
96 larr <- S.last s
96 larr <- S.last s
97− return $ case larr of
97+ case larr of
98− Nothing -> Nothing
98+ Nothing -> return Nothing
99 Just arr -> A.readIndex arr (A.length arr - 1)
99 Just arr -> A.readIndex arr (A.length arr - 1)
100 -- Note: this cannot be fairly compared with GNU wc -c or wc -m as
100 -- Note: this cannot be fairly compared with GNU wc -c or wc -m as
101 -- wc uses lseek to just determine the file size rather than reading
101 -- wc uses lseek to just determine the file size rather than reading
examples/HandleIO.hs+23−11
@@ -2,40 +2,52 @@ import qualified Streamly.Prelude as S
2import qualified Streamly.Fold as FL
2import qualified Streamly.Fold as FL
3import qualified Streamly.Mem.Array as A
3import qualified Streamly.Mem.Array as A
4import qualified Streamly.FileSystem.Handle as FH
4import qualified Streamly.FileSystem.Handle as FH
5+import qualified System.IO as FH
6+-- import qualified Streamly.FileSystem.FD as FH
5import qualified Streamly.String as SS
7import qualified Streamly.String as SS
6
8
7import Data.Char (ord)
9import Data.Char (ord)
8import System.Environment (getArgs)
10import System.Environment (getArgs)
9−import System.IO (openFile, IOMode(..), stdout, Handle, hSeek, SeekMode(..))
11+import System.IO (IOMode(..), hSeek, SeekMode(..))
10
12
11−cat :: Handle -> IO ()
13+cat :: FH.Handle -> IO ()
12−cat src = FH.writeArrays stdout $ FH.readArraysUpto (256*1024) src
14+cat src = FH.writeArrays FH.stdout $ FH.readArraysOfUpto (256*1024) src
15+{-
16+cat src =
17+ FH.writeByChunksOf (1024*1024) FH.stdout
18+ $ FH.readByChunksUpto (16*1024) src
19+-}
13
20
14−cp :: Handle -> Handle -> IO ()
21+cp :: FH.Handle -> FH.Handle -> IO ()
15−cp src dst = FH.writeArrays dst $ FH.readArraysUpto (256*1024) src
22+cp src dst = FH.writeArrays dst $ FH.readArraysOfUpto (256*1024) src
23+{-
24+cp src dst =
25+ FH.writeByChunksOf (1024*1024) dst
26+ $ FH.readByChunksUpto (16*1024) src
27+-}
16
28
17ord' :: Num a => Char -> a
29ord' :: Num a => Char -> a
18ord' = (fromIntegral . ord)
30ord' = (fromIntegral . ord)
19
31
20−wcl :: Handle -> IO ()
32+wcl :: FH.Handle -> IO ()
21wcl src = print =<< (S.length
33wcl src = print =<< (S.length
22 $ flip SS.foldLines FL.drain
34 $ flip SS.foldLines FL.drain
23 $ SS.decodeChar8
35 $ SS.decodeChar8
24 $ FH.read src)
36 $ FH.read src)
25
37
26−grepc :: String -> Handle -> IO ()
38+grepc :: String -> FH.Handle -> IO ()
27grepc pat src = print . (subtract 1) =<< (S.length
39grepc pat src = print . (subtract 1) =<< (S.length
28 $ FL.splitOn (A.fromList (map ord' pat)) FL.drain
40 $ FL.splitOn (A.fromList (map ord' pat)) FL.drain
29 $ FH.read src)
41 $ FH.read src)
30
42
31−avgll :: Handle -> IO ()
43+avgll :: FH.Handle -> IO ()
32avgll src = print =<< (FL.foldl' avg
44avgll src = print =<< (FL.foldl' avg
33 $ FL.splitSuffixBy (== ord' '\n') FL.length
45 $ FL.splitSuffixBy (== ord' '\n') FL.length
34 $ FH.read src)
46 $ FH.read src)
35 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
47 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
36 toDouble = fmap (fromIntegral :: Int -> Double)
48 toDouble = fmap (fromIntegral :: Int -> Double)
37
49
38−llhisto :: Handle -> IO ()
50+llhisto :: FH.Handle -> IO ()
39llhisto src = print =<< (FL.foldl' (FL.classify FL.length)
51llhisto src = print =<< (FL.foldl' (FL.classify FL.length)
40 $ S.map bucket
52 $ S.map bucket
41 $ FL.splitSuffixBy (== ord' '\n') FL.length
53 $ FL.splitSuffixBy (== ord' '\n') FL.length
@@ -46,7 +58,7 @@ llhisto src = print =<< (FL.foldl' (FL.classify FL.length)
46main :: IO ()
58main :: IO ()
47main = do
59main = do
48 name <- fmap head getArgs
60 name <- fmap head getArgs
49− src <- openFile name ReadMode
61+ src <- FH.openFile name ReadMode
50 let rewind = hSeek src AbsoluteSeek 0
62 let rewind = hSeek src AbsoluteSeek 0
51
63
52 rewind >> putStrLn "cat" >> cat src -- Unix cat program
64 rewind >> putStrLn "cat" >> cat src -- Unix cat program
@@ -55,5 +67,5 @@ main = do
55 rewind >> putStr "avgll " >> avgll src -- get average line length
67 rewind >> putStr "avgll " >> avgll src -- get average line length
56 rewind >> putStr "llhisto " >> llhisto src -- get line length histogram
68 rewind >> putStr "llhisto " >> llhisto src -- get line length histogram
57
69
58− dst <- openFile "dst-xyz.txt" WriteMode
70+ dst <- FH.openFile "dst-xyz.txt" WriteMode
59 rewind >> putStr "cp " >> cp src dst -- Unix cp program
71 rewind >> putStr "cp " >> cp src dst -- Unix cp program
src/Streamly/FileSystem/FD.hsnew file+561
@@ -0,0 +1,561 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE MagicHash #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# LANGUAGE UnboxedTuples #-}
6+
7+#include "inline.hs"
8+
9+-- |
10+-- Module : Streamly.FileSystem.FD
11+-- Copyright : (c) 2019 Composewell Technologies
12+--
13+-- License : BSD3
14+-- Maintainer : harendra.kumar@gmail.com
15+-- Stability : experimental
16+-- Portability : GHC
17+--
18+-- This module is a an experimental replacement for
19+-- "Streamly.FileSystem.Handle". The former module provides IO facilities based
20+-- on the GHC Handle type. The APIs in this module avoid the GHC handle layer
21+-- and provide more explicit control over buffering.
22+--
23+-- Read and write data as streams and arrays to and from files.
24+--
25+-- This module provides read and write APIs based on handles. Before reading or
26+-- writing, a file must be opened first using 'openFile'. The 'Handle' returned
27+-- by 'openFile' is then used to access the file. A 'Handle' is backed by an
28+-- operating system file descriptor. When the 'Handle' is garbage collected the
29+-- underlying file descriptor is automatically closed. A handle can be
30+-- explicitly closed using 'closeFile'.
31+--
32+-- Reading and writing APIs are divided into two categories, sequential
33+-- streaming APIs and random or seekable access APIs. File IO APIs are quite
34+-- similar to "Streamly.Mem.Array" read write APIs. In that regard, arrays can
35+-- be considered as in-memory files or files can be considered as on-disk
36+-- arrays.
37+--
38+-- > import qualified Streamly.FileSystem.FD as FD
39+--
40+
41+module Streamly.FileSystem.FD
42+ (
43+ -- * File Handles
44+ Handle
45+ , stdin
46+ , stdout
47+ , stderr
48+ , openFile
49+
50+ -- TODO file path based APIs
51+ -- , readFile
52+ -- , writeFile
53+
54+ -- * Streaming IO
55+ -- | Streaming APIs read or write data to or from a file or device
56+ -- sequentially, they never perform a seek to a random location. When
57+ -- reading, the stream is lazy and generated on-demand as the consumer
58+ -- consumes it. Read IO requests to the IO device are performed in chunks
59+ -- of 32KiB, this is referred to as @defaultChunkSize@ in the
60+ -- documentation. One IO request may or may not read the full chunk. If the
61+ -- whole stream is not consumed, it is possible that we may read slightly
62+ -- more from the IO device than what the consumer needed. Unless specified
63+ -- otherwise in the API, writes are collected into chunks of
64+ -- @defaultChunkSize@ before they are written to the IO device.
65+
66+ -- Streaming APIs work for all kind of devices, seekable or non-seekable;
67+ -- including disks, files, memory devices, terminals, pipes, sockets and
68+ -- fifos. While random access APIs work only for files or devices that have
69+ -- random access or seek capability for example disks, memory devices.
70+ -- Devices like terminals, pipes, sockets and fifos do not have random
71+ -- access capability.
72+
73+ -- ** Read File to Stream
74+ , read
75+ -- , readUtf8
76+ -- , readLines
77+ -- , readFrames
78+ , readByChunksUpto
79+
80+ -- -- * Array Read
81+ -- , readArrayUpto
82+ -- , readArrayOf
83+
84+ , readArrays
85+ , readArraysOfUpto
86+ -- , readArraysOf
87+
88+ -- ** Write File from Stream
89+ , write
90+ -- , writeUtf8
91+ -- , writeUtf8Lines
92+ -- , writeFrames
93+ , writeByChunksOf
94+
95+ -- -- * Array Write
96+ -- , writeArray
97+ , writeArrays
98+ , writeArraysOfUpto
99+ , writev
100+ , writevArraysOfUpto
101+
102+ -- -- * Random Access (Seek)
103+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
104+ -- files that have random access or seek capability. This type of devices
105+ -- include disks, files, memory devices and exclude terminals, pipes,
106+ -- sockets and fifos.
107+ --
108+ -- , readIndex
109+ -- , readSlice
110+ -- , readSliceRev
111+ -- , readAt -- read from a given position to th end of file
112+ -- , readSliceArrayUpto
113+ -- , readSliceArrayOf
114+
115+ -- , writeIndex
116+ -- , writeSlice
117+ -- , writeSliceRev
118+ -- , writeAt -- start writing at the given position
119+ -- , writeSliceArray
120+ )
121+where
122+
123+import Control.Monad.IO.Class (MonadIO(..))
124+import Data.Word (Word8)
125+import Foreign.ForeignPtr (withForeignPtr)
126+-- import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
127+import Foreign.Ptr (plusPtr, castPtr)
128+import Foreign.Storable (Storable(..))
129+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
130+-- import System.IO (Handle, hGetBufSome, hPutBuf)
131+import System.IO (IOMode)
132+import Prelude hiding (read)
133+
134+import qualified GHC.IO.FD as FD
135+import qualified GHC.IO.Device as RawIO hiding (write)
136+import qualified Streamly.FileSystem.FDIO as RawIO
137+
138+import Streamly.Mem.Array.Types (Array(..))
139+import Streamly.Streams.Serial (SerialT)
140+import Streamly.Streams.StreamD (toStreamD)
141+import Streamly.Streams.StreamD.Type (fromStreamD)
142+import Streamly.Streams.StreamK.Type (IsStream, mkStream)
143+import Streamly.Mem.Array.Types (byteLength, defaultChunkSize, groupIOVecsOf)
144+-- import Streamly.Fold (Fold)
145+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
146+
147+import qualified Streamly.Mem.Array as A
148+import qualified Streamly.Prelude as S
149+import qualified Streamly.Streams.StreamD.Type as D
150+
151+-------------------------------------------------------------------------------
152+-- References
153+-------------------------------------------------------------------------------
154+--
155+-- The following references may be useful to build an understanding about the
156+-- file API design:
157+--
158+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
159+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
160+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
161+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
162+-- https://www.w3.org/TR/FileAPI/ for http file API.
163+
164+-------------------------------------------------------------------------------
165+-- Handles
166+-------------------------------------------------------------------------------
167+
168+-- XXX attach a finalizer
169+-- | A 'Handle' is returned by 'openFile' and is subsequently used to perform
170+-- read and write operations on a file.
171+--
172+newtype Handle = Handle FD.FD
173+
174+-- | File handle for standard input
175+stdin :: Handle
176+stdin = Handle FD.stdin
177+
178+-- | File handle for standard output
179+stdout :: Handle
180+stdout = Handle FD.stdout
181+
182+-- | File handle for standard error
183+stderr :: Handle
184+stderr = Handle FD.stderr
185+
186+-- XXX we can support all the flags that the "open" system call supports.
187+-- Instead of using RTS locking mechanism can we use system provided locking
188+-- instead?
189+--
190+-- | Open a file that is not a directory and return a file handle.
191+-- 'openFile' enforces a multiple-reader single-writer locking on files. That
192+-- is, there may either be many handles on the same file which manage input, or
193+-- just one handle on the file which manages output. If any open handle is
194+-- managing a file for output, no new handle can be allocated for that file. If
195+-- any open handle is managing a file for input, new handles can only be
196+-- allocated if they do not manage output. Whether two files are the same is
197+-- implementation-dependent, but they should normally be the same if they have
198+-- the same absolute path name and neither has been renamed, for example.
199+--
200+openFile :: FilePath -> IOMode -> IO Handle
201+openFile path mode = fmap (Handle . fst) $ FD.openFile path mode True
202+
203+-------------------------------------------------------------------------------
204+-- Array IO (Input)
205+-------------------------------------------------------------------------------
206+
207+-- | Read a 'ByteArray' from a file handle. If no data is available on the
208+-- handle it blocks until some data becomes available. If data is available
209+-- then it immediately returns that data without blocking. It reads a maximum
210+-- of up to the size requested.
211+{-# INLINABLE readArrayUpto #-}
212+readArrayUpto :: Int -> Handle -> IO (Array Word8)
213+readArrayUpto size (Handle fd) = do
214+ ptr <- mallocPlainForeignPtrBytes size
215+ -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
216+ withForeignPtr ptr $ \p -> do
217+ -- n <- hGetBufSome h p size
218+ n <- RawIO.read fd p size
219+ let v = Array
220+ { aStart = ptr
221+ , aEnd = p `plusPtr` n
222+ , aBound = p `plusPtr` size
223+ }
224+ -- XXX shrink only if the diff is significant
225+ -- A.shrinkToFit v
226+ return v
227+
228+-------------------------------------------------------------------------------
229+-- Array IO (output)
230+-------------------------------------------------------------------------------
231+
232+-- | Write an 'Array' to a file handle.
233+--
234+-- @since 0.7.0
235+{-# INLINABLE writeArray #-}
236+writeArray :: Storable a => Handle -> Array a -> IO ()
237+writeArray _ arr | A.length arr == 0 = return ()
238+writeArray (Handle fd) arr = withForeignPtr (aStart arr) $ \p -> do
239+ RawIO.writeAll fd (castPtr p) aLen
240+ {-
241+ -- Experiment to compare "writev" based IO with "write" based IO.
242+ iov <- A.newArray 1
243+ let iov' = iov {aEnd = aBound iov}
244+ A.writeIndex iov' 0 (RawIO.IOVec (castPtr p) (fromIntegral aLen))
245+ RawIO.writevAll fd (unsafeForeignPtrToPtr (aStart iov')) 1
246+ -}
247+ where
248+ aLen = byteLength arr
249+
250+-- | Write an array of 'IOVec' to a file handle.
251+--
252+-- @since 0.7.0
253+{-# INLINABLE writeIOVec #-}
254+writeIOVec :: Handle -> Array RawIO.IOVec -> IO ()
255+writeIOVec _ iov | A.length iov == 0 = return ()
256+writeIOVec (Handle fd) iov =
257+ withForeignPtr (aStart iov) $ \p ->
258+ RawIO.writevAll fd p (A.length iov)
259+
260+-------------------------------------------------------------------------------
261+-- Stream of Arrays IO
262+-------------------------------------------------------------------------------
263+
264+-- | @readArraysOfUpto size h@ reads a stream of arrays from file handle @h@.
265+-- The maximum size of a single array is specified by @size@. The actual size
266+-- read may be less than or equal to @size@.
267+{-# INLINABLE _readArraysOfUpto #-}
268+_readArraysOfUpto :: (IsStream t, MonadIO m)
269+ => Int -> Handle -> t m (Array Word8)
270+_readArraysOfUpto size h = go
271+ where
272+ -- XXX use cons/nil instead
273+ go = mkStream $ \_ yld _ stp -> do
274+ arr <- liftIO $ readArrayUpto size h
275+ if A.length arr == 0
276+ then stp
277+ else yld arr go
278+
279+{-# INLINE_NORMAL readArraysOfUpto #-}
280+readArraysOfUpto :: (IsStream t, MonadIO m)
281+ => Int -> Handle -> t m (Array Word8)
282+readArraysOfUpto size h = D.fromStreamD (D.Stream step ())
283+ where
284+ {-# INLINE_LATE step #-}
285+ step _ _ = do
286+ arr <- liftIO $ readArrayUpto size h
287+ return $
288+ case A.length arr of
289+ 0 -> D.Stop
290+ _ -> D.Yield arr ()
291+
292+-- XXX read 'Array a' instead of Word8
293+--
294+-- | @readArrays h@ reads a stream of arrays from file handle @h@.
295+-- The maximum size of a single array is limited to @defaultChunkSize@.
296+-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
297+-- on the 'Handle'.
298+--
299+-- > readArrays = readArraysOfUpto defaultChunkSize
300+--
301+-- @since 0.7.0
302+{-# INLINE readArrays #-}
303+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
304+readArrays = readArraysOfUpto defaultChunkSize
305+
306+-------------------------------------------------------------------------------
307+-- Read File to Stream
308+-------------------------------------------------------------------------------
309+
310+-- TODO for concurrent streams implement readahead IO. We can send multiple
311+-- read requests at the same time. For serial case we can use async IO. We can
312+-- also control the read throughput in mbps or IOPS.
313+
314+-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file handle,
315+-- reads are performed in chunks of up to @chunkSize@. The stream ends as soon
316+-- as EOF is encountered.
317+--
318+{-# INLINE readByChunksUpto #-}
319+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
320+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h
321+
322+-- TODO
323+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
324+--
325+-- > read = 'readByChunks' A.defaultChunkSize
326+-- | Generate a stream of elements of the given type from a file 'Handle'. The
327+-- stream ends when EOF is encountered.
328+--
329+-- @since 0.7.0
330+{-# INLINE read #-}
331+read :: (IsStream t, MonadIO m) => Handle -> t m Word8
332+read = A.flattenArrays . readArrays
333+
334+-------------------------------------------------------------------------------
335+-- Writing
336+-------------------------------------------------------------------------------
337+
338+-- | Write a stream of arrays to a handle.
339+--
340+-- @since 0.7.0
341+{-# INLINE writeArrays #-}
342+writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
343+writeArrays h m = S.mapM_ (liftIO . writeArray h) m
344+
345+-- | Write a stream of 'IOVec' arrays to a handle.
346+--
347+-- @since 0.7.0
348+{-# INLINE writev #-}
349+writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()
350+writev h m = S.mapM_ (liftIO . writeIOVec h) m
351+
352+-- | Write a stream of arrays to a handle after coalescing them in chunks of
353+-- specified size. The chunk size is only a maximum and the actual writes could
354+-- be smaller than that as we do not split the arrays to fit them to the
355+-- specified size.
356+--
357+-- @since 0.7.0
358+{-# INLINE writeArraysOfUpto #-}
359+writeArraysOfUpto :: (MonadIO m, Storable a)
360+ => Int -> Handle -> SerialT m (Array a) -> m ()
361+writeArraysOfUpto n h xs = writeArrays h $ A.coalesceChunksOf n xs
362+
363+-- | Write a stream of arrays to a handle after grouping them in 'IOVec' arrays
364+-- of up to a maximum total size. Writes are performed using gather IO via
365+-- @writev@ system call. The maximum number of entries in each 'IOVec' group
366+-- limited to 512.
367+--
368+-- @since 0.7.0
369+{-# INLINE writevArraysOfUpto #-}
370+writevArraysOfUpto :: MonadIO m
371+ => Int -> Handle -> SerialT m (Array a) -> m ()
372+writevArraysOfUpto n h xs =
373+ writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)
374+
375+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
376+--
377+-- XXX test this
378+-- Note that if you use a chunk size less than 8K (GHC's default buffer
379+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
380+-- do not want buffering to occur at GHC level as well. Same thing applies to
381+-- writes as well.
382+
383+-- | Like 'write' but provides control over the write buffer. Output will
384+-- be written to the IO device as soon as we collect the specified number of
385+-- input elements.
386+--
387+-- @since 0.7.0
388+{-# INLINE writeByChunksOf #-}
389+writeByChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
390+writeByChunksOf n h m = writeArrays h $ A.arraysOf n m
391+
392+-- > write = 'writeByChunks' A.defaultChunkSize
393+--
394+-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
395+-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
396+-- depends on the 'IOMode' and the current seek position of the handle.
397+--
398+-- @since 0.7.0
399+{-# INLINE write #-}
400+write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
401+write = writeByChunksOf defaultChunkSize
402+
403+{-
404+{-# INLINE write #-}
405+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
406+write = toHandleWith A.defaultChunkSize
407+-}
408+
409+-------------------------------------------------------------------------------
410+-- IO with encoding/decoding Unicode characters
411+-------------------------------------------------------------------------------
412+
413+{-
414+-- |
415+-- > readUtf8 = decodeUtf8 . read
416+--
417+-- Read a UTF8 encoded stream of unicode characters from a file handle.
418+--
419+-- @since 0.7.0
420+{-# INLINE readUtf8 #-}
421+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
422+readUtf8 = decodeUtf8 . read
423+
424+-- |
425+-- > writeUtf8 h s = write h $ encodeUtf8 s
426+--
427+-- Encode a stream of unicode characters to UTF8 and write it to the given file
428+-- handle. Default block buffering applies to the writes.
429+--
430+-- @since 0.7.0
431+{-# INLINE writeUtf8 #-}
432+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
433+writeUtf8 h s = write h $ encodeUtf8 s
434+
435+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
436+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
437+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
438+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
439+-- option.
440+--
441+-- @since 0.7.0
442+{-# INLINE writeUtf8ByLines #-}
443+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
444+writeUtf8ByLines = undefined
445+
446+-- | Read UTF-8 lines from a file handle and apply the specified fold to each
447+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
448+--
449+-- @since 0.7.0
450+{-# INLINE readLines #-}
451+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
452+readLines h f = foldLines (readUtf8 h) f
453+
454+-------------------------------------------------------------------------------
455+-- Framing on a sequence
456+-------------------------------------------------------------------------------
457+
458+-- | Read a stream from a file handle and split it into frames delimited by
459+-- the specified sequence of elements. The supplied fold is applied on each
460+-- frame.
461+--
462+-- @since 0.7.0
463+{-# INLINE readFrames #-}
464+readFrames :: (IsStream t, MonadIO m, Storable a)
465+ => Array a -> Handle -> Fold m a b -> t m b
466+readFrames = undefined -- foldFrames . read
467+
468+-- | Write a stream to the given file handle buffering up to frames separated
469+-- by the given sequence or up to a maximum of @defaultChunkSize@.
470+--
471+-- @since 0.7.0
472+{-# INLINE writeByFrames #-}
473+writeByFrames :: (IsStream t, MonadIO m, Storable a)
474+ => Array a -> Handle -> t m a -> m ()
475+writeByFrames = undefined
476+
477+-------------------------------------------------------------------------------
478+-- Random Access IO (Seek)
479+-------------------------------------------------------------------------------
480+
481+-- XXX handles could be shared, so we may not want to use the handle state at
482+-- all for these APIs. we can use pread and pwrite instead. On windows we will
483+-- need to use readFile/writeFile with an offset argument.
484+
485+-------------------------------------------------------------------------------
486+
487+-- | Read the element at the given index treating the file as an array.
488+--
489+-- @since 0.7.0
490+{-# INLINE readIndex #-}
491+readIndex :: Storable a => Handle -> Int -> Maybe a
492+readIndex arr i = undefined
493+
494+-- NOTE: To represent a range to read we have chosen (start, size) instead of
495+-- (start, end). This helps in removing the ambiguity of whether "end" is
496+-- included in the range or not.
497+--
498+-- We could avoid specifying the range to be read and instead use "take size"
499+-- on the stream, but it may end up reading more and then consume it partially.
500+
501+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
502+-- from @handle@ starting at the offset @pos@ from the beginning of the file.
503+--
504+-- Reads are performed in chunks of size @chunkSize@. For block devices, to
505+-- avoid reading partial blocks @chunkSize@ must align with the block size of
506+-- the underlying device. If the underlying block size is unknown, it is a good
507+-- idea to keep it a multiple 4KiB. This API ensures that the start of each
508+-- chunk is aligned with @chunkSize@ from second chunk onwards.
509+--
510+{-# INLINE readSliceWith #-}
511+readSliceWith :: (IsStream t, MonadIO m, Storable a)
512+ => Int -> Handle -> Int -> Int -> t m a
513+readSliceWith chunkSize h pos len = undefined
514+
515+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
516+-- at index @i@ and reading up to @count@ elements in the forward direction
517+-- ending at the index @i + count - 1@.
518+--
519+-- @since 0.7.0
520+{-# INLINE readSlice #-}
521+readSlice :: (IsStream t, MonadIO m, Storable a)
522+ => Handle -> Int -> Int -> t m a
523+readSlice = readSliceWith A.defaultChunkSize
524+
525+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
526+-- at index @i@ and reading up to @count@ elements in the reverse direction
527+-- ending at the index @i - count + 1@.
528+--
529+-- @since 0.7.0
530+{-# INLINE readSliceRev #-}
531+readSliceRev :: (IsStream t, MonadIO m, Storable a)
532+ => Handle -> Int -> Int -> t m a
533+readSliceRev h i count = undefined
534+
535+-- | Write the given element at the given index in the file.
536+--
537+-- @since 0.7.0
538+{-# INLINE writeIndex #-}
539+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
540+writeIndex h i a = undefined
541+
542+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
543+-- starting at index @i@ and writing up to @count@ elements in the forward
544+-- direction ending at the index @i + count - 1@.
545+--
546+-- @since 0.7.0
547+{-# INLINE writeSlice #-}
548+writeSlice :: (IsStream t, Monad m, Storable a)
549+ => Handle -> Int -> Int -> t m a -> m ()
550+writeSlice h i len s = undefined
551+
552+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
553+-- starting at index @i@ and writing up to @count@ elements in the reverse
554+-- direction ending at the index @i - count + 1@.
555+--
556+-- @since 0.7.0
557+{-# INLINE writeSliceRev #-}
558+writeSliceRev :: (IsStream t, Monad m, Storable a)
559+ => Handle -> Int -> Int -> t m a -> m ()
560+writeSliceRev arr i len s = undefined
561+-}
src/Streamly/FileSystem/FDIO.hsnew file+200
@@ -0,0 +1,200 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE MagicHash #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# LANGUAGE UnboxedTuples #-}
6+
7+#include "inline.hs"
8+
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+--
21+
22+module Streamly.FileSystem.FDIO
23+ ( write
24+ , writeAll
25+ , IOVec (..)
26+ , writev
27+ , writevAll
28+ )
29+where
30+
31+import Control.Concurrent (threadWaitWrite)
32+import Control.Monad (when)
33+import Data.Int (Int64)
34+import Data.Word (Word8)
35+import Foreign.C.Error (throwErrnoIfMinus1RetryMayBlock)
36+import Foreign.C.Types (CSize(..), CInt(..), CBool(..))
37+import Foreign.Ptr (plusPtr, Ptr)
38+import System.Posix.Internals (c_write, c_safe_write)
39+
40+import GHC.IO.FD (FD(..))
41+
42+import Streamly.FileSystem.IOVec (IOVec(..), c_writev, c_safe_writev)
43+
44+-------------------------------------------------------------------------------
45+-- IO Routines
46+-------------------------------------------------------------------------------
47+
48+-- See System.POSIX.Internals in GHC base package
49+
50+-------------------------------------------------------------------------------
51+-- Write without blocking the underlying OS thread
52+-------------------------------------------------------------------------------
53+
54+foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
55+
56+isNonBlocking :: FD -> Bool
57+isNonBlocking fd = fdIsNonBlocking fd /= 0
58+
59+#if !defined(mingw32_HOST_OS)
60+
61+-- "poll"s the fd for data to become available or timeout
62+-- See cbits/inputReady.c in base package
63+foreign import ccall unsafe "fdReady"
64+ unsafe_fdReady :: CInt -> CBool -> Int64 -> CBool -> IO CInt
65+
66+writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
67+writeNonBlocking 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
76+
77+ where
78+
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)
84+
85+writevNonBlocking :: String -> FD -> Ptr IOVec -> Int -> IO CInt
86+writevNonBlocking 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
95+
96+ where
97+
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))
103+
104+#else
105+
106+foreign import WINDOWS_CCONV safe "recv"
107+ c_safe_recv :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
108+
109+foreign import WINDOWS_CCONV safe "send"
110+ c_safe_send :: CInt -> Ptr Word8 -> CInt -> CInt{-flags-} -> IO CInt
111+
112+blockingWriteRawBufferPtr :: String -> FD -> Ptr Word8-> Int -> CSize -> IO CInt
113+blockingWriteRawBufferPtr 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.
131+
132+-- NOTE: "safe" versions of the read/write calls for use by the threaded RTS.
133+-- These calls may block, but that's ok.
134+
135+asyncWriteRawBufferPtr :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
136+asyncWriteRawBufferPtr 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)
145+
146+writeNonBlocking :: String -> FD -> Ptr Word8 -> Int -> CSize -> IO CInt
147+writeNonBlocking loc !fd !buf !off !len
148+ | threaded = blockingWriteRawBufferPtr loc fd buf off len
149+ | otherwise = asyncWriteRawBufferPtr loc fd buf off len
150+
151+#endif
152+
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.
158+write :: FD -> Ptr Word8 -> Int -> CSize -> IO CInt
159+write = writeNonBlocking "Streamly.FileSystem.FDIO"
160+
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.
167+writeAll :: FD -> Ptr Word8 -> Int -> IO ()
168+writeAll 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 ()
174+
175+-------------------------------------------------------------------------------
176+-- Vector IO
177+-------------------------------------------------------------------------------
178+
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.
183+writev :: FD -> Ptr IOVec -> Int -> IO CInt
184+writev = writevNonBlocking "Streamly.FileSystem.FDIO"
185+
186+-- | Keep writing an iovec in a loop until all the iovec entries are written.
187+writevAll :: FD -> Ptr IOVec -> Int -> IO ()
188+writevAll 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 ()
src/Streamly/FileSystem/File.hs+8−8
@@ -63,7 +63,7 @@ module Streamly.FileSystem.File
63 -- -- * Array Read
63 -- -- * Array Read
64 -- , readArrayOf
64 -- , readArrayOf
65
65
66− , readArraysUpto
66+ , readArraysOfUpto
67 -- , readArraysOf
67 -- , readArraysOf
68 , readArrays
68 , readArrays
69
69
@@ -183,26 +183,26 @@ appendArray file arr = SIO.withFile file AppendMode (\h -> FH.writeArray h arr)
183-- Stream of Arrays IO
183-- Stream of Arrays IO
184-------------------------------------------------------------------------------
184-------------------------------------------------------------------------------
185
185
186−-- | @readArraysUpto size file@ reads a stream of arrays from file @file@.
186+-- | @readArraysOfUpto size file@ reads a stream of arrays from file @file@.
187-- The maximum size of a single array is specified by @size@. The actual size
187-- The maximum size of a single array is specified by @size@. The actual size
188-- read may be less than or equal to @size@.
188-- read may be less than or equal to @size@.
189−{-# INLINABLE readArraysUpto #-}
189+{-# INLINABLE readArraysOfUpto #-}
190−readArraysUpto :: (IsStream t, MonadCatch m, MonadIO m)
190+readArraysOfUpto :: (IsStream t, MonadCatch m, MonadIO m)
191 => Int -> FilePath -> t m (Array Word8)
191 => Int -> FilePath -> t m (Array Word8)
192−readArraysUpto size file = withFile file ReadMode (FH.readArraysUpto size)
192+readArraysOfUpto size file = withFile file ReadMode (FH.readArraysOfUpto size)
193
193
194-- XXX read 'Array a' instead of Word8
194-- XXX read 'Array a' instead of Word8
195--
195--
196-- | @readArrays file@ reads a stream of arrays from file @file@.
196-- | @readArrays file@ reads a stream of arrays from file @file@.
197-- The maximum size of a single array is limited to @defaultChunkSize@.
197-- The maximum size of a single array is limited to @defaultChunkSize@.
198--
198--
199−-- > readArrays = readArraysUpto defaultChunkSize
199+-- > readArrays = readArraysOfUpto defaultChunkSize
200--
200--
201-- @since 0.7.0
201-- @since 0.7.0
202{-# INLINE readArrays #-}
202{-# INLINE readArrays #-}
203readArrays :: (IsStream t, MonadCatch m, MonadIO m)
203readArrays :: (IsStream t, MonadCatch m, MonadIO m)
204 => FilePath -> t m (Array Word8)
204 => FilePath -> t m (Array Word8)
205−readArrays = readArraysUpto A.defaultChunkSize
205+readArrays = readArraysOfUpto A.defaultChunkSize
206
206
207-------------------------------------------------------------------------------
207-------------------------------------------------------------------------------
208-- Read File to Stream
208-- Read File to Stream
@@ -219,7 +219,7 @@ readArrays = readArraysUpto A.defaultChunkSize
219--
219--
220{-# INLINE readByChunksUpto #-}
220{-# INLINE readByChunksUpto #-}
221readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
221readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
222−readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
222+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h
223-}
223-}
224
224
225-- TODO
225-- TODO
src/Streamly/FileSystem/Handle.hs+18−20
@@ -53,13 +53,13 @@ module Streamly.FileSystem.Handle
53 -- , readUtf8
53 -- , readUtf8
54 -- , readLines
54 -- , readLines
55 -- , readFrames
55 -- , readFrames
56− -- , readByChunks
56+ , readByChunksUpto
57
57
58 -- -- * Array Read
58 -- -- * Array Read
59 -- , readArrayUpto
59 -- , readArrayUpto
60 -- , readArrayOf
60 -- , readArrayOf
61
61
62− , readArraysUpto
62+ , readArraysOfUpto
63 -- , readArraysOf
63 -- , readArraysOf
64 , readArrays
64 , readArrays
65
65
@@ -68,7 +68,7 @@ module Streamly.FileSystem.Handle
68 -- , writeUtf8
68 -- , writeUtf8
69 -- , writeUtf8ByLines
69 -- , writeUtf8ByLines
70 -- , writeByFrames
70 -- , writeByFrames
71− , writeByChunks
71+ , writeByChunksOf
72
72
73 -- -- * Array Write
73 -- -- * Array Write
74 , writeArray
74 , writeArray
@@ -172,13 +172,13 @@ writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
172-- Stream of Arrays IO
172-- Stream of Arrays IO
173-------------------------------------------------------------------------------
173-------------------------------------------------------------------------------
174
174
175−-- | @readArraysUpto size h@ reads a stream of arrays from file handle @h@.
175+-- | @readArraysOfUpto size h@ reads a stream of arrays from file handle @h@.
176-- The maximum size of a single array is specified by @size@. The actual size
176-- The maximum size of a single array is specified by @size@. The actual size
177-- read may be less than or equal to @size@.
177-- read may be less than or equal to @size@.
178−{-# INLINABLE _readArraysUpto #-}
178+{-# INLINABLE _readArraysOfUpto #-}
179−_readArraysUpto :: (IsStream t, MonadIO m)
179+_readArraysOfUpto :: (IsStream t, MonadIO m)
180 => Int -> Handle -> t m (Array Word8)
180 => Int -> Handle -> t m (Array Word8)
181−_readArraysUpto size h = go
181+_readArraysOfUpto size h = go
182 where
182 where
183 -- XXX use cons/nil instead
183 -- XXX use cons/nil instead
184 go = mkStream $ \_ yld _ stp -> do
184 go = mkStream $ \_ yld _ stp -> do
@@ -187,9 +187,9 @@ _readArraysUpto size h = go
187 then stp
187 then stp
188 else yld arr go
188 else yld arr go
189
189
190−{-# INLINE_NORMAL readArraysUpto #-}
190+{-# INLINE_NORMAL readArraysOfUpto #-}
191−readArraysUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
191+readArraysOfUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
192−readArraysUpto size h = D.fromStreamD (D.Stream step ())
192+readArraysOfUpto size h = D.fromStreamD (D.Stream step ())
193 where
193 where
194 {-# INLINE_LATE step #-}
194 {-# INLINE_LATE step #-}
195 step _ _ = do
195 step _ _ = do
@@ -206,12 +206,12 @@ readArraysUpto size h = D.fromStreamD (D.Stream step ())
206-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
206-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
207-- on the 'Handle'.
207-- on the 'Handle'.
208--
208--
209−-- > readArrays = readArraysUpto defaultChunkSize
209+-- > readArrays = readArraysOfUpto defaultChunkSize
210--
210--
211-- @since 0.7.0
211-- @since 0.7.0
212{-# INLINE readArrays #-}
212{-# INLINE readArrays #-}
213readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
213readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
214−readArrays = readArraysUpto A.defaultChunkSize
214+readArrays = readArraysOfUpto A.defaultChunkSize
215
215
216-------------------------------------------------------------------------------
216-------------------------------------------------------------------------------
217-- Read File to Stream
217-- Read File to Stream
@@ -221,15 +221,13 @@ readArrays = readArraysUpto A.defaultChunkSize
221-- read requests at the same time. For serial case we can use async IO. We can
221-- read requests at the same time. For serial case we can use async IO. We can
222-- also control the read throughput in mbps or IOPS.
222-- also control the read throughput in mbps or IOPS.
223
223
224−{-
225-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file
224-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file
226-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
225-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
227-- as soon as EOF is encountered.
226-- as soon as EOF is encountered.
228--
227--
229{-# INLINE readByChunksUpto #-}
228{-# INLINE readByChunksUpto #-}
230readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
229readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
231−readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
230+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h
232−-}
233
231
234-- TODO
232-- TODO
235-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
233-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
@@ -267,11 +265,11 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m
267-- input elements.
265-- input elements.
268--
266--
269-- @since 0.7.0
267-- @since 0.7.0
270−{-# INLINE writeByChunks #-}
268+{-# INLINE writeByChunksOf #-}
271−writeByChunks :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
269+writeByChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
272−writeByChunks n h m = writeArrays h $ A.arraysOf n m
270+writeByChunksOf n h m = writeArrays h $ A.arraysOf n m
273
271
274−-- > write = 'writeByChunks' A.defaultChunkSize
272+-- > write = 'writeByChunksOf' A.defaultChunkSize
275--
273--
276-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
274-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
277-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
275-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
@@ -280,7 +278,7 @@ writeByChunks n h m = writeArrays h $ A.arraysOf n m
280-- @since 0.7.0
278-- @since 0.7.0
281{-# INLINE write #-}
279{-# INLINE write #-}
282write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
280write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
283−write = writeByChunks A.defaultChunkSize
281+write = writeByChunksOf A.defaultChunkSize
284
282
285{-
283{-
286{-# INLINE write #-}
284{-# INLINE write #-}
src/Streamly/FileSystem/IOVec.hscnew file+66
@@ -0,0 +1,66 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE CApiFFI #-}
4+{-# LANGUAGE ScopedTypeVariables #-}
5+
6+-- |
7+-- Module : Streamly.FileSystem.IOVec
8+-- Copyright : (c) 2019 Composewell Technologies
9+--
10+-- License : BSD3
11+-- Maintainer : harendra.kumar@gmail.com
12+-- Stability : experimental
13+-- Portability : GHC
14+--
15+-- Low level IO routines interfacing the operating system.
16+--
17+
18+module Streamly.FileSystem.IOVec
19+ ( IOVec(..)
20+ , c_writev
21+ , c_safe_writev
22+ )
23+where
24+
25+import Data.Word (Word8, Word64)
26+import Foreign.C.Types (CInt(..))
27+import Foreign.Ptr (Ptr)
28+import Foreign.Storable (Storable(..))
29+import System.Posix.Types (CSsize(..))
30+
31+-------------------------------------------------------------------------------
32+-- IOVec
33+-------------------------------------------------------------------------------
34+
35+#if !defined(mingw32_HOST_OS)
36+
37+#include <sys/uio.h>
38+
39+data IOVec = IOVec
40+ { iovBase :: {-# UNPACK #-} !(Ptr Word8)
41+ , iovLen :: {-# UNPACK #-} !Word64
42+ } deriving (Eq, Show)
43+
44+instance Storable IOVec where
45+ sizeOf _ = #{size struct iovec}
46+ alignment _ = #{alignment struct iovec}
47+ peek ptr = do
48+ base <- #{peek struct iovec, iov_base} ptr
49+ len :: #{type size_t} <- #{peek struct iovec, iov_len} ptr
50+ return $ IOVec base len
51+ poke ptr vec = do
52+ let base = iovBase vec
53+ len :: #{type size_t} = iovLen vec
54+ #{poke struct iovec, iov_base} ptr base
55+ #{poke struct iovec, iov_len} ptr len
56+
57+foreign import capi unsafe "sys/uio.h writev"
58+ c_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize
59+
60+foreign import capi safe "sys/uio.h writev"
61+ c_safe_writev :: CInt -> Ptr IOVec -> CInt -> IO CSsize
62+
63+#else
64+c_writev = error "writev not implemented for windows"
65+c_safe_writev = error "writev not implemented for windows"
66+#endif
src/Streamly/Mem/Array.hs+108−59
@@ -93,27 +93,32 @@ module Streamly.Mem.Array
93 -- , newArray
93 -- , newArray
94 , writeN
94 , writeN
95 , write
95 , write
96− , fromListN
96+ , A.fromListN
97− , fromList
97+ , A.fromList
98
98
99 -- Folds
99 -- Folds
100− , toArrayN
100+ , A.toArrayN
101 -- , toArrays
101 -- , toArrays
102 , toArray
102 , toArray
103
103
104− -- Streams
104+ -- Streams of arrays
105 , arraysOf
105 , arraysOf
106
106
107 -- * Elimination
107 -- * Elimination
108 -- 'GHC.Exts.toList' from "GHC.Exts" can be used to convert an array to a
108 -- 'GHC.Exts.toList' from "GHC.Exts" can be used to convert an array to a
109 -- list.
109 -- list.
110
110
111− , read
111+ , A.read
112 , readRev
112 , readRev
113− , toList
113+ , A.toList
114+
115+ -- Streams of arrays
114 , flattenArrays
116 , flattenArrays
115 -- , flattenArraysRev
117 -- , flattenArraysRev
116− , spliceArrays
118+ , coalesceArrays
119+ , coalesceChunksOf
120+ , unlinesArraysBy
121+ , splitArraysOn
117
122
118 -- * Random Access
123 -- * Random Access
119 , length
124 , length
@@ -123,8 +128,10 @@ module Streamly.Mem.Array
123 {-
128 {-
124 , readSlice
129 , readSlice
125 , readSliceRev
130 , readSliceRev
131+ -}
126
132
127 , writeIndex
133 , writeIndex
134+ {-
128 , writeSlice
135 , writeSlice
129 , writeSliceRev
136 , writeSliceRev
130 -}
137 -}
@@ -140,13 +147,14 @@ where
140import Control.Monad (when)
147import Control.Monad (when)
141import Control.Monad.IO.Class (MonadIO(..))
148import Control.Monad.IO.Class (MonadIO(..))
142import Data.Functor.Identity (Identity)
149import Data.Functor.Identity (Identity)
150+import Data.Word (Word8)
143import Foreign.ForeignPtr (withForeignPtr)
151import Foreign.ForeignPtr (withForeignPtr)
144import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
152import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
145import Foreign.Ptr (minusPtr, plusPtr, castPtr)
153import Foreign.Ptr (minusPtr, plusPtr, castPtr)
146import Foreign.Storable (Storable(..))
154import Foreign.Storable (Storable(..))
147import Prelude hiding (length, null, last, map, (!!), read)
155import Prelude hiding (length, null, last, map, (!!), read)
148
156
149−import Streamly.Mem.Array.Types hiding (flattenArrays, newArray)
157+import Streamly.Mem.Array.Types (Array(..), length)
150import Streamly.Streams.Serial (SerialT)
158import Streamly.Streams.Serial (SerialT)
151import Streamly.Streams.StreamK.Type (IsStream)
159import Streamly.Streams.StreamK.Type (IsStream)
152import Streamly.Fold.Types (Fold(..))
160import Streamly.Fold.Types (Fold(..))
@@ -178,7 +186,7 @@ newArray len = undefined
178writeN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)
186writeN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)
179writeN n m = do
187writeN n m = do
180 if n < 0 then error "writeN: negative write count specified" else return ()
188 if n < 0 then error "writeN: negative write count specified" else return ()
181− fromStreamDN n $ D.toStreamD m
189+ A.fromStreamDN n $ D.toStreamD m
182
190
183-------------------------------------------------------------------------------
191-------------------------------------------------------------------------------
184-- Elimination
192-- Elimination
@@ -189,7 +197,7 @@ writeN n m = do
189-- @since 0.7.0
197-- @since 0.7.0
190{-# INLINE_EARLY readRev #-}
198{-# INLINE_EARLY readRev #-}
191readRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
199readRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
192−readRev = D.fromStreamD . toStreamDRev
200+readRev = D.fromStreamD . A.toStreamDRev
193-- XXX add fallback to StreamK rule
201-- XXX add fallback to StreamK rule
194-- {-# RULES "Streamly.Array.readRev fallback to StreamK" [1]
202-- {-# RULES "Streamly.Array.readRev fallback to StreamK" [1]
195-- forall a. S.toStreamK (readRev a) = K.revFromArray a #-}
203-- forall a. S.toStreamK (readRev a) = K.revFromArray a #-}
@@ -199,7 +207,7 @@ _null :: Storable a => Array a -> Bool
199_null arr = length arr == 0
207_null arr = length arr == 0
200
208
201{-# INLINE _last #-}
209{-# INLINE _last #-}
202−_last :: forall a. Storable a => Array a -> Maybe a
210+_last :: (MonadIO m, Storable a) => Array a -> m (Maybe a)
203_last arr = readIndex arr (length arr - 1)
211_last arr = readIndex arr (length arr - 1)
204
212
205-------------------------------------------------------------------------------
213-------------------------------------------------------------------------------
@@ -306,11 +314,12 @@ foldbWith level f = undefined
306--
314--
307-- @since 0.7.0
315-- @since 0.7.0
308{-# INLINE readIndex #-}
316{-# INLINE readIndex #-}
309−readIndex :: Storable a => Array a -> Int -> Maybe a
317+readIndex :: (MonadIO m, Storable a) => Array a -> Int -> m (Maybe a)
310readIndex arr i =
318readIndex arr i =
311 if i < 0 || i > length arr - 1
319 if i < 0 || i > length arr - 1
312− then Nothing
320+ then return Nothing
313− else Just $ unsafeIndex arr i
321+ else liftIO $ withForeignPtr (aStart arr) $ \p ->
322+ fmap Just $ peekElemOff p i
314
323
315{-
324{-
316-- | @readSlice arr i count@ streams a slice of the array @arr@ starting
325-- | @readSlice arr i count@ streams a slice of the array @arr@ starting
@@ -332,14 +341,25 @@ readSlice arr i len = undefined
332readSliceRev :: (IsStream t, Monad m, Storable a)
341readSliceRev :: (IsStream t, Monad m, Storable a)
333 => Array a -> Int -> Int -> t m a
342 => Array a -> Int -> Int -> t m a
334readSliceRev arr i len = undefined
343readSliceRev arr i len = undefined
344+-}
335
345
336-- | /O(1)/ Write the given element at the given index in the array.
346-- | /O(1)/ Write the given element at the given index in the array.
337--
347--
338-- @since 0.7.0
348-- @since 0.7.0
339{-# INLINE writeIndex #-}
349{-# INLINE writeIndex #-}
340writeIndex :: (MonadIO m, Storable a) => Array a -> Int -> a -> m ()
350writeIndex :: (MonadIO m, Storable a) => Array a -> Int -> a -> m ()
341−writeIndex arr i a = undefined
351+writeIndex arr i a = do
352+ let maxIndex = length arr - 1
353+ if i < 0
354+ then error "writeIndex: negative array index"
355+ else if i > maxIndex
356+ then error $ "writeIndex: specified array index " ++ show i
357+ ++ " is beyond the maximum index " ++ show maxIndex
358+ else
359+ liftIO $ withForeignPtr (aStart arr) $ \p ->
360+ pokeElemOff p i a
342
361
362+{-
343-- | @writeSlice arr i count stream@ writes a stream to the array @arr@
363-- | @writeSlice arr i count stream@ writes a stream to the array @arr@
344-- starting at index @i@ and writing up to @count@ elements in the forward
364-- starting at index @i@ and writing up to @count@ elements in the forward
345-- direction ending at the index @i + count - 1@.
365-- direction ending at the index @i + count - 1@.
@@ -381,7 +401,7 @@ toArraysInRange low high (Fold step initial extract) =
381{-# INLINE _toArraysOf #-}
401{-# INLINE _toArraysOf #-}
382_toArraysOf :: (MonadIO m, Storable a)
402_toArraysOf :: (MonadIO m, Storable a)
383 => Int -> Fold m a (SerialT Identity (Array a))
403 => Int -> Fold m a (SerialT Identity (Array a))
384−_toArraysOf n = FL.lchunksOf n (toArrayN n) FL.toStream
404+_toArraysOf n = FL.lchunksOf n (A.toArrayN n) FL.toStream
385
405
386-- XXX The realloc based implementation needs to make one extra copy if we use
406-- XXX The realloc based implementation needs to make one extra copy if we use
387-- shrinkToFit. On the other hand, the stream of arrays implementation may
407-- shrinkToFit. On the other hand, the stream of arrays implementation may
@@ -401,7 +421,7 @@ bytesToCount x n =
401 let elemSize = sizeOf x
421 let elemSize = sizeOf x
402 in n + elemSize - 1 `div` elemSize
422 in n + elemSize - 1 `div` elemSize
403
423
404−{-# INLINE toArrayMinChunk #-}
424+{-# INLINE_NORMAL toArrayMinChunk #-}
405toArrayMinChunk :: forall m a. (MonadIO m, Storable a)
425toArrayMinChunk :: forall m a. (MonadIO m, Storable a)
406 => Int -> Fold m a (Array a)
426 => Int -> Fold m a (Array a)
407-- toArrayMinChunk n = FL.mapM spliceArrays $ toArraysOf n
427-- toArrayMinChunk n = FL.mapM spliceArrays $ toArraysOf n
@@ -416,11 +436,14 @@ toArrayMinChunk elemCount = Fold step initial extract
416 initial = do
436 initial = do
417 when (elemCount < 0) $ error "toArrayMinChunk: elemCount is negative"
437 when (elemCount < 0) $ error "toArrayMinChunk: elemCount is negative"
418 liftIO $ A.newArray elemCount
438 liftIO $ A.newArray elemCount
419− step arr@(Array _ end bound) x | end == bound = do
439+ step arr@(Array start end bound) x | end == bound = do
440+ let p = unsafeForeignPtrToPtr start
441+ oldSize = end `minusPtr` p
442+ newSize = max (oldSize * 2) 1
420− arr1 <- liftIO $ reallocDouble 1 arr
443+ arr1 <- liftIO $ A.realloc newSize arr
421 insertElem arr1 x
444 insertElem arr1 x
422 step arr x = insertElem arr x
445 step arr x = insertElem arr x
423− extract = liftIO . shrinkToFit
446+ extract = liftIO . A.shrinkToFit
424
447
425-- | Fold the whole input to a single array.
448-- | Fold the whole input to a single array.
426--
449--
@@ -429,7 +452,7 @@ toArrayMinChunk elemCount = Fold step initial extract
429-- @since 0.7.0
452-- @since 0.7.0
430{-# INLINE toArray #-}
453{-# INLINE toArray #-}
431toArray :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)
454toArray :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)
432−toArray = toArrayMinChunk (bytesToCount (undefined :: a) (mkChunkSize 1024))
455+toArray = toArrayMinChunk (bytesToCount (undefined :: a) (A.mkChunkSize 1024))
433
456
434-- | Convert a stream of arrays into a stream of their elements.
457-- | Convert a stream of arrays into a stream of their elements.
435--
458--
@@ -449,6 +472,39 @@ flattenArrays m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)
449_flattenArraysRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
472_flattenArraysRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
450_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
473_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
451
474
475+-- XXX use an Array instead as separator? Or use a separate unlinesArraysBySeq
476+-- API for that?
477+--
478+-- | Flatten a stream of arrays appending the given element after each
479+-- array.
480+--
481+-- @since 0.7.0
482+{-# INLINE unlinesArraysBy #-}
483+unlinesArraysBy :: (MonadIO m, IsStream t, Storable a)
484+ => a -> t m (Array a) -> t m a
485+unlinesArraysBy x = D.fromStreamD . A.unlines x . D.toStreamD
486+
487+-- | Split a stream of arrays on a given separator byte, dropping the separator
488+-- and coalescing all the arrays between two separators into a single array.
489+--
490+-- @since 0.7.0
491+{-# INLINE splitArraysOn #-}
492+splitArraysOn
493+ :: (IsStream t, MonadIO m)
494+ => Word8
495+ -> t m (Array Word8)
496+ -> t m (Array Word8)
497+splitArraysOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s
498+
499+-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a
500+-- maximum specified size.
501+--
502+-- @since 0.7.0
503+{-# INLINE coalesceChunksOf #-}
504+coalesceChunksOf :: (MonadIO m, Storable a)
505+ => Int -> SerialT m (Array a) -> SerialT m (Array a)
506+coalesceChunksOf n xs = D.fromStreamD $ A.coalesceChunksOf n (D.toStreamD xs)
507+
452-- |
508-- |
453-- > arraysOf n = FL.groupsOf n (FL.toArrayN n)
509-- > arraysOf n = FL.groupsOf n (FL.toArrayN n)
454--
510--
@@ -459,18 +515,17 @@ _flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
459arraysOf :: (IsStream t, MonadIO m, Storable a)
515arraysOf :: (IsStream t, MonadIO m, Storable a)
460 => Int -> t m a -> t m (Array a)
516 => Int -> t m a -> t m (Array a)
461arraysOf n str =
517arraysOf n str =
462− D.fromStreamD $ fromStreamDArraysOf n (D.toStreamD str)
518+ D.fromStreamD $ A.fromStreamDArraysOf n (D.toStreamD str)
463
519
464-- XXX Both of these implementations of splicing seem to perform equally well.
520-- XXX Both of these implementations of splicing seem to perform equally well.
465-- We need to perform benchmarks over a range of sizes though.
521-- We need to perform benchmarks over a range of sizes though.
466
522
467−{-# INLINE _spliceArraysRealloced #-}
523+-- CAUTION! length must more than equal to lengths of all the arrays in the
524+-- stream.
525+{-# INLINE spliceArraysLenUnsafe #-}
468−_spliceArraysRealloced :: (MonadIO m, Storable a)
526+spliceArraysLenUnsafe :: (MonadIO m, Storable a)
469− => SerialT m (Array a) -> m (Array a)
527+ => Int -> SerialT m (Array a) -> m (Array a)
470−_spliceArraysRealloced s = do
528+spliceArraysLenUnsafe len buffered = do
471− buffered <- P.foldr S.cons S.nil s
472− len <- S.sum (S.map length buffered)
473−
474 arr <- liftIO $ A.newArray len
529 arr <- liftIO $ A.newArray len
475 end <- S.foldlM' writeArr (aEnd arr) buffered
530 end <- S.foldlM' writeArr (aEnd arr) buffered
476 return $ arr {aEnd = end}
531 return $ arr {aEnd = end}
@@ -479,42 +534,36 @@ _spliceArraysRealloced s = do
479
534
480 writeArr dst Array{..} =
535 writeArr dst Array{..} =
481 liftIO $ withForeignPtr aStart $ \src -> do
536 liftIO $ withForeignPtr aStart $ \src -> do
482− let len = aEnd `minusPtr` src
537+ let count = aEnd `minusPtr` src
483− memcpy (castPtr dst) (castPtr src) len
538+ A.memcpy (castPtr dst) (castPtr src) count
484− return $ dst `plusPtr` len
539+ return $ dst `plusPtr` count
485
540
486−{-# INLINE spliceArraysBuffered #-}
541+{-# INLINE _spliceArraysBuffered #-}
487−spliceArraysBuffered :: forall m a. (MonadIO m, Storable a)
542+_spliceArraysBuffered :: (MonadIO m, Storable a)
488 => SerialT m (Array a) -> m (Array a)
543 => SerialT m (Array a) -> m (Array a)
489−spliceArraysBuffered s = do
544+_spliceArraysBuffered s = do
490− idst <- liftIO $ A.newArray (bytesToCount (undefined :: a)
545+ buffered <- P.foldr S.cons S.nil s
491− (mkChunkSizeKB 4))
546+ len <- S.sum (S.map length buffered)
492− arr <- S.foldlM' appendArr idst s
547+ spliceArraysLenUnsafe len s
493− liftIO $ shrinkToFit arr
494−
495− where
496
548
497− appendArr dst@(Array _ end bound) src = liftIO $ do
549+{-# INLINE spliceArraysRealloced #-}
498− let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
550+spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)
499− dst1 <-
551+ => SerialT m (Array a) -> m (Array a)
500− if end `plusPtr` srcLen >= bound
552+spliceArraysRealloced s = do
501− then reallocDouble srcLen dst
553+ idst <- liftIO $ A.newArray (bytesToCount (undefined :: a)
502− else return dst
554+ (A.mkChunkSizeKB 4))
503
555
504− withForeignPtr (aStart dst1) $ \_ -> do
556+ arr <- S.foldlM' A.spliceWithDoubling idst s
505− withForeignPtr (aStart src) $ \psrc -> do
557+ liftIO $ A.shrinkToFit arr
506− let pdst = aEnd dst1
507− memcpy (castPtr pdst) (castPtr psrc) srcLen
508− return $ dst1 { aEnd = pdst `plusPtr` srcLen }
509
558
510-- | Given a stream of arrays, splice them all together to generate a single
559-- | Given a stream of arrays, splice them all together to generate a single
511-- array. The stream must be /finite/.
560-- array. The stream must be /finite/.
512--
561--
513-- @since 0.7.0
562-- @since 0.7.0
514−{-# INLINE spliceArrays #-}
563+{-# INLINE coalesceArrays #-}
515−spliceArrays :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)
564+coalesceArrays :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)
516−spliceArrays = spliceArraysBuffered
565+coalesceArrays = spliceArraysRealloced
517−-- spliceArrays = _spliceArraysRealloced
566+-- spliceArrays = _spliceArraysBuffered
518
567
519-- | Create an 'Array' from a stream. This is useful when we want to create a
568-- | Create an 'Array' from a stream. This is useful when we want to create a
520-- single array from a stream of unknown size. 'writeN' is at least twice
569-- single array from a stream of unknown size. 'writeN' is at least twice
@@ -548,18 +597,18 @@ write = FL.foldl' toArray
548{-# INLINE transformWith #-}
597{-# INLINE transformWith #-}
549transformWith :: (MonadIO m, Storable a, Storable b)
598transformWith :: (MonadIO m, Storable a, Storable b)
550 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
599 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
551−transformWith f arr = FL.foldl' (toArrayMinChunk (length arr)) $ f (read arr)
600+transformWith f arr = FL.foldl' (toArrayMinChunk (length arr)) $ f (A.read arr)
552
601
553-- | Fold an array using a 'Fold'.
602-- | Fold an array using a 'Fold'.
554--
603--
555-- @since 0.7.0
604-- @since 0.7.0
556{-# INLINE foldArray #-}
605{-# INLINE foldArray #-}
557foldArray :: (MonadIO m, Storable a) => Fold m a b -> Array a -> m b
606foldArray :: (MonadIO m, Storable a) => Fold m a b -> Array a -> m b
558−foldArray f arr = FL.foldl' f (read arr)
607+foldArray f arr = FL.foldl' f (A.read arr)
559
608
560-- | Fold an array using a stream fold operation.
609-- | Fold an array using a stream fold operation.
561--
610--
562-- @since 0.7.0
611-- @since 0.7.0
563{-# INLINE foldWith #-}
612{-# INLINE foldWith #-}
564foldWith :: (MonadIO m, Storable a) => (SerialT m a -> m b) -> Array a -> m b
613foldWith :: (MonadIO m, Storable a) => (SerialT m a -> m b) -> Array a -> m b
565−foldWith f arr = f (read arr)
614+foldWith f arr = f (A.read arr)
src/Streamly/Mem/Array/Types.hs+332−47
@@ -23,29 +23,34 @@ module Streamly.Mem.Array.Types
23 Array (..)
23 Array (..)
24
24
25 -- * Construction
25 -- * Construction
26− , unsafeInlineIO
27 , withNewArray
26 , withNewArray
28 , newArray
27 , newArray
29 , unsafeSnoc
28 , unsafeSnoc
30 , snoc
29 , snoc
31− , shrinkToFit
30+ , spliceWithDoubling
32− , memcpy
33− , memcmp
34
31
35 , fromList
32 , fromList
36 , fromListN
33 , fromListN
37 , fromStreamDN
34 , fromStreamDN
38 -- , fromStreamD
35 -- , fromStreamD
36+
37+ -- * Streams of arrays
39 , fromStreamDArraysOf
38 , fromStreamDArraysOf
40 , flattenArrays
39 , flattenArrays
41 , flattenArraysRev
40 , flattenArraysRev
41+ , coalesceChunksOf
42+ , groupIOVecsOf
43+ , splitOn
42
44
43 -- * Elimination
45 -- * Elimination
44 , unsafeIndexIO
46 , unsafeIndexIO
45 , unsafeIndex
47 , unsafeIndex
46 , length
48 , length
49+ , byteLength
50+ , byteCapacity
47 , foldl'
51 , foldl'
48 , foldr
52 , foldr
53+ , splitAt
49
54
50 , toStreamD
55 , toStreamD
51 , toStreamDRev
56 , toStreamDRev
@@ -59,7 +64,11 @@ module Streamly.Mem.Array.Types
59 , defaultChunkSize
64 , defaultChunkSize
60 , mkChunkSize
65 , mkChunkSize
61 , mkChunkSizeKB
66 , mkChunkSizeKB
62− , reallocDouble
67+ , unsafeInlineIO
68+ , realloc
69+ , shrinkToFit
70+ , memcpy
71+ , memcmp
63
72
64 , unlines
73 , unlines
65 )
74 )
@@ -74,11 +83,11 @@ import Data.Word (Word8)
74import Foreign.C.String (CString)
83import Foreign.C.String (CString)
75import Foreign.C.Types (CSize(..), CInt(..))
84import Foreign.C.Types (CSize(..), CInt(..))
76import Foreign.ForeignPtr
85import Foreign.ForeignPtr
77− (ForeignPtr, withForeignPtr, touchForeignPtr)
86+ (ForeignPtr, withForeignPtr, touchForeignPtr, plusForeignPtr)
78import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
87import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
79−import Foreign.Ptr (plusPtr, minusPtr, castPtr)
88+import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)
80import Foreign.Storable (Storable(..))
89import Foreign.Storable (Storable(..))
81−import Prelude hiding (length, foldr, read, unlines, unwords)
90+import Prelude hiding (length, foldr, read, unlines, splitAt)
82import Text.Read (readPrec, readListPrec, readListPrecDefault)
91import Text.Read (readPrec, readListPrec, readListPrecDefault)
83
92
84import GHC.Base (Addr#, realWorld#)
93import GHC.Base (Addr#, realWorld#)
@@ -89,6 +98,7 @@ import GHC.Ptr (Ptr(..))
89
98
90import Streamly.Fold.Types (Fold(..))
99import Streamly.Fold.Types (Fold(..))
91import Streamly.SVar (adaptState)
100import Streamly.SVar (adaptState)
101+import Streamly.FileSystem.FDIO (IOVec(..))
92
102
93import qualified Streamly.Streams.StreamD.Type as D
103import qualified Streamly.Streams.StreamD.Type as D
94import qualified Streamly.Streams.StreamK as K
104import qualified Streamly.Streams.StreamK as K
@@ -165,6 +175,9 @@ foreign import ccall unsafe "string.h memcpy" c_memcpy
165foreign import ccall unsafe "string.h strlen" c_strlen
175foreign import ccall unsafe "string.h strlen" c_strlen
166 :: CString -> IO CSize
176 :: CString -> IO CSize
167
177
178+foreign import ccall unsafe "string.h memchr" c_memchr
179+ :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)
180+
168-- XXX we are converting Int to CSize
181-- XXX we are converting Int to CSize
169memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
182memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
170memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
183memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
@@ -222,14 +235,14 @@ withNewArray count f = do
222unsafeSnoc :: forall a. Storable a => Array a -> a -> IO (Array a)
235unsafeSnoc :: forall a. Storable a => Array a -> a -> IO (Array a)
223unsafeSnoc arr@Array{..} x = do
236unsafeSnoc arr@Array{..} x = do
224 when (aEnd == aBound) $
237 when (aEnd == aBound) $
225− error "BUG: unsafeAppend: writing beyond array bounds"
238+ error "BUG: unsafeSnoc: writing beyond array bounds"
226 poke aEnd x
239 poke aEnd x
227 touchForeignPtr aStart
240 touchForeignPtr aStart
228 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
241 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
229
242
230{-# INLINE snoc #-}
243{-# INLINE snoc #-}
231−snoc :: forall a. Storable a => Array a -> a -> Array a
244+snoc :: forall a. Storable a => Array a -> a -> IO (Array a)
232−snoc arr@Array {..} x = unsafePerformIO $
245+snoc arr@Array {..} x = do
233 if (aEnd == aBound)
246 if (aEnd == aBound)
234 then do
247 then do
235 let oldStart = unsafeForeignPtrToPtr aStart
248 let oldStart = unsafeForeignPtrToPtr aStart
@@ -250,37 +263,14 @@ snoc arr@Array {..} x = unsafePerformIO $
250 touchForeignPtr aStart
263 touchForeignPtr aStart
251 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
264 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
252
265
253−
266+-- | Reallocate the array to the specified size in bytes. If the size is less
254−-- | Remove the free space from an Array.
267+-- than the original array the array gets truncated.
255−shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
256−shrinkToFit arr@Array{..} = do
257− assert (aEnd <= aBound) (return ())
258− if aEnd /= aBound
259− then do
260− let oldStart = unsafeForeignPtrToPtr aStart
261− let size = aEnd `minusPtr` oldStart
262− newPtr <- mallocPlainForeignPtrAlignedBytes
263− size (alignment (undefined :: a))
264− withForeignPtr newPtr $ \pNew -> do
265− memcpy (castPtr pNew) (castPtr oldStart) size
266− touchForeignPtr aStart
267− let end = pNew `plusPtr` size
268− return $ Array
269− { aStart = newPtr
270− , aEnd = end
271− , aBound = end
272− }
273− else return arr
274−
275−-- | Expand the free space in the array doubling the size with a minimum
276−-- exapnsion of the specified amount of bytes.
277−{-# NOINLINE reallocDouble #-}
268+{-# NOINLINE realloc #-}
278−reallocDouble :: forall a. Storable a => Int -> Array a -> IO (Array a)
269+realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
279−reallocDouble minIncrease Array{..} = do
270+realloc newSize Array{..} = do
280 assert (aEnd <= aBound) (return ())
271 assert (aEnd <= aBound) (return ())
281 let oldStart = unsafeForeignPtrToPtr aStart
272 let oldStart = unsafeForeignPtrToPtr aStart
282 let size = aEnd `minusPtr` oldStart
273 let size = aEnd `minusPtr` oldStart
283− newSize = max (size * 2) (size + minIncrease)
284 newPtr <- mallocPlainForeignPtrAlignedBytes
274 newPtr <- mallocPlainForeignPtrAlignedBytes
285 newSize (alignment (undefined :: a))
275 newSize (alignment (undefined :: a))
286 withForeignPtr newPtr $ \pNew -> do
276 withForeignPtr newPtr $ \pNew -> do
@@ -292,6 +282,17 @@ reallocDouble minIncrease Array{..} = do
292 , aBound = pNew `plusPtr` newSize
282 , aBound = pNew `plusPtr` newSize
293 }
283 }
294
284
285+-- | Remove the free space from an Array.
286+shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
287+shrinkToFit arr@Array{..} = do
288+ assert (aEnd <= aBound) (return ())
289+ if aEnd /= aBound
290+ then do
291+ let oldStart = unsafeForeignPtrToPtr aStart
292+ let size = aEnd `minusPtr` oldStart
293+ realloc size arr
294+ else return arr
295+
295-- XXX when converting an array of Word8 from a literal string we can simply
296-- XXX when converting an array of Word8 from a literal string we can simply
296-- refer to the literal string. Is it possible to write rules such that
297-- refer to the literal string. Is it possible to write rules such that
297-- fromList Word8 can be rewritten so that GHC does not first convert the
298-- fromList Word8 can be rewritten so that GHC does not first convert the
@@ -340,15 +341,30 @@ unsafeIndexIO Array {..} i =
340unsafeIndex :: forall a. Storable a => Array a -> Int -> a
341unsafeIndex :: forall a. Storable a => Array a -> Int -> a
341unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
342unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
342
343
344+-- | /O(1)/ Get the byte length of the array.
345+--
346+-- @since 0.7.0
347+{-# INLINE byteLength #-}
348+byteLength :: Array a -> Int
349+byteLength Array{..} =
350+ let p = unsafeForeignPtrToPtr aStart
351+ len = aEnd `minusPtr` p
352+ in assert (len >= 0) len
353+
343−-- | /O(1)/ Get the length of the array.
354+-- | /O(1)/ Get the length of the array i.e. the number of elements in the
355+-- array.
344--
356--
345-- @since 0.7.0
357-- @since 0.7.0
346{-# INLINE length #-}
358{-# INLINE length #-}
347length :: forall a. Storable a => Array a -> Int
359length :: forall a. Storable a => Array a -> Int
348−length Array{..} =
360+length arr = byteLength arr `div` sizeOf (undefined :: a)
361+
362+{-# INLINE byteCapacity #-}
363+byteCapacity :: Array a -> Int
364+byteCapacity Array{..} =
349 let p = unsafeForeignPtrToPtr aStart
365 let p = unsafeForeignPtrToPtr aStart
350− aLen = aEnd `minusPtr` p
366+ len = aBound `minusPtr` p
351− in assert (aLen >= 0) (aLen `div` sizeOf (undefined :: a))
367+ in assert (len >= 0) len
352
368
353{-# INLINE_NORMAL toStreamD #-}
369{-# INLINE_NORMAL toStreamD #-}
354toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
370toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
@@ -729,8 +745,9 @@ defaultChunkSize :: Int
729defaultChunkSize = mkChunkSizeKB 32
745defaultChunkSize = mkChunkSizeKB 32
730
746
731{-# INLINE_NORMAL unlines #-}
747{-# INLINE_NORMAL unlines #-}
748+unlines :: forall m a. (MonadIO m, Storable a)
732−unlines :: MonadIO m => D.Stream m (Array Char) -> D.Stream m Char
749+ => a -> D.Stream m (Array a) -> D.Stream m a
733−unlines (D.Stream step state) = D.Stream step' (OuterLoop state)
750+unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
734 where
751 where
735 {-# INLINE_LATE step' #-}
752 {-# INLINE_LATE step' #-}
736 step' gst (OuterLoop st) = do
753 step' gst (OuterLoop st) = do
@@ -743,7 +760,7 @@ unlines (D.Stream step state) = D.Stream step' (OuterLoop state)
743 D.Stop -> D.Stop
760 D.Stop -> D.Stop
744
761
745 step' _ (InnerLoop st _ p end) | p == end =
762 step' _ (InnerLoop st _ p end) | p == end =
746− return $ D.Yield '\n' $ OuterLoop st
763+ return $ D.Yield sep $ OuterLoop st
747
764
748 step' _ (InnerLoop st startf p end) = do
765 step' _ (InnerLoop st startf p end) = do
749 x <- liftIO $ do
766 x <- liftIO $ do
@@ -751,4 +768,272 @@ unlines (D.Stream step state) = D.Stream step' (OuterLoop state)
751 touchForeignPtr startf
768 touchForeignPtr startf
752 return r
769 return r
753 return $ D.Yield x (InnerLoop st startf
770 return $ D.Yield x (InnerLoop st startf
754− (p `plusPtr` (sizeOf (undefined :: Char))) end)
771+ (p `plusPtr` (sizeOf (undefined :: a))) end)
772+
773+{-# INLINE spliceTwo #-}
774+spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
775+spliceTwo arr1 arr2 = do
776+ let src1 = unsafeForeignPtrToPtr (aStart arr1)
777+ src2 = unsafeForeignPtrToPtr (aStart arr2)
778+ len1 = aEnd arr1 `minusPtr` src1
779+ len2 = aEnd arr2 `minusPtr` src2
780+
781+ arr <- liftIO $ newArray (len1 + len2)
782+ let dst = unsafeForeignPtrToPtr (aStart arr)
783+
784+ liftIO $ do
785+ memcpy (castPtr dst) (castPtr src1) len1
786+ touchForeignPtr (aStart arr1)
787+ memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
788+ touchForeignPtr (aStart arr2)
789+ return arr { aEnd = dst `plusPtr` (len1 + len2) }
790+
791+-- Splice a new array into a pre-reserved array. The user must ensure that
792+-- there is enough space in the array.
793+{-# INLINE spliceWith #-}
794+spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
795+spliceWith dst@(Array _ end bound) src = liftIO $ do
796+ let srcLen = byteLength src
797+ if end `plusPtr` srcLen > bound
798+ then error "Bug: spliceIntoUnsafe: Not enough space in the target array"
799+ else
800+ withForeignPtr (aStart dst) $ \_ -> do
801+ withForeignPtr (aStart src) $ \psrc -> do
802+ let pdst = aEnd dst
803+ memcpy (castPtr pdst) (castPtr psrc) srcLen
804+ return $ dst { aEnd = pdst `plusPtr` srcLen }
805+
806+-- Splice a new array into a preallocated array, doubling the space if there is
807+-- no space in the target array.
808+{-# INLINE spliceWithDoubling #-}
809+spliceWithDoubling :: (MonadIO m, Storable a)
810+ => Array a -> Array a -> m (Array a)
811+spliceWithDoubling dst@(Array start end bound) src = do
812+ assert (end <= bound) (return ())
813+ let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
814+
815+ dst1 <-
816+ if end `plusPtr` srcLen >= bound
817+ then do
818+ let oldStart = unsafeForeignPtrToPtr start
819+ oldSize = end `minusPtr` oldStart
820+ newSize = max (oldSize * 2) (oldSize + srcLen)
821+ liftIO $ realloc newSize dst
822+ else return dst
823+ spliceWith dst1 src
824+
825+data SpliceState s arr
826+ = SpliceInitial s
827+ | SpliceBuffering s arr
828+ | SpliceYielding arr (SpliceState s arr)
829+ | SpliceFinish
830+
831+-- XXX can use general grouping combinators to achieve this?
832+-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a
833+-- maximum specified size.
834+--
835+-- @since 0.7.0
836+{-# INLINE_NORMAL coalesceChunksOf #-}
837+coalesceChunksOf :: (MonadIO m, Storable a)
838+ => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
839+coalesceChunksOf n (D.Stream step state) = D.Stream step' (SpliceInitial state)
840+
841+ where
842+
843+ {-# INLINE_LATE step' #-}
844+ step' gst (SpliceInitial st) = do
845+ r <- step gst st
846+ case r of
847+ D.Yield arr s -> return $
848+ let len = byteLength arr
849+ in if len >= n
850+ then D.Skip (SpliceYielding arr (SpliceInitial s))
851+ else D.Skip (SpliceBuffering s arr)
852+ D.Skip s -> return $ D.Skip (SpliceInitial s)
853+ D.Stop -> return $ D.Stop
854+
855+ step' gst (SpliceBuffering st buf) = do
856+ r <- step gst st
857+ case r of
858+ D.Yield arr s -> do
859+ let len = byteLength buf + byteLength arr
860+ if len > n
861+ then return $ D.Skip (SpliceYielding buf (SpliceBuffering s arr))
862+ else do
863+ buf' <- if byteCapacity buf < n
864+ then liftIO $ realloc n buf
865+ else return buf
866+ buf'' <- spliceWith buf' arr
867+ return $ D.Skip (SpliceBuffering s buf'')
868+ D.Skip s -> return $ D.Skip (SpliceBuffering s buf)
869+ D.Stop -> return $ D.Skip (SpliceYielding buf SpliceFinish)
870+
871+ step' _ SpliceFinish = return D.Stop
872+
873+ step' _ (SpliceYielding arr next) = return $ D.Yield arr next
874+
875+data GatherState s arr
876+ = GatherInitial s
877+ | GatherBuffering s arr Int
878+ | GatherYielding arr (GatherState s arr)
879+ | GatherFinish
880+
881+-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
882+-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
883+-- each array and a maximum of @maxEntries@ entries in each array.
884+--
885+-- @since 0.7.0
886+{-# INLINE_NORMAL groupIOVecsOf #-}
887+groupIOVecsOf :: MonadIO m
888+ => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
889+groupIOVecsOf n maxIOVLen (D.Stream step state) =
890+ D.Stream step' (GatherInitial state)
891+
892+ where
893+
894+ {-# INLINE_LATE step' #-}
895+ step' gst (GatherInitial st) = do
896+ r <- step (adaptState gst) st
897+ case r of
898+ D.Yield arr s -> do
899+ let p = unsafeForeignPtrToPtr (aStart arr)
900+ len = byteLength arr
901+ iov <- liftIO $ newArray maxIOVLen
902+ iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
903+ (fromIntegral len))
904+ if len >= n
905+ then return $ D.Skip (GatherYielding iov' (GatherInitial s))
906+ else return $ D.Skip (GatherBuffering s iov' len)
907+ D.Skip s -> return $ D.Skip (GatherInitial s)
908+ D.Stop -> return $ D.Stop
909+
910+ step' gst (GatherBuffering st iov len) = do
911+ r <- step (adaptState gst) st
912+ case r of
913+ D.Yield arr s -> do
914+ let p = unsafeForeignPtrToPtr (aStart arr)
915+ alen = byteLength arr
916+ len' = len + alen
917+ if len' > n || length iov >= maxIOVLen
918+ then do
919+ iov' <- liftIO $ newArray maxIOVLen
920+ iov'' <- liftIO $ unsafeSnoc iov' (IOVec (castPtr p)
921+ (fromIntegral alen))
922+ return $ D.Skip (GatherYielding iov
923+ (GatherBuffering s iov'' alen))
924+ else do
925+ iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
926+ (fromIntegral alen))
927+ return $ D.Skip (GatherBuffering s iov' len')
928+ D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
929+ D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
930+
931+ step' _ GatherFinish = return D.Stop
932+
933+ step' _ (GatherYielding iov next) = return $ D.Yield iov next
934+
935+-- | Create two slices of an array without copying the original array. The
936+-- specified index @i@ is the first index of the second slice.
937+--
938+-- @since 0.7.0
939+splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
940+splitAt i arr@Array{..} =
941+ let maxIndex = length arr - 1
942+ in if i < 0
943+ then error "sliceAt: negative array index"
944+ else if i > maxIndex
945+ then error $ "sliceAt: specified array index " ++ show i
946+ ++ " is beyond the maximum index " ++ show maxIndex
947+ else let off = i * sizeOf (undefined :: a)
948+ p = unsafeForeignPtrToPtr aStart `plusPtr` off
949+ in ( Array
950+ { aStart = aStart
951+ , aEnd = p
952+ , aBound = p
953+ }
954+ , Array
955+ { aStart = aStart `plusForeignPtr` off
956+ , aEnd = aEnd
957+ , aBound = aBound
958+ }
959+ )
960+
961+-- Drops the separator byte
962+{-# INLINE breakOn #-}
963+breakOn :: MonadIO m
964+ => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
965+breakOn sep arr@Array{..} = liftIO $ do
966+ let p = unsafeForeignPtrToPtr aStart
967+ loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
968+ return $
969+ if loc == nullPtr
970+ then (arr, Nothing)
971+ else
972+ ( Array
973+ { aStart = aStart
974+ , aEnd = loc
975+ , aBound = loc
976+ }
977+ , Just $ Array
978+ { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
979+ , aEnd = aEnd
980+ , aBound = aBound
981+ }
982+ )
983+
984+data SplitState s arr
985+ = Initial s
986+ | Buffering s arr
987+ | Splitting s arr
988+ | Yielding arr (SplitState s arr)
989+ | Finishing
990+
991+-- | Split a stream of arrays on a given separator byte, dropping the separator
992+-- and coalescing all the arrays between two separators into a single array.
993+--
994+-- @since 0.7.0
995+{-# INLINE_NORMAL splitOn #-}
996+splitOn
997+ :: MonadIO m
998+ => Word8
999+ -> D.Stream m (Array Word8)
1000+ -> D.Stream m (Array Word8)
1001+splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
1002+
1003+ where
1004+
1005+ {-# INLINE_LATE step' #-}
1006+ step' gst (Initial st) = do
1007+ r <- step gst st
1008+ case r of
1009+ D.Yield arr s -> do
1010+ (arr1, marr2) <- breakOn byte arr
1011+ return $ case marr2 of
1012+ Nothing -> D.Skip (Buffering s arr1)
1013+ Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
1014+ D.Skip s -> return $ D.Skip (Initial s)
1015+ D.Stop -> return $ D.Stop
1016+
1017+ step' gst (Buffering st buf) = do
1018+ r <- step gst st
1019+ case r of
1020+ D.Yield arr s -> do
1021+ (arr1, marr2) <- breakOn byte arr
1022+ buf' <- spliceTwo buf arr1
1023+ return $ case marr2 of
1024+ Nothing -> D.Skip (Buffering s buf')
1025+ Just x -> D.Skip (Yielding buf' (Splitting s x))
1026+ D.Skip s -> return $ D.Skip (Buffering s buf)
1027+ D.Stop -> return $
1028+ if byteLength buf == 0
1029+ then D.Stop
1030+ else D.Skip (Yielding buf Finishing)
1031+
1032+ step' _ (Splitting st buf) = do
1033+ (arr1, marr2) <- breakOn byte buf
1034+ return $ case marr2 of
1035+ Nothing -> D.Skip $ Buffering st arr1
1036+ Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
1037+
1038+ step' _ (Yielding arr next) = return $ D.Yield arr next
1039+ step' _ Finishing = return $ D.Stop
src/Streamly/String.hs+1−1
@@ -187,7 +187,7 @@ words = FL.wordsBy isSpace toArray
187
187
188{-# INLINE unlines #-}
188{-# INLINE unlines #-}
189unlines :: (MonadIO m, IsStream t) => t m (Array Char) -> t m Char
189unlines :: (MonadIO m, IsStream t) => t m (Array Char) -> t m Char
190−unlines = D.fromStreamD . A.unlines . D.toStreamD
190+unlines = D.fromStreamD . A.unlines '\n' . D.toStreamD
191
191
192{-# INLINE unwords #-}
192{-# INLINE unwords #-}
193unwords :: (MonadAsync m, IsStream t) => t m (Array Char) -> t m Char
193unwords :: (MonadAsync m, IsStream t) => t m (Array Char) -> t m Char
streamly.cabal+5−1
@@ -207,6 +207,9 @@ library
207 , Streamly.Sink
207 , Streamly.Sink
208 , Streamly.Pipe.Types
208 , Streamly.Pipe.Types
209
209
210+ , Streamly.FileSystem.IOVec
211+ , Streamly.FileSystem.FDIO
212+
210 exposed-modules: Streamly.Prelude
213 exposed-modules: Streamly.Prelude
211 , Streamly.Time
214 , Streamly.Time
212 , Streamly
215 , Streamly
@@ -217,8 +220,9 @@ library
217
220
218 -- IO devices
221 -- IO devices
219 , Streamly.Mem.Array
222 , Streamly.Mem.Array
220− , Streamly.FileSystem.File
223+ , Streamly.FileSystem.FD
221 , Streamly.FileSystem.Handle
224 , Streamly.FileSystem.Handle
225+ , Streamly.FileSystem.File
222 , Streamly.Network.Socket
226 , Streamly.Network.Socket
223 , Streamly.Network.Server
227 , Streamly.Network.Server
224 , Streamly.Network.Client
228 , Streamly.Network.Client