[ moving to libraries(a)haskell.org from glasgow-haskell-users(a)haskell.org ]
Bulat Ziganshin wrote:
> Wednesday, April 19, 2006, 4:45:19 PM, you wrote:
>
>>Believe me I've looked in detail at your streams library.
>>Performance-wise it is great but the design needs to be reworked IMO.
>
>>The main problem is that it doesn't have enough type structure. There
>>are many combinations of stream transformers that don't make sense, and
>>should therefore be ruled out by the type system. There are operations
>>that don't work on some streams. There should at the least be a type
>>distinction between directly accessible memory streams, byte streams,
>>and text streams. Additionally I would add separate classes for
>>seekable and buffered streams. I believe these changes would improve
>>performance by reducing the size of dictionaries.
>
> you have written this in February, but this discussion was not
> finished due to my laziness. now i tried to split Stream interface to
> several parts. so
I've attached a sketched design. It doesn't compile, but it illustrates
the structure I have in mind. The main improvement since the new-io
library is the addition of memory streams. This is an idea from your
library and I like it a lot, although I changed the type of the methods:
-- | An input stream accessed directly via a memory buffer.
-- Ordinary 'InputStream's may be converted to 'MemInputStream's by
-- adding buffering; see 'bufferInputStream'.
class MemInputStream s where
-- | Consume some bytes from the memory stream. The second argument
-- is an IO action that is passed a buffer and its size (the size must
-- be non-zero), and it should return the number of bytes consumed.
withStreamInputBuffer :: s -> (Ptr Word8 -> Int -> IO Int) -> IO ()
-- | An output stream accessed directly via a memory buffer.
-- Ordinary 'OutputStream's may be converted to 'MemOutputStream's by
-- adding buffering; see 'bufferOutputStream'.
class MemOutputStream s where
-- | Write some bytes to a memory stream. The second argument
-- is an IO action that is passed a buffer and its size (the size must
-- be non-zero), and it should return the number of bytes written.
withStreamOutputBuffer :: s -> (Ptr Word8 -> Int -> IO Int) -> IO ()
> 1) that you think - Stream should be base for all other stream classes
> or each Stream class should be independent? i.e.
Superclasses aren't necessary, but they might help to reduce the size of
contexts in practice. We should probably experiment with both.
> 2) separation of Stream classes make some automatic definitions
> impossible. for example, released version contains vGetBuf
> implementation that is defined via vGetChar and works ok for streams
> that provide only vGetChar as base function.
No class should implement both reading bytes and reading chars.
Encoding/decoding should be a stream transformer that turns a byte
stream into a text stream. So there's no duplication of these methods.
I believe splitting up the classes should lead to less duplication, not
more, partly because you don't have to implement a lot of methods that
don't do anything or are errors (eg. writing to an input stream). I
admit I haven't actually written all the code, though.
> 3) the problems are substantially growed now - when i tried to
> separate input and output streams (the same will apply to detaching of
> seekable streams into the separate class). the problem is what i need
> either to provide 2 or 3 separate implementations for buffering of
> read-only, write-only and read-write streams or have some universal
> definition that should work even when base Stream don't provide part
> of operations. the last seems to be impossible - may be i don't
> understand enough Haskell's class system?
>
> let's see:
>
> data BufferedStream h = Buf h ....
>
> vClose (Buf h ...) = vPutBuf ... - flush buffer's contents
>
> how i can implement this if `h` may not support vPutBuf operation?
> especially to allow read/write streams to work???
You can only buffer a byte stream. See my sketch design.
> 4) what you mean by "There are many combinations of stream
> transformers that don't make sense" ? splitting Stream class to the
> BlockStream/TextStream/ByteStream or something else?
Yes - adding decoding to a TextStream doesn't make sense. Directly
accessing the memory of a byte stream doesn't make sense: you need to
buffer it first, or use a memory-mapped stream.
It is still possible to implement read/write files using this structure.
There's nothing stopping you having an type that is an instance of
both InputStream and OutputStream (eg. a read/write file), and layering
buffering on top of this would yield a buffered input/output stream in
which the buffer contains only input or output data.
Cheers,
Simon
{-# OPTIONS -fglasgow-exts -cpp #-}
-----------------------------------------------------------------------------
-- |
-- Module : System.IO.Stream
-- Copyright : (c) various
-- License : see libraries/base/LICENSE
--
-- Maintainer : simonmar(a)microsoft.com
-- Stability : experimental
-- Portability : non-portable (existentials, ghc extensions)
--
-- InputStreams and OutputStreams are classes of objects which support
-- input and output respectively. Streams can be layered on top of various
-- underlying I/O objects (such as files or sockets). Stream transformers
-- can be applied to turn streams of one type into streams of another type.
--
-----------------------------------------------------------------------------
module System.IO.Stream (
-- * Streams
{-class-} Stream(..),
{-class-} InputStream(..),
{-class-} OutputStream(..),
streamGet, streamReadBuffer,
streamPut, streamWriteBuffer,
-- * Stream connections
PipeInputStream, PipeOutputStream,
streamPipe,
streamConnect,
-- * Memory streams
{-class-} MemInputStream,
{-class-} MemOutputStream,
-- ** Converting memory streams to I/O streams
MemToInputStream, memToInputStream,
MemToOutputStream, memToOutputStream,
-- * Buffering
BufferMode(..),
BufferedInputStream, bufferIntputStream,
BufferedOutputStream, bufferOutputStream,
) where
import System.IO.Buffer
import Foreign
import Data.Word ( Word8 )
import System.IO ( BufferMode(..) )
import System.IO.Error ( mkIOError, eofErrorType )
import Control.Exception ( assert )
import Control.Monad ( when, liftM )
import Control.Concurrent
import Data.IORef
import GHC.Exts
import GHC.Ptr ( Ptr(..) )
import GHC.IOBase ( IO(..), ioException )
import GHC.Handle ( ioe_EOF )
#define UPK {-# UNPACK #-} !
-- -----------------------------------------------------------------------------
-- Streams
class Stream s where
-- | closes a stream
closeStream :: s -> IO ()
-- | returns 'True' if the stream is open
streamIsOpen :: s -> IO Bool
-- | ToDo: objections have been raised about this method, and
-- are still to be resolved. It doesn't make as much sense
-- for output streams as it does for input streams.
streamIsEOS :: s -> IO Bool
-- | Returns 'True' if there is data available to read from this
-- stream. Returns 'False' if either there is no data available, or
-- the end of the stream has been reached.
streamReady :: s -> IO Bool
-- | Returns the number of bytes that can be transfered to/from
-- this stream, if known.
streamRemaining :: s -> IO (Maybe Integer)
-- | An 'InputStream' is a basic I/O object which supports reading a
-- stream of 'Word8's. It is expected that 'InputStream's are unbuffered:
-- buffering is layered on top of one of these.
class Stream s => InputStream s where
-- | Grabs data without blocking, but only if there is data available.
-- If there is none, then waits for some. This function may only
-- return zero if either the requested length is zero or the end of stream
-- has been reached.
streamReadBufferNonBlocking :: s -> Integer -> Ptr Word8 -> IO Integer
-- | Reads a single 'Word8' from a stream.
streamGet :: InputStream s => s -> IO Word8
streamGet s =
alloca $ \p -> do
r <- streamReadBufferNonBlocking s 1 p
if r == 0 then ioe_EOF else peek p
-- | Reads data from the stream into a 'Buffer'. Returns the
-- number of elements that were read, which may only be less than the
-- requested length if the end of the stream was reached.
streamReadBuffer :: InputStream s => s -> Integer -> Ptr Word8 -> IO Integer
streamReadBuffer s 0 buf = return 0
streamReadBuffer s len ptr = streamReadBufferLoop s ptr 0 (fromIntegral len)
streamReadBufferLoop
:: InputStream s => s -> Ptr Word8
-> Integer -> Integer -> IO Integer
streamReadBufferLoop s ptr off len = do
r <- streamReadBufferNonBlocking s len ptr
if r == 0
then return (fromIntegral off)
else if (r < len)
then streamReadBufferLoop s
(ptr `plusPtr` fromIntegral r) (off+r) (len-r)
else return (off+r)
-- -----------------------------------------------------------------------------
-- Output streams
class Stream s => OutputStream s where
-- | Writes data to an output stream. It will write at least one
-- byte, but will only write further bytes if it can do so without
-- blocking.
--
-- The result may never be 0 if the requested write size was > 0.
-- If no bytes can be written to the stream, then the
-- 'streamWriteBufferNonBlocking' should raise an exception
-- indicating the cause of the problem (eg. the stream is closed).
streamWriteBufferNonBlocking :: s -> Integer -> Ptr Word8 -> IO Integer
-- | Writes a single byte to an output stream.
streamPut :: OutputStream s => s -> Word8 -> IO ()
streamPut s word = with word $ \p -> streamWriteBuffer s 1 p
-- | Writes data to a stream, only returns when all the data has been
-- written.
streamWriteBuffer :: OutputStream s => s -> Integer -> Ptr Word8 -> IO ()
streamWriteBuffer s 0 ptr = return ()
streamWriteBuffer s len ptr = streamWriteBufferLoop s ptr 0 len
streamWriteBufferLoop
:: OutputStream s => s -> Ptr Word8
-> Integer -> Integer -> IO ()
streamWriteBufferLoop s ptr off len =
seq off $ -- strictness hack
if len == 0
then return ()
else do
r <- streamWriteBufferNonBlocking s len ptr
assert (r /= 0) $ do
if (r < len)
then streamWriteBufferLoop s (ptr `plusPtr` fromIntegral r)
(off+r) (len-r)
else return ()
-- ----------------------------------------------------------------------------
-- Connecting streams
data PipeInputStream
data PipeOutputStream
instance Stream PipeInputStream -- ToDo
instance InputStream PipeInputStream -- ToDo
instance Stream PipeOutputStream -- ToDo
instance OutputStream PipeOutputStream -- ToDo
streamPipe :: IO (PipeInputStream, PipeOutputStream)
streamPipe = error "unimplemented: streamOutputToInput"
-- | Takes an output stream and an input stream, and pipes all the
-- data from the former into the latter.
streamConnect :: (OutputStream o, InputStream i) => o -> i -> IO ()
streamConnect = error "unimplemented: streamInputToOutput"
-- ----------------------------------------------------------------------------
-- Memory streams
-- | An input stream accessed directly via a memory buffer.
-- Ordinary 'InputStream's may be converted to 'MemInputStream's by
-- adding buffering; see 'bufferInputStream'.
class MemInputStream s where
-- | Consume some bytes from the memory stream. The second argument
-- is an IO action that is passed a buffer and its size (the size must
-- be non-zero), and it should return the number of bytes consumed.
withStreamInputBuffer :: s -> (Ptr Word8 -> Int -> IO Int) -> IO ()
-- | An output stream accessed directly via a memory buffer.
-- Ordinary 'OutputStream's may be converted to 'MemOutputStream's by
-- adding buffering; see 'bufferOutputStream'.
class MemOutputStream s where
-- | Write some bytes to a memory stream. The second argument
-- is an IO action that is passed a buffer and its size (the size must
-- be non-zero), and it should return the number of bytes written.
withStreamOutputBuffer :: s -> (Ptr Word8 -> Int -> IO Int) -> IO ()
-- -----------------------------------------------------------------------------
-- A memory stream can be converted to an ordinary byte stream
newtype MemToInputStream s = MemToInputStream s
deriving (Stream, MemInputStream)
newtype MemToOutputStream s = MemToOutputStream s
deriving (Stream, MemOutputStream)
-- Rationale: what we really want is
-- instance MemInputStream s => InputStream s
-- but that overlaps. So instead we provide a way to convert
-- every MemInputStream into something that is an instance of
-- InputStream.
instance InputStream (MemToInputStream s) where
instance OutputStream (MemToOutputStream s) where
memToInputStream :: MemInputStream s => MemToInputStream s
memToInputStream = MemToInputStream
memToOutputStream :: MemOutputStream s => MemToOutputStream s
memToOutputStream = MemToOutputStream
-- -----------------------------------------------------------------------------
-- Buffering
-- | Operations on a stream with a buffer
class Buffered s where
-- | Sets the buffering mode on the stream. Returns 'True' if
-- the buffereing mode was set, or 'False' if it wasn't. If the
-- stream does not support buffering, it may return 'False'.
setBufferMode :: s -> BufferMode -> IO Bool
-- | Returns the current buffering mode for a stream. On a
-- stream that does not support buffering, the result will always
-- be 'NoBuffering'.
getBufferMode :: s -> IO BufferMode
-- | Returns the number of bytes of data in the buffer
countBufferedBytes :: s -> IO Int
-- | Operations on an output stream with a buffer
class OutputBuffered s where
-- | Flushes the buffer to the operating system
flush :: s -> IO ()
-- | Flushes the buffered data as far as possible, even to the
-- physical media if it can. It returns 'True' if the data
-- has definitely been flushed as far as it can go: to the
-- disk for a disk file, to the screen for a terminal, and so on.
sync :: s -> IO Bool
-- | Operations on an input stream with a buffer
class InputBuffered s where
-- | Discards the input buffer
discard :: s -> IO ()
-- | Pushes back the buffered data, if possible. Returns 'True' if
-- the buffer could be pushed back, 'False' otherwise.
pushback :: s -> IO Bool
-- TODO: Seekable superclass allows pushback?
-- | An 'InputStream' with buffering added
data BufferedInputStream s = BufferedInputStream s BufferMode !(IORef (Buffer Word8))
-- | An 'OutputStream' with buffering added
data BufferedOutputStream s = BufferedOutputStream s BufferMode !(IORef (Buffer Word8))
bufferSize :: BufferMode -> Int
bufferSize (BlockBuffering (Just size)) = size
bufferSize _ = dEFAULT_BUFFER_SIZE
-- | Add buffering to an 'InputStream'
bufferInputStream :: InputStream s => s -> BufferMode -> BufferedInputStream s
bufferInputStream stream bmode = do
buffer <- allocateBuffer (bufferSize bmode)
r <- newIORef buffer
return (BufferedInputStream stream bmode r)
-- | Add buffering to an 'OutputStream'
bufferOutputStream :: OutputStream s => s -> BufferedOutputStream s
bufferOutputStream stream bmode = do
buffer <- allocateBuffer (bufferSize bmode)
r <- newIORef buffer
return (BufferedOutputStream stream bmode r)
instance Stream s => MemInputStream (BufferedInputStream s) where
withStreamInputBuffer b@(BufferedInputStream s bmode ref) action = do
buffer <- readIORef ref
if emptyBuffer buffer
then do buffer' <- fillReadBuffer s buffer
writeIORef ref buffer'
withStreamInputBuffer b action
else do let used = bufferUsed buffer
count <- withBuffer buffer $ \ptr ->
action (ptr `plusPtr` bufRPtr buffer) used
let buffer' = bufferRemove (min used count)
checkBufferInvariants buffer'
writeIORef ref $! buffer'
instance Stream s => MemOutputStream (BufferedOutputStream s) where
withStreamOutputBuffer b@(BufferedOutputStream s bmode ref) action = do
buffer <- readIORef ref
let avail = bufferAvailable buffer
count <- withBuffer buffer $ \ptr ->
action (ptr `plusPtr` bufWPtr buffer) avail
let buffer' = bufferAdd (max count avail)
if fullBuffer buffer'
then do writeBuffer s buffer; writeIORef ref (emptyBuffer buffer')
else writeIORef ref buffer'
instance Stream s => Stream (BufferedInputStream s)
instance Stream s => Stream (BufferedOutputStream s)
instance Buffered (BufferedOutputStream s)
instance OutputBuffered (BufferedOutputStream s)
instance Buffered (BufferedInputStream s)
instance InputBuffered (BufferedInputStream s)
