Reactor-netty: `Broken pipe` is not throwed with netty-transport-native-epoll dependency

[flisky]

Note: This issue is also reported in spring-projects/spring-framework#22501

Expected behavior

The client aborting error should be throwed, and then logged by Spring.

Actual behavior

The client aborting error is not throwed with netty-transport-native-epoll dependency.

Steps to reproduce

• clone https://github.com/flisky/webflux-demo and run it
• curl http://localhost:8080/test -m 1 and check the console logging

Reactor Netty version

0.8.5

JVM version (e.g. java -version)

OpenJDK 8 & 11

OS version (e.g. uname -a)

Linux archlinux 4.20.12-arch1-1-ARCH #1 SMP PREEMPT Sat Feb 23 15:11:34 UTC 2019 x86_64 GNU/Linux

[violetagg]

@flisky The Broken pipe is just logged in the log file and not propagated.
Do you expect the same behaviour with the native transport? i.e. to see in the logs exception about channel close?

https://github.com/reactor/reactor-netty/blob/master/src/main/java/reactor/netty/http/server/HttpServerOperations.java#L504

23:52:22.757 [reactor-http-nio-2] DEBUG r.n.http.server.HttpServerOperations - [id: 0x9e4cf128, L:/0:0:0:0:0:0:0:1:8080 ! R:/0:0:0:0:0:0:0:1:65461] Failed flushing last frame
java.io.IOException: Broken pipe
    at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
    at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47)
    at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
    at sun.nio.ch.IOUtil.write(IOUtil.java:51)
    at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471)
    at io.netty.channel.socket.nio.NioSocketChannel.doWrite(NioSocketChannel.java:405)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.flush0(AbstractChannel.java:938)
    at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.flush0(AbstractNioChannel.java:360)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.flush(AbstractChannel.java:905)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.flush(DefaultChannelPipeline.java:1370)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:776)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:768)
    at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:749)
    at io.netty.handler.logging.LoggingHandler.flush(LoggingHandler.java:265)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:776)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:768)
    at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:749)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.flush(CombinedChannelDuplexHandler.java:533)
    at io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelOutboundHandlerAdapter.java:115)
    at io.netty.channel.CombinedChannelDuplexHandler.flush(CombinedChannelDuplexHandler.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:776)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:768)
    at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:749)
    at io.netty.channel.ChannelDuplexHandler.flush(ChannelDuplexHandler.java:117)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:776)
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:768)
    at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:749)
    at reactor.netty.channel.ChannelOperationsHandler$PublisherSender.lambda$onComplete$0(ChannelOperationsHandler.java:556)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:495)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
    at java.lang.Thread.run(Thread.java:748)

[flisky]

Yes, I do. Besides logging, maybe this exception should be also throwed out?

And under netty-transport-native-epoll, I'm confirmed the https://github.com/reactor/reactor-netty/blob/15934fd42465b38ef94d10acdfa60eb40e2a3d11/src/main/java/reactor/netty/channel/ChannelOperations.java#L186 is not called.

[violetagg]

@flisky
I made some changes to your example:

1. I added a wire logging so that I can see the traffic between the peers.

@Component
public class MyNettyWebServerCustomizer
        implements WebServerFactoryCustomizer<NettyReactiveWebServerFactory> {
    @Override
    public void customize(NettyReactiveWebServerFactory factory) {
        factory.addServerCustomizers(httpServer -> httpServer.wiretap(true));
    }
}

1. I added logging to your Publisher

    @GetMapping("/test")
    public Mono<String> test() {
        return Mono.delay(Duration.ofSeconds(3))
                .thenReturn("OK")
                .log();
    }

So there is a difference between NIO and the native transport when propagating the closed connection event.

The native transport propagates immediately the event, so that Reactor Netty doesn't start sending anything but just cancels the subscription to the provided Publisher.
log-kqueue.txt

Exactly after 1s the event for channelInactive was sent and we sent cancel

2019-03-06 16:45:54.950  INFO 40150 --- [r-http-kqueue-2] reactor.Mono.IgnoreThen.1                : | onSubscribe([Fuseable] MonoIgnoreThen.ThenIgnoreMain)
2019-03-06 16:45:54.951  INFO 40150 --- [r-http-kqueue-2] reactor.Mono.IgnoreThen.1                : | request(unbounded)
2019-03-06 16:45:54.953 DEBUG 40150 --- [r-http-kqueue-2] reactor.netty.http.server.HttpServer     : [id: 0x0800f5b2, L:/0:0:0:0:0:0:0:1%0:8080 - R:/0:0:0:0:0:0:0:1%0:60283] READ COMPLETE
2019-03-06 16:45:55.818 DEBUG 40150 --- [r-http-kqueue-2] reactor.netty.http.server.HttpServer     : [id: 0x0800f5b2, L:/0:0:0:0:0:0:0:1%0:8080 - R:/0:0:0:0:0:0:0:1%0:60283] READ COMPLETE
2019-03-06 16:45:55.819 DEBUG 40150 --- [r-http-kqueue-2] reactor.netty.http.server.HttpServer     : [id: 0x0800f5b2, L:/0:0:0:0:0:0:0:1%0:8080 ! R:/0:0:0:0:0:0:0:1%0:60283] INACTIVE
2019-03-06 16:45:55.820  INFO 40150 --- [r-http-kqueue-2] reactor.Mono.IgnoreThen.1                : | cancel()

The NIO doesn't do that so that we are able to consume the published items and then when we try to flush we receive Broken pipe.
log-nio.txt

2019-03-06 16:46:49.954  INFO 40165 --- [ctor-http-nio-2] reactor.Mono.IgnoreThen.1                : | onSubscribe([Fuseable] MonoIgnoreThen.ThenIgnoreMain)
2019-03-06 16:46:49.955  INFO 40165 --- [ctor-http-nio-2] reactor.Mono.IgnoreThen.1                : | request(unbounded)
...
2019-03-06 16:46:53.000 DEBUG 40165 --- [ctor-http-nio-2] reactor.netty.http.server.HttpServer     : [id: 0x488ece85, L:/0:0:0:0:0:0:0:1:8080 ! R:/0:0:0:0:0:0:0:1:60297] INACTIVE
2019-03-06 16:46:53.001 DEBUG 40165 --- [ctor-http-nio-2] reactor.netty.http.server.HttpServer     : [id: 0x488ece85, L:/0:0:0:0:0:0:0:1:8080 ! R:/0:0:0:0:0:0:0:1:60297] UNREGISTERED

[flisky]

Thanks for clarifying, and it makes sense.

Anything we can improve here? just curious.

[violetagg]

@flisky yes give us some time for discussion :)

[violetagg]

@flisky With all transports, Reactor Netty sends cancel signal. This is the behaviour since version 0.8.15 and 0.9.3 (#940).
You can attach a callback for the cancel signal.