Fix queue capacity invariants

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java

@@ -50,6 +50,9 @@ public synchronized void increaseReference(final PipeParameters parameters)
   public synchronized void decreaseReference(final PipeParameters parameters)
       throws IllegalPathException {
     if (!ConfigRegionListeningFilter.parseListeningPlanTypeSet(parameters).isEmpty()) {
+      if (listeningQueueReferenceCount == 0) {
+        return;
+      }
       listeningQueueReferenceCount--;
       if (listeningQueueReferenceCount == 0) {
         listeningQueue.close();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorLock.java

@@ -24,74 +24,105 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
 /**
  * {@link PipeTaskCoordinatorLock} is a cross-thread lock for pipe task coordinator. It is used to
  * ensure that only one thread can execute the pipe task coordinator at the same time.
  *
- * <p>Uses {@link Semaphore} instead of {@link java.util.concurrent.locks.ReentrantLock} to support
- * cross-thread acquire/release, which is required by the procedure recovery mechanism: locks may be
- * acquired on the StateMachineUpdater thread during {@code restoreLock()} and released on a
- * ProcedureCoreWorker thread after execution.
+ * <p>Supports cross-thread acquire/release, which is required by the procedure recovery mechanism:
+ * locks may be acquired on the StateMachineUpdater thread during {@code restoreLock()} and released
+ * on a ProcedureCoreWorker thread after execution.
  */
 public class PipeTaskCoordinatorLock {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeTaskCoordinatorLock.class);
 
-  private final Semaphore semaphore = new Semaphore(1);
+  private boolean locked = false;
 
   public void lock() {
     LOGGER.debug(
         ManagerMessages.PIPETASKCOORDINATOR_LOCK_WAITING_FOR_THREAD,
         Thread.currentThread().getName());
-    try {
-      semaphore.acquire();
-      LOGGER.debug(
-          ManagerMessages.PIPETASKCOORDINATOR_LOCK_ACQUIRED_BY_THREAD,
-          Thread.currentThread().getName());
-    } catch (final InterruptedException e) {
+
+    boolean interrupted = false;
+    synchronized (this) {
+      while (locked) {
+        try {
+          wait();
+        } catch (final InterruptedException e) {
+          interrupted = true;
+          LOGGER.error(
+              ManagerMessages.INTERRUPTED_WHILE_WAITING_FOR_PIPETASKCOORDINATOR_LOCK_CURRENT_THREAD,
+              Thread.currentThread().getName());
+        }
+      }
+      locked = true;
+    }
+
+    if (interrupted) {
       Thread.currentThread().interrupt();
-      LOGGER.error(
-          ManagerMessages.INTERRUPTED_WHILE_WAITING_FOR_PIPETASKCOORDINATOR_LOCK_CURRENT_THREAD,
-          Thread.currentThread().getName());
     }
+    LOGGER.debug(
+        ManagerMessages.PIPETASKCOORDINATOR_LOCK_ACQUIRED_BY_THREAD,
+        Thread.currentThread().getName());
   }
 
   public boolean tryLock() {
-    try {
-      LOGGER.debug(
-          ManagerMessages.PIPETASKCOORDINATOR_LOCK_WAITING_FOR_THREAD,
-          Thread.currentThread().getName());
-      if (semaphore.tryAcquire(10, TimeUnit.SECONDS)) {
-        LOGGER.debug(
-            ManagerMessages.PIPETASKCOORDINATOR_LOCK_ACQUIRED_BY_THREAD,
-            Thread.currentThread().getName());
-        return true;
-      } else {
-        LOGGER.info(
-            ManagerMessages.PIPETASKCOORDINATOR_LOCK_FAILED_TO_ACQUIRE_BY_THREAD_BECAUSE_OF_TIMEOUT,
-            Thread.currentThread().getName());
-        return false;
-      }
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
+    LOGGER.debug(
+        ManagerMessages.PIPETASKCOORDINATOR_LOCK_WAITING_FOR_THREAD,
+        Thread.currentThread().getName());
+    if (Thread.currentThread().isInterrupted()) {
       LOGGER.error(
           ManagerMessages.INTERRUPTED_WHILE_WAITING_FOR_PIPETASKCOORDINATOR_LOCK_CURRENT_THREAD,
           Thread.currentThread().getName());
       return false;
     }
+
+    final long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+    synchronized (this) {
+      while (locked) {
+        final long remainingNanos = deadlineNanos - System.nanoTime();
+        if (remainingNanos <= 0) {
+          LOGGER.info(
+              ManagerMessages
+                  .PIPETASKCOORDINATOR_LOCK_FAILED_TO_ACQUIRE_BY_THREAD_BECAUSE_OF_TIMEOUT,
+              Thread.currentThread().getName());
+          return false;
+        }
+        try {
+          TimeUnit.NANOSECONDS.timedWait(this, remainingNanos);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          LOGGER.error(
+              ManagerMessages.INTERRUPTED_WHILE_WAITING_FOR_PIPETASKCOORDINATOR_LOCK_CURRENT_THREAD,
+              Thread.currentThread().getName());
+          return false;
+        }
+      }
+      locked = true;
+    }
+
+    LOGGER.debug(
+        ManagerMessages.PIPETASKCOORDINATOR_LOCK_ACQUIRED_BY_THREAD,
+        Thread.currentThread().getName());
+    return true;
   }
 
   public void unlock() {
-    semaphore.release();
+    synchronized (this) {
+      if (!locked) {
+        return;
+      }
+      locked = false;
+      notifyAll();
+    }
     LOGGER.debug(
         ManagerMessages.PIPETASKCOORDINATOR_LOCK_RELEASED_BY_THREAD,
         Thread.currentThread().getName());
   }
 
-  public boolean isLocked() {
-    return semaphore.availablePermits() == 0;
+  public synchronized boolean isLocked() {
+    return locked;
   }
 }

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListenerTest.java

@@ -0,0 +1,55 @@
+/*
+ * 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 org.apache.iotdb.confignode.manager.pipe.agent.runtime;
+
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+
+public class PipeConfigRegionListenerTest {
+
+  @Test
+  public void testRepeatedDecreaseDoesNotBreakFutureIncrease() throws Exception {
+    PipeConfigRegionListener listener = new PipeConfigRegionListener();
+    PipeParameters parameters =
+        new PipeParameters(
+            new HashMap<String, String>() {
+              {
+                put(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "schema.database.create");
+              }
+            });
+
+    listener.increaseReference(parameters);
+    Assert.assertTrue(listener.listener().isOpened());
+
+    listener.decreaseReference(parameters);
+    Assert.assertFalse(listener.listener().isOpened());
+
+    listener.decreaseReference(parameters);
+    Assert.assertFalse(listener.listener().isOpened());
+
+    listener.increaseReference(parameters);
+    Assert.assertTrue(listener.listener().isOpened());
+  }
+}

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorLockTest.java

@@ -0,0 +1,108 @@
+/*
+ * 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 org.apache.iotdb.confignode.manager.pipe.coordinator.task;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class PipeTaskCoordinatorLockTest {
+
+  @Test
+  public void testRepeatedUnlockDoesNotIncreaseCapacity() throws Exception {
+    PipeTaskCoordinatorLock lock = new PipeTaskCoordinatorLock();
+
+    lock.lock();
+    lock.unlock();
+    lock.unlock();
+    Assert.assertTrue(lock.tryLock());
+
+    AtomicBoolean acquired = new AtomicBoolean(false);
+    Thread waiter =
+        new Thread(
+            () -> {
+              lock.lock();
+              acquired.set(true);
+            });
+    waiter.start();
+    waitUntilState(waiter, Thread.State.WAITING);
+    Assert.assertFalse(acquired.get());
+
+    lock.unlock();
+    waiter.join(TimeUnit.SECONDS.toMillis(5));
+    Assert.assertFalse(waiter.isAlive());
+    Assert.assertTrue(acquired.get());
+    lock.unlock();
+  }
+
+  @Test
+  public void testInterruptedLockDoesNotReturnBeforeAcquired() throws Exception {
+    PipeTaskCoordinatorLock lock = new PipeTaskCoordinatorLock();
+    AtomicBoolean acquired = new AtomicBoolean(false);
+    AtomicBoolean interruptedAfterLock = new AtomicBoolean(false);
+
+    lock.lock();
+    Thread waiter =
+        new Thread(
+            () -> {
+              Thread.currentThread().interrupt();
+              lock.lock();
+              acquired.set(true);
+              interruptedAfterLock.set(Thread.currentThread().isInterrupted());
+            });
+    waiter.start();
+    waitUntilState(waiter, Thread.State.WAITING);
+    Assert.assertFalse(acquired.get());
+
+    lock.unlock();
+    waiter.join(TimeUnit.SECONDS.toMillis(5));
+    Assert.assertFalse(waiter.isAlive());
+    Assert.assertTrue(acquired.get());
+    Assert.assertTrue(interruptedAfterLock.get());
+    lock.unlock();
+  }
+
+  @Test
+  public void testInterruptedTryLockDoesNotAcquire() {
+    PipeTaskCoordinatorLock lock = new PipeTaskCoordinatorLock();
+
+    Thread.currentThread().interrupt();
+    try {
+      Assert.assertFalse(lock.tryLock());
+      Assert.assertFalse(lock.isLocked());
+      Assert.assertTrue(Thread.currentThread().isInterrupted());
+    } finally {
+      Thread.interrupted();
+    }
+  }
+
+  private void waitUntilState(Thread thread, Thread.State expectedState) throws Exception {
+    long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+    while (System.nanoTime() < deadline) {
+      if (thread.getState() == expectedState) {
+        return;
+      }
+      Thread.sleep(10);
+    }
+    Assert.assertEquals(expectedState, thread.getState());
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java

@@ -374,8 +374,10 @@ private Map<String, Map<String, Long>> mayGetTableInPreUpdateMap(
   private Map<String, Map<String, TsTable>> getTablesInConfigNode(
       final Map<String, Map<String, Long>> tableInput) {
     Map<String, Map<String, TsTable>> result = Collections.emptyMap();
+    boolean acquired = false;
     try {
       fetchTableSemaphore.acquire();
+      acquired = true;
       final TFetchTableResp resp =
           ClusterConfigTaskExecutor.getInstance()
               .fetchTables(
@@ -388,11 +390,11 @@ private Map<String, Map<String, TsTable>> getTablesInConfigNode(
     } catch (final InterruptedException e) {
       Thread.currentThread().interrupt();
       LOGGER.warn(DataNodeSchemaMessages.INTERRUPTED_ACQUIRE_SEMAPHORE_GET_TABLES);
-    } catch (final Exception e) {
-      fetchTableSemaphore.release();
-      throw e;
+    } finally {
+      if (acquired) {
+        fetchTableSemaphore.release();
+      }
     }
-    fetchTableSemaphore.release();
     return result;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/concurrent/FiniteSemaphore.java

@@ -20,15 +20,12 @@
 
 import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
 
-import java.util.concurrent.Semaphore;
-
 /**
  * FiniteSemaphore defines a special Semaphore that the upper limit of permit is capacity. If
  * permits exceed the capacity, the release request will be ignored.
  */
 public class FiniteSemaphore {
   private final int capacity;
-  private final Semaphore semaphore;
   private int permit;
 
   public FiniteSemaphore(int capacity, int permit) {
@@ -39,23 +36,20 @@ public FiniteSemaphore(int capacity, int permit) {
       throw new IllegalArgumentException(DataNodeMiscMessages.CAPACITY_LARGER_THAN_INITIAL_PERMITS);
     }
     this.capacity = capacity;
-    this.semaphore = new Semaphore(permit);
     this.permit = permit;
   }
 
-  public void release() {
-    synchronized (this) {
-      if (permit < capacity) {
-        permit++;
-        semaphore.release();
-      }
+  public synchronized void release() {
+    if (permit < capacity) {
+      permit++;
+      notifyAll();
     }
   }
 
-  public void acquire() throws InterruptedException {
-    semaphore.acquire();
-    synchronized (this) {
-      permit--;
+  public synchronized void acquire() throws InterruptedException {
+    while (permit == 0) {
+      wait();
     }
+    permit--;
   }
 }