add BarrierEvent

src/main/java/com/baidu/hugegraph/event/BarrierEvent.java → src/main/java/com/baidu/hugegraph/concurrent/BarrierEvent.java

@@ -18,7 +18,7 @@
  *  under the License.
  */
 
-package com.baidu.hugegraph.event;
+package com.baidu.hugegraph.concurrent;
 
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
@@ -33,70 +33,75 @@ public class BarrierEvent {
     private final Condition cond = lock.newCondition();
     private boolean signaled = false;
 
-    public void reset() {
+    /**
+     * Wait forever until the signal is received.
+     * @return true if signal is received
+     * @throws InterruptedException if interrupted.
+     */
+    public void await() throws InterruptedException {
         this.lock.lock();
         try {
-            this.signaled = false;
+            while (!this.signaled) {
+                this.cond.await();
+            }
         } finally {
             this.lock.unlock();
         }
     }
 
-    public void signal() {
+    /**
+     * Wait specified time in milliseconds.
+     * @param timeout: the time in millisecond to wait.
+     * @return true if signal is received, false if time out.
+     */
+    public boolean await(long timeout) throws InterruptedException {
+        E.checkArgument(timeout >= 0L,
+                        "The time must be >= 0, but got '%d'.",
+                        timeout);
+        long deadline = System.currentTimeMillis() + timeout;
         this.lock.lock();
         try {
-            this.signaled = true;
-            this.cond.signal();
+            while (!this.signaled) {
+                timeout = deadline - System.currentTimeMillis();
+                if (timeout > 0) {
+                    this.cond.await(timeout, TimeUnit.MILLISECONDS);
+                }
+                if (System.currentTimeMillis() >= deadline) {
+                    return this.signaled;
+                }
+            }
         } finally {
             this.lock.unlock();
         }
+        return true;
     }
 
-    public void signalAll() {
+    public void reset() {
         this.lock.lock();
         try {
-            this.signaled = true;
-            this.cond.signalAll();
+            this.signaled = false;
         } finally {
             this.lock.unlock();
         }
     }
 
-    /**
-     * Wait forever until the signal is received.
-     * @return true if signal is received
-     * @throws InterruptedException if interrupted.
-     */
-    public void await() throws InterruptedException {
+    public void signal() {
         this.lock.lock();
         try {
-            while (!this.signaled) {
-                this.cond.await();
-            }
+            this.signaled = true;
+            this.cond.signal();
         } finally {
             this.lock.unlock();
         }
     }
 
-    public boolean await(long time) throws InterruptedException {
-        E.checkArgument(time >= 0L,
-                        "The time must be >= 0, but got '%d'.",
-                        time);
-        long deadline = System.currentTimeMillis() + time;
+    public void signalAll() {
         this.lock.lock();
         try {
-            while (!this.signaled) {
-                time = deadline - System.currentTimeMillis();
-                if (time > 0) {
-                    this.cond.await(time, TimeUnit.MILLISECONDS);
-                }
-                if (System.currentTimeMillis() >= deadline) {
-                    return this.signaled;
-                }
-            }
+            this.signaled = true;
+            this.cond.signalAll();
         } finally {
             this.lock.unlock();
         }
-        return true;
     }
 }

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

@@ -32,7 +32,7 @@
 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.event.BarrierEventTest;
+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;

src/test/java/com/baidu/hugegraph/unit/event/BarrierEventTest.java → src/test/java/com/baidu/hugegraph/unit/concurrent/BarrierEventTest.java

@@ -1,30 +1,17 @@
-package com.baidu.hugegraph.unit.event;
+package com.baidu.hugegraph.unit.concurrent;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
-import com.baidu.hugegraph.event.BarrierEvent;
+import com.baidu.hugegraph.concurrent.BarrierEvent;
 import com.baidu.hugegraph.testutil.Assert;
 
 public class BarrierEventTest {
 
-    @Test
-    public void testWaitMillis() throws InterruptedException {
-        BarrierEvent barrierEvent = new BarrierEvent();
-        boolean signaled = barrierEvent.await(1L);
-        Assert.assertFalse(signaled);
-        barrierEvent.signal();
-        signaled = barrierEvent.await(1L);
-        Assert.assertTrue(signaled);
-        barrierEvent.reset();
-        signaled = barrierEvent.await(1L);
-        Assert.assertFalse(signaled);
-    }
-
     @Test(timeout = 5000)
-    public void testWait() throws InterruptedException {
+    public void testAWait() throws InterruptedException {
         BarrierEvent barrierEvent = new BarrierEvent();
         AtomicInteger result = new AtomicInteger(0);
         CountDownLatch latch = new CountDownLatch(2);
@@ -47,4 +34,44 @@ public void testWait() throws InterruptedException {
         latch.await();
         Assert.assertEquals(1, result.get());
     }
+
+    @Test
+    public void testAWaitWithTimeout() throws InterruptedException {
+        BarrierEvent barrierEvent = new BarrierEvent();
+        boolean signaled = barrierEvent.await(1L);
+        Assert.assertFalse(signaled);
+    }
+
+    @Test
+    public void testReset() throws InterruptedException {
+        BarrierEvent barrierEvent = new BarrierEvent();
+        boolean signaled = barrierEvent.await(1L);
+        Assert.assertFalse(signaled);
+        barrierEvent.signal();
+        signaled = barrierEvent.await(1L);
+        Assert.assertTrue(signaled);
+        barrierEvent.reset();
+        signaled = barrierEvent.await(1L);
+        Assert.assertFalse(signaled);
+    }
+
+    @Test
+    public void testSignal() throws InterruptedException {
+        BarrierEvent barrierEvent = new BarrierEvent();
+        boolean signaled = barrierEvent.await(1L);
+        Assert.assertFalse(signaled);
+        barrierEvent.signal();
+        signaled = barrierEvent.await(1L);
+        Assert.assertTrue(signaled);
+    }
+
+    @Test
+    public void testSignalAll() throws InterruptedException {
+        BarrierEvent barrierEvent = new BarrierEvent();
+        boolean signaled = barrierEvent.await(1L);
+        Assert.assertFalse(signaled);
+        barrierEvent.signalAll();
+        signaled = barrierEvent.await(1L);
+        Assert.assertTrue(signaled);
+    }
 }