Reduce the memory consumption in:

1. the read/write path of base-kv
2. inbox service helper's frequently-used method

Author: popduke

base-kv/base-kv-local-engine/src/main/java/com/baidu/bifromq/basekv/localengine/AbstractKVEngine.java

@@ -13,33 +13,43 @@
 
 package com.baidu.bifromq.basekv.localengine;
 
+import static com.google.common.collect.Lists.newArrayList;
+
+import com.baidu.bifromq.basekv.localengine.metrics.KVSpaceOpMeters;
 import com.baidu.bifromq.logger.SiftLogger;
+import com.google.common.collect.Iterables;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tags;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
 
-public abstract class AbstractKVEngine<T extends IKVSpace> implements IKVEngine<T> {
-    protected enum State {
-        INIT, STARTING, STARTED, FATAL_FAILURE, STOPPING, STOPPED
-    }
-
+/**
+ * The abstract class of KVEngine.
+ *
+ * @param <T> the type of KV space created by the engine
+ * @param <C> the type of configurator
+ */
+public abstract class AbstractKVEngine<T extends IKVSpace, C extends IKVEngineConfigurator> implements IKVEngine<T> {
     protected final String overrideIdentity;
+    protected final C configurator;
+    private final AtomicReference<State> state = new AtomicReference<>(State.INIT);
+    private final Map<String, T> kvSpaceMap = new ConcurrentHashMap<>();
     protected Logger log;
     protected String[] metricTags;
-    private final AtomicReference<State> state = new AtomicReference<>(State.INIT);
     private Gauge gauge;
 
-    public AbstractKVEngine(String overrideIdentity) {
+    public AbstractKVEngine(String overrideIdentity, C configurator) {
         this.overrideIdentity = overrideIdentity;
-    }
-
-    protected State state() {
-        return state.get();
+        this.configurator = configurator;
     }
 
     @Override
-    public void start(String... tags) {
+    public final void start(String... tags) {
         if (state.compareAndSet(State.INIT, State.STARTING)) {
             try {
                 log = SiftLogger.getLogger(this.getClass(), tags);
@@ -64,7 +74,7 @@ protected void afterStart() {
     }
 
     @Override
-    public void stop() {
+    public final void stop() {
         assertStarted();
         if (state.compareAndSet(State.STARTED, State.STOPPING)) {
             try {
@@ -76,9 +86,54 @@ public void stop() {
         }
     }
 
-    protected abstract void doStop();
+    protected void doStop() {
+        kvSpaceMap.values().forEach(IKVSpace::close);
+    }
 
     protected void assertStarted() {
         assert state.get() == State.STARTED : "Not started";
     }
+
+    @Override
+    public final Map<String, T> spaces() {
+        assertStarted();
+        return Collections.unmodifiableMap(kvSpaceMap);
+    }
+
+    @Override
+    public final T createIfMissing(String spaceId) {
+        assertStarted();
+        return kvSpaceMap.computeIfAbsent(spaceId,
+            k -> {
+                T space = buildKVSpace(spaceId, configurator, () -> kvSpaceMap.remove(spaceId), metricTags);
+                space.open();
+                return space;
+            });
+    }
+
+    protected final void load(String spaceId) {
+        T space = buildKVSpace(spaceId, configurator, () -> kvSpaceMap.remove(spaceId), metricTags);
+        space.open();
+        T prev = kvSpaceMap.put(spaceId, space);
+        assert prev == null;
+    }
+
+    private T buildKVSpace(String spaceId, C configurator, Runnable onDestroy, String... tags) {
+        String[] tagList =
+            newArrayList(Iterables.concat(List.of(tags), List.of("spaceId", spaceId))).toArray(String[]::new);
+        KVSpaceOpMeters opMeters = new KVSpaceOpMeters(spaceId, Tags.of(tagList));
+        Logger logger = SiftLogger.getLogger("space.logger", tagList);
+        return doBuildKVSpace(spaceId, configurator, onDestroy, opMeters, logger, tagList);
+    }
+
+    protected abstract T doBuildKVSpace(String spaceId,
+                                        C configurator,
+                                        Runnable onDestroy,
+                                        KVSpaceOpMeters opMeters,
+                                        Logger logger,
+                                        String... tags);
+
+    private enum State {
+        INIT, STARTING, STARTED, FATAL_FAILURE, STOPPING, STOPPED
+    }
 }

base-kv/base-kv-local-engine/src/main/java/com/baidu/bifromq/basekv/localengine/AbstractKVSpaceReader.java

@@ -13,34 +13,21 @@
 
 package com.baidu.bifromq.basekv.localengine;
 
-import static com.baidu.bifromq.basekv.localengine.metrics.KVSpaceMeters.getTimer;
-
-import com.baidu.bifromq.basekv.localengine.metrics.KVSpaceMetric;
+import com.baidu.bifromq.basekv.localengine.metrics.KVSpaceOpMeters;
 import com.baidu.bifromq.basekv.proto.Boundary;
-import com.baidu.bifromq.logger.SiftLogger;
 import com.google.protobuf.ByteString;
-import io.micrometer.core.instrument.Tags;
-import io.micrometer.core.instrument.Timer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
 import java.util.Optional;
 import org.slf4j.Logger;
 
 public abstract class AbstractKVSpaceReader implements IKVSpaceReader {
-    protected final Logger log;
     protected final String id;
-    protected final String[] metricTags;
-    private final MetricManager metricMgr;
+    protected final KVSpaceOpMeters opMeters;
+    protected final Logger logger;
 
-    protected AbstractKVSpaceReader(String id, String... tags) {
+    protected AbstractKVSpaceReader(String id, KVSpaceOpMeters opMeters, Logger logger) {
         this.id = id;
-        List<String> allTags = new ArrayList<>(Arrays.asList(tags));
-        allTags.add("spaceId");
-        allTags.add(id);
-        this.metricTags = allTags.toArray(String[]::new);
-        this.metricMgr = new MetricManager(Tags.of(metricTags));
-        this.log = SiftLogger.getLogger(getClass(), allTags.toArray(String[]::new));
+        this.opMeters = opMeters;
+        this.logger = logger;
     }
 
     @Override
@@ -50,7 +37,7 @@ public final String id() {
 
     @Override
     public final Optional<ByteString> metadata(ByteString metaKey) {
-        return metricMgr.metadataCallTimer.record(() -> doMetadata(metaKey));
+        return opMeters.metadataCallTimer.record(() -> doMetadata(metaKey));
     }
 
     protected abstract Optional<ByteString> doMetadata(ByteString metaKey);
@@ -62,35 +49,35 @@ public final long size() {
 
     @Override
     public final long size(Boundary boundary) {
-        return metricMgr.sizeCallTimer.record(() -> doSize(boundary));
+        return opMeters.sizeCallTimer.record(() -> doSize(boundary));
     }
 
     protected abstract long doSize(Boundary boundary);
 
     @Override
     public final boolean exist(ByteString key) {
-        return metricMgr.existCallTimer.record(() -> doExist(key));
+        return opMeters.existCallTimer.record(() -> doExist(key));
     }
 
     protected abstract boolean doExist(ByteString key);
 
     @Override
     public final Optional<ByteString> get(ByteString key) {
-        return metricMgr.getCallTimer.record(() -> doGet(key));
+        return opMeters.getCallTimer.record(() -> doGet(key));
     }
 
     protected abstract Optional<ByteString> doGet(ByteString key);
 
     @Override
     public final IKVSpaceIterator newIterator() {
-        return metricMgr.iterNewCallTimer.record(() -> new MonitoredKeyRangeIterator(doNewIterator()));
+        return opMeters.iterNewCallTimer.record(() -> new MonitoredKeyRangeIterator(doNewIterator()));
     }
 
     protected abstract IKVSpaceIterator doNewIterator();
 
     @Override
     public final IKVSpaceIterator newIterator(Boundary subBoundary) {
-        return metricMgr.iterNewCallTimer.record(() -> new MonitoredKeyRangeIterator(doNewIterator(subBoundary)));
+        return opMeters.iterNewCallTimer.record(() -> new MonitoredKeyRangeIterator(doNewIterator(subBoundary)));
     }
 
     protected abstract IKVSpaceIterator doNewIterator(Boundary subBoundary);
@@ -119,74 +106,42 @@ public boolean isValid() {
 
         @Override
         public void next() {
-            metricMgr.iterNextCallTimer.record(delegate::next);
+            opMeters.iterNextCallTimer.record(delegate::next);
         }
 
         @Override
         public void prev() {
-            metricMgr.iterPrevCallTimer.record(delegate::prev);
+            opMeters.iterPrevCallTimer.record(delegate::prev);
         }
 
         @Override
         public void seekToFirst() {
-            metricMgr.iterSeekToFirstCallTimer.record(delegate::seekToFirst);
+            opMeters.iterSeekToFirstCallTimer.record(delegate::seekToFirst);
         }
 
         @Override
         public void seekToLast() {
-            metricMgr.iterSeekToLastCallTimer.record(delegate::seekToLast);
+            opMeters.iterSeekToLastCallTimer.record(delegate::seekToLast);
         }
 
         @Override
         public void seek(ByteString target) {
-            metricMgr.iterSeekCallTimer.record(() -> delegate.seek(target));
+            opMeters.iterSeekCallTimer.record(() -> delegate.seek(target));
         }
 
         @Override
         public void seekForPrev(ByteString target) {
-            metricMgr.iterSeekForPrevCallTimer.record(() -> delegate.seekForPrev(target));
+            opMeters.iterSeekForPrevCallTimer.record(() -> delegate.seekForPrev(target));
         }
 
         @Override
         public void refresh() {
-            metricMgr.iterRefreshTimer.record(delegate::refresh);
+            opMeters.iterRefreshTimer.record(delegate::refresh);
         }
 
         @Override
         public void close() {
             delegate.close();
         }
     }
-
-    private class MetricManager {
-        private final Timer metadataCallTimer;
-        private final Timer sizeCallTimer;
-        private final Timer existCallTimer;
-        private final Timer getCallTimer;
-        private final Timer iterNewCallTimer;
-        private final Timer iterSeekCallTimer;
-        private final Timer iterSeekForPrevCallTimer;
-        private final Timer iterSeekToFirstCallTimer;
-        private final Timer iterSeekToLastCallTimer;
-        private final Timer iterNextCallTimer;
-        private final Timer iterPrevCallTimer;
-        private final Timer iterRefreshTimer;
-
-        MetricManager(Tags tags) {
-            metadataCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "metadata"));
-
-            sizeCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "size"));
-            existCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "exist"));
-            getCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "get"));
-            iterNewCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "newitr"));
-
-            iterSeekCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "seek"));
-            iterSeekForPrevCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "pseek"));
-            iterSeekToFirstCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "fseek"));
-            iterSeekToLastCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "lseek"));
-            iterNextCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "next"));
-            iterPrevCallTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "prev"));
-            iterRefreshTimer = getTimer(id, KVSpaceMetric.CallTimer, tags.and("op", "refresh"));
-        }
-    }
 }

base-kv/base-kv-local-engine/src/main/java/com/baidu/bifromq/basekv/localengine/ICPableKVSpace.java

@@ -15,9 +15,12 @@
 
 import java.util.Optional;
 
+/**
+ * KV space supports checkpoint.
+ */
 public interface ICPableKVSpace extends IKVSpace {
     /**
-     * Make a checkpoint of the current range state
+     * Make a checkpoint of the current range state.
      *
      * @return global unique id of the checkpoint
      */
@@ -31,5 +34,5 @@ public interface ICPableKVSpace extends IKVSpace {
      * @param checkpointId the checkpoint id
      * @return the range object for accessing the checkpoint
      */
-    Optional<IKVSpaceCheckpoint> open(String checkpointId);
+    Optional<IKVSpaceCheckpoint> openCheckpoint(String checkpointId);
 }

base-kv/base-kv-local-engine/src/main/java/com/baidu/bifromq/basekv/localengine/IKVEngine.java

@@ -15,41 +15,46 @@
 
 import java.util.Map;
 
+/**
+ * The interface of kv engine.
+ *
+ * @param <T> the type of kv space created by the engine
+ */
 public interface IKVEngine<T extends IKVSpace> {
 
     String DEFAULT_NS = "default";
 
     /**
-     * The unique identifier of the engine
+     * The unique identifier of the engine.
      *
      * @return id
      */
     String id();
 
     /**
-     * Find all currently available kv spaces
+     * Find all currently available kv spaces.
      *
      * @return the kv space list
      */
     Map<String, T> spaces();
 
     /**
-     * Create a new key range with specified spaceId and boundary or get existing key range
+     * Create a new key range with specified spaceId and boundary or get existing key range.
      *
      * @param spaceId the space id
      * @return the key range created
      */
     T createIfMissing(String spaceId);
 
     /**
-     * Start the kv engine and specifying additional tags for generated metrics
+     * Start the kv engine and specifying additional tags for generated metrics.
      *
      * @param metricTags the additional metric tags
      */
     void start(String... metricTags);
 
     /**
-     * Stop the engine
+     * Stop the engine.
      */
     void stop();
 }