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

Issue #1418 

Improved async handling to avoid race of setWriteListener->handle with async read callback.
The async read callback now calls into the HttpChannelState with a new onReadPossible() method.
A more detailed state machine is now kept for async reads, with and additional state for using
the handler thread to produce more content.
This commit is contained in:
Greg Wilkins 2017-03-23 15:55:20 +11:00
parent 08ee1b62d4
commit 00b42ca5ee
7 changed files with 202 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -381,6 +381,12 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
throw _state.getAsyncContextEvent().getThrowable();
}
case READ_PRODUCE:
{
_request.getHttpInput().produceContent();
break;
}
case READ_CALLBACK:
{
ContextHandler handler=_state.getContextHandler();

184
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java
View File

@ -78,6 +78,7 @@ public class HttpChannelState
ERROR_DISPATCH, // handle a normal error
ASYNC_ERROR, // handle an async error
WRITE_CALLBACK, // handle an IO write callback
READ_PRODUCE, // Check is a read is possible by parsing/filling
READ_CALLBACK, // handle an IO read callback
COMPLETE, // Complete the response
TERMINATED, // No further actions
@ -99,19 +100,15 @@ public class HttpChannelState
ERRORED // The error has been processed
}
private enum Interest
private enum AsyncRead
{
NONE(false),
NEEDED(true),
REGISTERED(true);
private final boolean _interested;
Interest(boolean interest)
{
_interested = interest;
}
private boolean isInterested() { return _interested;}
NONE, // No isReady; No data
AVAILABLE, // No isReady; onDataAvailable
NEEDED, // isReady()==false handling; No data
REGISTERED, // isReady()==false !handling; No data
POSSIBLE, // isReady()==false async callback called (http/1 only)
PRODUCING, // isReady()==false handling content production (http/1 only)
READY // isReady() was false, data now available
}
private final Locker _locker=new Locker();
@ -120,8 +117,7 @@ public class HttpChannelState
private State _state;
private Async _async;
private boolean _initial;
private boolean _asyncReadPossible;
private Interest _asyncRead=Interest.NONE;
private AsyncRead _asyncRead=AsyncRead.NONE;
private boolean _asyncWritePossible;
private long _timeoutMs=DEFAULT_TIMEOUT;
private AsyncContextEvent _event;
@ -187,14 +183,13 @@ public class HttpChannelState
public String toStringLocked()
{
return String.format("%s@%x{s=%s a=%s i=%b r=%s/%s w=%b}",
return String.format("%s@%x{s=%s a=%s i=%b r=%s w=%b}",
getClass().getSimpleName(),
hashCode(),
_state,
_async,
_initial,
_asyncRead,
_asyncReadPossible,
_asyncWritePossible);
}
@ -234,11 +229,18 @@ public class HttpChannelState
return Action.TERMINATED;
case ASYNC_WOKEN:
if (_asyncReadPossible && _asyncRead.isInterested())
switch (_asyncRead)
{
case POSSIBLE:
_state=State.ASYNC_IO;
_asyncRead=Interest.NONE;
_asyncRead=AsyncRead.PRODUCING;
return Action.READ_PRODUCE;
case READY:
_state=State.ASYNC_IO;
_asyncRead=AsyncRead.NONE;
return Action.READ_CALLBACK;
default:
break;
}
if (_asyncWritePossible)
@ -385,7 +387,7 @@ public class HttpChannelState
protected Action unhandle()
{
Action action;
boolean read_interested=false;
boolean read_interested = false;
try(Locker.Lock lock= _locker.lock())
{
@ -412,7 +414,7 @@ public class HttpChannelState
}
_initial=false;
switch(_async)
async: switch(_async)
{
case COMPLETE:
_state=State.COMPLETING;
@ -427,13 +429,31 @@ public class HttpChannelState
break;
case STARTED:
if (_asyncReadPossible && _asyncRead.isInterested())
switch(_asyncRead)
{
case READY:
_state=State.ASYNC_IO;
_asyncRead=Interest.NONE;
_asyncRead=AsyncRead.NONE;
action=Action.READ_CALLBACK;
break async;
case POSSIBLE:
_state=State.ASYNC_IO;
action=Action.READ_PRODUCE;
break async;
case NEEDED:
case PRODUCING:
_asyncRead=AsyncRead.REGISTERED;
read_interested=true;
case NONE:
case AVAILABLE:
case REGISTERED:
break;
}
else if (_asyncWritePossible)
if (_asyncWritePossible)
{
_state=State.ASYNC_IO;
_asyncWritePossible=false;
@ -443,11 +463,6 @@ public class HttpChannelState
{
_state=State.ASYNC_WAIT;
action=Action.WAIT;
if (_asyncRead==Interest.NEEDED)
{
_asyncRead=Interest.REGISTERED;
read_interested=true;
}
Scheduler scheduler=_channel.getScheduler();
if (scheduler!=null && _timeoutMs>0)
_event.setTimeoutTask(scheduler.schedule(_event,_timeoutMs,TimeUnit.MILLISECONDS));
@ -915,8 +930,7 @@ public class HttpChannelState
_state=State.IDLE;
_async=Async.NOT_ASYNC;
_initial=true;
_asyncReadPossible=false;
_asyncRead=Interest.NONE;
_asyncRead=AsyncRead.NONE;
_asyncWritePossible=false;
_timeoutMs=DEFAULT_TIMEOUT;
_event=null;
@ -943,8 +957,7 @@ public class HttpChannelState
_state=State.UPGRADED;
_async=Async.NOT_ASYNC;
_initial=true;
_asyncReadPossible=false;
_asyncRead=Interest.NONE;
_asyncRead=AsyncRead.NONE;
_asyncWritePossible=false;
_timeoutMs=DEFAULT_TIMEOUT;
_event=null;
@ -1133,19 +1146,30 @@ public class HttpChannelState
if (LOG.isDebugEnabled())
LOG.debug("onReadUnready {}",toStringLocked());
// We were already unready, this is not a state change, so do nothing
if (_asyncRead!=Interest.REGISTERED)
switch(_asyncRead)
{
_asyncReadPossible=false; // Assumes this has been checked in isReady() with lock held
case NONE:
case AVAILABLE:
case READY:
case NEEDED:
if (_state==State.ASYNC_WAIT)
{
interested=true;
_asyncRead=Interest.REGISTERED;
_asyncRead=AsyncRead.REGISTERED;
}
else
{
_asyncRead=Interest.NEEDED;
_asyncRead=AsyncRead.NEEDED;
}
break;
case REGISTERED:
case POSSIBLE:
case PRODUCING:
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
@ -1160,7 +1184,7 @@ public class HttpChannelState
* is returned.
* @return True IFF the channel was unready and in ASYNC_WAIT state
*/
public boolean onReadPossible()
public boolean onDataAvailable()
{
boolean woken=false;
try(Locker.Lock lock= _locker.lock())
@ -1168,12 +1192,34 @@ public class HttpChannelState
if (LOG.isDebugEnabled())
LOG.debug("onReadPossible {}",toStringLocked());
_asyncReadPossible=true;
if (_state==State.ASYNC_WAIT && _asyncRead.isInterested())
switch(_asyncRead)
{
case NONE:
_asyncRead=AsyncRead.AVAILABLE;
break;
case AVAILABLE:
break;
case PRODUCING:
_asyncRead=AsyncRead.READY;
break;
case NEEDED:
case REGISTERED:
case POSSIBLE:
case READY:
_asyncRead=AsyncRead.READY;
if (_state==State.ASYNC_WAIT)
{
woken=true;
_state=State.ASYNC_WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
@ -1181,7 +1227,7 @@ public class HttpChannelState
/**
* Called to signal that the channel is ready for a callback.
* This is similar to calling {@link #onReadUnready()} followed by
* {@link #onReadPossible()}, except that as content is already
* {@link #onDataAvailable()}, except that as content is already
* available, read interest is never set.
* @return true if woken
*/
@ -1193,13 +1239,52 @@ public class HttpChannelState
if (LOG.isDebugEnabled())
LOG.debug("onReadReady {}",toStringLocked());
_asyncRead=Interest.REGISTERED;
_asyncReadPossible=true;
switch(_asyncRead)
{
case NONE:
case AVAILABLE:
_asyncRead=AsyncRead.READY;
if (_state==State.ASYNC_WAIT)
{
woken=true;
_state=State.ASYNC_WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
/**
* Called to indicate that more content may be available,
* but that a handling thread may need to produce (fill/parse)
* it. Typically called by the async read success callback.
*/
public boolean onReadPossible()
{
boolean woken=false;
try(Locker.Lock lock= _locker.lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onReadReady {}",toStringLocked());
switch(_asyncRead)
{
case REGISTERED:
_asyncRead=AsyncRead.POSSIBLE;
if (_state==State.ASYNC_WAIT)
{
woken=true;
_state=State.ASYNC_WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
@ -1217,9 +1302,8 @@ public class HttpChannelState
if (LOG.isDebugEnabled())
LOG.debug("onEof {}",toStringLocked());
// Force read interest so onAllDataRead can be called
_asyncRead=Interest.REGISTERED;
_asyncReadPossible=true;
// Force read ready so onAllDataRead can be called
_asyncRead=AsyncRead.READY;
if (_state==State.ASYNC_WAIT)
{
woken=true;
@ -1229,14 +1313,6 @@ public class HttpChannelState
return woken;
}
public boolean isReadPossible()
{
try(Locker.Lock lock= _locker.lock())
{
return _asyncReadPossible;
}
}
public boolean onWritePossible()
{
boolean wake=false;

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

@ -625,10 +625,8 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
@Override
public void succeeded()
{
if (fillAndParseForContent())
if (_channel.getState().onReadPossible())
_channel.handle();
else if (!_input.isFinished() && !_input.hasContent())
asyncReadFillInterested();
}
@Override

6
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java
View File

@ -577,7 +577,7 @@ public class HttpInput extends ServletInputStream implements Runnable
if (_listener == null)
_inputQ.notify();
else
woken = _channelState.onReadPossible();
woken = _channelState.onDataAvailable();
}
return woken;
}
@ -612,7 +612,7 @@ public class HttpInput extends ServletInputStream implements Runnable
if (_listener == null)
_inputQ.notify();
else
woken = _channelState.onReadPossible();
woken = _channelState.onDataAvailable();
}
}
return woken;
@ -800,7 +800,7 @@ public class HttpInput extends ServletInputStream implements Runnable
if (_listener == null)
_inputQ.notify();
else
woken = _channelState.onReadPossible();
woken = _channelState.onDataAvailable();
}
return woken;

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

@ -120,9 +120,9 @@ public class HttpInputAsyncStateTest
}
@Override
public boolean onReadPossible()
public boolean onDataAvailable()
{
boolean wake = super.onReadPossible();
boolean wake = super.onDataAvailable();
__history.add("onReadPossible "+wake);
return wake;
}

59
jetty-server/src/test/java/org/eclipse/jetty/server/HttpInputTest.java
View File

@ -42,19 +42,19 @@ public class HttpInputTest
@Override
public void onError(Throwable t)
{
_history.add("onError:" + t);
_history.add("l.onError:" + t);
}
@Override
public void onDataAvailable() throws IOException
{
_history.add("onDataAvailable");
_history.add("l.onDataAvailable");
}
@Override
public void onAllDataRead() throws IOException
{
_history.add("onAllDataRead");
_history.add("l.onAllDataRead");
}
};
private HttpInput _in;
@ -99,21 +99,28 @@ public class HttpInputTest
@Override
public void onReadUnready()
{
_history.add("unready");
_history.add("s.onReadUnready");
super.onReadUnready();
}
@Override
public boolean onReadPossible()
{
_history.add("onReadPossible");
_history.add("s.onReadPossible");
return super.onReadPossible();
}
@Override
public boolean onDataAvailable()
{
_history.add("s.onDataAvailable");
return super.onDataAvailable();
}
@Override
public boolean onReadReady()
{
_history.add("ready");
_history.add("s.onReadReady");
return super.onReadReady();
}
})
@ -387,17 +394,17 @@ public class HttpInputTest
{
_in.setReadListener(_listener);
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
}
@ -406,21 +413,21 @@ public class HttpInputTest
{
_in.setReadListener(_listener);
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.addContent(new TContent("AB"));
_fillAndParseSimulate.add("CD");
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.run();
Assert.assertThat(_history.poll(), Matchers.equalTo("onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.equalTo("l.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));
@ -434,7 +441,7 @@ public class HttpInputTest
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 1"));
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.read(), Matchers.equalTo((int)'C'));
@ -446,7 +453,7 @@ public class HttpInputTest
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
}
@ -455,13 +462,13 @@ public class HttpInputTest
{
_in.setReadListener(_listener);
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.eof();
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));
Assert.assertThat(_in.isFinished(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.read(), Matchers.equalTo(-1));
@ -474,22 +481,22 @@ public class HttpInputTest
{
_in.setReadListener(_listener);
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.addContent(new TContent("AB"));
_fillAndParseSimulate.add("_EOF_");
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.run();
Assert.assertThat(_history.poll(), Matchers.equalTo("onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.equalTo("l.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));
@ -504,7 +511,7 @@ public class HttpInputTest
Assert.assertThat(_in.isFinished(), Matchers.equalTo(false));
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 1"));
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isFinished(), Matchers.equalTo(false));
@ -521,21 +528,21 @@ public class HttpInputTest
{
_in.setReadListener(_listener);
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(false));
Assert.assertThat(_history.poll(), Matchers.equalTo("produceContent 0"));
Assert.assertThat(_history.poll(), Matchers.equalTo("unready"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onReadUnready"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.failed(new TimeoutException());
Assert.assertThat(_history.poll(), Matchers.equalTo("onReadPossible"));
Assert.assertThat(_history.poll(), Matchers.equalTo("s.onDataAvailable"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
_in.run();
Assert.assertThat(_in.isFinished(), Matchers.equalTo(true));
Assert.assertThat(_history.poll(), Matchers.equalTo("onError:java.util.concurrent.TimeoutException"));
Assert.assertThat(_history.poll(), Matchers.equalTo("l.onError:java.util.concurrent.TimeoutException"));
Assert.assertThat(_history.poll(), Matchers.nullValue());
Assert.assertThat(_in.isReady(), Matchers.equalTo(true));

6
jetty-server/src/test/java/org/eclipse/jetty/server/session/FileSessionManagerTest.java
View File

@ -139,7 +139,7 @@ public class FileSessionManagerTest
Assert.assertTrue(f1.delete());
f1.createNewFile();
Thread.currentThread().sleep(20);
Thread.sleep(1100);
String name2 = "101__0.0.0.0_abc";
File f2 = new File(testDir, name2);
@ -147,7 +147,7 @@ public class FileSessionManagerTest
Assert.assertTrue(f2.delete());
f2.createNewFile();
Thread.currentThread().sleep(20);
Thread.sleep(1100);
String name3 = "102__0.0.0.0_abc";
File f3 = new File(testDir, name3);
@ -155,7 +155,7 @@ public class FileSessionManagerTest
Assert.assertTrue(f3.delete());
f3.createNewFile();
Thread.currentThread().sleep(20);
Thread.sleep(1100);
Session session = handler.getSession("abc");