Possible bug in TcpNioConnection.java

[maaxiim]

Before I create a jira issue for this, I figured I'd post some information about it in case it is SUE rather than an actual bug.

Here is the configuration for the offending connection:

<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:context="http://www.springframework.org/schema/context" xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:oxm="http://www.springframework.org/schema/oxm" xmlns:util="http://www.springframework.org/schema/util"
	xmlns:int-http="http://www.springframework.org/schema/integration/http" xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
	xmlns:int-xml="http://www.springframework.org/schema/integration/xml"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
		http://www.springframework.org/schema/integration/http http://www.springframework.org/schema/integration/http/spring-integration-http-2.0.xsd
		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd
		http://www.springframework.org/schema/oxm http://www.springframework.org/schema/oxm/spring-oxm-3.0.xsd
		http://www.springframework.org/schema/integration/xml http://www.springframework.org/schema/integration/xml/spring-integration-xml-2.0.xsd
		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd">

	<bean id="mllpSerializer" class="....adapter.serializer.SccMLLPStreamingConverter" />

	<int-ip:tcp-connection-factory id="pcd01ConnectionFactoryClient" type="client"
		host="${sc.scc.adapter.pcd01.host}" port="${sc.scc.adapter.pcd01.port}" input-converter="mllpSerializer"
		output-converter="mllpSerializer" so-receive-buffer-size="65534" so-send-buffer-size="65534" />

	<int-ip:tcp-connection-factory id="pcd01ConnectionFactory" type="server"
		port="${sc.scc.adapter.pcd01.port}" input-converter="mllpSerializer" output-converter="mllpSerializer"
		so-receive-buffer-size="65534" so-send-buffer-size="65534" />

	<int-ip:tcp-inbound-channel-adapter id="pcd01InboundChannelAdapter"
		connection-factory="pcd01ConnectionFactory" channel="pcd01InboundChannel" />

	<int-ip:tcp-outbound-channel-adapter id="pcd01OutboundChannelAdapter" channel="pcd01OutboundChannel"
		connection-factory="pcd01ConnectionFactory" />

	<int:bridge input-channel="pcd01InboundChannel" output-channel="extInboundChannel" />

</beans>

Initially, I was using the using-nio="true" attribute for the connection factory. My app connects successfully and I can see the properly established connection:

  TCP    0.0.0.0:8101           0.0.0.0:0              LISTENING
  TCP    165.226.105.29:8101    10.1.140.195:3601      ESTABLISHED

When data arrives on the connection, I would see the following debug message in the log:

8609 [main] DEBUG com.siemens.soarian.sc.scc.adapter.manager.SccAdapterEngine  - Starting the adapter-engine
10391 [pool-2-thread-2] DEBUG org.springframework.integration.ip.tcp.connection.TcpNioConnection  - Read 2540 into raw buffer

but the dispatcher would never invoke my chain components on the message. This 2540 is exactly the correct length for an inbound message of the type I was expecting. So, I dug around in the debugger a little and see that the rawBuffer field in TcpNioConnection.java is 'java.nio.HeapByteBuffer[pos=2540 lim=61440 cap=61440]' which is as expected. Then, the code flips the buffer and calls

pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());

This call is where the code hangs; stepping into it, I see that it calls PipedInputStream.sink.receive (b, off(0), len(2540));

the receive() method copies the first 1024 bytes into the sink buffer, then loops. the head of the loops sees that the buffer is at capacity, and so calls awaitSpace() ; It broadcasts a notify to all waiting readers but no other threads are awoken. Since the reader will not be created until after the call to pipedOutputStream.write() (the taskexecutor, right?) the call will block indefinitely. I don't know if you agree with my diagnosis, but I'm wondering if you should move the executor.execute() call above the pipe write? Assuming the reader will block until data is available, I suspect it won't cause a race condition?


I almost forgot to mention, that when I remove the using-nio="true" attribute, the code functions as expected.

[maaxiim]

Thinking about it a little more, if you move the reader above the write, the reader will be delivered partial messages, so perhaps the solution is to set the pipeSize of the reader equal to the incoming message size before writing into the pipe?

[Gary Russell]

@maaxiim,

Are you using M6? Sounds like it from your suggested fix. This was indeed a problem in M6 and was fixed in M7; the problem you describe is correct, for large messages, the write would block before we checked to see if there was a reader.

The fix wasn't quite as simple as moving the executor.execute() above the write but you were heading in the right direction.

Anyway, RC1 was just released (http://www.springsource.org/spring-integration); please upgrade and let us know if you still see problems.

Thanks

[maaxiim]

Hey Gary, thanks for the update, I apologize, I forgot that I needed to upgrade M6 to M7. Doing it now and will let you know how I fare ...

[Gary Russell]

RC1 (release candidate 1) is hot off the press - why not go all the way and get all the good stuff since M7?

:-)

[maaxiim]

Am updating my eclipse user libraries to reference RC1 as we speak :-)

[maaxiim]

Gary,

thanks for your help, that fixes the problem.