--- a/src/doveadm/doveadm-stats.c	Thu Sep 01 10:37:55 2011 +0300
+++ b/src/doveadm/doveadm-stats.c	Thu Sep 01 10:39:19 2011 +0300
@@ -1,18 +1,56 @@
 /* Copyright (c) 2011 Dovecot authors, see the included COPYING file */
 
 #include "lib.h"
+#include "array.h"
 #include "network.h"
+#include "ioloop.h"
 #include "istream.h"
+#include "hash.h"
 #include "str.h"
 #include "strescape.h"
 #include "write-full.h"
 #include "doveadm.h"
 #include "doveadm-print.h"
 
+#include <stdio.h>
+#include <stdlib.h>
 #include <unistd.h>
+#include <sys/ioctl.h>
+#include <termios.h>
+
+struct top_line {
+	char *id;
+	/* [headers_count] */
+	const char **prev_values, **cur_values;
+
+	unsigned int flip:1;
+};
+
+struct top_context {
+	const char *path;
+	int fd;
+	struct istream *input;
+	const char *sort_type;
 
-static const char *const*
-read_next_line(struct istream *input)
+	char **headers;
+	unsigned int headers_count;
+
+	pool_t prev_pool, cur_pool;
+	/* id => struct top_line. */
+	struct hash_table *sessions;
+	ARRAY_DEFINE(lines, struct top_line *);
+	int (*lines_sort)(struct top_line *const *, struct top_line *const *);
+
+	unsigned int last_update_idx, user_idx;
+	unsigned int sort_idx1, sort_idx2;
+
+	unsigned int flip:1;
+};
+
+static struct top_context *sort_ctx = NULL;
+
+static char **
+p_read_next_line(pool_t pool, struct istream *input)
 {
 	const char *line;
 	char **args;
@@ -22,10 +60,16 @@
 	if (line == NULL)
 		return NULL;
 
-	args = p_strsplit(pool_datastack_create(), line, "\t");
+	args = p_strsplit(pool, line, "\t");
 	for (i = 0; args[i] != NULL; i++)
 		args[i] = str_tabunescape(args[i]);
-	return (void *)args;
+	return args;
+}
+
+static const char *const*
+read_next_line(struct istream *input)
+{
+	return (const void *)p_read_next_line(pool_datastack_create(), input);
 }
 
 static void stats_dump(const char *path, const char *cmd)
@@ -83,12 +127,12 @@
 			path = optarg;
 			break;
 		default:
-			help(&doveadm_cmd_stats);
+			help(&doveadm_cmd_stats_dump);
 		}
 	}
 	argv += optind - 1;
 	if (argv[1] == NULL)
-		help(&doveadm_cmd_stats);
+		help(&doveadm_cmd_stats_dump);
 	cmd = t_strdup_printf("EXPORT\t%s\n",
 			      t_strarray_join((const void *)(argv+1), "\t"));
 
@@ -96,6 +140,390 @@
 	stats_dump(path, cmd);
 }
 
-struct doveadm_cmd doveadm_cmd_stats = {
+static void
+stats_line_set_prev_values(struct top_context *ctx,
+			   const struct top_line *old_line,
+			   struct top_line *line)
+{
+	const char **values;
+	unsigned int i;
+
+	if (old_line->prev_values == NULL ||
+	    strcmp(old_line->cur_values[ctx->last_update_idx],
+		   line->cur_values[ctx->last_update_idx]) != 0) {
+		values = old_line->cur_values;
+	} else {
+		values = old_line->prev_values;
+	}
+
+	/* line hasn't been updated, keep old values */
+	line->prev_values =
+		p_new(ctx->cur_pool, const char *, ctx->headers_count);
+	for (i = 0; i < ctx->headers_count; i++)
+		line->prev_values[i] = p_strdup(ctx->cur_pool, values[i]);
+}
+
+static void stats_read(struct top_context *ctx)
+{
+	struct top_line *old_line, *line;
+	unsigned int i;
+	char **args;
+
+	/* read lines */
+	while ((args = p_read_next_line(ctx->cur_pool, ctx->input)) != NULL) {
+		if (args[0] == NULL) {
+			/* end of stats */
+			return;
+		}
+		if (str_array_length((void *)args) != ctx->headers_count)
+			i_fatal("read(%s): invalid stats line", ctx->path);
+
+		line = p_new(ctx->cur_pool, struct top_line, 1);
+		line->id = args[0];
+		line->flip = ctx->flip;
+		line->cur_values = p_new(ctx->cur_pool, const char *, ctx->headers_count);
+		for (i = 0; i < ctx->headers_count; i++)
+			line->cur_values[i] = args[i];
+
+		old_line = hash_table_lookup(ctx->sessions, line->id);
+		if (old_line != NULL) {
+			stats_line_set_prev_values(ctx, old_line, line);
+			array_append(&ctx->lines, &line, 1);
+		}
+		hash_table_insert(ctx->sessions, line->id, line);
+	}
+
+	if (ctx->input->stream_errno != 0)
+		i_fatal("read(%s) failed: %m", ctx->path);
+	i_fatal("read(%s): unexpected EOF", ctx->path);
+}
+
+static void stats_drop_stale(struct top_context *ctx)
+{
+	struct hash_iterate_context *iter;
+	void *key, *value;
+
+	iter = hash_table_iterate_init(ctx->sessions);
+	while (hash_table_iterate(iter, &key, &value)) {
+		struct top_line *line = value;
+
+		if (line->flip != ctx->flip)
+			hash_table_remove(ctx->sessions, key);
+	}
+	hash_table_iterate_deinit(&iter);
+}
+
+static int get_double(const char *str, double *num_r)
+{
+	char *p;
+
+	*num_r = strtod(str, &p);
+	return *p == '\0' ? 0 : -1;
+}
+
+static double sort_cpu_diff(const struct top_line *line)
+{
+	double prev, cur, diff, prev_time, cur_time, time_multiplier;
+
+	if (get_double(line->prev_values[sort_ctx->last_update_idx], &prev_time) < 0 &&
+	    get_double(line->cur_values[sort_ctx->last_update_idx], &cur_time) < 0)
+		i_fatal("sorting: invalid last_update value");
+	time_multiplier = (cur_time - prev_time) * 100;
+
+	if (get_double(line->prev_values[sort_ctx->sort_idx1], &prev) < 0 ||
+	    get_double(line->cur_values[sort_ctx->sort_idx1], &cur) < 0)
+		i_fatal("sorting: not a double");
+
+	diff = cur - prev;
+
+	if (sort_ctx->sort_idx2 != 0) {
+		if (get_double(line->prev_values[sort_ctx->sort_idx2], &prev) < 0 ||
+		    get_double(line->cur_values[sort_ctx->sort_idx2], &cur) < 0)
+			i_fatal("sorting: not a double");
+		diff += cur - prev;
+	}
+	return diff * time_multiplier;
+}
+
+static int sort_cpu(struct top_line *const *l1, struct top_line *const *l2)
+{
+	double d1, d2;
+
+	d1 = sort_cpu_diff(*l1);
+	d2 = sort_cpu_diff(*l2);
+	if (d1 < d2)
+		return -1;
+	if (d1 > d2)
+		return 1;
+	return strcmp((*l1)->cur_values[sort_ctx->user_idx],
+		      (*l2)->cur_values[sort_ctx->user_idx]);
+}
+
+static double sort_num_diff(const struct top_line *line)
+{
+	uint64_t prev, cur, diff;
+
+	if (str_to_uint64(line->prev_values[sort_ctx->sort_idx1], &prev) < 0 ||
+	    str_to_uint64(line->cur_values[sort_ctx->sort_idx1], &cur) < 0)
+		i_fatal("sorting: not a number");
+	diff = cur - prev;
+
+	if (sort_ctx->sort_idx2 != 0) {
+		if (str_to_uint64(line->prev_values[sort_ctx->sort_idx2], &prev) < 0 ||
+		    str_to_uint64(line->cur_values[sort_ctx->sort_idx2], &cur) < 0)
+			i_fatal("sorting: not a number");
+		diff += cur - prev;
+	}
+	return diff;
+}
+
+static int sort_num(struct top_line *const *l1, struct top_line *const *l2)
+{
+	uint64_t n1, n2;
+
+	n1 = sort_num_diff(*l1);
+	n2 = sort_num_diff(*l2);
+	if (n1 < n2)
+		return -1;
+	if (n1 > n2)
+		return 1;
+	return strcmp((*l1)->cur_values[sort_ctx->user_idx],
+		      (*l2)->cur_values[sort_ctx->user_idx]);
+}
+
+static bool
+stats_header_find(struct top_context *ctx, const char *name,
+		  unsigned int *idx_r)
+{
+	unsigned int i;
+
+	for (i = 0; ctx->headers[i] != NULL; i++) {
+		if (strcmp(ctx->headers[i], name) == 0) {
+			*idx_r = i;
+			return TRUE;
+		}
+	}
+	return FALSE;
+}
+
+static void stats_top_get_sorting(struct top_context *ctx)
+{
+	if (stats_header_find(ctx, ctx->sort_type, &ctx->sort_idx1))
+		return;
+
+	if (strcmp(ctx->sort_type, "cpu") == 0) {
+		if (!stats_header_find(ctx, "user_cpu", &ctx->sort_idx1) ||
+		    !stats_header_find(ctx, "sys_cpu", &ctx->sort_idx2))
+			i_fatal("cpu sort type is missing fields");
+		return;
+	}
+	if (strcmp(ctx->sort_type, "disk") == 0) {
+		if (!stats_header_find(ctx, "disk_input", &ctx->sort_idx1) ||
+		    !stats_header_find(ctx, "disk_output", &ctx->sort_idx2))
+			i_fatal("disk sort type is missing fields");
+		return;
+	}
+	i_fatal("unknown sort type: %s", ctx->sort_type);
+}
+
+static bool stats_top_round(struct top_context *ctx)
+{
+#define TOP_CMD "EXPORT\tsession\tconnected\n"
+	const char *const *args;
+	pool_t tmp_pool;
+
+	if (write_full(ctx->fd, TOP_CMD, strlen(TOP_CMD)) < 0)
+		i_fatal("write(%s) failed: %m", ctx->path);
+
+	/* read header */
+	if (ctx->headers != NULL) {
+		args = read_next_line(ctx->input);
+		if (args == NULL)
+			i_fatal("read(%s) unexpectedly disconnected", ctx->path);
+		if (*args == NULL)
+			return TRUE;
+		if (str_array_length(args) != ctx->headers_count)
+			i_fatal("headers changed");
+	} else {
+		ctx->headers = p_read_next_line(default_pool, ctx->input);
+		if (ctx->headers == NULL)
+			i_fatal("read(%s) unexpectedly disconnected", ctx->path);
+		if (*ctx->headers == NULL) {
+			i_free_and_null(ctx->headers);
+			return FALSE;
+		}
+		ctx->headers_count = str_array_length((void *)ctx->headers);
+		if (!stats_header_find(ctx, "last_update", &ctx->last_update_idx))
+			i_fatal("last_update header missing");
+		if (!stats_header_find(ctx, "user", &ctx->user_idx))
+			i_fatal("user header missing");
+		stats_top_get_sorting(ctx);
+	}
+
+	array_clear(&ctx->lines);
+	p_clear(ctx->prev_pool);
+	tmp_pool = ctx->prev_pool;
+	ctx->prev_pool = ctx->cur_pool;
+	ctx->cur_pool = tmp_pool;
+
+	ctx->flip = !ctx->flip;
+	stats_read(ctx);
+	stats_drop_stale(ctx);
+
+	sort_ctx = ctx;
+	array_sort(&ctx->lines, ctx->lines_sort);
+	sort_ctx = NULL;
+	return TRUE;
+}
+
+static void
+stats_top_output_diff(struct top_context *ctx,
+		      const struct top_line *line, unsigned int i)
+{
+	uint64_t prev_num, cur_num;
+	double prev_double, cur_double, prev_time, cur_time;
+	char numstr[MAX_INT_STRLEN];
+
+	if (str_to_uint64(line->prev_values[i], &prev_num) == 0 &&
+	    str_to_uint64(line->cur_values[i], &cur_num) == 0) {
+		i_snprintf(numstr, sizeof(numstr), "%llu",
+			   (unsigned long long)(cur_num - prev_num));
+		doveadm_print(numstr);
+	} else if (get_double(line->prev_values[i], &prev_double) == 0 &&
+		   get_double(line->cur_values[i], &cur_double) == 0 &&
+		   get_double(line->prev_values[ctx->last_update_idx], &prev_time) == 0 &&
+		   get_double(line->cur_values[ctx->last_update_idx], &cur_time) == 0) {
+		/* %CPU */
+		i_snprintf(numstr, sizeof(numstr), "%d",
+			   (int)((cur_double - prev_double) *
+				 (cur_time - prev_time) * 100));
+		doveadm_print(numstr);
+	} else {
+		doveadm_print(line->cur_values[i]);
+	}
+}
+
+static void stats_top_output(struct top_context *ctx)
+{
+	static const char *names[] = {
+		"user", "service", "user_cpu", "sys_cpu",
+		"disk_input", "disk_output"
+	};
+	struct winsize ws;
+	struct top_line *const *lines;
+	unsigned int i, j, row, maxrow, count, indexes[N_ELEMENTS(names)];
+
+	/* ANSI clear screen and move cursor to top of screen */
+	printf("\x1b[2J\x1b[1;1H"); fflush(stdout);
+	doveadm_print_deinit();
+	doveadm_print_init(DOVEADM_PRINT_TYPE_TABLE);
+
+	doveadm_print_header_simple("USER");
+	doveadm_print_header_simple("SERVICE");
+	doveadm_print_header_simple("%CPU");
+	doveadm_print_header_simple("%SYS");
+	doveadm_print_header_simple("DISKIN");
+	doveadm_print_header_simple("DISKOUT");
+
+	if (!stats_top_round(ctx)) {
+		/* no connections yet */
+		return;
+	}
+
+	for (i = 0; i < N_ELEMENTS(names); i++) {
+		if (!stats_header_find(ctx, names[i], &indexes[i]))
+			indexes[i] = -1U;
+	}
+
+	if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &ws) < 0)
+		ws.ws_row = 24;
+	maxrow = ws.ws_row-1;
+
+	lines = array_get(&ctx->lines, &count);
+	for (i = 0, row = 1; row < maxrow && i < count; i++, row++) {
+		for (j = 0; j < N_ELEMENTS(names); j++) {
+			if (indexes[j] == -1U)
+				doveadm_print("?");
+			else
+				stats_top_output_diff(ctx, lines[i], indexes[j]);
+		}
+	}
+}
+
+static void stats_top_start(struct top_context *ctx)
+{
+	struct timeout *to;
+
+	stats_top_output(ctx);
+	to = timeout_add(1000, stats_top_output, ctx);
+	io_loop_run(current_ioloop);
+	timeout_remove(&to);
+}
+
+static void stats_top(const char *path, const char *sort_type)
+{
+	struct top_context ctx;
+
+	memset(&ctx, 0, sizeof(ctx));
+	ctx.path = path;
+	ctx.fd = doveadm_connect(path);
+	ctx.prev_pool = pool_alloconly_create("stats top", 1024*16);
+	ctx.cur_pool = pool_alloconly_create("stats top", 1024*16);
+	i_array_init(&ctx.lines, 128);
+	ctx.sessions =
+		hash_table_create(default_pool, default_pool, 0,
+				  str_hash, (hash_cmp_callback_t *)strcmp);
+	net_set_nonblock(ctx.fd, FALSE);
+
+	ctx.input = i_stream_create_fd(ctx.fd, (size_t)-1, TRUE);
+
+	if (strstr(sort_type, "cpu") != NULL)
+		ctx.lines_sort = sort_cpu;
+	else
+		ctx.lines_sort = sort_num;
+	ctx.sort_type = sort_type;
+
+	stats_top_start(&ctx);
+	i_stream_destroy(&ctx.input);
+	hash_table_destroy(&ctx.sessions);
+	array_free(&ctx.lines);
+	pool_unref(&ctx.prev_pool);
+	pool_unref(&ctx.cur_pool);
+	(void)close(ctx.fd);
+}
+
+static void cmd_stats_top(int argc, char *argv[])
+{
+	const char *path, *sort_type;
+	int c;
+
+	path = t_strconcat(doveadm_settings->base_dir, "/stats", NULL);
+
+	while ((c = getopt(argc, argv, "s:")) > 0) {
+		switch (c) {
+		case 's':
+			path = optarg;
+			break;
+		default:
+			help(&doveadm_cmd_stats_top);
+		}
+	}
+	argv += optind - 1;
+	if (argv[1] == NULL)
+		sort_type = "disk";
+	else if (argv[2] != NULL)
+		help(&doveadm_cmd_stats_top);
+	else
+		sort_type = argv[1];
+
+	doveadm_print_init(DOVEADM_PRINT_TYPE_TABLE);
+	stats_top(path, sort_type);
+}
+
+struct doveadm_cmd doveadm_cmd_stats_dump = {
 	cmd_stats_dump, "stats dump", "[-s <stats socket path>] <type> [<filter>]"
 };
+
+struct doveadm_cmd doveadm_cmd_stats_top = {
+	cmd_stats_top, "stats top", "[-s <stats socket path>] [<sort field>]"
+};

--- a/src/doveadm/doveadm.c	Thu Sep 01 10:37:55 2011 +0300
+++ b/src/doveadm/doveadm.c	Thu Sep 01 10:39:19 2011 +0300
@@ -260,7 +260,8 @@
 	&doveadm_cmd_mailbox_mutf7,
 	&doveadm_cmd_sis_deduplicate,
 	&doveadm_cmd_sis_find,
-	&doveadm_cmd_stats
+	&doveadm_cmd_stats_dump,
+	&doveadm_cmd_stats_top
 };
 
 int main(int argc, char *argv[])

--- a/src/doveadm/doveadm.h	Thu Sep 01 10:37:55 2011 +0300
+++ b/src/doveadm/doveadm.h	Thu Sep 01 10:39:19 2011 +0300
@@ -26,7 +26,8 @@
 extern struct doveadm_cmd doveadm_cmd_mailbox_mutf7;
 extern struct doveadm_cmd doveadm_cmd_sis_deduplicate;
 extern struct doveadm_cmd doveadm_cmd_sis_find;
-extern struct doveadm_cmd doveadm_cmd_stats;
+extern struct doveadm_cmd doveadm_cmd_stats_dump;
+extern struct doveadm_cmd doveadm_cmd_stats_top;
 
 void doveadm_register_cmd(const struct doveadm_cmd *cmd);