[
https://issues.apache.org/jira/browse/CASSANDRA-21019?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18108923#comment-18108923
]
Dmitry Konstantinov commented on CASSANDRA-21019:
-------------------------------------------------
original deadlock mechanism in more details:
h2. 1. The problem the design has to solve, and the escape path
Writes consume memtable memory. Memtable memory is only released when a flush
completes. A flush can only complete once the writes already in flight finish.
So "write waits for memory" and "flush waits for writes" are mutually
recursive, and without care that is a cycle.
Pre-patch, {{MemtableAllocator.SubAllocator.allocate}} resolved it with an
{*}escape hatch{*}:
{code:java}
public void allocate(long size, OpOrder.Group opGroup)
{
assert size >= 0;
while (true)
{
if (parent.tryAllocate(size)) { acquired(size); return; } // (1) room
available -> take it
if (opGroup.isBlocking()) { allocated(size); return; } // (2)
ESCAPE HATCH -> exceed the limit
WaitQueue.Signal signal =
parent.hasRoom().register(parent.blockedTimerContext(),
Timer.Context::stop); // (3) register for wake-up
opGroup.notifyIfBlocking(signal); //
...and for markBlocking
boolean allocated = parent.tryAllocate(size);
if (allocated) { signal.cancel(); acquired(size); return; }
else
signal.awaitThrowUncheckedOnInterrupt(); // (4) PARK
}
}
{code}
Line (2) is the invariant that keeps the system live: {*}if a flush is waiting
for my operation, I am allowed to blow through the memory limit so that the
flush can proceed{*}.
A flush announces this by calling {{markBlocking()}} before {{{}await(){}}}:
{code:java}
// ColumnFamilyStore.Flush.run(), lines 1276-1286
public void run()
{
// mark writes older than the barrier as blocking progress, permitting them
to exceed our memory limit
// if they are stuck waiting on it, then wait for them all to complete
writeBarrier.markBlocking();
writeBarrier.await();
}
{code}
Both interleavings are covered:
* Marked *before* the writer parks - caught at (2).
* Marked *after* the writer parks - {{notifyIfBlocking}}
(OpOrder.java:324-331) added the signal to the group's list, and
{{markBlocking()}} walks that list and signals it:
{code:java}
// OpOrder.Group, lines 324-339
public void notifyIfBlocking(WaitQueue.Signal signal)
{
if (blocking == null)
blockingUpdater.compareAndSet(this, null, new
ConcurrentLinkedQueue<>());
blocking.add(signal);
if (isBlocking() && blocking.remove(signal))
signal.signal();
}
private void markBlocking()
{
isBlocking = true;
ConcurrentLinkedQueue<WaitQueue.Signal> blocking = this.blocking;
if (blocking != null)
blocking.forEach(WaitQueue.Signal::signal);
}
{code}
So the guarantee is: *any thread parked in the allocator, whose op-group the
flush awaits, will be released.*
h2. 2. Where the guarantee leaks
{{TrieMemtable.MemtableShard.put}} takes a {{ReentrantLock}} and then allocates
*inside* it:
{code:java}
// TrieMemtable.java:550-593
boolean locked = writeLock.tryLock();
if (locked) { metrics.uncontendedPuts.inc(); }
else
{
metrics.contendedPuts.inc();
long lockStartTime = Clock.Global.nanoTime();
writeLock.lock(); // line
562 -- W1 queues HERE
metrics.contentionTime.addNano(Clock.Global.nanoTime() - lockStartTime);
}
try
{
long onHeap = data.sizeOnHeap();
long offHeap = data.sizeOffHeap();
data.putSingleton(key, update, updater::mergePartitions, ...); //
allocates via cloner + indexer callbacks
allocator.offHeap().adjust(data.sizeOffHeap() - offHeap, opGroup); //
line 576 -+ these reach
allocator.onHeap().adjust(data.sizeOnHeap() - onHeap, opGroup); //
line 577 -+ allocate()
}
finally { writeLock.unlock(); }
{code}
So a thread {*}can park on the memory pool while holding the shard lock{*}. The
escape hatch understands the pool's {{{}WaitQueue{}}}; it understands nothing
about {{{}ReentrantLock{}}}.
{{ShardedSkipListMemtable.Locking}} has the same shape with {{synchronized
(shard)}} at line 523.
h2. 3. The deadlock, step by step
{noformat}
Keyspace.writeOrder -- ONE static OpOrder for the whole node (Keyspace.java:102)
T0 W1 starts: g1 = writeOrder.start() g1 is the current group
W1 is a write to TABLE B
T1 TABLE A flushes. new Flush(false)
ColumnFamilyStore.java:1080
writeBarrier = Keyspace.writeOrder.newBarrier() line 1247
writeBarrier.issue() line 1271
-> orderOnOrBefore = g1
-> new current group = g2
flushExecutor.execute(flush) line 1081
T2 W2 starts: g2 = writeOrder.start() g2 is POST-barrier
W2 is also a write to TABLE B, same shard
T3 Table B is NOT flushing -> M_B.writeBarrier == null
-> accepts(g1) = accepts(g2) = true
Both W1 and W2 are routed to M_B: same MemtableShard, same writeLock.
T4 W2 wins the lock (tryLock succeeds).
Inside the lock it reaches adjust() -> allocate():
tryAllocate fails, pool is at its limit (1) FAIL
g2.isBlocking() == false (2) FAIL <--
post-barrier, never marked
registers a signal, parks (4) PARKED, HOLDING THE
LOCK
T5 W1 calls put(), tryLock fails, blocks at writeLock.lock()
TrieMemtable.java:562
W1 has NOT reached the allocator, so it registered NO signal.
T6 Flush A's run() executes on flushExecutor:
writeBarrier.markBlocking()
-> walks orderOnOrBefore(g1) and its prev chain
OpOrder.java:405-413
-> marks g1: isBlocking = true; blocking.forEach(signal) <-- list
is EMPTY
-> does NOT mark g2 (g2 is after the barrier)
writeBarrier.await() -> waits for g1 to close
T7 Nothing can move.
{noformat}
h2. 4. Why every rescue fails - and the symmetry is the point
||Thread||Would the escape hatch help?||Why it does not fire||
|*W2* - holds the lock, parked in the allocator|It is sitting exactly where the
hatch is checked|Its group {{g2}} is {*}post-barrier{*}, so {{markBlocking()}}
never marks it. {{isBlocking()}} stays false forever.|
|*W1* - queued on the lock|Its group {{g1}} *is* marked blocking, so the hatch
would let it straight through|It never reaches the allocator. It is parked on a
{{{}ReentrantLock{}}}, which {{markBlocking()}} cannot signal. Its signal list
is empty.|
Neither thread can be rescued, for *opposite* reasons: W1 qualifies for the
hatch but cannot reach it; W2 can reach it but does not qualify.
W2's wait is indefinite because a meaningful amount of pool memory is only
returned by {{SubAllocator.releaseAll()}} -> {{SubPool.released()}} ->
{{{}hasRoom.signalAll(){}}}, and {{releaseAll()}} runs from {{setDiscarded()}}
<- {{memtable.discard()}} <- {{Flush.reclaim}} - that is, only after some flush
has got *past* {{{}writeBarrier.await(){}}}.
{noformat}
Flush A ---awaits---> g1 ---held open by---> W1
^ |
| blocked on
| v
| MemtableShard.writeLock
| |
| held by
| v
+---- pool room <---only freed by---- W2 (parked in allocate())
(requires a COMPLETED flush)
{noformat}
h2. 5. Is it permanent?
Not automatically. A *later* flush's {{markBlocking()}} would cover {{g2}} -
any barrier issued after {{g2}} started has {{orderOnOrBefore >= g2}} - and
would release W2.
But {{markBlocking()}} lives inside {{{}Flush.run(){}}}, which executes on a
*static, node-wide* pool:
{code:java}
// ColumnFamilyStore.java:199-201
private static final ExecutorPlus flushExecutor =
DatabaseDescriptor.isDaemonInitialized()
? executorFactory().withJmxInternal().pooled("MemtableFlushWriter",
getFlushWriters())
: null;
{code}
sized by {{{}memtable_flush_writers{}}}:
{code:java}
// DatabaseDescriptor.java:929-931
if (conf.memtable_flush_writers == 0)
{
conf.memtable_flush_writers = conf.data_file_directories.length == 1 ? 2 :
1;
}
{code}
The stuck flush occupies one of those threads for the entire duration of
{{{}await(){}}}. So:
* *Multiple data directories* (default 1 writer): no thread is left to run any
other {{{}Flush.run(){}}}, so no later {{markBlocking()}} ever executes. The
deadlock is {*}unrecoverable{*}.
* *Single data directory* (default 2 writers): a second flush can run and
rescue W2 - unless the remaining slot is also consumed by a flush that gets
stuck the same way.
h2. 6. What the patch changes
The wait moves out of {{allocate()}} and up into
{{{}AbstractAllocatorMemtable.put{}}}, *before* {{performPut}} and therefore
before {{writeLock}} is ever touched:
{code:java}
// AbstractAllocatorMemtable.java:138-143
@Override
public final long put(PartitionUpdate update, UpdateTransaction indexer,
OpOrder.Group opGroup, boolean assumeMissing)
{
allocator.awaitRoomToStart(opGroup);
return performPut(update, indexer, opGroup, assumeMissing);
}
{code}
Step T4 becomes: W2 parks in {{awaitRoom}} {*}holding nothing{*}. The shard
lock stays free, W1 acquires it, completes, {{g1}} closes, and
{{writeBarrier.await()}} returns.
The set of locks held at the blocking point is now a strict subset of what it
was before, so the change cannot introduce a new lock-order cycle.
Nested writes (legacy 2i, which re-enters {{ColumnFamilyStore.apply}} from an
indexer callback *inside* the base memtable's lock) must skip the gate for the
same reason - hence {{{}putNested{}}}.
> Memtable allocator: separate memory usage tracking and limit checking
> ---------------------------------------------------------------------
>
> Key: CASSANDRA-21019
> URL: https://issues.apache.org/jira/browse/CASSANDRA-21019
> Project: Apache Cassandra
> Issue Type: Improvement
> Components: Local/Memtable
> Reporter: Dmitry Konstantinov
> Assignee: Michael Semb Wever
> Priority: High
> Fix For: 4.0.x, 4.1.x, 5.0.x, 7.x
>
> Attachments: ci_summary_thelastpickle_mck-21019-5.0_175.html,
> ci_summary_thelastpickle_mck-21019-6.0_101.html,
> ci_summary_thelastpickle_mck-21019-trunk_653.html,
> results_details_thelastpickle_mck-21019-5.0_175.tar.xz,
> results_details_thelastpickle_mck-21019-6.0_101.tar.xz,
> results_details_thelastpickle_mck-21019-trunk_653.tar.xz
>
> Time Spent: 4h
> Remaining Estimate: 0h
>
> The following optimization idea has been suggested by @blambov in
> CASSANDRA-20226:
> {quote}
> There's another option to consider here. The allocation mechanism does not
> need to check the limit for individual cell writes. We could just as well
> track the usage of a mutation in a single {{allocate}} call after it
> completes, or track the allocations with a {{LongAdder}} without checking if
> the limit is hit, and check if we need to wait for room before starting to
> apply a mutation.
> We use the {{allocate}} code to decide:
> - whether to initiate a flush, when the chosen memory limit is filled to
> some ratio
> - whether to pause accepting writes, when the chosen memory limit has been
> exhausted
> For the former use there is absolutely no benefit to make these decisions at
> the individual allocation level, as we will wait for the mutation to complete
> anyway before flushing anything. For the latter, I'd argue that the
> allocation-level tracking is actually hurting us. The reason for this is that
> we can have the limit be hit at any time during the application of a
> mutation, holding multiple locks (which necessitates the complexity of the
> {{isBlocking}} mutation signal), a partial copy of the mutation already
> written to the memtable structures, and a likely expanded version of the
> mutation to be applied on heap, keeping hold of more total memory than we
> would if we allowed the operation to continue.
> If, instead, we check the allocation limits _before starting_ a mutation and,
> once started, allow it to fully progress to completion, we can avoid this
> situation at the cost of being somewhat late to notice that the limit has
> been reached. This means that the limit will be breached, but this also
> happens as it stands now because we will permit operations to run to
> completion if the memtable they have been marked for is scheduled for a flush
> – which is effectively the same thing as not having noticed the memory limit
> would be breached by this mutation at the time when we decided to start it.
> {quote}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
