X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=src%2Futcp.c;h=20dd0aba049757ca30833779ec772c63ba685310;hb=HEAD;hp=5ba5553bd3e2e51ac824887b2f82039a16e4b02d;hpb=8d4b96efb7955eaa96174af4804597f92e124041;p=meshlink diff --git a/src/utcp.c b/src/utcp.c index 5ba5553b..ca91bee2 100644 --- a/src/utcp.c +++ b/src/utcp.c @@ -17,16 +17,7 @@ 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE - -#include -#include -#include -#include -#include -#include -#include -#include +#include "system.h" #include #include "utcp_priv.h" @@ -205,6 +196,16 @@ static bool buffer_wraps(struct buffer *buf) { } static bool buffer_resize(struct buffer *buf, uint32_t newsize) { + assert(!buf->external); + + if(!newsize) { + free(buf->data); + buf->data = NULL; + buf->size = 0; + buf->offset = 0; + return true; + } + char *newdata = realloc(buf->data, newsize); if(!newdata) { @@ -268,7 +269,7 @@ static ssize_t buffer_put_at(struct buffer *buf, size_t offset, const void *data uint32_t realoffset = buf->offset + offset; - if(buf->size - buf->offset < offset) { + if(buf->size - buf->offset <= offset) { // The offset wrapped realoffset -= buf->size; } @@ -305,7 +306,7 @@ static ssize_t buffer_copy(struct buffer *buf, void *data, size_t offset, size_t uint32_t realoffset = buf->offset + offset; - if(buf->size - buf->offset < offset) { + if(buf->size - buf->offset <= offset) { // The offset wrapped realoffset -= buf->size; } @@ -322,7 +323,11 @@ static ssize_t buffer_copy(struct buffer *buf, void *data, size_t offset, size_t } // Copy data from the buffer without removing it. -static ssize_t buffer_call(struct buffer *buf, utcp_recv_t cb, void *arg, size_t offset, size_t len) { +static ssize_t buffer_call(struct utcp_connection *c, struct buffer *buf, size_t offset, size_t len) { + if(!c->recv) { + return len; + } + // Ensure we don't copy more than is actually stored in the buffer if(offset >= buf->used) { return 0; @@ -334,20 +339,25 @@ static ssize_t buffer_call(struct buffer *buf, utcp_recv_t cb, void *arg, size_t uint32_t realoffset = buf->offset + offset; - if(buf->size - buf->offset < offset) { + if(buf->size - buf->offset <= offset) { // The offset wrapped realoffset -= buf->size; } if(buf->size - realoffset < len) { // The data is wrapped - ssize_t rx1 = cb(arg, buf->data + realoffset, buf->size - realoffset); + ssize_t rx1 = c->recv(c, buf->data + realoffset, buf->size - realoffset); if(rx1 < buf->size - realoffset) { return rx1; } - ssize_t rx2 = cb(arg, buf->data, len - (buf->size - realoffset)); + // The channel might have been closed by the previous callback + if(!c->recv) { + return len; + } + + ssize_t rx2 = c->recv(c, buf->data, len - (buf->size - realoffset)); if(rx2 < 0) { return rx2; @@ -355,7 +365,7 @@ static ssize_t buffer_call(struct buffer *buf, utcp_recv_t cb, void *arg, size_t return rx1 + rx2; } } else { - return cb(arg, buf->data + realoffset, len); + return c->recv(c, buf->data + realoffset, len); } } @@ -365,7 +375,7 @@ static ssize_t buffer_discard(struct buffer *buf, size_t len) { len = buf->used; } - if(buf->size - buf->offset < len) { + if(buf->size - buf->offset <= len) { buf->offset -= buf->size; } @@ -395,8 +405,99 @@ static bool buffer_set_size(struct buffer *buf, uint32_t minsize, uint32_t maxsi return buf->size >= minsize || buffer_resize(buf, minsize); } +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 - tailsize); + } else { + // Old situation: + // [....012345..] + // New situation: + // [012345......] + memcpy(newdata, buf->data + buf->offset, buf->used); + } + + buf->offset = 0; + buf->size = newsize; +} + +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; + + if(minsize) { + data = malloc(minsize); + + if(!data) { + // Cannot handle this + abort(); + } + + buffer_transfer(buf, data, minsize); + buf->data = data; + } else { + buf->data = NULL; + buf->size = 0; + } + + buf->external = false; + } else { + // Don't do anything if the buffer wraps + if(buffer_wraps(buf)) { + return; + } + + // Realloc internal storage + size_t minsize = max(DEFAULT_SNDBUFSIZE, buf->offset + buf->used); + + if(minsize) { + data = realloc(buf->data, minsize); + + if(data) { + buf->data = data; + buf->size = minsize; + } + } else { + free(buf->data); + buf->data = NULL; + buf->size = 0; + } + } +} + static void buffer_exit(struct buffer *buf) { - free(buf->data); + if(!buf->external) { + free(buf->data); + } + memset(buf, 0, sizeof(*buf)); } @@ -414,7 +515,6 @@ static int compare(const void *va, const void *vb) { const struct utcp_connection *b = *(struct utcp_connection **)vb; assert(a && b); - assert(a->src && b->src); int c = (int)a->src - (int)b->src; @@ -651,6 +751,7 @@ void utcp_accept(struct utcp_connection *c, utcp_recv_t recv, void *priv) { debug(c, "accepted %p %p\n", c, recv, priv); c->recv = recv; c->priv = priv; + c->do_poll = true; set_state(c, ESTABLISHED); } @@ -1110,13 +1211,11 @@ static void handle_in_order(struct utcp_connection *c, const void *data, size_t len = c->sacks[0].offset + c->sacks[0].len; size_t remainder = len - offset; - if(c->recv) { - ssize_t rxd = buffer_call(&c->rcvbuf, c->recv, c, offset, remainder); + ssize_t rxd = buffer_call(c, &c->rcvbuf, offset, remainder); - if(rxd != (ssize_t)remainder) { - // TODO: handle the application not accepting all data. - abort(); - } + if(rxd != (ssize_t)remainder) { + // TODO: handle the application not accepting all data. + abort(); } } } @@ -1161,8 +1260,8 @@ static void handle_unreliable(struct utcp_connection *c, const struct hdr *hdr, } // Send the packet if it's the final fragment - if(!(hdr->ctl & MF) && c->recv) { - buffer_call(&c->rcvbuf, c->recv, c, 0, hdr->wnd + len); + if(!(hdr->ctl & MF)) { + buffer_call(c, &c->rcvbuf, 0, hdr->wnd + len); } c->rcv.nxt = hdr->seq + len; @@ -1295,7 +1394,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; } @@ -1370,7 +1469,7 @@ synack: if(c->state == CLOSED) { debug(c, "got packet for closed connection\n"); - return 0; + goto reset; } // It is for an existing connection. @@ -1458,17 +1557,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) { @@ -1512,6 +1600,8 @@ synack: // The peer has aborted our connection. set_state(c, CLOSED); errno = ECONNRESET; + buffer_clear(&c->sndbuf); + buffer_clear(&c->rcvbuf); if(c->recv) { c->recv(c, NULL, 0); @@ -1558,6 +1648,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) { @@ -1721,6 +1816,7 @@ skip_ack: c->snd.last++; set_state(c, FIN_WAIT_1); } else { + c->do_poll = true; set_state(c, ESTABLISHED); } @@ -1779,8 +1875,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: @@ -1971,8 +2074,8 @@ static bool reset_connection(struct utcp_connection *c) { return false; } - c->recv = NULL; - c->poll = NULL; + buffer_clear(&c->sndbuf); + buffer_clear(&c->rcvbuf); switch(c->state) { case CLOSED: @@ -2002,17 +2105,27 @@ 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; + hdr.aux = 0; print_packet(c, "send", &hdr, sizeof(hdr)); c->utcp->send(c->utcp, &hdr, sizeof(hdr)); return true; } -// Closes all the opened connections -void utcp_abort_all_connections(struct utcp *utcp) { +static void set_reapable(struct utcp_connection *c) { + set_buffer_storage(&c->sndbuf, NULL, min(c->sndbuf.maxsize, DEFAULT_MAXSNDBUFSIZE)); + set_buffer_storage(&c->rcvbuf, NULL, min(c->rcvbuf.maxsize, DEFAULT_MAXRCVBUFSIZE)); + + c->recv = NULL; + c->poll = NULL; + c->reapable = true; +} + +// Resets all connections, but does not invalidate connection handles +void utcp_reset_all_connections(struct utcp *utcp) { if(!utcp) { errno = EINVAL; return; @@ -2025,19 +2138,16 @@ void utcp_abort_all_connections(struct utcp *utcp) { continue; } - utcp_recv_t old_recv = c->recv; - utcp_poll_t old_poll = c->poll; - reset_connection(c); - if(old_recv) { + if(c->recv) { errno = 0; - old_recv(c, NULL, 0); + c->recv(c, NULL, 0); } - if(old_poll && !c->reapable) { + if(c->poll && !c->reapable) { errno = 0; - old_poll(c, 0); + c->poll(c, 0); } } @@ -2049,9 +2159,7 @@ int utcp_close(struct utcp_connection *c) { return -1; } - c->recv = NULL; - c->poll = NULL; - c->reapable = true; + set_reapable(c); return 0; } @@ -2060,7 +2168,7 @@ int utcp_abort(struct utcp_connection *c) { return -1; } - c->reapable = true; + set_reapable(c); return 0; } @@ -2096,6 +2204,8 @@ struct timespec utcp_timeout(struct utcp *utcp) { if(timespec_isset(&c->conn_timeout) && timespec_lt(&c->conn_timeout, &now)) { errno = ETIMEDOUT; c->state = CLOSED; + buffer_clear(&c->sndbuf); + buffer_clear(&c->rcvbuf); if(c->recv) { c->recv(c, NULL, 0); @@ -2155,7 +2265,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; @@ -2181,7 +2291,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 @@ -2198,6 +2308,9 @@ void utcp_exit(struct utcp *utcp) { struct utcp_connection *c = utcp->connections[i]; if(!c->reapable) { + buffer_clear(&c->sndbuf); + buffer_clear(&c->rcvbuf); + if(c->recv) { c->recv(c, NULL, 0); } @@ -2315,16 +2428,12 @@ size_t utcp_get_sndbuf_free(struct utcp_connection *c) { } } -void utcp_set_sndbuf(struct utcp_connection *c, size_t size) { +void utcp_set_sndbuf(struct utcp_connection *c, void *data, size_t size) { if(!c) { return; } - c->sndbuf.maxsize = size; - - if(c->sndbuf.maxsize != size) { - c->sndbuf.maxsize = -1; - } + set_buffer_storage(&c->sndbuf, data, size); c->do_poll = is_reliable(c) && buffer_free(&c->sndbuf); } @@ -2341,16 +2450,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) { @@ -2398,10 +2503,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; } } @@ -2428,6 +2533,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);