Merge pull request #23 from bodgit/speedup

Implement pooling of decompressed folder readers

.golangci.yaml

@@ -2,17 +2,12 @@
 linters:
   enable-all: true
   disable:
-    - cyclop
     - exhaustivestruct
-    - funlen
-    - gochecknoglobals
-    - gochecknoinits
-    - gocognit
-    - gocyclo
+    - exhaustruct
+    - godox
     - goerr113
     - gomnd
-    - godox
     - ireturn
-    - nestif
+    - nonamedreturns
     - varnamelen
     - wrapcheck

internal/deflate/reader.go

@@ -5,20 +5,19 @@ import (
 	"errors"
 	"io"
 	"sync"
+
+	"github.com/bodgit/sevenzip/internal/util"
 )
 
+//nolint:gochecknoglobals
 var flateReaderPool sync.Pool
 
 type readCloser struct {
 	c  io.Closer
 	fr io.ReadCloser
-	mu sync.Mutex
 }
 
 func (rc *readCloser) Close() error {
-	rc.mu.Lock()
-	defer rc.mu.Unlock()
-
 	var err error
 
 	if rc.c != nil {
@@ -35,9 +34,6 @@ func (rc *readCloser) Close() error {
 }
 
 func (rc *readCloser) Read(p []byte) (int, error) {
-	rc.mu.Lock()
-	defer rc.mu.Unlock()
-
 	if rc.fr == nil {
 		return 0, errors.New("deflate: Read after Close")
 	}
@@ -55,12 +51,12 @@ func NewReader(_ []byte, _ uint64, readers []io.ReadCloser) (io.ReadCloser, erro
 	if ok {
 		frf, ok := fr.(flate.Resetter)
 		if ok {
-			if err := frf.Reset(readers[0], nil); err != nil {
+			if err := frf.Reset(util.ByteReadCloser(readers[0]), nil); err != nil {
 				return nil, err
 			}
 		}
 	} else {
-		fr = flate.NewReader(readers[0])
+		fr = flate.NewReader(util.ByteReadCloser(readers[0]))
 	}
 
 	return &readCloser{

internal/pool/pool.go

@@ -0,0 +1,137 @@
+package pool
+
+import (
+	"container/list"
+	"runtime"
+	"sort"
+	"sync"
+
+	"github.com/bodgit/sevenzip/internal/util"
+)
+
+// Pooler is the interface implemented by a pool.
+type Pooler interface {
+	Get(int64) (util.SizeReadSeekCloser, bool)
+	Put(int64, util.SizeReadSeekCloser) (bool, error)
+}
+
+// Constructor is the function prototype used to instantiate a pool.
+type Constructor func() (Pooler, error)
+
+type noopPool struct{}
+
+// NewNoopPool returns a Pooler that doesn't actually pool anything.
+func NewNoopPool() (Pooler, error) {
+	return new(noopPool), nil
+}
+
+func (noopPool) Get(_ int64) (util.SizeReadSeekCloser, bool) {
+	return nil, false
+}
+
+func (noopPool) Put(_ int64, rc util.SizeReadSeekCloser) (bool, error) {
+	return false, rc.Close()
+}
+
+type pool struct {
+	mutex     sync.Mutex
+	size      int
+	evictList *list.List
+	items     map[int64]*list.Element
+}
+
+type entry struct {
+	key   int64
+	value util.SizeReadSeekCloser
+}
+
+// NewPool returns a Pooler that uses a LRU strategy to maintain a fixed pool
+// of util.SizeReadSeekCloser's keyed by their stream offset.
+func NewPool() (Pooler, error) {
+	return &pool{
+		size:      runtime.NumCPU(),
+		evictList: list.New(),
+		items:     make(map[int64]*list.Element),
+	}, nil
+}
+
+func (p *pool) Get(offset int64) (util.SizeReadSeekCloser, bool) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	if ent, ok := p.items[offset]; ok {
+		_ = p.removeElement(ent, false)
+
+		return ent.Value.(*entry).value, true //nolint:forcetypeassert
+	}
+
+	// Sort keys in descending order
+	keys := p.keys()
+	sort.Slice(keys, func(i, j int) bool { return keys[i] > keys[j] })
+
+	for _, k := range keys {
+		// First key less than offset is the closest
+		if k < offset {
+			ent := p.items[k]
+			_ = p.removeElement(ent, false)
+
+			return ent.Value.(*entry).value, true //nolint:forcetypeassert
+		}
+	}
+
+	return nil, false
+}
+
+func (p *pool) Put(offset int64, rc util.SizeReadSeekCloser) (bool, error) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	if _, ok := p.items[offset]; ok {
+		return false, nil
+	}
+
+	ent := &entry{offset, rc}
+	entry := p.evictList.PushFront(ent)
+	p.items[offset] = entry
+
+	var err error
+
+	evict := p.evictList.Len() > p.size
+	if evict {
+		err = p.removeOldest()
+	}
+
+	return evict, err
+}
+
+func (p *pool) keys() []int64 {
+	keys := make([]int64, len(p.items))
+	i := 0
+
+	for ent := p.evictList.Back(); ent != nil; ent = ent.Prev() {
+		keys[i] = ent.Value.(*entry).key //nolint:forcetypeassert
+		i++
+	}
+
+	return keys
+}
+
+func (p *pool) removeOldest() error {
+	if ent := p.evictList.Back(); ent != nil {
+		return p.removeElement(ent, true)
+	}
+
+	return nil
+}
+
+func (p *pool) removeElement(e *list.Element, cb bool) error {
+	p.evictList.Remove(e)
+	kv := e.Value.(*entry) //nolint:forcetypeassert
+	delete(p.items, kv.key)
+
+	if cb {
+		return kv.value.Close()
+	}
+
+	return nil
+}

internal/util/reader.go

@@ -2,6 +2,15 @@ package util
 
 import "io"
 
+// SizeReadSeekCloser is an io.Reader, io.Seeker, and io.Closer with a Size
+// method.
+type SizeReadSeekCloser interface {
+	io.Reader
+	io.Seeker
+	io.Closer
+	Size() int64
+}
+
 // Reader is both an io.Reader and io.ByteReader.
 type Reader interface {
 	io.Reader