From b40f02d21477bd0593809ef7e72797e8532072ca Mon Sep 17 00:00:00 2001
From: Jean-Noel Rouvignac <jean-noel.rouvignac@forgerock.com>
Date: Mon, 16 Jun 2014 13:21:12 +0000
Subject: [PATCH] OPENDJ-1496 (CR-3767) ThreadInterruptedException (JE) when running replication tests
---
opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java | 11 ++---
opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java | 70 ++++++++++++++++++++++++-----------
2 files changed, 53 insertions(+), 28 deletions(-)
diff --git a/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java b/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java
index d0d1269..a3eaa0c 100644
--- a/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java
+++ b/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java
@@ -29,7 +29,7 @@
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.ConcurrentSkipListSet;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.api.DirectoryThread;
@@ -61,10 +61,13 @@
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
/**
- * If this is true, then the {@link #run()} method must clear its state.
- * Otherwise the run method executes normally.
+ * If it contains nothing, then the run method executes normally.
+ * Otherwise, the {@link #run()} method must clear its state
+ * for the supplied domain baseDNs. If a supplied domain is
+ * {@link DN#NULL_DN}, then all domains will be cleared.
*/
- private final AtomicBoolean doClear = new AtomicBoolean();
+ private final ConcurrentSkipListSet<DN> domainsToClear =
+ new ConcurrentSkipListSet<DN>();
private final ChangelogDB changelogDB;
/** Only used for initialization, and then discarded. */
private ChangelogState changelogState;
@@ -508,13 +511,17 @@
{
try
{
- if (doClear.get())
+ if (!domainsToClear.isEmpty())
{
- removeAllCursors();
+ while (!domainsToClear.isEmpty())
+ {
+ final DN baseDNToClear = domainsToClear.first();
+ removeCursors(baseDNToClear);
+ // Only release the waiting thread
+ // once this domain's state has been cleared.
+ domainsToClear.remove(baseDNToClear);
+ }
resetNextChangeForInsertDBCursor();
- // No need to use CAS here because it is only for unit tests and at
- // this point all will have been cleaned up anyway.
- doClear.set(false);
}
else
{
@@ -605,7 +612,7 @@
}
finally
{
- removeAllCursors();
+ removeCursors(DN.NULL_DN);
}
}
@@ -658,19 +665,33 @@
}
}
- private void removeAllCursors()
+ private void removeCursors(DN baseDN)
{
if (nextChangeForInsertDBCursor != null)
{
nextChangeForInsertDBCursor.close();
nextChangeForInsertDBCursor = null;
}
- for (Map<Integer, DBCursor<UpdateMsg>> map : allCursors.values())
+ if (DN.NULL_DN.equals(baseDN))
{
- StaticUtils.close(map.values());
+ // close all cursors
+ for (Map<Integer, DBCursor<UpdateMsg>> map : allCursors.values())
+ {
+ StaticUtils.close(map.values());
+ }
+ allCursors.clear();
+ newCursors.clear();
}
- allCursors.clear();
- newCursors.clear();
+ else
+ {
+ // close cursors for this DN
+ final Map<Integer, DBCursor<UpdateMsg>> map = allCursors.remove(baseDN);
+ if (map != null)
+ {
+ StaticUtils.close(map.values());
+ }
+ newCursors.remove(baseDN);
+ }
}
private Pair<DBCursor<UpdateMsg>, Iterator<Entry<Integer, DBCursor<UpdateMsg>>>>
@@ -738,22 +759,27 @@
}
/**
- * Asks the current thread to clear its state and blocks until state is
- * cleared.
+ * Asks the current thread to clear its state for the specified domain.
* <p>
- * This method is only useful for unit tests.
+ * Note: This method blocks the current thread until state is cleared.
+ *
+ * @param baseDN the baseDN to be cleared from this thread's state.
+ * {@code null} and {@link DN#NULL_DN} mean "clear all domains".
*/
- public void clear()
+ public void clear(DN baseDN)
{
- doClear.set(true);
- while (doClear.get() && !State.TERMINATED.equals(getState()))
+ // Use DN.NULL_DN to say "clear all domains"
+ final DN baseDNToClear = baseDN != null ? baseDN : DN.NULL_DN;
+ domainsToClear.add(baseDNToClear);
+ while (domainsToClear.contains(baseDNToClear)
+ && !State.TERMINATED.equals(getState()))
{
// wait until clear() has been done by thread, always waking it up
synchronized (this)
{
notify();
}
- // ensures unit tests wait that this thread's state is cleaned up
+ // ensures thread wait that this thread's state is cleaned up
Thread.yield();
}
}
diff --git a/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java b/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java
index a5540a2..32c0c08 100644
--- a/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java
+++ b/opendj-sdk/opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java
@@ -454,12 +454,6 @@
// - then throw the first encountered exception
ChangelogException firstException = null;
- final ChangeNumberIndexer indexer = cnIndexer.get();
- if (indexer != null)
- {
- indexer.clear();
- }
-
for (DN baseDN : this.domainToReplicaDBs.keySet())
{
removeDomain(baseDN);
@@ -546,6 +540,11 @@
Map<Integer, JEReplicaDB> domainMap = domainToReplicaDBs.get(baseDN);
if (domainMap != null)
{
+ final ChangeNumberIndexer indexer = this.cnIndexer.get();
+ if (indexer != null)
+ {
+ indexer.clear(baseDN);
+ }
synchronized (domainMap)
{
domainMap = domainToReplicaDBs.remove(baseDN);
--
Gitblit v1.10.0