Netty removal debug

Author: neocoretechs

src/main/java/org/ros/internal/node/topic/SubscriberHandshakeHandler.java

@@ -2,9 +2,6 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-//import org.jboss.netty.channel.ChannelHandlerContext;
-//import org.jboss.netty.channel.ChannelPipeline;
-//import org.jboss.netty.channel.MessageEvent;
 import org.ros.internal.transport.BaseClientHandshakeHandler;
 import org.ros.internal.transport.ChannelHandlerContext;
 import org.ros.internal.transport.ChannelPipeline;
@@ -16,13 +13,15 @@
 import org.ros.node.topic.Subscriber;
 
 import java.io.IOException;
-import java.net.SocketAddress;
 import java.util.concurrent.ExecutorService;
 
 /**
  * Performs a handshake with the connected {@link Publisher} and connects the
  * {@link Publisher} to the {@link IncomingMessageQueue} on success.
  * 
+ * In the AsynchTcpWorker thread that handles the read for each channel, it strobes the pipeline
+ * with the read notification and the handler here takes care of the processing.
+ * 
  * @author jg
  * 
  * @param <T>
@@ -44,6 +43,8 @@ public SubscriberHandshakeHandler(ConnectionHeader outgoingConnectionHeader,
 
   @Override
   protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandlerContext ctx) {
+	if( DEBUG )
+		log.info("SubscriberHandshakeHandler.onSuccess:"+ctx+" "+incomingConnectionHeader);
     ChannelPipeline pipeline = ctx.pipeline();
     pipeline.remove(SubscriberHandshakeHandler.this);
     NamedChannelHandler namedChannelHandler = incomingMessageQueue.getMessageReceiver();

src/main/java/org/ros/internal/transport/ChannelHandlerContext.java

@@ -108,7 +108,7 @@ public interface ChannelHandlerContext {
      * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      */
-    Future<Integer> read();
+    Future<Integer> read(ByteBuffer buf);
 
     /**
      * Request to write a message via this {@link ChannelHandlerContext} through the {@link ChannelPipeline}.
@@ -149,7 +149,8 @@ public interface ChannelHandlerContext {
      * 
      */
     AsynchronousChannelGroup getChannelGroup();
-    
+
+
 
 
 }

src/main/java/org/ros/internal/transport/ChannelHandlerContextImpl.java

@@ -30,14 +30,11 @@ public class ChannelHandlerContextImpl implements ChannelHandlerContext {
 	AsynchronousSocketChannel channel;
 	ChannelPipeline pipeline;
 
-	ByteBuffer buf;
-	int MAXBUF = 2000000;
 
 	public ChannelHandlerContextImpl(AsynchronousChannelGroup grp, AsynchronousSocketChannel ch, Executor exc) {
 		channelGroup = grp;
 		channel = ch;
 		executor = exc;
-		buf = ByteBuffer.allocate(MAXBUF);
 		pipeline = new ChannelPipelineImpl(this);
 	}
 
@@ -86,7 +83,7 @@ public void close() throws IOException {
 
 
 	@Override
-	public Future<Integer> read() {
+	public Future<Integer> read(ByteBuffer buf) {
 		return channel.read(buf);	
 	}
 
@@ -137,4 +134,5 @@ public String toString() {
 		return new String("ChannelHandlerContext:"+channel+" "+channelGroup+" "+executor+" "+pipeline);
 	}
 
+	
 }

src/main/java/org/ros/internal/transport/ChannelPipeline.java

@@ -9,6 +9,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingDeque;
 
+import org.ros.exception.RosRuntimeException;
+
 /**
  * Implementation of the ChannelPipeline interface to process the requests via the pluggable ChannelHandlers
  * named in the queue. The pipeline is per ChannelHandlerContext, per channel as it represents state in the form of the
@@ -37,26 +39,48 @@ public Iterator<Entry<String, ChannelHandler>> iterator() {
 	@Override
 	public ChannelPipeline addFirst(String name, ChannelHandler handler) {
 		queue.addFirst(new AbstractMap.SimpleEntry<String, ChannelHandler>(name, handler));
+		try {
+			queue.peekFirst().getValue().handlerAdded(ctx);
+		} catch (Exception e) {
+			throw new RosRuntimeException(e);
+		}
 		return this;
 	}
 
 
 	@Override
 	public ChannelPipeline addLast(String name, ChannelHandler handler) {
 		queue.addLast(new AbstractMap.SimpleEntry<String, ChannelHandler>(name, handler));
+		try {
+			queue.peekLast().getValue().handlerAdded(ctx);
+		} catch (Exception e) {
+			throw new RosRuntimeException(e);
+		}
 		return this;
 	}
 
 
-
+	/**
+	 * Remove and fire event handlerRemoved to handler after removed one
+	 */
 	@Override
 	public ChannelPipeline remove(ChannelHandler handler) {
 		Iterator<Map.Entry<String,ChannelHandler>> it = iterator();
+		boolean found = false;
 		while(it.hasNext()) {
 			Map.Entry<String,ChannelHandler> me = it.next();
-			if( me.getValue().equals(handler) ) {
+			if( !found && me.getValue().equals(handler) ) {
 				it.remove();
-				break;
+				found = true;
+				continue;
+			}
+			if( found ) {
+				try {
+						me.getValue().handlerRemoved(ctx);
+						break;
+				} catch (Exception e) {
+					throw new RosRuntimeException(e);
+				}
 			}
 		}
 		return this;
@@ -65,24 +89,40 @@ public ChannelPipeline remove(ChannelHandler handler) {
 	@Override
 	public ChannelHandler remove(String name) {
 		Iterator<Map.Entry<String,ChannelHandler>> it = iterator();
-		Map.Entry<String,ChannelHandler> me = null;
+		ChannelHandler val = null;
+		boolean found = false;
 		while(it.hasNext()) {
-			me = it.next();
-			if( me.getKey().equals(name) ) {
+			Map.Entry<String,ChannelHandler> me = it.next();
+			if( !found && me.getKey().equals(name) ) {
+				val = me.getValue();
 				it.remove();
-				break;
+				found = true;
+				continue;
+			}
+			if( found ) {
+				try {
+						me.getValue().handlerRemoved(ctx);
+						break;
+				} catch (Exception e) {
+					throw new RosRuntimeException(e);
+				}
 			}
 		}
-		if( me == null)
-			return null;
-		return me.getValue();
+		return val;
 	}
 
 
 	@Override
 	public ChannelHandler removeFirst() {
 		Map.Entry<String,ChannelHandler> me = queue.removeFirst();
 		if( me == null )  return null;
+		if( queue.isEmpty())
+			return me.getValue();
+		try {
+			queue.peekFirst().getValue().handlerRemoved(ctx);
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		}
 		return me.getValue();
 	}
 

src/main/java/org/ros/internal/transport/ChannelPipelineImpl.java

@@ -1,19 +1,3 @@
-/*
- * Copyright (C) 2011 Google Inc.
- * 
- * Licensed 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.ros.internal.transport.queue;
 
 import org.ros.concurrent.CircularBlockingDeque;
@@ -24,7 +8,8 @@
 import java.util.concurrent.ExecutorService;
 
 /**
- * @author damonkohler@google.com (Damon Kohler)
+ * Created in default subscriber to handle the incoming messages
+ * @author jg
  */
 public class IncomingMessageQueue<T> {
 
@@ -37,7 +22,7 @@ public class IncomingMessageQueue<T> {
    * {@link IncomingMessageQueue#addListener(MessageListener, int)} which are
    * consumed by user provided {@link MessageListener}s.
    */
-  private static final int DEQUE_CAPACITY = 16;
+  private static final int DEQUE_CAPACITY = 16384;
 
   private final MessageReceiver<T> messageReceiver;
   private final MessageDispatcher<T> messageDispatcher;

src/main/java/org/ros/internal/transport/queue/IncomingMessageQueue.java

@@ -2,7 +2,6 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousChannelGroup;
 import java.nio.channels.Channel;
 import java.nio.channels.CompletionHandler;
 import java.util.Iterator;
@@ -11,35 +10,29 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
-//import org.jboss.netty.buffer.ChannelBuffer;
-//import org.jboss.netty.channel.Channel;
-//import org.jboss.netty.channel.group.ChannelGroup;
-//import org.jboss.netty.channel.group.ChannelGroupFuture;
-//import org.jboss.netty.channel.group.ChannelGroupFutureListener;
-//import org.jboss.netty.channel.group.DefaultChannelGroup;
-
 import org.ros.concurrent.CancellableLoop;
 import org.ros.concurrent.CircularBlockingDeque;
-import org.ros.exception.RosRuntimeException;
 import org.ros.internal.message.MessageBufferPool;
 import org.ros.internal.message.MessageBuffers;
 import org.ros.internal.system.Utility;
 import org.ros.internal.transport.ChannelHandlerContext;
 
-
-
 /**
+ * The outgoing message queue of a publisher processing type T messages.
+ * A writer will be spun up in the executor that processes a deque with message entries,
+ * serializing them to a bytebuffer for outbound asynch channel transport.
+ * Each publisher created generates this writer which addresses all channels in the context array
+ * it was created with.
+ * @author jg
  */
 public class OutgoingMessageQueue<T> {
 
   private static final boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(OutgoingMessageQueue.class);
 
-  private static final int DEQUE_CAPACITY = 16;
+  private static final int DEQUE_CAPACITY = 16384;
 
   private final CircularBlockingDeque<T> deque;
-  private final AsynchronousChannelGroup channelGroup;
   private final Writer writer;
   private final MessageBufferPool messageBufferPool;
   private final ByteBuffer latchedBuffer;
@@ -49,39 +42,43 @@ public class OutgoingMessageQueue<T> {
   private T latchedMessage;
 
   private List<ChannelHandlerContext> channels;
-
+  /**
+   * This class is submitted to the executor to process the deque entries
+   * and serialize them to the latched buffer for outbound publisher channel
+   * @author jg
+   *
+   */
   private final class Writer extends CancellableLoop {
     @Override
     public void loop() throws InterruptedException {
       T message = deque.takeFirst();
       final ByteBuffer buffer = messageBufferPool.acquire();
+      //messageBufferPool.release(buffer);
+      latchedBuffer.clear();
       Utility.serialize(message, buffer);
       if (DEBUG  ) {
-        log.info(String.format("Writing %d bytes.", buffer.limit()));
+        log.info(String.format("Writing %d bytes.", buffer.position()));
       }
-      // we have to wait until the write
-      // operation is complete before returning the buffer to the pool.
       Iterator<ChannelHandlerContext> it = channels.iterator();
       while(it.hasNext()) {
     	  ChannelHandlerContext ctx = it.next();
     	  ctx.write(buffer, new CompletionHandler<Integer, Void>() {
-        @Override
-        public void completed(Integer a, Void b) {
-          messageBufferPool.release(buffer);
-        }
-		@Override
-		public void failed(Throwable arg0, Void arg1) {
-			log.error("Failed write");
-			throw new RosRuntimeException(arg0);
-		}
-      });
+			@Override
+			public void completed(Integer arg0, Void arg1) {
+				messageBufferPool.release(buffer);
+			}
+
+			@Override
+			public void failed(Throwable arg0, Void arg1) {
+				throw new RuntimeException(arg0);
+			}  
+    	  });
       }
     }
   }
 
   public OutgoingMessageQueue(ExecutorService executorService, List<ChannelHandlerContext> ctxs) throws IOException {
     deque = new CircularBlockingDeque<T>(DEQUE_CAPACITY);
-    channelGroup = AsynchronousChannelGroup.withThreadPool(executorService);
     writer = new Writer();
     messageBufferPool = new MessageBufferPool();
     latchedBuffer = MessageBuffers.dynamicBuffer();
@@ -105,7 +102,7 @@ public boolean getLatchMode() {
    */
   public void add(T message) {
     deque.addLast(message);
-    setLatchedMessage(message);
+    //setLatchedMessage(message);
   }
 
   private void setLatchedMessage(T message) {
@@ -119,7 +116,6 @@ private void setLatchedMessage(T message) {
    */
   public void shutdown() {
     writer.cancel();
-    channelGroup.shutdown();
   }
 
 
@@ -142,7 +138,4 @@ public int getNumberOfChannels() {
     return channels.size();
   }
 
-  public AsynchronousChannelGroup getChannelGroup() {
-    return channelGroup;
-  }
 }

src/main/java/org/ros/internal/transport/queue/OutgoingMessageQueue.java

@@ -67,7 +67,7 @@ public void run() {
 				// inject the handlers, start handshake
 				tcpserver.getFactoryStack().inject(ctx);
 				ctx.pipeline().fireChannelRegistered(); 
-                AsynchTCPWorker uworker = new AsynchTCPWorker(channel.get(), tcpserver);
+                AsynchTCPWorker uworker = new AsynchTCPWorker(ctx, channel.get());
                 exc.execute(uworker);  
                 if( DEBUG ) {
                     	log.info("ROS Asynch transport server worker starting");