write_queue: Check the result of osmo_wqueue_enqueue and free

The write_queue is designed to have a maximum amount of pending
messages and will refuse to take new messages when it has been
reached. The caller can decide if it wants to flush the queue
and add the message again, create a log. But in all cases the
ownership of the msgb has not been transferred. Fix the potential
memory leak in the failure situation.

2 files changed, 39 insertions, 9 deletions

diff --git a/src/osmo-bts-sysmo/l1_fwd_main.c b/src/osmo-bts-sysmo/l1_fwd_main.c
index cb05e15c..92d2ea47 100644
--- a/src/osmo-bts-sysmo/l1_fwd_main.c
+++ b/src/osmo-bts-sysmo/l1_fwd_main.c
@@ -77,7 +77,12 @@ int l1if_handle_l1prim(int wq, struct femtol1_hdl *fl1h, struct msgb *msg)
 	struct l1fwd_hdl *l1fh = fl1h->priv;
 
 	/* Enqueue message to UDP socket */
-	return osmo_wqueue_enqueue(&l1fh->udp_wq[wq], msg);
+	if (osmo_wqueue_enqueue(&l1fh->udp_wq[wq], msg) != 0) {
+		LOGP(DL1C, LOGL_ERROR, "Write queue %d full. dropping msg\n", wq);
+		msgb_free(msg);
+		return -EAGAIN;
+	}
+	return 0;
 }
 
 /* callback when there's a new SYS primitive coming in from the HW */
@@ -86,7 +91,12 @@ int l1if_handle_sysprim(struct femtol1_hdl *fl1h, struct msgb *msg)
 	struct l1fwd_hdl *l1fh = fl1h->priv;
 
 	/* Enqueue message to UDP socket */
-	return osmo_wqueue_enqueue(&l1fh->udp_wq[MQ_SYS_WRITE], msg);
+	if (osmo_wqueue_enqueue(&l1fh->udp_wq[MQ_SYS_WRITE], msg) != 0) {
+		LOGP(DL1C, LOGL_ERROR, "MQ_SYS_WRITE ful. dropping msg\n");
+		msgb_free(msg);
+		return -EAGAIN;
+	}
+	return 0;
 }
 
 
@@ -121,9 +131,13 @@ static int udp_read_cb(struct osmo_fd *ofd)
 		ofd->priv_nr);
 
 	/* put the message into the right queue */
-	rc = osmo_wqueue_enqueue(&fl1h->write_q[ofd->priv_nr], msg);
-	
-	return rc;
+	if (osmo_wqueue_enqueue(&fl1h->write_q[ofd->priv_nr], msg) != 0) {
+		LOGP(DL1C, LOGL_ERROR, "Write queue %d full. dropping msg\n",
+			ofd->priv_nr);
+		msgb_free(msg);
+		return -EAGAIN;
+	}
+	return 0;
 }
 
 /* callback when we can write to the UDP socket */
diff --git a/src/osmo-bts-sysmo/l1_if.c b/src/osmo-bts-sysmo/l1_if.c
index 9423cfe0..0159607a 100644
--- a/src/osmo-bts-sysmo/l1_if.c
+++ b/src/osmo-bts-sysmo/l1_if.c
@@ -238,7 +238,12 @@ static int _l1if_req_compl(struct femtol1_hdl *fl1h, struct msgb *msg,
 	}
 
 	/* enqueue the message in the queue and add wsc to list */
-	osmo_wqueue_enqueue(wqueue, msg);
+	if (osmo_wqueue_enqueue(wqueue, msg) != 0) {
+		/* So we will get a timeout but the log message might help */
+		LOGP(DL1C, LOGL_ERROR, "Write queue for %s full. dropping msg.\n",
+			is_system_prim ? "system primitive" : "gsm");
+		msgb_free(msg);
+	}
 	llist_add(&wlc->list, &fl1h->wlc_list);
 
 	/* schedule a timer for timeout_secs seconds. If DSP fails to respond, we terminate */
@@ -603,7 +608,10 @@ tx:
 	tx_to_gsmtap(fl1, resp_msg);
 
 	/* transmit */
-	osmo_wqueue_enqueue(&fl1->write_q[MQ_L1_WRITE], resp_msg);
+	if (osmo_wqueue_enqueue(&fl1->write_q[MQ_L1_WRITE], resp_msg) != 0) {
+		LOGP(DL1C, LOGL_ERROR, "MQ_L1_WRITE queue full. Dropping msg.\n");
+		msgb_free(resp_msg);
+	}
 
 	return 0;
 
@@ -1421,7 +1429,12 @@ int l1if_set_trace_flags(struct femtol1_hdl *hdl, uint32_t flags)
 	hdl->dsp_trace_f = flags;
 
 	/* There is no confirmation we could wait for */
-	return osmo_wqueue_enqueue(&hdl->write_q[MQ_SYS_WRITE], msg);
+	if (osmo_wqueue_enqueue(&hdl->write_q[MQ_SYS_WRITE], msg) != 0) {
+		LOGP(DL1C, LOGL_ERROR, "MQ_SYS_WRITE queue full. Dropping msg\n");
+		msgb_free(msg);
+		return -EAGAIN;
+	}
+	return 0;
 }
 
 /* send packet data request to L1 */
@@ -1459,7 +1472,10 @@ int l1if_pdch_req(struct gsm_bts_trx_ts *ts, int is_ptcch, uint32_t fn,
 	tx_to_gsmtap(fl1h, msg);
 
 	/* transmit */
-	osmo_wqueue_enqueue(&fl1h->write_q[MQ_L1_WRITE], msg);
+	if (osmo_wqueue_enqueue(&fl1h->write_q[MQ_L1_WRITE], msg) != 0) {
+		LOGP(DL1P, LOGL_ERROR, "MQ_L1_WRITE queue full. Dropping msg.\n");
+		msgb_free(msg);
+	}
 
 	return 0;
 }