concurrent

Documentation

Index

• type Writer
  • func NewWriter(w io.Writer) *Writer
  • func NewWriterAutoFlush(w io.Writer, size int, flushAt float32) *Writer
  • func NewWriterSize(w io.Writer, size int) *Writer
  • func (b *Writer) Available() int
  • func (b *Writer) Buffered() int
  • func (b *Writer) Flush() error
  • func (b *Writer) ReadFrom(r io.Reader) (n int64, err error)
  • func (b *Writer) Reset(w io.Writer)
  • func (b *Writer) Write(p []byte) (nn int, err error)
  • func (b *Writer) WriteByte(c byte) error
  • func (b *Writer) WriteRune(r rune) (size int, err error)
  • func (b *Writer) WriteString(s string) (int, error)

Types

type Writer

type Writer struct {
	// contains filtered or unexported fields
}

Writer implements highly concurrent buffering for an io.Writer object. In particular, writes will not block while a Flush() call is in progress as long as enough buffer space is available.

Note however that writes will still block in a number of cases, e.g. when another write larger than the buffer size (AKA a chunked write) is in progress. Also, concurrent Flush() calls (whether explicit or triggered by the buffer filling up) will block one another.

Optionally, to help avoid the buffer filling up in the first place, automatic asynchronous flushing may be enabled when a configured fraction of the buffer is filled and no other flush is in progress.

Implementation details:

In order to provide write isolation (i.e. make writes appear as if they were issued serially) Writer uses a global mutex to protect all internal state. In the general case, which we will call "base mode", in order to provide high concurrency, the mutex is released just before the write to the underlying io.Writer is issued and reacquired immediately after. A condition variable, notFlushing, is used to ensure at most one goroutine at a time is executing this mutex unprotected sequence of of code.

However, if a chunked write is begun by any of the write methods, Writer enters what we'll call "chunked write mode". In this state the goroutine doing the chunked write will never explicitly release the mutex until it completes, in order to prevent race conditions with other writers. But, because of the asynchronous nature of flushes, it may still have to wait for an in flight flush to complete (on a separate, higher priority condition variable, chunkedWriter). sync.Cond.Wait() will implicitly release the mutex, meaning that all other goroutines (except the chunked writer) will have to explicitly check whether in "chunked write mode" every single time they have just acquired the mutex and wait for the in-flight chunked write to complete (on a third condition variable, noChunkedWrite).

This arrangement ensures that (1) the chunked writer goroutine has priority over all other writers and flushers; and that (2) whenever a goroutine is holding the mutex it is either (a) the (one) chunked writer when in "chunked write state" or (b) it is a non-chunked writer or flusher and Writer is in the straightforward "base mode" where the mutex and notFlushing provide all necessary concurrency.

func NewWriter

func NewWriter(w io.Writer) *Writer

NewWriter returns a new Writer whose buffer has the default size.

func NewWriterAutoFlush

func NewWriterAutoFlush(w io.Writer, size int, flushAt float32) *Writer

NewWriterAutoFlush returns a new Writer whose buffer has at least the specified size and that will automatically trigger an asynchronous flush when the given buffer fraction is filled (e.g. 0.75 will flush when the buffer is 75% full). Panics if the argument io.Writer is already a Writer or bufio.Writer.

func NewWriterSize

func NewWriterSize(w io.Writer, size int) *Writer

NewWriterSize returns a new Writer whose buffer has at least the specified size. If the argument io.Writer is already a Writer with large enough size, it returns the underlying Writer.

func (*Writer) Available

func (b *Writer) Available() int

Available returns how many bytes are unused in the buffer.

func (*Writer) Buffered

func (b *Writer) Buffered() int

Buffered returns the number of bytes that have been written into the current buffer.

func (*Writer) Flush

func (b *Writer) Flush() error

Flush writes any buffered data to the underlying io.Writer. As long as the buffer has enough available space, writes can proceed concurrently.

func (*Writer) ReadFrom

func (b *Writer) ReadFrom(r io.Reader) (n int64, err error)

ReadFrom implements io.ReaderFrom.

func (*Writer) Reset

func (b *Writer) Reset(w io.Writer)

Reset discards any unflushed buffered data, clears any error, and resets b to write its output to w. If w.err is nil, in order to ensure no partial writes end up in w, it waits until any chunked write and/or flush complete before the output is redirected.

func (*Writer) Write

func (b *Writer) Write(p []byte) (nn int, err error)

Write writes the contents of p into the buffer. It returns the number of bytes written. If nn < len(p), it also returns an error explaining why the write is short.

func (*Writer) WriteByte

func (b *Writer) WriteByte(c byte) error

WriteByte writes a single byte.

func (*Writer) WriteRune

func (b *Writer) WriteRune(r rune) (size int, err error)

WriteRune writes a single Unicode code point, returning the number of bytes written and any error.

func (*Writer) WriteString

func (b *Writer) WriteString(s string) (int, error)

WriteString writes a string. It returns the number of bytes written. If the count is less than len(s), it also returns an error explaining why the write is short.