ipc: fix per channel start/stop, ipc uhd backend

Start/stop was previously fixed to channel 0, so proper multichannel
operation without using the mcbts mode was not possible.
UHD multichannel is rather annoying due to the alignment dance which
returns 0 reads, which may not be submitted to the ipc interface, since
empty buffers are treated as errors.

Change-Id: I441b1977e30a6c6c96b2e0543cedb3ce54d3ce31

5 files changed, 120 insertions, 25 deletions

diff --git a/Transceiver52M/device/ipc/IPCDevice.cpp b/Transceiver52M/device/ipc/IPCDevice.cpp
index 3b23d6a..a71adac 100644
--- a/Transceiver52M/device/ipc/IPCDevice.cpp
+++ b/Transceiver52M/device/ipc/IPCDevice.cpp
@@ -75,6 +75,7 @@ IPCDevice::IPCDevice(size_t tx_sps, size_t rx_sps, InterfaceType iface, size_t c
 		rx_buffers[i] = new smpl_buf(SAMPLE_BUF_SZ / sizeof(uint32_t));
 
 	memset(&sk_chan_state, 0, sizeof(sk_chan_state));
+	memset(&trx_is_started, 0, sizeof(trx_is_started));
 }
 
 IPCDevice::~IPCDevice()
@@ -202,7 +203,8 @@ int IPCDevice::ipc_tx_open_req(struct ipc_sock_state *state, uint32_t num_chans,
 	ipc_prim = (struct ipc_sk_if *)msg->data;
 	ipc_prim->u.open_req.num_chans = num_chans;
 
-	/* FIXME: pass fractional freq */
+	/* FIXME: this is actually the sps value, not the sample rate!
+	 * sample rate is looked up according to the sps rate by uhd backend */
 	ipc_prim->u.open_req.rx_sample_freq_num = rx_sps;
 	ipc_prim->u.open_req.rx_sample_freq_den = 1;
 	ipc_prim->u.open_req.tx_sample_freq_num = tx_sps;
@@ -324,8 +326,8 @@ int IPCDevice::ipc_rx_open_cnf(const struct ipc_sk_if_open_cnf *open_cnf)
 		int rc;
 		LOGC(DDEV, NOTICE) << "chan " << i << ": sk_path=" << open_cnf->chan_info[i].chan_ipc_sk_path;
 
-		/* FIXME: current limit 8 chans, make dynamic */
-		if (i < 8) {
+		/* FIXME: current limit IPC_MAX_NUM_TRX chans, make dynamic */
+		if (i < IPC_MAX_NUM_TRX) {
 			struct ipc_sock_state *state = &sk_chan_state[i];
 			memset(state, 0x00, sizeof(*state));
 
@@ -395,24 +397,50 @@ int IPCDevice::ipc_rx(uint8_t msg_type, struct ipc_sk_if *ipc_prim)
 
 int IPCDevice::ipc_rx_chan_start_cnf(ipc_sk_chan_if_op_rc *ret, uint8_t chan_nr)
 {
-	tmp_state = IPC_IF_MSG_START_CNF;
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal start response for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
+	trx_is_started[chan_nr] = true;
 	return 0;
 }
 int IPCDevice::ipc_rx_chan_stop_cnf(ipc_sk_chan_if_op_rc *ret, uint8_t chan_nr)
 {
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal stop response for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
+	trx_is_started[chan_nr] = false;
 	return 0;
 }
 int IPCDevice::ipc_rx_chan_setgain_cnf(ipc_sk_chan_if_gain *ret, uint8_t chan_nr)
 {
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal setgain response for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
 	ret->is_tx ? tx_gains[chan_nr] = ret->gain : rx_gains[chan_nr] = ret->gain;
 	return 0;
 }
 int IPCDevice::ipc_rx_chan_setfreq_cnf(ipc_sk_chan_if_freq_cnf *ret, uint8_t chan_nr)
 {
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal setfreq response for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
 	return 0;
 }
 int IPCDevice::ipc_rx_chan_notify_underflow(ipc_sk_chan_if_notfiy *ret, uint8_t chan_nr)
 {
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal underfloww notification for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
 	m_ctr[chan_nr].tx_underruns += 1;
 	osmo_signal_dispatch(SS_DEVICE, S_DEVICE_COUNTER_CHANGE, &m_ctr[chan_nr]);
 
@@ -420,6 +448,11 @@ int IPCDevice::ipc_rx_chan_notify_underflow(ipc_sk_chan_if_notfiy *ret, uint8_t
 }
 int IPCDevice::ipc_rx_chan_notify_overflow(ipc_sk_chan_if_notfiy *ret, uint8_t chan_nr)
 {
+	if(chan_nr >= ARRAY_SIZE(trx_is_started)) {
+		LOGC(DDEV, NOTICE) << "shm: illegal overflow notification for chan #" << chan_nr << " ?!?";
+		return 0;
+	}
+
 	m_ctr[chan_nr].rx_overruns += 1;
 	osmo_signal_dispatch(SS_DEVICE, S_DEVICE_COUNTER_CHANGE, &m_ctr[chan_nr]);
 	return 0;
@@ -784,6 +817,17 @@ out_close:
 	return -1;
 }
 
+/* the call stack is rather difficult here, we're already in select:
+>~"#0  IPCDevice::start (this=<optimized out>) at IPCDevice.cpp:789\n"
+>~"#1  in RadioInterface::start (this=0x614000001640) at radioInterface.cpp:187\n"
+>~"#2  in Transceiver::start (this=<optimized out>) at Transceiver.cpp:293\n"
+>~"#3  in Transceiver::ctrl_sock_handle_rx (this=0x61600000b180, chan=0) at Transceiver.cpp:838\n"
+>~"#4  in Transceiver::ctrl_sock_cb (bfd=<optimized out>, flags=1) at Transceiver.cpp:168\n"
+>~"#5  in osmo_fd_disp_fds (_rset=<optimized out>, _wset=<optimized out>, _eset=<optimized out>) at select.c:227\n"
+>~"#6  _osmo_select_main (polling=<optimized out>) at select.c:265\n"
+>~"#7  in osmo_select_main (polling=128) at select.c:274\n"
+>~"#8  in main (argc=<optimized out>, argv=<optimized out>) at osmo-trx.cpp:649\n"
+ * */
 bool IPCDevice::start()
 {
 	LOGC(DDEV, INFO) << "starting IPC...";
@@ -796,15 +840,33 @@ bool IPCDevice::start()
 	struct msgb *msg;
 	struct ipc_sk_chan_if *ipc_prim;
 
-	msg = ipc_msgb_alloc(IPC_IF_MSG_START_REQ);
-	if (!msg)
-		return -ENOMEM;
-	ipc_prim = (struct ipc_sk_chan_if *)msg->data;
-	ipc_prim->u.start_req.dummy = 0;
+	for(int i = 0; i < chans; i++) {
+		msg = ipc_msgb_alloc(IPC_IF_MSG_START_REQ);
+		if (!msg)
+			return -ENOMEM;
+		ipc_prim = (struct ipc_sk_chan_if *)msg->data;
+		ipc_prim->u.start_req.dummy = 0;
 
-	ipc_sock_send(&sk_chan_state[0], msg);
-	while (tmp_state != IPC_IF_MSG_START_CNF)
-		osmo_select_main(0);
+		ipc_sock_send(&sk_chan_state[i], msg);
+	}
+
+	int chan_started_count = 0, retrycount = 0;
+	while (chan_started_count != chans && retrycount < 5) {
+		chan_started_count = 0;
+
+		/* just poll here, we're already in select, so there is no other way to drive
+		 * the fds and "wait" for a response or retry */
+		usleep(100000);
+		osmo_select_main(1);
+
+		for(unsigned int i = 0; i < ARRAY_SIZE(trx_is_started); i++)
+			if(trx_is_started[i] == true)
+				chan_started_count++;
+		retrycount++;
+	}
+
+	if(retrycount >= 5)
+		return false;
 
 	flush_recv(10);
 
@@ -828,7 +890,9 @@ bool IPCDevice::stop()
 	ipc_prim = (struct ipc_sk_chan_if *)msg->data;
 	ipc_prim->u.start_req.dummy = 0;
 
-	ipc_sock_send(&sk_chan_state[0], msg);
+	for(int i = 0; i < chans; i++)
+		if(trx_is_started[i] == true)
+			ipc_sock_send(&sk_chan_state[i], msg);
 
 	started = false;
 	return true;
@@ -1016,6 +1080,9 @@ int IPCDevice::readSamples(std::vector<short *> &bufs, int len, bool *overrun, T
 			expect_timestamp = timestamp + avail_smpls;
 			thread_enable_cancel(true);
 
+			if(num_smpls == -ETIMEDOUT)
+				continue;
+
 			LOGCHAN(i, DDEV, DEBUG)
 			"Received timestamp = " << (TIMESTAMP)recv_timestamp << " (" << num_smpls << ")";
 
@@ -1035,7 +1102,7 @@ int IPCDevice::readSamples(std::vector<short *> &bufs, int len, bool *overrun, T
 
 			rc = rx_buffers[i]->write(bufs[i], num_smpls, (TIMESTAMP)recv_timestamp);
 			if (rc < 0) {
-				LOGCHAN(i, DDEV, ERROR) << rx_buffers[i]->str_code(rc);
+				LOGCHAN(i, DDEV, ERROR) << rx_buffers[i]->str_code(rc) << " num smpls: " << num_smpls << " chan: " << i;
 				LOGCHAN(i, DDEV, ERROR) << rx_buffers[i]->str_status(timestamp);
 				if (rc != smpl_buf::ERROR_OVERFLOW)
 					return 0;
diff --git a/Transceiver52M/device/ipc/IPCDevice.h b/Transceiver52M/device/ipc/IPCDevice.h
index 6bdb292..ed75d92 100644
--- a/Transceiver52M/device/ipc/IPCDevice.h
+++ b/Transceiver52M/device/ipc/IPCDevice.h
@@ -44,13 +44,17 @@ struct ipc_sock_state {
 	struct osmo_timer_list timer; /* socket connect retry timer */
 	struct llist_head upqueue; /* queue for sending messages */
 };
+#define IPC_MAX_NUM_TRX 8
+
+
 
 /** A class to handle a LimeSuite supported device */
 class IPCDevice : public RadioDevice {
     protected:
 	struct ipc_sock_state sk_state;
-	/* FIXME: current limit 8 chans, make dynamic */
-	struct ipc_sock_state sk_chan_state[8];
+	/* FIXME: current limit IPC_MAX_NUM_TRX chans, make dynamic */
+	struct ipc_sock_state sk_chan_state[IPC_MAX_NUM_TRX];
+	bool trx_is_started[IPC_MAX_NUM_TRX];
 	uint8_t tmp_state;
 	char shm_name[SHM_NAME_MAX];
 	int ipc_shm_connect(const char *shm_name);
diff --git a/Transceiver52M/device/ipc/ipc-driver-test.c b/Transceiver52M/device/ipc/ipc-driver-test.c
index a7300bf..93878fd 100644
--- a/Transceiver52M/device/ipc/ipc-driver-test.c
+++ b/Transceiver52M/device/ipc/ipc-driver-test.c
@@ -232,7 +232,7 @@ int ipc_rx_open_req(struct ipc_sk_if_open_req *open_req)
 	/* Here we verify num_chans, rx_path, tx_path, clockref, etc. */
 	int rc = ipc_shm_setup(DEFAULT_SHM_NAME, len);
 	len = ipc_shm_encode_region((struct ipc_shm_raw_region *)shm, open_req->num_chans, 4, shmbuflen);
-	LOGP(DMAIN, LOGL_NOTICE, "%s\n", osmo_hexdump((const unsigned char *)shm, 80));
+//	LOGP(DMAIN, LOGL_NOTICE, "%s\n", osmo_hexdump((const unsigned char *)shm, 80));
 
 	/* set up our own copy of the decoded area, we have to do it here,
 	 *  since the uhd wrapper does not allow starting single channels
@@ -249,10 +249,13 @@ int ipc_rx_open_req(struct ipc_sk_if_open_req *open_req)
 	return 0;
 }
 
+volatile int ul_running = 0;
+volatile int dl_running = 0;
+
 void *uplink_thread(void *x_void_ptr)
 {
 	uint32_t chann = decoded_region->num_chans;
-
+	ul_running = 1;
 	pthread_setname_np(pthread_self(), "uplink rx");
 
 	while (!ipc_exit_requested) {
@@ -266,6 +269,7 @@ void *uplink_thread(void *x_void_ptr)
 void *downlink_thread(void *x_void_ptr)
 {
 	int chann = decoded_region->num_chans;
+	dl_running = 1;
 	pthread_setname_np(pthread_self(), "downlink tx");
 
 	while (!ipc_exit_requested) {
@@ -279,14 +283,21 @@ int ipc_rx_chan_start_req(struct ipc_sk_chan_if_op_void *req, uint8_t chan_nr)
 {
 	struct msgb *msg;
 	struct ipc_sk_chan_if *ipc_prim;
-	int rc;
+	int rc = 0;
 
 	/* no per-chan start/stop */
-	rc = uhdwrap_start(global_dev, chan_nr);
-
-	pthread_t rx, tx;
-	pthread_create(&rx, NULL, uplink_thread, 0);
-	pthread_create(&tx, NULL, downlink_thread, 0);
+	if(!dl_running || !ul_running) {
+		rc = uhdwrap_start(global_dev, chan_nr);
+
+		/* chan != first chan start will "fail", which is fine, usrp can't start/stop chans independently */
+		if(rc) {
+			LOGP(DMAIN, LOGL_INFO, "starting rx/tx threads.. req for chan:%d\n", chan_nr);
+			pthread_t rx, tx;
+			pthread_create(&rx, NULL, uplink_thread, 0);
+			pthread_create(&tx, NULL, downlink_thread, 0);
+		}
+	} else
+		LOGP(DMAIN, LOGL_INFO, "starting rx/tx threads request ignored.. req for chan:%d\n", chan_nr);
 
 	msg = ipc_msgb_alloc(IPC_IF_MSG_START_CNF);
 	if (!msg)
@@ -462,6 +473,11 @@ int main(int argc, char **argv)
 	ipc_sock_init(ipc_msock_path, &global_ipc_sock_state, ipc_sock_accept, 0);
 	while (!ipc_exit_requested)
 		osmo_select_main(0);
+
+	if (global_dev)
+		for (unsigned int i = 0; i < decoded_region->num_chans; i++)
+			uhdwrap_stop(global_dev, i);
+
 	//ipc_sock_close()
 	return 0;
 }
diff --git a/Transceiver52M/device/ipc/ipc_chan.c b/Transceiver52M/device/ipc/ipc_chan.c
index 91aa52d..949cda9 100644
--- a/Transceiver52M/device/ipc/ipc_chan.c
+++ b/Transceiver52M/device/ipc/ipc_chan.c
@@ -99,6 +99,7 @@ static int ipc_chan_sock_read(struct osmo_fd *bfd)
 		return 0;
 	}
 
+	LOGP(DMAIN, LOGL_INFO, "rx on bf priv: %d\n", bfd->priv_nr);
 	rc = ipc_chan_rx(ipc_prim->msg_type, ipc_prim, bfd->priv_nr);
 
 	/* as we always synchronously process the message in IPC_rx() and
@@ -236,6 +237,9 @@ int ipc_chan_sock_accept(struct osmo_fd *bfd, unsigned int flags)
 	conn_bfd->cb = ipc_chan_sock_cb;
 	conn_bfd->data = state;
 
+	/* copy chan nr, required for proper bfd<->chan # mapping */
+	conn_bfd->priv_nr = bfd->priv_nr;
+
 	if (osmo_fd_register(conn_bfd) != 0) {
 		LOGP(DMAIN, LOGL_ERROR,
 		     "Failed to register new connection "
diff --git a/Transceiver52M/device/ipc/uhdwrap.cpp b/Transceiver52M/device/ipc/uhdwrap.cpp
index 1175bd1..4b796f5 100644
--- a/Transceiver52M/device/ipc/uhdwrap.cpp
+++ b/Transceiver52M/device/ipc/uhdwrap.cpp
@@ -106,6 +106,8 @@ extern "C" void *uhdwrap_open(struct ipc_sk_if_open_req *open_req)
 		rx_paths.push_back(open_req->chan_info[i].rx_path);
 	}
 
+	/* FIXME: this is actually the sps value, not the sample rate!
+	 * sample rate is looked up according to the sps rate by uhd backend */
 	rx_sps = open_req->rx_sample_freq_num / open_req->rx_sample_freq_den;
 	tx_sps = open_req->tx_sample_freq_num / open_req->tx_sample_freq_den;
 	uhd_wrap *uhd_wrap_dev =
@@ -136,7 +138,9 @@ extern "C" int32_t uhdwrap_read(void *dev, uint32_t num_chans)
 	}
 
 	int32_t read = d->wrap_read(&t);
-	if (read < 0)
+
+	/* multi channel rx on b210 will return 0 due to alignment adventures, do not put 0 samples into a ipc buffer... */
+	if (read <= 0)
 		return read;
 
 	for (uint32_t i = 0; i < num_chans; i++) {