ccanlint: infrastructure to run commands async.
authorRusty Russell <rusty@rustcorp.com.au>
Mon, 12 Sep 2011 04:50:26 +0000 (14:20 +0930)
committerRusty Russell <rusty@rustcorp.com.au>
Mon, 12 Sep 2011 04:50:26 +0000 (14:20 +0930)
We use the lbalance module to figure out how many to run in parallel.

tools/ccanlint/Makefile
tools/ccanlint/async.c [new file with mode: 0644]
tools/ccanlint/ccanlint.h

index 846adf627dc86277327477fe8b3c65507dc6cc3d..ee371e054c87bf084eb32e3c290a611fe63492b1 100644 (file)
@@ -9,6 +9,7 @@ CORE_OBJS := \
        ccan/grab_file/grab_file.o \
        ccan/hash/hash.o \
        ccan/htable/htable.o \
+       ccan/lbalance/lbalance.o \
        ccan/noerr/noerr.o \
        ccan/opt/helpers.o \
        ccan/opt/opt.o \
@@ -20,6 +21,7 @@ CORE_OBJS := \
        ccan/talloc/talloc.o \
        ccan/talloc_link/talloc_link.o \
        ccan/time/time.o \
+       tools/ccanlint/async.o \
        tools/ccanlint/ccanlint.o \
        tools/ccanlint/file_analysis.o \
        tools/ccanlint/licenses.o \
diff --git a/tools/ccanlint/async.c b/tools/ccanlint/async.c
new file mode 100644 (file)
index 0000000..32a67d2
--- /dev/null
@@ -0,0 +1,200 @@
+#include "ccanlint.h"
+#include "../tools.h"
+#include <sys/types.h>
+#include <sys/time.h>
+#include <sys/resource.h>
+#include <sys/wait.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <err.h>
+#include <assert.h>
+#include <ccan/lbalance/lbalance.h>
+#include <ccan/tlist/tlist.h>
+#include <ccan/grab_file/grab_file.h>
+#include <ccan/time/time.h>
+#include <ccan/talloc/talloc.h>
+
+static struct lbalance *lb;
+TLIST_TYPE(command, struct command);
+static struct tlist_command pending = TLIST_INIT(pending);
+static struct tlist_command running = TLIST_INIT(running);
+static unsigned int num_running = 0;
+static struct tlist_command done = TLIST_INIT(done);
+
+struct command {
+       struct list_node list;
+       char *command;
+       pid_t pid;
+       int output_fd;
+       unsigned int time_ms;
+       struct lbalance_task *task;
+       int status;
+       char *output;
+       bool done;
+       const void *ctx;
+};
+
+static void killme(int sig)
+{
+       kill(-getpid(), SIGKILL);
+}
+
+static void run_more(void)
+{
+       struct command *c;
+
+       while (num_running < lbalance_target(lb)) {
+               int p[2];
+               c = tlist_top(&pending, struct command, list);
+               if (!c)
+                       break;
+
+               if (pipe(p) != 0)
+                       err(1, "Pipe failed");
+               c->pid = fork();
+               if (c->pid == -1)
+                       err(1, "Fork failed");
+               if (c->pid == 0) {
+                       struct itimerval itim;
+
+                       if (dup2(p[1], STDOUT_FILENO) != STDOUT_FILENO
+                           || dup2(p[1], STDERR_FILENO) != STDERR_FILENO
+                           || close(p[0]) != 0
+                           || close(STDIN_FILENO) != 0
+                           || open("/dev/null", O_RDONLY) != STDIN_FILENO)
+                               exit(128);
+
+                       signal(SIGALRM, killme);
+                       itim.it_interval.tv_sec = itim.it_interval.tv_usec = 0;
+                       itim.it_value = time_from_msec(c->time_ms);
+                       setitimer(ITIMER_REAL, &itim, NULL);
+
+                       c->status = system(c->command);
+                       if (WIFEXITED(c->status))
+                               exit(WEXITSTATUS(c->status));
+                       /* Here's a hint... */
+                       exit(128 + WTERMSIG(c->status));
+               }
+
+               if (tools_verbose)
+                       printf("Running async: %s => %i\n", c->command, c->pid);
+
+               close(p[1]);
+               c->output_fd = p[0];
+               c->task = lbalance_task_new(lb);
+               tlist_del_from(&pending, c, list);
+               tlist_add_tail(&running, c, list);
+               num_running++;
+       }
+}
+
+static int destroy_command(struct command *command)
+{
+       if (!command->done && command->pid) {
+               kill(-command->pid, SIGKILL);
+               close(command->output_fd);
+               num_running--;
+       }
+
+       tlist_del(command, list);
+       return 0;
+}
+
+void run_command_async(const void *ctx, unsigned int time_ms,
+                      const char *fmt, ...)
+{
+       struct command *command;
+       va_list ap;
+
+       assert(ctx);
+
+       if (!lb)
+               lb = lbalance_new();
+
+       command = talloc(ctx, struct command);
+       command->ctx = ctx;
+       command->time_ms = time_ms;
+       command->pid = 0;
+       command->output = talloc_strdup(command, "");
+       va_start(ap, fmt);
+       command->command = talloc_vasprintf(command, fmt, ap);
+       va_end(ap);
+       tlist_add_tail(&pending, command, list);
+       command->done = false;
+       talloc_set_destructor(command, destroy_command);
+
+       run_more();
+}
+
+static void reap_output(void)
+{
+       fd_set in;
+       struct command *c, *next;
+       int max_fd = 0;
+
+       FD_ZERO(&in);
+
+       tlist_for_each(&running, c, list) {
+               FD_SET(c->output_fd, &in);
+               if (c->output_fd > max_fd)
+                       max_fd = c->output_fd;
+       }
+
+       if (select(max_fd+1, &in, NULL, NULL, NULL) < 0)
+               err(1, "select failed");
+
+       tlist_for_each_safe(&running, c, next, list) {
+               if (FD_ISSET(c->output_fd, &in)) {
+                       int old_len, len;
+                       old_len = talloc_array_length(c->output);
+                       c->output = talloc_realloc(c, c->output, char,
+                                                  old_len + 1024);
+                       len = read(c->output_fd, c->output + old_len, 1024);
+                       if (len < 0)
+                               err(1, "Reading from async command");
+                       c->output = talloc_realloc(c, c->output, char,
+                                                  old_len + len);
+                       if (len == 0) {
+                               struct rusage ru;
+                               wait4(c->pid, &c->status, 0, &ru);
+                               if (tools_verbose)
+                                       printf("Finished async %i: %s %u\n",
+                                              c->pid,
+                                              WIFEXITED(c->status)
+                                              ? "exit status"
+                                              : "killed by signal",
+                                              WIFEXITED(c->status)
+                                              ? WEXITSTATUS(c->status)
+                                              : WTERMSIG(c->status));
+                               lbalance_task_free(c->task, &ru);
+                               c->task = NULL;
+                               c->done = true;
+                               close(c->output_fd);
+                               tlist_del_from(&running, c, list);
+                               tlist_add_tail(&done, c, list);
+                               num_running--;
+                       }
+               }
+       }
+}
+
+void *collect_command(bool *ok, char **output)
+{
+       struct command *c;
+       const void *ctx;
+
+       while ((c = tlist_top(&done, struct command, list)) == NULL) {
+               if (tlist_empty(&pending) && tlist_empty(&running))
+                       return NULL;
+               reap_output();
+               run_more();
+       }
+
+       *ok = (WIFEXITED(c->status) && WEXITSTATUS(c->status) == 0);
+       ctx = c->ctx;
+       *output = talloc_steal(ctx, c->output);
+       talloc_free(c);
+       return (void *)ctx;
+}
index 9a4817092acecc6c08e67bffac7c43158bd3c1da..13b10dcdd2a3539ba19ad2f700225816111ec6ff 100644 (file)
@@ -235,6 +235,13 @@ char **per_file_options(const struct ccanlint *test, struct ccan_file *f);
 void score_file_error(struct score *, struct ccan_file *f, unsigned line,
                      const char *errorfmt, ...);
 
+/* Start a command in the background. */
+void run_command_async(const void *ctx, unsigned int time_ms,
+                      const char *fmt, ...);
+
+/* Get results of a command, returning ctx (and free it). */
+void *collect_command(bool *ok, char **output);
+
 /* Normal tests. */
 extern struct ccanlint trailing_whitespace;