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 2009/12/18 19:13:33 UTC
svn commit: r892331 - in
/mina/branches/3.0/core/src/main/java/org/apache/mina: service/
transport/socket/nio/
Author: jvermillard
Date: Fri Dec 18 18:13:32 2009
New Revision: 892331
URL: http://svn.apache.org/viewvc?rev=892331&view=rev
Log:
strategies for selector processor and a first implementation : OneThreadSelectorStrategy
Added:
mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java (contents, props changed)
- copied, changed from r889901, mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/OneThreadSelectorStrategy.java
mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java (with props)
mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java (with props)
Removed:
mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/OneThreadSelectorStrategy.java
Modified:
mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSelectorProcessor.java
mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSocketAcceptor.java
mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/SelectorStrategy.java
Copied: mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java (from r889901, mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/OneThreadSelectorStrategy.java)
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java?p2=mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java&p1=mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/OneThreadSelectorStrategy.java&r1=889901&r2=892331&rev=892331&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/OneThreadSelectorStrategy.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java Fri Dec 18 18:13:32 2009
@@ -1,30 +1,31 @@
-package org.apache.mina.transport.socket.nio;
+package org.apache.mina.service;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
+import java.io.IOException;
+import java.net.SocketAddress;
+
+import org.apache.mina.transport.socket.nio.SelectorStrategy;
public class OneThreadSelectorStrategy implements SelectorStrategy {
- private NioSelectorProcessor processor;
+ private SelectorProcessor processor;
- public OneThreadSelectorStrategy() {
- processor = new NioSelectorProcessor("uniqueSelector", this);
+ public OneThreadSelectorStrategy(SelectorFactory selectorFactory) {
+ this.processor = selectorFactory.getNewSelector("uniqueSelector", this);
}
@Override
- public void addServerSocket(ServerSocketChannel serverSocketChannel) {
- processor.add(serverSocketChannel);
+ public SelectorProcessor getSelectorForBindNewAddress() {
+ return processor;
}
@Override
- public void removeServerSocket(ServerSocketChannel serverSocketChannel) {
- processor.remove(serverSocketChannel);
+ public SelectorProcessor getSelectorForNewSession(SelectorProcessor acceptingProcessor) {
+ return processor;
}
@Override
- public void createSession(NioSelectorProcessor fromProcessor, SocketChannel newClient) {
- // send it back to the processor
-
+ public void unbind(SocketAddress address) throws IOException {
+ processor.unbind(address);
}
}
Propchange: mina/branches/3.0/core/src/main/java/org/apache/mina/service/OneThreadSelectorStrategy.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java?rev=892331&view=auto
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java (added)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java Fri Dec 18 18:13:32 2009
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.mina.service;
+
+import org.apache.log4j.Logger;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+import org.apache.mina.transport.socket.nio.SelectorStrategy;
+
+/**
+ * A factory used by {@link SelectorStrategy} for instantiating selectors when needed.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ *
+ */
+public class SelectorFactory {
+
+ private static final Logger LOG = Logger.getLogger(SelectorFactory.class);
+
+ private Constructor<?> constructor;
+
+ /**
+ * create a factory for the given {@link SelectorProcessor} type.
+ * @param selectorClass
+ */
+ public SelectorFactory(Class<SelectorProcessor> selectorClass) {
+ try {
+ constructor = selectorClass.getDeclaredConstructor(new Class<?>[]{String.class,SelectorStrategy.class});
+ } catch (NoSuchMethodException e) {
+ LOG.error("NoSuchMethodException while instantiating selector",e);
+ } catch (SecurityException e) {
+ LOG.error("SecurityException while instantiating selector",e);
+ }
+ }
+
+ public SelectorProcessor getNewSelector(String name,SelectorStrategy strategy) {
+ try {
+ return (SelectorProcessor)constructor.newInstance(name,strategy);
+ } catch (SecurityException e) {
+ LOG.error("SecurityException while instantiating selector",e);
+ } catch (IllegalArgumentException e) {
+ LOG.error("IllegalArgumentException while instantiating selector",e);
+ } catch (InstantiationException e) {
+ LOG.error("InstantiationException while instantiating selector",e);
+ } catch (IllegalAccessException e) {
+ LOG.error("IllegalAccessException while instantiating selector",e);
+ } catch (InvocationTargetException e) {
+ LOG.error("InvocationTargetException while instantiating selector",e);
+ }
+ return null;
+ }
+}
Propchange: mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java?rev=892331&view=auto
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java (added)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java Fri Dec 18 18:13:32 2009
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.mina.service;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+
+/**
+ * A processor in charge of a group of client session and server sockets.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ *
+ */
+public interface SelectorProcessor {
+
+ /**
+ * create a session for a freshly accepted client socket
+ * @param clientChannel
+ */
+ void createSession(Object clientSocket);
+
+ /**
+ * Bind and start processing this new server address
+ * @param address local address to bind
+ * @throws IOException exception thrown if any problem occurs while binding
+ */
+ void bindAndAcceptAddress(SocketAddress address) throws IOException;
+
+ /**
+ * Stop processing and unbind this server address
+ * @param address the local server address to unbind
+ * @throws IOException exception thrown if any problem occurs while unbinding
+ */
+ void unbind(SocketAddress address) throws IOException;
+}
Propchange: mina/branches/3.0/core/src/main/java/org/apache/mina/service/SelectorProcessor.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSelectorProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSelectorProcessor.java?rev=892331&r1=892330&r2=892331&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSelectorProcessor.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSelectorProcessor.java Fri Dec 18 18:13:32 2009
@@ -21,6 +21,7 @@
package org.apache.mina.transport.socket.nio;
import java.io.IOException;
+import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
@@ -28,9 +29,11 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.mina.IoSession;
+import org.apache.mina.service.SelectorProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,12 +42,14 @@
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*
*/
-public class NioSelectorProcessor {
+public class NioSelectorProcessor implements SelectorProcessor {
private SelectorStrategy strategy;
private Logger log;
+ private Map<SocketAddress,ServerSocketChannel> serverSocketChannels = new ConcurrentHashMap<SocketAddress, ServerSocketChannel>();
+
public NioSelectorProcessor(String name,SelectorStrategy strategy) {
this.strategy = strategy;
this.log = LoggerFactory.getLogger("SelectorProcessor["+name+"]");
@@ -75,13 +80,7 @@
serverToAdd.add(serverChannel);
wakeupWorker();
}
-
- public void remove(ServerSocketChannel serverChannel) {
- log.debug("removing a server channel "+serverChannel);
- serverToRemove.add(serverChannel);
- wakeupWorker();
- }
-
+
private Object workerLock = new Object();
private SelectorWorker worker = null;
@@ -98,6 +97,32 @@
}
}
+ @Override
+ public void bindAndAcceptAddress(SocketAddress address) throws IOException {
+ ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
+
+ serverSocketChannel.socket().bind(address);
+ serverSocketChannel.configureBlocking(false);
+ serverSocketChannels.put(address,serverSocketChannel);
+ add(serverSocketChannel);
+
+
+ }
+
+ @Override
+ public void unbind(SocketAddress address) throws IOException {
+ ServerSocketChannel channel = serverSocketChannels.get(address);
+ channel.socket().close();
+ serverSocketChannels.remove(channel);
+ log.debug("removing a server channel "+channel);
+ serverToRemove.add(channel);
+ wakeupWorker();
+ }
+
+ @Override
+ public void createSession(Object clientSocket) {
+ // TODO Auto-generated method stub
+ }
private class SelectorWorker extends Thread {
@@ -126,7 +151,6 @@
log.error("The server socket was already removed of the selector");
} else {
key.cancel();
-
}
}
}
@@ -152,6 +176,7 @@
SocketChannel newClientChannel = ((ServerSocketChannel)key.attachment()).accept();
log.debug("client accepted");
// and give it's to the strategy
+ strategy.getSelectorForNewSession(NioSelectorProcessor.this).createSession(newClientChannel);
}
}
@@ -169,5 +194,5 @@
}
}
}
- }
+ }
}
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSocketAcceptor.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSocketAcceptor.java?rev=892331&r1=892330&r2=892331&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSocketAcceptor.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/NioSocketAcceptor.java Fri Dec 18 18:13:32 2009
@@ -21,11 +21,9 @@
import java.io.IOException;
import java.net.SocketAddress;
-import java.nio.channels.ServerSocketChannel;
-import java.security.InvalidParameterException;
-import java.util.Map;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import org.apache.mina.service.AbstractIoAcceptor;
import org.slf4j.Logger;
@@ -41,8 +39,11 @@
static final Logger LOG = LoggerFactory.getLogger(NioSocketAcceptor.class);
+ // list of bound addresses
+ private Set<SocketAddress> addresses = Collections.synchronizedSet(new HashSet<SocketAddress>());
+
// map of the created selection keys, mainly used for cancelling them.
- private Map<SocketAddress,ServerSocketChannel> serverSocketChannels = new ConcurrentHashMap<SocketAddress, ServerSocketChannel>();
+ // private Map<SocketAddress,NioSelectorProcessor> serverSocketChannels = new ConcurrentHashMap<SocketAddress, NioSelectorProcessor>();
// the strategy for dispatching servers and client to selector threads.
private SelectorStrategy strategy;
@@ -58,23 +59,20 @@
for(SocketAddress address : localAddress) {
// check if the address is already bound
synchronized (this) {
- if (serverSocketChannels.containsKey(address)) {
+ if (addresses.contains(address)) {
throw new IOException("address "+address+" already bound");
}
- ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
LOG.debug("binding address {}",address);
- serverSocketChannel.socket().bind(address);
- serverSocketChannel.configureBlocking(false);
- serverSocketChannels.put(address,serverSocketChannel);
- // add the server socket to the acceptor selector thread
- strategy.addServerSocket(serverSocketChannel);
+ addresses.add(address);
+ NioSelectorProcessor processor = (NioSelectorProcessor)strategy.getSelectorForBindNewAddress();
+ processor.bindAndAcceptAddress(address);
}
}
}
@Override
public Set<SocketAddress> getLocalAddresses() {
- return serverSocketChannels.keySet();
+ return addresses;
}
@Override
@@ -82,21 +80,14 @@
for (SocketAddress socketAddress : localAddresses) {
LOG.debug("unbinding {}",socketAddress);
synchronized (this) {
- ServerSocketChannel channel = serverSocketChannels.get(socketAddress);
- if (channel == null) {
- throw new InvalidParameterException("localAddresses");
- }
- channel.socket().close();
- serverSocketChannels.remove(socketAddress);
- // remove the server socket form the selector accepting connections
- strategy.removeServerSocket(channel);
+ strategy.unbind(socketAddress);
}
}
}
@Override
public void unbindAll() throws IOException {
- for (SocketAddress socketAddress:serverSocketChannels.keySet()) {
+ for (SocketAddress socketAddress: addresses) {
unbind(socketAddress);
}
}
Modified: mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/SelectorStrategy.java
URL: http://svn.apache.org/viewvc/mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/SelectorStrategy.java?rev=892331&r1=892330&r2=892331&view=diff
==============================================================================
--- mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/SelectorStrategy.java (original)
+++ mina/branches/3.0/core/src/main/java/org/apache/mina/transport/socket/nio/SelectorStrategy.java Fri Dec 18 18:13:32 2009
@@ -19,20 +19,20 @@
*/
package org.apache.mina.transport.socket.nio;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
+import java.io.IOException;
+import java.net.SocketAddress;
+
+import org.apache.mina.service.SelectorProcessor;
/**
* Strategy for balancing server socket and client socket to different selecting/polling threads.
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
- *
*/
public interface SelectorStrategy {
-
- void addServerSocket(ServerSocketChannel serverSocketChannel);
- void removeServerSocket(ServerSocketChannel serverSocketChannel);
+ SelectorProcessor getSelectorForNewSession(SelectorProcessor acceptingProcessor);
- void createSession(NioSelectorProcessor fromProcessor,SocketChannel newClient);
+ SelectorProcessor getSelectorForBindNewAddress();
+ void unbind(SocketAddress address) throws IOException;
}