Merge pull request #6259 from tronprotocol/release_v4.8.0

merge Release v4.8.0 to Nile

Author: CodeNinjaEvan

common/build.gradle

@@ -38,8 +38,6 @@ dependencies {
     api group: 'com.typesafe', name: 'config', version: '1.3.2'
     api group: leveldbGroup, name: leveldbName, version: leveldbVersion
     api group: 'org.rocksdb', name: 'rocksdbjni', version: '5.15.10'
-    // https://mvnrepository.com/artifact/org.quartz-scheduler/quartz
-    api group: 'org.quartz-scheduler', name: 'quartz', version: '2.3.2'
     api group: 'io.prometheus', name: 'simpleclient', version: '0.15.0'
     api group: 'io.prometheus', name: 'simpleclient_httpserver', version: '0.15.0'
     api group: 'io.prometheus', name: 'simpleclient_hotspot', version: '0.15.0'

common/src/main/java/org/tron/common/cron/CronExpression.java

@@ -0,0 +1,7 @@
+package org.tron.common.cron;
+
+class ValueSet {
+  public int value;
+
+  public int pos;
+}

common/src/main/java/org/tron/common/cron/ValueSet.java

@@ -8,7 +8,7 @@
 import java.util.Set;
 import lombok.Getter;
 import lombok.Setter;
-import org.quartz.CronExpression;
+import org.tron.common.cron.CronExpression;
 import org.tron.common.args.GenesisBlock;
 import org.tron.common.config.DbBackupConfig;
 import org.tron.common.logsfilter.EventPluginConfig;

common/src/main/java/org/tron/common/parameter/CommonParameter.java

@@ -9,6 +9,8 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.bouncycastle.util.encoders.Hex;
 import org.pf4j.CompoundPluginDescriptorFinder;
@@ -31,6 +33,8 @@ public class EventPluginLoader {
 
   private static EventPluginLoader instance;
 
+  private long MAX_PENDING_SIZE = 50000;
+
   private PluginManager pluginManager = null;
 
   private List<IPluginEventListener> eventListeners;
@@ -73,6 +77,7 @@ public class EventPluginLoader {
 
   private FilterQuery filterQuery;
 
+  @Getter
   private boolean useNativeQueue = false;
 
   public static EventPluginLoader getInstance() {
@@ -537,6 +542,21 @@ public void postContractEventTrigger(ContractEventTrigger trigger) {
     }
   }
 
+  public boolean isBusy() {
+    if (useNativeQueue) {
+      return false;
+    }
+    int queueSize = 0;
+    for (IPluginEventListener listener : eventListeners) {
+      try {
+        queueSize += listener.getPendingSize();
+      } catch (AbstractMethodError error) {
+        break;
+      }
+    }
+    return queueSize >= MAX_PENDING_SIZE;
+  }
+
   private String toJsonString(Object data) {
     String jsonData = "";
 

framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java

@@ -13,6 +13,8 @@ public interface IPluginEventListener extends ExtensionPoint {
   // start should be called after setServerAddress, setTopic, setDBConfig
   void start();
 
+  int getPendingSize();
+
   void handleBlockEvent(Object trigger);
 
   void handleTransactionTrigger(Object trigger);

framework/src/main/java/org/tron/common/logsfilter/IPluginEventListener.java

@@ -42,7 +42,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.quartz.CronExpression;
+import org.tron.common.cron.CronExpression;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.args.Account;

framework/src/main/java/org/tron/core/config/args/Args.java

@@ -46,7 +46,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.bouncycastle.util.encoders.Hex;
-import org.quartz.CronExpression;
+import org.tron.common.cron.CronExpression;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.api.GrpcAPI.TransactionInfoList;

framework/src/main/java/org/tron/core/db/Manager.java

@@ -9,6 +9,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.tron.common.es.ExecutorServiceManager;
+import org.tron.common.logsfilter.EventPluginLoader;
 import org.tron.core.db.Manager;
 import org.tron.core.services.event.bo.BlockEvent;
 import org.tron.core.services.event.bo.Event;
@@ -26,13 +27,19 @@ public class BlockEventLoad {
   @Autowired
   private BlockEventGet blockEventGet;
 
+  private EventPluginLoader instance = EventPluginLoader.getInstance();
+
   private final ScheduledExecutorService executor = ExecutorServiceManager
       .newSingleThreadScheduledExecutor("event-load");
 
+  private long MAX_LOAD_NUM = 100;
+
   public void init() {
     executor.scheduleWithFixedDelay(() -> {
       try {
-        load();
+        if (!instance.isBusy()) {
+          load();
+        }
       } catch (Exception e) {
         close();
         logger.error("Event load service fail.", e);
@@ -62,6 +69,9 @@ public synchronized void load() throws Exception {
       if (cacheHeadNum >= tmpNum) {
         return;
       }
+      if (tmpNum > cacheHeadNum + MAX_LOAD_NUM) {
+        tmpNum = cacheHeadNum + MAX_LOAD_NUM;
+      }
       List<BlockEvent> l1 = new ArrayList<>();
       List<BlockEvent> l2 = new ArrayList<>();
       BlockEvent tmp = BlockEventCache.getHead();

framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java

@@ -55,12 +55,17 @@ private void syncEvent() {
       long tmp = instance.getStartSyncBlockNum();
       long endNum = manager.getDynamicPropertiesStore().getLatestSolidifiedBlockNum();
       while (tmp <= endNum) {
+        if (instance.isUseNativeQueue()) {
+          Thread.sleep(20);
+        } else if (instance.isBusy()) {
+          Thread.sleep(100);
+          continue;
+        }
         BlockEvent blockEvent = blockEventGet.getBlockEvent(tmp);
         realtimeEventService.flush(blockEvent, false);
         solidEventService.flush(blockEvent);
         tmp++;
         endNum = manager.getDynamicPropertiesStore().getLatestSolidifiedBlockNum();
-        Thread.sleep(30);
       }
       initEventService(manager.getChainBaseManager().getBlockIdByNum(endNum));
     } catch (InterruptedException e1) {

framework/src/main/java/org/tron/core/services/event/HistoryEventService.java

@@ -8,14 +8,14 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
-import com.mchange.v2.collection.MapEntry;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
+import lombok.Getter;
 import org.junit.Before;
 import org.junit.Test;
-import org.tron.core.db.ByteArrayWrapper;
 
 public class ByteArrayMapTest {
 
@@ -142,4 +142,32 @@ public void test() {
     assertTrue(testMap.hashCode() <= 0);
     assertNotNull(testMap.toString());
   }
+
+
+  @Getter
+  static class MapEntry<K, V> implements Map.Entry<K, V> {
+    K key;
+    V value;
+
+    public MapEntry(K key, V value) {
+      this.key = key;
+      this.value = value;
+    }
+
+    public V setValue(V o) {
+      throw new UnsupportedOperationException();
+    }
+
+    public boolean equals(Object o) {
+      if (o instanceof Map.Entry) {
+        Map.Entry<K, V> other = (Map.Entry<K, V>) o;
+        return Objects.equals(key, other.getKey()) && Objects.equals(value, other.getValue());
+      }
+      return false;
+    }
+
+    public int hashCode() {
+      return Objects.hashCode(key) ^ Objects.hashCode(value);
+    }
+  }
 }

framework/src/test/java/org/tron/common/utils/ByteArrayMapTest.java

@@ -31,7 +31,7 @@
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
-import org.quartz.CronExpression;
+import org.tron.common.cron.CronExpression;
 import org.tron.common.parameter.CommonParameter;
 import org.tron.common.runtime.ProgramResult;
 import org.tron.common.utils.Sha256Hash;

framework/src/test/java/org/tron/core/db/ManagerMockTest.java

@@ -0,0 +1,43 @@
+package org.tron.core.event;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.tron.common.logsfilter.EventPluginLoader;
+import org.tron.common.logsfilter.IPluginEventListener;
+import org.tron.common.utils.ReflectUtils;
+
+public class EventPluginLoaderTest {
+
+  @Test
+  public void testIsBusy() {
+
+    EventPluginLoader eventPluginLoader = EventPluginLoader.getInstance();
+    ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", true);
+    boolean flag = eventPluginLoader.isBusy();
+    Assert.assertFalse(flag);
+
+    ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", false);
+
+    IPluginEventListener p1 = mock(IPluginEventListener.class);
+    List<IPluginEventListener> list = new ArrayList<>();
+    list.add(p1);
+    ReflectUtils.setFieldValue(eventPluginLoader, "eventListeners", list);
+
+    Mockito.when(p1.getPendingSize()).thenReturn(100);
+    flag = eventPluginLoader.isBusy();
+    Assert.assertFalse(flag);
+
+    Mockito.when(p1.getPendingSize()).thenReturn(60000);
+    flag = eventPluginLoader.isBusy();
+    Assert.assertTrue(flag);
+
+    Mockito.when(p1.getPendingSize()).thenThrow(new AbstractMethodError());
+    flag = eventPluginLoader.isBusy();
+    Assert.assertFalse(flag);
+  }
+}

framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java

@@ -26,6 +26,9 @@ public class HistoryEventServiceTest {
   @Test
   public void test() throws Exception {
     EventPluginLoader instance = mock(EventPluginLoader.class);
+    Mockito.when(instance.isUseNativeQueue()).thenReturn(true);
+    Mockito.when(instance.isUseNativeQueue()).thenReturn(false);
+
     ReflectUtils.setFieldValue(historyEventService, "instance", instance);
 
     DynamicPropertiesStore dynamicPropertiesStore = mock(DynamicPropertiesStore.class);
@@ -39,6 +42,7 @@ public void test() throws Exception {
     SolidEventService solidEventService = new SolidEventService();
     RealtimeEventService realtimeEventService = new RealtimeEventService();
     BlockEventLoad blockEventLoad = new BlockEventLoad();
+    ReflectUtils.setFieldValue(blockEventLoad, "instance", instance);
 
     ReflectUtils.setFieldValue(historyEventService, "solidEventService", solidEventService);
     ReflectUtils.setFieldValue(historyEventService, "realtimeEventService", realtimeEventService);
@@ -77,11 +81,16 @@ public void test() throws Exception {
     Mockito.when(chainBaseManager.getBlockIdByNum(1L))
         .thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 1));
 
+    Mockito.when(instance.isUseNativeQueue()).thenReturn(true);
+
     Method method1 = historyEventService.getClass().getDeclaredMethod("syncEvent");
     method1.setAccessible(true);
     method1.invoke(historyEventService);
 
+    Mockito.when(instance.isUseNativeQueue()).thenReturn(false);
+    Mockito.when(instance.isBusy()).thenReturn(true);
     historyEventService.init();
+    Thread.sleep(1000);
     historyEventService.close();
   }
 }

framework/src/test/java/org/tron/core/event/HistoryEventServiceTest.java

@@ -5,7 +5,8 @@
 import java.util.Date;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.Assert;
-import org.quartz.CronExpression;
+import org.junit.Test;
+import org.tron.common.cron.CronExpression;
 import org.tron.common.parameter.CommonParameter;
 
 @Slf4j
@@ -25,6 +26,21 @@ public class BlockTimeStopTest extends ConditionallyStopTest {
     }
   }
 
+  @Test
+  public void isValidExpression() {
+    Assert.assertTrue(CronExpression.isValidExpression(cronExpression.getCronExpression()));
+    ParseException err = Assert.assertThrows(ParseException.class, () ->
+        CronExpression.validateExpression("invalid expression"));
+    Assert.assertEquals("Illegal characters for this position: 'INV'", err.getMessage());
+  }
+
+  @Test
+  public void  getNextTime() {
+    Date date = cronExpression.getNextValidTimeAfter(new Date());
+    Date invalidDate = cronExpression.getNextInvalidTimeAfter(new Date());
+    Assert.assertNotEquals(date, invalidDate);
+  }
+
 
   protected void initParameter(CommonParameter parameter) {
     parameter.setShutdownBlockTime(cronExpression);

framework/src/test/java/org/tron/core/services/stop/BlockTimeStopTest.java

@@ -2088,19 +2088,6 @@
             <sha256 value="bca2bb252c6ec5db92584af7ab26f787b14a155f587c3e30ec1e1da0d4164694" origin="Generated by Gradle"/>
          </artifact>
       </component>
-      <component group="org.quartz-scheduler" name="quartz" version="2.3.2">
-         <artifact name="quartz-2.3.2.jar">
-            <sha256 value="639c6a675bc472e1568df9d8c954ff702da6f83ed27da0ff9a7bd12ed73b8bf0" origin="Generated by Gradle"/>
-         </artifact>
-         <artifact name="quartz-2.3.2.pom">
-            <sha256 value="f5bca37862760be888bdeed62e9f1061d7486a7d14917949ce0a91f3899a11dc" origin="Generated by Gradle"/>
-         </artifact>
-      </component>
-      <component group="org.quartz-scheduler" name="quartz-parent" version="2.3.2">
-         <artifact name="quartz-parent-2.3.2.pom">
-            <sha256 value="24286b15786e3b5a71899f7ff0df17449820257475e3e74ece97f43a684a4463" origin="Generated by Gradle"/>
-         </artifact>
-      </component>
       <component group="org.reactivestreams" name="reactive-streams" version="1.0.3">
          <artifact name="reactive-streams-1.0.3.jar">
             <sha256 value="1dee0481072d19c929b623e155e14d2f6085dc011529a0a0dbefc84cf571d865" origin="Generated by Gradle"/>