natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

More efficient executor service for fast operations 

Patch by Benedict Elliott Smith; reviewed by Jason Brown for CASSANDRA 4718
This commit is contained in:
belliottsmith 2014-05-29 22:16:03 +01:00
parent a8d18c9c19
commit 5420b7a229
27 changed files with 1575 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.0-rc1
* More efficient executor service for fast operations (CASSANDRA-4718)
* Move less common tools into a new cassandra-tools package (CASSANDRA-7160)
* Support more concurrent requests in native protocol (CASSANDRA-7231)
* Add tab-completion to debian nodetool packaging (CASSANDRA-6421)

227
src/java/org/apache/cassandra/concurrent/AbstractTracingAwareExecutorService.java Normal file
View File

@ -0,0 +1,227 @@
/*
* 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.concurrent;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static org.apache.cassandra.tracing.Tracing.isTracing;
public abstract class AbstractTracingAwareExecutorService implements TracingAwareExecutorService
{
private static final Logger logger = LoggerFactory.getLogger(AbstractTracingAwareExecutorService.class);
protected abstract void addTask(FutureTask<?> futureTask);
protected abstract void onCompletion();
/** Task Submission / Creation / Objects **/
public <T> FutureTask<T> submit(Callable<T> task)
{
return submit(newTaskFor(task));
}
public FutureTask<?> submit(Runnable task)
{
return submit(newTaskFor(task, null));
}
public <T> FutureTask<T> submit(Runnable task, T result)
{
return submit(newTaskFor(task, result));
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
{
throw new UnsupportedOperationException();
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException
{
throw new UnsupportedOperationException();
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException
{
throw new UnsupportedOperationException();
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
throw new UnsupportedOperationException();
}
protected <T> FutureTask<T> newTaskFor(Runnable runnable, T result)
{
return newTaskFor(runnable, result, null);
}
protected <T> FutureTask<T> newTaskFor(Runnable runnable, T result, TraceState traceState)
{
if (traceState != null || isTracing())
{
if (runnable instanceof TraceSessionFutureTask)
return (TraceSessionFutureTask<T>) runnable;
return new TraceSessionFutureTask<T>(runnable, result, traceState);
}
if (runnable instanceof FutureTask)
return (FutureTask<T>) runnable;
return new FutureTask<>(runnable, result);
}
protected <T> FutureTask<T> newTaskFor(Callable<T> callable)
{
if (isTracing())
{
if (callable instanceof TraceSessionFutureTask)
return (TraceSessionFutureTask<T>) callable;
return new TraceSessionFutureTask<T>(callable, null);
}
if (callable instanceof java.util.concurrent.FutureTask)
return (FutureTask<T>) callable;
return new FutureTask<>(callable);
}
private class TraceSessionFutureTask<T> extends FutureTask<T>
{
private final TraceState state;
public TraceSessionFutureTask(Callable<T> callable, TraceState state)
{
super(callable);
this.state = state;
}
public TraceSessionFutureTask(Runnable runnable, T result, TraceState state)
{
super(runnable, result);
this.state = state;
}
public void run()
{
Tracing.instance.set(state);
try
{
super.run();
}
finally
{
Tracing.instance.set(null);
}
}
}
class FutureTask<T> extends SimpleCondition implements Future<T>, Runnable
{
private boolean failure;
private Object result = this;
private final Callable<T> callable;
public FutureTask(Callable<T> callable)
{
this.callable = callable;
}
public FutureTask(Runnable runnable, T result)
{
this(Executors.callable(runnable, result));
}
public void run()
{
try
{
result = callable.call();
}
catch (Throwable t)
{
logger.warn("Uncaught exception on thread {}: {}", Thread.currentThread(), t);
result = t;
failure = true;
}
finally
{
signalAll();
onCompletion();
}
}
public boolean cancel(boolean mayInterruptIfRunning)
{
return false;
}
public boolean isCancelled()
{
return false;
}
public boolean isDone()
{
return isSignaled();
}
public T get() throws InterruptedException, ExecutionException
{
await();
Object result = this.result;
if (failure)
throw new ExecutionException((Throwable) result);
return (T) result;
}
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
await(timeout, unit);
Object result = this.result;
if (failure)
throw new ExecutionException((Throwable) result);
return (T) result;
}
}
private <T> FutureTask<T> submit(FutureTask<T> task)
{
addTask(task);
return task;
}
public void execute(Runnable command)
{
addTask(newTaskFor(command, null));
}
public void execute(Runnable command, TraceState state)
{
addTask(newTaskFor(command, null, state));
}
}

5
src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
View File

@ -138,6 +138,11 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor implements
: new TraceSessionWrapper<Object>(command, state));
}
public void maybeExecuteImmediately(Runnable command)
{
execute(command);
}
// execute does not call newTaskFor
@Override
public void execute(Runnable command)

114
src/java/org/apache/cassandra/concurrent/JMXEnabledSharedExecutorPool.java Normal file
View File

@ -0,0 +1,114 @@
/*
* 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.concurrent;
import java.lang.management.ManagementFactory;
import java.util.List;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.metrics.SEPMetrics;
public class JMXEnabledSharedExecutorPool extends SharedExecutorPool
{
public static final JMXEnabledSharedExecutorPool SHARED = new JMXEnabledSharedExecutorPool("SharedPool");
public JMXEnabledSharedExecutorPool(String poolName)
{
super(poolName);
}
public interface JMXEnabledSEPExecutorMBean extends JMXEnabledThreadPoolExecutorMBean
{
}
public class JMXEnabledSEPExecutor extends SEPExecutor implements JMXEnabledSEPExecutorMBean
{
private final SEPMetrics metrics;
private final String mbeanName;
public JMXEnabledSEPExecutor(int poolSize, int maxQueuedLength, String name, String jmxPath)
{
super(JMXEnabledSharedExecutorPool.this, poolSize, maxQueuedLength);
metrics = new SEPMetrics(this, jmxPath, name);
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbeanName = "org.apache.cassandra." + jmxPath + ":type=" + name;
try
{
mbs.registerMBean(this, new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
private void unregisterMBean()
{
try
{
ManagementFactory.getPlatformMBeanServer().unregisterMBean(new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
// release metrics
metrics.release();
}
@Override
public synchronized void shutdown()
{
// synchronized, because there is no way to access super.mainLock, which would be
// the preferred way to make this threadsafe
if (!isShutdown())
{
unregisterMBean();
}
super.shutdown();
}
public int getCoreThreads()
{
return 0;
}
public void setCoreThreads(int number)
{
throw new UnsupportedOperationException();
}
public void setMaximumThreads(int number)
{
throw new UnsupportedOperationException();
}
}
public TracingAwareExecutorService newExecutor(int maxConcurrency, int maxQueuedTasks, String name, String jmxPath)
{
JMXEnabledSEPExecutor executor = new JMXEnabledSEPExecutor(maxConcurrency, maxQueuedTasks, name, jmxPath);
executors.add(executor);
return executor;
}
}

274
src/java/org/apache/cassandra/concurrent/SEPExecutor.java Normal file
View File

@ -0,0 +1,274 @@
package org.apache.cassandra.concurrent;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
public class SEPExecutor extends AbstractTracingAwareExecutorService
{
private final SharedExecutorPool pool;
private final int maxWorkers;
private final int maxTasksQueued;
// stores both a set of work permits and task permits:
// bottom 32 bits are number of queued tasks, in the range [0..maxTasksQueued] (initially 0)
// top 32 bits are number of work permits available in the range [0..maxWorkers] (initially maxWorkers)
private final AtomicLong permits = new AtomicLong();
// producers wait on this when there is no room on the queue
private final WaitQueue hasRoom = new WaitQueue();
private final AtomicLong totalBlocked = new AtomicLong();
private final AtomicInteger currentlyBlocked = new AtomicInteger();
private final AtomicLong completedTasks = new AtomicLong();
volatile boolean shuttingDown = false;
final SimpleCondition shutdown = new SimpleCondition();
// TODO: see if other queue implementations might improve throughput
protected final ConcurrentLinkedQueue<FutureTask<?>> tasks = new ConcurrentLinkedQueue<>();
SEPExecutor(SharedExecutorPool pool, int maxWorkers, int maxTasksQueued)
{
this.pool = pool;
this.maxWorkers = maxWorkers;
this.maxTasksQueued = maxTasksQueued;
this.permits.set(combine(0, maxWorkers));
}
protected void onCompletion()
{
completedTasks.incrementAndGet();
}
// schedules another worker for this pool if there is work outstanding and there are no spinning threads that
// will self-assign to it in the immediate future
boolean maybeSchedule()
{
if (pool.spinningCount.get() > 0 || !takeWorkPermit(true))
return false;
pool.schedule(new Work(this));
return true;
}
protected void addTask(FutureTask<?> task)
{
// we add to the queue first, so that when a worker takes a task permit it can be certain there is a task available
// this permits us to schedule threads non-spuriously; it also means work is serviced fairly
tasks.add(task);
int taskPermits;
while (true)
{
long current = permits.get();
taskPermits = taskPermits(current);
// because there is no difference in practical terms between the work permit being added or not (the work is already in existence)
// we always add our permit, but block after the fact if we breached the queue limit
if (permits.compareAndSet(current, updateTaskPermits(current, taskPermits + 1)))
break;
}
if (taskPermits == 0)
{
// we only need to schedule a thread if there are no tasks already waiting to be processed, as
// the original enqueue will have started a thread to service its work which will have itself
// spawned helper workers that would have either exhausted the available tasks or are still being spawned.
// to avoid incurring any unnecessary signalling penalties we also do not take any work to hand to the new
// worker, we simply start a worker in a spinning state
pool.maybeStartSpinningWorker();
}
else if (taskPermits >= maxTasksQueued)
{
// register to receive a signal once a task is processed bringing the queue below its threshold
WaitQueue.Signal s = hasRoom.register();
// we will only be signalled once the queue drops below full, so this creates equivalent external behaviour
// however the advantage is that we never wake-up spuriously;
// we choose to always sleep, even if in the intervening time the queue has dropped below limit,
// so long as we _will_ eventually receive a signal
if (taskPermits(permits.get()) > maxTasksQueued)
{
// if we're blocking, we might as well directly schedule a worker if we aren't already at max
if (takeWorkPermit(true))
pool.schedule(new Work(this));
totalBlocked.incrementAndGet();
currentlyBlocked.incrementAndGet();
s.awaitUninterruptibly();
currentlyBlocked.decrementAndGet();
}
else // don't propagate our signal when we cancel, just cancel
s.cancel();
}
}
// takes permission to perform a task, if any are available; once taken it is guaranteed
// that a proceeding call to tasks.poll() will return some work
boolean takeTaskPermit()
{
while (true)
{
long current = permits.get();
int taskPermits = taskPermits(current);
if (taskPermits == 0)
return false;
if (permits.compareAndSet(current, updateTaskPermits(current, taskPermits - 1)))
{
if (taskPermits == maxTasksQueued && hasRoom.hasWaiters())
hasRoom.signalAll();
return true;
}
}
}
// takes a worker permit and (optionally) a task permit simultaneously; if one of the two is unavailable, returns false
boolean takeWorkPermit(boolean takeTaskPermit)
{
int taskDelta = takeTaskPermit ? 1 : 0;
while (true)
{
long current = permits.get();
int workPermits = workPermits(current);
int taskPermits = taskPermits(current);
if (workPermits == 0 || taskPermits == 0)
return false;
if (permits.compareAndSet(current, combine(taskPermits - taskDelta, workPermits - 1)))
{
if (takeTaskPermit && taskPermits == maxTasksQueued && hasRoom.hasWaiters())
hasRoom.signalAll();
return true;
}
}
}
// gives up a work permit
void returnWorkPermit()
{
while (true)
{
long current = permits.get();
int workPermits = workPermits(current);
if (permits.compareAndSet(current, updateWorkPermits(current, workPermits + 1)))
return;
}
}
public void maybeExecuteImmediately(Runnable command)
{
FutureTask<?> ft = newTaskFor(command, null);
if (!takeWorkPermit(false))
{
addTask(ft);
}
else
{
try
{
ft.run();
}
finally
{
returnWorkPermit();
// we have to maintain our invariant of always scheduling after any work is performed
// in this case in particular we are not processing the rest of the queue anyway, and so
// the work permit may go wasted if we don't immediately attempt to spawn another worker
maybeSchedule();
}
}
}
public synchronized void shutdown()
{
shuttingDown = true;
pool.executors.remove(this);
if (getActiveCount() == 0)
shutdown.signalAll();
}
public synchronized List<Runnable> shutdownNow()
{
shutdown();
List<Runnable> aborted = new ArrayList<>();
while (takeTaskPermit())
aborted.add(tasks.poll());
return aborted;
}
public boolean isShutdown()
{
return shuttingDown;
}
public boolean isTerminated()
{
return shuttingDown && shutdown.isSignaled();
}
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
{
shutdown.await(timeout, unit);
return isTerminated();
}
public long getPendingTasks()
{
return taskPermits(permits.get());
}
public long getCompletedTasks()
{
return completedTasks.get();
}
public int getActiveCount()
{
return maxWorkers - workPermits(permits.get());
}
public int getTotalBlockedTasks()
{
return (int) totalBlocked.get();
}
public int getMaximumThreads()
{
return maxWorkers;
}
public int getCurrentlyBlockedTasks()
{
return currentlyBlocked.get();
}
private static int taskPermits(long both)
{
return (int) both;
}
private static int workPermits(long both)
{
return (int) (both >>> 32);
}
private static long updateTaskPermits(long prev, int taskPermits)
{
return (prev & (-1L << 32)) | taskPermits;
}
private static long updateWorkPermits(long prev, int workPermits)
{
return (((long) workPermits) << 32) | (prev & (-1L >>> 32));
}
private static long combine(int taskPermits, int workPermits)
{
return (((long) workPermits) << 32) | taskPermits;
}
}

368
src/java/org/apache/cassandra/concurrent/SEPWorker.java Normal file
View File

@ -0,0 +1,368 @@
package org.apache.cassandra.concurrent;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnable
{
private static final Logger logger = LoggerFactory.getLogger(SEPWorker.class);
final Long workerId;
final Thread thread;
final SharedExecutorPool pool;
// prevStopCheck stores the value of pool.stopCheck after we last incremented it; if it hasn't changed,
// we know nobody else was spinning in the interval, so we increment our soleSpinnerSpinTime accordingly,
// and otherwise we set it to zero; this is then used to terminate the final spinning thread, as the coordinated
// strategy can only work when there are multiple threads spinning (as more sleep time must elapse than real time)
long prevStopCheck = 0;
long soleSpinnerSpinTime = 0;
SEPWorker(Long workerId, Work initialState, SharedExecutorPool pool)
{
this.pool = pool;
this.workerId = workerId;
thread = new Thread(this, pool.poolName + "-Worker-" + workerId);
thread.setDaemon(true);
set(initialState);
thread.start();
}
public void run()
{
/**
* we maintain two important invariants:
* 1) after exiting spinning phase, we ensure at least one more task on _each_ queue will be processed
* promptly after we begin, assuming any are outstanding on any pools. this is to permit producers to
* avoid signalling if there are _any_ spinning threads. we achieve this by simply calling maybeSchedule()
* on each queue if on decrementing the spin counter we hit zero.
* 2) before processing a task on a given queue, we attempt to assign another worker to the _same queue only_;
* this allows a producer to skip signalling work if the task queue is currently non-empty, and in conjunction
* with invariant (1) ensures that if any thread was spinning when a task was added to any executor, that
* task will be processed immediately if work permits are available
*/
SEPExecutor assigned = null;
Runnable task = null;
try
{
while (true)
{
if (isSpinning() && !selfAssign())
{
doWaitSpin();
continue;
}
// if stop was signalled, go to sleep (don't try self-assign; being put to sleep is rare, so let's obey it
// whenever we receive it - though we don't apply this constraint to producers, who may reschedule us before
// we go to sleep)
if (stop())
while (isStopped())
LockSupport.park();
// we can be assigned any state from STOPPED, so loop if we don't actually have any tasks assigned
assigned = get().assigned;
if (assigned == null)
continue;
task = assigned.tasks.poll();
// if we do have tasks assigned, nobody will change our state so we can simply set it to WORKING
// (which is also a state that will never be interrupted externally)
set(Work.WORKING);
boolean shutdown;
while (true)
{
// before we process any task, we maybe schedule a new worker _to our executor only_; this
// ensures that even once all spinning threads have found work, if more work is left to be serviced
// and permits are available, it will be dealt with immediately.
assigned.maybeSchedule();
// we know there is work waiting, as we have a work permit, so poll() will always succeed
task.run();
task = null;
// if we're shutting down, or we fail to take a permit, we don't perform any more work
if ((shutdown = assigned.shuttingDown) || !assigned.takeTaskPermit())
break;
task = assigned.tasks.poll();
}
// return our work permit, and maybe signal shutdown
assigned.returnWorkPermit();
if (shutdown && assigned.getActiveCount() == 0)
assigned.shutdown.signalAll();
assigned = null;
// try to immediately reassign ourselves some work; if we fail, start spinning
if (!selfAssign())
startSpinning();
}
}
catch (Throwable t)
{
while (true)
{
if (get().assigned != null)
{
assigned = get().assigned;
set(Work.WORKING);
}
if (assign(Work.STOPPED, true))
break;
}
if (assigned != null)
assigned.returnWorkPermit();
if (task != null)
logger.error("Failed to execute task, unexpected exception killed worker: {}", t);
else
logger.error("Unexpected exception killed worker: {}", t);
}
}
// try to assign this worker the provided work
// valid states to assign are SPINNING, STOP_SIGNALLED, (ASSIGNED);
// restores invariants of the various states (e.g. spinningCount, descheduled collection and thread park status)
boolean assign(Work work, boolean self)
{
Work state = get();
while (state.canAssign(self))
{
if (!compareAndSet(state, work))
{
state = get();
continue;
}
// if we were spinning, exit the state (decrement the count); this is valid even if we are already spinning,
// as the assigning thread will have incremented the spinningCount
if (state.isSpinning())
stopSpinning();
// if we're being descheduled, place ourselves in the descheduled collection
if (work.isStop())
pool.descheduled.put(workerId, this);
// if we're currently stopped, and the new state is not a stop signal
// (which we can immediately convert to stopped), unpark the worker
if (state.isStopped() && (!work.isStop() || !stop()))
LockSupport.unpark(thread);
return true;
}
return false;
}
// try to assign ourselves an executor with work available
private boolean selfAssign()
{
// if we aren't permitted to assign in this state, fail
if (!get().canAssign(true))
return false;
for (SEPExecutor exec : pool.executors)
{
if (exec.takeWorkPermit(true))
{
Work work = new Work(exec);
// we successfully started work on this executor, so we must either assign it to ourselves or ...
if (assign(work, true))
return true;
// ... if we fail, schedule it to another worker
pool.schedule(work);
// and return success as we must have already been assigned a task
assert get().assigned != null;
return true;
}
}
return false;
}
// we can only call this when our state is WORKING, and no other thread may change our state in this case;
// so in this case only we do not need to CAS. We increment the spinningCount and add ourselves to the spinning
// collection at the same time
private void startSpinning()
{
assert get() == Work.WORKING;
pool.spinningCount.incrementAndGet();
set(Work.SPINNING);
}
// exit the spinning state; if there are no remaining spinners, we immediately try and schedule work for all executors
// so that any producer is safe to not spin up a worker when they see a spinning thread (invariant (1) above)
private void stopSpinning()
{
if (pool.spinningCount.decrementAndGet() == 0)
for (SEPExecutor executor : pool.executors)
executor.maybeSchedule();
prevStopCheck = soleSpinnerSpinTime = 0;
}
// perform a sleep-spin, incrementing pool.stopCheck accordingly
private void doWaitSpin()
{
// pick a random sleep interval based on the number of threads spinning, so that
// we should always have a thread about to wake up, but most threads are sleeping
long sleep = 10000 * pool.spinningCount.get();
sleep = Math.min(1000000, sleep);
sleep *= Math.random();
sleep = Math.max(10000, sleep);
long start = System.nanoTime();
// place ourselves in the spinning collection; if we clash with another thread just exit
Long target = start + sleep;
if (pool.spinning.putIfAbsent(target, this) != null)
return;
LockSupport.parkNanos(sleep);
// remove ourselves (if haven't been already) - we should be at or near the front, so should be cheap-ish
pool.spinning.remove(target, this);
// finish timing and grab spinningTime (before we finish timing so it is under rather than overestimated)
long end = System.nanoTime();
long spin = end - start;
long stopCheck = pool.stopCheck.addAndGet(spin);
maybeStop(stopCheck, end);
if (prevStopCheck + spin == stopCheck)
soleSpinnerSpinTime += spin;
else
soleSpinnerSpinTime = 0;
prevStopCheck = stopCheck;
}
private static final long stopCheckInterval = TimeUnit.MILLISECONDS.toNanos(10L);
// stops a worker if elapsed real time is less than elapsed spin time, as this implies the equivalent of
// at least one worker achieved nothing in the interval. we achieve this by maintaining a stopCheck which
// is initialised to a negative offset from realtime; as we spin we add to this value, and if we ever exceed
// realtime we have spun too much and deschedule; if we get too far behind realtime, we reset to our initial offset
private void maybeStop(long stopCheck, long now)
{
long delta = now - stopCheck;
if (delta <= 0)
{
// if stopCheck has caught up with present, we've been spinning too much, so if we can atomically
// set it to the past again, we should stop a worker
if (pool.stopCheck.compareAndSet(stopCheck, now - stopCheckInterval))
{
// try and stop ourselves;
// if we've already been assigned work stop another worker
if (!assign(Work.STOP_SIGNALLED, true))
pool.schedule(Work.STOP_SIGNALLED);
}
}
else if (soleSpinnerSpinTime > stopCheckInterval && pool.spinningCount.get() == 1)
{
// permit self-stopping
assign(Work.STOP_SIGNALLED, true);
}
else
{
// if stop check has gotten too far behind present, update it so new spins can affect it
while (delta > stopCheckInterval * 2 && !pool.stopCheck.compareAndSet(stopCheck, now - stopCheckInterval))
{
stopCheck = pool.stopCheck.get();
delta = now - stopCheck;
}
}
}
private boolean isSpinning()
{
return get().isSpinning();
}
private boolean stop()
{
return get().isStop() && compareAndSet(Work.STOP_SIGNALLED, Work.STOPPED);
}
private boolean isStopped()
{
return get().isStopped();
}
/**
* Represents, and communicates changes to, a worker's work state - there are three non-actively-working
* states (STOP_SIGNALLED, STOPPED, AND SPINNING) and two working states: WORKING, and (ASSIGNED), the last
* being represented by a non-static instance with its "assigned" executor set.
*
* STOPPED: indicates the worker is descheduled, and whilst accepts work in this state (causing it to
* be rescheduled) it will generally not be considered for work until all other worker threads are busy.
* In this state we should be present in the pool.descheduled collection, and should be parked
* -> (ASSIGNED)|SPINNING
* STOP_SIGNALLED: the worker has been asked to deschedule itself, but has not yet done so; only entered from a SPINNING
* state, and generally communicated to itself, but maybe set from any worker. this state may be preempted
* and replaced with (ASSIGNED) or SPINNING
* In this state we should be present in the pool.descheduled collection
* -> (ASSIGNED)|STOPPED|SPINNING
* SPINNING: indicates the worker has no work to perform, so is performing a friendly wait-based-spinning
* until it either is (ASSIGNED) some work (by itself or another thread), or sent STOP_SIGNALLED
* In this state we _may_ be in the pool.spinning collection (but only if we are in the middle of a sleep)
* -> (ASSIGNED)|STOP_SIGNALLED|SPINNING
* (ASSIGNED): asks the worker to perform some work against the specified executor, and preassigns a task permit
* from that executor so that in this state there is always work to perform.
* In general a worker assigns itself this state, but sometimes it may assign another worker the state
* either if there is work outstanding and no-spinning threads, or there is a race to self-assign
* -> WORKING
* WORKING: indicates the worker is actively processing an executor's task queue; in this state it accepts
* no state changes/communications, except from itself; it usually exits this mode into SPINNING,
* but if work is immediately available on another executor it self-triggers (ASSIGNED)
* -> SPINNING|(ASSIGNED)
*/
static final class Work
{
static final Work STOP_SIGNALLED = new Work();
static final Work STOPPED = new Work();
static final Work SPINNING = new Work();
static final Work WORKING = new Work();
final SEPExecutor assigned;
Work(SEPExecutor executor)
{
this.assigned = executor;
}
private Work()
{
this.assigned = null;
}
boolean canAssign(boolean self)
{
// we can assign work if there isn't new work already assigned and either
// 1) we are assigning to ourselves
// 2) the worker we are assigning to is not already in the middle of WORKING
return assigned == null && (self || !isWorking());
}
boolean isSpinning()
{
return this == Work.SPINNING;
}
boolean isWorking()
{
return this == Work.WORKING;
}
boolean isStop()
{
return this == Work.STOP_SIGNALLED;
}
boolean isStopped()
{
return this == Work.STOPPED;
}
boolean isAssigned()
{
return assigned != null;
}
}
}

99
src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java Normal file
View File

@ -0,0 +1,99 @@
/*
* 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.concurrent;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
/**
* A pool of worker threads that are shared between all Executors created with it. Each executor is treated as a distinct
* unit, with its own concurrency and task queue limits, but the threads that service the tasks on each executor are
* free to hop between executors at will.
*
* To keep producers from incurring unnecessary delays, once an executor is "spun up" (i.e. is processing tasks at a steady
* rate), adding tasks to the executor often involves only placing the task on the work queue and updating the
* task permits (which imposes our max queue length constraints). Only when it cannot be guaranteed the task will be serviced
* promptly does the producer have to signal a thread itself to perform the work.
*
* We do this by scheduling only if
*
* The worker threads schedule themselves as far as possible: when they are assigned a task, they will attempt to spawn
* a partner worker to service any other work outstanding on the queue (if any); once they have finished the task they
* will either take another (if any remaining) and repeat this, or they will attempt to assign themselves to another executor
* that does have tasks remaining. If both fail, it will enter a non-busy-spinning phase, where it will sleep for a short
* random interval (based upon the number of threads in this mode, so that the total amount of non-sleeping time remains
* approximately fixed regardless of the number of spinning threads), and upon waking up will again try to assign themselves
* an executor with outstanding tasks to perform.
*/
public class SharedExecutorPool
{
// the name assigned to workers in the pool, and the id suffix
final String poolName;
final AtomicLong workerId = new AtomicLong();
// the collection of executors serviced by this pool; periodically ordered by traffic volume
final List<SEPExecutor> executors = new CopyOnWriteArrayList<>();
// the number of workers currently in a spinning state
final AtomicInteger spinningCount = new AtomicInteger();
// see SEPWorker.maybeStop() - used to self coordinate stopping of threads
final AtomicLong stopCheck = new AtomicLong();
// the collection of threads that are (most likely) in a spinning state - new workers are scheduled from here first
// TODO: consider using a queue partially-ordered by scheduled wake-up time
// (a full-fledged correctly ordered SkipList is overkill)
final ConcurrentSkipListMap<Long, SEPWorker> spinning = new ConcurrentSkipListMap<>();
// the collection of threads that have been asked to stop/deschedule - new workers are scheduled from here last
final ConcurrentSkipListMap<Long, SEPWorker> descheduled = new ConcurrentSkipListMap<>();
public SharedExecutorPool(String poolName)
{
this.poolName = poolName;
}
void schedule(Work work)
{
// we try to hand-off our work to the spinning queue before the descheduled queue, even though we expect it to be empty
// all we're doing here is hoping to find a worker without work to do, but it doesn't matter too much what we find;
// we atomically set the task so even if this were a collection of all workers it would be safe, and if they are both
// empty we schedule a new thread
Map.Entry<Long, SEPWorker> e;
while (null != (e = spinning.pollFirstEntry()) || null != (e = descheduled.pollFirstEntry()))
if (e.getValue().assign(work, false))
return;
if (!work.isStop())
new SEPWorker(workerId.incrementAndGet(), work, this);
}
void maybeStartSpinningWorker()
{
// in general the workers manage spinningCount directly; however if it is zero, we increment it atomically
// ourselves to avoid starting a worker unless we have to
int current = spinningCount.get();
if (current == 0 && spinningCount.compareAndSet(0, 1))
schedule(Work.SPINNING);
}
}

22
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -45,10 +45,10 @@ public class StageManager
static
{
stages.put(Stage.MUTATION, multiThreadedConfigurableStage(Stage.MUTATION, getConcurrentWriters()));
stages.put(Stage.COUNTER_MUTATION, multiThreadedConfigurableStage(Stage.COUNTER_MUTATION, getConcurrentCounterWriters()));
stages.put(Stage.READ, multiThreadedConfigurableStage(Stage.READ, getConcurrentReaders()));
stages.put(Stage.REQUEST_RESPONSE, multiThreadedStage(Stage.REQUEST_RESPONSE, FBUtilities.getAvailableProcessors()));
stages.put(Stage.MUTATION, multiThreadedLowSignalStage(Stage.MUTATION, getConcurrentWriters()));
stages.put(Stage.COUNTER_MUTATION, multiThreadedLowSignalStage(Stage.COUNTER_MUTATION, getConcurrentCounterWriters()));
stages.put(Stage.READ, multiThreadedLowSignalStage(Stage.READ, getConcurrentReaders()));
stages.put(Stage.REQUEST_RESPONSE, multiThreadedLowSignalStage(Stage.REQUEST_RESPONSE, FBUtilities.getAvailableProcessors()));
stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, FBUtilities.getAvailableProcessors()));
// the rest are all single-threaded
stages.put(Stage.GOSSIP, new JMXEnabledThreadPoolExecutor(Stage.GOSSIP));
@ -87,14 +87,9 @@ public class StageManager
stage.getJmxType());
}
private static JMXConfigurableThreadPoolExecutor multiThreadedConfigurableStage(Stage stage, int numThreads)
private static TracingAwareExecutorService multiThreadedLowSignalStage(Stage stage, int numThreads)
{
return new JMXConfigurableThreadPoolExecutor(numThreads,
KEEPALIVE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory(stage.getJmxName()),
stage.getJmxType());
return JMXEnabledSharedExecutorPool.SHARED.newExecutor(numThreads, Integer.MAX_VALUE, stage.getJmxName(), stage.getJmxType());
}
/**
@ -134,6 +129,11 @@ public class StageManager
super.execute(command);
}
public void maybeExecuteImmediately(Runnable command)
{
execute(command);
}
@Override
public Future<?> submit(Runnable task)
{

3
src/java/org/apache/cassandra/concurrent/TracingAwareExecutorService.java
View File

@ -30,4 +30,7 @@ public interface TracingAwareExecutorService extends ExecutorService
// we need a way to inject a TraceState directly into the Executor context without going through
// the global Tracing sessions; see CASSANDRA-5668
public void execute(Runnable command, TraceState state);
// permits executing in the context of the submitting thread
public void maybeExecuteImmediately(Runnable command);
}

98
src/java/org/apache/cassandra/metrics/SEPMetrics.java Normal file
View File

@ -0,0 +1,98 @@
/*
* 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.metrics;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import org.apache.cassandra.concurrent.SEPExecutor;
public class SEPMetrics
{
/** Number of active tasks. */
public final Gauge<Integer> activeTasks;
/** Number of tasks that had blocked before being accepted (or rejected). */
public final Gauge<Integer> totalBlocked;
/**
* Number of tasks currently blocked, waiting to be accepted by
* the executor (because all threads are busy and the backing queue is full).
*/
public final Gauge<Long> currentBlocked;
/** Number of completed tasks. */
public final Gauge<Long> completedTasks;
/** Number of tasks waiting to be executed. */
public final Gauge<Long> pendingTasks;
private MetricNameFactory factory;
/**
* Create metrics for the given LowSignalExecutor.
*
* @param executor Thread pool
* @param path Type of thread pool
* @param poolName Name of thread pool to identify metrics
*/
public SEPMetrics(final SEPExecutor executor, String path, String poolName)
{
this.factory = new ThreadPoolMetricNameFactory("ThreadPools", path, poolName);
activeTasks = Metrics.newGauge(factory.createMetricName("ActiveTasks"), new Gauge<Integer>()
{
public Integer value()
{
return executor.getActiveCount();
}
});
pendingTasks = Metrics.newGauge(factory.createMetricName("PendingTasks"), new Gauge<Long>()
{
public Long value()
{
return executor.getPendingTasks();
}
});
totalBlocked = Metrics.newGauge(factory.createMetricName("TotalBlockedTasks"), new Gauge<Integer>()
{
public Integer value()
{
return executor.getTotalBlockedTasks();
}
});
currentBlocked = Metrics.newGauge(factory.createMetricName("CurrentlyBlockedTasks"), new Gauge<Long>()
{
public Long value()
{
return (long) executor.getCurrentlyBlockedTasks();
}
});
completedTasks = Metrics.newGauge(factory.createMetricName("CompletedTasks"), new Gauge<Long>()
{
public Long value()
{
return executor.getCompletedTasks();
}
});
}
public void release()
{
Metrics.defaultRegistry().removeMetric(factory.createMetricName("ActiveTasks"));
Metrics.defaultRegistry().removeMetric(factory.createMetricName("PendingTasks"));
Metrics.defaultRegistry().removeMetric(factory.createMetricName("CompletedTasks"));
Metrics.defaultRegistry().removeMetric(factory.createMetricName("TotalBlockedTasks"));
Metrics.defaultRegistry().removeMetric(factory.createMetricName("CurrentlyBlockedTasks"));
}
}

47
src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.metrics;
import com.yammer.metrics.core.MetricName;
class ThreadPoolMetricNameFactory implements MetricNameFactory
{
private final String type;
private final String path;
private final String poolName;
ThreadPoolMetricNameFactory(String type, String path, String poolName)
{
this.type = type;
this.path = path;
this.poolName = poolName;
}
public MetricName createMetricName(String metricName)
{
String groupName = ThreadPoolMetrics.class.getPackage().getName();
StringBuilder mbeanName = new StringBuilder();
mbeanName.append(groupName).append(":");
mbeanName.append("type=").append(type);
mbeanName.append(",path=").append(path);
mbeanName.append(",scope=").append(poolName);
mbeanName.append(",name=").append(metricName);
return new MetricName(groupName, type, metricName, path + "." + poolName, mbeanName.toString());
}
}

28
src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java
View File

@ -52,7 +52,7 @@ public class ThreadPoolMetrics
*/
public ThreadPoolMetrics(final ThreadPoolExecutor executor, String path, String poolName)
{
this.factory = new ThreadPoolMetricNameFactory(path, poolName);
this.factory = new ThreadPoolMetricNameFactory("ThreadPools", path, poolName);
activeTasks = Metrics.newGauge(factory.createMetricName("ActiveTasks"), new Gauge<Integer>()
{
@ -87,30 +87,4 @@ public class ThreadPoolMetrics
Metrics.defaultRegistry().removeMetric(factory.createMetricName("TotalBlockedTasks"));
Metrics.defaultRegistry().removeMetric(factory.createMetricName("CurrentlyBlockedTasks"));
}
class ThreadPoolMetricNameFactory implements MetricNameFactory
{
private final String path;
private final String poolName;
ThreadPoolMetricNameFactory(String path, String poolName)
{
this.path = path;
this.poolName = poolName;
}
public MetricName createMetricName(String metricName)
{
String groupName = ThreadPoolMetrics.class.getPackage().getName();
String type = "ThreadPools";
StringBuilder mbeanName = new StringBuilder();
mbeanName.append(groupName).append(":");
mbeanName.append("type=").append(type);
mbeanName.append(",path=").append(path);
mbeanName.append(",scope=").append(poolName);
mbeanName.append(",name=").append(metricName);
return new MetricName(groupName, type, metricName, path + "." + poolName, mbeanName.toString());
}
}
}

1
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -64,6 +64,7 @@ import org.apache.cassandra.service.paxos.PrepareResponse;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
public final class MessagingService implements MessagingServiceMBean
{

2
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -33,7 +33,7 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationRequest;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
* RepairJob runs repair on given ColumnFamily.

1
src/java/org/apache/cassandra/repair/RepairSession.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.exceptions.RepairException;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
* Coordinates the (active) repair of a token range.

9
src/java/org/apache/cassandra/service/AbstractReadExecutor.java
View File

@ -77,12 +77,12 @@ public abstract class AbstractReadExecutor
protected void makeDataRequests(Iterable<InetAddress> endpoints)
{
boolean readLocal = false;
for (InetAddress endpoint : endpoints)
{
if (isLocalRequest(endpoint))
{
logger.trace("reading data locally");
StageManager.getStage(Stage.READ).execute(new LocalReadRunnable(command, handler));
readLocal = true;
}
else
{
@ -90,6 +90,11 @@ public abstract class AbstractReadExecutor
MessagingService.instance().sendRR(command.createMessage(), endpoint, handler);
}
}
if (readLocal)
{
logger.trace("reading data locally");
StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
}
}
protected void makeDigestRequests(Iterable<InetAddress> endpoints)

2
src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
View File

@ -30,7 +30,7 @@ import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.net.IAsyncCallback;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
public abstract class AbstractWriteResponseHandler implements IAsyncCallback
{

2
src/java/org/apache/cassandra/service/ReadCallback.java
View File

@ -41,7 +41,7 @@ import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessage>
{

8
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -780,6 +780,7 @@ public class StorageProxy implements StorageProxyMBean
// only need to create a Message for non-local writes
MessageOut<Mutation> message = null;
boolean insertLocal = false;
for (InetAddress destination : targets)
{
// avoid OOMing due to excess hints. we need to do this check even for "live" nodes, since we can
@ -797,7 +798,7 @@ public class StorageProxy implements StorageProxyMBean
{
if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
{
insertLocal(mutation, responseHandler);
insertLocal = true;
}
else
{
@ -835,6 +836,9 @@ public class StorageProxy implements StorageProxyMBean
}
}
if (insertLocal)
insertLocal(mutation, responseHandler);
if (dcGroups != null)
{
// for each datacenter, send the message to one node to relay the write to other replicas
@ -944,7 +948,7 @@ public class StorageProxy implements StorageProxyMBean
private static void insertLocal(final Mutation mutation, final AbstractWriteResponseHandler responseHandler)
{
StageManager.getStage(Stage.MUTATION).execute(new LocalMutationRunnable()
StageManager.getStage(Stage.MUTATION).maybeExecuteImmediately(new LocalMutationRunnable()
{
public void runMayThrow()
{

2
src/java/org/apache/cassandra/service/TruncateResponseHandler.java
View File

@ -27,7 +27,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.net.IAsyncCallback;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
public class TruncateResponseHandler implements IAsyncCallback
{

2
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -59,7 +59,7 @@ import org.apache.cassandra.service.*;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.streaming.StreamManagerMBean;
import org.apache.cassandra.streaming.management.StreamStateCompositeData;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
* JMX client operations for Cassandra.

2
src/java/org/apache/cassandra/transport/Message.java
View File

@ -335,7 +335,7 @@ public abstract class Message
}
void start()
{
if (running.compareAndSet(false, true))
if (!running.get() && running.compareAndSet(false, true))
{
this.eventLoop.execute(this);
}

19
src/java/org/apache/cassandra/transport/RequestThreadPoolExecutor.java
View File

@ -18,29 +18,24 @@
package org.apache.cassandra.transport;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import io.netty.util.concurrent.AbstractEventExecutor;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.TracingAwareExecutorService;
import org.apache.cassandra.config.DatabaseDescriptor;
import static org.apache.cassandra.concurrent.JMXEnabledSharedExecutorPool.SHARED;
public class RequestThreadPoolExecutor extends AbstractEventExecutor
{
private final static int CORE_THREAD_TIMEOUT_SEC = 30;
// Number of request we accept to queue before blocking. We could allow this to be configured...
private final static int MAX_QUEUED_REQUESTS = 128;
private final static String THREAD_FACTORY_ID = "Native-Transport-Requests";
private final JMXEnabledThreadPoolExecutor wrapped = new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
CORE_THREAD_TIMEOUT_SEC, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(MAX_QUEUED_REQUESTS),
new NamedThreadFactory(THREAD_FACTORY_ID),
"transport");
private final TracingAwareExecutorService wrapped = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
MAX_QUEUED_REQUESTS,
THREAD_FACTORY_ID,
"transport");
public boolean isShuttingDown()
{

55
src/java/org/apache/cassandra/utils/SimpleCondition.java → src/java/org/apache/cassandra/utils/concurrent/SimpleCondition.java
View File

@ -15,10 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.utils;
package org.apache.cassandra.utils.concurrent;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Condition;
// fulfils the Condition interface without spurious wakeup problems
@ -26,29 +27,40 @@ import java.util.concurrent.locks.Condition;
// _after_ signal(), it will work as desired.)
public class SimpleCondition implements Condition
{
private boolean set;
private static final AtomicReferenceFieldUpdater<SimpleCondition, WaitQueue> waitingUpdater = AtomicReferenceFieldUpdater.newUpdater(SimpleCondition.class, WaitQueue.class, "waiting");
public synchronized void await() throws InterruptedException
private volatile WaitQueue waiting;
private volatile boolean signaled = false;
public void await() throws InterruptedException
{
while (!set)
wait();
if (isSignaled())
return;
if (waiting == null)
waitingUpdater.compareAndSet(this, null, new WaitQueue());
WaitQueue.Signal s = waiting.register();
if (isSignaled())
s.cancel();
else
s.await();
assert isSignaled();
}
public synchronized void reset()
{
set = false;
}
public synchronized boolean await(long time, TimeUnit unit) throws InterruptedException
public boolean await(long time, TimeUnit unit) throws InterruptedException
{
if (isSignaled())
return true;
long start = System.nanoTime();
long timeout = unit.toNanos(time);
long elapsed;
while (!set && (elapsed = System.nanoTime() - start) < timeout)
long until = start + unit.toNanos(time);
if (waiting == null)
waitingUpdater.compareAndSet(this, null, new WaitQueue());
WaitQueue.Signal s = waiting.register();
if (isSignaled())
{
TimeUnit.NANOSECONDS.timedWait(this, timeout - elapsed);
s.cancel();
return true;
}
return set;
return s.awaitUntil(until) || isSignaled();
}
public void signal()
@ -56,15 +68,16 @@ public class SimpleCondition implements Condition
throw new UnsupportedOperationException();
}
public synchronized void signalAll()
public boolean isSignaled()
{
set = true;
notifyAll();
return signaled;
}
public synchronized boolean isSignaled()
public void signalAll()
{
return set;
signaled = true;
if (waiting != null)
waiting.signalAll();
}
public void awaitUninterruptibly()

72
src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java
View File

@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.LockSupport;
@ -73,8 +74,6 @@ import java.util.concurrent.locks.LockSupport;
public final class WaitQueue
{
private static final Logger logger = LoggerFactory.getLogger(WaitQueue.class);
private static final int CANCELLED = -1;
private static final int SIGNALLED = 1;
private static final int NOT_SET = 0;
@ -82,11 +81,11 @@ public final class WaitQueue
private static final AtomicIntegerFieldUpdater signalledUpdater = AtomicIntegerFieldUpdater.newUpdater(RegisteredSignal.class, "state");
// the waiting signals
private final ConcurrentLinkedDeque<RegisteredSignal> queue = new ConcurrentLinkedDeque<>();
private final ConcurrentLinkedQueue<RegisteredSignal> queue = new ConcurrentLinkedQueue<>();
/**
* The calling thread MUST be the thread that uses the signal
* @return
* @return x
*/
public Signal register()
{
@ -119,7 +118,7 @@ public final class WaitQueue
while (true)
{
RegisteredSignal s = queue.poll();
if (s == null || s.signal())
if (s == null || s.signal() != null)
return s != null;
}
}
@ -129,42 +128,40 @@ public final class WaitQueue
*/
public void signalAll()
{
RegisteredSignal last = queue.peekLast();
if (last == null)
if (!hasWaiters())
return;
List<Thread> woke = null;
if (logger.isTraceEnabled())
woke = new ArrayList<>();
long start = System.nanoTime();
// we wake up only a snapshot of the queue, to avoid a race where the condition is not met and the woken thread
// immediately waits on the queue again
// to avoid a race where the condition is not met and the woken thread managed to wait on the queue before
// we finish signalling it all, we pick a random thread we have woken-up and hold onto it, so that if we encounter
// it again we know we're looping. We reselect a random thread periodically, progressively less often.
// the "correct" solution to this problem is to use a queue that permits snapshot iteration, but this solution is sufficient
int i = 0, s = 5;
Thread randomThread = null;
Iterator<RegisteredSignal> iter = queue.iterator();
while (iter.hasNext())
{
RegisteredSignal signal = iter.next();
if (logger.isTraceEnabled())
Thread signalled = signal.signal();
if (signalled != null)
{
Thread thread = signal.thread;
if (signal.signal())
woke.add(thread);
if (signalled == randomThread)
break;
if (++i == s)
{
randomThread = signalled;
s <<= 1;
}
}
else
signal.signal();
iter.remove();
if (signal == last)
break;
}
long end = System.nanoTime();
if (woke != null)
logger.trace("Woke up {} in {}ms from {}", woke, (end - start) * 0.000001d, Thread.currentThread().getStackTrace()[2]);
}
private void cleanUpCancelled()
{
// attempt to remove the cancelled from the beginning only, but if we fail to remove any proceed to cover
// the whole list
// TODO: attempt to remove the cancelled from the beginning only (need atomic cas of head)
Iterator<RegisteredSignal> iter = queue.iterator();
while (iter.hasNext())
{
@ -185,7 +182,7 @@ public final class WaitQueue
*/
public int getWaiting()
{
if (queue.isEmpty())
if (!hasWaiters())
return 0;
Iterator<RegisteredSignal> iter = queue.iterator();
int count = 0;
@ -264,11 +261,11 @@ public final class WaitQueue
* isSignalled() will be true on exit, and the method will return true; if timedout, the method will return
* false and isCancelled() will be true; if interrupted an InterruptedException will be thrown and isCancelled()
* will be true.
* @param until System.currentTimeMillis() to wait until
* @param nanos System.nanoTime() to wait until
* @return true if signalled, false if timed out
* @throws InterruptedException
*/
public boolean awaitUntil(long until) throws InterruptedException;
public boolean awaitUntil(long nanos) throws InterruptedException;
}
/**
@ -302,10 +299,12 @@ public final class WaitQueue
public boolean awaitUntil(long until) throws InterruptedException
{
while (until < System.currentTimeMillis() && !isSignalled())
long now;
while (until > (now = System.nanoTime()) && !isSignalled())
{
checkInterrupted();
LockSupport.parkUntil(until);
long delta = until - now;
LockSupport.parkNanos(delta);
}
return checkAndClear();
}
@ -343,15 +342,16 @@ public final class WaitQueue
return state != NOT_SET;
}
private boolean signal()
private Thread signal()
{
if (!isSet() && signalledUpdater.compareAndSet(this, NOT_SET, SIGNALLED))
{
Thread thread = this.thread;
LockSupport.unpark(thread);
thread = null;
return true;
this.thread = null;
return thread;
}
return false;
return null;
}
public boolean checkAndClear()

228
test/long/org/apache/cassandra/concurrent/LongSharedExecutorPoolTest.java Normal file
View File

@ -0,0 +1,228 @@
/*
* 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.concurrent;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.LockSupport;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.math3.distribution.WeibullDistribution;
import org.junit.Test;
public class LongSharedExecutorPoolTest
{
private static final class WaitTask implements Runnable
{
final long nanos;
private WaitTask(long nanos)
{
this.nanos = nanos;
}
public void run()
{
LockSupport.parkNanos(nanos);
}
}
private static final class Result implements Comparable<Result>
{
final Future<?> future;
final long forecastedCompletion;
private Result(Future<?> future, long forecastedCompletion)
{
this.future = future;
this.forecastedCompletion = forecastedCompletion;
}
public int compareTo(Result that)
{
int c = Long.compare(this.forecastedCompletion, that.forecastedCompletion);
if (c != 0)
return c;
c = Integer.compare(this.hashCode(), that.hashCode());
if (c != 0)
return c;
return Integer.compare(this.future.hashCode(), that.future.hashCode());
}
}
private static final class Batch implements Comparable<Batch>
{
final TreeSet<Result> results;
final long timeout;
final int executorIndex;
private Batch(TreeSet<Result> results, long timeout, int executorIndex)
{
this.results = results;
this.timeout = timeout;
this.executorIndex = executorIndex;
}
public int compareTo(Batch that)
{
int c = Long.compare(this.timeout, that.timeout);
if (c != 0)
return c;
c = Integer.compare(this.results.size(), that.results.size());
if (c != 0)
return c;
return Integer.compare(this.hashCode(), that.hashCode());
}
}
@Test
public void testPromptnessOfExecution() throws InterruptedException, ExecutionException, TimeoutException
{
testPromptnessOfExecution(TimeUnit.MINUTES.toNanos(2L), 0.5f);
}
private void testPromptnessOfExecution(long intervalNanos, float loadIncrement) throws InterruptedException, ExecutionException, TimeoutException
{
final int executorCount = 4;
int threadCount = 8;
int maxQueued = 1024;
final WeibullDistribution workTime = new WeibullDistribution(3, 200000);
final long minWorkTime = TimeUnit.MICROSECONDS.toNanos(1);
final long maxWorkTime = TimeUnit.MILLISECONDS.toNanos(1);
final int[] threadCounts = new int[executorCount];
final WeibullDistribution[] workCount = new WeibullDistribution[executorCount];
final ExecutorService[] executors = new ExecutorService[executorCount];
for (int i = 0 ; i < executors.length ; i++)
{
executors[i] = JMXEnabledSharedExecutorPool.SHARED.newExecutor(threadCount, maxQueued, "test" + i, "test" + i);
threadCounts[i] = threadCount;
workCount[i] = new WeibullDistribution(2, maxQueued);
threadCount *= 2;
maxQueued *= 2;
}
long runs = 0;
long events = 0;
final TreeSet<Batch> pending = new TreeSet<>();
final BitSet executorsWithWork = new BitSet(executorCount);
long until = 0;
// basic idea is to go through different levels of load on the executor service; initially is all small batches
// (mostly within max queue size) of very short operations, moving to progressively larger batches
// (beyond max queued size), and longer operations
for (float multiplier = 0f ; multiplier < 2.01f ; )
{
if (System.nanoTime() > until)
{
System.out.println(String.format("Completed %.0fK batches with %.1fM events", runs * 0.001f, events * 0.000001f));
events = 0;
until = System.nanoTime() + intervalNanos;
multiplier += loadIncrement;
System.out.println(String.format("Running for %ds with load multiplier %.1f", TimeUnit.NANOSECONDS.toSeconds(intervalNanos), multiplier));
}
// wait a random amount of time so we submit new tasks in various stages of
long timeout;
if (pending.isEmpty()) timeout = 0;
else if (Math.random() > 0.98) timeout = Long.MAX_VALUE;
else if (pending.size() == executorCount) timeout = pending.first().timeout;
else timeout = (long) (Math.random() * pending.last().timeout);
while (!pending.isEmpty() && timeout > System.nanoTime())
{
Batch first = pending.first();
boolean complete = false;
try
{
for (Result result : first.results.descendingSet())
result.future.get(timeout - System.nanoTime(), TimeUnit.NANOSECONDS);
complete = true;
}
catch (TimeoutException e)
{
}
if (!complete && System.nanoTime() > first.timeout)
{
for (Result result : first.results)
if (!result.future.isDone())
throw new AssertionError();
complete = true;
}
if (complete)
{
pending.pollFirst();
executorsWithWork.clear(first.executorIndex);
}
}
// if we've emptied the executors, give all our threads an opportunity to spin down
if (timeout == Long.MAX_VALUE)
Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
// submit a random batch to the first free executor service
int executorIndex = executorsWithWork.nextClearBit(0);
if (executorIndex >= executorCount)
continue;
executorsWithWork.set(executorIndex);
ExecutorService executor = executors[executorIndex];
TreeSet<Result> results = new TreeSet<>();
int count = (int) (workCount[executorIndex].sample() * multiplier);
long targetTotalElapsed = 0;
long start = System.nanoTime();
long baseTime;
if (Math.random() > 0.5) baseTime = 2 * (long) (workTime.sample() * multiplier);
else baseTime = 0;
for (int j = 0 ; j < count ; j++)
{
long time;
if (baseTime == 0) time = (long) (workTime.sample() * multiplier);
else time = (long) (baseTime * Math.random());
if (time < minWorkTime)
time = minWorkTime;
if (time > maxWorkTime)
time = maxWorkTime;
targetTotalElapsed += time;
Future<?> future = executor.submit(new WaitTask(time));
results.add(new Result(future, System.nanoTime() + time));
}
long end = start + (long) Math.ceil(targetTotalElapsed / (double) threadCounts[executorIndex])
+ TimeUnit.MILLISECONDS.toNanos(100L);
long now = System.nanoTime();
if (runs++ > executorCount && now > end)
throw new AssertionError();
events += results.size();
pending.add(new Batch(results, end, executorIndex));
// System.out.println(String.format("Submitted batch to executor %d with %d items and %d permitted millis", executorIndex, count, TimeUnit.NANOSECONDS.toMillis(end - start)));
}
}
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException
{
// do longer test
new LongSharedExecutorPoolTest().testPromptnessOfExecution(TimeUnit.MINUTES.toNanos(10L), 0.1f);
}
}

2
test/unit/org/apache/cassandra/repair/ValidatorTest.java
View File

@ -46,7 +46,7 @@ import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.ValidationComplete;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.SimpleCondition;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static org.junit.Assert.*;