Implement PausableScheduledThreadPool

src/main/java/com/baidu/hugegraph/threadpool/PausableScheduledThreadPool.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.threadpool;
+
+import java.util.List;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+
+import org.slf4j.Logger;
+
+import com.baidu.hugegraph.util.Log;
+
+public class PausableScheduledThreadPool extends ScheduledThreadPoolExecutor {
+
+    private static final Logger LOG = Log.logger(
+                                      PausableScheduledThreadPool.class);
+
+    private volatile boolean paused = false;
+
+    public PausableScheduledThreadPool(int corePoolSize,
+                                       ThreadFactory factory) {
+        super(corePoolSize, factory);
+    }
+
+    public synchronized void pauseSchedule() {
+        this.paused = true;
+        LOG.info("PausableScheduledThreadPool was paused");
+    }
+
+    public synchronized void resumeSchedule() {
+        this.paused = false;
+        this.notifyAll();
+        LOG.info("PausableScheduledThreadPool was resumed");
+    }
+
+    @Override
+    protected void beforeExecute(Thread t, Runnable r) {
+        synchronized (this) {
+            while (this.paused) {
+                try {
+                    this.wait();
+                } catch (InterruptedException e) {
+                    LOG.warn("PausableScheduledThreadPool was interrupted");
+                }
+            }
+        }
+        super.beforeExecute(t, r);
+    }
+
+    @Override
+    public void shutdown() {
+        if (this.paused) {
+            this.resumeSchedule();
+        }
+        super.shutdown();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+        if (this.paused) {
+            this.resumeSchedule();
+        }
+        return super.shutdownNow();
+    }
+}

src/main/java/com/baidu/hugegraph/util/ExecutorUtil.java

@@ -26,6 +26,8 @@
 
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 
+import com.baidu.hugegraph.threadpool.PausableScheduledThreadPool;
+
 public final class ExecutorUtil {
 
     public static ExecutorService newFixedThreadPool(String name) {
@@ -50,4 +52,17 @@ public static ScheduledExecutorService newScheduledThreadPool(int size,
                                                       .build();
         return Executors.newScheduledThreadPool(size, factory);
     }
+
+    public static PausableScheduledThreadPool newPausableScheduledThreadPool(
+                                              String name) {
+        return newPausableScheduledThreadPool(1, name);
+    }
+
+    public static PausableScheduledThreadPool newPausableScheduledThreadPool(
+                                              int size, String name) {
+        ThreadFactory factory = new BasicThreadFactory.Builder()
+                                                      .namingPattern(name)
+                                                      .build();
+        return new PausableScheduledThreadPool(size, factory);
+    }
 }

src/test/java/com/baidu/hugegraph/unit/UnitTestSuite.java

@@ -25,14 +25,14 @@
 import com.baidu.hugegraph.testutil.AssertTest;
 import com.baidu.hugegraph.testutil.WhiteboxTest;
 import com.baidu.hugegraph.unit.concurrent.AtomicLockTest;
+import com.baidu.hugegraph.unit.concurrent.BarrierEventTest;
 import com.baidu.hugegraph.unit.concurrent.KeyLockTest;
 import com.baidu.hugegraph.unit.concurrent.LockGroupTest;
 import com.baidu.hugegraph.unit.concurrent.LockManagerTest;
 import com.baidu.hugegraph.unit.concurrent.RowLockTest;
 import com.baidu.hugegraph.unit.config.HugeConfigTest;
 import com.baidu.hugegraph.unit.config.OptionSpaceTest;
 import com.baidu.hugegraph.unit.date.SafeDateFormatTest;
-import com.baidu.hugegraph.unit.concurrent.BarrierEventTest;
 import com.baidu.hugegraph.unit.event.EventHubTest;
 import com.baidu.hugegraph.unit.iterator.BatchMapperIteratorTest;
 import com.baidu.hugegraph.unit.iterator.ExtendableIteratorTest;
@@ -49,6 +49,7 @@
 import com.baidu.hugegraph.unit.perf.PerfUtilTest;
 import com.baidu.hugegraph.unit.rest.RestClientTest;
 import com.baidu.hugegraph.unit.rest.RestResultTest;
+import com.baidu.hugegraph.unit.threadpool.PausableScheduledThreadPoolTest;
 import com.baidu.hugegraph.unit.util.BytesTest;
 import com.baidu.hugegraph.unit.util.CollectionUtilTest;
 import com.baidu.hugegraph.unit.util.DateUtilTest;
@@ -71,6 +72,8 @@
     KeyLockTest.class,
     RowLockTest.class,
 
+    PausableScheduledThreadPoolTest.class,
+
     HugeConfigTest.class,
     OptionSpaceTest.class,
     SafeDateFormatTest.class,

src/test/java/com/baidu/hugegraph/unit/threadpool/PausableScheduledThreadPoolTest.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.unit.threadpool;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.baidu.hugegraph.threadpool.PausableScheduledThreadPool;
+import com.baidu.hugegraph.util.ExecutorUtil;
+
+public class PausableScheduledThreadPoolTest {
+
+    @Test
+    public void testscheduleWithFixedDelay() throws InterruptedException {
+        PausableScheduledThreadPool executor =
+                ExecutorUtil.newPausableScheduledThreadPool("test");
+        AtomicInteger counter = new AtomicInteger(0);
+        executor.scheduleWithFixedDelay(() -> {
+            System.out.println("counter: " + counter.incrementAndGet());
+        }, 2, 2, TimeUnit.SECONDS);
+
+        Thread.sleep(4500);
+        Assert.assertEquals(2, counter.get());
+
+        // pause
+        executor.pauseSchedule();
+        Thread.sleep(2000);
+        Assert.assertEquals(2, counter.get());
+
+        // resume
+        executor.resumeSchedule();
+        Thread.sleep(2000);
+        Assert.assertEquals(3, counter.get());
+
+        // pause again
+        executor.pauseSchedule();
+
+        executor.shutdown();
+        executor.awaitTermination(3, TimeUnit.SECONDS);
+    }
+
+    @Test
+    public void testscheduleWithFixedRate() throws InterruptedException {
+        PausableScheduledThreadPool executor =
+                ExecutorUtil.newPausableScheduledThreadPool(2, "test");
+        AtomicInteger counter = new AtomicInteger(0);
+        executor.scheduleAtFixedRate(() -> {
+            System.out.println("counter: " + counter.incrementAndGet());
+        }, 2, 2, TimeUnit.SECONDS);
+        Thread.sleep(4500);
+        Assert.assertEquals(2, counter.get());
+
+        // pause
+        executor.pauseSchedule();
+        Thread.sleep(2000);
+        Assert.assertEquals(2, counter.get());
+
+        // resume
+        executor.resumeSchedule();
+        Thread.sleep(2000);
+        Assert.assertEquals(4, counter.get());
+
+        // pause again
+        executor.pauseSchedule();
+
+        executor.shutdownNow();
+        executor.awaitTermination(3, TimeUnit.SECONDS);
+    }
+}