RFC wait() support for GenericIOBuffer/PipeBuffer

[samoconnor]

This proposal adds wait(io::GenericIOBuffer) to wait for something to be written to the buffer and modifies eof(io::GenericIOBuffer) to be false if io.writeable is true.

The aim is to support an async producer / consumer pattern like this:

io = PipeBuffer()

@async while true
    if done
        close(io)
        break
    end
    println(io, now())
    sleep(1)
end

while !eof(io)
    wait(io)
    println(String(readavailable(io)))
end

@quinnj has a FIFOBuffer class in HTTP.jl that attempts to do something like this, but isn't working 100% yet: JuliaWeb/HTTP.jl#74 .

It seems to me that reading and writing from a buffered pipe is generally useful and a good thing to have in Base.

Note: I haven't tried to compile or run the code in this PR, at this point it's just intended as a sketch of how it might look. If I there is support for a change like this, I'll make sure it works and add tests etc.

[vtjnash]

This already exists. See BufferStream. :)

[StefanKarpinski]

@vtjnash can you elaborate on that, pls?

[vtjnash]

The requested / proposed functionality is already implemented and exported from Base.

[samoconnor]

@vtjnash that's great!
I guess I missed it because it is almost undocumented.

Is there any point in having PipeBuffer and IOBuffer and BufferStream?

See also #23398

[vtjnash]

Condition is an expensive object to construct, and isn't necessary (or used) when using IOBuffer as a drop-in for a file object, only when using it as a drop-in for a pipe.

[samoconnor]

@quinnj Any reason you couldn't use BufferStream instead of FIFOBuffer in HTTP.jl?

@vtjnash whats the deal with BufferStream <: LibuvStream? -- this comment seems to conflict with that.

[quinnj]

I definitely took inspiration from BufferStream. FIFOBuffer provides independent read/write positions to provide the "FIFO" functionality; i.e. one task can be writing while another task is reading and they can work independently of each other w/o getting things trampled/run over, or without having to be seeking all over the place. Whatever you put into a FIFOBuffer will be read out in the exact order it was put in, no seeking necessary.

[samoconnor]

Is there any reason that something like the eof() and close() changes in this PR shouldn't be made for IOBuffer/PipeBuffer? i.e. eof() not true until after close() is called, and read() is allowed after close().

The current close() behaviour seems wrong:

julia> io = PipeBuffer()
julia> write(io, "Hello")
julia> close(io)
julia> readstring(io)
ERROR: ArgumentError: read failed, IOBuffer is not readable
julia> take!(io)
ERROR: ArgumentError: read failed, IOBuffer is not readable

vs

julia> io = BufferStream()
julia> write(io, "Hello")
julia> close(io)
julia> readstring(io)
"Hello"

[samoconnor]

@quinnj as far as I understand, BufferStream does all the things you describe. ("one task can be writing while another task is reading" etc).

e.g.

julia> io = BufferStream()
BufferStream() bytes waiting:0, isopen:true

julia> @async while true
           println(io, now())
           sleep(1)
       end
Task (runnable) @0x000000011c52c250

julia> while !eof(io)
           println(String(readavailable(io)))
       end

2017-08-23T07:09:15.929
2017-08-23T07:09:16.935
2017-08-23T07:09:17.939

2017-08-23T07:09:18.94

2017-08-23T07:09:19.946

...

Do you have an example where BufferStream won't work?

[samoconnor]

@vtjnash Perhaps it's just the naming that is causing confusion. I assumed that PipeBuffer was intended as a "drop-in for a pipe" (and therefore expected it to support async read/write).

Perhaps the PipeBuffer constructor should be removed and the documentation should explain that BufferStream is intended as a "drop-in for a pipe" whereas IOBuffer is "a drop-in for a file object"?

[quinnj]

One big difference is FIFOBuffer allows you to use a fixed # of bytes for the buffer. I.e. writes are only allowed to fill up to the max # of bytes before they'll "wait" for bytes to be read before writing more. This is seamlessly integrated w/ the chunked-transfer encoding in HTTP.jl in that you can set a "max chunk size" that will be enforced via the underlying FIFOBuffer: meaning an upload can truly be sent in 1MB chunks if you so desire.