Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'jetty-9' of ssh://git.eclipse.org/gitroot/jetty/org.eclipse.jetty.project into jetty-plugin

This commit is contained in:
Thomas Becker 2012-09-04 12:10:56 +02:00
parent 694a07dac6 4fce4d0e53
commit 0b77a69571
17 changed files with 565 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
jetty-io/src/test/java/org/eclipse/jetty/io/ByteArrayEndPointTest.java
View File

@ -37,7 +37,7 @@ import org.eclipse.jetty.toolchain.test.annotation.Slow;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.FutureCallback;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -51,7 +51,7 @@ public class ByteArrayEndPointTest
@Before
public void before() throws Exception
{
_scheduler = new SimpleScheduler();
_scheduler = new TimerScheduler();
_scheduler.start();
}

4
jetty-io/src/test/java/org/eclipse/jetty/io/SelectChannelEndPointInterestsTest.java
View File

@ -37,7 +37,7 @@ import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
@ -54,7 +54,7 @@ public class SelectChannelEndPointInterestsTest
threadPool = new QueuedThreadPool();
threadPool.start();
scheduler = new SimpleScheduler();
scheduler = new TimerScheduler();
scheduler.start();
connector = ServerSocketChannel.open();

4
jetty-io/src/test/java/org/eclipse/jetty/io/SelectChannelEndPointTest.java
View File

@ -44,7 +44,7 @@ import org.eclipse.jetty.util.FutureCallback;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -56,7 +56,7 @@ public class SelectChannelEndPointTest
protected volatile EndPoint _lastEndPoint;
protected ServerSocketChannel _connector;
protected QueuedThreadPool _threadPool = new QueuedThreadPool();
protected Scheduler _scheduler = new SimpleScheduler();
protected Scheduler _scheduler = new TimerScheduler();
protected SelectorManager _manager = new SelectorManager()
{
@Override

4
jetty-io/src/test/java/org/eclipse/jetty/io/SslConnectionTest.java
View File

@ -43,7 +43,7 @@ import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
@ -60,7 +60,7 @@ public class SslConnectionTest
private volatile FutureCallback<Void> _writeCallback;
protected ServerSocketChannel _connector;
protected QueuedThreadPool _threadPool = new QueuedThreadPool();
protected Scheduler _scheduler = new SimpleScheduler();
protected Scheduler _scheduler = new TimerScheduler();
protected SelectorManager _manager = new SelectorManager()
{
@Override

4
jetty-server/src/main/java/org/eclipse/jetty/server/AbstractConnector.java
View File

@ -39,7 +39,7 @@ import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
/**
* <p>Partial implementation of {@link Connector}</p>
@ -79,7 +79,7 @@ public abstract class AbstractConnector extends AggregateLifeCycle implements Co
{
_server=server;
_executor=executor!=null?executor:_server.getThreadPool();
_scheduler=scheduler!=null?scheduler:new SimpleScheduler();
_scheduler=scheduler!=null?scheduler:new TimerScheduler();
_byteBufferPool = pool!=null?pool:new MappedByteBufferPool();
_sslContextFactory = sslContextFactory;

2
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java
View File

@ -233,7 +233,7 @@ public class HttpChannel<T> implements HttpParser.RequestHandler<T>, Runnable
{
try
{
_request.setHandled(false); // TODO: is this right here ?
_request.setHandled(false);
_response.getHttpOutput().reopen();
if (_state.isInitial())

4
jetty-server/src/test/java/org/eclipse/jetty/server/ResponseTest.java
View File

@ -48,7 +48,7 @@ import org.eclipse.jetty.server.session.HashSessionIdManager;
import org.eclipse.jetty.server.session.HashSessionManager;
import org.eclipse.jetty.server.session.HashedSession;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Assert;
@ -65,7 +65,7 @@ public class ResponseTest
public void init() throws Exception
{
_server = new Server();
_scheduler = new SimpleScheduler();
_scheduler = new TimerScheduler();
LocalConnector connector = new LocalConnector(_server, null, _scheduler, null, null, 1);
_server.addConnector(connector);
_server.setHandler(new DumpHandler());

6
jetty-spdy/spdy-core/src/test/java/org/eclipse/jetty/spdy/AsyncTimeoutTest.java
View File

@ -38,7 +38,7 @@ import org.eclipse.jetty.toolchain.test.AdvancedRunner;
import org.eclipse.jetty.toolchain.test.annotation.Slow;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -55,7 +55,7 @@ public class AsyncTimeoutTest
ByteBufferPool bufferPool = new MappedByteBufferPool();
Executor threadPool = Executors.newCachedThreadPool();
Scheduler scheduler = new SimpleScheduler();
Scheduler scheduler = new TimerScheduler();
scheduler.start(); // TODO need to use jetty lifecycles better here
Generator generator = new Generator(bufferPool, new StandardCompressionFactory.StandardCompressor());
Session session = new StandardSession(SPDY.V2, bufferPool, threadPool, scheduler, new TestController(), null, 1, null, generator, new FlowControlStrategy.None())
@ -102,7 +102,7 @@ public class AsyncTimeoutTest
ByteBufferPool bufferPool = new MappedByteBufferPool();
Executor threadPool = Executors.newCachedThreadPool();
Scheduler scheduler = new SimpleScheduler();
Scheduler scheduler = new TimerScheduler();
scheduler.start();
Generator generator = new Generator(bufferPool, new StandardCompressionFactory.StandardCompressor());
Session session = new StandardSession(SPDY.V2, bufferPool, threadPool, scheduler, new TestController(), null, 1, null, generator, new FlowControlStrategy.None())

4
jetty-spdy/spdy-core/src/test/java/org/eclipse/jetty/spdy/StandardSessionTest.java
View File

@ -49,7 +49,7 @@ import org.eclipse.jetty.spdy.frames.SynStreamFrame;
import org.eclipse.jetty.spdy.generator.Generator;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
@ -87,7 +87,7 @@ public class StandardSessionTest
{
bufferPool = new MappedByteBufferPool();
threadPool = Executors.newCachedThreadPool();
scheduler = new SimpleScheduler();
scheduler = new TimerScheduler();
scheduler.start();
generator = new Generator(bufferPool, new StandardCompressionFactory.StandardCompressor());
session = new StandardSession(SPDY.V2,bufferPool,threadPool,scheduler,controller,null,1,null,generator,new FlowControlStrategy.None());

4
jetty-spdy/spdy-jetty/src/main/java/org/eclipse/jetty/spdy/SPDYClient.java
View File

@ -51,7 +51,7 @@ import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
public class SPDYClient
{
@ -167,7 +167,7 @@ public class SPDYClient
{
private final Queue<Session> sessions = new ConcurrentLinkedQueue<>();
private final ByteBufferPool bufferPool = new MappedByteBufferPool();
private final Scheduler scheduler = new SimpleScheduler();
private final Scheduler scheduler = new TimerScheduler();
private final Executor executor;
private final SslContextFactory sslContextFactory;
private final SelectorManager selector;

121
jetty-util/src/main/java/org/eclipse/jetty/util/thread/ConcurrentScheduler.java
View File

@ -59,15 +59,26 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
});
private final Queue _delayQ;
public ConcurrentScheduler()
{
this(null,8192);
}
public ConcurrentScheduler(Executor executor)
{
this(executor,8192);
}
public ConcurrentScheduler(int delayQms)
{
this(null,delayQms);
}
public ConcurrentScheduler(Executor executor,int delayQms)
{
_executor = executor;
addBean(_executor,false);
if (_executor!=null)
addBean(_executor,false);
_delayQ=new Queue(delayQms);
}
@ -75,7 +86,10 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
protected void doStart() throws Exception
{
super.doStart();
_executor.execute(this);
if (_executor==null)
new Thread(this).start();
else
_executor.execute(this);
}
@ -110,10 +124,9 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
long interval=event._executeAt-now;
// Should we execute this event?
if (interval<=0 && event._state.compareAndSet(State.NEW,State.DONE))
{
_executor.execute(event._task);
}
if (interval<=0 && event.compareAndSet(State.NEW,State.DONE))
event.execute();
// Should we delay this event
else if (_delayQ._delay>0 && interval>_delayQ._delay)
{
@ -121,7 +134,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
_delayQ.add(event,dequeue_at);
}
// else we schedule the event
else if (event._state.compareAndSet(State.NEW,State.SCHEDULED))
else if (event.compareAndSet(State.NEW,State.SCHEDULED))
{
_timerQ.add(event);
if (interval<=MAX_SLEEP)
@ -144,6 +157,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
_runner=Thread.currentThread();
while(isRunning())
{
try
{
// Work out how long to sleep for and execute expired events
@ -160,7 +174,20 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
{
Event event=next.dequeue();
if (event!=null)
_timerQ.add(event);
{
if (event._executeAt<=now)
{
if (event.compareAndSet(State.SCHEDULED,State.DONE))
event.execute();
}
else
{
long interval=event._executeAt-now;
_timerQ.add(event);
if (interval<sleep)
sleep=interval;
}
}
}
else
{
@ -185,10 +212,8 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
else if (event._executeAt<=now)
{
i.remove();
if (event._state.compareAndSet(State.SCHEDULED,State.DONE))
{
_executor.execute(event._task);
}
if (event.compareAndSet(State.SCHEDULED,State.DONE))
event.execute();
}
// else how long do we need to wait?
else
@ -218,27 +243,33 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
}
}
@Override
public String toString()
{
return String.format("%s@%x{%d,%s}",this.getClass().getSimpleName(),hashCode(),_delayQ._delay,_executor);
}
enum State { NEW, DELAYED, SCHEDULED, CANCELLED, DONE };
private class Event implements Scheduler.Task
private class Event extends AtomicReference<State> implements Scheduler.Task
{
/* extends AtomicReference as a minor optimisation rather than holding a _state field */
final Runnable _task;
final long _executeAt;
final AtomicReference<State> _state=new AtomicReference<>(State.NEW);
volatile QNode _node;
public Event(Runnable task, long executeAt)
{
super();
super(State.NEW);
_task = task;
_executeAt = executeAt;
}
public boolean isScheduled()
{
return _state.get()==State.SCHEDULED;
return get()==State.SCHEDULED;
}
@Override
@ -246,7 +277,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
{
while(true)
{
switch(_state.get())
switch(get())
{
case NEW:
throw new IllegalStateException();
@ -255,14 +286,14 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
case CANCELLED:
return false;
case DELAYED:
if (_state.compareAndSet(State.DELAYED,State.CANCELLED))
if (compareAndSet(State.DELAYED,State.CANCELLED))
{
_node.cancel();
return true;
}
break;
case SCHEDULED:
if (_state.compareAndSet(State.SCHEDULED,State.CANCELLED))
if (compareAndSet(State.SCHEDULED,State.CANCELLED))
{
_timerQ.remove(this);
return true;
@ -272,10 +303,18 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
}
}
public void execute()
{
if (_executor==null)
_task.run();
else
_executor.execute(_task);
}
@Override
public String toString()
{
return String.format("Event@%x{%s,%d,%s}",hashCode(),_state,_executeAt,_task);
return String.format("Event@%x{%s,%d,%s}",hashCode(),get(),_executeAt,_task);
}
}
@ -304,19 +343,19 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
Queue(int delay)
{
_delay=delay;
_head._next.set(_tail);
_head.set(_tail);
_tail._prev=_head;
}
void clear()
{
_head._next.set(_tail);
_head.set(_tail);
_tail._prev=_head;
}
void add(Event event, long dequeue_at)
{
if (event._state.compareAndSet(State.NEW,State.DELAYED))
if (event.compareAndSet(State.NEW,State.DELAYED))
{
while (true)
{
@ -325,7 +364,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
if (prev!=null)
{
QNode node = new QNode(event,dequeue_at,prev,_tail);
if (prev._next.compareAndSet(_tail,node))
if (prev.compareAndSet(_tail,node))
{
_tail._prev=node;
event._node=node;
@ -368,19 +407,19 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
* Roughly based on public domain lock free queue algorithm from:
* http://www.java2s.com/Code/Java/Collections-Data-Structure/ConcurrentDoublyLinkedList.htm
*/
private static class QNode
private static class QNode extends AtomicReference<QNode>
{
/* extends AtomicReference as a minor optimisation rather than holding a _next field */
final Event _event;
final long _dequeueAt;
final AtomicReference<QNode> _next=new AtomicReference<>();
volatile QNode _prev;
QNode(Event event, long dequeue_at, QNode prev, QNode next)
{
super(next);
_event=event;
_dequeueAt=dequeue_at;
_prev=prev;
_next.set(next);
}
/**
@ -400,8 +439,8 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
return event.scanForPrevOf(this);
// If the previous next is this (still linked normally)
QNode prev_next = prev._next.get();
if (prev_next==this)
QNode prev_next = prev.get();
if (prev_next==this && prev.isDelayed())
return prev;
if (prev_next==null || prev_next.isDelayed())
@ -448,7 +487,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
*/
QNode next()
{
QNode next = _next.get();
QNode next = get();
while (true)
{
if (next == null)
@ -459,21 +498,21 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
next._prev=this;
return next;
}
QNode next_next = next._next.get();
_next.compareAndSet(next, next_next);
QNode next_next = next.get();
compareAndSet(next, next_next);
next = next_next;
}
}
public boolean cancel()
{
if (_event._state.compareAndSet(State.DELAYED,State.CANCELLED))
if (_event.compareAndSet(State.DELAYED,State.CANCELLED))
{
QNode prev = _prev;
QNode next = _next.get();
QNode next = get();
if (prev != null && next != null && next.isDelayed())
{
if (prev._next.compareAndSet(this, next))
if (prev.compareAndSet(this, next))
next._prev=prev;
}
return true;
@ -483,13 +522,13 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
public Event dequeue()
{
if (_event._state.compareAndSet(State.DELAYED,State.SCHEDULED))
if (_event.compareAndSet(State.DELAYED,State.SCHEDULED))
{
QNode prev = _prev;
QNode next = _next.get();
QNode next = get();
if (prev != null && next != null && next.isDelayed())
{
if (prev._next.compareAndSet(this, next))
if (prev.compareAndSet(this, next))
next._prev=prev;
}
return _event;
@ -499,12 +538,12 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
public boolean isDelayed()
{
return _event!=null && _event._state.get()==State.DELAYED;
return _event!=null && _event.get()==State.DELAYED;
}
public boolean isTail()
{
return _event==null && _next.get()==null;
return _event==null && get()==null;
}
@ -512,7 +551,7 @@ public class ConcurrentScheduler extends AggregateLifeCycle implements Runnable,
public String toString()
{
QNode p=_prev;
QNode n=_next.get();
QNode n=get();
return String.format("QNode@%x{%x<-%s->%x}",hashCode(),p==null?0:p.hashCode(),_event,n==null?0:n.hashCode());
}
}

63
jetty-util/src/main/java/org/eclipse/jetty/util/thread/ScheduledExecutionServiceScheduler.java Normal file
View File

@ -0,0 +1,63 @@
//
// ========================================================================
// Copyright (c) 1995-2012 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.util.thread;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
public class ScheduledExecutionServiceScheduler extends AbstractLifeCycle implements Scheduler
{
ScheduledExecutorService _service;
public ScheduledExecutionServiceScheduler()
{
}
@Override
protected void doStart() throws Exception
{
_service=new ScheduledThreadPoolExecutor(1);
super.doStart();
}
@Override
protected void doStop() throws Exception
{
super.doStop();
_service=null;
}
@Override
public Task schedule(final Runnable task, final long delay, final TimeUnit units)
{
final Future<?> future = _service.schedule(task,delay,units);
return new Task()
{
@Override
public boolean cancel()
{
return future.cancel(true);
}
};
}
}

6
jetty-util/src/main/java/org/eclipse/jetty/util/thread/SimpleScheduler.java → jetty-util/src/main/java/org/eclipse/jetty/util/thread/TimerScheduler.java
View File

@ -27,7 +27,7 @@ import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
public class SimpleScheduler extends AbstractLifeCycle implements Scheduler
public class TimerScheduler extends AbstractLifeCycle implements Scheduler
{
/* this class uses the Timer class rather than an ScheduledExecutionService because
* it uses the same algorithm internally and the signature is cheaper to use as there are no
@ -37,12 +37,12 @@ public class SimpleScheduler extends AbstractLifeCycle implements Scheduler
Timer _timer;
final String _name;
public SimpleScheduler()
public TimerScheduler()
{
this(null);
}
public SimpleScheduler(String name)
public TimerScheduler(String name)
{
_name=name;
}

290
jetty-util/src/test/java/org/eclipse/jetty/util/BenchmarkHelper.java Normal file
View File

@ -0,0 +1,290 @@
//
// ========================================================================
// Copyright (c) 1995-2012 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.util;
import java.lang.management.CompilationMXBean;
import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryPoolMXBean;
import java.lang.management.MemoryUsage;
import java.lang.management.OperatingSystemMXBean;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class BenchmarkHelper implements Runnable
{
private final OperatingSystemMXBean operatingSystem;
private final CompilationMXBean jitCompiler;
private final MemoryMXBean heapMemory;
private final AtomicInteger starts = new AtomicInteger();
private volatile MemoryPoolMXBean youngMemoryPool;
private volatile MemoryPoolMXBean survivorMemoryPool;
private volatile MemoryPoolMXBean oldMemoryPool;
private volatile boolean hasMemoryPools;
private volatile ScheduledFuture<?> memoryPoller;
private volatile GarbageCollectorMXBean youngCollector;
private volatile GarbageCollectorMXBean oldCollector;
private volatile boolean hasCollectors;
private volatile ScheduledExecutorService scheduler;
private volatile boolean polling;
private volatile long lastYoungUsed;
private volatile long startYoungCollections;
private volatile long startYoungCollectionsTime;
private volatile long totalYoungUsed;
private volatile long lastSurvivorUsed;
private volatile long totalSurvivorUsed;
private volatile long lastOldUsed;
private volatile long startOldCollections;
private volatile long startOldCollectionsTime;
private volatile long totalOldUsed;
private volatile long startTime;
private volatile long startProcessCPUTime;
private volatile long startJITCTime;
public BenchmarkHelper()
{
this.operatingSystem = ManagementFactory.getOperatingSystemMXBean();
this.jitCompiler = ManagementFactory.getCompilationMXBean();
this.heapMemory = ManagementFactory.getMemoryMXBean();
List<MemoryPoolMXBean> memoryPools = ManagementFactory.getMemoryPoolMXBeans();
for (MemoryPoolMXBean memoryPool : memoryPools)
{
if ("PS Eden Space".equals(memoryPool.getName()) ||
"Par Eden Space".equals(memoryPool.getName()) ||
"G1 Eden".equals(memoryPool.getName()))
youngMemoryPool = memoryPool;
else if ("PS Survivor Space".equals(memoryPool.getName()) ||
"Par Survivor Space".equals(memoryPool.getName()) ||
"G1 Survivor".equals(memoryPool.getName()))
survivorMemoryPool = memoryPool;
else if ("PS Old Gen".equals(memoryPool.getName()) ||
"CMS Old Gen".equals(memoryPool.getName()) ||
"G1 Old Gen".equals(memoryPool.getName()))
oldMemoryPool = memoryPool;
}
hasMemoryPools = youngMemoryPool != null && survivorMemoryPool != null && oldMemoryPool != null;
List<GarbageCollectorMXBean> garbageCollectors = ManagementFactory.getGarbageCollectorMXBeans();
for (GarbageCollectorMXBean garbageCollector : garbageCollectors)
{
if ("PS Scavenge".equals(garbageCollector.getName()) ||
"ParNew".equals(garbageCollector.getName()) ||
"G1 Young Generation".equals(garbageCollector.getName()))
youngCollector = garbageCollector;
else if ("PS MarkSweep".equals(garbageCollector.getName()) ||
"ConcurrentMarkSweep".equals(garbageCollector.getName()) ||
"G1 Old Generation".equals(garbageCollector.getName()))
oldCollector = garbageCollector;
}
hasCollectors = youngCollector != null && oldCollector != null;
}
public void run()
{
if (!hasMemoryPools)
return;
long young = youngMemoryPool.getUsage().getUsed();
long survivor = survivorMemoryPool.getUsage().getUsed();
long old = oldMemoryPool.getUsage().getUsed();
if (!polling)
{
polling = true;
}
else
{
if (lastYoungUsed <= young)
{
totalYoungUsed += young - lastYoungUsed;
}
if (lastSurvivorUsed <= survivor)
{
totalSurvivorUsed += survivor - lastSurvivorUsed;
}
if (lastOldUsed <= old)
{
totalOldUsed += old - lastOldUsed;
}
else
{
// May need something more here, like "how much was collected"
}
}
lastYoungUsed = young;
lastSurvivorUsed = survivor;
lastOldUsed = old;
}
public boolean startStatistics()
{
// Support for multiple nodes requires to ignore start requests after the first
// but also requires that requests after the first wait until the initialization
// is completed (otherwise node #2 may start the run while the server is GC'ing)
synchronized (this)
{
if (starts.incrementAndGet() > 1)
return false;
System.gc();
System.err.println("\n========================================");
System.err.println("Statistics Started at " + new Date());
System.err.println("Operative System: " + operatingSystem.getName() + " " + operatingSystem.getVersion() + " " + operatingSystem.getArch());
System.err.println("JVM : " + System.getProperty("java.vm.vendor") + " " + System.getProperty("java.vm.name") + " runtime " + System.getProperty("java.vm.version") + " " + System.getProperty("java.runtime.version"));
System.err.println("Processors: " + operatingSystem.getAvailableProcessors());
if (operatingSystem instanceof com.sun.management.OperatingSystemMXBean)
{
com.sun.management.OperatingSystemMXBean os = (com.sun.management.OperatingSystemMXBean)operatingSystem;
long totalMemory = os.getTotalPhysicalMemorySize();
long freeMemory = os.getFreePhysicalMemorySize();
System.err.println("System Memory: " + percent(totalMemory - freeMemory, totalMemory) + "% used of " + gibiBytes(totalMemory) + " GiB");
}
else
{
System.err.println("System Memory: N/A");
}
MemoryUsage heapMemoryUsage = heapMemory.getHeapMemoryUsage();
System.err.println("Used Heap Size: " + mebiBytes(heapMemoryUsage.getUsed()) + " MiB");
System.err.println("Max Heap Size: " + mebiBytes(heapMemoryUsage.getMax()) + " MiB");
if (hasMemoryPools)
{
long youngGenerationHeap = heapMemoryUsage.getMax() - oldMemoryPool.getUsage().getMax();
System.err.println("Young Generation Heap Size: " + mebiBytes(youngGenerationHeap) + " MiB");
}
else
{
System.err.println("Young Generation Heap Size: N/A");
}
System.err.println("- - - - - - - - - - - - - - - - - - - - ");
scheduler = Executors.newSingleThreadScheduledExecutor();
polling = false;
memoryPoller = scheduler.scheduleWithFixedDelay(this, 0, 250, TimeUnit.MILLISECONDS);
lastYoungUsed = 0;
if (hasCollectors)
{
startYoungCollections = youngCollector.getCollectionCount();
startYoungCollectionsTime = youngCollector.getCollectionTime();
}
totalYoungUsed = 0;
lastSurvivorUsed = 0;
totalSurvivorUsed = 0;
lastOldUsed = 0;
if (hasCollectors)
{
startOldCollections = oldCollector.getCollectionCount();
startOldCollectionsTime = oldCollector.getCollectionTime();
}
totalOldUsed = 0;
startTime = System.nanoTime();
if (operatingSystem instanceof com.sun.management.OperatingSystemMXBean)
{
com.sun.management.OperatingSystemMXBean os = (com.sun.management.OperatingSystemMXBean)operatingSystem;
startProcessCPUTime = os.getProcessCpuTime();
}
startJITCTime = jitCompiler.getTotalCompilationTime();
return true;
}
}
public boolean stopStatistics()
{
synchronized (this)
{
if (starts.decrementAndGet() > 0)
return false;
memoryPoller.cancel(false);
scheduler.shutdown();
System.err.println("- - - - - - - - - - - - - - - - - - - - ");
System.err.println("Statistics Ended at " + new Date());
long elapsedTime = System.nanoTime() - startTime;
System.err.println("Elapsed time: " + TimeUnit.NANOSECONDS.toMillis(elapsedTime) + " ms");
long elapsedJITCTime = jitCompiler.getTotalCompilationTime() - startJITCTime;
System.err.println("\tTime in JIT compilation: " + elapsedJITCTime + " ms");
if (hasCollectors)
{
long elapsedYoungCollectionsTime = youngCollector.getCollectionTime() - startYoungCollectionsTime;
long youngCollections = youngCollector.getCollectionCount() - startYoungCollections;
System.err.println("\tTime in Young Generation GC: " + elapsedYoungCollectionsTime + " ms (" + youngCollections + " collections)");
long elapsedOldCollectionsTime = oldCollector.getCollectionTime() - startOldCollectionsTime;
long oldCollections = oldCollector.getCollectionCount() - startOldCollections;
System.err.println("\tTime in Old Generation GC: " + elapsedOldCollectionsTime + " ms (" + oldCollections + " collections)");
}
else
{
System.err.println("\tTime in GC: N/A");
}
if (hasMemoryPools)
{
System.err.println("Garbage Generated in Young Generation: " + mebiBytes(totalYoungUsed) + " MiB");
System.err.println("Garbage Generated in Survivor Generation: " + mebiBytes(totalSurvivorUsed) + " MiB");
System.err.println("Garbage Generated in Old Generation: " + mebiBytes(totalOldUsed) + " MiB");
}
else
{
System.err.println("Garbage Generated: N/A");
}
if (operatingSystem instanceof com.sun.management.OperatingSystemMXBean)
{
com.sun.management.OperatingSystemMXBean os = (com.sun.management.OperatingSystemMXBean)operatingSystem;
long elapsedProcessCPUTime = os.getProcessCpuTime() - startProcessCPUTime;
System.err.println("Average CPU Load: " + ((float)elapsedProcessCPUTime * 100 / elapsedTime) + "/" + (100 * operatingSystem.getAvailableProcessors()));
}
else
{
System.err.println("Average CPU Load: N/A");
}
System.err.println("----------------------------------------\n");
return true;
}
}
public float percent(long dividend, long divisor)
{
return (float)dividend * 100 / divisor;
}
public float mebiBytes(long bytes)
{
return (float)bytes / 1024 / 1024;
}
public float gibiBytes(long bytes)
{
return (float)bytes / 1024 / 1024 / 1024;
}
}

132
jetty-util/src/test/java/org/eclipse/jetty/util/thread/SchedulerTest.java
View File

@ -28,11 +28,13 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.eclipse.jetty.toolchain.test.annotation.Slow;
import org.eclipse.jetty.util.BenchmarkHelper;
import org.eclipse.jetty.util.statistic.SampleStatistic;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@ -41,15 +43,18 @@ import org.junit.runners.Parameterized;
@RunWith(value = Parameterized.class)
public class SchedulerTest
{
public static Executor executor = Executors.newFixedThreadPool(256);
private static final BenchmarkHelper benchmark = new BenchmarkHelper();
private static final Executor executor = Executors.newFixedThreadPool(256);
@Parameterized.Parameters
public static Collection<Object[]> data()
{
Object[][] data = new Object[][]{
{new SimpleScheduler()},
{new ConcurrentScheduler(executor,0)},
{new ConcurrentScheduler(executor,2000)}
{new TimerScheduler()},
{new ScheduledExecutionServiceScheduler()},
{new ConcurrentScheduler(0)},
{new ConcurrentScheduler(1500)},
{new ConcurrentScheduler(executor,1500)}
};
return Arrays.asList(data);
}
@ -95,6 +100,42 @@ public class SchedulerTest
}
@Test
public void testTwoExecution() throws Exception
{
final AtomicLong executed = new AtomicLong();
long expected=System.currentTimeMillis()+3000;
Scheduler.Task task=_scheduler.schedule(new Runnable()
{
@Override
public void run()
{
executed.set(System.currentTimeMillis());
}
},3000,TimeUnit.MILLISECONDS);
Thread.sleep(4000);
Assert.assertFalse(task.cancel());
Assert.assertThat(executed.get(),Matchers.greaterThanOrEqualTo(expected));
Assert.assertThat(expected-executed.get(),Matchers.lessThan(1000L));
final AtomicLong executed1 = new AtomicLong();
long expected1=System.currentTimeMillis()+3000;
Scheduler.Task task1=_scheduler.schedule(new Runnable()
{
@Override
public void run()
{
executed1.set(System.currentTimeMillis());
}
},3000,TimeUnit.MILLISECONDS);
Thread.sleep(4000);
Assert.assertFalse(task1.cancel());
Assert.assertThat(executed1.get(),Matchers.greaterThanOrEqualTo(expected1));
Assert.assertThat(expected1-executed1.get(),Matchers.lessThan(1000L));
}
@Test
public void testQuickCancel() throws Exception
{
@ -106,11 +147,11 @@ public class SchedulerTest
{
executed.set(System.currentTimeMillis());
}
},3000,TimeUnit.MILLISECONDS);
},2000,TimeUnit.MILLISECONDS);
Thread.sleep(100);
Assert.assertTrue(task.cancel());
Thread.sleep(3500);
Thread.sleep(2500);
Assert.assertEquals(0,executed.get());
}
@ -125,21 +166,37 @@ public class SchedulerTest
{
executed.set(System.currentTimeMillis());
}
},3000,TimeUnit.MILLISECONDS);
},2000,TimeUnit.MILLISECONDS);
Thread.sleep(2100);
Thread.sleep(1600);
Assert.assertTrue(task.cancel());
Thread.sleep(1500);
Thread.sleep(1000);
Assert.assertEquals(0,executed.get());
}
@Test
@Slow
public void testManySchedulesAndCancels() throws Exception
{
final Random random = new Random();
Thread[] test = new Thread[2000];
schedule(100,10000,3800,200);
}
@Test
@Slow
@Ignore
public void testBenchmark() throws Exception
{
schedule(2000,10000,2000,50);
benchmark.startStatistics();
System.err.println(_scheduler);
schedule(2000,30000,2000,50);
benchmark.stopStatistics();
}
private void schedule(int threads,final int duration, final int delay, final int interval) throws Exception
{
final Random random = new Random(1);
Thread[] test = new Thread[threads];
final AtomicInteger schedules = new AtomicInteger();
final SampleStatistic executions = new SampleStatistic();
@ -156,18 +213,22 @@ public class SchedulerTest
{
long now = System.currentTimeMillis();
long start=now;
long end=start+5000;
while (now<end)
long end=start+duration;
boolean last=false;
while (!last)
{
final int delay=random.nextInt((int)(end-now));
final long expected = now+delay;
final long expected=now+delay;
int cancel=random.nextInt(interval);
final boolean expected_to_execute;
int cancel=random.nextInt(100);
if (cancel==0)
cancel=(int)(end-now)+1000;
last=now+2*interval>end;
if (cancel==0 || last)
{
expected_to_execute=true;
cancel=delay+1000;
}
else
cancel=cancel/4;
expected_to_execute=false;
schedules.incrementAndGet();
Scheduler.Task task=_scheduler.schedule(new Runnable()
@ -175,8 +236,12 @@ public class SchedulerTest
@Override
public void run()
{
long lateness=System.currentTimeMillis()-expected;
executions.set(lateness);
long lateness=System.currentTimeMillis()-expected;
if (expected_to_execute)
executions.set(lateness);
else
executions.set(6666);
}
},delay,TimeUnit.MILLISECONDS);
@ -185,9 +250,20 @@ public class SchedulerTest
if (task.cancel())
{
long lateness=now-expected;
cancellations.set(lateness);
if (expected_to_execute)
cancellations.set(lateness);
else
cancellations.set(0);
}
else
{
if (!expected_to_execute)
{
cancellations.set(9999);
}
}
Thread.yield();
}
}
catch (InterruptedException e)
@ -205,9 +281,9 @@ public class SchedulerTest
for (Thread thread : test)
thread.join();
//System.err.println(schedules);
//System.err.println(executions);
//System.err.println(cancellations);
// System.err.println(schedules);
// System.err.println(executions);
// System.err.println(cancellations);
// there were some executions and cancellations
Assert.assertThat(executions.getCount(),Matchers.greaterThan(0L));

6
jetty-websocket/websocket-client/src/main/java/org/eclipse/jetty/websocket/client/WebSocketClientFactory.java
View File

@ -35,7 +35,7 @@ import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.eclipse.jetty.websocket.api.Extension;
import org.eclipse.jetty.websocket.api.ExtensionRegistry;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
@ -69,7 +69,7 @@ public class WebSocketClientFactory extends AggregateLifeCycle
public WebSocketClientFactory(Executor threadPool)
{
this(threadPool,new SimpleScheduler());
this(threadPool,new TimerScheduler());
}
public WebSocketClientFactory(Executor threadPool, Scheduler scheduler)
@ -110,7 +110,7 @@ public class WebSocketClientFactory extends AggregateLifeCycle
public WebSocketClientFactory(SslContextFactory sslContextFactory)
{
this(new QueuedThreadPool(),new SimpleScheduler(),sslContextFactory);
this(new QueuedThreadPool(),new TimerScheduler(),sslContextFactory);
}
@Override

4
jetty-websocket/websocket-server/src/main/java/org/eclipse/jetty/websocket/server/WebSocketServerFactory.java
View File

@ -43,7 +43,7 @@ import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.SimpleScheduler;
import org.eclipse.jetty.util.thread.TimerScheduler;
import org.eclipse.jetty.websocket.annotations.WebSocket;
import org.eclipse.jetty.websocket.api.Extension;
import org.eclipse.jetty.websocket.api.ExtensionRegistry;
@ -78,7 +78,7 @@ public class WebSocketServerFactory extends AggregateLifeCycle implements WebSoc
/**
* Have the factory maintain 1 and only 1 scheduler. All connections share this scheduler.
*/
private final Scheduler scheduler = new SimpleScheduler();
private final Scheduler scheduler = new TimerScheduler();
private final String supportedVersions;
private final WebSocketPolicy basePolicy;
private final EventMethodsCache methodsCache;