Use NoBuffering in File/FileIo APIs

Author: harendra-kumar

core/src/Streamly/FileSystem/FileIO.hs

@@ -27,8 +27,9 @@ module Streamly.FileSystem.FileIO
     -- documentation. One IO request may or may not read the full
     -- chunk. If the whole stream is not consumed, it is possible that we may
     -- read slightly more from the IO device than what the consumer needed.
-    -- Unless specified otherwise in the API, writes are collected into chunks
-    -- of @defaultChunkSize@ before they are written to the IO device.
+    -- When writing, unless specified otherwise in the API, writes are
+    -- collected into chunks of @defaultChunkSize@ before they are written to
+    -- the IO device.
 
     -- Streaming APIs work for all kind of devices, seekable or non-seekable;
     -- including disks, files, memory devices, terminals, pipes, sockets and

core/src/Streamly/FileSystem/Handle.hs

@@ -14,7 +14,11 @@
 -- Read and write byte streams and array streams to and from file handles
 -- ('Handle').
 --
--- The 'TextEncoding', 'NewLineMode', and 'Buffering' options of the underlying
+-- Please set NoBuffering mode on the handle as buffering is explicitly
+-- controlled by the streaming API and double buffering can sometimes cause
+-- unexpected results.
+--
+-- Also note that the 'TextEncoding', 'NewLineMode' options of the underlying
 -- GHC 'Handle' are ignored by these APIs. Please use "Streamly.Unicode.Stream"
 -- module for encoding and decoding a byte stream, use stream splitting
 -- operations in "Streamly.Data.Stream" to create a stream of lines or to split

core/src/Streamly/Internal/FileSystem/File.hs

@@ -98,7 +98,8 @@ import Control.Monad.Catch (MonadCatch)
 import Control.Monad.IO.Class (MonadIO(..))
 import Data.Kind (Type)
 import Data.Word (Word8)
-import System.IO (Handle, openFile, IOMode(..), hClose)
+import System.IO
+    (Handle, IOMode(..), openFile, hClose, hSetBuffering, BufferMode(..))
 import Prelude hiding (read)
 
 import qualified Control.Monad.Catch as MC
@@ -150,7 +151,14 @@ import qualified Streamly.Internal.FileSystem.Handle as FH
 {-# INLINE withFile #-}
 withFile :: (MonadIO m, MonadCatch m)
     => FilePath -> IOMode -> (Handle -> Stream m a) -> Stream m a
-withFile file mode = S.bracketIO (openFile file mode) hClose
+withFile file mode = S.bracketIO open hClose
+
+    where
+
+    open = do
+        h <- openFile file mode
+        hSetBuffering h NoBuffering
+        return h
 
 -- | Transform an 'Unfold' from a 'Handle' to an unfold from a 'FilePath'.  The
 -- resulting unfold opens a handle in 'ReadMode', uses it using the supplied
@@ -163,7 +171,15 @@ withFile file mode = S.bracketIO (openFile file mode) hClose
 {-# INLINE usingFile #-}
 usingFile :: (MonadIO m, MonadCatch m)
     => Unfold m Handle a -> Unfold m FilePath a
-usingFile = UF.bracketIO (`openFile` ReadMode) hClose
+usingFile = UF.bracketIO open hClose
+
+    where
+
+    open file = do
+        h <- openFile file ReadMode
+        hSetBuffering h NoBuffering
+        return h
+
 
 {-# INLINE usingFile2 #-}
 usingFile2 :: (MonadIO m, MonadCatch m)
@@ -174,6 +190,7 @@ usingFile2 = UF.bracketIO before after
 
     before (x, file) =  do
         h <- openFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, h)
 
     after (_, h) = hClose h
@@ -187,6 +204,7 @@ usingFile3 = UF.bracketIO before after
 
     before (x, y, z, file) =  do
         h <- openFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, y, z, h)
 
     after (_, _, _, h) = hClose h
@@ -439,6 +457,7 @@ writeChunks path = Fold step initial extract final
     where
     initial = do
         h <- liftIO (openFile path WriteMode)
+        liftIO $ hSetBuffering h NoBuffering
         fld <- FL.reduce (FH.writeChunks h)
                 `MC.onException` liftIO (hClose h)
         return $ FL.Partial (fld, h)

core/src/Streamly/Internal/FileSystem/FileIO.hs

@@ -68,9 +68,9 @@ module Streamly.Internal.FileSystem.FileIO
     , writeChunks
 
     -- ** Writing Streams
-    , fromBytes -- putBytes?
-    , fromBytesWith
-    , fromChunks
+    , fromBytes -- XXX putBytes?
+    , fromBytesWith -- putBytesWith
+    , fromChunks -- putChunks?
 
     -- ** Append To File
     , writeAppend
@@ -84,7 +84,7 @@ where
 import Control.Monad.Catch (MonadCatch)
 import Control.Monad.IO.Class (MonadIO(..))
 import Data.Word (Word8)
-import System.IO (Handle, IOMode(..), hClose)
+import System.IO (Handle, IOMode(..), hClose, hSetBuffering, BufferMode(..))
 import Prelude hiding (read)
 
 import qualified Control.Monad.Catch as MC
@@ -130,32 +130,52 @@ import qualified Streamly.Internal.FileSystem.Windows.File as File
 -- Safe file reading
 -------------------------------------------------------------------------------
 
--- | @'withFile' name mode act@ opens a file using 'openFile' and passes
--- the resulting handle to the computation @act@.  The handle will be
--- closed on exit from 'withFile', whether by normal termination or by
--- raising an exception.  If closing the handle raises an exception, then
--- this exception will be raised by 'withFile' rather than any exception
--- raised by 'act'.
+-- | @'withFile' name mode act@ opens a file with NoBuffering set on the handle
+-- and passes the resulting handle to the computation @act@. The handle will be
+-- closed on exit from 'withFile', whether by normal termination or by raising
+-- an exception.  If closing the handle raises an exception, then that
+-- exception is raised by 'withFile' rather than any exception raised by 'act'.
+--
+-- The file is opened without buffering as buffering can be controlled by the
+-- streaming APIs.
 --
 -- /Pre-release/
 --
 {-# INLINE withFile #-}
 withFile :: (MonadIO m, MonadCatch m)
     => Path -> IOMode -> (Handle -> Stream m a) -> Stream m a
-withFile file mode = S.bracketIO (File.openFile file mode) hClose
+withFile file mode = S.bracketIO open hClose
+
+    where
 
--- | Transform an 'Unfold' from a 'Handle' to an unfold from a 'Path'.  The
--- resulting unfold opens a handle in 'ReadMode', uses it using the supplied
--- unfold and then makes sure that the handle is closed on normal termination
--- or in case of an exception.  If closing the handle raises an exception, then
--- this exception will be raised by 'usingFile'.
+    open = do
+        h <- File.openFile file mode
+        hSetBuffering h NoBuffering
+        return h
+
+-- | Transform an 'Unfold' that takes 'Handle' as input to an unfold that takes
+-- a 'Path' as input.  The resulting unfold opens the file in 'ReadMode',
+-- passes it to the supplied unfold and then makes sure that the handle is
+-- closed on normal termination or in case of an exception.  If closing the
+-- handle raises an exception, then this exception will be raised by
+-- 'usingFile'.
+--
+-- The file is opened without buffering as buffering can be controlled by the
+-- streaming APIs.
 --
 -- /Pre-release/
 --
 {-# INLINE usingFile #-}
 usingFile :: (MonadIO m, MonadCatch m)
     => Unfold m Handle a -> Unfold m Path a
-usingFile = UF.bracketIO (`File.openFile` ReadMode) hClose
+usingFile = UF.bracketIO open hClose
+
+    where
+
+    open file = do
+        h <- File.openFile file ReadMode
+        hSetBuffering h NoBuffering
+        return h
 
 {-# INLINE usingFile2 #-}
 usingFile2 :: (MonadIO m, MonadCatch m)
@@ -166,6 +186,7 @@ usingFile2 = UF.bracketIO before after
 
     before (x, file) =  do
         h <- File.openFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, h)
 
     after (_, h) = hClose h
@@ -179,6 +200,7 @@ usingFile3 = UF.bracketIO before after
 
     before (x, y, z, file) =  do
         h <- File.openFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, y, z, h)
 
     after (_, _, _, h) = hClose h
@@ -201,7 +223,7 @@ usingFile3 = UF.bracketIO before after
 putChunk :: Path -> Array a -> IO ()
 putChunk file arr = File.withFile file WriteMode (`FH.putChunk` arr)
 
--- | append an array to a file.
+-- | Append an array to a file.
 --
 -- /Pre-release/
 --
@@ -378,17 +400,18 @@ write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()
 write = toHandleWith A.defaultChunkSize
 -}
 
--- | Write a stream of chunks to a handle. Each chunk in the stream is written
--- to the device as a separate IO request.
+-- | Write a stream of chunks to a file. Each chunk in the stream is written
+-- immediately to the device as a separate IO request, without coalescing or
+-- buffering.
 --
--- /Pre-release/
 {-# INLINE writeChunks #-}
 writeChunks :: (MonadIO m, MonadCatch m)
     => Path -> Fold m (Array a) ()
 writeChunks path = Fold step initial extract final
     where
     initial = do
         h <- liftIO (File.openFile path WriteMode)
+        liftIO $ hSetBuffering h NoBuffering
         fld <- FL.reduce (FH.writeChunks h)
                 `MC.onException` liftIO (hClose h)
         return $ FL.Partial (fld, h)