AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Turns out presents' datagram support doesn't make sense at an individual datagram level, and

Browse Source 
requires a supporting Connection provided by the ConnectionManager.  As such, there's no point in
adding an additional selector for datagrams outside the ConnectionManager, and this moves it back
into BindingConectionManager.



git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6166 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Charlie Groves
2010-09-16 22:47:58 +00:00
parent 08a4a5281d
commit 308a6f9d52
5 changed files with 185 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/nio/DatagramAcceptor.java
-92
View File
@@ -1,92 +0,0 @@
package com.threerings.nio;
import java.util.List;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import com.google.common.collect.Lists;
import com.threerings.nio.SelectorIterable.SelectFailureHandler;
import static com.threerings.presents.Log.log;
/**
* Listens for datagrams arriving on a set of ports for a hostname and passes those datagrams on
* to a listener when they're ready to read. {@link #listen()} must be called after creating the
* acceptor to open the channels, and then tick must be called periodically to process new
* datagrams.
*/
public class DatagramAcceptor extends SelectAcceptor
{
/**
* Callback when datagrams are ready to be read.
*/
public interface DatagramHandler
{
/**
* Reads the message currentnly in the channel that was selected at the given time.
*/
void handleDatagram (DatagramChannel channel, long when);
}
/**
* Creates an acceptor that passes datagrams on to the given handler.
* @param failureHandler - called when the selector is irredemably broken.
* @param hostname - the hostname to bind to or null for all interfaces
* @param ports - the ports to bind to, or an empty array to skip binding
* @param selectLoopTime - the amount of time to wait in select, or 0 to skip the wait at all.
*/
public DatagramAcceptor (DatagramHandler dgramHandler, SelectFailureHandler failureHandler,
String hostname, int[] ports, int selectLoopTime)
throws IOException
{
super(failureHandler, hostname, ports, selectLoopTime);
_dgramHandler = dgramHandler;
}
/**
* Opens datagram channels on the configured addresses.
*/
public boolean listen ()
{
// open up the datagram ports as well
for (int port : _ports) {
try {
// create a channel and add it to the select set
final DatagramChannel channel = DatagramChannel.open();
channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY
channel.configureBlocking(false);
InetSocketAddress isa = getAddress(port);
channel.socket().bind(isa);
SelectionKey sk = channel.register(_selector, SelectionKey.OP_READ);
_handlers.put(sk, new SelectionHandler() {
public void handle (long when) {
_dgramHandler.handleDatagram(channel, when);
}
});
_datagramChannels.add(channel);
log.info("Server accepting datagrams on " + isa + ".");
} catch (IOException ioe) {
log.warning("Failure opening datagram channel", "hostname", _bindHostname,
"port", port, ioe);
}
}
return true;
}
/**
* Closes any open channels.
*/
public void shutdown()
{
// unbind sockets, if any
for (DatagramChannel datagramChannel : _datagramChannels) {
datagramChannel.socket().close();
}
}
protected final List<DatagramChannel> _datagramChannels = Lists.newArrayList();
protected final DatagramHandler _dgramHandler;
}
src/java/com/threerings/nio/SelectAcceptor.java
-66
View File
@@ -1,66 +0,0 @@
package com.threerings.nio;
import java.util.Map;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.google.common.primitives.Ints;
import com.samskivert.util.StringUtil;
import com.threerings.nio.SelectorIterable.SelectFailureHandler;
/**
* Base class for binding to ports with a selector and processing events on that selector.
*/
public abstract class SelectAcceptor
{
public SelectAcceptor (SelectFailureHandler failureHandler, String bindHostname, int[] ports,
int selectLoopTime)
throws IOException
{
Preconditions.checkNotNull(ports, "Ports must be non-null.");
_bindHostname = bindHostname;
_ports = ports;
_selectorSelector = new SelectorIterable(_selector, selectLoopTime, failureHandler);
}
/**
* Checks the selector for ready keys and passes any through to the handlers.
*/
public void tick (long when)
{
for (SelectionKey key : _selectorSelector) {
_handlers.get(key).handle(when);
}
}
public Iterable<Integer> getPorts()
{
return Ints.asList(_ports);
}
/** Helper function for creating proper bindable socket addresses. */
protected InetSocketAddress getAddress (int port)
{
return StringUtil.isBlank(_bindHostname) ?
new InetSocketAddress(port) : new InetSocketAddress(_bindHostname, port);
}
protected interface SelectionHandler {
void handle(long when);
}
protected final int[] _ports;
protected final String _bindHostname;
protected final Map<SelectionKey, SelectionHandler> _handlers = Maps.newHashMap();
protected final Selector _selector = Selector.open();
protected final SelectorIterable _selectorSelector;
}
src/java/com/threerings/nio/SocketChannelAcceptor.java
+59 -26
View File
@@ -1,14 +1,22 @@
package com.threerings.nio;
import java.util.List;
import java.util.Map;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.primitives.Ints;
import com.samskivert.util.StringUtil;
import com.threerings.nio.SelectorIterable.SelectFailureHandler;
import static com.threerings.presents.Log.log;
@@ -20,8 +28,18 @@ import static com.threerings.presents.Log.log;
* creating the acceptor to open the channels, and then tick must be called periodically to process
* new connections.
*/
public class SocketChannelAcceptor extends SelectAcceptor
public class SocketChannelAcceptor
{
/**
* Creates a address to the given host, or the wildcard host if the hostname is
* {@link StringUtil#blank}.
*/
public static InetSocketAddress getAddress (String hostname, int port)
{
return StringUtil.isBlank(hostname) ?
new InetSocketAddress(port) : new InetSocketAddress(hostname, port);
}
public interface SocketChannelHandler
{
void handleSocketChannel (SocketChannel channel, long when);
@@ -39,53 +57,62 @@ public class SocketChannelAcceptor extends SelectAcceptor
SelectFailureHandler failureHandler, String bindHostname, int[] ports, int selectLoopTime)
throws IOException
{
super(failureHandler, bindHostname, ports, selectLoopTime);
Preconditions.checkNotNull(ports, "Ports must be non-null.");
_bindHostname = bindHostname;
_ports = ports;
_selectorSelector = new SelectorIterable(_selector, selectLoopTime, failureHandler);
_connHandler = connectionHandler;
}
protected void acceptConnection (ServerSocketChannel listener, long when)
/**
* Checks the selector for ready keys and passes any through to the handlers.
*/
public void tick (long when)
{
SocketChannel channel = null;
try {
channel = listener.accept();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
if (channel == null) {
// in theory this shouldn't happen because we got an ACCEPT_READY event...
log.info("Psych! Got ACCEPT_READY, but no connection.");
return;
}
for (SelectionKey key : _selectorSelector) {
ServerSocketChannel ssocket = _channels.get(key);
SocketChannel channel = null;
try {
channel = ssocket.accept();
} catch (IOException e) {
log.warning("Got exception on accept", e);
continue;
}
if (channel == null) {
// in theory this shouldn't happen because we got an ACCEPT_READY event...
log.info("Psych! Got ACCEPT_READY, but no connection.");
continue;
}
// log.debug("Accepted connection " + channel + ".");
// log.debug("Accepted connection " + channel + ".");
_connHandler.handleSocketChannel(channel, when);
_connHandler.handleSocketChannel(channel, when);
}
}
public Iterable<Integer> getPorts ()
{
return Ints.asList(_ports);
}
/**
* Listens for socket connections on the configured addresses.
*/
public boolean listen ()
{
int successes = 0;
for (int port : _ports) {
try {
// create a listening socket and add it to the select set
final ServerSocketChannel ssocket = ServerSocketChannel.open();
ssocket.configureBlocking(false);
InetSocketAddress isa = getAddress(port);
InetSocketAddress isa = getAddress(_bindHostname, port);
ssocket.socket().bind(isa);
SelectionKey sk = ssocket.register(_selector, SelectionKey.OP_ACCEPT);
_handlers.put(sk, new SelectionHandler() {
public void handle (long when) {
acceptConnection(ssocket, when);
}
});
successes++;
_channels.put(sk, ssocket);
log.info("Server listening on " + isa + ".");
_ssockets.add(ssocket);
@@ -122,7 +149,7 @@ public class SocketChannelAcceptor extends SelectAcceptor
// }
// if we failed to listen on at least one port, give up the ghost
return successes > 0;
return !_channels.isEmpty();
}
public void shutdown()
@@ -139,6 +166,12 @@ public class SocketChannelAcceptor extends SelectAcceptor
}
}
protected final int[] _ports;
protected final String _bindHostname;
protected final Map<SelectionKey, ServerSocketChannel> _channels = Maps.newHashMap();
protected final Selector _selector = Selector.open();
protected final SelectorIterable _selectorSelector;
protected final List<ServerSocketChannel> _ssockets = Lists.newArrayList();
protected final SocketChannelHandler _connHandler;
}
src/java/com/threerings/presents/server/net/BindingConnectionManager.java
+114 -29
View File
@@ -1,14 +1,21 @@
package com.threerings.presents.server.net;
import java.util.List;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.samskivert.util.Lifecycle;
import com.threerings.presents.server.ReportManager;
import com.threerings.nio.DatagramAcceptor;
import com.threerings.nio.SocketChannelAcceptor;
import static com.threerings.presents.Log.log;
@@ -26,25 +33,6 @@ public class BindingConnectionManager extends ConnectionManager
super(cycle, repmgr, incomingEventWait);
}
@Override
protected void willStart ()
{
if (!_socketAcceptor.listen()) {
log.warning("ConnectionManager failed to bind to any ports. Shutting down.");
_server.queueShutdown();
} else {
_dgramAcceptor.listen();
}
}
@Override
protected void processIncomingEvents (long iterStamp)
{
_socketAcceptor.tick(iterStamp);
_dgramAcceptor.tick(iterStamp);
super.processIncomingEvents(iterStamp);
}
/**
* Configures the connection manager with the hostname and ports on which it will listen for
* socket connections and datagram packets. This must be called before the connection manager
@@ -61,17 +49,43 @@ public class BindingConnectionManager extends ConnectionManager
int[] datagramPorts)
throws IOException
{
Preconditions.checkNotNull(socketPorts, "Ports must be non-null.");
Preconditions.checkNotNull(socketPorts, "Socket ports must be non-null.");
Preconditions.checkNotNull(datagramPorts, "Datagram ports must be non-null. " +
"Pass a zero-length array to bind no datagram ports.");
// Listen for socket connections and datagram connections, but don't wait for anything to
// show up since that check is occurring as part of ConnectionManager's incoming event loop
// that already has a wait.
// Listen for socket connections, but don't wait to select on them. The connection check
// occurs as part of ConnectionManager's incoming event loop, which already has a wait.
_socketAcceptor = new SocketChannelAcceptor(this, _failureHandler, socketHostname,
socketPorts, 0);
_dgramAcceptor = new DatagramAcceptor(this, _failureHandler, datagramHostname,
datagramPorts, 0);
_datagramPorts = datagramPorts;
_datagramHostname = datagramHostname;
}
@Override
protected void willStart ()
{
if (!_socketAcceptor.listen()) {
log.warning("ConnectionManager failed to bind to any ports. Shutting down.");
_server.queueShutdown();
} else {
// open up the datagram ports as well
for (int port : _datagramPorts) {
try {
acceptDatagrams(port);
} catch (IOException ioe) {
log.warning("Failure opening datagram channel", "hostname", _datagramHostname,
"port", port, ioe);
}
}
}
}
@Override
protected void processIncomingEvents (long iterStamp)
{
_socketAcceptor.tick(iterStamp);
super.processIncomingEvents(iterStamp);
}
@Override
@@ -81,10 +95,81 @@ public class BindingConnectionManager extends ConnectionManager
// TODO: consider closing the listen sockets earlier, like in the shutdown method
_socketAcceptor.shutdown();
_dgramAcceptor.shutdown();
// unbind datagram channels, if any
for (DatagramChannel datagramChannel : _datagramChannels) {
datagramChannel.socket().close();
}
}
protected void acceptDatagrams (int port)
throws IOException
{
// create a channel and add it to the select set
final DatagramChannel channel = DatagramChannel.open();
channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY
channel.configureBlocking(false);
InetSocketAddress isa = SocketChannelAcceptor.getAddress(_datagramHostname, port);
channel.socket().bind(isa);
SelectionKey sk = channel.register(_selector, SelectionKey.OP_READ);
_handlers.put(sk, new NetEventHandler() {
public int handleEvent (long when) {
return handleDatagram(channel, when);
}
public boolean checkIdle (long now) {
return false;// Can't be idle
}
public void becameIdle () {}
});
_datagramChannels.add(channel);
log.info("Server accepting datagrams on " + isa + ".");
}
/**
* Called when a datagram message is ready to be read off its channel.
*/
protected int handleDatagram (DatagramChannel listener, long when)
{
InetSocketAddress source;
_databuf.clear();
try {
source = (InetSocketAddress)listener.receive(_databuf);
} catch (IOException ioe) {
log.warning("Failure receiving datagram.", ioe);
return 0;
}
// make sure we actually read a packet
if (source == null) {
log.info("Psych! Got READ_READY, but no datagram.");
return 0;
}
// flip the buffer and record the size (which must be at least 14 to contain the connection
// id, authentication hash, and a class reference)
int size = _databuf.flip().remaining();
if (size < 14) {
log.warning("Received undersized datagram", "source", source, "size", size);
return 0;
}
// the first four bytes are the connection id
int connectionId = _databuf.getInt();
Connection conn = _connections.get(connectionId);
if (conn != null) {
conn.handleDatagram(source, listener, _databuf, when);
} else {
log.debug("Received datagram for unknown connection", "id", connectionId,
"source", source);
}
return size;
}
protected SocketChannelAcceptor _socketAcceptor;
protected DatagramAcceptor _dgramAcceptor;
}
protected int[] _datagramPorts;
protected String _datagramHostname;
protected final List<DatagramChannel> _datagramChannels = Lists.newArrayList();
}
src/java/com/threerings/presents/server/net/ConnectionManager.java
+12 -62
View File
@@ -29,7 +29,6 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
@@ -69,32 +68,27 @@ import com.threerings.presents.server.PresentsServer;
import com.threerings.presents.server.ReportManager;
import com.threerings.presents.util.DatagramSequencer;
import com.threerings.nio.DatagramAcceptor;
import com.threerings.nio.SocketChannelAcceptor;
import com.threerings.nio.SocketChannelAcceptor.SocketChannelHandler;
import com.threerings.nio.DatagramAcceptor.DatagramHandler;
import com.threerings.nio.SelectorIterable;
import com.threerings.nio.SelectorIterable.SelectFailureHandler;
import static com.threerings.presents.Log.log;
/**
* Manages socket connections and datagram messages. It creates connection objects for each socket
* connection, but those connection objects interact closely with the connection manager because
* network I/O is done via a poll()-like mechanism rather than via threads.<p>
* Manages socket connections. It creates connection objects for each socket connection, but those
* connection objects interact closely with the connection manager because network I/O is done via
* a poll()-like mechanism rather than via threads.<p>
*
* ConnectionManager doesn't handle accepting tcp connections or listening for datagrams; it
* expects an external entity to do so and call its <code>handleSocketChannel</code> and
* <code>handleDatagram</code> methods.
* ConnectionManager doesn't handle accepting tcp connections; it expects an external entity to do
* so and call its <code>handleSocketChannel</code> method.
*
* @see BindingConnectionManager
* @see SocketChannelAcceptor
* @see DatagramAcceptor
*/
@Singleton
public class ConnectionManager extends LoopingThread
implements Lifecycle.ShutdownComponent, ReportManager.Reporter, DatagramHandler,
SocketChannelHandler
implements Lifecycle.ShutdownComponent, ReportManager.Reporter, SocketChannelHandler
{
/**
* Creates a connection manager instance.
@@ -266,47 +260,6 @@ public class ConnectionManager extends LoopingThread
}
}
/**
* Called by our DatagramAcceptor when a datagram message is ready to be read off its channel.
*/
public void handleDatagram (DatagramChannel listener, long when)
{
InetSocketAddress source;
_databuf.clear();
try {
source = (InetSocketAddress)listener.receive(_databuf);
} catch (IOException ioe) {
log.warning("Failure receiving datagram.", ioe);
return;
}
// make sure we actually read a packet
if (source == null) {
log.info("Psych! Got READ_READY, but no datagram.");
return;
}
// flip the buffer and record the size (which must be at least 14 to contain the connection
// id, authentication hash, and a class reference)
int size = _databuf.flip().remaining();
if (size < 14) {
log.warning("Received undersized datagram", "source", source, "size", size);
return;
}
// the first four bytes are the connection id
int connectionId = _databuf.getInt();
Connection conn = _connections.get(connectionId);
if (conn != null) {
conn.handleDatagram(source, listener, _databuf, when);
} else {
log.debug("Received datagram for unknown connection", "id", connectionId,
"source", source);
}
noteRead(size, 1, 1);
}
/**
* Performs the authentication process on the specified connection. This is called by {@link
* AuthingConnection} itself once it receives its auth request.
@@ -538,7 +491,12 @@ public class ConnectionManager extends LoopingThread
}
}
noteRead(bytesIn, msgsIn, eventCount);
synchronized (this) {
// update our stats
_stats.eventCount += eventCount;
_stats.bytesIn += bytesIn;
_stats.msgsIn += msgsIn;
}
}
/**
@@ -701,14 +659,6 @@ public class ConnectionManager extends LoopingThread
_stats.bytesOut += bytes;
}
protected synchronized void noteRead (long bytesIn, long msgsIn, long eventCount)
{
// update our stats
_stats.eventCount += eventCount;
_stats.bytesIn += bytesIn;
_stats.msgsIn += msgsIn;
}
/**
* Called by a connection when it has a downstream message that needs to be delivered.
* <em>Note:</em> this method is called as a result of a call to {@link Connection#postMessage}