batchwriter

README

batchwriter-go

batchwriter-go implements the io.Writer which batch writes (maybe writev if it's net.Conn) to writer by channel

go test -v -benchmem -run="^$" github.com/detailyang/batchwriter-go -bench "^Benchmark"
goos: darwin
goarch: amd64
pkg: github.com/detailyang/batchwriter-go
BenchmarkBatchWrite-8           	 2929246	       369 ns/op	     20997 failure	   2908249 success	     139 B/op	       1 allocs/op
BenchmarkBatchWriteParallel-8   	45410478	        35.0 ns/op	  45330196 failure	     80282 success	       0 B/op	       0 allocs/op
PASS
ok  	github.com/detailyang/batchwriter-go	3.131s

Documentation

Overview

Package batchwriter implements the io.Writer which batch writes (maybe writev if it's net.Conn) to writer by channel

nolint

Index

• Variables
• func BatchWrite(v interface{})
• type BatchWriter
  • func NewBatchWriter(o *Option, w io.Writer) (*BatchWriter, error)
  • func (bw *BatchWriter) Close() error
  • func (bw *BatchWriter) DoWrite() error
  • func (bw *BatchWriter) DoWritev() error
  • func (bw *BatchWriter) GetBytesWritten() int64
  • func (bw *BatchWriter) GetInflightsCap() int
  • func (bw *BatchWriter) GetInflightsLen() int
  • func (bw *BatchWriter) GetNumRequests() int64
  • func (bw *BatchWriter) GetPendingRequests() int64
  • func (bw *BatchWriter) IsClosed() bool
  • func (bw *BatchWriter) Write(d []byte) (int, error)
• type Option
  • func NewOption() *Option
  • func (o *Option) BlockWriteForever() *Option
  • func (o *Option) SetMaxFlushDelay(d time.Duration) *Option
  • func (o *Option) SetMaxinflights(max int) *Option
  • func (o *Option) SetNumRequests(i64 *int64) *Option
  • func (o *Option) SetNumWrite(i64 *int64) *Option
  • func (o *Option) SetPendingRequests(i64 *int64) *Option
  • func (o *Option) SetTimeout(d time.Duration) *Option
  • func (o *Option) SetWorkerPool(wp *workerpool.WorkerPool) *Option
  • func (o *Option) SetWriteMode(m WriteMode) *Option
• type WriteMode

Variables

var (
	// ErrBatchWriterClosed indicates the writer was closed.
	ErrBatchWriterClosed = errors.New("batchwriter: writer was closed")

	// ErrBatchWriterTooManyWrite indicates writer cannot process the write because of
	// too many write.
	ErrBatchWriterTooManyWrite = errors.New("batchwriter: writer wrote too fast")

	// ErrBatchWriterAtivelyClose indicates caller actively close the writer.
	ErrBatchWriterAtivelyClose = errors.New("batchwriter: writer close actively")
)

Functions

func BatchWrite

func BatchWrite(v interface{})

BatchWrite wraps the bw.DoWritev to use workerPool.

nolint

Types

type BatchWriter

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

BatchWriter wraps a writer and batch write it.

nolint

func NewBatchWriter

func NewBatchWriter(o *Option, w io.Writer) (*BatchWriter, error)

NewBatchWriter returns a new batch writer.

nolint

func (*BatchWriter) Close

func (bw *BatchWriter) Close() error

Close closes the writer.

func (*BatchWriter) DoWrite

func (bw *BatchWriter) DoWrite() error

func (*BatchWriter) DoWritev

func (bw *BatchWriter) DoWritev() error

func (*BatchWriter) GetBytesWritten

func (bw *BatchWriter) GetBytesWritten() int64

GetBytesWritten gets bytes of written.

func (*BatchWriter) GetInflightsCap

func (bw *BatchWriter) GetInflightsCap() int

GetInflightsCap gets the cap of the inflights.

func (*BatchWriter) GetInflightsLen

func (bw *BatchWriter) GetInflightsLen() int

GetInflightsLen gets the length of the inflights.

func (*BatchWriter) GetNumRequests

func (bw *BatchWriter) GetNumRequests() int64

GetNumRequests gets number of processed requests.

func (*BatchWriter) GetPendingRequests

func (bw *BatchWriter) GetPendingRequests() int64

GetPendingRequests gets the pending requests.

func (*BatchWriter) IsClosed

func (bw *BatchWriter) IsClosed() bool

IsClosed indicates whether writer was closed.

func (*BatchWriter) Write

func (bw *BatchWriter) Write(d []byte) (int, error)

Write implements io.Writer.

type Option

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

Option configruates the option of write.

func NewOption

func NewOption() *Option

NewOption returns a new Option.

func (*Option) BlockWriteForever

func (o *Option) BlockWriteForever() *Option

BlockWriteForever indicates caller blockly write to io.Writer.

func (*Option) SetMaxFlushDelay

func (o *Option) SetMaxFlushDelay(d time.Duration) *Option

SetMaxFlushDelay sets the max delay of flush.

func (*Option) SetMaxinflights

func (o *Option) SetMaxinflights(max int) *Option

SetMaxinflights sets the max of inflights (default CPU * 2)

func (*Option) SetNumRequests

func (o *Option) SetNumRequests(i64 *int64) *Option

SetNumRequests sets the numrequests metric.

func (*Option) SetNumWrite

func (o *Option) SetNumWrite(i64 *int64) *Option

SetNumWrite sets the numwrite metric.

func (*Option) SetPendingRequests

func (o *Option) SetPendingRequests(i64 *int64) *Option

SetPendingRequests sets the pendingrequests metric.

func (*Option) SetTimeout

func (o *Option) SetTimeout(d time.Duration) *Option

SetTimeout sets the timeout for write if it's net.Conn

func (*Option) SetWorkerPool

func (o *Option) SetWorkerPool(wp *workerpool.WorkerPool) *Option

SetWorkerPool sets the workerpool to option.

func (*Option) SetWriteMode

func (o *Option) SetWriteMode(m WriteMode) *Option

SetWriteMode sets the mode of writer which is batch or flush.

type WriteMode

type WriteMode uint8

const (
	BatchWriteMode  WriteMode = 0
	FlushWriteMode  WriteMode = 1
	ManualWriteMode WriteMode = 3
)