lib: initial support of yield/resume in vtysh

Support YIELD operation for long-running cli operations. A new
api allows a daemon to supply a callback and context pointer;
the vty lib will reschedule and call through to the daemon
allowing it to resume a long-running show operation. The vty
lib avoids new reads until the daemon is done.

Signed-off-by: Mark Stapp <mjs@cisco.com>

Author: Mark Stapp

lib/command.h

@@ -246,6 +246,7 @@ struct cmd_node {
 #define CMD_NOT_MY_INSTANCE	14
 #define CMD_NO_LEVEL_UP 15
 #define CMD_ERR_NO_DAEMON 16
+#define CMD_YIELD 17 /* Command has yielded but is not complete yet */
 
 /* Argc max counts. */
 #define CMD_ARGC_MAX   256

lib/vty.c

@@ -83,6 +83,9 @@ int (*nb_cli_apply_changes_mgmt_cb)(struct vty *vty, const char *xpath_base_abs)
 int (*nb_cli_rpc_mgmt_cb)(struct vty *vty, const char *xpath, const struct lyd_node *input);
 
 
+/* Master of the tasks. */
+static struct event_loop *vty_master;
+
 PREDECL_DLIST(vtyservs);
 
 struct vty_serv {
@@ -139,22 +142,27 @@ void vty_resume_response(struct vty *vty, int ret)
 	uint8_t header[4] = {0, 0, 0, 0};
 
 	if (vty->type != VTY_FILE) {
+		/* Send end-of-output marker */
 		header[3] = ret;
 		buffer_put(vty->obuf, header, 4);
-		if (!vty->t_write && (vtysh_flush(vty) < 0)) {
-			zlog_err("failed to vtysh_flush");
-			/* Try to flush results; exit if a write error occurs */
-			return;
+
+		/* Try to flush results; exit if a write error occurs */
+		if (vty->t_write == NULL) {
+			if (vtysh_flush(vty) < 0) {
+				zlog_err("failed to vtysh_flush");
+				return;
+			}
 		}
 	}
 
+	/* Resume reading if possible */
 	if (vty->status == VTY_CLOSE)
 		vty_close(vty);
 	else if (vty->type != VTY_FILE)
 		vty_event(VTYSH_READ, vty);
 	else
 		/* should we assert here? */
-		zlog_err("mgmtd: unexpected resume while reading config file");
+		zlog_err("vty: unexpected resume while reading config file");
 }
 
 void vty_frame(struct vty *vty, const char *format, ...)
@@ -175,6 +183,86 @@ void vty_endframe(struct vty *vty, const char *endtext)
 	vty->frame_pos = 0;
 }
 
+/*
+ * Callback function for yield/resume; call through to application's
+ * callback.
+ */
+static void yield_resume_cb(struct event *event)
+{
+	struct vty *vty = EVENT_ARG(event);
+	struct vty_yield_resume_s ctx;
+
+	/* Capture application callback info */
+	ctx = vty->yield_resume;
+
+	/* Clear vty's yield callback info before calling into the application:
+	 * the application may call the "finish" or the "yield" api, so we can't
+	 * make any assumptions about the state of the 'vty' info after calling in.
+	 */
+	vty->yield_resume.app_cb = NULL;
+	vty->yield_resume.arg = NULL;
+
+	/* Call through to application */
+	ctx.app_cb(vty, ctx.arg);
+}
+
+/*
+ * Application process wants to yield while sending results/replies.
+ * We'll schedule a task to resume, and call the application's callback.
+ */
+bool vty_yield(struct vty *vty, void (*func)(struct vty *vty, void *arg),
+	       void *arg)
+{
+	vty->yield_resume.app_cb = func;
+	vty->yield_resume.arg = arg;
+
+	event_add_event(vty_master, yield_resume_cb, vty, 0,
+			&vty->yield_resume.t_resume);
+
+	/* Ensure reading new input is disabled */
+	event_cancel(&vty->t_read);
+
+	/* Try to send buffered output */
+	if (vty->type == VTY_SHELL_SERV)
+		vtysh_flush(vty);
+	else
+		vty_event(VTY_WRITE, vty);
+
+	return true;
+}
+
+/*
+ *
+ */
+static void yield_finish_internal(struct vty *vty)
+{
+	event_cancel(&(vty->yield_resume.t_resume));
+
+	/* Clear vty's yield/resume callback info */
+	vty->yield_resume.app_cb = NULL;
+	vty->yield_resume.arg = NULL;
+}
+
+/*
+ * Yield/resume is complete; cancel any scheduled resume task, and return
+ * to normal vty operation (turn on reads, e.g.)
+ */
+void vty_yield_finish(struct vty *vty, int retcode)
+{
+	if (vty->status != VTY_CLOSE) {
+		if (vty->type == VTY_SHELL_SERV) {
+
+			/* Send end-of-output marker and resume normal processing */
+			vty_resume_response(vty, retcode);
+			vty_event(VTYSH_READ, vty);
+		} else {
+			vty_event(VTY_READ, vty);
+		}
+
+		yield_finish_internal(vty);
+	}
+}
+
 bool vty_set_include(struct vty *vty, const char *regexp)
 {
 	int errcode;
@@ -1618,7 +1706,7 @@ static void vty_flush(struct event *event)
 	buffer_status_t flushrc;
 	struct vty *vty = EVENT_ARG(event);
 
-	/* Tempolary disable read event. */
+	/* Temporarily disable reads. */
 	if (vty->lines == 0)
 		event_cancel(&vty->t_read);
 
@@ -2300,10 +2388,11 @@ static void vtysh_read(struct event *event)
 			if (*p == '\0') {
 				/* Pass this line to parser. */
 				ret = vty_execute(vty);
-/* Note that vty_execute clears the command buffer and resets
-   vty->length to 0. */
+				/* Note that vty_execute clears the command buffer and
+				 * resets vty->length to 0.
+				 */
 
-/* Return result. */
+				/* Return result. */
 #ifdef VTYSH_DEBUG
 				printf("result: %d\n", ret);
 				printf("vtysh node: %d\n", vty->node);
@@ -2352,6 +2441,12 @@ static void vtysh_read(struct event *event)
 					return;
 				}
 
+				/* If the application has asked to yield during
+				 * processing, don't continue reading.
+				 */
+				if (event_is_scheduled(vty->yield_resume.t_resume))
+					break;
+
 				/* warning: watchfrr hardcodes this result write
 				 */
 				header[3] = ret;
@@ -2369,6 +2464,12 @@ static void vtysh_read(struct event *event)
 		}
 	}
 
+	/* If the application has asked to yield during
+	 * processing, don't continue reading.
+	 */
+	if (event_is_scheduled(vty->yield_resume.t_resume))
+		return;
+
 	if (vty->status == VTY_CLOSE)
 		vty_close(vty);
 	else
@@ -2428,6 +2529,14 @@ void vty_close(struct vty *vty)
 
 	vty->status = VTY_CLOSE;
 
+	/* If application was doing yield/resume, notify it by calling
+	 * its callback with a NULL vty argument.
+	 */
+	if (vty->yield_resume.app_cb) {
+		(vty->yield_resume.app_cb)(NULL, vty->yield_resume.arg);
+		yield_finish_internal(vty);
+	}
+
 	/*
 	 * If we reach here with pending config to commit we will be losing it
 	 * so warn the user.
@@ -2753,6 +2862,7 @@ FILE *vty_open_config(const char *config_file, char *config_default_dir)
 		}
 #endif /* VTYSH */
 		confp = fopen(config_default_dir, "r");
+
 		if (confp == NULL) {
 			flog_err(
 				EC_LIB_SYSTEM_CALL,
@@ -2911,9 +3021,6 @@ int vty_config_node_exit(struct vty *vty)
 	return 1;
 }
 
-/* Master of the threads. */
-static struct event_loop *vty_master;
-
 static void vty_event_serv(enum vty_event event, struct vty_serv *vty_serv)
 {
 	switch (event) {

lib/vty.h

@@ -54,6 +54,22 @@ enum vty_status {
 
 PREDECL_DLIST(vtys);
 
+/* Data for yield/resume of large show outputs. Application handlers can
+ * call the yield api before returning; this will schedule the resume callback.
+ * When the handler returns, the vty library code will try to avoid reading
+ * any new input.
+ * When the application is done with the yield/resume cycles, it calls the finish
+ * api, and resumes normal processing.
+ */
+struct vty_yield_resume_s {
+	/* Application callback and argument */
+	void *arg;
+	void (*app_cb)(struct vty *vty, void *arg);
+
+	/* Event to schedule the resume callback */
+	struct event *t_resume;
+};
+
 /* VTY struct. */
 struct vty {
 	struct vtys_item itm;
@@ -231,13 +247,17 @@ struct vty {
 	uintptr_t mgmt_req_pending_data;
 	bool mgmt_locked_candidate_ds;
 	bool mgmt_locked_running_ds;
+
 	/* Incremental write/flush limit and current accumulator. When
 	 * producing large outputs, we try to avoid buffering the entire
 	 * output, sending incremental output periodically
 	 * as the application code produces it.
 	 */
 	size_t vty_buf_threshold;
 	size_t vty_buf_size_accum;
+
+	/* Data for yield/resume of large show outputs */
+	struct vty_yield_resume_s yield_resume;
 };
 
 static inline void vty_push_context(struct vty *vty, int node, uint64_t id)
@@ -449,6 +469,25 @@ extern void (*vty_config_node_exit_mgmt_cb)(struct vty *vty);
 extern int (*nb_cli_apply_changes_mgmt_cb)(struct vty *vty, const char *xpath_base_abs);
 extern int (*nb_cli_rpc_mgmt_cb)(struct vty *vty, const char *xpath, const struct lyd_node *input);
 
+/*
+ * Application process wants to yield while sending results/replies.
+ * We'll schedule a task to resume, and call the application's callback
+ * with the 'vty' and its 'arg'. We won't restart reads on the vty until
+ * the application tells us to.
+ * If the vty is being closed or deleted, we'll call the callback with a NULL
+ * 'vty', so the application can clean up, free memory, if necessary.
+ */
+bool vty_yield(struct vty *vty, void (*func)(struct vty *vty, void *arg),
+	       void *arg);
+
+/*
+ * Yield/resume is complete; cancel any scheduled resume task, and return
+ * to normal vty operation (turn on reads, e.g.). For clients of vtysh,
+ * send 'retcode' to signal that output is complete.
+ * Called from the application, so we expect the application to have
+ * cleaned-up any context, memory, etc.
+ */
+void vty_yield_finish(struct vty *vty, int retcode);
 
 #ifdef __cplusplus
 }