OpOrder.java
/*
* 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.cassandra.utils.concurrent;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
/**
* <p>A class for providing synchronization between producers and consumers that do not
* communicate directly with each other, but where the consumers need to process their
* work in contiguous batches. In particular this is useful for both CommitLog and Memtable
* where the producers (writing threads) are modifying a structure that the consumer
* (flush executor) only batch syncs, but needs to know what 'position' the work is at
* for co-ordination with other processes,
*
* <p>The typical usage is something like:
* <pre>
* {@code
public final class ExampleShared
{
final OpOrder order = new OpOrder();
volatile SharedState state;
static class SharedState
{
volatile Barrier barrier;
// ...
}
public void consume()
{
SharedState state = this.state;
state.setReplacement(new State())
state.doSomethingToPrepareForBarrier();
state.barrier = order.newBarrier();
// seal() MUST be called after newBarrier() else barrier.isAfter()
// will always return true, and barrier.await() will fail
state.barrier.issue();
// wait for all producer work started prior to the barrier to complete
state.barrier.await();
// change the shared state to its replacement, as the current state will no longer be used by producers
this.state = state.getReplacement();
state.doSomethingWithExclusiveAccess();
}
public void produce()
{
try (Group opGroup = order.start())
{
SharedState s = state;
while (s.barrier != null && !s.barrier.isAfter(opGroup))
s = s.getReplacement();
s.doProduceWork();
}
}
}
* }
* </pre>
*/
public class OpOrder
{
/**
* Constant that when an Ordered.running is equal to, indicates the Ordered is complete
*/
private static final int FINISHED = -1;
/**
* A linked list starting with the most recent Ordered object, i.e. the one we should start new operations from,
* with (prev) links to any incomplete Ordered instances, and (next) links to any potential future Ordered instances.
* Once all operations started against an Ordered instance and its ancestors have been finished the next instance
* will unlink this one
*/
private volatile Group current = new Group();
/**
* Start an operation against this OpOrder.
* Once the operation is completed Ordered.close() MUST be called EXACTLY once for this operation.
*
* @return the Ordered instance that manages this OpOrder
*/
public Group start()
{
while (true)
{
Group current = this.current;
if (current.register())
return current;
}
}
/**
* Creates a new barrier. The barrier is only a placeholder until barrier.issue() is called on it,
* after which all new operations will start against a new Group that will not be accepted
* by barrier.isAfter(), and barrier.await() will return only once all operations started prior to the issue
* have completed.
*/
public Barrier newBarrier()
{
return new Barrier();
}
public Group getCurrent()
{
return current;
}
public void awaitNewBarrier()
{
Barrier barrier = newBarrier();
barrier.issue();
barrier.await();
}
/**
* Represents a group of identically ordered operations, i.e. all operations started in the interval between
* two barrier issuances. For each register() call this is returned, close() must be called exactly once.
* It should be treated like taking a lock().
*/
public static final class Group implements Comparable<Group>, AutoCloseable
{
/**
* In general this class goes through the following stages:
* 1) LIVE: many calls to register() and close()
* 2) FINISHING: a call to expire() (after a barrier issue), means calls to register() will now fail,
* and we are now 'in the past' (new operations will be started against a new Ordered)
* 3) FINISHED: once the last close() is called, this Ordered is done. We call unlink().
* 4) ZOMBIE: all our operations are finished, but some operations against an earlier Ordered are still
* running, or tidying up, so unlink() fails to remove us
* 5) COMPLETE: all operations started on or before us are FINISHED (and COMPLETE), so we are unlinked
* <p/>
* Another parallel states is ISBLOCKING:
* <p/>
* isBlocking => a barrier that is waiting on us (either directly, or via a future Ordered) is blocking general
* progress. This state is entered by calling Barrier.markBlocking(). If the running operations are blocked
* on a Signal that is also registered with the isBlockingSignal (probably through isSafeBlockingSignal)
* then they will be notified that they are blocking forward progress, and may take action to avoid that.
*/
private volatile Group prev, next;
private final long id; // monotonically increasing id for compareTo()
private volatile int running = 0; // number of operations currently running. < 0 means we're expired, and the count of tasks still running is -(running + 1)
private volatile boolean isBlocking; // indicates running operations are blocking future barriers
private volatile ConcurrentLinkedQueue<WaitQueue.Signal> blocking; // signal to wait on to indicate isBlocking is true
private final WaitQueue waiting = newWaitQueue(); // signal to wait on for completion
static final AtomicIntegerFieldUpdater<Group> runningUpdater = AtomicIntegerFieldUpdater.newUpdater(Group.class, "running");
static final AtomicReferenceFieldUpdater<Group, ConcurrentLinkedQueue> blockingUpdater = AtomicReferenceFieldUpdater.newUpdater(Group.class, ConcurrentLinkedQueue.class, "blocking");
// constructs first instance only
private Group()
{
this.id = 0;
}
private Group(Group prev)
{
this.id = prev.id + 1;
this.prev = prev;
}
// prevents any further operations starting against this Ordered instance
// if there are no running operations, calls unlink; otherwise, we let the last op to close call it.
// this means issue() won't have to block for ops to finish.
private void expire()
{
while (true)
{
int current = running;
if (current < 0)
throw new IllegalStateException();
if (runningUpdater.compareAndSet(this, current, -1 - current))
{
// if we're already finished (no running ops), unlink ourselves
if (current == 0)
unlink();
return;
}
}
}
// attempts to start an operation against this Ordered instance, and returns true if successful.
private boolean register()
{
while (true)
{
int current = running;
if (current < 0)
return false;
if (runningUpdater.compareAndSet(this, current, current + 1))
return true;
}
}
/**
* To be called exactly once for each register() call this object is returned for, indicating the operation
* is complete
*/
public void close()
{
while (true)
{
int current = running;
if (current < 0)
{
if (runningUpdater.compareAndSet(this, current, current + 1))
{
if (current + 1 == FINISHED)
{
// if we're now finished, unlink ourselves
unlink();
}
return;
}
}
else if (runningUpdater.compareAndSet(this, current, current - 1))
{
return;
}
}
}
public boolean isFinished()
{
return next.prev == null;
}
public boolean isOldestLiveGroup()
{
return prev == null;
}
public void await()
{
while (!isFinished())
{
WaitQueue.Signal signal = waiting.register();
if (isFinished())
{
signal.cancel();
return;
}
else
signal.awaitUninterruptibly();
}
assert running == FINISHED;
}
public OpOrder.Group prev()
{
return prev;
}
/**
* called once we know all operations started against this Ordered have completed,
* however we do not know if operations against its ancestors have completed, or
* if its descendants have completed ahead of it, so we attempt to create the longest
* chain from the oldest still linked Ordered. If we can't reach the oldest through
* an unbroken chain of completed Ordered, we abort, and leave the still completing
* ancestor to tidy up.
*/
private void unlink()
{
// walk back in time to find the start of the list
Group start = this;
while (true)
{
Group prev = start.prev;
if (prev == null)
break;
// if we haven't finished this Ordered yet abort and let it clean up when it's done
if (prev.running != FINISHED)
return;
start = prev;
}
// now walk forwards in time, in case we finished up late
Group end = this.next;
while (end.running == FINISHED)
end = end.next;
// now walk from first to last, unlinking the prev pointer and waking up any blocking threads
while (start != end)
{
Group next = start.next;
next.prev = null;
start.waiting.signalAll();
start = next;
}
}
/**
* @return true if a barrier we are behind is, or may be, blocking general progress,
* so we should try more aggressively to progress
*/
public boolean isBlocking()
{
return isBlocking;
}
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);
}
public int compareTo(Group that)
{
// we deliberately use subtraction, as opposed to Long.compareTo() as we care about ordering
// not which is the smaller value, so this permits wrapping in the unlikely event we exhaust the long space
long c = this.id - that.id;
if (c > 0)
return 1;
else if (c < 0)
return -1;
else
return 0;
}
}
/**
* This class represents a synchronisation point providing ordering guarantees on operations started
* against the enclosing OpOrder. When issue() is called upon it (may only happen once per Barrier), the
* Barrier atomically partitions new operations from those already running (by expiring the current Group),
* and activates its isAfter() method
* which indicates if an operation was started before or after this partition. It offers methods to
* determine, or block until, all prior operations have finished, and a means to indicate to those operations
* that they are blocking forward progress. See {@link OpOrder} for idiomatic usage.
*/
public final class Barrier
{
// this Barrier was issued after all Group operations started against orderOnOrBefore
private volatile Group orderOnOrBefore;
/**
* @return true if @param group was started prior to the issuing of the barrier.
*
* (Until issue is called, always returns true, but if you rely on this behavior you are probably
* Doing It Wrong.)
*/
public boolean isAfter(Group group)
{
if (orderOnOrBefore == null)
return true;
// we subtract to permit wrapping round the full range of Long - so we only need to ensure
// there are never Long.MAX_VALUE * 2 total Group objects in existence at any one timem which will
// take care of itself
return orderOnOrBefore.id - group.id >= 0;
}
/**
* Issues (seals) the barrier, meaning no new operations may be issued against it, and expires the current
* Group. Must be called before await() for isAfter() to be properly synchronised.
*/
public void issue()
{
if (orderOnOrBefore != null)
throw new IllegalStateException("Can only call issue() once on each Barrier");
final Group current;
synchronized (OpOrder.this)
{
current = OpOrder.this.current;
orderOnOrBefore = current;
OpOrder.this.current = current.next = new Group(current);
}
current.expire();
}
/**
* Mark all prior operations as blocking, potentially signalling them to more aggressively make progress
*/
public void markBlocking()
{
Group current = orderOnOrBefore;
while (current != null)
{
current.markBlocking();
current = current.prev;
}
}
/**
* wait for all operations started prior to issuing the barrier to complete
*/
public void await()
{
Group current = orderOnOrBefore;
if (current == null)
throw new IllegalStateException("This barrier needs to have issue() called on it before prior operations can complete");
current.await();
}
/**
* returns the Group we are waiting on - any Group with {@code .compareTo(getSyncPoint()) <= 0}
* must complete before await() returns
*/
public Group getSyncPoint()
{
return orderOnOrBefore;
}
}
}