lib: Memory spike reduction for sh cmds at scale

The output buffer vty->obuf is a linked list where
each element is of 4KB.
Currently, when a huge sh command  like “show ip route json”
is executed on a large scale, all the vty_out’s are
processed and the entire data is accumulated.
After the entire vty execution, vtysh_flush proceeses
and puts this data in the socket (131KB at a time).

Problem here is the memory spike for such heavy duty
show commands.

The fix here is to chunkify the output on VTY shell by
flushing it intermediately for every 128 KB of output
accumulated and free the memory allocated for the buffer data.

This way, we achieve ~25-30% reduction in the memory spike.

Signed-off-by: Srujana <[email protected]>
Signed-off-by: Rajasekar Raja <[email protected]>
Signed-off-by: Srujana <[email protected]>

lib/buffer.c

@@ -43,9 +43,6 @@ struct buffer_data {
 
 /* It should always be true that: 0 <= sp <= cp <= size */
 
-/* Default buffer size (used if none specified).  It is rounded up to the
-   next page boundary. */
-#define BUFFER_SIZE_DEFAULT		4096
 
 #define BUFFER_DATA_FREE(D) XFREE(MTYPE_BUFFER_DATA, (D))
 

lib/buffer.h

@@ -11,9 +11,15 @@
 extern "C" {
 #endif
 
-/* Create a new buffer.  Memory will be allocated in chunks of the given
-   size.  If the argument is 0, the library will supply a reasonable
-   default size suitable for buffering socket I/O. */
+/* Default buffer size (used if none specified).  It is rounded up to the
+ * next page boundary.
+ */
+#define BUFFER_SIZE_DEFAULT 4096
+
+/* Create a new buffer. Memory will be allocated in chunks of the given
+ * size.  If the argument is 0, the library will supply a reasonable
+ * default size suitable for buffering socket I/O.
+ */
 extern struct buffer *buffer_new(size_t size);
 
 /* Free all data in the buffer. */

lib/vty.c

@@ -67,7 +67,8 @@ enum vty_event {
 #ifdef VTYSH
 	VTYSH_SERV,
 	VTYSH_READ,
-	VTYSH_WRITE
+	VTYSH_WRITE,
+	VTYSH_WRITE_INTERMEDIATE
 #endif /* VTYSH */
 };
 
@@ -94,7 +95,7 @@ DECLARE_DLIST(vtyservs, struct vty_serv, itm);
 static void vty_event_serv(enum vty_event event, struct vty_serv *);
 static void vty_event(enum vty_event, struct vty *);
 static int vtysh_flush(struct vty *vty);
-
+static void vtysh_flush_intermediate(struct vty *vty);
 /* Extern host structure from command.c */
 extern struct host host;
 
@@ -345,8 +346,17 @@ int vty_out(struct vty *vty, const char *format, ...)
 	case VTY_SHELL_SERV:
 	case VTY_FILE:
 	default:
+		vty->vty_buf_size_written += strlen(filtered);
 		/* print without crlf replacement */
 		buffer_put(vty->obuf, (uint8_t *)filtered, strlen(filtered));
+		/* For every 32 *4KB invoke vtysh_flush_intermediate where we
+		 * put the date of collective vty->obuf Linked List items on the
+		 * socket and free the vty->obuf data. This way, we are flushing
+		 * data every 128KB for the show command.
+		 */
+		if (vty->vty_buf_size_written >= 32 * BUFFER_SIZE_DEFAULT)
+			vtysh_flush_intermediate(vty);
+
 		break;
 	}
 
@@ -1833,6 +1843,7 @@ void vty_stdio_suspend(void)
 
 	EVENT_OFF(stdio_vty->t_write);
 	EVENT_OFF(stdio_vty->t_read);
+	EVENT_OFF(stdio_vty->t_write_intermediate);
 	EVENT_OFF(stdio_vty->t_timeout);
 
 	if (stdio_termios)
@@ -2141,6 +2152,23 @@ static void vtysh_accept(struct event *thread)
 		close(sock);
 		return;
 	}
+	/* Setting the SND buffer size to 16MB to increase the buffer size
+	 * the socket can hold before sending it to VTY shell
+	 */
+	uint32_t sndbufsize = BUFFER_SIZE_DEFAULT * BUFFER_SIZE_DEFAULT;
+	int ret;
+
+	ret = setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&sndbufsize,
+			 sizeof(sndbufsize));
+	if (ret < 0) {
+		flog_err(EC_LIB_SOCKET,
+			 "Cannot set socket %d send buffer size, %s", sock,
+			 safe_strerror(errno));
+		close(sock);
+		return;
+	}
+
+
 	set_cloexec(sock);
 
 #ifdef VTYSH_DEBUG
@@ -2207,6 +2235,34 @@ static int vtysh_do_pass_fd(struct vty *vty)
 	return BUFFER_EMPTY;
 }
 
+/* This function is similar to vtysh_flush exceot this is used
+ * to chunkify the vty_outs intermediately when the buffer size
+ * is 128KB or more (since only 131 KB is flushed at once).
+ *
+ * When a BUFFER_EMPTY is received, we clean up the buffer data
+ * When a BUFFER_PENDING is received, we schedule to call the same
+ * function again to process the data in the pending buffer
+ */
+static void vtysh_flush_intermediate(struct vty *vty)
+{
+	int ret;
+
+	ret = buffer_flush_available(vty->obuf, vty->wfd);
+	switch (ret) {
+	case BUFFER_EMPTY:
+		vty->vty_buf_size_written = 0;
+		buffer_reset(vty->lbuf);
+		buffer_reset(vty->obuf);
+		break;
+	case BUFFER_PENDING:
+		vty_event(VTYSH_WRITE_INTERMEDIATE, vty);
+		break;
+	case BUFFER_ERROR:
+		flog_err(EC_LIB_SOCKET, "%s: Write intermediate error on fd %d",
+			 __func__, vty->fd);
+		break;
+	}
+}
 static int vtysh_flush(struct vty *vty)
 {
 	int ret;
@@ -2443,6 +2499,12 @@ static void vtysh_read(struct event *thread)
 		vty_event(VTYSH_READ, vty);
 }
 
+static void vtysh_write_intermediate(struct event *thread)
+{
+	struct vty *vty = EVENT_ARG(thread);
+
+	vtysh_flush_intermediate(vty);
+}
 static void vtysh_write(struct event *thread)
 {
 	struct vty *vty = EVENT_ARG(thread);
@@ -2517,6 +2579,7 @@ void vty_close(struct vty *vty)
 	/* Cancel threads.*/
 	EVENT_OFF(vty->t_read);
 	EVENT_OFF(vty->t_write);
+	EVENT_OFF(vty->t_write_intermediate);
 	EVENT_OFF(vty->t_timeout);
 
 	if (vty->pass_fd != -1) {
@@ -3029,6 +3092,7 @@ static void vty_event_serv(enum vty_event event, struct vty_serv *vty_serv)
 	case VTY_TIMEOUT_RESET:
 	case VTYSH_READ:
 	case VTYSH_WRITE:
+	case VTYSH_WRITE_INTERMEDIATE:
 		assert(!"vty_event_serv() called incorrectly");
 	}
 }
@@ -3045,6 +3109,10 @@ static void vty_event(enum vty_event event, struct vty *vty)
 		event_add_write(vty_master, vtysh_write, vty, vty->wfd,
 				&vty->t_write);
 		break;
+	case VTYSH_WRITE_INTERMEDIATE:
+		event_add_write(vty_master, vtysh_write_intermediate, vty,
+				vty->wfd, &vty->t_write_intermediate);
+		break;
 #endif /* VTYSH */
 	case VTY_READ:
 		event_add_read(vty_master, vty_read, vty, vty->fd,

lib/vty.h

@@ -212,6 +212,7 @@ struct vty {
 	/* Read and write thread. */
 	struct event *t_read;
 	struct event *t_write;
+	struct event *t_write_intermediate;
 
 	/* Timeout seconds and thread. */
 	unsigned long v_timeout;
@@ -236,6 +237,7 @@ struct vty {
 	uintptr_t mgmt_req_pending_data;
 	bool mgmt_locked_candidate_ds;
 	bool mgmt_locked_running_ds;
+	uint64_t vty_buf_size_written;
 };
 
 static inline void vty_push_context(struct vty *vty, int node, uint64_t id)

vtysh/vtysh.c

@@ -4721,6 +4721,24 @@ static int vtysh_connect(struct vtysh_client *vclient)
 		close(sock);
 		return -1;
 	}
+
+	/* Setting the RCV buffer size to 16MB to increase the buffer size
+	 * the socket can hold after receving from other process
+	 */
+	uint32_t rcvbufsize = BUFFER_SIZE_DEFAULT * BUFFER_SIZE_DEFAULT;
+
+	ret = setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&rcvbufsize,
+			 sizeof(rcvbufsize));
+	if (ret < 0) {
+#ifdef DEBUG
+		fprintf(stderr,
+			"Cannot set socket %d rcv buffer size, %s\n",
+			sock, safe_strerror(errno));
+#endif /* DEBUG */
+		close(sock);
+		return -1;
+	}
+
 	vclient->fd = sock;
 
 	return 0;