Repository: petermattis/pebble
Pull request #172 (2019-07-03 12:02:41)

Reduce concurrency bottlenecks in the table and block caches
* Shard tableCache and cache.Cache
* Update tableCacheNode.refCount atomically.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/172)
<!-- Reviewable:end -->

petermattis: I've add a 3rd commit which implements the idea mentioned in https://github.com/petermattis/pebble/issues/11#issuecomment-508116863 for `tableCache`. The results were even better than my hacky experiment suggested.

Patch:
diff --git a/cache/clockpro.go b/cache/clockpro.go
index e17686e..c515ab1 100644
--- a/cache/clockpro.go
+++ b/cache/clockpro.go
@@ -20,6 +20,7 @@ It is MIT licensed, like the original.
 package cache // import "github.com/petermattis/pebble/cache"
 
 import (
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"unsafe"
@@ -147,9 +148,8 @@ type WeakHandle interface {
 	Get() []byte
 }
 
-// Cache ...
-type Cache struct {
-	mu sync.Mutex
+type shard struct {
+	mu sync.RWMutex
 
 	maxSize  int64
 	coldSize int64
@@ -165,43 +165,17 @@ type Cache struct {
 	countTest int64
 }
 
-// New creates a new cache of the specified size. Memory for the cache is
-// allocated on demand, not during initialization.
-func New(size int64) *Cache {
-	return &Cache{
-		maxSize:  size,
-		coldSize: size,
-		blocks:   make(map[key]*entry),
-		files:    make(map[uint64]*entry),
-	}
-}
-
-// Get retrieves the cache value for the specified file and offset, returning
-// nil if no value is present.
-func (c *Cache) Get(fileNum, offset uint64) []byte {
-	if c == nil {
-		return nil
-	}
-
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
+func (c *shard) Get(fileNum, offset uint64) []byte {
+	c.mu.RLock()
 	e := c.blocks[key{fileNum: fileNum, offset: offset}]
+	c.mu.RUnlock()
 	if e == nil {
 		return nil
 	}
 	return e.Get()
 }
 
-// Set sets the cache value for the specified file and offset, overwriting an
-// existing value if present. A WeakHandle is returned which provides faster
-// retrieval of the cached value than Get (lock-free and avoidance of the map
-// lookup).
-func (c *Cache) Set(fileNum, offset uint64, value []byte) WeakHandle {
-	if c == nil {
-		return nil
-	}
-
+func (c *shard) Set(fileNum, offset uint64, value []byte) WeakHandle {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -248,11 +222,7 @@ func (c *Cache) Set(fileNum, offset uint64, value []byte) WeakHandle {
 }
 
 // EvictFile evicts all of the cache values for the specified file.
-func (c *Cache) EvictFile(fileNum uint64) {
-	if c == nil {
-		return
-	}
-
+func (c *shard) EvictFile(fileNum uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -277,23 +247,15 @@ func (c *Cache) EvictFile(fileNum uint64) {
 	}
 }
 
-// MaxSize returns the max size of the cache.
-func (c *Cache) MaxSize() int64 {
-	if c == nil {
-		return 0
-	}
-	return c.maxSize
-}
-
 // Size returns the current space used by the cache.
-func (c *Cache) Size() int64 {
+func (c *shard) Size() int64 {
 	c.mu.Lock()
 	size := c.countHot + c.countCold
 	c.mu.Unlock()
 	return size
 }
 
-func (c *Cache) metaAdd(key key, e *entry) {
+func (c *shard) metaAdd(key key, e *entry) {
 	c.evict()
 
 	c.blocks[key] = e
@@ -318,7 +280,7 @@ func (c *Cache) metaAdd(key key, e *entry) {
 	}
 }
 
-func (c *Cache) metaDel(e *entry) {
+func (c *shard) metaDel(e *entry) {
 	delete(c.blocks, e.key)
 
 	if e == c.handHot {
@@ -345,13 +307,13 @@ func (c *Cache) metaDel(e *entry) {
 	}
 }
 
-func (c *Cache) evict() {
+func (c *shard) evict() {
 	for c.maxSize <= c.countHot+c.countCold {
 		c.runHandCold()
 	}
 }
 
-func (c *Cache) runHandCold() {
+func (c *shard) runHandCold() {
 	if c.handCold == nil {
 		return
 	}
@@ -381,7 +343,7 @@ func (c *Cache) runHandCold() {
 	}
 }
 
-func (c *Cache) runHandHot() {
+func (c *shard) runHandHot() {
 	if c.handHot == c.handTest {
 		c.runHandTest()
 	}
@@ -403,7 +365,7 @@ func (c *Cache) runHandHot() {
 	c.handHot = c.handHot.next()
 }
 
-func (c *Cache) runHandTest() {
+func (c *shard) runHandTest() {
 	if c.countCold > 0 && c.handTest == c.handCold {
 		c.runHandCold()
 	}
@@ -426,3 +388,101 @@ func (c *Cache) runHandTest() {
 
 	c.handTest = c.handTest.next()
 }
+
+// Cache ...
+type Cache struct {
+	maxSize int64
+	shards  []shard
+}
+
+// New creates a new cache of the specified size. Memory for the cache is
+// allocated on demand, not during initialization.
+func New(size int64) *Cache {
+	return newShards(size, runtime.NumCPU())
+}
+
+func newShards(size int64, shards int) *Cache {
+	c := &Cache{
+		maxSize: size,
+		shards:  make([]shard, shards),
+	}
+	for i := range c.shards {
+		c.shards[i] = shard{
+			maxSize:  size / int64(len(c.shards)),
+			coldSize: size / int64(len(c.shards)),
+			blocks:   make(map[key]*entry),
+			files:    make(map[uint64]*entry),
+		}
+	}
+	return c
+}
+
+func (c *Cache) getShard(fileNum, offset uint64) *shard {
+	// Inlined version of fnv.New64 + Write.
+	const offset64 = 14695981039346656037
+	const prime64 = 1099511628211
+
+	h := uint64(offset64)
+	for i := 0; i < 8; i++ {
+		h *= prime64
+		h ^= uint64(fileNum & 0xff)
+		fileNum >>= 8
+	}
+	for i := 0; i < 8; i++ {
+		h *= prime64
+		h ^= uint64(offset & 0xff)
+		offset >>= 8
+	}
+
+	return &c.shards[h%uint64(len(c.shards))]
+}
+
+// Get retrieves the cache value for the specified file and offset, returning
+// nil if no value is present.
+func (c *Cache) Get(fileNum, offset uint64) []byte {
+	if c == nil {
+		return nil
+	}
+	return c.getShard(fileNum, offset).Get(fileNum, offset)
+}
+
+// Set sets the cache value for the specified file and offset, overwriting an
+// existing value if present. A WeakHandle is returned which provides faster
+// retrieval of the cached value than Get (lock-free and avoidance of the map
+// lookup).
+func (c *Cache) Set(fileNum, offset uint64, value []byte) WeakHandle {
+	if c == nil {
+		return nil
+	}
+	return c.getShard(fileNum, offset).Set(fileNum, offset, value)
+}
+
+// EvictFile evicts all of the cache values for the specified file.
+func (c *Cache) EvictFile(fileNum uint64) {
+	if c == nil {
+		return
+	}
+	for i := range c.shards {
+		c.shards[i].EvictFile(fileNum)
+	}
+}
+
+// MaxSize returns the max size of the cache.
+func (c *Cache) MaxSize() int64 {
+	if c == nil {
+		return 0
+	}
+	return c.maxSize
+}
+
+// Size returns the current space used by the cache.
+func (c *Cache) Size() int64 {
+	if c == nil {
+		return 0
+	}
+	var size int64
+	for i := range c.shards {
+		size += c.shards[i].Size()
+	}
+	return size
+}
diff --git a/open.go b/open.go
index 5d4475c..4de07c4 100644
--- a/open.go
+++ b/open.go
@@ -73,7 +73,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	if tableCacheSize < minTableCacheSize {
 		tableCacheSize = minTableCacheSize
 	}
-	d.tableCache.init(dirname, opts.FS, d.opts, tableCacheSize)
+	d.tableCache.init(dirname, opts.FS, d.opts, tableCacheSize, defaultTableCacheHitBuffer)
 	d.newIters = d.tableCache.newIters
 	d.commit = newCommitPipeline(commitEnv{
 		logSeqNum:     &d.mu.versions.logSeqNum,
diff --git a/table_cache.go b/table_cache.go
index 53a26bd..e72bc3e 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"runtime"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
@@ -16,41 +17,87 @@ import (
 	"github.com/petermattis/pebble/vfs"
 )
 
+const defaultTableCacheHitBuffer = 64
+
 var emptyIter = &errorIter{err: nil}
 
 type tableCache struct {
+	shards []tableCacheShard
+}
+
+func (c *tableCache) init(dirname string, fs vfs.FS, opts *Options, size, hitBuffer int) {
+	c.shards = make([]tableCacheShard, runtime.NumCPU())
+	for i := range c.shards {
+		c.shards[i].init(dirname, fs, opts, size/len(c.shards), hitBuffer)
+	}
+}
+
+func (c *tableCache) getShard(fileNum uint64) *tableCacheShard {
+	return &c.shards[fileNum%uint64(len(c.shards))]
+}
+
+func (c *tableCache) newIters(
+	meta *fileMetadata, opts *IterOptions,
+) (internalIterator, internalIterator, error) {
+	return c.getShard(meta.fileNum).newIters(meta, opts)
+}
+
+func (c *tableCache) evict(fileNum uint64) {
+	c.getShard(fileNum).evict(fileNum)
+}
+
+func (c *tableCache) Close() error {
+	for i := range c.shards {
+		err := c.shards[i].Close()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type tableCacheShard struct {
 	dirname string
 	fs      vfs.FS
 	opts    *Options
 	size    int
 
 	mu struct {
-		sync.Mutex
-		cond      sync.Cond
-		nodes     map[uint64]*tableCacheNode
-		iterCount int32
-		iters     map[*sstable.Iterator][]byte
-		dummy     tableCacheNode
-		releasing int
+		sync.RWMutex
+		nodes map[uint64]*tableCacheNode
+		// The iters map is only created and populated in race builds.
+		iters map[*sstable.Iterator][]byte
+		lru   tableCacheNode
 	}
+
+	iterCount int32
+	releasing sync.WaitGroup
+	hitsPool  *sync.Pool
 }
 
-func (c *tableCache) init(dirname string, fs vfs.FS, opts *Options, size int) {
+func (c *tableCacheShard) init(dirname string, fs vfs.FS, opts *Options, size, hitBuffer int) {
 	c.dirname = dirname
 	c.fs = fs
 	c.opts = opts
 	c.size = size
-	c.mu.cond.L = &c.mu.Mutex
 	c.mu.nodes = make(map[uint64]*tableCacheNode)
-	c.mu.dummy.next = &c.mu.dummy
-	c.mu.dummy.prev = &c.mu.dummy
+	c.mu.lru.next = &c.mu.lru
+	c.mu.lru.prev = &c.mu.lru
+	c.hitsPool = &sync.Pool{
+		New: func() interface{} {
+			return &tableCacheHits{
+				hits:  make([]*tableCacheNode, 0, hitBuffer),
+				shard: c,
+			}
+		},
+	}
 
 	if raceEnabled {
 		c.mu.iters = make(map[*sstable.Iterator][]byte)
 	}
 }
 
-func (c *tableCache) newIters(
+func (c *tableCacheShard) newIters(
 	meta *fileMetadata, opts *IterOptions,
 ) (internalIterator, internalIterator, error) {
 	// Calling findNode gives us the responsibility of decrementing n's
@@ -60,12 +107,7 @@ func (c *tableCache) newIters(
 	n := c.findNode(meta)
 	<-n.loaded
 	if n.err != nil {
-		if !c.unrefNode(n) {
-			// Try loading the table again; the error may be transient.
-			//
-			// TODO(peter): This could loop forever. That doesn't seem right.
-			go n.load(c)
-		}
+		c.unrefNode(n)
 		return nil, nil, n.err
 	}
 
@@ -78,7 +120,7 @@ func (c *tableCache) newIters(
 	}
 
 	iter := n.reader.NewIter(opts.GetLowerBound(), opts.GetUpperBound())
-	atomic.AddInt32(&c.mu.iterCount, 1)
+	atomic.AddInt32(&c.iterCount, 1)
 	if raceEnabled {
 		c.mu.Lock()
 		c.mu.iters[iter] = debug.Stack()
@@ -86,17 +128,13 @@ func (c *tableCache) newIters(
 	}
 
 	iter.SetCloseHook(func() error {
-		c.mu.Lock()
-		atomic.AddInt32(&c.mu.iterCount, -1)
 		if raceEnabled {
+			c.mu.Lock()
 			delete(c.mu.iters, iter)
+			c.mu.Unlock()
 		}
-		n.refCount--
-		if n.refCount == 0 {
-			c.mu.releasing++
-			go n.release(c)
-		}
-		c.mu.Unlock()
+		c.unrefNode(n)
+		atomic.AddInt32(&c.iterCount, -1)
 		return nil
 	})
 
@@ -109,16 +147,27 @@ func (c *tableCache) newIters(
 	return iter, nil, nil
 }
 
-// releaseNode releases a node from the tableCache.
+// releaseNode releases a node from the tableCacheShard.
 //
 // c.mu must be held when calling this.
-func (c *tableCache) releaseNode(n *tableCacheNode) {
+func (c *tableCacheShard) releaseNode(n *tableCacheNode) {
 	delete(c.mu.nodes, n.meta.fileNum)
 	n.next.prev = n.prev
 	n.prev.next = n.next
-	n.refCount--
-	if n.refCount == 0 {
-		c.mu.releasing++
+	n.prev = nil
+	n.next = nil
+	c.unrefNode(n)
+}
+
+// unrefNode decrements the reference count for the specified node, releasing
+// it if the reference count fell to 0. Note that the node has a reference if
+// it is present in tableCacheShard.mu.nodes, so a reference count of 0 means the
+// node has already been removed from that map.
+//
+// Returns true if the node was released and false otherwise.
+func (c *tableCacheShard) unrefNode(n *tableCacheNode) {
+	if atomic.AddInt32(&n.refCount, -1) == 0 {
+		c.releasing.Add(1)
 		go n.release(c)
 	}
 }
@@ -126,10 +175,39 @@ func (c *tableCache) releaseNode(n *tableCacheNode) {
 // findNode returns the node for the table with the given file number, creating
 // that node if it didn't already exist. The caller is responsible for
 // decrementing the returned node's refCount.
-func (c *tableCache) findNode(meta *fileMetadata) *tableCacheNode {
+func (c *tableCacheShard) findNode(meta *fileMetadata) *tableCacheNode {
+	// Fast-path for a hit in the cache. We grab the lock in shared mode, and use
+	// a batching mechanism to perform updates to the LRU list.
+	c.mu.RLock()
+	if n := c.mu.nodes[meta.fileNum]; n != nil {
+		// The caller is responsible for decrementing the refCount.
+		atomic.AddInt32(&n.refCount, 1)
+		c.mu.RUnlock()
+
+		// Record a hit for the node. This has to be done with tableCacheShard.mu
+		// unlocked as it might result in a call to
+		// tableCacheShard.recordHits. Note that the sync.Pool acts as a
+		// thread-local cache of the accesses. This is lossy (a GC can result in
+		// the sync.Pool be cleared), but that is ok as we don't need perfect
+		// accuracy for the LRU list.
+		hits := c.hitsPool.Get().(*tableCacheHits)
+		hits.recordHit(n)
+		c.hitsPool.Put(hits)
+		return n
+	}
+	c.mu.RUnlock()
+
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	{
+		// Flush the thread-local hits buffer as we already have the shard locked
+		// exclusively.
+		hits := c.hitsPool.Get().(*tableCacheHits)
+		hits.flushLocked()
+		c.hitsPool.Put(hits)
+	}
+
 	n := c.mu.nodes[meta.fileNum]
 	if n == nil {
 		n = &tableCacheNode{
@@ -140,7 +218,7 @@ func (c *tableCache) findNode(meta *fileMetadata) *tableCacheNode {
 		c.mu.nodes[meta.fileNum] = n
 		if len(c.mu.nodes) > c.size {
 			// Release the tail node.
-			c.releaseNode(c.mu.dummy.prev)
+			c.releaseNode(c.mu.lru.prev)
 		}
 		go n.load(c)
 	} else {
@@ -149,49 +227,53 @@ func (c *tableCache) findNode(meta *fileMetadata) *tableCacheNode {
 		n.prev.next = n.next
 	}
 	// Insert n at the front of the doubly-linked list.
-	n.next = c.mu.dummy.next
-	n.prev = &c.mu.dummy
+	n.next = c.mu.lru.next
+	n.prev = &c.mu.lru
 	n.next.prev = n
 	n.prev.next = n
 	// The caller is responsible for decrementing the refCount.
-	n.refCount++
+	atomic.AddInt32(&n.refCount, 1)
 	return n
 }
 
-// unrefNode decrements the reference count for the specified node, releasing
-// it if the reference count fell to 0. Note that the node has a reference if
-// it is present in tableCache.mu.nodes, so a reference count of 0 means the
-// node has already been removed from that map.
-//
-// Returns true if the node was released and false otherwise.
-func (c *tableCache) unrefNode(n *tableCacheNode) bool {
+func (c *tableCacheShard) evict(fileNum uint64) {
 	c.mu.Lock()
-	n.refCount--
-	res := false
-	if n.refCount == 0 {
-		c.mu.releasing++
-		go n.release(c)
-		res = true
+	if n := c.mu.nodes[fileNum]; n != nil {
+		c.releaseNode(n)
 	}
 	c.mu.Unlock()
-	return res
+
+	c.opts.Cache.EvictFile(fileNum)
 }
 
-func (c *tableCache) evict(fileNum uint64) {
+func (c *tableCacheShard) recordHits(hits []*tableCacheNode) {
 	c.mu.Lock()
-	if n := c.mu.nodes[fileNum]; n != nil {
-		c.releaseNode(n)
-	}
+	c.recordHitsLocked(hits)
 	c.mu.Unlock()
+}
 
-	c.opts.Cache.EvictFile(fileNum)
+func (c *tableCacheShard) recordHitsLocked(hits []*tableCacheNode) {
+	for _, n := range hits {
+		if n.next == nil || n.prev == nil {
+			// The node is no longer on the LRU list.
+			continue
+		}
+		// Remove n from the doubly-linked list.
+		n.next.prev = n.prev
+		n.prev.next = n.next
+		// Insert n at the front of the doubly-linked list.
+		n.next = c.mu.lru.next
+		n.prev = &c.mu.lru
+		n.next.prev = n
+		n.prev.next = n
+	}
 }
 
-func (c *tableCache) Close() error {
+func (c *tableCacheShard) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	if v := atomic.LoadInt32(&c.mu.iterCount); v > 0 {
+	if v := atomic.LoadInt32(&c.iterCount); v > 0 {
 		if !raceEnabled {
 			return fmt.Errorf("leaked iterators: %d", v)
 		}
@@ -203,20 +285,17 @@ func (c *tableCache) Close() error {
 		return errors.New(buf.String())
 	}
 
-	for n := c.mu.dummy.next; n != &c.mu.dummy; n = n.next {
-		n.refCount--
-		if n.refCount == 0 {
-			c.mu.releasing++
+	for n := c.mu.lru.next; n != &c.mu.lru; n = n.next {
+		if atomic.AddInt32(&n.refCount, -1) == 0 {
+			c.releasing.Add(1)
 			go n.release(c)
 		}
 	}
 	c.mu.nodes = nil
-	c.mu.dummy.next = nil
-	c.mu.dummy.prev = nil
+	c.mu.lru.next = nil
+	c.mu.lru.prev = nil
 
-	for c.mu.releasing > 0 {
-		c.mu.cond.Wait()
-	}
+	c.releasing.Wait()
 	return nil
 }
 
@@ -229,10 +308,10 @@ type tableCacheNode struct {
 	// The remaining fields are protected by the tableCache mutex.
 
 	next, prev *tableCacheNode
-	refCount   int
+	refCount   int32
 }
 
-func (n *tableCacheNode) load(c *tableCache) {
+func (n *tableCacheNode) load(c *tableCacheShard) {
 	// Try opening the fileTypeTable first.
 	f, err := c.fs.Open(dbFilename(c.dirname, fileTypeTable, n.meta.fileNum))
 	if err != nil {
@@ -248,15 +327,32 @@ func (n *tableCacheNode) load(c *tableCache) {
 	close(n.loaded)
 }
 
-func (n *tableCacheNode) release(c *tableCache) {
+func (n *tableCacheNode) release(c *tableCacheShard) {
 	<-n.loaded
 	// Nothing to be done about an error at this point. Close the reader if it is
 	// open.
 	if n.reader != nil {
 		_ = n.reader.Close()
 	}
-	c.mu.Lock()
-	c.mu.releasing--
-	c.mu.Unlock()
-	c.mu.cond.Signal()
+	c.releasing.Done()
+}
+
+// tableCacheHits batches a set of node accesses in order to amortize exclusive
+// lock acquisition.
+type tableCacheHits struct {
+	hits  []*tableCacheNode
+	shard *tableCacheShard
+}
+
+func (f *tableCacheHits) recordHit(n *tableCacheNode) {
+	f.hits = append(f.hits, n)
+	if len(f.hits) == cap(f.hits) {
+		f.shard.recordHits(f.hits)
+		f.hits = f.hits[:0]
+	}
+}
+
+func (f *tableCacheHits) flushLocked() {
+	f.shard.recordHitsLocked(f.hits)
+	f.hits = f.hits[:0]
 }


Test patch:
diff --git a/cache/clockpro_test.go b/cache/clockpro_test.go
index 6b6f9b1..c35dd6e 100644
--- a/cache/clockpro_test.go
+++ b/cache/clockpro_test.go
@@ -18,7 +18,7 @@ func TestCache(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	cache := New(200)
+	cache := newShards(200, 1)
 	scanner := bufio.NewScanner(f)
 
 	for scanner.Scan() {
@@ -47,7 +47,7 @@ func TestCache(t *testing.T) {
 }
 
 func TestWeakHandle(t *testing.T) {
-	cache := New(5)
+	cache := newShards(5, 1)
 	cache.Set(1, 0, bytes.Repeat([]byte("a"), 5))
 	h := cache.Set(0, 0, bytes.Repeat([]byte("b"), 5))
 	if v := h.Get(); string(v) != "bbbbb" {
@@ -60,7 +60,7 @@ func TestWeakHandle(t *testing.T) {
 }
 
 func TestEvictFile(t *testing.T) {
-	cache := New(100)
+	cache := newShards(100, 1)
 	cache.Set(0, 0, bytes.Repeat([]byte("a"), 5))
 	cache.Set(1, 0, bytes.Repeat([]byte("a"), 5))
 	cache.Set(2, 0, bytes.Repeat([]byte("a"), 5))
diff --git a/table_cache_test.go b/table_cache_test.go
index 12eac38..9c31990 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -117,8 +117,9 @@ func (fs *tableCacheTestFS) validateNoneStillOpen() error {
 }
 
 const (
-	tableCacheTestNumTables = 300
-	tableCacheTestCacheSize = 100
+	tableCacheTestNumTables     = 300
+	tableCacheTestCacheSize     = 100
+	tableCacheTestHitBufferSize = 64
 )
 
 func newTableCache() (*tableCache, *tableCacheTestFS, error) {
@@ -149,7 +150,7 @@ func newTableCache() (*tableCache, *tableCacheTestFS, error) {
 	opts := &Options{}
 	opts.EnsureDefaults()
 	c := &tableCache{}
-	c.init("", fs, opts, tableCacheTestCacheSize)
+	c.init("", fs, opts, tableCacheTestCacheSize, tableCacheTestHitBufferSize)
 	return c, fs, nil
 }