Simple sync

elastic · Aug 14, 2018 · 88dee76 · 88dee76
1 parent 9533679
commit 88dee76
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 20 deletions.
diff --git a/server/src/main/java/org/elasticsearch/index/shard/GlobalCheckpointListeners.java b/server/src/main/java/org/elasticsearch/index/shard/GlobalCheckpointListeners.java
@@ -91,12 +91,12 @@ public interface GlobalCheckpointListener {
      */
     synchronized void add(final long currentGlobalCheckpoint, final GlobalCheckpointListener listener) {
         if (closed) {
-            executor.execute(() -> listener.accept(UNASSIGNED_SEQ_NO, new IndexShardClosedException(shardId)));
+            executor.execute(() -> notifyListener(listener, UNASSIGNED_SEQ_NO, new IndexShardClosedException(shardId)));
             return;
         }
         if (lastKnownGlobalCheckpoint > currentGlobalCheckpoint) {
             // notify directly
-            executor.execute(() -> listener.accept(lastKnownGlobalCheckpoint, null));
+            executor.execute(() -> notifyListener(listener, lastKnownGlobalCheckpoint, null));
             return;
         } else {
             if (listeners == null) {
@@ -107,10 +107,8 @@ synchronized void add(final long currentGlobalCheckpoint, final GlobalCheckpoint
     }
 
     @Override
-    public void close() throws IOException {
-        synchronized (this) {
-            closed = true;
-        }
+    public synchronized void close() throws IOException {
+        closed = true;
         notifyListeners(UNASSIGNED_SEQ_NO, new IndexShardClosedException(shardId));
     }
 
@@ -123,25 +121,22 @@ synchronized int pendingListeners() {
      *
      * @param globalCheckpoint the updated global checkpoint
      */
-    void globalCheckpointUpdated(final long globalCheckpoint) {
+    synchronized void globalCheckpointUpdated(final long globalCheckpoint) {
         assert globalCheckpoint >= NO_OPS_PERFORMED;
-        synchronized (this) {
-            assert globalCheckpoint > lastKnownGlobalCheckpoint
-                    : "updated global checkpoint [" + globalCheckpoint + "]"
-                    + " is not more than the last known global checkpoint [" + lastKnownGlobalCheckpoint + "]";
-            lastKnownGlobalCheckpoint = globalCheckpoint;
-        }
+        assert globalCheckpoint > lastKnownGlobalCheckpoint
+                : "updated global checkpoint [" + globalCheckpoint + "]"
+                + " is not more than the last known global checkpoint [" + lastKnownGlobalCheckpoint + "]";
+        lastKnownGlobalCheckpoint = globalCheckpoint;
         notifyListeners(globalCheckpoint, null);
     }
 
     private void notifyListeners(final long globalCheckpoint, final IndexShardClosedException e) {
+        assert Thread.holdsLock(this);
         assert (globalCheckpoint == UNASSIGNED_SEQ_NO && e != null) || (globalCheckpoint >= NO_OPS_PERFORMED && e == null);
         if (listeners != null) {
-            final List<GlobalCheckpointListener> currentListeners;
-            synchronized (this) {
-                currentListeners = listeners;
-                listeners = null;
-            }
+            // capture the current listeners
+            final List<GlobalCheckpointListener> currentListeners = listeners;
+            listeners = null;
             if (currentListeners != null) {
                 executor.execute(() -> {
                     for (final GlobalCheckpointListener listener : currentListeners) {

diff --git a/server/src/test/java/org/elasticsearch/index/shard/GlobalCheckpointListenersTests.java b/server/src/test/java/org/elasticsearch/index/shard/GlobalCheckpointListenersTests.java
@@ -78,7 +78,7 @@ public void testGlobalCheckpointUpdated() throws IOException {
         }
 
         // test the listeners are not invoked twice
-        final long nextGlobalCheckpoint = randomValueOtherThan(globalCheckpoint, () -> randomLongBetween(NO_OPS_PERFORMED, Long.MAX_VALUE));
+        final long nextGlobalCheckpoint = randomLongBetween(globalCheckpoint + 1, Long.MAX_VALUE);
         globalCheckpointListeners.globalCheckpointUpdated(nextGlobalCheckpoint);
         for (int i = 0; i < numberOfListeners; i++) {
             assertThat(globalCheckpoints[i], equalTo(globalCheckpoint));
@@ -115,7 +115,7 @@ public void testListenersReadyToBeNotified() throws IOException {
         }
 
         // test the listeners are not invoked twice
-        final long nextGlobalCheckpoint = randomValueOtherThan(globalCheckpoint, () -> randomLongBetween(NO_OPS_PERFORMED, Long.MAX_VALUE));
+        final long nextGlobalCheckpoint = randomLongBetween(globalCheckpoint + 1, Long.MAX_VALUE);
         globalCheckpointListeners.globalCheckpointUpdated(nextGlobalCheckpoint);
         for (int i = 0; i < numberOfListeners; i++) {
             assertThat(globalCheckpoints[i], equalTo(globalCheckpoint));