try to fix bug in dmap merger

dmap.go

@@ -73,7 +73,7 @@ func (db *Olric) createDMap(part *partition, name string) (*dmap, error) {
 	if !ok {
 		return nil, fmt.Errorf("storage engine could not be found: %s", nm.config.storageEngine)
 	}
-	nm.storage, err = engine.Fork()
+	nm.storage, err = engine.Fork(nil)
 	if err != nil {
 		return nil, err
 	}

dmap_test.go

@@ -121,33 +121,25 @@ func newDB(c *config.Config, peers ...*Olric) (*Olric, error) {
 		return nil, err
 	}
 
+	ctx, cancel := context.WithCancel(context.Background())
+	c.Started = func() {
+		defer cancel()
+	}
+
 	db, err := New(c)
 	if err != nil {
 		return nil, err
 	}
 
-	db.wg.Add(1)
-	go db.callStartedCallback()
-
 	db.wg.Add(1)
 	go func() {
 		defer db.wg.Done()
-		err = db.server.ListenAndServe()
-		if err != nil {
-			db.log.V(2).Printf("[ERROR] Failed to run TCP server")
+		serr := db.Start()
+		if serr != nil {
+			db.log.V(2).Printf("[ERROR] Failed to start Olric node: %s", serr)
 		}
 	}()
-	<-db.server.StartedCtx.Done()
-	db.passCheckpoint()
-
-	err = db.startDiscovery()
-	if err != nil {
-		return nil, err
-	}
-	// Wait some time for goroutines
-	<-time.After(100 * time.Millisecond)
-	db.passCheckpoint()
-
+	<-ctx.Done()
 	return db, nil
 }
 

internal/kvstore/kvstore.go

@@ -16,6 +16,7 @@
 package kvstore
 
 import (
+	"errors"
 	"regexp"
 
 	"github.com/buraksezer/olric/pkg/storage"
@@ -48,16 +49,24 @@ func (kv *KVStore) SetConfig(c *storage.Config) {
 	kv.config = c
 }
 
-func (kv *KVStore) Start() error { return nil }
+func (kv *KVStore) Start() error {
+	if kv.config == nil {
+		return errors.New("config cannot be nil")
+	}
+	return nil
+}
 
 // Fork creates a new KVStore instance.
-func (kv *KVStore) Fork() (storage.Engine, error) {
-	size, err := kv.config.Get("tableSize")
+func (kv *KVStore) Fork(c *storage.Config) (storage.Engine, error) {
+	if c == nil {
+		c = kv.config.Copy()
+	}
+	size, err := c.Get("tableSize")
 	if err != nil {
 		return nil, err
 	}
 	child := &KVStore{
-		config: kv.config,
+		config: c,
 	}
 	t := newTable(size.(int))
 	child.tables = append(kv.tables, t)
@@ -305,19 +314,20 @@ func (kv *KVStore) Import(data []byte) (storage.Engine, error) {
 		return nil, err
 	}
 
-	options := kv.config.Copy()
-	options.Add("tableSize", tr.Allocated)
-	fresh, err := kv.Fork()
+	c := kv.config.Copy()
+	c.Add("tableSize", tr.Allocated)
+
+	child, err := kv.Fork(c)
 	if err != nil {
 		return nil, err
 	}
-	t := fresh.(*KVStore).tables[0]
+	t := child.(*KVStore).tables[0]
 	t.hkeys = tr.HKeys
 	t.offset = tr.Offset
 	t.inuse = tr.Inuse
 	t.garbage = tr.Garbage
 	copy(t.memory, tr.Memory)
-	return fresh, nil
+	return child, nil
 }
 
 // Stats is a function which provides memory allocation and garbage ratio of a storage instance.

internal/kvstore/kvstore_test.go

@@ -39,7 +39,7 @@ func bval(i int) []byte {
 func testKVStore() (storage.Engine, error) {
 	kv := &KVStore{}
 	kv.SetConfig(DefaultConfig())
-	return kv.Fork()
+	return kv.Fork(nil)
 }
 
 func Test_Put(t *testing.T) {

pkg/storage/config.go

@@ -62,7 +62,7 @@ func (c *Config) Copy() *Config {
 		m: make(map[string]interface{}),
 	}
 	for key, value := range c.m {
-		c.m[key] = value
+		n.m[key] = value
 	}
 	return n
 }

pkg/storage/engine.go

@@ -19,7 +19,7 @@ type Engine interface {
 	Start() error
 	NewEntry() Entry
 	Name() string
-	Fork() (Engine, error)
+	Fork(*Config) (Engine, error)
 	PutRaw(uint64, []byte) error
 	Put(uint64, Entry) error
 	GetRaw(uint64) ([]byte, error)

rebalancer.go

@@ -129,6 +129,7 @@ func (db *Olric) mergeDMaps(part *partition, data *dmapbox) error {
 			mergeErr = err
 			return false
 		}
+		// TODO: Don't put the winner again if it comes from dm.storage
 		mergeErr = dm.storage.Put(hkey, winner)
 		if mergeErr == storage.ErrFragmented {
 			db.wg.Add(1)