eth, freezer: add immutable chain freezer

eth/backend.go

@@ -24,6 +24,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/common"
@@ -41,6 +42,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/freezer"
 	"github.com/ethereum/go-ethereum/internal/ethapi"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/miner"
@@ -69,6 +71,7 @@ type Ethereum struct {
 	// Handlers
 	txPool          *core.TxPool
 	blockchain      *core.BlockChain
+	freezer         *freezer.Freezer
 	protocolManager *ProtocolManager
 	lesServer       LesServer
 
@@ -168,6 +171,12 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	if err != nil {
 		return nil, err
 	}
+	eth.freezer, err = freezer.New(ctx.ResolvePath("freezer"))
+	if err != nil {
+		return nil, err
+	}
+	go eth.freezer.Freeze(eth.blockchain, chainDb, time.Minute, 60000)
+
 	// Rewind the chain in case of an incompatible config upgrade.
 	if compat, ok := genesisErr.(*params.ConfigCompatError); ok {
 		log.Warn("Rewinding chain to upgrade configuration", "err", compat)

freezer/freezer.go

@@ -0,0 +1,213 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// Package freezer implements an append-only immutable mmap chain database.
+package freezer
+
+import (
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+// Freezer is an memory mapped append-only database to store immutable chain data
+// into flat files:
+//
+// - The append only nature ensures that disk writes are minimized.
+// - The memory mapping ensures we can max out system memory for caching without
+//   reserving it for go-ethereum. This would also reduce the memory requirements
+//   of Geth, and thus also GC overhead.
+type Freezer struct {
+	frozen uint64 // Number of blocks already frozen
+
+	headers  *table // Data table for storing the block headers
+	bodies   *table // Data table for storing the block bodies
+	receipts *table // Data table for storing the block receipts
+	diffs    *table // Data table for storing the block tds
+
+	logger log.Logger // Contextual logger for the freezer database
+}
+
+// New creates a chain freezer that moves ancient chain data into immutable flat
+// file containers.
+func New(datadir string) (*Freezer, error) {
+	// Create the initial freezer object
+	var (
+		freezer = &Freezer{
+			logger: log.New("path", datadir),
+		}
+		readMeter  = metrics.NewRegisteredMeter("eth/db/freezer/read", nil)
+		writeMeter = metrics.NewRegisteredMeter("eth/db/freezer/write", nil)
+		err        error
+	)
+	// Open all the supported data tables
+	if freezer.headers, err = newTable(datadir, "headers", readMeter, writeMeter); err != nil {
+		return nil, err
+	}
+	if freezer.bodies, err = newTable(datadir, "bodies", readMeter, writeMeter); err != nil {
+		freezer.headers.Close()
+		return nil, err
+	}
+	if freezer.receipts, err = newTable(datadir, "receipts", readMeter, writeMeter); err != nil {
+		freezer.bodies.Close()
+		freezer.headers.Close()
+		return nil, err
+	}
+	if freezer.diffs, err = newTable(datadir, "diffs", readMeter, writeMeter); err != nil {
+		freezer.receipts.Close()
+		freezer.bodies.Close()
+		freezer.headers.Close()
+		return nil, err
+	}
+	return freezer, nil
+}
+
+// Close terminates the chain freezer, unmapping all the data files.
+func (f *Freezer) Close() error {
+	f.diffs.Close()
+	f.receipts.Close()
+	f.bodies.Close()
+	f.headers.Close()
+
+	return nil
+}
+
+// Freeze is a background thread that periodically checks the blockchain for any
+// import progress and moves ancient data from the fast database into the freezer.
+//
+// This functionality is deliberately broken off from block importing to avoid
+// incurring additional data shuffling delays on block propagation.
+func (f *Freezer) Freeze(chain *core.BlockChain, db ethdb.Database, recheck time.Duration, delay uint64) {
+	for {
+		// Retrieve the freezing threshold. In theory we're interested only in full
+		// blocks post-sync, but that would keep the live database enormous during
+		// dast sync. By picking the fast block, we still get to deep freeze all the
+		// final immutable data without having to wait for sync to finish.
+		head := chain.CurrentFastBlock()
+		if head == nil {
+			log.Error("Current fast block is nil")
+			time.Sleep(recheck)
+			continue
+		}
+		if head.NumberU64() < delay {
+			log.Debug("Current block not old enough", "number", head.Number(), "hash", head.Hash(), "age", common.PrettyAge(time.Unix(head.Time().Int64(), 0)), "delay", delay)
+			time.Sleep(recheck)
+			continue
+		}
+		limit := head.NumberU64() - delay
+		if limit <= f.frozen {
+			log.Debug("Ancient blocks frozen already")
+			time.Sleep(recheck)
+			continue
+		}
+		// Seems we have data ready to be frozen, process in usable batches
+		if limit-f.frozen > 30000 {
+			limit = f.frozen + 30000
+		}
+		var (
+			start = time.Now()
+			first = f.frozen
+			last  *types.Block
+		)
+		for f.frozen < limit {
+			// Deep freeze the next canonical block if it's available
+			if block := chain.GetBlockByNumber(f.frozen); block != nil {
+				// Deep freeze the block header and body
+				blob, _ := rlp.EncodeToBytes(block.Header())
+				if err := f.headers.Append(f.frozen, blob); err != nil {
+					log.Error("Failed to deep freeze header", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err)
+					break
+				}
+				blob, _ = rlp.EncodeToBytes(block.Body())
+				if err := f.bodies.Append(f.frozen, blob); err != nil {
+					log.Error("Failed to deep freeze body", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err)
+					break
+				}
+				// Deep freeze the block receipts and total difficulty
+				if receipts := chain.GetReceiptsByHash(block.Hash()); receipts != nil {
+					blob, _ = rlp.EncodeToBytes(receipts)
+					if err := f.receipts.Append(f.frozen, blob); err != nil {
+						log.Error("Failed to deep freeze receipts", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err)
+						break
+					}
+				}
+				if td := chain.GetTd(block.Hash(), block.NumberU64()); td != nil {
+					blob, _ = rlp.EncodeToBytes(td)
+					if err := f.diffs.Append(f.frozen, blob); err != nil {
+						log.Error("Failed to deep freeze difficulty", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err)
+						break
+					}
+				}
+				log.Trace("Deep froze ancient block", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)))
+				f.frozen++
+
+				// If it's the last block, save for reporting
+				if f.frozen == limit-1 {
+					last = block
+				}
+			}
+		}
+		// Batch of blocks have been frozen, flush them before wiping from leveldb
+		if err := f.headers.Flush(); err != nil {
+			f.logger.Error("Failed to flush frozen headers", "err", err)
+			time.Sleep(recheck)
+			continue
+		}
+		if err := f.bodies.Flush(); err != nil {
+			f.logger.Error("Failed to flush frozen bodies", "err", err)
+			time.Sleep(recheck)
+			continue
+		}
+		if err := f.receipts.Flush(); err != nil {
+			f.logger.Error("Failed to flush frozen receipts", "err", err)
+			time.Sleep(recheck)
+			continue
+		}
+		if err := f.diffs.Flush(); err != nil {
+			f.logger.Error("Failed to flush frozen diffs", "err", err)
+			time.Sleep(recheck)
+			continue
+		}
+		// Wipe out all data from the active database
+		for number := first; number < f.frozen; number++ {
+			if number == 0 {
+				// Skip deleting the genesis for the PoC
+				continue
+			}
+			rawdb.DeleteBlock(db, rawdb.ReadCanonicalHash(db, number), number)
+		}
+		// Log something friendly for the user
+		context := []interface{}{
+			"count", f.frozen - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", f.frozen - 1,
+		}
+		if last != nil {
+			context = append(context, []interface{}{"hash", last.Hash(), "age", common.PrettyAge(time.Unix(last.Time().Int64(), 0))}...)
+		}
+		log.Info("Deep froze chain segment", context...)
+
+		// Avoid database thrashing with tiny writes
+		if f.frozen-first < 30000 {
+			time.Sleep(recheck)
+		}
+	}
+}

freezer/mmap.go

@@ -0,0 +1,86 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package freezer
+
+import (
+	"os"
+	"reflect"
+	"unsafe"
+
+	"github.com/edsrzf/mmap-go"
+)
+
+// openWithSize opens a file and ensures it is at least the provided bytes in
+// size, growing it if necessary.
+func openWithSize(path string, size uint64) (*os.File, error) {
+	// Open the file for writing, potentially creating it
+	file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return nil, err
+	}
+	// Ensure the file's size is at least as much as requested
+	info, err := file.Stat()
+	if err != nil {
+		file.Close()
+		return nil, err
+	}
+	if info.Size() < int64(size) {
+		if err := file.Truncate(int64(size)); err != nil {
+			file.Close()
+			return nil, err
+		}
+	}
+	// File is correctly open and of the correct size
+	return file, err
+}
+
+// mmapBytes tries to memory map a file, creating it if it's non existent.
+func mmapBytes(path string, size uint64) (*os.File, mmap.MMap, []byte, error) {
+	// Open the file to memory map and ensure it's large enough
+	file, err := openWithSize(path, size)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	// Memory map the file, cast to a byte slice and return
+	mem, err := mmap.Map(file, mmap.RDWR, 0)
+	if err != nil {
+		file.Close()
+		return nil, nil, nil, err
+	}
+	header := *(*reflect.SliceHeader)(unsafe.Pointer(&mem))
+	return file, mem, *(*[]byte)(unsafe.Pointer(&header)), nil
+}
+
+// mmapUints tries to memory map a file, creating it if it's non existent.
+func mmapUints(path string, size uint64) (*os.File, mmap.MMap, []uint64, error) {
+	// Open the file to memory map and ensure it's large enough
+	file, err := openWithSize(path, size)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	// Memory map the file, cast to an uint64 slice and return
+	mem, err := mmap.Map(file, mmap.RDWR, 0)
+	if err != nil {
+		file.Close()
+		return nil, nil, nil, err
+	}
+	header := *(*reflect.SliceHeader)(unsafe.Pointer(&mem))
+	header.Len /= 8
+	header.Cap /= 8
+
+	return file, mem, *(*[]uint64)(unsafe.Pointer(&header)), nil
+}

freezer/table.go

@@ -0,0 +1,290 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package freezer
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/edsrzf/mmap-go"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/golang/snappy"
+)
+
+const (
+	growthIndex uint64 = 1024 * 1024       // Growth rate of the index file when remapping
+	growthData  uint64 = 128 * 1024 * 1024 // Growth rate of the data file when remapping
+)
+
+var (
+	// errAlreadyExists is returned if the user attempts to append an item to the
+	// freezer table that already exists (i.e. it's offset is lower or equal to
+	// the head of the immutable file).
+	errAlreadyExists = errors.New("item already exists")
+
+	// errGappedWrite is returned if the user attempts to append an item to the
+	// freezer table that would produce a data gap (i.e. it's offset is larger than
+	// the head of the immutable file).
+	errGappedWrite = errors.New("item produces data gap")
+
+	// errTableInaccessible is returned if a previously well functioning freezer
+	// table becomes non-accessible after a memory remap (resize).
+	errTableInaccessible = errors.New("table not accessible")
+
+	// errOutOfBounds is returned if the item requested is not contained within the
+	// freezer table.
+	errOutOfBounds = errors.New("out of bounds")
+)
+
+// table represents a single chained data table within the freezer (e.g. blocks).
+// It consists of a data file (snappy encoded arbitrary data blobs) and an index
+// file (uncompressed 64 bit indices into the data file). In addition a counter
+// (binary) file is also created which simply contains the number of items.
+type table struct {
+	path string // Database folder to store the files into
+	name string // Table name to multiplex multiple tables into the same folder
+
+	fileData    *os.File // File descriptor for the data region
+	fileIndex   *os.File // File descriptor for the offset region
+	fileCounter *os.File // File descriptor for the counter region
+
+	mmapData    mmap.MMap // Memory mapping for the data region
+	mmapIndex   mmap.MMap // Memory mapping for the offset region
+	mmapCounter mmap.MMap // Memory mapping for the counter region
+
+	rawData    []byte   // Direct memory region for the data file
+	rawIndex   []uint64 // Direct memory region for the offset file
+	rawCounter []uint64 // Direct memory region for the counter (single item)
+
+	readMeter  metrics.Meter // Meter for measuring the effective amount of data read
+	writeMeter metrics.Meter // Meter for measuring the effective amount of data written
+
+	logger log.Logger   // Logger with database path and table name ambedded
+	lock   sync.RWMutex // Lock protecting the reads from remaps
+}
+
+// newTable attempts to new freezer table by memory mapping the composing files.
+// If no file exists, it will create new ones.
+func newTable(path string, name string, readMeter metrics.Meter, writeMeter metrics.Meter) (*table, error) {
+	if err := os.MkdirAll(path, 0755); err != nil {
+		return nil, err
+	}
+	tab := &table{
+		path:       path,
+		name:       name,
+		readMeter:  readMeter,
+		writeMeter: writeMeter,
+		logger:     log.New("path", path, "table", name),
+	}
+	if err := tab.ensure(); err != nil {
+		return nil, err
+	}
+	return tab, nil
+}
+
+// ensure checks all the memory region limits and ensures that they conform to
+// the required data counts. If anything is off, this method will recreate and
+// remap accordingly.
+func (t *table) ensure() error {
+	// Figure out if we need to remap or not
+	// If the regions are already open and large enough, leave as is
+	if t.rawCounter != nil {
+		// Ensure the data file is large enough, unmap it otherwise
+		wantData := (t.rawIndex[t.rawCounter[0]]/growthData + 1) * growthData
+		if uint64(len(t.rawData)) < wantData {
+			if err := t.mmapData.Unmap(); err != nil {
+				t.logger.Error("Failed to unmap freezer datastore", "err", err)
+			}
+			if err := t.fileData.Close(); err != nil {
+				t.logger.Error("Failed to close freezer datastore", "err", err)
+			}
+			t.fileData, t.mmapData, t.rawData = nil, nil, nil
+		}
+		// Ensure the index file is large enough, unmap it otherwise
+		wantIndex := (((t.rawCounter[0]+1)*8)/growthIndex + 1) * growthIndex
+		if uint64(len(t.rawIndex)) < wantIndex {
+			if err := t.mmapIndex.Unmap(); err != nil {
+				t.logger.Error("Failed to unmap freezer index", "err", err)
+			}
+			if err := t.fileIndex.Close(); err != nil {
+				t.logger.Error("Failed to close freezer index", "err", err)
+			}
+			t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil
+		}
+	}
+	// Memory map the counter and retrieve the size of the offset file
+	var err error
+
+	if t.rawCounter == nil {
+		if t.fileCounter, t.mmapCounter, t.rawCounter, err = mmapUints(filepath.Join(t.path, t.name+".len"), 8); err != nil {
+			return err
+		}
+	}
+	// Memory map the index file and retrieve the size of the data file
+	if t.rawIndex == nil {
+		size := (((t.rawCounter[0]+1)*8)/growthIndex + 1) * growthIndex
+		if t.fileIndex, t.mmapIndex, t.rawIndex, err = mmapUints(filepath.Join(t.path, t.name+".idx"), size); err != nil {
+			if err := t.mmapCounter.Unmap(); err != nil {
+				t.logger.Error("Failed to unmap freezer counter", "err", err)
+			}
+			if err := t.fileCounter.Close(); err != nil {
+				t.logger.Error("Failed to close freezer counter", "err", err)
+			}
+			t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil
+			return err
+		}
+	}
+	// Memory map the data file and return the final freezer table
+	if t.rawData == nil {
+		size := growthData
+		if t.rawCounter[0] > 0 {
+			size = (t.rawIndex[t.rawCounter[0]]/growthData + 1) * growthData
+		}
+		if t.fileData, t.mmapData, t.rawData, err = mmapBytes(filepath.Join(t.path, t.name+".dat"), size); err != nil {
+			if err := t.mmapIndex.Unmap(); err != nil {
+				t.logger.Error("Failed to unmap freezer index", "err", err)
+			}
+			if err := t.fileIndex.Close(); err != nil {
+				t.logger.Error("Failed to close freezer index", "err", err)
+			}
+			t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil
+
+			if err := t.mmapCounter.Unmap(); err != nil {
+				t.logger.Error("Failed to unmap freezer counter", "err", err)
+			}
+			if err := t.fileCounter.Close(); err != nil {
+				t.logger.Error("Failed to close freezer counter", "err", err)
+			}
+			t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil
+
+			return err
+		}
+	}
+	return nil
+}
+
+// Close unmaps all active memory mapped regions.
+func (t *table) Close() error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	if t.mmapData != nil {
+		if err := t.mmapData.Unmap(); err != nil {
+			t.logger.Error("Failed to unmap freezer datastore", "err", err)
+		}
+		if err := t.fileData.Close(); err != nil {
+			t.logger.Error("Failed to close freezer datastore", "err", err)
+		}
+		t.fileData, t.mmapData, t.rawData = nil, nil, nil
+	}
+	if t.mmapIndex != nil {
+		if err := t.mmapIndex.Unmap(); err != nil {
+			t.logger.Error("Failed to unmap freezer index", "err", err)
+		}
+		if err := t.fileIndex.Close(); err != nil {
+			t.logger.Error("Failed to close freezer index", "err", err)
+		}
+		t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil
+	}
+	if t.mmapCounter != nil {
+		if err := t.mmapCounter.Unmap(); err != nil {
+			t.logger.Error("Failed to unmap freezer counter", "err", err)
+		}
+		if err := t.fileCounter.Close(); err != nil {
+			t.logger.Error("Failed to close freezer counter", "err", err)
+		}
+		t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil
+	}
+	return nil
+}
+
+// Append injects a binary blob at the end of the freezer table. The item index
+// is a precautionary parameter to ensure data correctness, but the table will
+// reject already existing data.
+//
+// Note, this method will *not* flush any data to disk (unless the files need to
+// be resized). Be sure to explicitly flush before irreversibly deleting data
+// from the fast chain database.
+func (t *table) Append(item uint64, blob []byte) error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	// Ensure the table is still accessible
+	if t.rawCounter == nil {
+		if err := t.ensure(); err != nil {
+			t.logger.Error("Failed to re-access table", "err", err)
+			return errTableInaccessible
+		}
+	}
+	// Ensure only the next item can be written, nothing else
+	items := t.rawCounter[0]
+	if items > item {
+		return errAlreadyExists
+	}
+	if items < item {
+		return errGappedWrite
+	}
+	// Encode the blob and write it into the data file
+	blob = snappy.Encode(nil, blob)
+
+	t.rawIndex[items+1] = t.rawIndex[items] + uint64(len(blob))
+	copy(t.rawData[t.rawIndex[items]:], blob)
+	t.rawCounter[0]++
+
+	t.writeMeter.Mark(int64(len(blob)))
+
+	// Ensure we have enough space for future appends
+	return t.ensure()
+}
+
+// Retrieve looks up the data offset of an item with the given index and retrieves
+// the raw binary blob from the data file.
+func (t *table) Retrieve(item uint64) ([]byte, error) {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	// Ensure the table and the item is accessible
+	if t.rawCounter == nil {
+		return nil, errTableInaccessible
+	}
+	if t.rawCounter[0] <= item {
+		return nil, errOutOfBounds
+	}
+	// Item reachable, retrive and return to the user
+	blob := t.rawData[t.rawIndex[item]:t.rawIndex[item+1]]
+	t.writeMeter.Mark(int64(len(blob)))
+
+	return snappy.Decode(nil, blob)
+}
+
+// Flush pushes any pending data from memory out to disk. This is an expensive
+// operation, so use it with care.
+func (t *table) Flush() error {
+	if err := t.mmapData.Flush(); err != nil {
+		return err
+	}
+	if err := t.mmapIndex.Flush(); err != nil {
+		return err
+	}
+	if err := t.mmapCounter.Flush(); err != nil {
+		return err
+	}
+	return nil
+}