Merge pull request #2272 from garazdawi/lukas/erts/fix_active_n_close_win32/ERL-960/OTP-15901

Fix {active,N} close race condition on windows

2 files changed, 78 insertions, 15 deletions

diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c
index 1bf6f9f1af..311c5fdd6a 100644
--- a/erts/emulator/drivers/common/inet_drv.c
+++ b/erts/emulator/drivers/common/inet_drv.c
@@ -1077,6 +1077,7 @@ typedef struct {
     long forced_events;           /* Mask of events that are forcefully signalled 
 				   on windows see winsock_event_select 
 				   for details */
+    int err;                      /* Keeps error code for closed socket */
     int send_would_block;       /* Last send attempt failed with "WOULDBLOCK" */
 #endif
     ErlDrvPort  port;           /* the port identifier */
@@ -9128,6 +9129,7 @@ static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol)
     desc->event_mask = 0;
 #ifdef __WIN32__
     desc->forced_events = 0;
+    desc->err = 0;
     desc->send_would_block = 0;
 #endif
     desc->port = port;
@@ -10977,7 +10979,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on)
 {
     int save_event_mask = desc->event_mask;
     
-    desc->forced_events = 0;
+    desc->forced_events &= FD_CLOSE;
     if (on) 
 	desc->event_mask |= flags;
     else
@@ -11029,7 +11031,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on)
 		TIMEVAL tmo = {0,0};
 		FD_SET fds;
 		int ret;
-		
+
 		FD_ZERO(&fds);
 		FD_SET(desc->s,&fds);
 		do_force = (select(desc->s+1,0,&fds,0,&tmo) > 0);
@@ -11046,7 +11048,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on)
 	    FD_SET fds;
 	    int ret;
 	    unsigned long arg;
-	  
+
 	    FD_ZERO(&fds);
 	    FD_SET(desc->s,&fds);
 	    ret = select(desc->s+1,&fds,0,0,&tmo);
@@ -11085,13 +11087,16 @@ static void tcp_inet_event(ErlDrvData e, ErlDrvEvent event)
 	goto error;
     }
 
-    DEBUGF((" => event=%02X, mask=%02X\r\n",
-	    netEv.lNetworkEvents, desc->inet.event_mask));
+    DEBUGF((" => event=%02X, mask=%02X, forced=%02X\r\n",
+	    netEv.lNetworkEvents, desc->inet.event_mask,
+            desc->inet.forced_events));
 
     /* Add the forced events. */
-
     netEv.lNetworkEvents |= desc->inet.forced_events;
 
+    if (desc->inet.forced_events & FD_CLOSE)
+        netEv.iErrorCode[FD_CLOSE_BIT] = desc->inet.err;
+
     /*
      * Calling WSAEventSelect() with a mask of 0 doesn't always turn off
      * all events.  To avoid acting on events we don't want, we mask
@@ -11111,16 +11116,29 @@ static void tcp_inet_event(ErlDrvData e, ErlDrvEvent event)
 	    goto error;
 	}
 	if (netEv.lNetworkEvents & FD_CLOSE) {
-	    /*
+
+            /* We may not get any more FD_CLOSE events so we
+               keep this event and always signal it from
+               this moment on. */
+            if ((desc->inet.forced_events & FD_CLOSE) == 0) {
+                desc->inet.forced_events |= FD_CLOSE;
+                desc->inet.err = netEv.iErrorCode[FD_CLOSE_BIT];
+            }
+
+            /*
 	     * We must loop to read out the remaining packets (if any).
 	     */
 	    for (;;) {
-		DEBUGF(("Retrying read due to closed port\r\n"));
-		/* XXX The buffer will be thrown away on error (empty que).
-		   Possible SMP FIXME. */
-		if (!desc->inet.active && (desc->inet.opt) == NULL) {
-		    goto error;
-		}
+
+                /* if passive and no subscribers, break loop */
+                if (!desc->inet.active && desc->inet.opt == NULL) {
+                    /* do not trigger close event when socket is
+                       transitioned to passive */
+                    netEv.lNetworkEvents &= ~FD_CLOSE;
+                    break;
+                }
+
+		DEBUGF(("Retrying read due to FD_CLOSE\r\n"));
 		if (tcp_inet_input(desc, event) < 0) {
 		    goto error;
 		}
diff --git a/lib/kernel/test/gen_tcp_misc_SUITE.erl b/lib/kernel/test/gen_tcp_misc_SUITE.erl
index 8d25ac5dde..421510f9d6 100644
--- a/lib/kernel/test/gen_tcp_misc_SUITE.erl
+++ b/lib/kernel/test/gen_tcp_misc_SUITE.erl
@@ -25,6 +25,7 @@
 	 init_per_group/2,end_per_group/2, 
 	 controlling_process/1, controlling_process_self/1,
 	 no_accept/1, close_with_pending_output/1, active_n/1,
+         active_n_closed/1,
 	 data_before_close/1,
 	 iter_max_socks/0, iter_max_socks/1,
 	 get_status/1,
@@ -74,7 +75,7 @@ suite() ->
 all() -> 
     [controlling_process, controlling_process_self, no_accept,
      close_with_pending_output, data_before_close,
-     iter_max_socks, passive_sockets, active_n,
+     iter_max_socks, passive_sockets, active_n, active_n_closed,
      accept_closed_by_other_process, otp_3924, closed_socket,
      shutdown_active, shutdown_passive, shutdown_pending,
      show_econnreset_active, show_econnreset_active_once,
@@ -2619,7 +2620,51 @@ active_once_closed(Config) when is_list(Config) ->
 	     ok = inet:setopts(A,[{active,once}]),
 	     ok = receive {tcp_closed, A} -> ok after 1000 -> error end
      end)().
-   
+
+%% Check that active n and tcp_close messages behave as expected.
+active_n_closed(Config) when is_list(Config) ->
+    {ok, L} = gen_tcp:listen(0, [binary, {active, false}]),
+
+    P = self(),
+
+    {ok,Port} = inet:port(L),
+
+    spawn_link(fun() ->
+                       Payload = <<0:50000/unit:8>>,
+                       Cnt = 10000,
+                       P ! {size,Cnt * byte_size(Payload)},
+                       {ok, S} = gen_tcp:connect("localhost", Port, [binary, {active, false}]),
+                       _ = [gen_tcp:send(S, Payload) || _ <- lists:seq(1, Cnt)],
+                       gen_tcp:close(S)
+          end),
+
+    receive {size,SendSize} -> SendSize end,
+    {ok, S} = gen_tcp:accept(L),
+    inet:setopts(S, [{active, 10}]),
+    RecvSize =
+        (fun Server(Size) ->
+                 receive
+                     {tcp, S, Bin} ->
+                         Server(byte_size(Bin) + Size);
+                     {tcp_closed, S} ->
+                         Size;
+                     {tcp_passive, S} ->
+                         inet:setopts(S, [{active, 10}]),
+                         Server(Size);
+                     Msg ->
+                         io:format("~p~n", [Msg]),
+                         Server(Size)
+                 end
+         end)(0),
+
+    gen_tcp:close(L),
+
+    if SendSize =:= RecvSize ->
+            ok;
+       true ->
+            ct:fail("Send and Recv size not equal: ~p ~p",[SendSize, RecvSize])
+    end.
+
 %% Test the send_timeout socket option.
 send_timeout(Config) when is_list(Config) ->
     Dir = filename:dirname(code:which(?MODULE)),