summaryrefslogtreecommitdiff
path: root/code/core/rbuf.c
diff options
context:
space:
mode:
authorUros Majstorovic <majstor@majstor.org>2017-08-18 20:35:21 +0200
committerUros Majstorovic <majstor@majstor.org>2017-08-18 20:35:21 +0200
commit5d20e9bafc3571f37eb0d9b74699d023d2d3d13a (patch)
treebb26005ff3ed2564212ac84cfa336ee6e97ffb93 /code/core/rbuf.c
parent8f44e2151cb3f91b220c4a3393a06068d0ee7302 (diff)
timer fixed; rbuf almost implemented
Diffstat (limited to 'code/core/rbuf.c')
-rw-r--r--code/core/rbuf.c114
1 files changed, 88 insertions, 26 deletions
diff --git a/code/core/rbuf.c b/code/core/rbuf.c
index b384774..ee9a7fc 100644
--- a/code/core/rbuf.c
+++ b/code/core/rbuf.c
@@ -37,33 +37,88 @@ ssize_t ecp_rbuf_msg_store(ECPRBuffer *rbuf, ecp_seq_t seq, int idx, unsigned ch
if (rbuf->msg == NULL) return 0;
if (test_flags && (test_flags & rbuf->msg[idx].flags)) return ECP_ERR_RBUF_DUP;
- if (ECP_RBUF_SEQ_LT(rbuf->seq_max, seq)) rbuf->seq_max = seq;
-
- if (!(set_flags & ECP_RBUF_FLAG_DELIVERED)) {
- memcpy(rbuf->msg[idx].msg, msg, msg_size);
- rbuf->msg[idx].size = msg_size;
- }
+ if (msg_size) memcpy(rbuf->msg[idx].msg, msg, msg_size);
+ rbuf->msg[idx].size = msg_size;
rbuf->msg[idx].flags = set_flags;
return msg_size;
}
-ssize_t ecp_conn_rbuf_pkt_send(ECPConnection *conn, ECPNetAddr *addr, unsigned char *packet, size_t pkt_size, ecp_seq_t seq, int idx) {
- int do_send = 1;
- ECPRBSend *buf = conn->rbuf.send;
+ssize_t ecp_rbuf_pld_send(ECPConnection *conn, unsigned char *payload, size_t payload_size, ecp_seq_t seq) {
+ unsigned char packet[ECP_MAX_PKT];
+ ECPSocket *sock = conn->sock;
+ ECPContext *ctx = sock->ctx;
+ ECPNetAddr addr;
+ ECPRBInfo rbuf_info;
+ ssize_t rv;
- ssize_t rv = ecp_rbuf_msg_store(&buf->rbuf, seq, idx, packet, pkt_size, 0, 0);
+ int _rv = ecp_rbuf_info_init(&rbuf_info);
+ if (_rv) return _rv;
+
+ rbuf_info.seq = seq;
+ rbuf_info.seq_force = 1;
+
+ rv = ctx->pack(conn, packet, ECP_MAX_PKT, ECP_ECDH_IDX_INV, ECP_ECDH_IDX_INV, payload, payload_size, &addr, &rbuf_info);
if (rv < 0) return rv;
- if (buf->win_size) {
+ rv = ecp_pkt_send(sock, &addr, packet, rv);
+ if (rv < 0) return rv;
+
+ return rv;
+}
+
+int ecp_rbuf_info_init(ECPRBInfo *rbuf_info) {
+ memset(rbuf_info, 0, sizeof(ECPRBInfo));
+
+ return ECP_OK;
+}
+
+int ecp_rbuf_pkt_prep(ECPRBSend *buf, ecp_seq_t seq, unsigned char mtype, ECPRBInfo *rbuf_info) {
+ if (buf == NULL) return ECP_ERR;
+
+#ifdef ECP_WITH_PTHREAD
+ pthread_mutex_lock(&buf->mutex);
+#endif
+ int idx = ecp_rbuf_msg_idx(&buf->rbuf, seq);
+#ifdef ECP_WITH_PTHREAD
+ pthread_mutex_unlock(&buf->mutex);
+#endif
+
+ if (idx < 0) return idx;
+
+ rbuf_info->seq = seq;
+ rbuf_info->idx = idx;
+ rbuf_info->mtype = mtype;
+ buf->rbuf.msg[idx].size = 0;
+ buf->rbuf.msg[idx].flags = 0;
+
+ return ECP_OK;
+}
+
+ssize_t ecp_rbuf_pkt_send(ECPRBSend *buf, ECPSocket *sock, ECPNetAddr *addr, ECPTimerItem *ti, unsigned char *packet, size_t pkt_size, ECPRBInfo *rbuf_info) {
+ unsigned char flags = 0;
+ int do_send = 1;
+ ssize_t rv = 0;
+ ecp_seq_t seq = rbuf_info->seq;
+ unsigned int idx = rbuf_info->idx;
+ unsigned char mtype = rbuf_info->mtype & ECP_MTYPE_MASK;
+
+ if (mtype < ECP_MAX_MTYPE_SYS) flags |= ECP_RBUF_FLAG_SYS;
+
+ rv = ecp_rbuf_msg_store(&buf->rbuf, seq, idx, packet, pkt_size, 0, flags);
+ if (rv < 0) return rv;
+
+ if (buf->flags & ECP_RBUF_FLAG_CCONTROL) {
#ifdef ECP_WITH_PTHREAD
pthread_mutex_lock(&buf->mutex);
#endif
- if (buf->cc_wait || (buf->in_transit >= buf->win_size)) {
- if (!buf->cc_wait) buf->seq_cc = seq;
- buf->cc_wait++;
- buf->rbuf.msg[idx].flags |= ECP_RBUF_FLAG_CCWAIT;
+ if (ECP_RBUF_SEQ_LT(buf->rbuf.seq_max, seq)) buf->rbuf.seq_max = seq;
+
+ if (buf->cnt_cc || (buf->in_transit >= buf->win_size)) {
+ if (!buf->cnt_cc) buf->seq_cc = seq;
+ buf->cnt_cc++;
+ buf->rbuf.msg[idx].flags |= ECP_RBUF_FLAG_IN_CCONTROL;
do_send = 0;
} else {
buf->in_transit++;
@@ -74,36 +129,43 @@ ssize_t ecp_conn_rbuf_pkt_send(ECPConnection *conn, ECPNetAddr *addr, unsigned c
#endif
}
- if (do_send) rv = ecp_pkt_send(conn->sock, addr, packet, pkt_size);
+ if (do_send) {
+ int _rv;
+ if (ti) {
+ _rv = ecp_timer_push(ti);
+ if (_rv) return _rv;
+ }
+ rv = ecp_pkt_send(sock, addr, packet, pkt_size);
+ }
return rv;
}
-int ecp_conn_rbuf_create(ECPConnection *conn, ECPRBSend *buf_s, ECPRBMessage *msg_s, unsigned int msg_s_size, ECPRBRecv *buf_r, ECPRBMessage *msg_r, unsigned int msg_r_size) {
+int ecp_rbuf_conn_create(ECPConnection *conn, ECPRBSend *buf_s, ECPRBMessage *msg_s, unsigned int msg_s_size, ECPRBRecv *buf_r, ECPRBMessage *msg_r, unsigned int msg_r_size) {
int rv;
- rv = ecp_conn_rbuf_send_create(conn, buf_s, msg_s, msg_s_size);
+ rv = ecp_rbuf_send_create(conn, buf_s, msg_s, msg_s_size);
if (rv) return rv;
- rv = ecp_conn_rbuf_recv_create(conn, buf_r, msg_r, msg_r_size);
+ rv = ecp_rbuf_recv_create(conn, buf_r, msg_r, msg_r_size);
if (rv) {
- ecp_conn_rbuf_send_destroy(conn);
+ ecp_rbuf_send_destroy(conn);
return rv;
}
return ECP_OK;
}
-void ecp_conn_rbuf_destroy(ECPConnection *conn) {
- ecp_conn_rbuf_send_destroy(conn);
- ecp_conn_rbuf_recv_destroy(conn);
+void ecp_rbuf_conn_destroy(ECPConnection *conn) {
+ ecp_rbuf_send_destroy(conn);
+ ecp_rbuf_recv_destroy(conn);
}
-int ecp_conn_rbuf_start(ECPConnection *conn, ecp_seq_t seq) {
- int rv = ecp_conn_rbuf_send_start(conn);
+int ecp_rbuf_conn_start(ECPConnection *conn, ecp_seq_t seq) {
+ int rv = ecp_rbuf_send_start(conn);
if (rv) return rv;
if (!conn->out) {
- rv = ecp_conn_rbuf_recv_start(conn, seq);
+ rv = ecp_rbuf_recv_start(conn, seq);
if (rv) return rv;
}