[socket-nif] Fixed (stream) recv

Fixed handling of closed in the recv function.
We still need to properly handle when we get
0 bytes of data for other types ock sockets
then stream (its valid for dgram for instance).

OTP-14831

2 files changed, 179 insertions, 31 deletions

diff --git a/lib/kernel/test/socket_client.erl b/lib/kernel/test/socket_client.erl
index 13e87f4109..a284777046 100644
--- a/lib/kernel/test/socket_client.erl
+++ b/lib/kernel/test/socket_client.erl
@@ -10,6 +10,9 @@
 
 -export([start/1]).
 
+-define(REQ, 0).
+-define(REP, 1).
+
 start(Port) ->
     start_tcp(Port).
 
@@ -19,12 +22,16 @@ start_tcp(Port) ->
 start(Domain, Type, Proto, Port) ->
     try do_init(Domain, Type, Proto) of
         Sock ->
-            connect(Sock, Domain, Port)
+            connect(Sock, Domain, Port),
+            %% Give the server some time...
+            p("wait some", []),
+            %% sleep(5000),
+            %% ok = socket:close(Sock),
+            send_loop(Sock)
     catch
-        throw:E:P ->
+        throw:E ->
             e("Failed initiate: "
-              "~n   Error: ~p"
-              "~n   Path:  ~p", [E, P])
+              "~n   Error: ~p", [E])
     end.
 
 do_init(Domain, Type, Proto) ->
@@ -58,11 +65,11 @@ connect(Sock, Domain, Port) ->
     SA = #{family => Domain,
            addr   => Addr,
            port   => Port},
-    i("try (socket) connect to ~p", [SA]),
+    i("try (socket) connect to:"
+      "~n   ~p", [SA]),
     case socket:connect(Sock, SA) of
         ok ->
             i("connected"),
-            send_loop(Sock),
             ok;
         {error, Reason} ->
             e("connect failure: "
@@ -74,22 +81,37 @@ connect(Sock, Domain, Port) ->
 send_loop(Sock) ->
     send_loop(Sock, 1).
 
-send_loop(Sock, N) ->
-    case socket:send(Sock, <<0:32, N:32, "hejsan">>) of
+send_loop(Sock, N) when (N =< 10) ->
+    i("try send request ~w", [N]),
+    Req = enc_req_msg(N, "hejsan"),
+    case socket:send(Sock, Req) of
         ok ->
-            case send:recv(Sock, 0) of
-                {ok, <<1:32, N:32, "hejsan">>} ->
-                    send_loop(Sock, N+1);
+            i("request ~w sent - now try read answer", [N]),
+            case socket:recv(Sock, 0) of
+                {ok, Msg} ->
+                    i("received ~w bytes of data", [size(Msg)]),
+                    case dec_msg(Msg) of
+                        {reply, N, Reply} ->
+                            i("received reply ~w: ~p", [N, Reply]),
+                            send_loop(Sock, N+1)
+                    end;
                 {error, RReason} ->
                     e("Failed recv response for request ~w: "
-                      "~n   ~p", [RReason]),
+                      "~n   ~p", [N, RReason]),
                     exit({failed_recv, RReason})
             end;
         {error, SReason} ->
             e("Failed send request ~w: "
               "~n   ~p", [SReason]),
             exit({failed_send, SReason})
-    end.
+    end;
+send_loop(Sock, _N) ->
+    i("we are done - close the socket when: "
+      "~n   ~p", [socket:info()]),
+    ok = socket:close(Sock),
+    i("we are done - socket closed when: "
+      "~n   ~p", [socket:info()]).
+
 
 which_addr(_Domain, []) ->
     throw(no_address);
@@ -106,6 +128,61 @@ which_addr2(Domain, [_|IFO]) ->
     which_addr2(Domain, IFO).
 
 
+%% ---
+
+enc_req_msg(N, Data) ->
+    enc_msg(?REQ, N, Data).
+
+enc_rep_msg(N, Data) ->
+    enc_msg(?REP, N, Data).
+
+enc_msg(Type, N, Data) when is_list(Data) ->
+    enc_msg(Type, N, list_to_binary(Data));
+enc_msg(Type, N, Data) 
+  when is_integer(Type) andalso is_integer(N) andalso is_binary(Data) ->
+    <<Type:32/integer, N:32/integer, Data/binary>>.
+    
+dec_msg(<<?REQ:32/integer, N:32/integer, Data/binary>>) ->
+    {request, N, Data};
+dec_msg(<<?REP:32/integer, N:32/integer, Data/binary>>) ->
+    {reply, N, Data}.
+
+
+%% ---
+
+sleep(T) ->
+    receive after T -> ok end.
+
+                     
+%% ---
+
+formated_timestamp() ->
+    format_timestamp(os:timestamp()).
+
+format_timestamp(Now) ->
+    N2T = fun(N) -> calendar:now_to_local_time(N) end,
+    format_timestamp(Now, N2T, true).
+
+format_timestamp({_N1, _N2, N3} = N, N2T, true) ->
+    FormatExtra = ".~.2.0w",
+    ArgsExtra   = [N3 div 10000],
+    format_timestamp(N, N2T, FormatExtra, ArgsExtra);
+format_timestamp({_N1, _N2, _N3} = N, N2T, false) ->
+    FormatExtra = "",
+    ArgsExtra   = [],
+    format_timestamp(N, N2T, FormatExtra, ArgsExtra).
+
+format_timestamp(N, N2T, FormatExtra, ArgsExtra) ->
+    {Date, Time}   = N2T(N),
+    {YYYY,MM,DD}   = Date,
+    {Hour,Min,Sec} = Time,
+    FormatDate =
+        io_lib:format("~.4w-~.2.0w-~.2.0w ~.2.0w:~.2.0w:~.2.0w" ++ FormatExtra,
+                      [YYYY, MM, DD, Hour, Min, Sec] ++ ArgsExtra),
+    lists:flatten(FormatDate).
+
+
+%% ---
 
 e(F, A) ->
     p("<ERROR> " ++ F, A).
@@ -116,5 +193,5 @@ i(F, A) ->
     p("*** " ++ F, A).
     
 p(F, A) ->
-    io:format("[client] " ++ F ++ "~n", A).
+    io:format("[client,~p][~s] " ++ F ++ "~n", [self(),formated_timestamp()|A]).
     
diff --git a/lib/kernel/test/socket_server.erl b/lib/kernel/test/socket_server.erl
index 0effc7c0ff..64bd6396e4 100644
--- a/lib/kernel/test/socket_server.erl
+++ b/lib/kernel/test/socket_server.erl
@@ -10,6 +10,9 @@
 
 -export([start/0]).
 
+-define(REQ, 0).
+-define(REP, 1).
+
 -record(handler, {socket, parent}).
 
 start() ->
@@ -82,14 +85,17 @@ which_addr2(Domain, [_|IFO]) ->
 
 
 accept_loop(LSock) ->
-    put(sname, "accept-loop"),
+    put(sname, "acceptor"),
     accept_loop(LSock, []).
 
 accept_loop(LSock, Handlers) ->
     i("try accept"),
     case socket:accept(LSock, infinity) of
         {ok, Sock} ->
-            i("accepted: ~p", [Sock]),
+            i("accepted: "
+              "~n   ~p"
+              "~nwhen"
+              "~n   ~p", [Sock, socket:info()]),
             case handle_accept_success(Sock) of
                 {ok, Handler} ->
                     accept_loop(LSock, [Handler|Handlers]);
@@ -127,6 +133,7 @@ handler_init(Parent, Socket) ->
     put(sname, "handler"),
     receive
         {handler, Parent, continue} ->
+            socket:setopt(Socket, otp, debug, true),
             handler_loop(#handler{parent = Parent,
                                   socket = Socket})
     end.
@@ -135,27 +142,90 @@ handler_continue(Handler) ->
     Handler ! {handler, self(), continue}.
 
 handler_loop(#handler{socket = Socket} = H) ->
-    case socket:read(Socket, 0) of
-        {ok, <<0:32, N:32, ReqData/binary>>} ->
-            i("received request ~w: "
-              "~n   ~p", [N, ReqData]),
-            Reply = <<1:32, N:32, ReqData/binary>>,
-            case socket:send(Socket, Reply) of
-                ok ->
-                    i("successfully sent reply ~w", [N]),
-                    handler_loop(H);
-                {error, SReason} ->
-                    e("failed sending reply ~w:"
-                      "~n   ~p", [N, SReason]),
-                    exit({failed_sending_reply, SReason})
+    case socket:recv(Socket, 0) of
+        {ok, Msg} when (size(Msg) =:= 0) ->
+            i("received empty msg - hickup? - try again", []),
+            handler_loop(H);
+        {ok, Msg} ->
+            i("received ~w bytes of data", [size(Msg)]),
+            case dec_msg(Msg) of
+                {request, N, Req} ->
+                    i("received request ~w: "
+                      "~n   ~p", [N, Req]),
+                    Reply = enc_rep_msg(N, "hoppsan"),
+                    case socket:send(Socket, Reply) of
+                        ok ->
+                            i("successfully sent reply ~w", [N]),
+                            handler_loop(H);
+                        {error, SReason} ->
+                            e("failed sending reply ~w:"
+                              "~n   ~p", [N, SReason]),
+                            exit({failed_sending_reply, SReason})
+                    end
             end;
+
+        {error, closed} ->
+            i("closed when"
+              "~n   ~p", [socket:info()]),
+            exit(normal);
+        
         {error, RReason} ->
             e("failed reading request: "
               "~n   ~p", [RReason]),
-            exit({failed_sending_reply, RReason})
+            exit({failed_reading_request, RReason})
     end.
             
     
+%% ---
+
+enc_req_msg(N, Data) ->
+    enc_msg(?REQ, N, Data).
+
+enc_rep_msg(N, Data) ->
+    enc_msg(?REP, N, Data).
+
+enc_msg(Type, N, Data) when is_list(Data) ->
+    enc_msg(Type, N, list_to_binary(Data));
+enc_msg(Type, N, Data) 
+  when is_integer(Type) andalso is_integer(N) andalso is_binary(Data) ->
+    <<Type:32/integer, N:32/integer, Data/binary>>.
+    
+dec_msg(<<?REQ:32/integer, N:32/integer, Data/binary>>) ->
+    {request, N, Data};
+dec_msg(<<?REP:32/integer, N:32/integer, Data/binary>>) ->
+    {reply, N, Data}.
+
+
+%% ---
+
+formated_timestamp() ->
+    format_timestamp(os:timestamp()).
+
+format_timestamp(Now) ->
+    N2T = fun(N) -> calendar:now_to_local_time(N) end,
+    format_timestamp(Now, N2T, true).
+
+format_timestamp({_N1, _N2, N3} = N, N2T, true) ->
+    FormatExtra = ".~.2.0w",
+    ArgsExtra   = [N3 div 10000],
+    format_timestamp(N, N2T, FormatExtra, ArgsExtra);
+format_timestamp({_N1, _N2, _N3} = N, N2T, false) ->
+    FormatExtra = "",
+    ArgsExtra   = [],
+    format_timestamp(N, N2T, FormatExtra, ArgsExtra).
+
+format_timestamp(N, N2T, FormatExtra, ArgsExtra) ->
+    {Date, Time}   = N2T(N),
+    {YYYY,MM,DD}   = Date,
+    {Hour,Min,Sec} = Time,
+    FormatDate =
+        io_lib:format("~.4w-~.2.0w-~.2.0w ~.2.0w:~.2.0w:~.2.0w" ++ FormatExtra,
+                      [YYYY, MM, DD, Hour, Min, Sec] ++ ArgsExtra),
+    lists:flatten(FormatDate).
+
+
+%% ---
+
 e(F, A) ->
     p("<ERROR> " ++ F, A).
 
@@ -168,5 +238,6 @@ p(F, A) ->
     p(get(sname), F, A).
 
 p(SName, F, A) ->
-    io:format("[server,~s] " ++ F ++ "~n", [SName|A]).
+    io:format("[server:~s,~p][~s] " ++ F ++ "~n", 
+              [SName,self(),formated_timestamp()|A]).