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

430654 avoid dispatching failure callbacks

This commit is contained in:
Greg Wilkins 2014-03-27 19:08:16 +11:00
parent cda4af3ec9
commit 86d13b91a5
8 changed files with 114 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java
View File

@ -21,12 +21,14 @@ package org.eclipse.jetty.io;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.NonBlockingThread;
/**
* <p>A convenience base implementation of {@link Connection}.</p>
@ -87,6 +89,33 @@ public abstract class AbstractConnection implements Connection
return _executor;
}
protected void failedCallback(final Callback callback, final Throwable x)
{
if (NonBlockingThread.isNonBlockingThread())
{
try
{
getExecutor().execute(new Runnable()
{
@Override
public void run()
{
callback.failed(x);
}
});
}
catch(RejectedExecutionException e)
{
LOG.debug(e);
callback.failed(x);
}
}
else
{
callback.failed(x);
}
}
/**
* <p>Utility method to be called to register read interest.</p>
* <p>After a call to this method, {@link #onFillable()} or {@link #onFillInterestedFailed(Throwable)}

3
jetty-io/src/main/java/org/eclipse/jetty/io/AbstractEndPoint.java
View File

@ -143,9 +143,10 @@ public abstract class AbstractEndPoint extends IdleTimeout implements EndPoint
protected void onIdleExpired(TimeoutException timeout)
{
boolean output_shutdown=isOutputShutdown();
boolean input_shutdown=isInputShutdown();
_fillInterest.onFail(timeout);
_writeFlusher.onFail(timeout);
if (output_shutdown)
if (isOpen() && output_shutdown || input_shutdown)
close();
}

22
jetty-io/src/main/java/org/eclipse/jetty/io/SelectorManager.java
View File

@ -46,6 +46,7 @@ import org.eclipse.jetty.util.component.ContainerLifeCycle;
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.NonBlockingThread;
import org.eclipse.jetty.util.thread.Scheduler;
/**
@ -66,7 +67,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
private final ManagedSelector[] _selectors;
private long _connectTimeout = DEFAULT_CONNECT_TIMEOUT;
private long _selectorIndex;
protected SelectorManager(Executor executor, Scheduler scheduler)
{
this(executor, scheduler, (Runtime.getRuntime().availableProcessors() + 1) / 2);
@ -475,14 +476,21 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
public void run()
{
_thread = Thread.currentThread();
String name = _thread.getName();
final String name = _thread.getName();
try
{
_thread.setName(name + "-selector-" + SelectorManager.this.getClass().getSimpleName()+"@"+Integer.toHexString(SelectorManager.this.hashCode())+"/"+_id);
LOG.debug("Starting {} on {}", _thread, this);
while (isRunning())
select();
runChanges();
NonBlockingThread.runAsNonBlocking(new Runnable()
{
@Override
public void run()
{
_thread.setName(name + "-selector-" + SelectorManager.this.getClass().getSimpleName()+"@"+Integer.toHexString(SelectorManager.this.hashCode())+"/"+_id);
LOG.debug("Starting {} on {}", _thread, this);
while (isRunning())
select();
runChanges();
}
});
}
finally
{

15
jetty-io/src/main/java/org/eclipse/jetty/io/ssl/SslConnection.java
View File

@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.Arrays;
import java.util.concurrent.Executor;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
@ -38,6 +39,7 @@ import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.FillInterest;
import org.eclipse.jetty.io.RuntimeIOException;
import org.eclipse.jetty.io.SelectChannelEndPoint;
import org.eclipse.jetty.io.SelectorManager;
import org.eclipse.jetty.io.WriteFlusher;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
@ -300,17 +302,22 @@ public class SslConnection extends AbstractConnection
final boolean filler_failed=fail_filler;
getExecutor().execute(new Runnable()
failedCallback(new Callback()
{
@Override
public void run()
{
public void succeeded()
{
}
@Override
public void failed(Throwable x)
{
if (filler_failed)
getFillInterest().onFail(x);
getWriteFlusher().onFail(x);
}
});
},x);
}
};

35
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnection.java
View File

@ -34,6 +34,7 @@ import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.SelectorManager;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.IteratingCallback;
@ -609,22 +610,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
public void failed(final Throwable x)
{
super.failed(x);
try
{
getExecutor().execute(new Runnable()
{
@Override
public void run()
{
_callback.failed(x);
}
});
}
catch(RejectedExecutionException e)
{
LOG.debug(e);
_callback.failed(x);
}
failedCallback(_callback,x);
}
}
@ -720,22 +706,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
public void failed(final Throwable x)
{
super.failed(x);
try
{
getExecutor().execute(new Runnable()
{
@Override
public void run()
{
_callback.failed(x);
}
});
}
catch (RejectedExecutionException e)
{
LOG.debug(e);
_callback.failed(x);
}
failedCallback(_callback,x);
}
}

9
jetty-util/src/main/java/org/eclipse/jetty/util/BlockingCallback.java
View File

@ -24,12 +24,18 @@ import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.NonBlockingThread;
/* ------------------------------------------------------------ */
/**
* An implementation of Callback that blocks until success or failure.
*/
public class BlockingCallback implements Callback
{
private static final Logger LOG = Log.getLogger(BlockingCallback.class);
private static Throwable SUCCEEDED=new Throwable()
{
@Override
@ -64,6 +70,9 @@ public class BlockingCallback implements Callback
*/
public void block() throws IOException
{
if (NonBlockingThread.isNonBlockingThread())
LOG.warn("Blocking a NonBlockingThread: ",new Throwable());
try
{
_latch.await();

4
jetty-util/src/main/java/org/eclipse/jetty/util/SharedBlockingCallback.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.NonBlockingThread;
/* ------------------------------------------------------------ */
@ -168,6 +169,9 @@ public class SharedBlockingCallback
*/
public void block() throws IOException
{
if (NonBlockingThread.isNonBlockingThread())
LOG.warn("Blocking a NonBlockingThread: ",new Throwable());
_lock.lock();
try
{

41
jetty-util/src/main/java/org/eclipse/jetty/util/thread/NonBlockingThread.java Normal file
View File

@ -0,0 +1,41 @@
//
// ========================================================================
// Copyright (c) 1995-2014 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;
public class NonBlockingThread
{
private final static ThreadLocal<Boolean> __nonBlockingThread = new ThreadLocal<>();
public static boolean isNonBlockingThread()
{
return Boolean.TRUE.equals(__nonBlockingThread.get());
}
public static void runAsNonBlocking(Runnable runnable)
{
try
{
__nonBlockingThread.set(Boolean.TRUE);
runnable.run();
}
finally
{
__nonBlockingThread.remove();
}
}
}