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-9

This commit is contained in:
Joakim Erdfelt 2012-07-20 08:22:54 -07:00
parent b5d23b1c64 b08bca2abb
commit 1c2c1b7346
4 changed files with 164 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,8 +25,11 @@ import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -303,7 +306,8 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
*/
public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dumpable
{
private final ConcurrentLinkedQueue<Runnable> _changes = new ConcurrentLinkedQueue<>();
private final Queue<Runnable> _changes = new ConcurrentLinkedQueue<>();
private final Set<AsyncEndPoint> _endPoints = Collections.newSetFromMap(new ConcurrentHashMap<AsyncEndPoint, Boolean>());
private final int _id;
private Selector _selector;
private Thread _thread;
@ -342,7 +346,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
{
if (Thread.currentThread() != _thread)
{
_changes.add(change);
_changes.offer(change);
LOG.debug("Queued change {}", change);
boolean wakeup = _needsWakeup;
if (wakeup)
@ -362,10 +366,8 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
{
Runnable change;
while ((change = _changes.poll()) != null)
{
runChange(change);
}
}
protected void runChange(Runnable change)
{
@ -531,9 +533,11 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
private AsyncEndPoint createEndPoint(SocketChannel channel, SelectionKey selectionKey) throws IOException
{
AsyncEndPoint endPoint = newEndPoint(channel, this, selectionKey);
_endPoints.add(endPoint);
endPointOpened(endPoint);
endPoint.setAsyncConnection(newConnection(channel, endPoint, selectionKey.attachment()));
endPoint.getAsyncConnection().onOpen();
AsyncConnection asyncConnection = newConnection(channel, endPoint, selectionKey.attachment());
endPoint.setAsyncConnection(asyncConnection);
asyncConnection.onOpen();
LOG.debug("Created {}", endPoint);
return endPoint;
}
@ -541,6 +545,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
public void destroyEndPoint(AsyncEndPoint endPoint)
{
LOG.debug("Destroyed {}", endPoint);
_endPoints.remove(endPoint);
endPoint.getAsyncConnection().onClose();
endPointClosed(endPoint);
}
@ -607,13 +612,11 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
private void timeoutCheck()
{
// We cannot use the _selector.keys() because the returned Set is not thread
// safe so it may be modified by the selector thread while we iterate here.
long now = System.currentTimeMillis();
for (SelectionKey key : _selector.keys())
{
Object attachment = key.attachment();
if (attachment instanceof AsyncEndPoint)
((AsyncEndPoint)attachment).checkTimeout(now);
}
for (AsyncEndPoint endPoint : _endPoints)
endPoint.checkTimeout(now);
}
private class DumpKeys implements Runnable

76
jetty-io/src/test/java/org/eclipse/jetty/io/IOTest.java
View File

@ -13,11 +13,6 @@
package org.eclipse.jetty.io;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@ -30,34 +25,33 @@ import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Future;
import junit.framework.Assert;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.toolchain.test.OS;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.IO;
import org.junit.Assert;
import org.junit.Test;
/**
*
*/
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class IOTest
{
@Test
public void testIO() throws InterruptedException
{
// Only a little test
ByteArrayInputStream in = new ByteArrayInputStream
("The quick brown fox jumped over the lazy dog".getBytes());
ByteArrayInputStream in = new ByteArrayInputStream("The quick brown fox jumped over the lazy dog".getBytes());
ByteArrayOutputStream out = new ByteArrayOutputStream();
IO.copyThread(in, out);
Thread.sleep(1500);
// System.err.println(out);
assertEquals( "copyThread",
out.toString(),
"The quick brown fox jumped over the lazy dog");
assertEquals("copyThread", out.toString(), "The quick brown fox jumped over the lazy dog");
}
@Test
@ -85,19 +79,31 @@ public class IOTest
assertEquals(-1, server.getInputStream().read());
// but cannot write
try { client.getOutputStream().write(1); fail("exception expected"); } catch (SocketException e) {}
try
{
client.getOutputStream().write(1);
fail("exception expected");
}
catch (SocketException e)
{
}
// but can still write in opposite direction.
server.getOutputStream().write(1);
assertEquals(1, client.getInputStream().read());
// server can shutdown input to match the shutdown out of client
server.shutdownInput();
// now we EOF instead of reading -1
try { server.getInputStream().read(); fail("exception expected"); } catch (SocketException e) {}
try
{
server.getInputStream().read();
fail("exception expected");
}
catch (SocketException e)
{
}
// but can still write in opposite direction.
server.getOutputStream().write(1);
@ -107,7 +113,14 @@ public class IOTest
client.shutdownInput();
// now we EOF instead of reading -1
try { client.getInputStream().read(); fail("exception expected"); } catch (SocketException e) {}
try
{
client.getInputStream().read();
fail("exception expected");
}
catch (SocketException e)
{
}
// But we can still write at the server (data which will never be read)
server.getOutputStream().write(1);
@ -119,7 +132,14 @@ public class IOTest
server.shutdownOutput();
// and now we can't write
try { server.getOutputStream().write(1); fail("exception expected"); } catch (SocketException e) {}
try
{
server.getOutputStream().write(1);
fail("exception expected");
}
catch (SocketException e)
{
}
// but the sockets are still open
assertFalse(client.isClosed());
@ -137,10 +157,8 @@ public class IOTest
// which has to be closed explictly
server.close();
assertTrue(server.isClosed());
}
@Test
public void testHalfCloseClientServer() throws Exception
{
@ -332,19 +350,21 @@ public class IOTest
Future<AsynchronousSocketChannel> acceptor = connector.accept();
AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
client.connect(connector.getLocalAddress());
client.connect(connector.getLocalAddress()).get(5, TimeUnit.SECONDS);
AsynchronousSocketChannel server = acceptor.get();
AsynchronousSocketChannel server = acceptor.get(5, TimeUnit.SECONDS);
ByteBuffer read = ByteBuffer.allocate(1024);
Future<Integer> reading = server.read(read);
ByteBuffer write= BufferUtil.toBuffer("Testing 1 2 3");
byte[] data = "Testing 1 2 3".getBytes("UTF-8");
ByteBuffer write = BufferUtil.toBuffer(data);
Future<Integer> writing = client.write(write);
reading.get();
writing.get();
writing.get(5, TimeUnit.SECONDS);
reading.get(5, TimeUnit.SECONDS);
read.flip();
Assert.assertEquals(ByteBuffer.wrap(data), read);
}
}

30
jetty-server/src/main/java/org/eclipse/jetty/server/LocalHttpConnector.java
View File

@ -34,20 +34,17 @@ public class LocalHttpConnector extends HttpConnector
private final BlockingQueue<LocalEndPoint> _connects = new LinkedBlockingQueue<LocalEndPoint>();
private LocalExecutor _executor;
/* ------------------------------------------------------------ */
public LocalHttpConnector()
{
setIdleTimeout(30000);
}
/* ------------------------------------------------------------ */
@Override
public Object getTransport()
{
return this;
}
/* ------------------------------------------------------------ */
/** Sends requests and get's responses based on thread activity.
* Returns all the responses received once the thread activity has
* returned to the level it was before the requests.
@ -61,7 +58,6 @@ public class LocalHttpConnector extends HttpConnector
return result==null?null:BufferUtil.toString(result,StringUtil.__UTF8_CHARSET);
}
/* ------------------------------------------------------------ */
/** Sends requests and get's responses based on thread activity.
* Returns all the responses received once the thread activity has
* returned to the level it was before the requests.
@ -78,7 +74,6 @@ public class LocalHttpConnector extends HttpConnector
return request.takeOutput();
}
/* ------------------------------------------------------------ */
/**
* Execute a request and return the EndPoint through which
* responses can be received.
@ -88,14 +83,13 @@ public class LocalHttpConnector extends HttpConnector
public LocalEndPoint executeRequest(String rawRequest)
{
Phaser phaser=_executor._phaser;
int phase = phaser.register(); // the corresponding arrival will be done by the acceptor thread when it takes
phaser.register(); // the corresponding arrival will be done by the acceptor thread when it takes
LocalEndPoint endp = new LocalEndPoint();
endp.setInput(BufferUtil.toBuffer(rawRequest,StringUtil.__UTF8_CHARSET));
_connects.add(endp);
return endp;
}
/* ------------------------------------------------------------ */
@Override
protected void accept(int acceptorID) throws IOException, InterruptedException
{
@ -104,11 +98,11 @@ public class LocalHttpConnector extends HttpConnector
HttpConnection connection=new HttpConnection(this,endp,getServer());
endp.setAsyncConnection(connection);
endp.onOpen();
connection.onOpen();
connectionOpened(connection);
_executor._phaser.arriveAndDeregister(); // arrive for the register done in getResponses
}
/* ------------------------------------------------------------ */
@Override
protected void doStart() throws Exception
{
@ -116,7 +110,6 @@ public class LocalHttpConnector extends HttpConnector
_executor=new LocalExecutor(findExecutor());
}
/* ------------------------------------------------------------ */
@Override
protected void doStop() throws Exception
{
@ -124,28 +117,25 @@ public class LocalHttpConnector extends HttpConnector
_executor=null;
}
/* ------------------------------------------------------------ */
@Override
public Executor findExecutor()
{
return _executor==null?super.findExecutor():_executor;
}
/* ------------------------------------------------------------ */
class LocalExecutor implements Executor
private class LocalExecutor implements Executor
{
Phaser _phaser=new Phaser()
private final Phaser _phaser=new Phaser()
{
@Override
protected boolean onAdvance(int phase, int registeredParties)
{
return false;
}
};
final Executor _executor;
LocalExecutor(Executor e)
private final Executor _executor;
private LocalExecutor(Executor e)
{
_executor=e;
}
@ -173,7 +163,6 @@ public class LocalHttpConnector extends HttpConnector
}
}
/* ------------------------------------------------------------ */
public class LocalEndPoint extends AsyncByteArrayEndPoint
{
private CountDownLatch _closed = new CountDownLatch(1);
@ -184,7 +173,6 @@ public class LocalHttpConnector extends HttpConnector
setIdleTimeout(LocalHttpConnector.this.getIdleTimeout());
}
/* ------------------------------------------------------------ */
public void addInput(String s)
{
// TODO this is a busy wait
@ -193,7 +181,6 @@ public class LocalHttpConnector extends HttpConnector
setInput(BufferUtil.toBuffer(s, StringUtil.__UTF8_CHARSET));
}
/* ------------------------------------------------------------ */
@Override
public void close()
{
@ -206,7 +193,6 @@ public class LocalHttpConnector extends HttpConnector
}
}
/* ------------------------------------------------------------ */
@Override
public void onClose()
{
@ -214,7 +200,6 @@ public class LocalHttpConnector extends HttpConnector
_closed.countDown();
}
/* ------------------------------------------------------------ */
@Override
public void shutdownOutput()
{
@ -222,7 +207,6 @@ public class LocalHttpConnector extends HttpConnector
close();
}
/* ------------------------------------------------------------ */
public void waitUntilClosed()
{
while (isOpen())

11
jetty-server/src/test/java/org/eclipse/jetty/server/handler/StatisticsHandlerTest.java
View File

@ -13,17 +13,11 @@
package org.eclipse.jetty.server.handler;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@ -38,6 +32,11 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class StatisticsHandlerTest
{
private Server _server;