gen_tcp passive sockets discard unread bytes on remote write shutdown

[liamwhite]

Describe the bug
When the remote side of a TCP socket shuts the socket down for writing, all unread bytes on a passive socket which have been received by gen_tcp are immediately discarded and no longer accessible.

To Reproduce
The following escript demonstrates the issue:

listen() ->
  {ok, Listener} = gen_tcp:listen(5678, [binary]),
  {ok, Socket} = gen_tcp:accept(Listener),
  ok = gen_tcp:send(Socket, <<"hello world">>),
  ok = gen_tcp:shutdown(Socket, write),
  timer:sleep(timer:seconds(2)).

connect() ->
  {ok, Socket} = gen_tcp:connect("localhost", 5678, [binary]),
  ok = timer:sleep(timer:seconds(1)),
  {ok, <<"hello world">>} = gen_tcp:recv(Socket, 1460),
  ok.

main(_) ->
  _ = spawn(fun() -> listen() end),
  ok = timer:sleep(100),
  _ = spawn(fun() -> connect() end),
  timer:sleep(2000).

This produces the error:

{{badmatch,{error,closed}},
 [{erl_eval,expr,6,[{file,"erl_eval.erl"},{line,498}]},
  {escript,eval_exprs,5,[{file,"escript.erl"},{line,869}]}]}

Expected behavior
gen_tcp should return the remaining bytes before returning {error,closed}

Affected versions
25, 26, 27 known, likely all

Additional context

C++ program which shows the expected behavior

#include <algorithm>
#include <thread>

#include <sys/socket.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <netinet/ip.h>

static struct sockaddr_in g_addr;
static const char g_buf[] = "hello world";
static const int reuseaddr = 1;

ssize_t check(ssize_t i, const char* x) {
	if (i < 0) {
		fprintf(stderr, "%s = %d [errno = %d, %s]\n", x, i, errno, strerror(errno));
		abort();
	}
	return i;
}

#define X(x) check(x, #x)

void listen() {
	int fd = X(socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
	X(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr, sizeof(reuseaddr)));

	g_addr.sin_family = AF_INET;
	g_addr.sin_port = htons(5678);
	inet_aton("127.0.0.1", &g_addr.sin_addr);

	X(bind(fd, (struct sockaddr *)&g_addr, sizeof(g_addr)));
	X(listen(fd, 0));

	struct sockaddr_in addr;
	socklen_t addrlen;
	int client_fd = X(accept(fd, (struct sockaddr *)&addr, &addrlen));

	send(client_fd, g_buf, sizeof(g_buf), 0);
	shutdown(client_fd, SHUT_WR);

	sleep(2);
	shutdown(client_fd, SHUT_RDWR);
	close(client_fd);
	close(fd);
}

void connect() {
	int fd = X(socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));

	X(connect(fd, (struct sockaddr *)&g_addr, sizeof(g_addr)));
	sleep(1);

	char local_buf[128] = {};
	size_t recv_len = X(recv(fd, local_buf, 128, 0));

	int cmp = memcmp(local_buf, g_buf, std::min(recv_len, sizeof(g_buf)));
	printf("cmp = %d\n", cmp);
	if (cmp == 0) {
		printf("%s\n", local_buf);
	}

	close(fd);
}

int main() {
	std::thread listen_thread([]() { listen(); });
	usleep(100);
	std::thread connect_thread([]() { connect(); });
	listen_thread.join();
	connect_thread.join();
	return 0;
}

[RaimoNiskanen]

Hello @liamwhite!

Firstly, a gen_tcp socket is per default {active,true}, so if you change the connect/0 function to handle that it works:

connect() ->
     {ok, Socket} = gen_tcp:connect("localhost", 5678, [binary]),
     ok = timer:sleep(timer:seconds(1)),
     {tcp, Socket, <<"hello world">>} = receive Msg -> Msg end.

Secondly, gen_tcp, probably in the name of usability, has a subtly different meaning of the Length argument compared to C socket recv(). It means a length that must be fulfilled; if less is received it is considered a failure and if the socket gets closed before Length bytes has been received, then none are returned. The function signature doesn't permit anything like {error, closed, PartialData}.

But if you use Length = 11 (length of "hello world"), or Length = 0 (return whatever available), in combination with {active,false} it also works:

connect() ->
   {ok, Socket} = gen_tcp:connect("localhost", 5678, [binary,{active,false}]),
   ok = timer:sleep(timer:seconds(1)),
   {ok, <<"hello world">>} = gen_tcp:recv(Socket, 0).

[liamwhite]

Length = 0 looks like the behavior I need. Thanks for the clarification.