4 files changed, 639 insertions, 11 deletions

diff --git a/Makefile b/Makefile
index 08f488c..481b1a7 100644
--- a/Makefile
+++ b/Makefile
@@ -140,7 +140,7 @@ endef
         $(Q)$(call check_gui)
 
 
-TRACE_CMD_OBJS = trace-cmd.o trace-read.o trace-split.o
+TRACE_CMD_OBJS = trace-cmd.o trace-read.o trace-split.o trace-listen.o
 TRACE_VIEW_OBJS = trace-view.o trace-view-store.o trace-filter.o trace-compat.o \
         trace-hash.o
 TRACE_GRAPH_OBJS = trace-graph.o trace-compat.o trace-hash.o trace-filter.o \
diff --git a/trace-cmd.c b/trace-cmd.c
index e89362d..cdd78f4 100644
--- a/trace-cmd.c
+++ b/trace-cmd.c
@@ -27,6 +27,8 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
+#include <sys/socket.h>
+#include <netdb.h>
 #include <pthread.h>
 #include <fcntl.h>
 #include <unistd.h>
@@ -48,7 +50,7 @@
 #define ITER_CTRL       "trace_options"
 #define MAX_LATENCY     "tracing_max_latency"
 
-unsigned int page_size;
+static unsigned int page_size;
 
 static const char *output_file = "trace.dat";
 
@@ -57,6 +59,10 @@ static int sleep_time = 1000;
 static int cpu_count;
 static int *pids;
 
+static char *host;
+static int *client_ports;
+static int sfd;
+
 static int filter_task;
 static int filter_pid = -1;
 
@@ -116,7 +122,7 @@ static void kill_threads(void)
 {
         int i;
 
-        if (!cpu_count)
+        if (!cpu_count || !pids)
                 return;
 
         for (i = 0; i < cpu_count; i++) {
@@ -895,6 +901,41 @@ static void flush(int sig)
                 tracecmd_stop_recording(recorder);
 }
 
+static void connect_port(int cpu)
+{
+        struct addrinfo hints;
+        struct addrinfo *results, *rp;
+        int s;
+        char buf[BUFSIZ];
+
+        snprintf(buf, BUFSIZ, "%d", client_ports[cpu]);
+
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_DGRAM;
+
+        s = getaddrinfo(host, buf, &hints, &results);
+        if (s != 0)
+                die("connecting to UDP server %s:%s", host, buf);
+
+        for (rp = results; rp != NULL; rp = rp->ai_next) {
+                sfd = socket(rp->ai_family, rp->ai_socktype,
+                             rp->ai_protocol);
+                if (sfd == -1)
+                        continue;
+                if (connect(sfd, rp->ai_addr, rp->ai_addrlen) != 1)
+                        break;
+                close(sfd);
+        }
+
+        if (rp == NULL)
+                die("Can not connect to UDP server %s:%s", host, buf);
+
+        freeaddrinfo(results);
+
+        client_ports[cpu] = sfd;
+}
+
 static int create_recorder(int cpu)
 {
         char *file;
@@ -913,10 +954,14 @@ static int create_recorder(int cpu)
         /* do not kill tasks on error */
         cpu_count = 0;
 
-        file = get_temp_file(cpu);
-
-        recorder = tracecmd_create_recorder(file, cpu);
-        put_temp_file(file);
+        if (client_ports) {
+                connect_port(cpu);
+                recorder = tracecmd_create_recorder_fd(client_ports[cpu], cpu);
+        } else {
+                file = get_temp_file(cpu);
+                recorder = tracecmd_create_recorder(file, cpu);
+                put_temp_file(file);
+        }
 
         if (!recorder)
                 die ("can't create recorder");
@@ -927,18 +972,125 @@ static int create_recorder(int cpu)
         exit(0);
 }
 
+static void setup_network(void)
+{
+        struct tracecmd_output *handle;
+        struct addrinfo hints;
+        struct addrinfo *result, *rp;
+        int sfd, s;
+        ssize_t n;
+        char buf[BUFSIZ];
+        char *server;
+        char *port;
+        char *p;
+        int cpu;
+        int i;
+
+        if (!strchr(host, ':')) {
+                server = strdup("localhost");
+                if (!server)
+                        die("alloctating server");
+                port = host;
+                host = server;
+        } else {
+                host = strdup(host);
+                if (!host)
+                        die("alloctating server");
+                server = strtok_r(host, ":", &p);
+                port = strtok_r(NULL, ":", &p);
+        }
+
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_STREAM;
+
+        s = getaddrinfo(server, port, &hints, &result);
+        if (s != 0)
+                die("getaddrinfo: %s", gai_strerror(s));
+
+        for (rp = result; rp != NULL; rp = rp->ai_next) {
+                sfd = socket(rp->ai_family, rp->ai_socktype,
+                             rp->ai_protocol);
+                if (sfd == -1)
+                        continue;
+
+                if (connect(sfd, rp->ai_addr, rp->ai_addrlen) != -1)
+                        break;
+                close(sfd);
+        }
+
+        if (!rp)
+                die("Can not connect to %s:%d", server, port);
+
+        freeaddrinfo(result);
+
+        n = read(sfd, buf, 8);
+
+        /* Make sure the server is the tracecmd server */
+        if (memcmp(buf, "tracecmd", 8) != 0)
+                die("server not tracecmd server");
+
+        /* write the number of CPUs we have (in ASCII) */
+
+        sprintf(buf, "%d", cpu_count);
+
+        /* include \0 */
+        write(sfd, buf, strlen(buf)+1);
+
+        /* write the pagesize (in ASCII) */
+
+        page_size = getpagesize();
+        sprintf(buf, "%d", page_size);
+
+        /* include \0 */
+        write(sfd, buf, strlen(buf)+1);
+
+        client_ports = malloc_or_die(sizeof(int) * cpu_count);
+
+        /*
+         * Now we will receive back a comma deliminated list
+         * of client ports to connect to.
+         */
+        for (cpu = 0; cpu < cpu_count; cpu++) {
+                for (i = 0; i < BUFSIZ; i++) {
+                        n = read(sfd, buf+i, 1);
+                        if (n != 1)
+                                die("Error, reading server ports");
+                        if (!buf[i] || buf[i] == ',')
+                                break;
+                }
+                if (i == BUFSIZ)
+                        die("read bad port number");
+                buf[i] = 0;
+                client_ports[cpu] = atoi(buf);
+        }
+
+        /* Now create the handle through this socket */
+        handle = tracecmd_create_init_fd(sfd, cpu_count);
+
+        /* OK, we are all set, let'r rip! */
+}
+
+static void finish_network(void)
+{
+        close(sfd);
+        free(host);
+}
+
 static void start_threads(void)
 {
         int i;
 
         cpu_count = count_cpus();
 
+        if (host)
+                setup_network();
+
         /* make a thread for every CPU we have */
         pids = malloc_or_die(sizeof(*pids) * cpu_count);
 
         memset(pids, 0, sizeof(*pids) * cpu_count);
 
-
         for (i = 0; i < cpu_count; i++) {
                 pids[i] = create_recorder(i);
         }
@@ -950,6 +1102,11 @@ static void record_data(void)
         char **temp_files;
         int i;
 
+        if (host) {
+                finish_network();
+                return;
+        }
+
         if (latency)
                 handle = tracecmd_create_file_latency(output_file, cpu_count);
         else {
@@ -1066,7 +1223,7 @@ void usage(char **argv)
                "usage:\n"
                " %s record [-v][-e event [-f filter]][-p plugin][-F][-d][-o file] \\\n"
                "           [-s usecs][-O option ][-l func][-g func][-n func]\n"
-               "           [-P pid][command ...]\n"
+               "           [-P pid][-N host:port][command ...]\n"
                "          -e run command with event enabled\n"
                "          -f filter for previous -e event\n"
                "          -p run command with plugin enabled\n"
@@ -1080,6 +1237,7 @@ void usage(char **argv)
                "          -o data output file [default trace.dat]\n"
                "          -O option to enable (or disable)\n"
                "          -s sleep interval between recording (in usecs) [default: 1000]\n"
+               "          -N host:port to connect to (see listen)\n"
                "\n"
                " %s start [-e event][-p plugin][-d][-O option ][-P pid]\n"
                "          Uses same options as record, but does not run a command.\n"
@@ -1122,11 +1280,17 @@ void usage(char **argv)
                "                  if left out, will start at beginning of file\n"
                "          end   - decimal end time in seconds\n"
                "\n"
+               " %s listen -p port[-D][-o file][-d dir]\n"
+               "          Creates a socket to listen for clients.\n"
+               "          -D create it in daemon mode.\n"
+               "          -o file name to use for clients.\n"
+               "          -d diretory to store client files.\n"
+               "\n"
                " %s list [-e][-p]\n"
                "          -e list available events\n"
                "          -p list available plugins\n"
                "          -o list available options\n"
-               "\n", p, VERSION_STRING, p, p, p, p, p, p, p, p);
+               "\n", p, VERSION_STRING, p, p, p, p, p, p, p, p, p);
         exit(-1);
 }
 
@@ -1159,6 +1323,9 @@ int main (int argc, char **argv)
         if (strcmp(argv[1], "report") == 0) {
                 trace_report(argc, argv);
                 exit(0);
+        } else if (strcmp(argv[1], "listen") == 0) {
+                trace_listen(argc, argv);
+                exit(0);
         } else if (strcmp(argv[1], "split") == 0) {
                 trace_split(argc, argv);
                 exit(0);
@@ -1166,7 +1333,7 @@ int main (int argc, char **argv)
                    (strcmp(argv[1], "start") == 0) ||
                    ((extract = strcmp(argv[1], "extract") == 0))) {
 
-                while ((c = getopt(argc-1, argv+1, "+he:f:Fp:do:O:s:vg:l:n:P:")) >= 0) {
+                while ((c = getopt(argc-1, argv+1, "+he:f:Fp:do:O:s:vg:l:n:P:N:")) >= 0) {
                         switch (c) {
                         case 'h':
                                 usage(argv);
@@ -1241,6 +1408,8 @@ int main (int argc, char **argv)
                                 disable = 1;
                                 break;
                         case 'o':
+                                if (host)
+                                        die("-o incompatible with -N");
                                 if (!record && !extract)
                                         die("start does not take output\n"
                                             "Did you mean 'record'?");
@@ -1257,6 +1426,13 @@ int main (int argc, char **argv)
                                         usage(argv);
                                 sleep_time = atoi(optarg);
                                 break;
+                        case 'N':
+                                if (!record)
+                                        die("-N only available with record");
+                                if (output)
+                                        die("-N incompatible with -o");
+                                host = optarg;
+                                break;
                         }
                 }
 
diff --git a/trace-listen.c b/trace-listen.c
new file mode 100644
index 0000000..d1314a5
--- /dev/null
+++ b/trace-listen.c
@@ -0,0 +1,450 @@
+/*
+ * Copyright (C) 2009, 2010 Red Hat Inc, Steven Rostedt <srostedt@redhat.com>
+ *
+ * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License (not later!)
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *
+ * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+ */
+#define _LARGEFILE64_SOURCE
+#define _GNU_SOURCE
+#include <dirent.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <getopt.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <signal.h>
+#include <errno.h>
+
+#include "trace-local.h"
+
+static char *default_output_dir = ".";
+static char *output_dir;
+static char *default_output_file = "trace";
+static char *output_file;
+
+static int backlog = 5;
+
+#define  TEMP_FILE_STR "%s.%s:%s.cpu%d", output_file, host, port, cpu
+static char *get_temp_file(const char *host, const char *port, int cpu)
+{
+        char *file = NULL;
+        int size;
+
+        size = snprintf(file, 0, TEMP_FILE_STR);
+        file = malloc_or_die(size + 1);
+        sprintf(file, TEMP_FILE_STR);
+
+        return file;
+}
+
+static void put_temp_file(char *file)
+{
+        free(file);
+}
+
+#define MAX_PATH 1024
+
+static void delete_temp_file(const char *host, const char *port, int cpu)
+{
+        char file[MAX_PATH];
+
+        snprintf(file, MAX_PATH, TEMP_FILE_STR);
+        unlink(file);
+}
+
+static int read_string(int fd, char *buf, size_t size)
+{
+        size_t i;
+        int n;
+
+        for (i = 0; i < size; i++) {
+                n = read(fd, buf+i, 1);
+                if (!buf[i] || n <= 0)
+                        break;
+        }
+
+        return i;
+}
+
+static int done;
+static void finish(int sig)
+{
+        done = 1;
+}
+
+static void process_udp_child(int sfd, const char *host, const char *port,
+                              int cpu, int page_size)
+{
+        struct sockaddr_storage peer_addr;
+        socklen_t peer_addr_len;
+        char buf[page_size];
+        char *tempfile;
+        int fd;
+        int n;
+        int once = 0;
+
+        signal(SIGUSR1, finish);
+
+        tempfile = get_temp_file(host, port, cpu);
+        fd = open(tempfile, O_WRONLY | O_TRUNC | O_CREAT, 0644);
+        if (fd < 0)
+                die("creating %s", tempfile);
+
+        do {
+                peer_addr_len = sizeof(peer_addr);
+                /* TODO, make this copyless! */
+                n = read(sfd, buf, page_size);
+#if 0
+                n = recvfrom(sfd, buf, page_size, 0,
+                             (struct sockaddr *)&peer_addr, &peer_addr_len);
+#endif
+                if (!n)
+                        break;
+                if (n < page_size && !once) {
+                        once = 1;
+                        warning("read %d bytes, expected %d", n, page_size);
+                }
+                write(fd, buf, n);
+        } while (!done);
+
+        put_temp_file(tempfile);
+        exit(0);
+}
+
+#define START_PORT_SEARCH 1500
+#define MAX_PORT_SEARCH 6000
+
+static int open_udp(const char *node, const char *port, int *pid,
+                    int cpu, int pagesize)
+{
+        struct addrinfo hints;
+        struct addrinfo *result, *rp;
+        int sfd, s;
+        char buf[BUFSIZ];
+        int num_port = START_PORT_SEARCH;
+
+ again:
+        snprintf(buf, BUFSIZ, "%d", num_port);
+
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_DGRAM;
+        hints.ai_flags = AI_PASSIVE;
+
+        s = getaddrinfo(NULL, buf, &hints, &result);
+        if (s != 0)
+                die("getaddrinfo: error opening udp socket");
+
+        for (rp = result; rp != NULL; rp = rp->ai_next) {
+                sfd = socket(rp->ai_family, rp->ai_socktype,
+                             rp->ai_protocol);
+                if (sfd < 0)
+                        continue;
+
+                if (bind(sfd, rp->ai_addr, rp->ai_addrlen) == 0)
+                        break;
+
+                close(sfd);
+        }
+
+        if (rp == NULL) {
+                freeaddrinfo(result);
+                if (++num_port > MAX_PORT_SEARCH)
+                        die("No available ports to bind");
+                goto again;
+        }
+
+        freeaddrinfo(result);
+
+        *pid = fork();
+
+        if (*pid < 0)
+                die("creating udp reader");
+
+        if (!*pid)
+                process_udp_child(sfd, node, port, cpu, pagesize);
+
+        close(sfd);
+
+        return num_port;
+}
+
+static void process_client(const char *node, const char *port, int fd)
+{
+        char **temp_files;
+        char buf[BUFSIZ];
+        int *port_array;
+        int *pid_array;
+        int pagesize;
+        int udp_port;
+        int cpus;
+        int cpu;
+        int pid;
+        int ofd;
+        int n, s, t;
+
+        /* Let the client know what we are */
+        write(fd, "tracecmd", 8);
+
+        /* read back the CPU count */
+        n = read_string(fd, buf, BUFSIZ);
+        if (n == BUFSIZ)
+                /** ERROR **/
+                return;
+
+        cpus = atoi(buf);
+
+        printf("cpus=%d\n", cpus);
+        if (cpus < 0)
+                return;
+
+        /* next read the page size */
+        n = read_string(fd, buf, BUFSIZ);
+        if (n == BUFSIZ)
+                /** ERROR **/
+                return;
+
+        pagesize = atoi(buf);
+
+        printf("pagesize=%d\n", pagesize);
+        if (pagesize <= 0)
+                return;
+
+        /* Create the client file */
+        snprintf(buf, BUFSIZ, "%s.%s:%s.dat", output_file, node, port);
+
+        ofd = open(buf, O_RDWR | O_CREAT | O_TRUNC, 0644);
+        if (ofd < 0)
+                die("Can not create file %s", buf);
+
+        port_array = malloc_or_die(sizeof(int) * cpus);
+        pid_array = malloc_or_die(sizeof(int) * cpus);
+        memset(pid_array, 0, sizeof(int) * cpus);
+
+        /* Now create a UDP port for each CPU */
+        for (cpu = 0; cpu < cpus; cpu++) {
+                udp_port = open_udp(node, port, &pid, cpu, pagesize);
+                if (udp_port < 0)
+                        goto out_free;
+                port_array[cpu] = udp_port;
+                pid_array[cpu] = pid;
+        }
+
+        /* send the client a comma deliminated set of port numbers */
+        for (cpu = 0; cpu < cpus; cpu++) {
+                snprintf(buf, BUFSIZ, "%s%d",
+                         cpu ? "," : "", port_array[cpu]);
+                write(fd, buf, strlen(buf));
+        }
+        /* end with null terminator */
+        write(fd, "\0", 1);
+
+        /* Now we are ready to start reading data from the client */
+        do {
+                n = read(fd, buf, BUFSIZ);
+                t = n;
+                s = 0;
+                do {
+                        s = write(ofd, buf+s, t);
+                        if (s < 0)
+                                die("writing to file");
+                        t -= s;
+                        s = n - t;
+                } while (t);
+        } while (n > 0);
+
+        /* wait a little to let our readers finish reading */
+        sleep(1);
+
+        /* stop our readers */
+        for (cpu = 0; cpu < cpus; cpu++) {
+                if (pid_array[cpu] > 0)
+                        kill(pid_array[cpu], SIGUSR1);
+        }
+
+        /* wait a little to have the readers clean up */
+        sleep(1);
+
+        /* Now put together the file */
+        temp_files = malloc_or_die(sizeof(*temp_files) * cpus);
+
+        for (cpu = 0; cpu < cpus; cpu++)
+                temp_files[cpu] = get_temp_file(node, port, cpu);
+
+        tracecmd_attach_cpu_data_fd(ofd, cpus, temp_files);
+
+ out_free:
+        for (cpu = 0; cpu < cpus; cpu++) {
+                if (pid_array[cpu] > 0) {
+                        kill(pid_array[cpu], SIGKILL);
+                        delete_temp_file(node, port, cpu);
+                        pid_array[cpu] = 0;
+                }
+        }
+}
+
+static void do_listen(char *port)
+{
+        struct addrinfo hints;
+        struct addrinfo *result, *rp;
+        int sfd, s, cfd;
+        struct sockaddr_storage peer_addr;
+        socklen_t peer_addr_len;
+        ssize_t nread;
+        char buf[BUFSIZ];
+        char host[NI_MAXHOST], service[NI_MAXSERV];
+
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_STREAM;
+        hints.ai_flags = AI_PASSIVE;
+
+        s = getaddrinfo(NULL, port, &hints, &result);
+        if (s != 0)
+                die("getaddrinfo: error opening %s", port);
+
+        for (rp = result; rp != NULL; rp = rp->ai_next) {
+                sfd = socket(rp->ai_family, rp->ai_socktype,
+                             rp->ai_protocol);
+                if (sfd < 0)
+                        continue;
+
+                if (bind(sfd, rp->ai_addr, rp->ai_addrlen) == 0)
+                        break;
+
+                close(sfd);
+        }
+
+        if (rp == NULL)
+                die("Could not bind");
+
+        freeaddrinfo(result);
+
+        if (listen(sfd, backlog) < 0)
+                die("listen");
+
+        peer_addr_len = sizeof(peer_addr);
+
+        do {
+                cfd = accept(sfd, (struct sockaddr *)&peer_addr, &peer_addr_len);
+                if (cfd < 0)
+                        die("connecting");
+                s = getnameinfo((struct sockaddr *)&peer_addr, peer_addr_len,
+                                host, NI_MAXHOST,
+                                service, NI_MAXSERV, NI_NUMERICSERV);
+
+                if (s == 0)
+                        printf("Connected with %s:%s\n",
+                               host, service);
+                else {
+                        printf("Error with getnameinfo: %s\n",
+                               gai_strerror(s));
+                        close(cfd);
+                        close(sfd);
+                        return;
+                }
+
+                process_client(host, service, cfd);
+
+                do {
+                        if (nread > 0)
+                                nread = read(cfd, buf, BUFSIZ);
+                        if (cfd < 0)
+                                die("client");
+                        if (nread > 0)
+                                write(1, buf, nread);
+                } while (nread);
+
+                close(cfd);
+        } while (0);
+}
+
+static void start_daemon(void)
+{
+}
+
+void trace_listen(int argc, char **argv)
+{
+        char *port = NULL;
+        char *iface;
+        int daemon = 0;
+        int c;
+
+        if (argc < 2)
+                usage(argv);
+
+        if (strcmp(argv[1], "listen") != 0)
+                usage(argv);
+
+        for (;;) {
+                int option_index = 0;
+                static struct option long_options[] = {
+                        {"port", required_argument, NULL, 'p'},
+                        {"help", no_argument, NULL, '?'},
+                        {NULL, 0, NULL, 0}
+                };
+
+                c = getopt_long (argc-1, argv+1, "+hp:o:d:i:D",
+                        long_options, &option_index);
+                if (c == -1)
+                        break;
+                switch (c) {
+                case 'h':
+                        usage(argv);
+                        break;
+                case 'p':
+                        port = optarg;
+                        break;
+                case 'i':
+                        iface = optarg;
+                        break;
+                case 'd':
+                        output_dir = optarg;
+                        break;
+                case 'o':
+                        output_file = optarg;
+                        break;
+                case 'D':
+                        daemon = 1;
+                default:
+                        usage(argv);
+                }
+        }
+
+        if (!port)
+                usage(argv);
+
+        if ((argc - optind) >= 2)
+                usage(argv);
+
+        if (!output_file)
+                output_file = default_output_file;
+
+        if (!output_dir)
+                output_dir = default_output_dir;
+
+        if (daemon)
+                start_daemon();
+
+        do_listen(port);
+
+        return;
+}
diff --git a/trace-local.h b/trace-local.h
index 0e49fa4..4e44238 100644
--- a/trace-local.h
+++ b/trace-local.h
@@ -37,4 +37,6 @@ void trace_report(int argc, char **argv);
 
 void trace_split(int argc, char **argv);
 
+void trace_listen(int argc, char **argv);
+
 #endif /* __TRACE_LOCAL_H */