storage: introduce PutChangeSet and use it for Persist

We're using batches in wrong way during persist, we already have all changes
accumulated in two maps and then we move them to batch and then this is
applied. For some DBs like BoltDB this batch is just another MemoryStore, so
we essentially just shuffle the changeset from one map to another, for others
like LevelDB batch is just a serialized set of KV pairs, it doesn't help much
on subsequent PutBatch, we just duplicate the changeset again.

So introduce PutChangeSet that allows to take two maps with sets and deletes
directly. It also allows to simplify MemCachedStore logic.

neo-bench for single node with 10 workers, LevelDB:

  Reference:

  RPS    30189.132 30556.448 30390.482 ≈ 30379    ±  0.61%
  TPS    29427.344 29418.687 29434.273 ≈ 29427    ±  0.03%
  CPU %     33.304    27.179    33.860 ≈    31.45 ± 11.79%
  Mem MB   800.677   798.389   715.042 ≈   771    ±  6.33%

  Patched:

  RPS    30264.326 30386.364 30166.231 ≈ 30272    ± 0.36% ⇅
  TPS    29444.673 29407.440 29452.478 ≈ 29435    ± 0.08% ⇅
  CPU %     34.012    32.597    33.467 ≈   33.36  ± 2.14% ⇅
  Mem MB   549.126   523.656   517.684 ≈  530     ± 3.15% ↓ 31.26%

BoltDB:

  Reference:

  RPS    31937.647 31551.684 31850.408 ≈ 31780    ±  0.64%
  TPS    31292.049 30368.368 31307.724 ≈ 30989    ±  1.74%
  CPU %     33.792    22.339    35.887 ≈    30.67 ± 23.78%
  Mem MB  1271.687  1254.472  1215.639 ≈  1247    ±  2.30%

  Patched:

  RPS    31746.818 30859.485 31689.761 ≈ 31432    ± 1.58% ⇅
  TPS    31271.499 30340.726 30342.568 ≈ 30652    ± 1.75% ⇅
  CPU %     34.611    34.414    31.553 ≈    33.53 ± 5.11% ⇅
  Mem MB  1262.960  1231.389  1335.569 ≈  1277    ± 4.18% ⇅

pkg/core/storage/badgerdb_store.go

@@ -102,6 +102,25 @@ func (b *BadgerDBStore) PutBatch(batch Batch) error {
 	return batch.(*BadgerDBBatch).batch.Flush()
 }
 
+// PutChangeSet implements the Store interface.
+func (b *BadgerDBStore) PutChangeSet(puts map[string][]byte, dels map[string]bool) error {
+	return b.db.Update(func(txn *badger.Txn) error {
+		for k, v := range puts {
+			err := txn.Set([]byte(k), v)
+			if err != nil {
+				return err
+			}
+		}
+		for k := range dels {
+			err := txn.Delete([]byte(k))
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 // Seek implements the Store interface.
 func (b *BadgerDBStore) Seek(key []byte, f func(k, v []byte)) {
 	err := b.db.View(func(txn *badger.Txn) error {

pkg/core/storage/boltdb_store.go

@@ -84,15 +84,21 @@ func (s *BoltDBStore) Delete(key []byte) error {
 
 // PutBatch implements the Store interface.
 func (s *BoltDBStore) PutBatch(batch Batch) error {
+	memBatch := batch.(*MemoryBatch)
+	return s.PutChangeSet(memBatch.mem, memBatch.del)
+}
+
+// PutChangeSet implements the Store interface.
+func (s *BoltDBStore) PutChangeSet(puts map[string][]byte, dels map[string]bool) error {
 	return s.db.Batch(func(tx *bbolt.Tx) error {
 		b := tx.Bucket(Bucket)
-		for k, v := range batch.(*MemoryBatch).mem {
+		for k, v := range puts {
 			err := b.Put([]byte(k), v)
 			if err != nil {
 				return err
 			}
 		}
-		for k := range batch.(*MemoryBatch).del {
+		for k := range dels {
 			err := b.Delete([]byte(k))
 			if err != nil {
 				return err

pkg/core/storage/leveldb_store.go

@@ -61,6 +61,29 @@ func (s *LevelDBStore) PutBatch(batch Batch) error {
 	return s.db.Write(lvldbBatch, nil)
 }
 
+// PutChangeSet implements the Store interface.
+func (s *LevelDBStore) PutChangeSet(puts map[string][]byte, dels map[string]bool) error {
+	tx, err := s.db.OpenTransaction()
+	if err != nil {
+		return err
+	}
+	for k := range puts {
+		err = tx.Put([]byte(k), puts[k], nil)
+		if err != nil {
+			tx.Discard()
+			return err
+		}
+	}
+	for k := range dels {
+		err = tx.Delete([]byte(k), nil)
+		if err != nil {
+			tx.Discard()
+			return err
+		}
+	}
+	return tx.Commit()
+}
+
 // Seek implements the Store interface.
 func (s *LevelDBStore) Seek(key []byte, f func(k, v []byte)) {
 	iter := s.db.NewIterator(util.BytesPrefix(key), nil)

pkg/core/storage/memcached_store.go

@@ -121,32 +121,8 @@ func (s *MemCachedStore) Persist() (int, error) {
 	s.del = make(map[string]bool)
 	s.mut.Unlock()
 
-	memStore, ok := tempstore.ps.(*MemoryStore)
-	if !ok {
-		memCachedStore, ok := tempstore.ps.(*MemCachedStore)
-		if ok {
-			memStore = &memCachedStore.MemoryStore
-		}
-	}
-	if memStore != nil {
-		memStore.mut.Lock()
-		for k := range tempstore.mem {
-			memStore.put(k, tempstore.mem[k])
-		}
-		for k := range tempstore.del {
-			memStore.drop(k)
-		}
-		memStore.mut.Unlock()
-	} else {
-		batch := tempstore.ps.Batch()
-		for k := range tempstore.mem {
-			batch.Put([]byte(k), tempstore.mem[k])
-		}
-		for k := range tempstore.del {
-			batch.Delete([]byte(k))
-		}
-		err = tempstore.ps.PutBatch(batch)
-	}
+	err = tempstore.ps.PutChangeSet(tempstore.mem, tempstore.del)
+
 	s.mut.Lock()
 	if err == nil {
 		// tempstore.mem and tempstore.del are completely flushed now

pkg/core/storage/memcached_store_test.go

@@ -200,6 +200,9 @@ func (b *BadStore) Put(k, v []byte) error {
 	return nil
 }
 func (b *BadStore) PutBatch(Batch) error {
+	return nil
+}
+func (b *BadStore) PutChangeSet(_ map[string][]byte, _ map[string]bool) error {
 	b.onPutBatch()
 	return ErrKeyNotFound
 }

pkg/core/storage/memory_store.go

@@ -85,14 +85,19 @@ func (s *MemoryStore) Delete(key []byte) error {
 // PutBatch implements the Store interface. Never returns an error.
 func (s *MemoryStore) PutBatch(batch Batch) error {
 	b := batch.(*MemoryBatch)
+	return s.PutChangeSet(b.mem, b.del)
+}
+
+// PutChangeSet implements the Store interface. Never returns an error.
+func (s *MemoryStore) PutChangeSet(puts map[string][]byte, dels map[string]bool) error {
 	s.mut.Lock()
-	defer s.mut.Unlock()
-	for k := range b.del {
+	for k := range puts {
+		s.put(k, puts[k])
+	}
+	for k := range dels {
 		s.drop(k)
 	}
-	for k, v := range b.mem {
-		s.put(k, v)
-	}
+	s.mut.Unlock()
 	return nil
 }
 

pkg/core/storage/redis_store.go

@@ -62,11 +62,17 @@ func (s *RedisStore) Put(k, v []byte) error {
 
 // PutBatch implements the Store interface.
 func (s *RedisStore) PutBatch(b Batch) error {
+	memBatch := b.(*MemoryBatch)
+	return s.PutChangeSet(memBatch.mem, memBatch.del)
+}
+
+// PutChangeSet implements the Store interface.
+func (s *RedisStore) PutChangeSet(puts map[string][]byte, dels map[string]bool) error {
 	pipe := s.client.Pipeline()
-	for k, v := range b.(*MemoryBatch).mem {
+	for k, v := range puts {
 		pipe.Set(k, v, 0)
 	}
-	for k := range b.(*MemoryBatch).del {
+	for k := range dels {
 		pipe.Del(k)
 	}
 	_, err := pipe.Exec()

pkg/core/storage/store.go

@@ -44,6 +44,8 @@ type (
 		Get([]byte) ([]byte, error)
 		Put(k, v []byte) error
 		PutBatch(Batch) error
+		// PutChangeSet allows to push prepared changeset to the Store.
+		PutChangeSet(puts map[string][]byte, dels map[string]bool) error
 		// Seek can guarantee that provided key (k) and value (v) are the only valid until the next call to f.
 		// Key and value slices should not be modified.
 		Seek(k []byte, f func(k, v []byte))