You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by jv...@apache.org on 2011/05/23 17:44:33 UTC
svn commit: r1126544 - in
/mina/branches/3.0/core/src/main/java/org/apache/mina: IoSession.java
session/AbstractIoSession.java transport/tcp/NioSelectorProcessor.java
Author: jvermillard
Date: Mon May 23 15:44:33 2011
New Revision: 1126544
URL: http://svn.apache.org/viewvc?rev=1126544&view=rev
Log:
registering session for writing
Modified:
mina/branches/3.0/core/src/main/java/org/apache/mina/IoSession.java
mina/branches/3.0/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
mina/branches/3.0/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/IoSession.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/IoSession.java?rev=1126544&r1=1126543&r2=1126544&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/IoSession.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/IoSession.java Mon May 23 15:44:33 2011
@@ -23,6 +23,7 @@ import java.net.SocketAddress;
import java.util.Set;
import org.apache.mina.service.SelectorProcessor;
+import org.apache.mina.session.WriteQueue;
/**
* A handle which represents a connection between two end-points regardless of
@@ -280,4 +281,10 @@ public interface IoSession {
* @return the {@link WriteFuture} for tracking this asynchronous operation
*/
public WriteFuture writeWithFuture(Object message);
+
+ /**
+ * Get the {@link WriteQueue} of this session. The write queue contains the pending writes.
+ * @return the write queue of this session
+ */
+ public WriteQueue getWriteQueue();
}
\ No newline at end of file
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/session/AbstractIoSession.java?rev=1126544&r1=1126543&r2=1126544&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/session/AbstractIoSession.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/session/AbstractIoSession.java Mon May 23 15:44:33 2011
@@ -22,6 +22,7 @@ package org.apache.mina.session;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.mina.IoService;
@@ -50,6 +51,9 @@ public abstract class AbstractIoSession
/** The service this session is associated with */
private final IoService service;
+ /** The {@link SelectorProcessor} used for handling this session writing */
+ private SelectorProcessor writeProcessor;
+
/** The number of bytes read since this session has been created */
private volatile long readBytes;
@@ -70,6 +74,9 @@ public abstract class AbstractIoSession
protected Object stateMonitor = new Object();
protected SessionState state;
+
+ /** is this session registered for being polled for write ready events */
+ AtomicBoolean registeredForWrite = new AtomicBoolean();
/** the queue of pending writes for the session, to be dequeued by the {@link SelectorProcessor} */
private WriteQueue writeQueue = new DefaultWriteQueue();
@@ -210,7 +217,11 @@ public abstract class AbstractIoSession
}
writeQueue.add(new DefaultWriteRequest(message));
- // mark as interested to write
+ // If it wasn't, we register this session as interested to write.
+ // It's done in atomic fashion for avoiding two concurent registering.
+ if (!registeredForWrite.getAndSet(true)) {
+ writeProcessor.flush(this);
+ }
}
@Override
@@ -219,4 +230,9 @@ public abstract class AbstractIoSession
// TODO implements WriteFuture
return null;
}
+
+ @Override
+ public WriteQueue getWriteQueue() {
+ return writeQueue;
+ }
}
\ No newline at end of file
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java?rev=1126544&r1=1126543&r2=1126544&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java Mon May 23 15:44:33 2011
@@ -39,12 +39,16 @@ import org.apache.mina.IoService;
import org.apache.mina.IoSession;
import org.apache.mina.service.SelectorProcessor;
import org.apache.mina.service.SelectorStrategy;
+import org.apache.mina.session.WriteQueue;
+import org.apache.mina.session.WriteRequest;
import org.apache.mina.transport.tcp.nio.NioTcpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
+ * A {@link SelectorProcessor} for processing NIO based {@link IoSession}.
+ *
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*
*/
@@ -63,18 +67,23 @@ public class NioSelectorProcessor implem
private ByteBuffer readBuffer;
- // new binded server to add to the selector
- // {ServerSocketChannel, IoServer}
+ /**
+ * new binded server to add to the selector
+ * {ServerSocketChannel, IoServer}
+ */
private final Queue<Object[]> serversToAdd = new ConcurrentLinkedQueue<Object[]>();
- // server to remove of the selector
+ /** server to remove of the selector */
private final Queue<ServerSocketChannel> serversToRemove = new ConcurrentLinkedQueue<ServerSocketChannel>();
- // new session freshly accepted, placed here for being added to the selector
+ /** new session freshly accepted, placed here for being added to the selector */
private final Queue<NioTcpSession> sessionsToConnect = new ConcurrentLinkedQueue<NioTcpSession>();
- // session to be removed of the selector
+ /** session to be removed of the selector */
private final Queue<NioTcpSession> sessionsToClose = new ConcurrentLinkedQueue<NioTcpSession>();
+
+ /** A queue used to store the sessions to be flushed */
+ private final Queue<NioTcpSession> flushingSessions = new ConcurrentLinkedQueue<NioTcpSession>();
private Selector selector;
@@ -162,8 +171,8 @@ public class NioSelectorProcessor implem
// map for finding the keys associated with a given server
private Map<ServerSocketChannel, SelectionKey> serverKey = new HashMap<ServerSocketChannel, SelectionKey>();
- // map for finding keys associated with a given session
- private Map<NioTcpSession, SelectionKey> sessionKey = new HashMap<NioTcpSession, SelectionKey>();
+ // map for finding read keys associated with a given session
+ private Map<NioTcpSession, SelectionKey> sessionReadKey = new HashMap<NioTcpSession, SelectionKey>();
@Override
public void run() {
@@ -209,7 +218,7 @@ public class NioSelectorProcessor implem
NioTcpSession session = sessionsToConnect.poll();
SelectionKey key = session.getSocketChannel().register(selector, SelectionKey.OP_READ);
key.attach(session);
- sessionKey.put(session, key);
+ sessionReadKey.put(session, key);
session.setConnected();
}
}
@@ -219,7 +228,7 @@ public class NioSelectorProcessor implem
while (!sessionsToClose.isEmpty()) {
NioTcpSession session = sessionsToClose.poll();
- SelectionKey key = sessionKey.remove(session);
+ SelectionKey key = sessionReadKey.remove(session);
key.cancel();
// needed ?
@@ -230,7 +239,7 @@ public class NioSelectorProcessor implem
log.debug("selecting...");
int readyCount = selector.select(SELECT_TIMEOUT);
log.debug("... done selecting : {}", readyCount);
-
+
if (readyCount > 0) {
// process selected keys
@@ -263,6 +272,22 @@ public class NioSelectorProcessor implem
}
}
+ if (key.isWritable()) {
+ log.debug("writable client {}", key);
+ NioTcpSession session = (NioTcpSession) key.attachment();
+ // write from the session write queue
+ WriteQueue queue = session.getWriteQueue();
+ do {
+ // get a write request from the queue
+ WriteRequest wr = queue.poll();
+ if (wr == null) {
+ break;
+ }
+ // TODO : write it
+
+ } while(!queue.isEmpty());
+
+ }
if (key.isAcceptable()) {
log.debug("acceptable new client {}", key);
@@ -278,6 +303,24 @@ public class NioSelectorProcessor implem
}
}
+
+ // registering session with data in the write queue for writing
+ while (!flushingSessions.isEmpty()) {
+ NioTcpSession session = flushingSessions.poll();
+ // a key registered for read ? (because we can have a Selector for reads and another for the writes
+ SelectionKey readKey = sessionReadKey.get(session);
+ if (readKey != null) {
+ readKey.cancel();
+ // register for read/write
+ SelectionKey key = session.getSocketChannel().register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
+ key.attach(session);
+ sessionReadKey.put(session, key);
+
+ } else {
+ SelectionKey key = session.getSocketChannel().register(selector, SelectionKey.OP_WRITE);
+ key.attach(session);
+ }
+ }
} catch (IOException e) {
log.error("IOException while selecting selector", e);
}
@@ -295,7 +338,10 @@ public class NioSelectorProcessor implem
@Override
public void flush(IoSession session) {
+ log.debug("scheduling session {} for writing",session.toString());
// add the session to the list of session to be registered for writing
// wake the selector
+ flushingSessions.add((NioTcpSession)session);
+ wakeupWorker();
}
}