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

Merged branch 'jetty-9.2.x' into 'master'.

This commit is contained in:
Simone Bordet 2015-03-23 22:32:40 +01:00
parent e255bda694 517b575c7f
commit 161317a5ef
8 changed files with 443 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
jetty-client/src/main/java/org/eclipse/jetty/client/ConnectionPool.java
View File

@ -37,8 +37,9 @@ import org.eclipse.jetty.util.component.Dumpable;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.SpinLock;
import org.eclipse.jetty.util.thread.Sweeper;
public class ConnectionPool implements Closeable, Dumpable
public class ConnectionPool implements Closeable, Dumpable, Sweeper.Sweepable
{
protected static final Logger LOG = Log.getLogger(ConnectionPool.class);
@ -294,6 +295,36 @@ public class ConnectionPool implements Closeable, Dumpable
ContainerLifeCycle.dump(out, indent, actives, idles);
}
@Override
public boolean sweep()
{
List<Sweeper.Sweepable> toSweep = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
{
for (Connection connection : getActiveConnections())
{
if (connection instanceof Sweeper.Sweepable)
toSweep.add(((Sweeper.Sweepable)connection));
}
}
for (Sweeper.Sweepable candidate : toSweep)
{
if (candidate.sweep())
{
boolean removed = getActiveConnections().remove(candidate);
LOG.warn("Connection swept: {}{}{} from active connections{}{}",
candidate,
System.lineSeparator(),
removed ? "Removed" : "Not removed",
System.lineSeparator(),
dump());
}
}
return false;
}
@Override
public String toString()
{

81
jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java
View File

@ -36,11 +36,16 @@ public class HttpExchange
private final HttpRequest request;
private final List<Response.ResponseListener> listeners;
private final HttpResponse response;
enum State { PENDING, COMPLETED, TERMINATED } ;
enum State
{
PENDING, COMPLETED, TERMINATED
}
;
private final SpinLock _lock = new SpinLock();
private State requestState=State.PENDING;
private State responseState=State.PENDING;
private State requestState = State.PENDING;
private State responseState = State.PENDING;
private Throwable requestFailure;
private Throwable responseFailure;
@ -67,7 +72,7 @@ public class HttpExchange
public Throwable getRequestFailure()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
return requestFailure;
}
@ -85,7 +90,7 @@ public class HttpExchange
public Throwable getResponseFailure()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
return responseFailure;
}
@ -105,32 +110,32 @@ public class HttpExchange
public boolean requestComplete()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
if (requestState!=State.PENDING)
if (requestState != State.PENDING)
return false;
requestState=State.COMPLETED;
requestState = State.COMPLETED;
return true;
}
}
public boolean responseComplete()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
if (responseState!=State.PENDING)
if (responseState != State.PENDING)
return false;
responseState=State.COMPLETED;
responseState = State.COMPLETED;
return true;
}
}
public Result terminateRequest(Throwable failure)
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
requestState=State.TERMINATED;
requestFailure=failure;
requestState = State.TERMINATED;
requestFailure = failure;
if (State.TERMINATED.equals(responseState))
return new Result(getRequest(), requestFailure, getResponse(), responseFailure);
}
@ -139,11 +144,11 @@ public class HttpExchange
public Result terminateResponse(Throwable failure)
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
responseState=State.TERMINATED;
responseFailure=failure;
if (State.TERMINATED.equals(requestState))
responseState = State.TERMINATED;
responseFailure = failure;
if (requestState == State.TERMINATED)
return new Result(getRequest(), requestFailure, getResponse(), responseFailure);
}
return null;
@ -173,23 +178,23 @@ public class HttpExchange
private boolean fail(Throwable cause)
{
boolean notify=false;
try(SpinLock.Lock lock = _lock.lock())
boolean notify = false;
try (SpinLock.Lock lock = _lock.lock())
{
if (!Boolean.TRUE.equals(requestState))
if (requestState != State.TERMINATED)
{
requestState=State.TERMINATED;
notify=true;
requestFailure=cause;
requestState = State.TERMINATED;
notify = true;
requestFailure = cause;
}
if (!Boolean.TRUE.equals(responseState))
if (responseState != State.TERMINATED)
{
responseState=State.TERMINATED;
notify=true;
responseFailure=cause;
responseState = State.TERMINATED;
notify = true;
responseFailure = cause;
}
}
if (notify)
{
if (LOG.isDebugEnabled())
@ -209,10 +214,10 @@ public class HttpExchange
public void resetResponse()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
responseState=State.PENDING;
responseFailure=null;
responseState = State.PENDING;
responseFailure = null;
}
}
@ -226,13 +231,13 @@ public class HttpExchange
@Override
public String toString()
{
try(SpinLock.Lock lock = _lock.lock())
try (SpinLock.Lock lock = _lock.lock())
{
return String.format("%s@%x req=%s/%s res=%s/%s",
HttpExchange.class.getSimpleName(),
hashCode(),
requestState,requestFailure,
responseState,responseFailure);
HttpExchange.class.getSimpleName(),
hashCode(),
requestState, requestFailure,
responseState, responseFailure);
}
}
}

4
jetty-client/src/main/java/org/eclipse/jetty/client/PoolingHttpDestination.java
View File

@ -25,6 +25,7 @@ import org.eclipse.jetty.client.api.Connection;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.component.ContainerLifeCycle;
import org.eclipse.jetty.util.thread.Sweeper;
public abstract class PoolingHttpDestination<C extends Connection> extends HttpDestination implements Callback
{
@ -34,6 +35,9 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
{
super(client, origin);
this.connectionPool = newConnectionPool(client);
Sweeper sweeper = client.getBean(Sweeper.class);
if (sweeper != null)
sweeper.offer(connectionPool);
}
protected ConnectionPool newConnectionPool(HttpClient client)

20
jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpConnectionOverHTTP.java
View File

@ -21,6 +21,7 @@ package org.eclipse.jetty.client.http;
import java.nio.channels.AsynchronousCloseException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.client.HttpConnection;
import org.eclipse.jetty.client.HttpDestination;
@ -33,13 +34,15 @@ import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.util.Promise;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.Sweeper;
public class HttpConnectionOverHTTP extends AbstractConnection implements Connection
public class HttpConnectionOverHTTP extends AbstractConnection implements Connection, Sweeper.Sweepable
{
private static final Logger LOG = Log.getLogger(HttpConnectionOverHTTP.class);
private final AtomicBoolean closed = new AtomicBoolean();
private final Promise<Connection> promise;
private final AtomicInteger sweeps = new AtomicInteger();
private final Delegate delegate;
private final HttpChannelOverHTTP channel;
private long idleTimeout;
@ -158,14 +161,27 @@ public class HttpConnectionOverHTTP extends AbstractConnection implements Connec
return exchange != null && exchange.getRequest().abort(failure);
}
@Override
public boolean sweep()
{
if (!closed.get())
return false;
if (sweeps.incrementAndGet() < 4)
return false;
return true;
}
@Override
public String toString()
{
return String.format("%s@%h(l:%s <-> r:%s)[%s]",
return String.format("%s@%h(l:%s <-> r:%s,closed=%b)[%s]",
getClass().getSimpleName(),
this,
getEndPoint().getLocalAddress(),
getEndPoint().getRemoteAddress(),
closed.get(),
channel);
}

26
jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java
View File

@ -343,17 +343,23 @@ public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dump
LOG.debug("Connected {} {}", connected, channel);
if (connected)
{
connect.timeout.cancel();
key.interestOps(0);
return new CreateEndPoint(channel, key)
if (connect.timeout.cancel())
{
@Override
protected void failed(Throwable failure)
key.interestOps(0);
return new CreateEndPoint(channel, key)
{
super.failed(failure);
connect.failed(failure);
}
};
@Override
protected void failed(Throwable failure)
{
super.failed(failure);
connect.failed(failure);
}
};
}
else
{
throw new SocketTimeoutException("Concurrent Connect Timeout");
}
}
else
{
@ -648,7 +654,7 @@ public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dump
{
if (LOG.isDebugEnabled())
LOG.debug("Channel {} timed out while connecting, closing it", channel);
connect.failed(new SocketTimeoutException());
connect.failed(new SocketTimeoutException("Connect Timeout"));
}
}
}

194
jetty-util/src/main/java/org/eclipse/jetty/util/thread/Sweeper.java Normal file
View File

@ -0,0 +1,194 @@
//
// ========================================================================
// Copyright (c) 1995-2015 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.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
/**
* <p>A utility class to perform periodic sweeping of resources.</p>
* <p>{@link Sweepable} resources may be added to or removed from a
* {@link Sweeper} and the resource implementation decides whether
* it should be swept or not.</p>
* <p>If a {@link Sweepable} resources is itself a container of
* other sweepable resources, it will forward the sweep operation
* to children resources, and so on recursively.</p>
* <p>Typical usage is to add {@link Sweeper} as a bean to an existing
* container:</p>
* <pre>
* Server server = new Server();
* server.addBean(new Sweeper(), true);
* server.start();
* </pre>
* Code that knows it has sweepable resources can then lookup the
* {@link Sweeper} and offer the sweepable resources to it:
* <pre>
* class MyComponent implements Sweeper.Sweepable
* {
* private final long creation;
* private volatile destroyed;
*
* MyComponent(Server server)
* {
* this.creation = System.nanoTime();
* Sweeper sweeper = server.getBean(Sweeper.class);
* sweeper.offer(this);
* }
*
* void destroy()
* {
* destroyed = true;
* }
*
* &#64;Override
* public boolean sweep()
* {
* return destroyed;
* }
* }
* </pre>
*/
public class Sweeper extends AbstractLifeCycle implements Runnable
{
private static final Logger LOG = Log.getLogger(Sweeper.class);
private final AtomicReference<List<Sweepable>> items = new AtomicReference<>();
private final AtomicReference<Scheduler.Task> task = new AtomicReference<>();
private final Scheduler scheduler;
private final long period;
public Sweeper(Scheduler scheduler, long period)
{
this.scheduler = scheduler;
this.period = period;
}
@Override
protected void doStart() throws Exception
{
super.doStart();
items.set(new CopyOnWriteArrayList<Sweepable>());
activate();
}
@Override
protected void doStop() throws Exception
{
deactivate();
items.set(null);
super.doStop();
}
public int getSize()
{
List<Sweepable> refs = items.get();
return refs == null ? 0 : refs.size();
}
public boolean offer(Sweepable sweepable)
{
List<Sweepable> refs = items.get();
if (refs == null)
return false;
refs.add(sweepable);
if (LOG.isDebugEnabled())
LOG.debug("Resource offered {}", sweepable);
return true;
}
public boolean remove(Sweepable sweepable)
{
List<Sweepable> refs = items.get();
return refs != null && refs.remove(sweepable);
}
@Override
public void run()
{
List<Sweepable> refs = items.get();
if (refs == null)
return;
for (Sweepable sweepable : refs)
{
try
{
if (sweepable.sweep())
{
refs.remove(sweepable);
if (LOG.isDebugEnabled())
LOG.debug("Resource swept {}", sweepable);
}
}
catch (Throwable x)
{
LOG.info("Exception while sweeping " + sweepable, x);
}
}
activate();
}
private void activate()
{
if (isRunning())
{
Scheduler.Task t = scheduler.schedule(this, period, TimeUnit.MILLISECONDS);
if (LOG.isDebugEnabled())
LOG.debug("Scheduled in {} ms sweep task {}", period, t);
task.set(t);
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("Skipping sweep task scheduling");
}
}
private void deactivate()
{
Scheduler.Task t = task.getAndSet(null);
if (t != null)
{
boolean cancelled = t.cancel();
if (LOG.isDebugEnabled())
LOG.debug("Cancelled ({}) sweep task {}", cancelled, t);
}
}
/**
* <p>A {@link Sweepable} resource implements this interface to
* signal to a {@link Sweeper} or to a parent container if it
* needs to be swept or not.</p>
* <p>Typical implementations will check their own internal state
* and return true or false from {@link #sweep()} to indicate
* whether they should be swept.</p>
*/
public interface Sweepable
{
/**
* @return whether this resource should be swept
*/
public boolean sweep();
}
}

124
jetty-util/src/test/java/org/eclipse/jetty/util/thread/SweeperTest.java Normal file
View File

@ -0,0 +1,124 @@
//
// ========================================================================
// Copyright (c) 1995-2015 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.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class SweeperTest
{
private Scheduler scheduler;
@Before
public void prepare() throws Exception
{
scheduler = new ScheduledExecutorScheduler();
scheduler.start();
}
@After
public void dispose() throws Exception
{
scheduler.stop();
}
@Test
public void testResourceNotSweptIsNotRemoved() throws Exception
{
testResourceSweepRemove(false);
}
@Test
public void testResourceSweptIsRemoved() throws Exception
{
testResourceSweepRemove(true);
}
private void testResourceSweepRemove(final boolean sweep) throws Exception
{
long period = 1000;
final CountDownLatch taskLatch = new CountDownLatch(1);
Sweeper sweeper = new Sweeper(scheduler, period)
{
@Override
public void run()
{
super.run();
taskLatch.countDown();
}
};
sweeper.start();
final CountDownLatch sweepLatch = new CountDownLatch(1);
sweeper.offer(new Sweeper.Sweepable()
{
@Override
public boolean sweep()
{
sweepLatch.countDown();
return sweep;
}
});
Assert.assertTrue(sweepLatch.await(2 * period, TimeUnit.MILLISECONDS));
Assert.assertTrue(taskLatch.await(2 * period, TimeUnit.MILLISECONDS));
Assert.assertEquals(sweep ? 0 : 1, sweeper.getSize());
sweeper.stop();
}
@Test
public void testSweepThrows() throws Exception
{
long period = 500;
final CountDownLatch taskLatch = new CountDownLatch(2);
Sweeper sweeper = new Sweeper(scheduler, period)
{
@Override
public void run()
{
super.run();
taskLatch.countDown();
}
};
sweeper.start();
final CountDownLatch sweepLatch = new CountDownLatch(2);
sweeper.offer(new Sweeper.Sweepable()
{
@Override
public boolean sweep()
{
sweepLatch.countDown();
throw new NullPointerException();
}
});
Assert.assertTrue(sweepLatch.await(4 * period, TimeUnit.MILLISECONDS));
Assert.assertTrue(taskLatch.await(4 * period, TimeUnit.MILLISECONDS));
Assert.assertEquals(1, sweeper.getSize());
sweeper.stop();
}
}

12
pom.xml
View File

@ -1035,5 +1035,17 @@
<alpn.version>8.1.3.v20150130</alpn.version>
</properties>
</profile>
<profile>
<id>8u40</id>
<activation>
<property>
<name>java.version</name>
<value>1.8.0_40</value>
</property>
</activation>
<properties>
<alpn.version>8.1.3.v20150130</alpn.version>
</properties>
</profile>
</profiles>
</project>