peripherals: Create and use flow context

Introduce a flow control context per peripheral and pass this to
dm_broadcast() for data channels in order for any queued messages to be
accounted for in the flow control context.

Associate the same flow control context with the read watcher for the
peripheral's incoming data descriptors to make the watch runner
selectively read from the given peripheral only when the number of
outstanding, queued, messages are below the threshold.

Signed-off-by: Bjorn Andersson <bjorn.andersson@linaro.org>

router/diag.h

@@ -83,6 +83,8 @@ struct peripheral {
 	int cmd_fd;
 	int dci_cmd_fd;
 
+	struct watch_flow *flow;
+
 	int diag_id;
 
 	bool sockets;

router/peripheral-qrtr.c

@@ -243,7 +243,7 @@ static int qrtr_data_recv(int fd, void *data)
 			fprintf(stderr, "non-HDLC frame is not truncated\n");
 			break;
 		}
-		dm_broadcast(frame->payload, frame->length, NULL);
+		dm_broadcast(frame->payload, frame->length, perif->flow);
 		break;
 	case QRTR_TYPE_BYE:
 		watch_remove_writeq(perif->data_fd);
@@ -275,13 +275,17 @@ void qrtr_perif_close(struct peripheral *perif)
 static int qrtr_perif_init_subsystem(const char *name, int instance_base)
 {
 	struct peripheral *perif;
+	struct watch_flow *flow;
 
 	perif = calloc(1, sizeof(*perif));
 
+	flow = watch_flow_new();
+
 	perif->name = strdup(name);
 	perif->send = qrtr_perif_send;
 	perif->close = qrtr_perif_close;
 	perif->sockets = true;
+	perif->flow = flow;
 
 	list_init(&perif->cmdq);
 	list_init(&perif->cntlq);
@@ -316,7 +320,7 @@ static int qrtr_perif_init_subsystem(const char *name, int instance_base)
 
 	watch_add_readfd(perif->cntl_fd, qrtr_cntl_recv, perif, NULL);
 	watch_add_readfd(perif->cmd_fd, qrtr_cmd_recv, perif, NULL);
-	watch_add_readfd(perif->data_fd, qrtr_data_recv, perif, NULL);
+	watch_add_readfd(perif->data_fd, qrtr_data_recv, perif, flow);
 	list_add(&peripherals, &perif->node);
 	return 0;
 }

router/peripheral-rpmsg.c

@@ -127,7 +127,7 @@ static int diag_data_recv_hdlc(int fd, struct peripheral *peripheral)
 			if (!msg)
 				break;
 
-			dm_broadcast(msg, msglen, NULL);
+			dm_broadcast(msg, msglen, peripheral->flow);
 		}
 	}
 
@@ -144,7 +144,7 @@ static int diag_data_recv_raw(int fd, struct peripheral *peripheral)
 		if (n < 0)
 			return -errno;
 
-		dm_broadcast(buf, n, NULL);
+		dm_broadcast(buf, n, peripheral->flow);
 	}
 
 	/* Not reached */
@@ -326,7 +326,7 @@ static void peripheral_open(void *data)
 	watch_add_writeq(peripheral->cntl_fd, &peripheral->cntlq);
 	watch_add_writeq(peripheral->data_fd, &peripheral->dataq);
 	watch_add_readfd(peripheral->cntl_fd, rpmsg_perif_cntl_recv, peripheral, NULL);
-	watch_add_readfd(peripheral->data_fd, diag_data_recv, peripheral, NULL);
+	watch_add_readfd(peripheral->data_fd, diag_data_recv, peripheral, peripheral->flow);
 	if (peripheral->cmd_fd >= 0) {
 		watch_add_readfd(peripheral->cmd_fd, diag_cmd_recv, peripheral, NULL);
 		watch_add_writeq(peripheral->cmd_fd, &peripheral->cmdq);
@@ -356,6 +356,7 @@ static void perif_rpmsg_close(struct peripheral *peripheral)
 static int peripheral_create(const char *rproc, const char *channel)
 {
 	struct peripheral *peripheral;
+	struct watch_flow *flow;
 	struct list_head *item;
 
 	/* Only trigger the creation of a peripheral on primary channels */
@@ -371,12 +372,15 @@ static int peripheral_create(const char *rproc, const char *channel)
 	peripheral = malloc(sizeof(*peripheral));
 	memset(peripheral, 0, sizeof(*peripheral));
 
+	flow = watch_flow_new();
+
 	peripheral->name = strdup(rproc);
 	peripheral->data_fd = -1;
 	peripheral->cntl_fd = -1;
 	peripheral->cmd_fd = -1;
 	peripheral->send = perif_rpmsg_send;
 	peripheral->close = perif_rpmsg_close;
+	peripheral->flow = flow;
 	list_init(&peripheral->cmdq);
 	list_init(&peripheral->cntlq);
 	list_init(&peripheral->dataq);