abis_nm: Band aid on OML initialisation by queuing messages

Instead of sending many messages we will queue the OML
messages and wait for the ACK/NACK before sending the
next message from the queue. We tag the msgb to remember
if we need to wait for an ack or not.

We keep the order of all messages, on ACKs and similiar
occassions we will drown the queue until we reach a message
that needs to be acked and then wait for that ack again.

Possible breakage can appear when we send an OML (e.g.
BS11 specific message) msg which does not need to be acked
through the abis_nm_sendmsg call. The fix will be to use
the _direct version of this method.

Conflicts:

	openbsc/include/openbsc/abis_nm.h
	openbsc/include/openbsc/gsm_data.h
	openbsc/src/gsm_data.c
	openbsc/src/input/ipaccess.c

6 files changed, 80 insertions, 9 deletions

diff --git a/openbsc/include/openbsc/abis_nm.h b/openbsc/include/openbsc/abis_nm.h
index c20e4e172..79d27301d 100644
--- a/openbsc/include/openbsc/abis_nm.h
+++ b/openbsc/include/openbsc/abis_nm.h
@@ -170,4 +170,6 @@ int nm_state_event(enum nm_evt evt, u_int8_t obj_class, void *obj,
 const char *nm_opstate_name(u_int8_t os);
 const char *nm_avail_name(u_int8_t avail);
 int nm_is_running(struct gsm_nm_state *s);
+void abis_nm_clear_queue(struct gsm_bts *bts);
+
 #endif /* _NM_H */
diff --git a/openbsc/include/openbsc/gsm_data.h b/openbsc/include/openbsc/gsm_data.h
index 9e9dc6d52..9cf03efa9 100644
--- a/openbsc/include/openbsc/gsm_data.h
+++ b/openbsc/include/openbsc/gsm_data.h
@@ -81,6 +81,7 @@ enum bts_gprs_mode {
 	BTS_GPRS_EGPRS = 2,
 };
 
+#define OBSC_NM_W_ACK_CB(__msgb) (__msgb)->cb[3]
 struct msgb;
 typedef int gsm_cbfn(unsigned int hooknum,
 		     unsigned int event,
@@ -540,6 +541,10 @@ struct gsm_bts {
 	/* transceivers */
 	int num_trx;
 	struct llist_head trx_list;
+
+	/* Abis NM queue */
+	struct llist_head abis_queue;
+	int abis_nm_pend;
 };
 
 /* Some statistics of our network */
diff --git a/openbsc/src/abis_nm.c b/openbsc/src/abis_nm.c
index 868502d5f..c07d516cf 100644
--- a/openbsc/src/abis_nm.c
+++ b/openbsc/src/abis_nm.c
@@ -410,11 +410,31 @@ static struct msgb *nm_msgb_alloc(void)
 }
 
 /* Send a OML NM Message from BSC to BTS */
-int abis_nm_sendmsg(struct gsm_bts *bts, struct msgb *msg)
+static int abis_nm_queue_msg(struct gsm_bts *bts, struct msgb *msg)
 {
 	msg->trx = bts->c0;
 
-	return _abis_nm_sendmsg(msg);
+	/* queue OML messages */
+	if (llist_empty(&bts->abis_queue) && !bts->abis_nm_pend) {
+		bts->abis_nm_pend = OBSC_NM_W_ACK_CB(msg);
+		return _abis_nm_sendmsg(msg);
+	} else {
+		msgb_enqueue(&bts->abis_queue, msg);
+		return 0;
+	}
+
+}
+
+int abis_nm_sendmsg(struct gsm_bts *bts, struct msgb *msg)
+{
+	OBSC_NM_W_ACK_CB(msg) = 1;
+	return abis_nm_queue_msg(bts, msg);
+}
+
+static int abis_nm_sendmsg_direct(struct gsm_bts *bts, struct msgb *msg)
+{
+	OBSC_NM_W_ACK_CB(msg) = 0;
+	return abis_nm_queue_msg(bts, msg);
 }
 
 static int abis_nm_rcvmsg_sw(struct msgb *mb);
@@ -952,12 +972,30 @@ static int abis_nm_rx_lmt_event(struct msgb *mb)
 	return 0;
 }
 
+static void abis_nm_queue_send_next(struct gsm_bts *bts)
+{
+	int wait = 0;
+	struct msgb *msg;
+	/* the queue is empty */
+	while (!llist_empty(&bts->abis_queue)) {
+		msg = msgb_dequeue(&bts->abis_queue);
+		wait = OBSC_NM_W_ACK_CB(msg);
+		_abis_nm_sendmsg(msg);
+
+		if (wait)
+			break;
+	}
+
+	bts->abis_nm_pend = wait;
+}
+
 /* Receive a OML NM Message from BTS */
 static int abis_nm_rcvmsg_fom(struct msgb *mb)
 {
 	struct abis_om_hdr *oh = msgb_l2(mb);
 	struct abis_om_fom_hdr *foh = msgb_l3(mb);
 	u_int8_t mt = foh->msg_type;
+	int ret = 0;
 
 	/* check for unsolicited message */
 	if (is_report(mt))
@@ -981,6 +1019,7 @@ static int abis_nm_rcvmsg_fom(struct msgb *mb)
 			DEBUGPC(DNM, "\n");
 
 		dispatch_signal(SS_NM, S_NM_NACK, (void*) &mt);
+		abis_nm_queue_send_next(mb->trx->bts);
 		return 0;
 	}
 #if 0
@@ -1002,13 +1041,13 @@ static int abis_nm_rcvmsg_fom(struct msgb *mb)
 
 	switch (mt) {
 	case NM_MT_CHG_ADM_STATE_ACK:
-		return abis_nm_rx_chg_adm_state_ack(mb);
+		ret = abis_nm_rx_chg_adm_state_ack(mb);
 		break;
 	case NM_MT_SW_ACT_REQ:
-		return abis_nm_rx_sw_act_req(mb);
+		ret = abis_nm_rx_sw_act_req(mb);
 		break;
 	case NM_MT_BS11_LMT_SESSION:
-		return abis_nm_rx_lmt_event(mb);
+		ret = abis_nm_rx_lmt_event(mb);
 		break;
 	case NM_MT_CONN_MDROP_LINK_ACK:
 		DEBUGP(DNM, "CONN MDROP LINK ACK\n");
@@ -1021,7 +1060,8 @@ static int abis_nm_rcvmsg_fom(struct msgb *mb)
 		break;
 	}
 
-	return 0;
+	abis_nm_queue_send_next(mb->trx->bts);
+	return ret;
 }
 
 static int abis_nm_rx_ipacc(struct msgb *mb);
@@ -1034,6 +1074,7 @@ static int abis_nm_rcvmsg_manuf(struct msgb *mb)
 	switch (bts_type) {
 	case GSM_BTS_TYPE_NANOBTS:
 		rc = abis_nm_rx_ipacc(mb);
+		abis_nm_queue_send_next(mb->trx->bts);
 		break;
 	default:
 		LOGP(DNM, LOGL_ERROR, "don't know how to parse OML for this "
@@ -1277,7 +1318,7 @@ static int sw_load_segment(struct abis_nm_sw *sw)
 			sw->obj_instance[0], sw->obj_instance[1],
 			sw->obj_instance[2]);
 
-	return abis_nm_sendmsg(sw->bts, msg);
+	return abis_nm_sendmsg_direct(sw->bts, msg);
 }
 
 /* 6.2.4 / 8.3.4 Load Data End */
@@ -1470,6 +1511,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 					 sw->cb_data, NULL);
 			rc = sw_fill_window(sw);
 			sw->state = SW_STATE_WAIT_SEGACK;
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		case NM_MT_LOAD_INIT_NACK:
 			if (sw->forced) {
@@ -1490,6 +1532,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 						 sw->cb_data, NULL);
 				sw->state = SW_STATE_ERROR;
 			}
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		}
 		break;
@@ -1510,6 +1553,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 				sw->state = SW_STATE_WAIT_ENDACK;
 				rc = sw_load_end(sw);
 			}
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		case NM_MT_LOAD_ABORT:
 			if (sw->cbfn)
@@ -1531,6 +1575,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 					 NM_MT_LOAD_END_ACK, mb,
 					 sw->cb_data, NULL);
 			rc = 0;
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		case NM_MT_LOAD_END_NACK:
 			if (sw->forced) {
@@ -1550,6 +1595,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 						 NM_MT_LOAD_END_NACK, mb,
 						 sw->cb_data, NULL);
 			}
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		}
 	case SW_STATE_WAIT_ACTACK:
@@ -1563,6 +1609,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 				sw->cbfn(GSM_HOOK_NM_SWLOAD,
 					 NM_MT_ACTIVATE_SW_ACK, mb,
 					 sw->cb_data, NULL);
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		case NM_MT_ACTIVATE_SW_NACK:
 			DEBUGP(DNM, "Activate Software NACK\n");
@@ -1572,6 +1619,7 @@ static int abis_nm_rcvmsg_sw(struct msgb *mb)
 				sw->cbfn(GSM_HOOK_NM_SWLOAD,
 					 NM_MT_ACTIVATE_SW_NACK, mb,
 					 sw->cb_data, NULL);
+			abis_nm_queue_send_next(mb->trx->bts);
 			break;
 		}
 	case SW_STATE_NONE:
@@ -2014,7 +2062,7 @@ int abis_nm_sw_act_req_ack(struct gsm_bts *bts, u_int8_t obj_class, u_int8_t i1,
 	if (nack)
 		msgb_tv_put(msg, NM_ATT_NACK_CAUSES, NM_NACK_OBJCLASS_NOTSUPP);
 
-	return abis_nm_sendmsg(bts, msg);
+	return abis_nm_sendmsg_direct(bts, msg);
 }
 
 int abis_nm_raw_msg(struct gsm_bts *bts, int len, u_int8_t *rawmsg)
@@ -3016,3 +3064,15 @@ int ipac_parse_bcch_info(struct ipac_bcch_info *binf, u_int8_t *buf)
 
 	return 0;
 }
+
+void abis_nm_clear_queue(struct gsm_bts *bts)
+{
+	struct msgb *msg;
+
+	while (!llist_empty(&bts->abis_queue)) {
+		msg = msgb_dequeue(&bts->abis_queue);
+		msgb_free(msg);
+	}
+
+	bts->abis_nm_pend = 0;
+}
diff --git a/openbsc/src/bsc_init.c b/openbsc/src/bsc_init.c
index 35c861176..037845b11 100644
--- a/openbsc/src/bsc_init.c
+++ b/openbsc/src/bsc_init.c
@@ -979,6 +979,8 @@ void input_event(int event, enum e1inp_sign_type type, struct gsm_bts_trx *trx)
 		trx->nm_state.availability = 0;
 		trx->bb_transc.nm_state.operational = 0;
 		trx->bb_transc.nm_state.availability = 0;
+
+		abis_nm_clear_queue(trx->bts);
 		break;
 	default:
 		break;
diff --git a/openbsc/src/gsm_data.c b/openbsc/src/gsm_data.c
index baeb36dde..6284e2a78 100644
--- a/openbsc/src/gsm_data.c
+++ b/openbsc/src/gsm_data.c
@@ -234,6 +234,8 @@ struct gsm_bts *gsm_bts_alloc(struct gsm_network *net, enum gsm_bts_type type,
 	bts->rach_b_thresh = -1;
 	bts->rach_ldavg_slots = -1;
 
+	INIT_LLIST_HEAD(&bts->abis_queue);
+
 	llist_add_tail(&bts->list, &net->bts_list);
 
 	return bts;
diff --git a/openbsc/src/input/ipaccess.c b/openbsc/src/input/ipaccess.c
index 8c99157e0..aa3d85a64 100644
--- a/openbsc/src/input/ipaccess.c
+++ b/openbsc/src/input/ipaccess.c
@@ -573,7 +573,7 @@ static int handle_ts1_write(struct bsc_fd *bfd)
 	e1i_ts->sign.tx_timer.data = e1i_ts;
 
 	/* Reducing this might break the nanoBTS 900 init. */
-	bsc_schedule_timer(&e1i_ts->sign.tx_timer, 0, e1i_ts->sign.delay);
+	bsc_schedule_timer(&e1i_ts->sign.tx_timer, 0, 0);
 
 	return ret;
 }