Use NoBuffering in File/FileIo APIs

core/src/Streamly/FileSystem/FileIO.hs

@@ -15,6 +15,12 @@
 -- the handle based APIs as there is no possibility of a file descriptor
 -- leakage.
 --
+-- The file is opened in binary mode as encoding, decoding, and newline
+-- translation can be handled explicitly by the streaming APIs.
+--
+-- The file is opened without buffering as buffering can be controlled
+-- explicitly by the streaming APIs.
+--
 -- >> import qualified Streamly.FileSystem.FileIO as File
 --
 module Streamly.FileSystem.FileIO
@@ -27,8 +33,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

@@ -6,16 +6,6 @@
 -- Maintainer  : streamly@composewell.com
 -- Portability : GHC
 --
--- Read and write streams and arrays to and from files specified by their paths
--- in the file system. Unlike the handle based APIs which can have a read/write
--- session consisting of multiple reads and writes to the handle, these APIs
--- are one shot read or write APIs. These APIs open the file handle, perform
--- the requested operation and close the handle. Thease are safer compared to
--- the handle based APIs as there is no possibility of a file descriptor
--- leakage.
---
--- > import qualified Streamly.Internal.FileSystem.FileIO as File
---
 
 module Streamly.Internal.FileSystem.FileIO
     (
@@ -68,9 +58,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 +74,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 +120,58 @@ 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 and passes the resulting handle to
+-- the computation @act@. The handle is 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 in binary mode as encoding, decoding, and newline
+-- translation can be handled explicitly by the streaming APIs.
+--
+-- The file is opened without buffering as buffering can be controlled
+-- explicitly 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
+
+    open = do
+        h <- File.openBinaryFile file mode
+        hSetBuffering h NoBuffering
+        return h
 
--- | 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'.
+-- | 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 in binary mode as encoding, decoding, and newline
+-- translation can be handled explicitly by the streaming APIs.
+--
+-- The file is opened without buffering as buffering can be controlled
+-- explicitly 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.openBinaryFile file ReadMode
+        hSetBuffering h NoBuffering
+        return h
 
 {-# INLINE usingFile2 #-}
 usingFile2 :: (MonadIO m, MonadCatch m)
@@ -165,7 +181,8 @@ usingFile2 = UF.bracketIO before after
     where
 
     before (x, file) =  do
-        h <- File.openFile file ReadMode
+        h <- File.openBinaryFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, h)
 
     after (_, h) = hClose h
@@ -178,7 +195,8 @@ usingFile3 = UF.bracketIO before after
     where
 
     before (x, y, z, file) =  do
-        h <- File.openFile file ReadMode
+        h <- File.openBinaryFile file ReadMode
+        hSetBuffering h NoBuffering
         return (x, y, z, h)
 
     after (_, _, _, h) = hClose h
@@ -201,7 +219,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 +396,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)
+        h <- liftIO (File.openBinaryFile 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/Posix/File.hsc

@@ -66,8 +66,8 @@ module Streamly.Internal.FileSystem.Posix.File
     -- * Handle based
     , openFile
     , withFile
-    -- , openBinaryFile
-    -- , withBinaryFile
+    , openBinaryFile
+    , withBinaryFile
 
     -- Re-exported
     , Fd
@@ -306,13 +306,11 @@ openFile = File.openFile False openFileHandle
 withFile :: PosixPath -> IOMode -> (Handle -> IO r) -> IO r
 withFile = File.withFile False openFileHandle
 
-{-
 -- | Like openBinaryFile in base package but using Path instead of FilePath.
 openBinaryFile :: PosixPath -> IOMode -> IO Handle
 openBinaryFile = File.openFile True openFileHandle
 
 -- | Like withBinaryFile in base package but using Path instead of FilePath.
 withBinaryFile :: PosixPath -> IOMode -> (Handle -> IO r) -> IO r
 withBinaryFile = File.withFile True openFileHandle
--}
 #endif

core/src/Streamly/Internal/FileSystem/Windows/File.hsc

@@ -6,8 +6,8 @@ module Streamly.Internal.FileSystem.Windows.File
     -- * Handle based
       openFile
     , withFile
-    -- , openBinaryFile
-    -- , withBinaryFile
+    , openBinaryFile
+    , withBinaryFile
 #endif
     ) where
 
@@ -192,13 +192,11 @@ withFile = File.withFile False openFileHandle
 openFile :: WindowsPath -> IOMode -> IO Handle
 openFile = File.openFile False openFileHandle
 
-{-
 -- | Like withBinaryFile in base package but using Path instead of FilePath.
 withBinaryFile :: WindowsPath -> IOMode -> (Handle -> IO r) -> IO r
 withBinaryFile = File.withFile True openFileHandle
 
 -- | Like openBinaryFile in base package but using Path instead of FilePath.
 openBinaryFile :: WindowsPath -> IOMode -> IO Handle
 openBinaryFile = File.openFile True openFileHandle
--}
 #endif