diff options
author | Zandra Hird <zandra@erlang.org> | 2015-06-02 15:40:58 +0200 |
---|---|---|
committer | Zandra Hird <zandra@erlang.org> | 2015-06-02 15:40:58 +0200 |
commit | b4774c9d55a1606a1abc726e0f7ce563065284b3 (patch) | |
tree | 8f6a8834dc96f5e47cab64da9337efbb90f076f1 | |
parent | 65bf43e82a8ec731aafdbac74d41d65b63c7f49c (diff) | |
parent | 25bd6312491fc9153a16f74b5d1d39609426ae60 (diff) | |
download | otp-b4774c9d55a1606a1abc726e0f7ce563065284b3.tar.gz otp-b4774c9d55a1606a1abc726e0f7ce563065284b3.tar.bz2 otp-b4774c9d55a1606a1abc726e0f7ce563065284b3.zip |
Merge branch 'nybek/fix_gen_tcp_shutdown'
OTP-12797
-rw-r--r-- | erts/emulator/drivers/common/inet_drv.c | 58 | ||||
-rw-r--r-- | erts/preloaded/src/prim_inet.erl | 21 | ||||
-rw-r--r-- | lib/kernel/doc/src/gen_tcp.xml | 15 | ||||
-rw-r--r-- | lib/kernel/test/gen_tcp_api_SUITE.erl | 32 |
4 files changed, 99 insertions, 27 deletions
diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c index 5196eb51c6..e001f31932 100644 --- a/erts/emulator/drivers/common/inet_drv.c +++ b/erts/emulator/drivers/common/inet_drv.c @@ -293,6 +293,10 @@ static BOOL (WINAPI *fpSetHandleInformation)(HANDLE,DWORD,DWORD); static unsigned long zero_value = 0; static unsigned long one_value = 1; +#define TCP_SHUT_WR SD_SEND +#define TCP_SHUT_RD SD_RECEIVE +#define TCP_SHUT_RDWR SD_BOTH + #elif defined (__OSE__) /* @@ -421,6 +425,10 @@ typedef unsigned long u_long; inet_driver_select((d), (flags), (onoff)); \ } while(0) +#define TCP_SHUT_WR SHUT_WR +#define TCP_SHUT_RD SHUT_RD +#define TCP_SHUT_RDWR SHUT_RDWR + #else /* !__OSE__ && !__WIN32__ */ #include <sys/time.h> @@ -691,6 +699,9 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n) inet_driver_select((d)->port, (ErlDrvEvent)(long)(d)->event, (flags), (onoff)); \ } while(0) +#define TCP_SHUT_WR SHUT_WR +#define TCP_SHUT_RD SHUT_RD +#define TCP_SHUT_RDWR SHUT_RDWR #endif /* !__WIN32__ && !__OSE__ */ @@ -820,6 +831,10 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n) #define TCP_ADDF_CLOSE_SENT 2 /* Close sent (active mode only) */ #define TCP_ADDF_DELAYED_CLOSE_RECV 4 /* If receive fails, report {error,closed} (passive mode) */ #define TCP_ADDF_DELAYED_CLOSE_SEND 8 /* If send fails, report {error,closed} (passive mode) */ +#define TCP_ADDF_PENDING_SHUT_WR 16 /* Call shutdown(sock, SHUT_WR) when queue empties */ +#define TCP_ADDF_PENDING_SHUT_RDWR 32 /* Call shutdown(sock, SHUT_RDWR) when queue empties */ +#define TCP_ADDF_PENDING_SHUTDOWN \ + (TCP_ADDF_PENDING_SHUT_WR | TCP_ADDF_PENDING_SHUT_RDWR) /* *_REQ_* replies */ #define INET_REP_ERROR 0 @@ -1407,6 +1422,8 @@ static int tcp_sendv(tcp_descriptor* desc, ErlIOVec* ev); static int tcp_recv(tcp_descriptor* desc, int request_len); static int tcp_deliver(tcp_descriptor* desc, int len); +static int tcp_shutdown_error(tcp_descriptor* desc, int err); + static int tcp_inet_output(tcp_descriptor* desc, HANDLE event); static int tcp_inet_input(tcp_descriptor* desc, HANDLE event); @@ -9473,10 +9490,18 @@ static ErlDrvSSizeT tcp_inet_ctl(ErlDrvData e, unsigned int cmd, return ctl_error(EINVAL, rbuf, rsize); } how = buf[0]; - if (sock_shutdown(INETP(desc)->s, how) == 0) { + if (how != TCP_SHUT_RD && driver_sizeq(desc->inet.port) > 0) { + if (how == TCP_SHUT_WR) { + desc->tcp_add_flags |= TCP_ADDF_PENDING_SHUT_WR; + } else if (how == TCP_SHUT_RDWR) { + desc->tcp_add_flags |= TCP_ADDF_PENDING_SHUT_RDWR; + } return ctl_reply(INET_REP_OK, NULL, 0, rbuf, rsize); - } else { + } + if (IS_SOCKET_ERROR(sock_shutdown(INETP(desc)->s, how))) { return ctl_error(sock_errno(), rbuf, rsize); + } else { + return ctl_reply(INET_REP_OK, NULL, 0, rbuf, rsize); } } default: @@ -9613,6 +9638,8 @@ static void tcp_inet_commandv(ErlDrvData e, ErlIOVec* ev) else inet_reply_error(INETP(desc), ENOTCONN); } + else if (desc->tcp_add_flags & TCP_ADDF_PENDING_SHUTDOWN) + tcp_shutdown_error(desc, EPIPE); else if (tcp_sendv(desc, ev) == 0) inet_reply_ok(INETP(desc)); DEBUGF(("tcp_inet_commandv(%ld) }\r\n", (long)desc->inet.port)); @@ -10506,7 +10533,7 @@ static int tcp_inet_input(tcp_descriptor* desc, HANDLE event) return ret; } -static int tcp_send_error(tcp_descriptor* desc, int err) +static int tcp_send_or_shutdown_error(tcp_descriptor* desc, int err) { /* * If the port is busy, we must do some clean-up before proceeding. @@ -10563,6 +10590,16 @@ static int tcp_send_error(tcp_descriptor* desc, int err) return -1; } +static int tcp_send_error(tcp_descriptor* desc, int err) +{ + return tcp_send_or_shutdown_error(desc, err); +} + +static int tcp_shutdown_error(tcp_descriptor* desc, int err) +{ + return tcp_send_or_shutdown_error(desc, err); +} + /* ** Send non-blocking vector data */ @@ -10763,6 +10800,19 @@ static int tcp_send(tcp_descriptor* desc, char* ptr, ErlDrvSizeT len) return 0; } +/* shutdown on the socket: +** Assume caller has confirmed TCP_ADDF_PENDING_SHUTDOWN is set. +*/ +static void tcp_shutdown_async(tcp_descriptor* desc) +{ + int how; + + how = (desc->tcp_add_flags & TCP_ADDF_PENDING_SHUT_WR) ? + TCP_SHUT_WR : TCP_SHUT_RDWR; + if (IS_SOCKET_ERROR(sock_shutdown(INETP(desc)->s, how))) + tcp_shutdown_error(desc, sock_errno()); +} + #ifdef __OSE__ static void tcp_inet_drv_output_ose(ErlDrvData data, ErlDrvEvent event) @@ -10891,6 +10941,8 @@ static int tcp_inet_output(tcp_descriptor* desc, HANDLE event) if ((iov = driver_peekq(ix, &vsize)) == NULL) { sock_select(INETP(desc), FD_WRITE, 0); send_empty_out_q_msgs(INETP(desc)); + if (desc->tcp_add_flags & TCP_ADDF_PENDING_SHUTDOWN) + tcp_shutdown_async(desc); goto done; } vsize = vsize > MAX_VSIZE ? MAX_VSIZE : vsize; diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl index 79ff013c77..622e1be869 100644 --- a/erts/preloaded/src/prim_inet.erl +++ b/erts/preloaded/src/prim_inet.erl @@ -127,37 +127,18 @@ drv2protocol(_) -> undefined. %% TODO: shutdown equivalent for SCTP %% shutdown(S, read) when is_port(S) -> - shutdown_2(S, 0); + shutdown_1(S, 0); shutdown(S, write) when is_port(S) -> shutdown_1(S, 1); shutdown(S, read_write) when is_port(S) -> shutdown_1(S, 2). shutdown_1(S, How) -> - case subscribe(S, [subs_empty_out_q]) of - {ok,[{subs_empty_out_q,N}]} when N > 0 -> - shutdown_pend_loop(S, N); %% wait for pending output to be sent - _Other -> ok - end, - shutdown_2(S, How). - -shutdown_2(S, How) -> case ctl_cmd(S, ?TCP_REQ_SHUTDOWN, [How]) of {ok, []} -> ok; {error,_}=Error -> Error end. -shutdown_pend_loop(S, N0) -> - receive - {empty_out_q,S} -> ok - after ?INET_CLOSE_TIMEOUT -> - case getstat(S, [send_pend]) of - {ok,[{send_pend,N0}]} -> ok; - {ok,[{send_pend,N}]} -> shutdown_pend_loop(S, N); - _ -> ok - end - end. - %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% %% CLOSE(insock()) -> ok diff --git a/lib/kernel/doc/src/gen_tcp.xml b/lib/kernel/doc/src/gen_tcp.xml index 820ecd1e30..71ef5cd48f 100644 --- a/lib/kernel/doc/src/gen_tcp.xml +++ b/lib/kernel/doc/src/gen_tcp.xml @@ -347,11 +347,22 @@ do_recv(Sock, Bs) -> </func> <func> <name name="shutdown" arity="2"/> - <fsummary>Immediately close a socket</fsummary> + <fsummary>Asynchronously close a socket</fsummary> <desc> - <p>Immediately close a socket in one or two directions.</p> + <p>Close a socket in one or two directions.</p> <p><c><anno>How</anno> == write</c> means closing the socket for writing, reading from it is still possible.</p> + <p>If <c><anno>How</anno> == read</c>, or there is no outgoing + data buffered in the <c><anno>Socket</anno></c> port, + then the socket is shutdown immediately and any error encountered + is returned in <c><anno>Reason</anno></c>.</p> + <p>If there is data buffered in the socket port, then the attempt + to shutdown the socket is postponed until that data is written to the + kernel socket send buffer. Any errors encountered will result + in the socket being closed and <c>{error, closed}</c> being returned + on the next + <seealso marker="gen_tcp#recv/2">recv/2</seealso> or + <seealso marker="gen_tcp#send/2">send/2</seealso>.</p> <p>To be able to handle that the peer has done a shutdown on the write side, the <c>{exit_on_close, false}</c> option is useful.</p> diff --git a/lib/kernel/test/gen_tcp_api_SUITE.erl b/lib/kernel/test/gen_tcp_api_SUITE.erl index c27d265550..4a527e2f51 100644 --- a/lib/kernel/test/gen_tcp_api_SUITE.erl +++ b/lib/kernel/test/gen_tcp_api_SUITE.erl @@ -32,6 +32,7 @@ t_connect_bad/1, t_recv_timeout/1, t_recv_eof/1, t_shutdown_write/1, t_shutdown_both/1, t_shutdown_error/1, + t_shutdown_async/1, t_fdopen/1, t_fdconnect/1, t_implicit_inet6/1]). -export([getsockfd/0,closesockfd/1]). @@ -41,7 +42,7 @@ suite() -> [{ct_hooks,[ts_install_cth]}]. all() -> [{group, t_accept}, {group, t_connect}, {group, t_recv}, t_shutdown_write, t_shutdown_both, t_shutdown_error, - t_fdopen, t_fdconnect, t_implicit_inet6]. + t_shutdown_async, t_fdopen, t_fdconnect, t_implicit_inet6]. groups() -> [{t_accept, [], [t_accept_timeout]}, @@ -155,7 +156,34 @@ t_shutdown_error(Config) when is_list(Config) -> ?line ok = gen_tcp:close(L), ?line {error, closed} = gen_tcp:shutdown(L, read_write), ok. - + +t_shutdown_async(Config) when is_list(Config) -> + ?line {OS, _} = os:type(), + ?line {ok, L} = gen_tcp:listen(0, [{sndbuf, 4096}]), + ?line {ok, Port} = inet:port(L), + ?line {ok, Client} = gen_tcp:connect(localhost, Port, + [{recbuf, 4096}, + {active, false}]), + ?line {ok, S} = gen_tcp:accept(L), + ?line PayloadSize = 1024 * 1024, + ?line Payload = lists:duplicate(PayloadSize, $.), + ?line ok = gen_tcp:send(S, Payload), + ?line case erlang:port_info(S, queue_size) of + {queue_size, N} when N > 0 -> ok; + {queue_size, 0} when OS =:= win32 -> ok; + {queue_size, 0} = T -> ?t:fail({unexpected, T}) + end, + + ?line ok = gen_tcp:shutdown(S, write), + ?line {ok, Buf} = gen_tcp:recv(Client, PayloadSize), + ?line {error, closed} = gen_tcp:recv(Client, 0), + ?line case length(Buf) of + PayloadSize -> ok; + Sz -> ?t:fail({payload_size, + {expected, PayloadSize}, + {received, Sz}}) + end. + %%% gen_tcp:fdopen/2 |