Skip to content

Commit

Permalink
improved tcp closure
Browse files Browse the repository at this point in the history
  • Loading branch information
robert committed Aug 17, 2023
1 parent 8c03aa2 commit 4e9073c
Show file tree
Hide file tree
Showing 6 changed files with 104 additions and 29 deletions.
56 changes: 44 additions & 12 deletions mongoose.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8122,6 +8122,7 @@ struct mg_tcpip_driver mg_tcpip_driver_w5500 = {w5500_init, w5500_tx, w5500_rx,
#define MIP_TCP_ACK_MS 150 // Timeout for ACKing
#define MIP_TCP_ARP_MS 100 // Timeout for ARP response
#define MIP_TCP_SYN_MS 15000 // Timeout for connection establishment
#define MIP_TCP_FIN_MS 1000 // Timeout for closing connection

struct connstate {
uint32_t seq, ack; // TCP seq/ack counters
Expand All @@ -8132,8 +8133,9 @@ struct connstate {
#define MIP_TTYPE_ACK 1 // Peer sent us data, we have to ack it soon
#define MIP_TTYPE_ARP 2 // ARP resolve sent, waiting for response
#define MIP_TTYPE_SYN 3 // SYN sent, waiting for response
uint8_t tmiss; // Number of keep-alive misses
struct mg_iobuf raw; // For TLS only. Incoming raw data
#define MIP_TTYPE_FIN 4 // FIN sent, waiting until terminating the connection
uint8_t tmiss; // Number of keep-alive misses
struct mg_iobuf raw; // For TLS only. Incoming raw data
};

#pragma pack(push, 1)
Expand Down Expand Up @@ -8271,6 +8273,7 @@ static void settmout(struct mg_connection *c, uint8_t type) {
unsigned n = type == MIP_TTYPE_ACK ? MIP_TCP_ACK_MS
: type == MIP_TTYPE_ARP ? MIP_TCP_ARP_MS
: type == MIP_TTYPE_SYN ? MIP_TCP_SYN_MS
: type == MIP_TTYPE_FIN ? MIP_TCP_FIN_MS
: MIP_TCP_KEEPALIVE_MS;
s->timer = ifp->now + n;
s->ttype = type;
Expand Down Expand Up @@ -8694,18 +8697,34 @@ static void read_conn(struct mg_connection *c, struct pkt *pkt) {
struct mg_iobuf *io = c->is_tls ? &s->raw : &c->recv;
uint32_t seq = mg_ntohl(pkt->tcp->seq);
s->raw.align = c->recv.align;
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
if (pkt->tcp->flags & TH_FIN) {
s->ack = mg_htonl(pkt->tcp->seq) + 1, s->seq = mg_htonl(pkt->tcp->ack);
c->is_closing = 1;
// If we initiated the closure, we reply with ACK upon receiving FIN
// If we didn't initiate it, we reply with FIN as part of the normal TCP
// closure process
uint8_t flags = TH_ACK;
s->ack = mg_htonl(pkt->tcp->seq) + 1;
if (c->is_close_wait) {
if (s->seq == mg_htonl(pkt->tcp->ack))
// Checking for simultaneous closure
s->seq++;
else
s->seq = mg_htonl(pkt->tcp->ack);
} else {
flags |= TH_FIN;
c->is_closing = c->is_close_wait = 1;
settmout(c, MIP_TTYPE_FIN);
}
tx_tcp((struct mg_tcpip_if *) c->mgr->priv, s->mac, rem_ip, flags,
c->loc.port, c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), "", 0);
} else if (pkt->pay.len == 0) {
// TODO(cpq): handle this peer's ACK
} else if (seq != s->ack) {
uint32_t ack = (uint32_t) (mg_htonl(pkt->tcp->seq) + pkt->pay.len);
if (s->ack == ack) {
MG_VERBOSE(("ignoring duplicate pkt"));
} else {
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
MG_VERBOSE(("SEQ != ACK: %x %x %x", seq, s->ack, ack));
tx_tcp((struct mg_tcpip_if *) c->mgr->priv, s->mac, rem_ip, TH_ACK,
c->loc.port, c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), "",
Expand Down Expand Up @@ -8764,6 +8783,7 @@ static void read_conn(struct mg_connection *c, struct pkt *pkt) {
}

static void rx_tcp(struct mg_tcpip_if *ifp, struct pkt *pkt) {
//mg_hexdump(pkt->raw.ptr, pkt->raw.len);
struct mg_connection *c = getpeer(ifp->mgr, pkt, false);
struct connstate *s = c == NULL ? NULL : (struct connstate *) (c + 1);
#if 0
Expand Down Expand Up @@ -8872,7 +8892,7 @@ static void mg_tcpip_rx(struct mg_tcpip_if *ifp, void *buf, size_t len) {
pkt.raw.ptr = (char *) buf;
pkt.raw.len = len;
pkt.eth = (struct eth *) buf;
//mg_hexdump(buf, len > 16 ? 16: len);
// mg_hexdump(buf, len > 16 ? 16: len);
if (pkt.raw.len < sizeof(*pkt.eth)) return; // Truncated - runt?
if (ifp->enable_mac_check &&
memcmp(pkt.eth->dst, ifp->mac, sizeof(pkt.eth->dst)) != 0 &&
Expand Down Expand Up @@ -8973,6 +8993,9 @@ static void mg_tcpip_poll(struct mg_tcpip_if *ifp, uint64_t uptime_ms) {
mg_error(c, "ARP timeout");
} else if (s->ttype == MIP_TTYPE_SYN) {
mg_error(c, "Connection timeout");
} else if (s->ttype == MIP_TTYPE_FIN) {
c->is_close_wait = 0;
continue;
} else {
if (s->tmiss++ > 2) {
mg_error(c, "keepalive");
Expand All @@ -8982,6 +9005,7 @@ static void mg_tcpip_poll(struct mg_tcpip_if *ifp, uint64_t uptime_ms) {
mg_htonl(s->seq - 1), mg_htonl(s->ack), "", 0);
}
}

settmout(c, MIP_TTYPE_KEEPALIVE);
}
}
Expand Down Expand Up @@ -9106,18 +9130,24 @@ static void write_conn(struct mg_connection *c) {
}
}

static void close_conn(struct mg_connection *c) {
static void init_closure(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
mg_iobuf_free(&s->raw); // For TLS connections, release raw data
if (c->is_udp == false && c->is_listening == false &&
c->is_connecting == false) { // For TCP conns,
struct mg_tcpip_if *ifp =
(struct mg_tcpip_if *) c->mgr->priv; // send TCP FIN
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
tx_tcp(ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port, c->rem.port,
mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
c->is_close_wait = 1;
settmout(c, MIP_TTYPE_FIN);
}
}

static void close_conn(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
mg_iobuf_free(&s->raw); // For TLS connections, release raw data
mg_close_conn(c);
}

Expand All @@ -9133,14 +9163,16 @@ void mg_mgr_poll(struct mg_mgr *mgr, int ms) {
mg_timer_poll(&mgr->timers, now);
for (c = mgr->conns; c != NULL; c = tmp) {
tmp = c->next;
struct connstate *s = (struct connstate *) (c + 1);
mg_call(c, MG_EV_POLL, &now);
MG_VERBOSE(("%lu .. %c%c%c%c%c", c->id, c->is_tls ? 'T' : 't',
c->is_connecting ? 'C' : 'c', c->is_tls_hs ? 'H' : 'h',
c->is_resolving ? 'R' : 'r', c->is_closing ? 'C' : 'c'));
if (c->is_tls_hs) mg_tls_handshake(c);
if (can_write(c)) write_conn(c);
if (c->is_draining && c->send.len == 0) c->is_closing = 1;
if (c->is_closing) close_conn(c);
if (c->is_closing && s->ttype != MIP_TTYPE_FIN) init_closure(c);
if (c->is_closing && c->is_close_wait == 0) close_conn(c);
}
(void) ms;
}
Expand Down
1 change: 1 addition & 0 deletions mongoose.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1208,6 +1208,7 @@ struct mg_connection {
unsigned is_mqtt5 : 1; // For MQTT connection, v5 indicator
unsigned is_hexdumping : 1; // Hexdump in/out traffic
unsigned is_draining : 1; // Send remaining data, then close and free
unsigned is_close_wait : 1; // Set when initializing TCP closure
unsigned is_closing : 1; // Close and free the connection immediately
unsigned is_full : 1; // Stop reads, until cleared
unsigned is_resp : 1; // Response is still being generated
Expand Down
1 change: 1 addition & 0 deletions src/net.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ struct mg_connection {
unsigned is_mqtt5 : 1; // For MQTT connection, v5 indicator
unsigned is_hexdumping : 1; // Hexdump in/out traffic
unsigned is_draining : 1; // Send remaining data, then close and free
unsigned is_close_wait : 1; // Set when initializing TCP closure
unsigned is_closing : 1; // Close and free the connection immediately
unsigned is_full : 1; // Stop reads, until cleared
unsigned is_resp : 1; // Response is still being generated
Expand Down
56 changes: 44 additions & 12 deletions src/tcpip/tcpip.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
#define MIP_TCP_ACK_MS 150 // Timeout for ACKing
#define MIP_TCP_ARP_MS 100 // Timeout for ARP response
#define MIP_TCP_SYN_MS 15000 // Timeout for connection establishment
#define MIP_TCP_FIN_MS 1000 // Timeout for closing connection

struct connstate {
uint32_t seq, ack; // TCP seq/ack counters
Expand All @@ -22,8 +23,9 @@ struct connstate {
#define MIP_TTYPE_ACK 1 // Peer sent us data, we have to ack it soon
#define MIP_TTYPE_ARP 2 // ARP resolve sent, waiting for response
#define MIP_TTYPE_SYN 3 // SYN sent, waiting for response
uint8_t tmiss; // Number of keep-alive misses
struct mg_iobuf raw; // For TLS only. Incoming raw data
#define MIP_TTYPE_FIN 4 // FIN sent, waiting until terminating the connection
uint8_t tmiss; // Number of keep-alive misses
struct mg_iobuf raw; // For TLS only. Incoming raw data
};

#pragma pack(push, 1)
Expand Down Expand Up @@ -161,6 +163,7 @@ static void settmout(struct mg_connection *c, uint8_t type) {
unsigned n = type == MIP_TTYPE_ACK ? MIP_TCP_ACK_MS
: type == MIP_TTYPE_ARP ? MIP_TCP_ARP_MS
: type == MIP_TTYPE_SYN ? MIP_TCP_SYN_MS
: type == MIP_TTYPE_FIN ? MIP_TCP_FIN_MS
: MIP_TCP_KEEPALIVE_MS;
s->timer = ifp->now + n;
s->ttype = type;
Expand Down Expand Up @@ -584,18 +587,34 @@ static void read_conn(struct mg_connection *c, struct pkt *pkt) {
struct mg_iobuf *io = c->is_tls ? &s->raw : &c->recv;
uint32_t seq = mg_ntohl(pkt->tcp->seq);
s->raw.align = c->recv.align;
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
if (pkt->tcp->flags & TH_FIN) {
s->ack = mg_htonl(pkt->tcp->seq) + 1, s->seq = mg_htonl(pkt->tcp->ack);
c->is_closing = 1;
// If we initiated the closure, we reply with ACK upon receiving FIN
// If we didn't initiate it, we reply with FIN as part of the normal TCP
// closure process
uint8_t flags = TH_ACK;
s->ack = mg_htonl(pkt->tcp->seq) + 1;
if (c->is_close_wait) {
if (s->seq == mg_htonl(pkt->tcp->ack))
// Checking for simultaneous closure
s->seq++;
else
s->seq = mg_htonl(pkt->tcp->ack);
} else {
flags |= TH_FIN;
c->is_closing = c->is_close_wait = 1;
settmout(c, MIP_TTYPE_FIN);
}
tx_tcp((struct mg_tcpip_if *) c->mgr->priv, s->mac, rem_ip, flags,
c->loc.port, c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), "", 0);
} else if (pkt->pay.len == 0) {
// TODO(cpq): handle this peer's ACK
} else if (seq != s->ack) {
uint32_t ack = (uint32_t) (mg_htonl(pkt->tcp->seq) + pkt->pay.len);
if (s->ack == ack) {
MG_VERBOSE(("ignoring duplicate pkt"));
} else {
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
MG_VERBOSE(("SEQ != ACK: %x %x %x", seq, s->ack, ack));
tx_tcp((struct mg_tcpip_if *) c->mgr->priv, s->mac, rem_ip, TH_ACK,
c->loc.port, c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), "",
Expand Down Expand Up @@ -654,6 +673,7 @@ static void read_conn(struct mg_connection *c, struct pkt *pkt) {
}

static void rx_tcp(struct mg_tcpip_if *ifp, struct pkt *pkt) {
//mg_hexdump(pkt->raw.ptr, pkt->raw.len);
struct mg_connection *c = getpeer(ifp->mgr, pkt, false);
struct connstate *s = c == NULL ? NULL : (struct connstate *) (c + 1);
#if 0
Expand Down Expand Up @@ -762,7 +782,7 @@ static void mg_tcpip_rx(struct mg_tcpip_if *ifp, void *buf, size_t len) {
pkt.raw.ptr = (char *) buf;
pkt.raw.len = len;
pkt.eth = (struct eth *) buf;
//mg_hexdump(buf, len > 16 ? 16: len);
// mg_hexdump(buf, len > 16 ? 16: len);
if (pkt.raw.len < sizeof(*pkt.eth)) return; // Truncated - runt?
if (ifp->enable_mac_check &&
memcmp(pkt.eth->dst, ifp->mac, sizeof(pkt.eth->dst)) != 0 &&
Expand Down Expand Up @@ -863,6 +883,9 @@ static void mg_tcpip_poll(struct mg_tcpip_if *ifp, uint64_t uptime_ms) {
mg_error(c, "ARP timeout");
} else if (s->ttype == MIP_TTYPE_SYN) {
mg_error(c, "Connection timeout");
} else if (s->ttype == MIP_TTYPE_FIN) {
c->is_close_wait = 0;
continue;
} else {
if (s->tmiss++ > 2) {
mg_error(c, "keepalive");
Expand All @@ -872,6 +895,7 @@ static void mg_tcpip_poll(struct mg_tcpip_if *ifp, uint64_t uptime_ms) {
mg_htonl(s->seq - 1), mg_htonl(s->ack), "", 0);
}
}

settmout(c, MIP_TTYPE_KEEPALIVE);
}
}
Expand Down Expand Up @@ -996,18 +1020,24 @@ static void write_conn(struct mg_connection *c) {
}
}

static void close_conn(struct mg_connection *c) {
static void init_closure(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
mg_iobuf_free(&s->raw); // For TLS connections, release raw data
if (c->is_udp == false && c->is_listening == false &&
c->is_connecting == false) { // For TCP conns,
struct mg_tcpip_if *ifp =
(struct mg_tcpip_if *) c->mgr->priv; // send TCP FIN
uint32_t rem_ip;
memcpy(&rem_ip, c->rem.ip, sizeof(uint32_t));
tx_tcp(ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port, c->rem.port,
mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
c->is_close_wait = 1;
settmout(c, MIP_TTYPE_FIN);
}
}

static void close_conn(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
mg_iobuf_free(&s->raw); // For TLS connections, release raw data
mg_close_conn(c);
}

Expand All @@ -1023,14 +1053,16 @@ void mg_mgr_poll(struct mg_mgr *mgr, int ms) {
mg_timer_poll(&mgr->timers, now);
for (c = mgr->conns; c != NULL; c = tmp) {
tmp = c->next;
struct connstate *s = (struct connstate *) (c + 1);
mg_call(c, MG_EV_POLL, &now);
MG_VERBOSE(("%lu .. %c%c%c%c%c", c->id, c->is_tls ? 'T' : 't',
c->is_connecting ? 'C' : 'c', c->is_tls_hs ? 'H' : 'h',
c->is_resolving ? 'R' : 'r', c->is_closing ? 'C' : 'c'));
if (c->is_tls_hs) mg_tls_handshake(c);
if (can_write(c)) write_conn(c);
if (c->is_draining && c->send.len == 0) c->is_closing = 1;
if (c->is_closing) close_conn(c);
if (c->is_closing && s->ttype != MIP_TTYPE_FIN) init_closure(c);
if (c->is_closing && c->is_close_wait == 0) close_conn(c);
}
(void) ms;
}
Expand Down
14 changes: 9 additions & 5 deletions test/mip_tap_test.c
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
#define MG_ENABLE_TCPIP 1
#define MG_ENABLE_SOCKET 0
#define MG_USING_DHCP 1
#define MG_USING_DHCP 0
#define MG_ENABLE_PACKED_FS 0
#define MG_ENABLE_LINES 1

Expand Down Expand Up @@ -101,10 +101,13 @@ char *fetch(struct mg_mgr *mgr, const char *url, const char *fn_data) {
mg_mgr_poll(mgr, 100);
usleep(10000); // 10 ms. Slow down poll loop to ensure packets transit
}

if (mgr->conns != 0) {
conn->is_closing = 1;
mg_mgr_poll(mgr, 0);
usleep(100000); // 100 ms. Wait to receive FIN-ACK
}

mg_mgr_poll(mgr, 0);
if (!post_reply.http_responses_received)
return 0;
Expand Down Expand Up @@ -175,9 +178,9 @@ static void test_http_fetch(void) {

#if MG_USING_DHCP == 1
#else
mif.ip = 0x0220a8c0; // 192.168.32.2 // Triggering a network failure
mif.mask = 0x00ffffff; // 255.255.255.0
mif.gw = 0x0120a8c0; // 192.168.32.1
mif.ip = mg_htonl(MG_U32(192, 168, 32, 2)); // Triggering a network failure
mif.mask = mg_htonl(MG_U32(255, 255, 255, 0));
mif.gw = mg_htonl(MG_U32(192, 168, 32, 1));
#endif

sscanf(mac, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx", &mif.mac[0], &mif.mac[1],
Expand Down Expand Up @@ -234,8 +237,9 @@ static void test_http_fetch(void) {
}

// Clear
usleep(1000000); // Waiting until all connections finally close
mg_mgr_free(&mgr);
mg_tcpip_free(&mif); // Release after mg_mgr
mg_tcpip_free(&mif); // Release after mg_mgr
ASSERT(mgr.conns == NULL); // Deconstruction OK
close(fd);
}
Expand Down
5 changes: 5 additions & 0 deletions test/mip_test.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,11 @@ static void test_retransmit(void) {
ASSERT((t->ack == mg_htonl(1003))); // OK

s_driver_data.len = 0;
// closing the connections immediately
for (c = mgr.conns; c != NULL; c = c->next) {
c->is_closing = 1;
settmout(c, MIP_TTYPE_FIN);
}
mg_mgr_free(&mgr);
mg_tcpip_free(&mif);
}
Expand Down

0 comments on commit 4e9073c

Please sign in to comment.