51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
-#define _GNU_SOURCE
-
-#include <assert.h>
-#include <errno.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <stdint.h>
-#include <stdbool.h>
-#include <string.h>
-#include <unistd.h>
+#include "system.h"
#include <time.h>
#include "utcp_priv.h"
uint32_t realoffset = buf->offset + offset;
- if(buf->size - buf->offset < offset) {
+ if(buf->size - buf->offset <= offset) {
// The offset wrapped
realoffset -= buf->size;
}
uint32_t realoffset = buf->offset + offset;
- if(buf->size - buf->offset < offset) {
+ if(buf->size - buf->offset <= offset) {
// The offset wrapped
realoffset -= buf->size;
}
}
// 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;
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;
return rx1 + rx2;
}
} else {
- return cb(arg, buf->data + realoffset, len);
+ return c->recv(c, buf->data + realoffset, len);
}
}
len = buf->used;
}
- if(buf->size - buf->offset < len) {
+ if(buf->size - buf->offset <= len) {
buf->offset -= buf->size;
}
}
static void buffer_exit(struct buffer *buf) {
- free(buf->data);
+ if(!buf->external) {
+ free(buf->data);
+ }
+
memset(buf, 0, sizeof(*buf));
}
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;
debug(c, "accepted %p %p\n", c, recv, priv);
c->recv = recv;
c->priv = priv;
+ c->do_poll = true;
set_state(c, ESTABLISHED);
}
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();
}
}
}
}
// 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;
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;
}
if(c->state == CLOSED) {
debug(c, "got packet for closed connection\n");
- return 0;
+ goto reset;
}
// It is for an existing connection.
}
}
- 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) {
// 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) {
c->snd.last++;
set_state(c, FIN_WAIT_1);
} else {
+ c->do_poll = true;
set_state(c, ESTABLISHED);
}
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:
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;
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;
}
utcp->accept = accept;
- utcp->pre_accept = pre_accept;
+ utcp->listen = listen;
utcp->send = send;
utcp->priv = priv;
utcp->timeout = DEFAULT_USER_TIMEOUT; // sec
}
}
-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;
+}
+
+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;
+ }
- if(c->sndbuf.maxsize != size) {
- c->sndbuf.maxsize = -1;
+ // 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);
}
}
}
-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) {
}
}
-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;
}
}
}
}
+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);