Merge branch 'develop' of github.com:tronprotocol/java-tron into fix_msg_process

Author: xxo1shine

chainbase/src/main/java/org/tron/core/db2/core/AbstractSnapshot.java

@@ -15,6 +15,8 @@ public abstract class AbstractSnapshot<K, V> implements Snapshot {
 
   protected WeakReference<Snapshot> next;
 
+  protected boolean isOptimized;
+
   @Override
   public Snapshot advance() {
     return new SnapshotImpl(this);
@@ -34,4 +36,9 @@ public void setNext(Snapshot next) {
   public String getDbName() {
     return db.getDbName();
   }
+
+  @Override
+  public boolean isOptimized(){
+    return isOptimized;
+  }
 }

chainbase/src/main/java/org/tron/core/db2/core/Snapshot.java

@@ -46,4 +46,8 @@ static boolean isImpl(Snapshot snapshot) {
   void updateSolidity();
 
   String getDbName();
+
+  boolean isOptimized();
+
+  void reloadToMem();
 }

chainbase/src/main/java/org/tron/core/db2/core/SnapshotImpl.java

@@ -30,6 +30,10 @@ public class SnapshotImpl extends AbstractSnapshot<Key, Value> {
     }
     previous = snapshot;
     snapshot.setNext(this);
+    isOptimized = snapshot.isOptimized();
+    if (isOptimized &&  root == previous) {
+      Streams.stream(root.iterator()).forEach( e -> put(e.getKey(),e.getValue()));
+    }
   }
 
   @Override
@@ -40,6 +44,7 @@ public byte[] get(byte[] key) {
   private byte[] get(Snapshot head, byte[] key) {
     Snapshot snapshot = head;
     Value value;
+
     while (Snapshot.isImpl(snapshot)) {
       if ((value = ((SnapshotImpl) snapshot).db.get(Key.of(key))) != null) {
         return value.getBytes();
@@ -83,6 +88,19 @@ public void merge(Snapshot from) {
     Streams.stream(fromImpl.db).forEach(e -> db.put(e.getKey(), e.getValue()));
   }
 
+  public void mergeAhead(Snapshot from) {
+    if (from instanceof SnapshotRoot) {
+      return ;
+    }
+    SnapshotImpl fromImpl = (SnapshotImpl) from;
+    Streams.stream(fromImpl.db).forEach(e -> {
+      if (db.get(e.getKey()) == null) {
+        db.put(e.getKey(), e.getValue());
+      }
+    }
+    );
+  }
+
   @Override
   public Snapshot retreat() {
     return previous;
@@ -177,4 +195,9 @@ public String getDbName() {
   public Snapshot newInstance() {
     return new SnapshotImpl(this);
   }
+
+  @Override
+  public void reloadToMem() {
+      mergeAhead(previous);
+  }
 }

chainbase/src/main/java/org/tron/core/db2/core/SnapshotManager.java

@@ -221,6 +221,12 @@ public synchronized void commit() {
     }
 
     --activeSession;
+
+    dbs.forEach(db -> {
+      if (db.getHead().isOptimized()) {
+        db.getHead().reloadToMem();
+      }
+    });
   }
 
   public synchronized void pop() {

chainbase/src/main/java/org/tron/core/db2/core/SnapshotRoot.java

@@ -38,6 +38,7 @@ public SnapshotRoot(DB<byte[], byte[]> db) {
     if (CACHE_DBS.contains(this.db.getDbName())) {
       this.cache = CacheManager.allocate(CacheType.findByType(this.db.getDbName()));
     }
+    isOptimized = "properties".equalsIgnoreCase(db.getDbName());
   }
 
   private boolean needOptAsset() {
@@ -221,4 +222,7 @@ public String getDbName() {
   public Snapshot newInstance() {
     return new SnapshotRoot(db.newInstance());
   }
+
+  @Override
+  public void reloadToMem() { }
 }

common/src/main/java/org/tron/common/logsfilter/trigger/TransactionLogTrigger.java

@@ -1,6 +1,8 @@
 package org.tron.common.logsfilter.trigger;
 
 import java.util.List;
+import java.util.Map;
+
 import lombok.Getter;
 import lombok.Setter;
 
@@ -91,6 +93,10 @@ public class TransactionLogTrigger extends Trigger {
   @Setter
   private long energyUnitPrice;
 
+  @Getter
+  @Setter
+  private Map<String, Long> extMap;
+
   public TransactionLogTrigger() {
     setTriggerName(Trigger.TRANSACTION_TRIGGER_NAME);
   }

framework/src/main/java/org/tron/common/application/Application.java

@@ -34,6 +34,10 @@ public interface Application {
 
   void startServices();
 
+  // DO NOT USE THIS METHOD IN TEST CASES MAIN-THREAD
+  default void blockUntilShutdown() {
+  }
+
   void shutdownServices();
 
   void addService(Service service);

framework/src/main/java/org/tron/common/application/ApplicationImpl.java

@@ -3,11 +3,9 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import org.tron.common.logsfilter.EventPluginLoader;
 import org.tron.common.parameter.CommonParameter;
 import org.tron.core.ChainBaseManager;
 import org.tron.core.config.args.Args;
-import org.tron.core.config.args.DynamicArgs;
 import org.tron.core.consensus.ConsensusService;
 import org.tron.core.db.Manager;
 import org.tron.core.metrics.MetricsUtil;
@@ -31,9 +29,6 @@ public class ApplicationImpl implements Application {
   @Autowired
   private ConsensusService consensusService;
 
-  @Autowired
-  private DynamicArgs dynamicArgs;
-
   @Override
   public void setOptions(Args args) {
     // not used
@@ -64,35 +59,30 @@ public void startup() {
     }
     consensusService.start();
     MetricsUtil.init();
-    dynamicArgs.init();
+    this.initServices(Args.getInstance());
+    this.startServices();
   }
 
   @Override
   public void shutdown() {
-    logger.info("******** start to shutdown ********");
+    this.shutdownServices();
+    consensusService.stop();
     if (!Args.getInstance().isSolidityNode() && (!Args.getInstance().p2pDisable)) {
       tronNetService.close();
     }
-    consensusService.stop();
-    synchronized (dbManager.getRevokingStore()) {
-      dbManager.getSession().reset();
-      closeRevokingStore();
-    }
-    dbManager.stopRePushThread();
-    dbManager.stopRePushTriggerThread();
-    EventPluginLoader.getInstance().stopPlugin();
-    dbManager.stopFilterProcessThread();
-    dbManager.stopValidateSignThread();
-    getChainBaseManager().shutdown();
-    dynamicArgs.close();
-    logger.info("******** end to shutdown ********");
+    dbManager.close();
   }
 
   @Override
   public void startServices() {
     services.start();
   }
 
+  @Override
+  public void blockUntilShutdown() {
+    services.blockUntilShutdown();
+  }
+
   @Override
   public void shutdownServices() {
     services.stop();
@@ -108,9 +98,4 @@ public ChainBaseManager getChainBaseManager() {
     return chainBaseManager;
   }
 
-  private void closeRevokingStore() {
-    logger.info("******** start to closeRevokingStore ********");
-    dbManager.getRevokingStore().shutdown();
-  }
-
 }

framework/src/main/java/org/tron/common/application/HttpService.java

@@ -0,0 +1,81 @@
+/*
+ * java-tron is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * java-tron 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package org.tron.common.application;
+
+import com.google.common.base.Objects;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.jetty.server.Server;
+
+@Slf4j(topic = "rpc")
+public abstract class HttpService implements Service {
+
+  protected Server apiServer;
+  protected int port;
+
+  @Override
+  public void blockUntilShutdown() {
+    if (apiServer != null) {
+      try {
+        apiServer.join();
+      } catch (InterruptedException e) {
+        logger.warn("{}", e.getMessage());
+        Thread.currentThread().interrupt();
+      }
+    }
+  }
+
+  @Override
+  public void start() {
+    if (apiServer != null) {
+      try {
+        apiServer.start();
+        logger.info("{} started, listening on {}", this.getClass().getSimpleName(), port);
+      } catch (Exception e) {
+        logger.error("{}", this.getClass().getSimpleName(), e);
+      }
+    }
+  }
+
+  @Override
+  public void stop() {
+    if (apiServer != null) {
+      logger.info("{} shutdown...", this.getClass().getSimpleName());
+      try {
+        apiServer.stop();
+      } catch (Exception e) {
+        logger.warn("{}", this.getClass().getSimpleName(), e);
+      }
+      logger.info("{} shutdown complete", this.getClass().getSimpleName());
+    }
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    HttpService that = (HttpService) o;
+    return port == that.port;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(getClass().getSimpleName(), port);
+  }
+}

framework/src/main/java/org/tron/common/application/RpcService.java

@@ -0,0 +1,85 @@
+/*
+ * java-tron is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * java-tron 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package org.tron.common.application;
+
+import com.google.common.base.Objects;
+import io.grpc.Server;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j(topic = "rpc")
+public abstract class RpcService implements Service {
+
+  protected Server apiServer;
+  protected int port;
+
+  @Override
+  public void blockUntilShutdown() {
+    if (apiServer != null) {
+      try {
+        apiServer.awaitTermination();
+      } catch (InterruptedException e) {
+        logger.warn("{}", e.getMessage());
+        Thread.currentThread().interrupt();
+      }
+    }
+  }
+
+  @Override
+  public void start() {
+    if (apiServer != null) {
+      try {
+        apiServer.start();
+        logger.info("{} started, listening on {}", this.getClass().getSimpleName(), port);
+      } catch (IOException e) {
+        logger.error("{}", this.getClass().getSimpleName(), e);
+      }
+    }
+  }
+
+  @Override
+  public void stop() {
+    if (apiServer != null) {
+      logger.info("{} shutdown...", this.getClass().getSimpleName());
+      try {
+        apiServer.shutdown().awaitTermination(5, TimeUnit.SECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        logger.warn("{}", this.getClass().getSimpleName(), e);
+      }
+      logger.info("{} shutdown complete", this.getClass().getSimpleName());
+    }
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    RpcService that = (RpcService) o;
+    return port == that.port;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(getClass().getSimpleName(), port);
+  }
+
+}