diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendImpl.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendImpl.java index 40b0c0f966..03cd9303a7 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendImpl.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendImpl.java @@ -29,7 +29,6 @@ import java.util.Set; import java.util.SortedSet; import java.util.concurrent.ExecutionException; -import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.i18n.LocalizableException; import org.forgerock.i18n.LocalizableMessage; @@ -93,9 +92,14 @@ public abstract class BackendImpl extends LocalBa /** The root container to use for this backend. */ private RootContainer rootContainer; - // FIXME: this is broken. Replace with read-write lock. - /** A count of the total operation threads currently in the backend. */ - private final AtomicInteger threadTotalCount = new AtomicInteger(0); + /** + * A count of the total operation threads currently in the backend. Bumped + * twice per operation by all worker threads, so it uses a striped counter + * to avoid contending on a single cache line; it is only read when waiting + * for the backend to become quiescent, which is why it is not a LongAdder — + * see {@link StripedCounter}. + */ + private final StripedCounter threadTotalCount = new StripedCounter(); /** The base DNs defined for this backend instance. */ private Set baseDNs; @@ -146,14 +150,14 @@ private EntryContainer accessBegin(Operation operation, DN entryDN, ResultCode n throw new DirectoryException( noEntryContainerResultCode, ERR_BACKEND_ENTRY_DOESNT_EXIST.get(entryDN, getBackendID())); } - threadTotalCount.getAndIncrement(); + threadTotalCount.increment(); return ec; } /** End a Backend API method that accesses the EntryContainer. */ private void accessEnd() { - threadTotalCount.getAndDecrement(); + threadTotalCount.decrement(); } /** @@ -163,7 +167,7 @@ private void accessEnd() */ private void waitUntilQuiescent() { - while (threadTotalCount.get() > 0) + while (threadTotalCount.sum() > 0) { // Still have threads accessing the storage so sleep a little try @@ -268,7 +272,7 @@ public void closeBackend() } // Make sure the thread counts are zero for next initialization. - threadTotalCount.set(0); + threadTotalCount.reset(); // Log an informational message. logger.info(NOTE_BACKEND_OFFLINE, cfg.getBackendId()); @@ -356,7 +360,7 @@ public ConditionResult hasSubordinates(DN entryDN) throws DirectoryException throw de; } - container.sharedLock.lock(); + container.beginSharedAccess(); try { return ConditionResult.valueOf(container.hasSubordinates(entryDN)); @@ -367,7 +371,7 @@ public ConditionResult hasSubordinates(DN entryDN) throws DirectoryException } finally { - container.sharedLock.unlock(); + container.endSharedAccess(); accessEnd(); } } @@ -378,7 +382,7 @@ public long getNumberOfEntriesInBaseDN(DN baseDN) throws DirectoryException checkNotNull(baseDN, "baseDN must not be null"); final EntryContainer ec = accessBegin(null, baseDN); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { return ec.getNumberOfEntriesInBaseDN(); @@ -390,7 +394,7 @@ public long getNumberOfEntriesInBaseDN(DN baseDN) throws DirectoryException } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -417,7 +421,7 @@ public long getNumberOfChildren(DN parentDN) throws DirectoryException throw de; } - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { return ec.getNumberOfChildren(parentDN); @@ -428,7 +432,7 @@ public long getNumberOfChildren(DN parentDN) throws DirectoryException } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -437,7 +441,7 @@ public long getNumberOfChildren(DN parentDN) throws DirectoryException public boolean entryExists(final DN entryDN) throws DirectoryException { EntryContainer ec = accessBegin(null, entryDN); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { return ec.entryExists(entryDN); @@ -448,7 +452,7 @@ public boolean entryExists(final DN entryDN) throws DirectoryException } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -457,7 +461,7 @@ public boolean entryExists(final DN entryDN) throws DirectoryException public Entry getEntry(DN entryDN) throws DirectoryException { EntryContainer ec = accessBegin(null, entryDN); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { return ec.getEntry(entryDN); @@ -468,7 +472,7 @@ public Entry getEntry(DN entryDN) throws DirectoryException } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -478,7 +482,7 @@ public void addEntry(Entry entry, AddOperation addOperation) throws DirectoryExc { EntryContainer ec = accessBegin(addOperation, entry.getName()); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { ec.addEntry(entry, addOperation); @@ -489,7 +493,7 @@ public void addEntry(Entry entry, AddOperation addOperation) throws DirectoryExc } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -500,7 +504,7 @@ public void deleteEntry(DN entryDN, DeleteOperation deleteOperation) { EntryContainer ec = accessBegin(deleteOperation, entryDN); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { ec.deleteEntry(entryDN, deleteOperation); @@ -511,7 +515,7 @@ public void deleteEntry(DN entryDN, DeleteOperation deleteOperation) } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -522,7 +526,7 @@ public void replaceEntry(Entry oldEntry, Entry newEntry, ModifyOperation modifyO { EntryContainer ec = accessBegin(modifyOperation, newEntry.getName()); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { @@ -534,7 +538,7 @@ public void replaceEntry(Entry oldEntry, Entry newEntry, ModifyOperation modifyO } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } @@ -554,7 +558,7 @@ public void renameEntry(DN currentDN, Entry entry, ModifyDNOperation modifyDNOpe throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM, WARN_FUNCTION_NOT_SUPPORTED.get()); } - currentContainer.sharedLock.lock(); + currentContainer.beginSharedAccess(); try { currentContainer.renameEntry(currentDN, entry, modifyDNOperation); @@ -565,7 +569,7 @@ public void renameEntry(DN currentDN, Entry entry, ModifyDNOperation modifyDNOpe } finally { - currentContainer.sharedLock.unlock(); + currentContainer.endSharedAccess(); accessEnd(); } } @@ -577,7 +581,7 @@ public void search(SearchOperation searchOperation) throws DirectoryException, C // is concerned: report it as such instead of the UNDEFINED result code used internally. EntryContainer ec = accessBegin(searchOperation, searchOperation.getBaseDN(), ResultCode.NO_SUCH_OBJECT); - ec.sharedLock.lock(); + ec.beginSharedAccess(); try { @@ -589,7 +593,7 @@ public void search(SearchOperation searchOperation) throws DirectoryException, C } finally { - ec.sharedLock.unlock(); + ec.endSharedAccess(); accessEnd(); } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java index 04827b488c..4cb8614ef4 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java @@ -236,7 +236,7 @@ public ConfigChangeResult applyConfigurationDelete(final BackendIndexCfg cfg) { final ConfigChangeResult ccr = new ConfigChangeResult(); - exclusiveLock.lock(); + EntryContainer.this.lock(); try { storage.write(new WriteOperation() @@ -256,7 +256,7 @@ public void run(WriteableTransaction txn) throws Exception } finally { - exclusiveLock.unlock(); + EntryContainer.this.unlock(); } return ccr; @@ -318,7 +318,7 @@ public boolean isConfigurationDeleteAcceptable(BackendVLVIndexCfg cfg, List + * The increment and the matching decrement of one logical access must be + * performed by the same thread; the stripe is a pure function of the thread, + * so both land in the same slot. A scan reads each slot once, i.e. observes a + * prefix of each slot's modification history, and within one slot a decrement + * can never be observed without the increment that preceded it, so every + * per-slot subtotal is non-negative. {@link java.util.concurrent.atomic.LongAdder} + * does not provide this: the two halves of a pair may land in different cells + * (probe rehash after CAS contention, cell table growth), letting a scan + * observe the decrement while missing the increment and under-count to a + * false zero. + */ +final class StripedCounter +{ + /** 16 longs = 128 bytes between slots, to keep them on distinct cache lines. */ + private static final int SPACING = 16; + private static final int STRIPES = nextPowerOfTwo(Runtime.getRuntime().availableProcessors()); + + private final AtomicLongArray counts = new AtomicLongArray(STRIPES * SPACING); + + private static int nextPowerOfTwo(int n) + { + int p = 1; + while (p < n) + { + p <<= 1; + } + return p; + } + + private static int slot() + { + final long id = Thread.currentThread().getId(); + return (((int) ((id * 0x9E3779B97F4A7C15L) >>> 32)) & (STRIPES - 1)) * SPACING; + } + + void increment() + { + counts.getAndIncrement(slot()); + } + + /** Must be called by the same thread that did the paired {@link #increment()}. */ + void decrement() + { + counts.getAndDecrement(slot()); + } + + /** + * Returns the current count. Concurrent updates may cause over-estimation, + * but a paired increment/decrement is never observed half-way in the + * decrement-only direction, so the result is zero only if every access + * whose increment is visible has completed. + */ + long sum() + { + long s = 0; + for (int i = 0; i < counts.length(); i += SPACING) + { + s += counts.get(i); + } + return s; + } + + /** Resets the count to zero. Only safe when no accesses are in flight. */ + void reset() + { + for (int i = 0; i < counts.length(); i += SPACING) + { + counts.set(i, 0); + } + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java index cae7c05f1d..eef3190788 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java @@ -31,9 +31,12 @@ import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import com.google.common.io.Resources; import org.forgerock.opendj.ldap.*; @@ -629,6 +632,50 @@ private int getTotalNumberOfLDIFEntries() return topEntries.size() + entries.size() + workEntries.size(); } + @Test(timeOut = 30000) + public void testExclusiveLockDrainsSharedAccessDespiteInterrupt() throws Exception + { + final EntryContainer ec = backend.getRootContainer().getEntryContainer(testBaseDN); + final CountDownLatch lockAcquired = new CountDownLatch(1); + final AtomicBoolean interruptPreserved = new AtomicBoolean(); + + ec.beginSharedAccess(); + final Thread exclusiveLocker = new Thread("Test exclusive locker") + { + @Override + public void run() + { + ec.lock(); + try + { + interruptPreserved.set(Thread.currentThread().isInterrupted()); + lockAcquired.countDown(); + } + finally + { + ec.unlock(); + } + } + }; + + try + { + exclusiveLocker.start(); + exclusiveLocker.interrupt(); + assertFalse(lockAcquired.await(200, TimeUnit.MILLISECONDS), + "lock() returned while a shared access was still in flight"); + } + finally + { + ec.endSharedAccess(); + } + + assertTrue(lockAcquired.await(10, TimeUnit.SECONDS), + "lock() did not complete after the shared access ended"); + assertTrue(interruptPreserved.get(), "lock() must preserve the caller's interrupt status"); + exclusiveLocker.join(10000); + } + @Test public void testHasSubordinates() throws Exception {