Do flow-specific setsockopt() before accept()

adsbus/flow.c

@@ -1,9 +1,17 @@
 #include <unistd.h>
 
 #include "buf.h"
+#include "socket.h"
 
 #include "flow.h"
 
+void flow_socket_connected(int fd, struct flow *flow) {
+	socket_connected(fd);
+	if (flow->socket_connected) {
+		flow->socket_connected(fd);
+	}
+}
+
 bool flow_hello(int fd, struct flow *flow, void *passthrough) {
 	if (!flow->get_hello) {
 		return true;

adsbus/flow.h

@@ -6,15 +6,13 @@
 struct buf;
 struct peer;
 
-typedef void (*flow_bound_socket_init)(int);
-typedef void (*flow_new)(int, void *, struct peer *);
-typedef void (*flow_get_hello)(struct buf **, void *);
 struct flow {
 	const char *name;
-	flow_bound_socket_init bound_socket_init;
+	void (*socket_connected)(int);
-	flow_new new;
+	void (*new)(int, void *, struct peer *);
-	flow_get_hello get_hello;
+	void (*get_hello)(struct buf **, void *);
 	uint32_t *ref_count;
 };
 
+void flow_socket_connected(int, struct flow *);
 bool flow_hello(int, struct flow *, void *);

adsbus/incoming.c

@@ -95,7 +95,7 @@ static void incoming_listen(struct incoming *incoming) {
 		incoming->peer.fd = socket(addr->ai_family, addr->ai_socktype | SOCK_CLOEXEC, addr->ai_protocol);
 		assert(incoming->peer.fd >= 0);
 
-		socket_pre_bind_init(incoming->peer.fd);
+		socket_pre_bind(incoming->peer.fd);
 
 		if (bind(incoming->peer.fd, addr->ai_addr, addr->ai_addrlen) != 0) {
 			fprintf(stderr, "I %s: Failed to bind to %s/%s: %s\n", incoming->id, hbuf, sbuf, strerror(errno));
@@ -103,7 +103,9 @@ static void incoming_listen(struct incoming *incoming) {
 			continue;
 		}
 
-		socket_bound_init(incoming->peer.fd);
+		socket_pre_listen(incoming->peer.fd);
+		// Options are inherited through accept()
+		flow_socket_connected(incoming->peer.fd, incoming->flow);
 
 		assert(listen(incoming->peer.fd, 255) == 0);
 		break;

adsbus/outgoing.c

@@ -96,7 +96,7 @@ static void outgoing_connect_result(struct outgoing *outgoing, int result) {
 		case 0:
 			fprintf(stderr, "O %s: Connected to %s/%s\n", outgoing->id, hbuf, sbuf);
 			freeaddrinfo(outgoing->addrs);
-			socket_connected_init(outgoing->peer.fd);
+			flow_socket_connected(outgoing->peer.fd, outgoing->flow);
 			outgoing->attempt = 0;
 			int fd = outgoing->peer.fd;
 			outgoing->peer.fd = -1;

adsbus/receive.c

@@ -137,7 +137,7 @@ static void receive_read(struct peer *peer) {
 static void receive_new(int fd, void __attribute__((unused)) *passthrough, struct peer *on_close) {
 	peer_count_in++;
 
-	socket_receive_init(fd);
+	socket_receive(fd);
 
 	struct receive *receive = malloc(sizeof(*receive));
 	assert(receive);

adsbus/send.c

@@ -38,6 +38,7 @@ static void send_get_hello(struct buf **, void *);
 
 static struct flow _send_flow = {
 	.name = "send",
+	.socket_connected = socket_connected_send,
 	.new = send_new,
 	.get_hello = send_get_hello,
 	.ref_count = &peer_count_out,
@@ -104,7 +105,7 @@ static void send_new(int fd, void *passthrough, struct peer *on_close) {
 
 	peer_count_out++;
 
-	socket_send_init(fd);
+	socket_send(fd);
 
 	struct send *send = malloc(sizeof(*send));
 	assert(send);

adsbus/socket.c

@@ -7,23 +7,19 @@
 
 #include "socket.h"
 
-void socket_pre_bind_init(int fd) {
+void socket_pre_bind(int fd) {
 	// Called by transport code; safe to assume that fd is a socket
 	int optval = 1;
 	assert(!setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval)));
 }
 
-void socket_bound_init(int fd) {
+void socket_pre_listen(int fd) {
 	// Called by transport code; safe to assume that fd is a socket
 	int qlen = 5;
 	assert(!setsockopt(fd, SOL_TCP, TCP_FASTOPEN, &qlen, sizeof(qlen)));
-
-	// These options are inherited through accept()
-	// Verified by flamingcow on kernel 4.2.0
-	socket_connected_init(fd);
 }
 
-void socket_connected_init(int fd) {
+void socket_connected(int fd) {
 	// Called by transport code; safe to assume that fd is a socket
 	int optval = 1;
 	assert(!setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval)));
@@ -35,25 +31,30 @@ void socket_connected_init(int fd) {
 	assert(!setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &optval, sizeof(optval)));
 }
 
-void socket_send_init(int fd) {
+void socket_connected_send(int fd) {
-	// Called by data flow code; NOT safe to assume that fd is a socket
-	int res = shutdown(fd, SHUT_RD);
-	assert(res == 0 || (res == -1 && errno == ENOTSOCK));
-
 	int optval = 128; // Lowest value that the kernel will accept
-	res = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &optval, sizeof(optval));
+	int res = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &optval, sizeof(optval));
-	assert(res == 0 || (res == -1 && errno == ENOTSOCK));
+	if (res == -1 && errno == ENOTSOCK) {
+		return;
+	}
+	assert(res == 0);
 
 	optval = 128; // Lowest value that the kernel will accept
 	res = setsockopt(fd, IPPROTO_TCP, TCP_WINDOW_CLAMP, &optval, sizeof(optval));
-	assert(res == 0 || (res == -1 && errno == ENOTSOCK));
+	assert(res == 0);
 
 	optval = 60000; // 60s
 	res = setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &optval, sizeof(optval));
+	assert(res == 0);
+}
+
+void socket_send(int fd) {
+	// Called by data flow code; NOT safe to assume that fd is a socket
+	int res = shutdown(fd, SHUT_RD);
 	assert(res == 0 || (res == -1 && errno == ENOTSOCK));
 }
 
-void socket_receive_init(int fd) {
+void socket_receive(int fd) {
 	// Called by data flow code; NOT safe to assume that fd is a socket
 	int res = shutdown(fd, SHUT_WR);
 	assert(res == 0 || (res == -1 && errno == ENOTSOCK));

adsbus/socket.h

@@ -1,7 +1,8 @@
 #pragma once
 
-void socket_pre_bind_init(int);
+void socket_pre_bind(int);
-void socket_bound_init(int);
+void socket_pre_listen(int);
-void socket_connected_init(int);
+void socket_connected(int);
-void socket_send_init(int);
+void socket_connected_send(int);
-void socket_receive_init(int);
+void socket_send(int);
+void socket_receive(int);