1 files changed, 57 insertions, 11 deletions

diff --git a/lib/libtle_l4p/tcp_ctl.h b/lib/libtle_l4p/tcp_ctl.h
index bec1e76..3196470 100644
--- a/lib/libtle_l4p/tcp_ctl.h
+++ b/lib/libtle_l4p/tcp_ctl.h
@@ -22,6 +22,7 @@
 
 #include "tcp_stream.h"
 #include "tcp_ofo.h"
+#include "tcp_timer.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -97,10 +98,10 @@ calc_rx_wnd(const struct tle_tcp_stream *s, uint32_t scale)
 
 	/* peer doesn't support WSCALE option, wnd size is limited to 64K */
 	if (scale == TCP_WSCALE_NONE) {
-		wnd = _rte_ring_get_mask(s->rx.q) << TCP_WSCALE_DEFAULT;
+		wnd = rte_ring_free_count(s->rx.q) << TCP_WSCALE_DEFAULT;
 		return RTE_MIN(wnd, (uint32_t)UINT16_MAX);
 	} else
-		return  _rte_ring_get_mask(s->rx.q) << scale;
+		return rte_ring_free_count(s->rx.q) << scale;
 }
 
 /* empty stream's send queue */
@@ -144,31 +145,34 @@ static inline void
 tcp_stream_reset(struct tle_ctx *ctx, struct tle_tcp_stream *s)
 {
 	struct stbl *st;
-	uint16_t uop;
+	uint16_t state;
+	uint8_t i;
 
 	st = CTX_TCP_STLB(ctx);
 
-	/* reset TX armed */
-	rte_atomic32_set(&s->tx.arm, 0);
+	for (i = 0; i < TIMER_NUM; i++)
+		timer_stop(s, i);
 
 	/* reset TCB */
-	uop = s->tcb.uop & ~TCP_OP_CLOSE;
+	state = s->tcb.state;
 	memset(&s->tcb, 0, sizeof(s->tcb));
 
 	/* reset cached destination */
 	memset(&s->tx.dst, 0, sizeof(s->tx.dst));
 
-	if (uop != TCP_OP_ACCEPT) {
+	/* state could be ESTABLISHED, CLOSED or LISTEN
+	 * stream in CLOSED state has already been cleared by stream_term
+	 * stream in ESTABLISHED state is accepted stream, and doesn't need clear
+	 */
+	if (state == TCP_ST_LISTEN) {
 		/* free stream's destination port */
 		stream_clear_ctx(ctx, &s->s);
-		if (uop == TCP_OP_LISTEN)
-			empty_lq(s);
+		empty_lq(s);
 	}
 
 	if (s->ste != NULL) {
 		/* remove entry from RX streams table */
-		stbl_del_stream(st, s->ste, s,
-			(s->flags & TLE_CTX_FLAG_ST) == 0);
+		stbl_del_stream(st, s->ste, &s->s);
 		s->ste = NULL;
 		empty_rq(s);
 	}
@@ -184,6 +188,48 @@ tcp_stream_reset(struct tle_ctx *ctx, struct tle_tcp_stream *s)
 	put_stream(ctx, &s->s, TCP_STREAM_TX_FINISHED(s));
 }
 
+static inline void
+stream_term(struct tle_tcp_stream *s)
+{
+	struct sdr *dr;
+
+	/* 1) recv a RST packet; 2) keepalive timeout */
+	if (s->tcb.state == TCP_ST_ESTABLISHED) {
+		TCP_DEC_STATS_ATOMIC(TCP_MIB_CURRESTAB);
+		TCP_INC_STATS(TCP_MIB_ESTABRESETS);
+	}
+
+	s->tcb.state = TCP_ST_CLOSED;
+	rte_smp_wmb();
+
+	/* close() was already invoked, schedule final cleanup */
+	if ((s->tcb.uop & TCP_OP_CLOSE) != 0) {
+		if ((s->tcb.uop & TCP_OP_ACCEPT) == 0) {
+			/* free stream's destination port */
+			stream_clear_ctx(s->s.ctx, &s->s);
+			if ((s->tcb.uop & TCP_OP_LISTEN) != 0)
+				empty_lq(s);
+		}
+
+		if (s->ste != NULL) {
+			/* remove entry from RX streams table */
+			stbl_del_stream(CTX_TCP_STLB(s->s.ctx), s->ste, &s->s);
+			s->ste = NULL;
+			empty_rq(s);
+		}
+
+		dr = CTX_TCP_SDR(s->s.ctx);
+		rte_spinlock_lock(&dr->lock);
+		STAILQ_INSERT_TAIL(&dr->be, &s->s, link);
+		rte_spinlock_unlock(&dr->lock);
+
+	/* notify user that stream need to be closed */
+	} else if (s->err.ev != NULL)
+		tle_event_raise(s->err.ev);
+	else if (s->err.cb.func != NULL)
+		s->err.cb.func(s->err.cb.data, &s->s);
+}
+
 #ifdef __cplusplus
 }
 #endif