honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

work on scaling and blocking thread pool

This commit is contained in:
kimchy 2010-06-29 17:50:05 +03:00
parent 297001de53
commit 9051ae3a65
8 changed files with 311 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

199
modules/elasticsearch/src/main/java/org/elasticsearch/common/util/concurrent/ScalingThreadPoolExecutor.java → modules/elasticsearch/src/main/java/org/elasticsearch/common/util/concurrent/TransferThreadPoolExecutor.java
View File

@ -19,11 +19,14 @@
package org.elasticsearch.common.util.concurrent;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.jsr166y.LinkedTransferQueue;
import org.elasticsearch.common.util.concurrent.jsr166y.TransferQueue;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@ -31,9 +34,9 @@ import java.util.concurrent.locks.ReentrantLock;
/**
* @author kimchy (shay.banon)
*/
public class ScalingThreadPoolExecutor extends AbstractExecutorService {
public class TransferThreadPoolExecutor extends AbstractExecutorService {
private final BlockingQueue<Runnable> workQueue = new LinkedTransferQueue<Runnable>();
private final TransferQueue<Runnable> workQueue = new LinkedTransferQueue<Runnable>();
private final AtomicInteger queueSize = new AtomicInteger();
@ -98,6 +101,12 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
static final int TERMINATED = 3;
private final boolean blocking;
private final int blockingCapacity;
private final long blockingTime;
/**
* Core pool size, updated only while holding mainLock, but
* volatile to allow concurrent readability even during updates.
@ -122,39 +131,130 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
* Current pool size, updated only while holding mainLock but
* volatile to allow concurrent readability even during updates.
*/
private volatile int poolSize;
private final AtomicInteger poolSize = new AtomicInteger();
public static TransferThreadPoolExecutor newScalingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {
return new TransferThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, false, 0, TimeUnit.NANOSECONDS, 0, threadFactory);
}
private final ScheduledFuture scheduledFuture;
public static TransferThreadPoolExecutor newBlockingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
long blockingTime, TimeUnit blockingUnit, int blockingCapacity,
ThreadFactory threadFactory) {
return new TransferThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, true, blockingTime, blockingUnit, blockingCapacity, threadFactory);
}
public ScalingThreadPoolExecutor(int corePoolSize, int maximumPoolSize, TimeValue keepAlive,
ThreadFactory threadFactory,
ScheduledExecutorService scheduler, TimeValue schedulerInterval) {
private TransferThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
boolean blocking, long blockingTime, TimeUnit blockingUnit, int blockingCapacity,
ThreadFactory threadFactory) {
this.blocking = blocking;
this.blockingTime = blockingUnit.toNanos(blockingTime);
this.blockingCapacity = blockingCapacity;
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.keepAliveTime = keepAlive.nanos();
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
for (int i = 0; i < corePoolSize; i++) {
Thread t = addThread();
if (t != null)
Thread t = addWorker();
if (t != null) {
poolSize.incrementAndGet();
t.start();
}
}
this.scheduledFuture = scheduler.scheduleWithFixedDelay(new Scheduler(), schedulerInterval.nanos(),
schedulerInterval.nanos(), TimeUnit.NANOSECONDS);
}
@Override public void execute(Runnable command) {
queueSize.incrementAndGet();
workQueue.add(command);
if (blocking) {
executeBlocking(command);
} else {
executeNonBlocking(command);
}
}
private void executeNonBlocking(Runnable command) {
// note, there might be starvation of some commands that were added to the queue,
// while others are being transferred directly
boolean succeeded = workQueue.tryTransfer(command);
if (succeeded) {
return;
}
int currentPoolSize = poolSize.get();
if (currentPoolSize < maximumPoolSize) {
// if we manage to add a worker, add it, and tryTransfer again
if (poolSize.compareAndSet(currentPoolSize, currentPoolSize + 1)) {
Thread t = addWorker();
if (t == null) {
poolSize.decrementAndGet();
workQueue.add(command);
} else {
t.start();
succeeded = workQueue.tryTransfer(command);
if (!succeeded) {
workQueue.add(command);
}
}
} else {
succeeded = workQueue.tryTransfer(command);
if (!succeeded) {
workQueue.add(command);
}
}
} else {
workQueue.add(command);
}
}
private void executeBlocking(Runnable command) {
int currentCapacity = queueSize.getAndIncrement();
boolean succeeded = workQueue.tryTransfer(command);
if (succeeded) {
return;
}
int currentPoolSize = poolSize.get();
if (currentPoolSize < maximumPoolSize) {
// if we manage to add a worker, add it, and tryTransfer again
if (poolSize.compareAndSet(currentPoolSize, currentPoolSize + 1)) {
Thread t = addWorker();
if (t == null) {
poolSize.decrementAndGet();
workQueue.add(command);
} else {
t.start();
succeeded = workQueue.tryTransfer(command);
if (!succeeded) {
transferOrAddBlocking(command, currentCapacity);
}
}
} else {
succeeded = workQueue.tryTransfer(command);
if (!succeeded) {
transferOrAddBlocking(command, currentCapacity);
}
}
} else {
transferOrAddBlocking(command, currentCapacity);
}
}
private void transferOrAddBlocking(Runnable command, int currentCapacity) {
if (currentCapacity < blockingCapacity) {
workQueue.add(command);
} else {
boolean succeeded;
try {
succeeded = workQueue.tryTransfer(command, blockingTime, TimeUnit.NANOSECONDS);
if (!succeeded) {
throw new RejectedExecutionException("Rejected execution after waiting "
+ TimeUnit.NANOSECONDS.toSeconds(blockingTime) + "ms for task [" + command.getClass() + "] to be executed.");
}
} catch (InterruptedException e) {
throw new RejectedExecutionException(e);
}
}
}
@Override public void shutdown() {
if (!scheduledFuture.isCancelled()) {
scheduledFuture.cancel(false);
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
@ -179,9 +279,6 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
}
@Override public List<Runnable> shutdownNow() {
if (!scheduledFuture.isCancelled()) {
scheduledFuture.cancel(false);
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
@ -238,7 +335,7 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
* @return the number of threads
*/
public int getPoolSize() {
return poolSize;
return poolSize.get();
}
/**
@ -262,31 +359,17 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
}
}
private final class Scheduler implements Runnable {
@Override public void run() {
if (queueSize.get() > 0 && poolSize < maximumPoolSize) {
final ReentrantLock mainLock = ScalingThreadPoolExecutor.this.mainLock;
mainLock.lock();
try {
int currentQueueSize = queueSize.get();
if (currentQueueSize > 0 && poolSize < maximumPoolSize) {
int incrementBy = currentQueueSize;
if (poolSize + incrementBy > maximumPoolSize) {
incrementBy = maximumPoolSize - poolSize;
}
for (int i = 0; i < incrementBy; i++) {
Thread t = addThread();
if (t != null)
t.start();
}
}
} finally {
mainLock.unlock();
}
}
}
public int getCorePoolSize() {
return corePoolSize;
}
public int getMaximumPoolSize() {
return maximumPoolSize;
}
public int getQueueSize() {
return queueSize.get();
}
private final class Worker implements Runnable {
/**
@ -380,7 +463,7 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
Runnable r;
if (state == SHUTDOWN) // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize)
else if (poolSize.get() > corePoolSize)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
@ -445,7 +528,7 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
mainLock.lock();
try {
workers.remove(w);
if (--poolSize == 0)
if (poolSize.decrementAndGet() == 0)
tryTerminate();
} finally {
mainLock.unlock();
@ -464,11 +547,12 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
* shutdown or shutdownNow, if there are no live threads.
*/
private void tryTerminate() {
if (poolSize == 0) {
if (poolSize.get() == 0) {
int state = runState;
if (state < STOP && queueSize.get() > 0) {
state = RUNNING; // disable termination check below
Thread t = addThread();
poolSize.incrementAndGet();
if (t != null)
t.start();
}
@ -479,6 +563,20 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
}
}
/**
* Creates and returns a new thread running firstTask as its first
* task. Executed under mainLock.
*/
private Thread addWorker() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return addThread();
} finally {
mainLock.unlock();
}
}
/**
* Creates and returns a new thread running firstTask as its first
* task. Call only while holding mainLock.
@ -489,7 +587,6 @@ public class ScalingThreadPoolExecutor extends AbstractExecutorService {
if (t != null) {
w.thread = t;
workers.add(w);
++poolSize;
}
return t;
}

4
modules/elasticsearch/src/main/java/org/elasticsearch/threadpool/blocking/BlockingThreadPool.java
View File

@ -25,10 +25,12 @@ import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.DynamicExecutors;
import org.elasticsearch.common.util.concurrent.TransferThreadPoolExecutor;
import org.elasticsearch.threadpool.support.AbstractThreadPool;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static org.elasticsearch.common.settings.ImmutableSettings.Builder.*;
import static org.elasticsearch.common.unit.TimeValue.*;
@ -61,7 +63,7 @@ public class BlockingThreadPool extends AbstractThreadPool {
this.waitTime = componentSettings.getAsTime("wait_time", timeValueSeconds(60));
this.keepAlive = componentSettings.getAsTime("keep_alive", timeValueSeconds(60));
logger.debug("Initializing {} thread pool with min[{}], max[{}], keep_alive[{}], capacity[{}], wait_time[{}], scheduled_size[{}]", getType(), min, max, keepAlive, capacity, waitTime, scheduledSize);
executorService = DynamicExecutors.newBlockingThreadPool(min, max, keepAlive.millis(), capacity, waitTime.millis(), DynamicExecutors.daemonThreadFactory(settings, "[tp]"));
executorService = TransferThreadPoolExecutor.newBlockingExecutor(min, max, keepAlive.millis(), TimeUnit.MILLISECONDS, waitTime.millis(), TimeUnit.MILLISECONDS, capacity, DynamicExecutors.daemonThreadFactory(settings, "[tp]"));
scheduledExecutorService = Executors.newScheduledThreadPool(scheduledSize, DynamicExecutors.daemonThreadFactory(settings, "[sc]"));
started = true;
}

14
modules/elasticsearch/src/main/java/org/elasticsearch/threadpool/scaling/ScalingThreadPool.java
View File

@ -23,24 +23,24 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.DynamicExecutors;
import org.elasticsearch.common.util.concurrent.ScalingThreadPoolExecutor;
import org.elasticsearch.common.util.concurrent.TransferThreadPoolExecutor;
import org.elasticsearch.threadpool.support.AbstractThreadPool;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static org.elasticsearch.common.settings.ImmutableSettings.Builder.*;
import static org.elasticsearch.common.unit.TimeValue.*;
/**
* @author kimchy (Shay Banon)
* @author kimchy (shay.banon)
*/
public class ScalingThreadPool extends AbstractThreadPool {
final int min;
final int max;
final TimeValue keepAlive;
final TimeValue interval;
final int scheduledSize;
@ -53,12 +53,10 @@ public class ScalingThreadPool extends AbstractThreadPool {
this.min = componentSettings.getAsInt("min", 10);
this.max = componentSettings.getAsInt("max", 100);
this.keepAlive = componentSettings.getAsTime("keep_alive", timeValueSeconds(60));
this.interval = componentSettings.getAsTime("interval", timeValueSeconds(5));
this.scheduledSize = componentSettings.getAsInt("scheduled_size", 20);
logger.debug("Initializing {} thread pool with min[{}], max[{}], keep_alive[{}], scheduled_size[{}]", getType(), min, max, keepAlive, scheduledSize);
scheduledExecutorService = Executors.newScheduledThreadPool(scheduledSize, DynamicExecutors.daemonThreadFactory(settings, "[sc]"));
executorService = new ScalingThreadPoolExecutor(min, max, keepAlive, DynamicExecutors.daemonThreadFactory(settings, "[tp]"), scheduledExecutorService,
interval);
executorService = TransferThreadPoolExecutor.newScalingExecutor(min, max, keepAlive.nanos(), TimeUnit.NANOSECONDS, DynamicExecutors.daemonThreadFactory(settings, "[tp]"));
started = true;
}
@ -75,11 +73,11 @@ public class ScalingThreadPool extends AbstractThreadPool {
}
@Override public int getPoolSize() {
return ((ScalingThreadPoolExecutor) executorService).getPoolSize();
return ((TransferThreadPoolExecutor) executorService).getPoolSize();
}
@Override public int getActiveCount() {
return ((ScalingThreadPoolExecutor) executorService).getActiveCount();
return ((TransferThreadPoolExecutor) executorService).getActiveCount();
}
@Override public int getSchedulerPoolSize() {

15
modules/elasticsearch/src/test/java/org/elasticsearch/util/concurrent/BlockingThreadPoolTest.java → modules/elasticsearch/src/test/java/org/elasticsearch/common/util/concurrent/BlockingThreadPoolTest.java
View File

@ -17,20 +17,19 @@
* under the License.
*/
package org.elasticsearch.util.concurrent;
package org.elasticsearch.common.util.concurrent;
import org.elasticsearch.common.util.concurrent.DynamicExecutors;
import org.elasticsearch.common.util.concurrent.ThreadBarrier;
import org.testng.annotations.Test;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static org.hamcrest.MatcherAssert.*;
import static org.hamcrest.Matchers.*;
/**
* @author kimchy (Shay Banon)
* @author kimchy (shay.banon)
*/
public class BlockingThreadPoolTest {
@ -40,7 +39,7 @@ public class BlockingThreadPoolTest {
final long waitTime = 1000; //1 second
final ThreadBarrier barrier = new ThreadBarrier(max + 1);
ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newBlockingThreadPool(min, max, 60000, 1, waitTime);
TransferThreadPoolExecutor pool = TransferThreadPoolExecutor.newBlockingExecutor(min, max, 60000, TimeUnit.MILLISECONDS, waitTime, TimeUnit.MILLISECONDS, 1, Executors.defaultThreadFactory());
assertThat("Min property", pool.getCorePoolSize(), equalTo(min));
assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max));
@ -67,13 +66,13 @@ public class BlockingThreadPoolTest {
assertThat("wrong active size", pool.getActiveCount(), equalTo(max));
//Queue should be empty, lets occupy it's only free space
assertThat("queue isn't empty", pool.getQueue().size(), equalTo(0));
assertThat("queue isn't empty", pool.getQueueSize(), equalTo(0));
pool.execute(new Runnable() {
public void run() {
//dummy task
}
});
assertThat("queue isn't full", pool.getQueue().size(), equalTo(1));
assertThat("queue isn't full", pool.getQueueSize(), equalTo(1));
//request should block since queue is full
try {

31
modules/elasticsearch/src/test/java/org/elasticsearch/util/concurrent/ScalingThreadPoolTest.java → modules/elasticsearch/src/test/java/org/elasticsearch/common/util/concurrent/ScalingThreadPoolTest.java
View File

@ -17,19 +17,19 @@
* under the License.
*/
package org.elasticsearch.util.concurrent;
package org.elasticsearch.common.util.concurrent;
import org.elasticsearch.common.util.concurrent.DynamicExecutors;
import org.elasticsearch.common.util.concurrent.ThreadBarrier;
import org.testng.annotations.Test;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.hamcrest.MatcherAssert.*;
import static org.hamcrest.Matchers.*;
/**
* @author kimchy (Shay Banon)
* @author kimchy (shay.banon)
*/
public class ScalingThreadPoolTest {
@ -38,7 +38,8 @@ public class ScalingThreadPoolTest {
final int max = 4;
final ThreadBarrier barrier = new ThreadBarrier(max + 1);
ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, Long.MAX_VALUE);
// ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, Long.MAX_VALUE);
TransferThreadPoolExecutor pool = TransferThreadPoolExecutor.newScalingExecutor(min, max, Long.MAX_VALUE, TimeUnit.NANOSECONDS, Executors.defaultThreadFactory());
assertThat("Min property", pool.getCorePoolSize(), equalTo(min));
assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max));
@ -72,7 +73,8 @@ public class ScalingThreadPoolTest {
final int max = 4;
final ThreadBarrier barrier = new ThreadBarrier(max + 1);
ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, 0 /*keep alive*/);
// ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, 0 /*keep alive*/);
TransferThreadPoolExecutor pool = TransferThreadPoolExecutor.newScalingExecutor(min, max, 0, TimeUnit.NANOSECONDS, Executors.defaultThreadFactory());
assertThat("Min property", pool.getCorePoolSize(), equalTo(min));
assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max));
@ -100,7 +102,7 @@ public class ScalingThreadPoolTest {
barrier.await();
Thread.sleep(1000);
assertThat("not all tasks completed", pool.getCompletedTaskCount(), equalTo((long) max));
// assertThat("not all tasks completed", pool.getCompletedTaskCount(), equalTo((long) max));
assertThat("wrong active count", pool.getActiveCount(), equalTo(0));
//Assert.assertEquals("wrong pool size. ", min, pool.getPoolSize()); //BUG in ThreadPool - Bug ID: 6458662
assertThat("idle threads didn't shrink below max. (" + pool.getPoolSize() + ")", pool.getPoolSize(), greaterThan(0));
@ -114,14 +116,17 @@ public class ScalingThreadPoolTest {
final int ntasks = 16;
final ThreadBarrier barrier = new ThreadBarrier(max + 1);
ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, Long.MAX_VALUE);
// ThreadPoolExecutor pool = (ThreadPoolExecutor) DynamicExecutors.newScalingThreadPool(min, max, Long.MAX_VALUE);
TransferThreadPoolExecutor pool = TransferThreadPoolExecutor.newScalingExecutor(min, max, Long.MAX_VALUE, TimeUnit.NANOSECONDS, Executors.defaultThreadFactory());
assertThat("Min property", pool.getCorePoolSize(), equalTo(min));
assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max));
final AtomicInteger tasksExecuted = new AtomicInteger();
for (int i = 0; i < ntasks; ++i) {
final int id = i;
pool.execute(new Runnable() {
public void run() {
tasksExecuted.incrementAndGet();
try {
if (id < max) {
barrier.await();
@ -138,13 +143,13 @@ public class ScalingThreadPoolTest {
Thread.sleep(100);
}
assertThat("wrong number of pooled tasks", pool.getQueue().size(), equalTo(ntasks - max));
assertThat("wrong number of pooled tasks", pool.getQueueSize(), equalTo(ntasks - max));
barrier.await();
//wait around for one second
Thread.sleep(1000);
assertThat("tasks not complete", pool.getCompletedTaskCount(), equalTo((long) ntasks));
assertThat("didn't scale above core pool size. (" + pool.getLargestPoolSize() + ")", pool.getLargestPoolSize(), greaterThan(min));
assertThat("Largest pool size exceeds max. (" + pool.getLargestPoolSize() + ")", pool.getLargestPoolSize(), lessThanOrEqualTo(max));
assertThat("tasks not complete", tasksExecuted.get(), equalTo(ntasks));
// assertThat("didn't scale above core pool size. (" + pool.getLargestPoolSize() + ")", pool.getLargestPoolSize(), greaterThan(min));
// assertThat("Largest pool size exceeds max. (" + pool.getLargestPoolSize() + ")", pool.getLargestPoolSize(), lessThanOrEqualTo(max));
}
}

3
modules/elasticsearch/src/test/java/org/elasticsearch/transport/netty/benchmark/BenchmarkNettyClient.java
View File

@ -38,7 +38,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author kimchy (Shay Banon)
* @author kimchy (shay.banon)
*/
public class BenchmarkNettyClient {
@ -57,6 +57,7 @@ public class BenchmarkNettyClient {
.build();
final ThreadPool threadPool = new CachedThreadPool(settings);
// final ThreadPool threadPool = new ScalingThreadPool(settings);
final TimerService timerService = new TimerService(settings, threadPool);
final TransportService transportService = new TransportService(new NettyTransport(settings, threadPool), threadPool, timerService).start();

125
modules/elasticsearch/src/test/java/org/elasticsearch/transport/netty/benchmark/BenchmarkNettyClientBlocking.java Normal file
View File

@ -0,0 +1,125 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.elasticsearch.transport.netty.benchmark;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.StopWatch;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.threadpool.cached.CachedThreadPool;
import org.elasticsearch.timer.TimerService;
import org.elasticsearch.transport.BaseTransportResponseHandler;
import org.elasticsearch.transport.RemoteTransportException;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.transport.netty.NettyTransport;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author kimchy (shay.banon)
*/
public class BenchmarkNettyClientBlocking {
public static void main(String[] args) {
final ByteSizeValue payloadSize = new ByteSizeValue(100, ByteSizeUnit.BYTES);
final int NUMBER_OF_CLIENTS = 3;
final int NUMBER_OF_ITERATIONS = 100000;
final byte[] payload = new byte[(int) payloadSize.bytes()];
final AtomicLong idGenerator = new AtomicLong();
final BenchmarkTransportResponseHandler responseHandler = new BenchmarkTransportResponseHandler();
Settings settings = ImmutableSettings.settingsBuilder()
.put("network.server", false)
.build();
final ThreadPool threadPool = new CachedThreadPool(settings);
// final ThreadPool threadPool = new ScalingThreadPool(settings);
final TimerService timerService = new TimerService(settings, threadPool);
final TransportService transportService = new TransportService(new NettyTransport(settings, threadPool), threadPool, timerService).start();
final DiscoveryNode node = new DiscoveryNode("server", new InetSocketTransportAddress("localhost", 9999));
transportService.connectToNode(node);
// warm things up
for (int i = 0; i < 10000; i++) {
BenchmarkMessage message = new BenchmarkMessage(1, payload);
transportService.submitRequest(node, "benchmark", message, responseHandler).txGet();
}
Thread[] clients = new Thread[NUMBER_OF_CLIENTS];
final CountDownLatch latch = new CountDownLatch(NUMBER_OF_CLIENTS);
for (int i = 0; i < NUMBER_OF_CLIENTS; i++) {
clients[i] = new Thread(new Runnable() {
@Override public void run() {
for (int j = 0; j < NUMBER_OF_ITERATIONS; j++) {
final long id = idGenerator.incrementAndGet();
BenchmarkMessage message = new BenchmarkMessage(id, payload);
transportService.submitRequest(node, "benchmark", message, responseHandler).txGet();
}
latch.countDown();
}
});
}
StopWatch stopWatch = new StopWatch().start();
for (int i = 0; i < NUMBER_OF_CLIENTS; i++) {
clients[i].start();
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
stopWatch.stop();
System.out.println("Ran [" + NUMBER_OF_CLIENTS + "], each with [" + NUMBER_OF_ITERATIONS + "] iterations, payload [" + payloadSize + "]: took [" + stopWatch.totalTime() + "], TPS: " + (NUMBER_OF_CLIENTS * NUMBER_OF_ITERATIONS) / stopWatch.totalTime().secondsFrac());
transportService.close();
threadPool.shutdownNow();
}
private static final class BenchmarkTransportResponseHandler extends BaseTransportResponseHandler<BenchmarkMessage> {
@Override public BenchmarkMessage newInstance() {
return new BenchmarkMessage();
}
@Override public void handleResponse(BenchmarkMessage response) {
}
@Override public void handleException(RemoteTransportException exp) {
exp.printStackTrace();
}
@Override public boolean spawn() {
return false;
}
}
}

3
modules/elasticsearch/src/test/java/org/elasticsearch/transport/netty/benchmark/BenchmarkNettyServer.java
View File

@ -30,7 +30,7 @@ import org.elasticsearch.transport.TransportService;
import org.elasticsearch.transport.netty.NettyTransport;
/**
* @author kimchy (Shay Banon)
* @author kimchy (shay.banon)
*/
public class BenchmarkNettyServer {
@ -42,6 +42,7 @@ public class BenchmarkNettyServer {
.build();
final ThreadPool threadPool = new CachedThreadPool(settings);
// final ThreadPool threadPool = new ScalingThreadPool(settings);
final TimerService timerService = new TimerService(settings, threadPool);
final TransportService transportService = new TransportService(new NettyTransport(settings, threadPool), threadPool, timerService).start();