X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=src%2Futcp.c;h=9e2229ad522fb48b2d5431b6658e82671422680e;hb=refs%2Fheads%2Ffeature%2Fhuge-udp-channel-packets;hp=7ac96c626d6dedbf9849d4fe41eb11e7a24763ac;hpb=f84fc9da827afbc7bc7bd14fa94b84ee9922e575;p=meshlink diff --git a/src/utcp.c b/src/utcp.c index 7ac96c62..9e2229ad 100644 --- a/src/utcp.c +++ b/src/utcp.c @@ -396,7 +396,10 @@ static bool buffer_set_size(struct buffer *buf, uint32_t minsize, uint32_t maxsi } static void buffer_exit(struct buffer *buf) { - free(buf->data); + if(!buf->external) { + free(buf->data); + } + memset(buf, 0, sizeof(*buf)); } @@ -1293,7 +1296,7 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { if(hdr.ctl & SYN && !(hdr.ctl & ACK) && utcp->accept) { // If we don't want to accept it, send a RST back - if((utcp->pre_accept && !utcp->pre_accept(utcp, hdr.dst))) { + if((utcp->listen && !utcp->listen(utcp, hdr.dst))) { len = 1; goto reset; } @@ -1368,7 +1371,7 @@ synack: if(c->state == CLOSED) { debug(c, "got packet for closed connection\n"); - return 0; + goto reset; } // It is for an existing connection. @@ -1456,17 +1459,6 @@ synack: } } - if(hdr.ctl & ACK && (seqdiff(hdr.ack, c->snd.last) > 0 || seqdiff(hdr.ack, c->snd.una) < 0)) { - debug(c, "packet ack seqno out of range, %u <= %u < %u\n", c->snd.una, hdr.ack, c->snd.una + c->sndbuf.used); - - // Ignore unacceptable RST packets. - if(hdr.ctl & RST) { - return 0; - } - - goto reset; - } - // 2. Handle RST packets if(hdr.ctl & RST) { @@ -1556,6 +1548,11 @@ synack: // 3. Advance snd.una + if(seqdiff(hdr.ack, c->snd.last) > 0 || seqdiff(hdr.ack, c->snd.una) < 0) { + debug(c, "packet ack seqno out of range, %u <= %u < %u\n", c->snd.una, hdr.ack, c->snd.una + c->sndbuf.used); + goto reset; + } + advanced = seqdiff(hdr.ack, c->snd.una); if(advanced) { @@ -1778,8 +1775,15 @@ skip_ack: return 0; case ESTABLISHED: + break; + case FIN_WAIT_1: case FIN_WAIT_2: + if(c->reapable) { + // We already closed the connection and are not interested in more data. + goto reset; + } + break; case CLOSE_WAIT: @@ -2001,7 +2005,7 @@ static bool reset_connection(struct utcp_connection *c) { hdr.src = c->src; hdr.dst = c->dst; hdr.seq = c->snd.nxt; - hdr.ack = 0; + hdr.ack = c->rcv.nxt; hdr.wnd = 0; hdr.ctl = RST; @@ -2044,10 +2048,6 @@ void utcp_abort_all_connections(struct utcp *utcp) { } int utcp_close(struct utcp_connection *c) { - if(c->rcvbuf.used) { - return reset_connection(c) ? 0 : -1; - } - if(utcp_shutdown(c, SHUT_RDWR) && errno != ENOTCONN) { return -1; } @@ -2158,7 +2158,7 @@ bool utcp_is_active(struct utcp *utcp) { return false; } -struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_send_t send, void *priv) { +struct utcp *utcp_init(utcp_accept_t accept, utcp_listen_t listen, utcp_send_t send, void *priv) { if(!send) { errno = EFAULT; return NULL; @@ -2184,7 +2184,7 @@ struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_ } utcp->accept = accept; - utcp->pre_accept = pre_accept; + utcp->listen = listen; utcp->send = send; utcp->priv = priv; utcp->timeout = DEFAULT_USER_TIMEOUT; // sec @@ -2318,17 +2318,73 @@ size_t utcp_get_sndbuf_free(struct utcp_connection *c) { } } -void utcp_set_sndbuf(struct utcp_connection *c, size_t size) { - if(!c) { - return; +static void buffer_transfer(struct buffer *buf, char *newdata, size_t newsize) { + if(buffer_wraps(buf)) { + // Old situation: + // [345......012] + // New situation: + // [012345......] + uint32_t tailsize = buf->size - buf->offset; + memcpy(newdata, buf->data + buf->offset, tailsize); + memcpy(newdata + tailsize, buf->data, buf->used - buf->offset); + } else { + // Old situation: + // [....012345..] + // New situation: + // [012345......] + memcpy(newdata, buf->data + buf->offset, buf->used); } - c->sndbuf.maxsize = size; + buf->offset = 0; + buf->size = newsize; +} - if(c->sndbuf.maxsize != size) { - c->sndbuf.maxsize = -1; +static void set_buffer_storage(struct buffer *buf, char *data, size_t size) { + if(size > UINT32_MAX) { + size = UINT32_MAX; } + buf->maxsize = size; + + if(data) { + if(buf->external) { + // Don't allow resizing an external buffer + abort(); + } + + if(size < buf->used) { + // Ignore requests for an external buffer if we are already using more than it can store + return; + } + + // Transition from internal to external buffer + buffer_transfer(buf, data, size); + free(buf->data); + buf->data = data; + buf->external = true; + } else if(buf->external) { + // Transition from external to internal buf + size_t minsize = buf->used < DEFAULT_SNDBUFSIZE ? DEFAULT_SNDBUFSIZE : buf->used; + data = malloc(minsize); + + if(!data) { + // Cannot handle this + abort(); + } + + buffer_transfer(buf, data, minsize); + buf->data = data; + buf->external = false; + } +} + +void utcp_set_sndbuf(struct utcp_connection *c, void *data, size_t size) { + if(!c) { + return; + } + + set_buffer_storage(&c->sndbuf, data, size); + c->do_poll = is_reliable(c) && buffer_free(&c->sndbuf); } @@ -2344,16 +2400,12 @@ size_t utcp_get_rcvbuf_free(struct utcp_connection *c) { } } -void utcp_set_rcvbuf(struct utcp_connection *c, size_t size) { +void utcp_set_rcvbuf(struct utcp_connection *c, void *data, size_t size) { if(!c) { return; } - c->rcvbuf.maxsize = size; - - if(c->rcvbuf.maxsize != size) { - c->rcvbuf.maxsize = -1; - } + set_buffer_storage(&c->rcvbuf, data, size); } size_t utcp_get_sendq(struct utcp_connection *c) { @@ -2401,10 +2453,10 @@ void utcp_set_poll_cb(struct utcp_connection *c, utcp_poll_t poll) { } } -void utcp_set_accept_cb(struct utcp *utcp, utcp_accept_t accept, utcp_pre_accept_t pre_accept) { +void utcp_set_accept_cb(struct utcp *utcp, utcp_accept_t accept, utcp_listen_t listen) { if(utcp) { utcp->accept = accept; - utcp->pre_accept = pre_accept; + utcp->listen = listen; } } @@ -2431,6 +2483,11 @@ void utcp_expect_data(struct utcp_connection *c, bool expect) { } } +void utcp_set_flags(struct utcp_connection *c, uint32_t flags) { + c->flags &= ~UTCP_CHANGEABLE_FLAGS; + c->flags |= flags & UTCP_CHANGEABLE_FLAGS; +} + void utcp_offline(struct utcp *utcp, bool offline) { struct timespec now; clock_gettime(UTCP_CLOCK, &now);