syncpool

README

zstdpool-syncpool

github.com/klauspost/compress/zstd is a native Go implementation of Zstandard, with an API that leaks memory and goroutines if you don't call Close() on every Encoder and Decoder that you create, and Decoders cannot be reused after they are closed.

zstdpool-syncpool is a Go wrapper for github.com/klauspost/compress/zstd and sync.Pool, which automatically frees resources if you forget to call Close() and/or when items are dropped from the sync.Pool.

Background: https://github.com/klauspost/compress/issues/264

Basic usage

https://pkg.go.dev/github.com/mostynb/zstdpool-syncpool

import (
	"github.com/klauspost/compress/zstd"
	syncpool "github.com/mostynb/zstdpool-syncpool"
)

// Create a sync.Pool which returns wrapped *zstd.Decoder's.
decoderPool := syncpool.NewDecoderPool(zstd.WithDecoderConcurrency(1))

// Get a *DecoderWrapper and use it.
decoder := decoderPool.Get().(*syncpool.DecoderWrapper)
decoder.Reset(compressedDataReader)
_, err = io.Copy(uncompressedDataWriter, decoder)

// Return the decoder to the pool. If we forget this, then the zstd.Decoder
// won't leak resources.
decoderPool.Put(decoder)


// Create a sync.Pool which returns wrapped *zstd.Endoder's.
encoderPool := syncpool.NewEncoderPool(zstd.WithEncoderConcurrency(1))

// Get an *EncoderWrapper and use it.
encoder := encoderPool.Get().(*syncpool.EncoderWrapper)
encoder.Reset(compressedDataWriter)
_, err = io.Copy(encoder, uncompressedDataReader)

// Return the encoder to the pool. If we forget this, then the zstd.Encoder
// won't leak resources.
encoderPool.Put(encoder)

Simplified usage

If you would like to avoid type assertions, you can use NewDecoderPoolWapper and NewEncoderPoolWrapper:

import (
	"github.com/klauspost/compress/zstd"
	syncpool "github.com/mostynb/zstdpool-syncpool"
)

// Create a sync.Pool which returns wrapped *zstd.Decoder's.
decoderPool := syncpool.NewDecoderPoolWrapper(zstd.WithDecoderConcurrency(1))

// Get a *DecoderWrapper and use it.
decoder := decoderPool.Get(compressedDataReader)
_, err = io.Copy(uncompressedDataWriter, decoder)

// Return the decoder to the pool. If we forget this, then the zstd.Decoder
// won't leak resources.
decoderPool.Put(decoder)

// Create a sync.Pool which returns wrapped *zstd.Endoder's.
encoderPool := syncpool.NewEncoderPoolWrapper(zstd.WithEncoderConcurrency(1))

// Get an EncoderWrapper and use it.
encoder := encoderPool.Get(compressedDataWriter)
_, err = io.Copy(encoder, uncompressedDataReader)

// Return the encoder to the pool. If we forget this, then the zstd.Encoder
// won't leak resources.
encoderPool.Put(encoder)

Contributing

Bug reports, feature requests, PRs welcome.

License

Licensed under the Apache License, Version 2.0. See the LICENSE file.

Documentation

Overview

Package syncpool provides a non-leaky sync.Pool for github.com/klauspost/compress/zstd's Encoder and Decoder types, using wrappers (EncoderWrapper and DecoderWrapper).

Index

• func NewDecoderPool(options ...zstd.DOption) *sync.Pool
• func NewEncoderPool(options ...zstd.EOption) *sync.Pool
• type DecoderPoolWrapper
  • func NewDecoderPoolWrapper(options ...zstd.DOption) *DecoderPoolWrapper
  • func (d *DecoderPoolWrapper) Get(r io.Reader) *DecoderWrapper
  • func (d *DecoderPoolWrapper) Put(w *DecoderWrapper)
• type DecoderWrapper
  • func (w *DecoderWrapper) Close()
  • func (w *DecoderWrapper) IOReadCloser() io.ReadCloser
• type EncoderPoolWrapper
  • func NewEncoderPoolWrapper(options ...zstd.EOption) *EncoderPoolWrapper
  • func (e *EncoderPoolWrapper) Get(w io.Writer) *EncoderWrapper
  • func (e *EncoderPoolWrapper) Put(w *EncoderWrapper)
• type EncoderWrapper

Functions

func NewDecoderPool

func NewDecoderPool(options ...zstd.DOption) *sync.Pool

NewDecoderPool returns a sync.Pool that provides DecoderWrapper objects, which embed *zstd.Decoders. You probably want to include zstd.WithDecoderConcurrency(1) in the list of options.

func NewEncoderPool

func NewEncoderPool(options ...zstd.EOption) *sync.Pool

NewEncoderPool returns a sync.Pool that provides EncoderWrapper objects which embed a *zstd.Encoder. You probably want to include zstd.WithEncoderConcurrency(1) in the list of options.

Types

type DecoderPoolWrapper

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

DecoderPoolWrapper is a convenience wrapper for sync.Pool which only accepts and returns *DecoderWrapper's.

func NewDecoderPoolWrapper

func NewDecoderPoolWrapper(options ...zstd.DOption) *DecoderPoolWrapper

NewDecoderPoolWapper returns a *DecoderPoolWrapper that provides *DecoderWrapper objects that do not need to be type asserted. As with NewDecoderPool, you probably want to include zstd.WithDecoderConcurrency(1) in the list of options.

func (*DecoderPoolWrapper) Get

func (d *DecoderPoolWrapper) Get(r io.Reader) *DecoderWrapper

Get returns a *DecoderWrapper that has been Reset to use r.

func (*DecoderPoolWrapper) Put

func (d *DecoderPoolWrapper) Put(w *DecoderWrapper)

Put returns a *DecoderWrapper to the pool.

type DecoderWrapper

type DecoderWrapper struct {
	// *zstd.Decoder is not safe for use in a sync.Pool directly, since it
	// leaks data and goroutines. Finalizers on the *zstd.Decoder don't help
	// because the aforementioned goroutines reference the *zstd.Decoder and
	// prevent it from being garbage collected (so the finalizers don't run).
	//
	// We can work around this by storing this wrapper with an embedded
	// *zstd.Decoder in the sync.Pool, and using a finalizer on the wrapper
	// to Close the *zstd.Decoder.
	*zstd.Decoder
	// contains filtered or unexported fields
}

DecoderWrapper is a wrapper that embeds a *zstd.Decoder, and is safe for use in a sync.Pool.

func (*DecoderWrapper) Close

func (w *DecoderWrapper) Close()

Close does not close the embedded *zstd.Decoder (once closed, they cannot be reused), but instead resets it and places this *DecoderWrapper back in the pool.

func (*DecoderWrapper) IOReadCloser

func (w *DecoderWrapper) IOReadCloser() io.ReadCloser

IOReadCloser returns an io.ReadCloser that will return this *DecoderWrapper to the pool when it is closed.

type EncoderPoolWrapper

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

EncoderPoolWrapper is a convenience wrapper for sync.Pool which only accepts and returns *EncoderWrapper's.

func NewEncoderPoolWrapper

func NewEncoderPoolWrapper(options ...zstd.EOption) *EncoderPoolWrapper

NewEncoderPoolWapper returns an *EncoderPoolWrapper that provides *EncoderWrapper objects that do not need to be type asserted. As with NewEncoderPool, you probably want to include zstd.WithEncoderConcurrency(1) in the list of options.

func (*EncoderPoolWrapper) Get

func (e *EncoderPoolWrapper) Get(w io.Writer) *EncoderWrapper

Get returns an *EncoderWrapper that has been Reset to use w.

func (*EncoderPoolWrapper) Put

func (e *EncoderPoolWrapper) Put(w *EncoderWrapper)

Put returns an *EncoderWrapper to the pool.

type EncoderWrapper

type EncoderWrapper struct {
	*zstd.Encoder
}

EncoderWrapper is a wrapper that embeds a *zstd.Encoder, and is safe for use in a sync.Pool.