* block queue tracing application
*
* Copyright (C) 2005 Jens Axboe <axboe@suse.de>
+ * Copyright (C) 2006 Jens Axboe <axboe@kernel.dk>
*
* 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
#include <sys/param.h>
#include <sys/statfs.h>
#include <sys/poll.h>
+#include <sys/mman.h>
+#include <sys/socket.h>
#include <stdio.h>
#include <stdlib.h>
#include <sched.h>
#include <ctype.h>
#include <getopt.h>
#include <errno.h>
-#include <assert.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <sys/sendfile.h>
#include "blktrace.h"
+#include "barrier.h"
-static char blktrace_version[] = "0.99";
+static char blktrace_version[] = "0.99.1";
/*
* You may want to increase this even more, if you are logging at a high
#define OFILE_BUF (128 * 1024)
-#define RELAYFS_TYPE 0xF0B4A981
+#define DEBUGFS_TYPE 0x64626720
-#define RING_INIT_NR (2)
-#define RING_MAX_NR (16UL)
-
-#define S_OPTS "d:a:A:r:o:kw:Vb:n:D:"
+#define S_OPTS "d:a:A:r:o:kw:Vb:n:D:lh:p:s"
static struct option l_opts[] = {
{
.name = "dev",
.flag = NULL,
.val = 'D'
},
+ {
+ .name = "listen",
+ .has_arg = no_argument,
+ .flag = NULL,
+ .val = 'l'
+ },
+ {
+ .name = "host",
+ .has_arg = required_argument,
+ .flag = NULL,
+ .val = 'h'
+ },
+ {
+ .name = "port",
+ .has_arg = required_argument,
+ .flag = NULL,
+ .val = 'p'
+ },
+ {
+ .name = "no-sendfile",
+ .has_arg = no_argument,
+ .flag = NULL,
+ .val = 's'
+ },
{
.name = NULL,
}
};
+struct tip_subbuf {
+ void *buf;
+ unsigned int len;
+ unsigned int max_len;
+};
+
+#define FIFO_SIZE (1024) /* should be plenty big! */
+#define CL_SIZE (128) /* cache line, any bigger? */
+
+struct tip_subbuf_fifo {
+ int tail __attribute__((aligned(CL_SIZE)));
+ int head __attribute__((aligned(CL_SIZE)));
+ struct tip_subbuf *q[FIFO_SIZE];
+};
+
struct thread_information {
int cpu;
pthread_t thread;
int fd;
void *fd_buf;
- unsigned long fd_off;
- unsigned long fd_size;
- unsigned long fd_max_size;
char fn[MAXPATHLEN + 64];
- pthread_mutex_t *fd_lock;
FILE *ofile;
char *ofile_buffer;
+ off_t ofile_offset;
+ int ofile_stdout;
+ int ofile_mmap;
+
+ int (*get_subbuf)(struct thread_information *, unsigned int);
+ int (*flush_subbuf)(struct thread_information *, struct tip_subbuf *);
+ int (*read_data)(struct thread_information *, void *, unsigned int);
unsigned long events_processed;
+ unsigned long long data_read;
+ unsigned long long data_queued;
struct device_information *device;
+
+ int exited;
+
+ /*
+ * piped fifo buffers
+ */
+ struct tip_subbuf_fifo fifo;
+ struct tip_subbuf *leftover_ts;
+
+ /*
+ * mmap controlled output files
+ */
+ unsigned long long fs_size;
+ unsigned long long fs_max_size;
+ unsigned long fs_off;
+ void *fs_buf;
+ unsigned long fs_buf_len;
+
+ struct net_connection *nc;
};
struct device_information {
volatile int trace_started;
unsigned long drop_count;
struct thread_information *threads;
+ unsigned long buf_size;
+ unsigned long buf_nr;
+ unsigned int page_size;
+
+ struct cl_host *ch;
+ u32 cl_id;
+ time_t cl_connect_time;
};
static int ncpus;
static struct device_information *device_information;
/* command line option globals */
-static char *relay_path;
+static char *debugfs_path;
static char *output_name;
static char *output_dir;
static int act_mask = ~0U;
static int kill_running_trace;
static unsigned long buf_size = BUF_SIZE;
static unsigned long buf_nr = BUF_NR;
+static unsigned int page_size;
#define is_done() (*(volatile int *)(&done))
static volatile int done;
#define is_stat_shown() (*(volatile int *)(&stat_shown))
static volatile int stat_shown;
-static pthread_mutex_t stdout_mutex = PTHREAD_MUTEX_INITIALIZER;
+int data_is_native = -1;
static void exit_trace(int status);
#define dip_tracing(dip) (*(volatile int *)(&(dip)->trace_started))
#define dip_set_tracing(dip, v) ((dip)->trace_started = (v))
-#define __for_each_dip(__d, __i, __e) \
- for (__i = 0, __d = device_information; __i < __e; __i++, __d++)
+#define __for_each_dip(__d, __di, __e, __i) \
+ for (__i = 0, __d = __di; __i < __e; __i++, __d++)
-#define for_each_dip(__d, __i) __for_each_dip(__d, __i, ndevs)
-#define for_each_tip(__d, __t, __i) \
- for (__i = 0, __t = (__d)->threads; __i < ncpus; __i++, __t++)
+#define for_each_dip(__d, __i) \
+ __for_each_dip(__d, device_information, ndevs, __i)
+#define for_each_nc_dip(__nc, __d, __i) \
+ __for_each_dip(__d, (__nc)->ch->device_information, (__nc)->ch->ndevs, __i)
+
+#define __for_each_tip(__d, __t, __ncpus, __j) \
+ for (__j = 0, __t = (__d)->threads; __j < __ncpus; __j++, __t++)
+#define for_each_tip(__d, __t, __j) \
+ __for_each_tip(__d, __t, ncpus, __j)
+#define for_each_cl_host(__c) \
+ for (__c = cl_host_list; __c; __c = __c->list_next)
+
+/*
+ * networking stuff follows. we include a magic number so we know whether
+ * to endianness convert or not
+ */
+struct blktrace_net_hdr {
+ u32 magic; /* same as trace magic */
+ char buts_name[32]; /* trace name */
+ u32 cpu; /* for which cpu */
+ u32 max_cpus;
+ u32 len; /* length of following trace data */
+ u32 cl_id; /* id for set of client per-cpu connections */
+ u32 buf_size; /* client buf_size for this trace */
+ u32 buf_nr; /* client buf_nr for this trace */
+ u32 page_size; /* client page_size for this trace */
+};
+
+#define TRACE_NET_PORT (8462)
+
+enum {
+ Net_none = 0,
+ Net_server,
+ Net_client,
+};
+
+/*
+ * network cmd line params
+ */
+static char hostname[MAXHOSTNAMELEN];
+static int net_port = TRACE_NET_PORT;
+static int net_mode = 0;
+static int net_use_sendfile = 1;
+
+struct cl_host {
+ struct cl_host *list_next;
+ struct in_addr cl_in_addr;
+ struct net_connection *net_connections;
+ int nconn;
+ struct device_information *device_information;
+ int ndevs;
+ int ncpus;
+ int ndevs_done;
+};
+
+struct net_connection {
+ int in_fd;
+ struct pollfd pfd;
+ time_t connect_time;
+ struct cl_host *ch;
+ int ncpus;
+};
+
+#define NET_MAX_CL_HOSTS (1024)
+static struct cl_host *cl_host_list;
+static int cl_hosts;
+static int net_connects;
+
+static int *net_out_fd;
+
+static void handle_sigint(__attribute__((__unused__)) int sig)
+{
+ struct device_information *dip;
+ int i;
+
+ /*
+ * stop trace so we can reap currently produced data
+ */
+ for_each_dip(dip, i) {
+ if (dip->fd == -1)
+ continue;
+ if (ioctl(dip->fd, BLKTRACESTOP) < 0)
+ perror("BLKTRACESTOP");
+ }
+
+ done = 1;
+}
static int get_dropped_count(const char *buts_name)
{
char tmp[MAXPATHLEN + 64];
snprintf(tmp, sizeof(tmp), "%s/block/%s/dropped",
- relay_path, buts_name);
+ debugfs_path, buts_name);
fd = open(tmp, O_RDONLY);
if (fd < 0) {
struct blk_user_trace_setup buts;
memset(&buts, 0, sizeof(buts));
- buts.buf_size = buf_size;
- buts.buf_nr = buf_nr;
+ buts.buf_size = dip->buf_size;
+ buts.buf_nr = dip->buf_nr;
buts.act_mask = act_mask;
- if (ioctl(dip->fd, BLKSTARTTRACE, &buts) < 0) {
- perror("BLKSTARTTRACE");
+ if (ioctl(dip->fd, BLKTRACESETUP, &buts) < 0) {
+ perror("BLKTRACESETUP");
+ return 1;
+ }
+
+ if (ioctl(dip->fd, BLKTRACESTART) < 0) {
+ perror("BLKTRACESTART");
return 1;
}
if (dip_tracing(dip) || kill_running_trace) {
dip_set_tracing(dip, 0);
- if (ioctl(dip->fd, BLKSTOPTRACE) < 0)
- perror("BLKSTOPTRACE");
+ /*
+ * should be stopped, just don't complain if it isn't
+ */
+ ioctl(dip->fd, BLKTRACESTOP);
+
+ if (ioctl(dip->fd, BLKTRACETEARDOWN) < 0)
+ perror("BLKTRACETEARDOWN");
close(dip->fd);
dip->fd = -1;
}
}
-static void wait_for_data(struct thread_information *tip)
+static void wait_for_data(struct thread_information *tip, int timeout)
{
struct pollfd pfd = { .fd = tip->fd, .events = POLLIN };
while (!is_done()) {
- poll(&pfd, 1, 10);
+ if (poll(&pfd, 1, timeout) < 0) {
+ perror("poll");
+ break;
+ }
if (pfd.revents & POLLIN)
break;
+ if (tip->ofile_stdout)
+ break;
}
}
-static int __read_data(struct thread_information *tip, void *buf, int len,
- int block)
+static int read_data_file(struct thread_information *tip, void *buf,
+ unsigned int len)
{
- char *p = buf;
- int ret, bytes_done = 0;
+ int ret = 0;
- while (!is_done() && bytes_done < len) {
- ret = read(tip->fd, p, len - bytes_done);
- if (ret == (len - bytes_done))
- return len;
-
- if (ret < 0) {
- if (errno == EAGAIN) {
- if (bytes_done || !block)
- break;
- wait_for_data(tip);
- } else {
+ do {
+ wait_for_data(tip, 100);
+
+ ret = read(tip->fd, buf, len);
+ if (!ret)
+ continue;
+ else if (ret > 0)
+ return ret;
+ else {
+ if (errno != EAGAIN) {
perror(tip->fn);
fprintf(stderr,"Thread %d failed read of %s\n",
tip->cpu, tip->fn);
break;
}
- } else if (ret > 0) {
- p += ret;
- bytes_done += ret;
- } else if (bytes_done || !block)
- break;
- else
- wait_for_data(tip);
- }
+ continue;
+ }
+ } while (!is_done());
- if (bytes_done)
- return bytes_done;
- if (!block)
- return 0;
+ return ret;
- return -1;
}
-#define can_grow_ring(tip) ((tip)->fd_max_size < RING_MAX_NR * buf_size * buf_nr)
-
-static int resize_ringbuffer(struct thread_information *tip)
+static int read_data_net(struct thread_information *tip, void *buf,
+ unsigned int len)
{
- if (!can_grow_ring(tip))
- return 1;
+ struct net_connection *nc = tip->nc;
+ unsigned int bytes_left = len;
+ int ret = 0;
- tip->fd_buf = realloc(tip->fd_buf, 2 * tip->fd_max_size);
+ do {
+ ret = recv(nc->in_fd, buf, bytes_left, MSG_WAITALL);
- /*
- * if the ring currently wraps, copy range over
- */
- if (tip->fd_off + tip->fd_size > tip->fd_max_size) {
- unsigned long wrap_size = tip->fd_size - (tip->fd_max_size - tip->fd_off);
- memmove(tip->fd_buf + tip->fd_off, tip->fd_buf, wrap_size);
- }
+ if (!ret)
+ continue;
+ else if (ret < 0) {
+ if (errno != EAGAIN) {
+ perror(tip->fn);
+ fprintf(stderr, "server: failed read\n");
+ return 0;
+ }
+ continue;
+ } else {
+ buf += ret;
+ bytes_left -= ret;
+ }
+ } while (!is_done() && bytes_left);
- tip->fd_max_size <<= 1;
- return 0;
+ return len - bytes_left;
}
-static int __refill_ringbuffer(struct thread_information *tip, unsigned int len,
- int block)
+static inline struct tip_subbuf *
+subbuf_fifo_dequeue(struct thread_information *tip)
{
- unsigned long off;
- int ret;
+ const int head = tip->fifo.head;
+ const int next = (head + 1) & (FIFO_SIZE - 1);
- off = (tip->fd_size + tip->fd_off) & (tip->fd_max_size - 1);
- if (off + len > tip->fd_max_size)
- len = tip->fd_max_size - off;
+ if (head != tip->fifo.tail) {
+ struct tip_subbuf *ts = tip->fifo.q[head];
- assert(len > 0);
+ store_barrier();
+ tip->fifo.head = next;
+ return ts;
+ }
- ret = __read_data(tip, tip->fd_buf + off, len, block);
- if (ret < 0)
- return -1;
+ return NULL;
+}
- tip->fd_size += ret;
- if (ret == len)
+static inline int subbuf_fifo_queue(struct thread_information *tip,
+ struct tip_subbuf *ts)
+{
+ const int tail = tip->fifo.tail;
+ const int next = (tail + 1) & (FIFO_SIZE - 1);
+
+ if (next != tip->fifo.head) {
+ tip->fifo.q[tail] = ts;
+ store_barrier();
+ tip->fifo.tail = next;
return 0;
+ }
+ fprintf(stderr, "fifo too small!\n");
return 1;
}
/*
- * keep filling ring until we get a short read
+ * For file output, truncate and mmap the file appropriately
*/
-static void refill_ringbuffer(struct thread_information *tip, int block)
+static int mmap_subbuf(struct thread_information *tip, unsigned int maxlen)
{
- int len = buf_size;
+ int ofd = fileno(tip->ofile);
int ret;
-
- do {
- if (len + tip->fd_size > tip->fd_max_size)
- resize_ringbuffer(tip);
-
- ret = __refill_ringbuffer(tip, len, block);
- } while ((ret = len) && !is_done());
-}
-
-static int read_data(struct thread_information *tip, void *buf, int len)
-{
- unsigned int start_size, end_size;
-
- refill_ringbuffer(tip, len > tip->fd_size);
-
- if (len > tip->fd_size)
- return -1;
+ unsigned long nr;
/*
- * see if we wrap the ring
+ * extend file, if we have to. use chunks of 16 subbuffers.
*/
- start_size = len;
- end_size = 0;
- if (len > (tip->fd_max_size - tip->fd_off)) {
- start_size = tip->fd_max_size - tip->fd_off;
- end_size = len - start_size;
- }
-
- memcpy(buf, tip->fd_buf + tip->fd_off, start_size);
- if (end_size)
- memcpy(buf + start_size, tip->fd_buf, end_size);
-
- tip->fd_off = (tip->fd_off + len) & (tip->fd_max_size - 1);
- tip->fd_size -= len;
- return 0;
-}
+ if (tip->fs_off + maxlen > tip->fs_buf_len) {
+ if (tip->fs_buf) {
+ munlock(tip->fs_buf, tip->fs_buf_len);
+ munmap(tip->fs_buf, tip->fs_buf_len);
+ tip->fs_buf = NULL;
+ }
-static int write_data(FILE *file, void *buf, unsigned int buf_len)
-{
- int ret, bytes_left;
- char *p = buf;
+ tip->fs_off = tip->fs_size & (tip->device->page_size - 1);
+ nr = max(16, tip->device->buf_nr);
+ tip->fs_buf_len = (nr * tip->device->buf_size) - tip->fs_off;
+ tip->fs_max_size += tip->fs_buf_len;
- bytes_left = buf_len;
- while (bytes_left > 0) {
- ret = fwrite(p, bytes_left, 1, file);
- if (ret == 1)
- break;
+ if (ftruncate(ofd, tip->fs_max_size) < 0) {
+ perror("ftruncate");
+ return -1;
+ }
- if (ret < 0) {
- perror("write");
- return 1;
+ tip->fs_buf = mmap(NULL, tip->fs_buf_len, PROT_WRITE,
+ MAP_SHARED, ofd, tip->fs_size - tip->fs_off);
+ if (tip->fs_buf == MAP_FAILED) {
+ perror("mmap");
+ return -1;
}
+ mlock(tip->fs_buf, tip->fs_buf_len);
}
- return 0;
-}
-
-static void *extract_data(struct thread_information *tip, int nb)
-{
- unsigned char *buf;
-
- buf = malloc(nb);
- if (!read_data(tip, buf, nb))
- return buf;
+ ret = tip->read_data(tip, tip->fs_buf + tip->fs_off, maxlen);
+ if (ret >= 0) {
+ tip->data_read += ret;
+ tip->fs_size += ret;
+ tip->fs_off += ret;
+ return 0;
+ }
- free(buf);
- return NULL;
+ return -1;
}
/*
- * trace may start inside 'bit' or may need to be gotten further on
+ * Use the copy approach for pipes and network
*/
-static int get_event_slow(struct thread_information *tip,
- struct blk_io_trace *bit)
+static int get_subbuf(struct thread_information *tip, unsigned int maxlen)
{
- const int inc = sizeof(__u32);
- struct blk_io_trace foo;
- unsigned int offset;
- void *p;
-
- /*
- * check if trace is inside
- */
- offset = 0;
- p = bit;
- while (offset < sizeof(*bit)) {
- p += inc;
- offset += inc;
+ struct tip_subbuf *ts = malloc(sizeof(*ts));
+ int ret;
- memcpy(&foo, p, inc);
+ ts->buf = malloc(tip->device->buf_size);
+ ts->max_len = maxlen;
- if (CHECK_MAGIC(&foo))
- break;
+ ret = tip->read_data(tip, ts->buf, ts->max_len);
+ if (ret > 0) {
+ ts->len = ret;
+ tip->data_read += ret;
+ if (subbuf_fifo_queue(tip, ts))
+ ret = -1;
}
- /*
- * part trace found inside, read the rest
- */
- if (offset < sizeof(*bit)) {
- int good_bytes = sizeof(*bit) - offset;
-
- memmove(bit, p, good_bytes);
- p = (void *) bit + good_bytes;
-
- return read_data(tip, p, offset);
+ if (ret <= 0) {
+ free(ts->buf);
+ free(ts);
}
- /*
- * nothing found, keep looking for start of trace
- */
- do {
- if (read_data(tip, bit, sizeof(bit->magic)))
- return -1;
- } while (!CHECK_MAGIC(bit));
-
- /*
- * now get the rest of it
- */
- p = &bit->sequence;
- if (read_data(tip, p, sizeof(*bit) - inc))
- return -1;
-
- return 0;
-}
-
-/*
- * Sometimes relayfs screws us a little, if an event crosses a sub buffer
- * boundary. So keep looking forward in the trace data until an event
- * is found
- */
-static int get_event(struct thread_information *tip, struct blk_io_trace *bit)
-{
- /*
- * optimize for the common fast case, a full trace read that
- * succeeds
- */
- if (read_data(tip, bit, sizeof(*bit)))
- return -1;
-
- if (CHECK_MAGIC(bit))
- return 0;
-
- /*
- * ok that didn't work, the event may start somewhere inside the
- * trace itself
- */
- return get_event_slow(tip, bit);
-}
-
-static inline void tip_fd_unlock(struct thread_information *tip)
-{
- if (tip->fd_lock)
- pthread_mutex_unlock(tip->fd_lock);
-}
-
-static inline void tip_fd_lock(struct thread_information *tip)
-{
- if (tip->fd_lock)
- pthread_mutex_lock(tip->fd_lock);
+ return ret;
}
static void close_thread(struct thread_information *tip)
tip->fd_buf = NULL;
}
-static void *extract(void *arg)
+static void tip_ftrunc_final(struct thread_information *tip)
+{
+ /*
+ * truncate to right size and cleanup mmap
+ */
+ if (tip->ofile_mmap && tip->ofile) {
+ int ofd = fileno(tip->ofile);
+
+ if (tip->fs_buf)
+ munmap(tip->fs_buf, tip->fs_buf_len);
+
+ ftruncate(ofd, tip->fs_size);
+ }
+}
+
+static void *thread_main(void *arg)
{
struct thread_information *tip = arg;
- int pdu_len;
- char *pdu_data;
- struct blk_io_trace t;
pid_t pid = getpid();
cpu_set_t cpu_mask;
}
snprintf(tip->fn, sizeof(tip->fn), "%s/block/%s/trace%d",
- relay_path, tip->device->buts_name, tip->cpu);
- tip->fd = open(tip->fn, O_RDONLY | O_NONBLOCK);
+ debugfs_path, tip->device->buts_name, tip->cpu);
+ tip->fd = open(tip->fn, O_RDONLY);
if (tip->fd < 0) {
perror(tip->fn);
fprintf(stderr,"Thread %d failed open of %s\n", tip->cpu,
exit_trace(1);
}
- /*
- * start with a ringbuffer that is twice the size of the kernel side
- */
- tip->fd_max_size = buf_size * buf_nr * RING_INIT_NR;
- tip->fd_buf = malloc(tip->fd_max_size);
- tip->fd_off = 0;
- tip->fd_size = 0;
-
- pdu_data = NULL;
- while (1) {
- if (get_event(tip, &t))
- break;
-
- if (verify_trace(&t))
- break;
-
- pdu_len = t.pdu_len;
-
- trace_to_be(&t);
-
- if (pdu_len) {
- pdu_data = extract_data(tip, pdu_len);
- if (!pdu_data)
- break;
- }
-
- /*
- * now we have both trace and payload, get a lock on the
- * output descriptor and send it off
- */
- tip_fd_lock(tip);
-
- if (write_data(tip->ofile, &t, sizeof(t))) {
- tip_fd_unlock(tip);
- break;
- }
-
- if (pdu_data && write_data(tip->ofile, pdu_data, pdu_len)) {
- tip_fd_unlock(tip);
+ while (!is_done()) {
+ if (tip->get_subbuf(tip, tip->device->buf_size) < 0)
break;
- }
-
- tip_fd_unlock(tip);
-
- if (pdu_data) {
- free(pdu_data);
- pdu_data = NULL;
- }
-
- tip->events_processed++;
}
- close_thread(tip);
+ /*
+ * trace is stopped, pull data until we get a short read
+ */
+ while (tip->get_subbuf(tip, tip->device->buf_size) > 0)
+ ;
+
+ tip_ftrunc_final(tip);
+ tip->exited = 1;
return NULL;
}
-static int start_threads(struct device_information *dip)
+static int write_data_net(int fd, void *buf, unsigned int buf_len)
{
- struct thread_information *tip;
- char op[64];
- int j, pipeline = output_name && !strcmp(output_name, "-");
- int len, mode;
-
- for_each_tip(dip, tip, j) {
- tip->cpu = j;
- tip->device = dip;
- tip->fd_lock = NULL;
- tip->events_processed = 0;
-
- if (pipeline) {
- tip->ofile = fdopen(STDOUT_FILENO, "w");
- tip->fd_lock = &stdout_mutex;
- mode = _IOLBF;
- buf_size = 512;
- } else {
- len = 0;
-
- if (output_dir)
- len = sprintf(op, "%s/", output_dir);
-
- if (output_name) {
- sprintf(op + len, "%s.blktrace.%d", output_name,
- tip->cpu);
- } else {
- sprintf(op + len, "%s.blktrace.%d",
- dip->buts_name, tip->cpu);
- }
- tip->ofile = fopen(op, "w");
- mode = _IOFBF;
- buf_size = OFILE_BUF;
- }
-
- if (tip->ofile == NULL) {
- perror(op);
- return 1;
- }
+ unsigned int bytes_left = buf_len;
+ int ret;
- tip->ofile_buffer = malloc(buf_size);
- if (setvbuf(tip->ofile, tip->ofile_buffer, mode, buf_size)) {
- perror("setvbuf");
- close_thread(tip);
+ while (bytes_left) {
+ ret = send(fd, buf, bytes_left, 0);
+ if (ret < 0) {
+ perror("send");
return 1;
}
- if (pthread_create(&tip->thread, NULL, extract, tip)) {
- perror("pthread_create");
- close_thread(tip);
- return 1;
- }
+ buf += ret;
+ bytes_left -= ret;
}
return 0;
}
-static void stop_threads(struct device_information *dip)
+static int net_send_header(struct thread_information *tip, unsigned int len)
{
- struct thread_information *tip;
- unsigned long ret;
- int i;
-
- for_each_tip(dip, tip, i)
- (void) pthread_join(tip->thread, (void *) &ret);
+ struct blktrace_net_hdr hdr;
+
+ hdr.magic = BLK_IO_TRACE_MAGIC;
+ strcpy(hdr.buts_name, tip->device->buts_name);
+ hdr.cpu = tip->cpu;
+ hdr.max_cpus = ncpus;
+ hdr.len = len;
+ hdr.cl_id = getpid();
+ hdr.buf_size = tip->device->buf_size;
+ hdr.buf_nr = tip->device->buf_nr;
+ hdr.page_size = tip->device->page_size;
+
+ return write_data_net(net_out_fd[tip->cpu], &hdr, sizeof(hdr));
}
-static void stop_all_threads(void)
+/*
+ * send header with 0 length to signal end-of-run
+ */
+static void net_client_send_close(void)
{
struct device_information *dip;
+ struct blktrace_net_hdr hdr;
int i;
- for_each_dip(dip, i)
- stop_threads(dip);
+ for_each_dip(dip, i) {
+ hdr.magic = BLK_IO_TRACE_MAGIC;
+ hdr.max_cpus = ncpus;
+ hdr.len = 0;
+ strcpy(hdr.buts_name, dip->buts_name);
+ hdr.cpu = get_dropped_count(dip->buts_name);
+ hdr.cl_id = getpid();
+ hdr.buf_size = dip->buf_size;
+ hdr.buf_nr = dip->buf_nr;
+ hdr.page_size = dip->page_size;
+
+ write_data_net(net_out_fd[0], &hdr, sizeof(hdr));
+ }
+
}
-static void stop_all_tracing(void)
+static int flush_subbuf_net(struct thread_information *tip,
+ struct tip_subbuf *ts)
{
- struct device_information *dip;
- int i;
+ if (net_send_header(tip, ts->len))
+ return -1;
+ if (write_data_net(net_out_fd[tip->cpu], ts->buf, ts->len))
+ return -1;
- for_each_dip(dip, i)
- stop_trace(dip);
+ free(ts->buf);
+ free(ts);
+ return 1;
}
-static void exit_trace(int status)
+static int net_sendfile(struct thread_information *tip, struct tip_subbuf *ts)
{
- if (!is_trace_stopped()) {
- trace_stopped = 1;
- stop_all_threads();
- stop_all_tracing();
+ int ret = sendfile(net_out_fd[tip->cpu], tip->fd, NULL, ts->len);
+
+ if (ret < 0) {
+ perror("sendfile");
+ return 1;
+ } else if (ret < (int) ts->len) {
+ fprintf(stderr, "short sendfile send (%d of %d)\n", ret, ts->len);
+ return 1;
+ }
+
+ return 0;
+}
+
+static int flush_subbuf_sendfile(struct thread_information *tip,
+ struct tip_subbuf *ts)
+{
+ int ret = -1;
+
+ if (net_send_header(tip, ts->len))
+ goto err;
+ if (net_sendfile(tip, ts))
+ goto err;
+
+ tip->data_read += ts->len;
+ ret = 1;
+err:
+ free(ts);
+ return ret;
+}
+
+static int get_subbuf_sendfile(struct thread_information *tip,
+ __attribute__((__unused__)) unsigned int maxlen)
+{
+ struct tip_subbuf *ts;
+ struct stat sb;
+ unsigned int ready;
+
+ wait_for_data(tip, -1);
+
+ if (fstat(tip->fd, &sb) < 0) {
+ perror("trace stat");
+ return -1;
+ }
+
+ ready = sb.st_size - tip->data_queued;
+ if (!ready) {
+ usleep(1000);
+ return 0;
+ }
+
+ ts = malloc(sizeof(*ts));
+ ts->buf = NULL;
+ ts->max_len = 0;
+ ts->len = ready;
+ tip->data_queued += ready;
+
+ if (flush_subbuf_sendfile(tip, ts) < 0)
+ return -1;
+
+ return ready;
+}
+
+static int write_data(struct thread_information *tip, void *buf,
+ unsigned int buf_len)
+{
+ int ret;
+
+ if (!buf_len)
+ return 0;
+
+ ret = fwrite(buf, buf_len, 1, tip->ofile);
+ if (ferror(tip->ofile) || ret != 1) {
+ perror("fwrite");
+ clearerr(tip->ofile);
+ return 1;
+ }
+
+ if (tip->ofile_stdout)
+ fflush(tip->ofile);
+
+ return 0;
+}
+
+static int flush_subbuf_file(struct thread_information *tip,
+ struct tip_subbuf *ts)
+{
+ unsigned int offset = 0;
+ struct blk_io_trace *t;
+ int pdu_len, events = 0;
+
+ /*
+ * surplus from last run
+ */
+ if (tip->leftover_ts) {
+ struct tip_subbuf *prev_ts = tip->leftover_ts;
+
+ if (prev_ts->len + ts->len > prev_ts->max_len) {
+ prev_ts->max_len += ts->len;
+ prev_ts->buf = realloc(prev_ts->buf, prev_ts->max_len);
+ }
+
+ memcpy(prev_ts->buf + prev_ts->len, ts->buf, ts->len);
+ prev_ts->len += ts->len;
+
+ free(ts->buf);
+ free(ts);
+
+ ts = prev_ts;
+ tip->leftover_ts = NULL;
+ }
+
+ while (offset + sizeof(*t) <= ts->len) {
+ t = ts->buf + offset;
+
+ if (verify_trace(t)) {
+ write_data(tip, ts->buf, offset);
+ return -1;
+ }
+
+ pdu_len = t->pdu_len;
+
+ if (offset + sizeof(*t) + pdu_len > ts->len)
+ break;
+
+ offset += sizeof(*t) + pdu_len;
+ tip->events_processed++;
+ tip->data_read += sizeof(*t) + pdu_len;
+ events++;
+ }
+
+ if (write_data(tip, ts->buf, offset))
+ return -1;
+
+ /*
+ * leftover bytes, save them for next time
+ */
+ if (offset != ts->len) {
+ tip->leftover_ts = ts;
+ ts->len -= offset;
+ memmove(ts->buf, ts->buf + offset, ts->len);
+ } else {
+ free(ts->buf);
+ free(ts);
+ }
+
+ return events;
+}
+
+static int write_tip_events(struct thread_information *tip)
+{
+ struct tip_subbuf *ts = subbuf_fifo_dequeue(tip);
+
+ if (ts)
+ return tip->flush_subbuf(tip, ts);
+
+ return 0;
+}
+
+/*
+ * scans the tips we know and writes out the subbuffers we accumulate
+ */
+static void get_and_write_events(void)
+{
+ struct device_information *dip;
+ struct thread_information *tip;
+ int i, j, events, ret, tips_running;
+
+ while (!is_done()) {
+ events = 0;
+
+ for_each_dip(dip, i) {
+ for_each_tip(dip, tip, j) {
+ ret = write_tip_events(tip);
+ if (ret > 0)
+ events += ret;
+ }
+ }
+
+ if (!events)
+ usleep(100000);
+ }
+
+ /*
+ * reap stored events
+ */
+ do {
+ events = 0;
+ tips_running = 0;
+ for_each_dip(dip, i) {
+ for_each_tip(dip, tip, j) {
+ ret = write_tip_events(tip);
+ if (ret > 0)
+ events += ret;
+ tips_running += !tip->exited;
+ }
+ }
+ usleep(10);
+ } while (events || tips_running);
+}
+
+static void wait_for_threads(void)
+{
+ /*
+ * for piped or network output, poll and fetch data for writeout.
+ * for files, we just wait around for trace threads to exit
+ */
+ if ((output_name && !strcmp(output_name, "-")) ||
+ ((net_mode == Net_client) && !net_use_sendfile))
+ get_and_write_events();
+ else {
+ struct device_information *dip;
+ struct thread_information *tip;
+ int i, j, tips_running;
+
+ do {
+ tips_running = 0;
+ usleep(100000);
+
+ for_each_dip(dip, i)
+ for_each_tip(dip, tip, j)
+ tips_running += !tip->exited;
+ } while (tips_running);
+ }
+
+ if (net_mode == Net_client)
+ net_client_send_close();
+}
+
+static int fill_ofname(struct device_information *dip,
+ struct thread_information *tip, char *dst,
+ char *buts_name)
+{
+ struct stat sb;
+ int len = 0;
+
+ if (output_dir)
+ len = sprintf(dst, "%s/", output_dir);
+ else
+ len = sprintf(dst, "./");
+
+ if (net_mode == Net_server) {
+ struct net_connection *nc = tip->nc;
+
+ len += sprintf(dst + len, "%s-", inet_ntoa(nc->ch->cl_in_addr));
+ len += strftime(dst + len, 64, "%F-%T/", gmtime(&dip->cl_connect_time));
+ }
+
+ if (stat(dst, &sb) < 0) {
+ if (errno != ENOENT) {
+ perror("stat");
+ return 1;
+ }
+ if (mkdir(dst, 0755) < 0) {
+ perror(dst);
+ fprintf(stderr, "Can't make output dir\n");
+ return 1;
+ }
+ }
+
+ if (output_name)
+ sprintf(dst + len, "%s.blktrace.%d", output_name, tip->cpu);
+ else
+ sprintf(dst + len, "%s.blktrace.%d", buts_name, tip->cpu);
+
+ return 0;
+}
+
+static void fill_ops(struct thread_information *tip)
+{
+ /*
+ * setup ops
+ */
+ if (net_mode == Net_client) {
+ if (net_use_sendfile) {
+ tip->get_subbuf = get_subbuf_sendfile;
+ tip->flush_subbuf = NULL;
+ } else {
+ tip->get_subbuf = get_subbuf;
+ tip->flush_subbuf = flush_subbuf_net;
+ }
+ } else {
+ if (tip->ofile_mmap)
+ tip->get_subbuf = mmap_subbuf;
+ else
+ tip->get_subbuf = get_subbuf;
+
+ tip->flush_subbuf = flush_subbuf_file;
+ }
+
+ if (net_mode == Net_server)
+ tip->read_data = read_data_net;
+ else
+ tip->read_data = read_data_file;
+}
+
+static int tip_open_output(struct device_information *dip,
+ struct thread_information *tip)
+{
+ int pipeline = output_name && !strcmp(output_name, "-");
+ int mode, vbuf_size;
+ char op[128];
+
+ if (net_mode == Net_client) {
+ tip->ofile = NULL;
+ tip->ofile_stdout = 0;
+ tip->ofile_mmap = 0;
+ goto done;
+ } else if (pipeline) {
+ tip->ofile = fdopen(STDOUT_FILENO, "w");
+ tip->ofile_stdout = 1;
+ tip->ofile_mmap = 0;
+ mode = _IOLBF;
+ vbuf_size = 512;
+ } else {
+ if (fill_ofname(dip, tip, op, dip->buts_name))
+ return 1;
+ tip->ofile = fopen(op, "w+");
+ tip->ofile_stdout = 0;
+ tip->ofile_mmap = 1;
+ mode = _IOFBF;
+ vbuf_size = OFILE_BUF;
+ }
+
+ if (tip->ofile == NULL) {
+ perror(op);
+ return 1;
+ }
+
+ tip->ofile_buffer = malloc(vbuf_size);
+ if (setvbuf(tip->ofile, tip->ofile_buffer, mode, vbuf_size)) {
+ perror("setvbuf");
+ close_thread(tip);
+ return 1;
+ }
+
+done:
+ fill_ops(tip);
+ return 0;
+}
+
+static int start_threads(struct device_information *dip)
+{
+ struct thread_information *tip;
+ int j;
+
+ for_each_tip(dip, tip, j) {
+ tip->cpu = j;
+ tip->device = dip;
+ tip->events_processed = 0;
+ tip->fd = -1;
+ memset(&tip->fifo, 0, sizeof(tip->fifo));
+ tip->leftover_ts = NULL;
+
+ if (tip_open_output(dip, tip))
+ return 1;
+
+ if (pthread_create(&tip->thread, NULL, thread_main, tip)) {
+ perror("pthread_create");
+ close_thread(tip);
+ return 1;
+ }
+ }
+
+ return 0;
+}
+
+static void stop_threads(struct device_information *dip)
+{
+ struct thread_information *tip;
+ unsigned long ret;
+ int i;
+
+ for_each_tip(dip, tip, i) {
+ (void) pthread_join(tip->thread, (void *) &ret);
+ close_thread(tip);
+ }
+}
+
+static void stop_all_threads(void)
+{
+ struct device_information *dip;
+ int i;
+
+ for_each_dip(dip, i)
+ stop_threads(dip);
+}
+
+static void stop_all_tracing(void)
+{
+ struct device_information *dip;
+ int i;
+
+ for_each_dip(dip, i)
+ stop_trace(dip);
+}
+
+static void exit_trace(int status)
+{
+ if (!is_trace_stopped()) {
+ trace_stopped = 1;
+ stop_all_threads();
+ stop_all_tracing();
}
exit(status);
perror(dip->path);
return 1;
}
+ dip->buf_size = buf_size;
+ dip->buf_nr = buf_nr;
+ dip->page_size = page_size;
}
return 0;
}
if (i != ndevs) {
- __for_each_dip(dip, j, i)
+ __for_each_dip(dip, device_information, i, j)
stop_trace(dip);
return 1;
}
if (i != ndevs) {
- __for_each_dip(dip, j, i)
+ __for_each_dip(dip, device_information, i, j)
stop_threads(dip);
for_each_dip(dip, i)
stop_trace(dip);
return 0;
}
-static void show_stats(void)
+static void show_stats(struct device_information *dips, int ndips, int cpus)
{
- int i, j, no_stdout = 0;
struct device_information *dip;
struct thread_information *tip;
- unsigned long long events_processed;
+ unsigned long long events_processed, data_read;
unsigned long total_drops;
+ int i, j, no_stdout = 0;
if (is_stat_shown())
return;
- stat_shown = 1;
-
if (output_name && !strcmp(output_name, "-"))
no_stdout = 1;
+ stat_shown = 1;
+
total_drops = 0;
- for_each_dip(dip, i) {
+ __for_each_dip(dip, dips, ndips, i) {
if (!no_stdout)
printf("Device: %s\n", dip->path);
events_processed = 0;
- for_each_tip(dip, tip, j) {
+ data_read = 0;
+ __for_each_tip(dip, tip, cpus, j) {
if (!no_stdout)
- printf(" CPU%3d: %20ld events\n",
- tip->cpu, tip->events_processed);
+ printf(" CPU%3d: %20lu events, %8llu KiB data\n",
+ tip->cpu, tip->events_processed,
+ (tip->data_read + 1023) >> 10);
events_processed += tip->events_processed;
+ data_read += tip->data_read;
}
total_drops += dip->drop_count;
if (!no_stdout)
- printf(" Total: %20lld events (dropped %lu)\n",
- events_processed, dip->drop_count);
+ printf(" Total: %20llu events (dropped %lu), %8llu KiB data\n",
+ events_processed, dip->drop_count,
+ (data_read + 1023) >> 10);
}
if (total_drops)
fprintf(stderr, "You have dropped events, consider using a larger buffer size (-b)\n");
}
+static struct device_information *net_get_dip(struct net_connection *nc,
+ struct blktrace_net_hdr *bnh)
+{
+ struct device_information *dip, *cl_dip = NULL;
+ struct cl_host *ch = nc->ch;
+ int i;
+
+ for (i = 0; i < ch->ndevs; i++) {
+ dip = &ch->device_information[i];
+
+ if (!strcmp(dip->buts_name, bnh->buts_name))
+ return dip;
+
+ if (dip->cl_id == bnh->cl_id)
+ cl_dip = dip;
+ }
+
+ ch->device_information = realloc(ch->device_information, (ch->ndevs + 1) * sizeof(*dip));
+ dip = &ch->device_information[ch->ndevs];
+ memset(dip, 0, sizeof(*dip));
+ dip->fd = -1;
+ dip->ch = ch;
+ dip->cl_id = bnh->cl_id;
+ dip->buf_size = bnh->buf_size;
+ dip->buf_nr = bnh->buf_nr;
+ dip->page_size = bnh->page_size;
+
+ if (cl_dip)
+ dip->cl_connect_time = cl_dip->cl_connect_time;
+ else
+ dip->cl_connect_time = nc->connect_time;
+ strcpy(dip->buts_name, bnh->buts_name);
+ dip->path = strdup(bnh->buts_name);
+ dip->trace_started = 1;
+ ch->ndevs++;
+ dip->threads = malloc(nc->ncpus * sizeof(struct thread_information));
+ memset(dip->threads, 0, nc->ncpus * sizeof(struct thread_information));
+
+ /*
+ * open all files
+ */
+ for (i = 0; i < nc->ncpus; i++) {
+ struct thread_information *tip = &dip->threads[i];
+
+ tip->cpu = i;
+ tip->device = dip;
+ tip->fd = -1;
+ tip->nc = nc;
+
+ if (tip_open_output(dip, tip))
+ return NULL;
+
+ tip->nc = NULL;
+ }
+
+ return dip;
+}
+
+static struct thread_information *net_get_tip(struct net_connection *nc,
+ struct blktrace_net_hdr *bnh)
+{
+ struct device_information *dip;
+ struct thread_information *tip;
+
+ dip = net_get_dip(nc, bnh);
+ if (!dip->trace_started) {
+ fprintf(stderr, "Events for closed devices %s\n", dip->buts_name);
+ return NULL;
+ }
+
+ tip = &dip->threads[bnh->cpu];
+ if (!tip->nc)
+ tip->nc = nc;
+
+ return tip;
+}
+
+static int net_get_header(struct net_connection *nc,
+ struct blktrace_net_hdr *bnh)
+{
+ int fl = fcntl(nc->in_fd, F_GETFL);
+ int bytes_left, ret;
+ void *p = bnh;
+
+ fcntl(nc->in_fd, F_SETFL, fl | O_NONBLOCK);
+ bytes_left = sizeof(*bnh);
+ while (bytes_left && !is_done()) {
+ ret = recv(nc->in_fd, p, bytes_left, MSG_WAITALL);
+ if (ret < 0) {
+ if (errno != EAGAIN) {
+ perror("recv header");
+ return 1;
+ }
+ usleep(1000);
+ continue;
+ } else if (!ret) {
+ usleep(1000);
+ continue;
+ } else {
+ p += ret;
+ bytes_left -= ret;
+ }
+ }
+ fcntl(nc->in_fd, F_SETFL, fl & ~O_NONBLOCK);
+ return bytes_left;
+}
+
+/*
+ * finalize a net client: truncate files, show stats, cleanup, etc
+ */
+static void device_done(struct net_connection *nc, struct device_information *dip)
+{
+ struct thread_information *tip;
+ int i;
+
+ __for_each_tip(dip, tip, nc->ncpus, i)
+ tip_ftrunc_final(tip);
+
+ show_stats(dip, 1, nc->ncpus);
+
+ /*
+ * cleanup for next run
+ */
+ __for_each_tip(dip, tip, nc->ncpus, i) {
+ if (tip->ofile)
+ fclose(tip->ofile);
+ }
+
+ free(dip->threads);
+ free(dip->path);
+
+ close(nc->in_fd);
+ nc->in_fd = -1;
+
+ stat_shown = 0;
+}
+
+static inline int in_addr_eq(struct in_addr a, struct in_addr b)
+{
+ return a.s_addr == b.s_addr;
+}
+
+static void net_add_client_host(struct cl_host *ch)
+{
+ ch->list_next = cl_host_list;
+ cl_host_list = ch;
+ cl_hosts++;
+}
+
+static void net_remove_client_host(struct cl_host *ch)
+{
+ struct cl_host *p, *c;
+
+ for (p = c = cl_host_list; c; c = c->list_next) {
+ if (c == ch) {
+ if (p == c)
+ cl_host_list = c->list_next;
+ else
+ p->list_next = c->list_next;
+ cl_hosts--;
+ return;
+ }
+ p = c;
+ }
+}
+
+static struct cl_host *net_find_client_host(struct in_addr cl_in_addr)
+{
+ struct cl_host *ch = cl_host_list;
+
+ while (ch) {
+ if (in_addr_eq(ch->cl_in_addr, cl_in_addr))
+ return ch;
+ ch = ch->list_next;
+ }
+
+ return NULL;
+}
+
+static void net_client_host_done(struct cl_host *ch)
+{
+ free(ch->device_information);
+ free(ch->net_connections);
+ net_connects -= ch->nconn;
+ net_remove_client_host(ch);
+ free(ch);
+}
+
+/*
+ * handle incoming events from a net client
+ */
+static int net_client_data(struct net_connection *nc)
+{
+ struct thread_information *tip;
+ struct blktrace_net_hdr bnh;
+
+ if (net_get_header(nc, &bnh))
+ return 1;
+
+ if (data_is_native == -1 && check_data_endianness(bnh.magic)) {
+ fprintf(stderr, "server: received data is bad\n");
+ return 1;
+ }
+
+ if (!data_is_native) {
+ bnh.magic = be32_to_cpu(bnh.magic);
+ bnh.cpu = be32_to_cpu(bnh.cpu);
+ bnh.max_cpus = be32_to_cpu(bnh.max_cpus);
+ bnh.len = be32_to_cpu(bnh.len);
+ bnh.cl_id = be32_to_cpu(bnh.cl_id);
+ bnh.buf_size = be32_to_cpu(bnh.buf_size);
+ bnh.buf_nr = be32_to_cpu(bnh.buf_nr);
+ bnh.page_size = be32_to_cpu(bnh.page_size);
+ }
+
+ if ((bnh.magic & 0xffffff00) != BLK_IO_TRACE_MAGIC) {
+ fprintf(stderr, "server: bad data magic\n");
+ return 1;
+ }
+
+ if (nc->ncpus == -1)
+ nc->ncpus = bnh.max_cpus;
+
+ /*
+ * len == 0 means that the other end signalled end-of-run
+ */
+ if (!bnh.len) {
+ /*
+ * overload cpu count with dropped events
+ */
+ struct device_information *dip;
+
+ dip = net_get_dip(nc, &bnh);
+ dip->drop_count = bnh.cpu;
+ dip->trace_started = 0;
+
+ printf("server: end of run for %s\n", dip->buts_name);
+
+ device_done(nc, dip);
+
+ if (++nc->ch->ndevs_done == nc->ch->ndevs)
+ net_client_host_done(nc->ch);
+
+ return 0;
+ }
+
+ tip = net_get_tip(nc, &bnh);
+ if (!tip)
+ return 1;
+
+ if (mmap_subbuf(tip, bnh.len))
+ return 1;
+
+ return 0;
+}
+
+static void net_add_connection(int listen_fd, struct sockaddr_in *addr)
+{
+ socklen_t socklen = sizeof(*addr);
+ struct net_connection *nc;
+ struct cl_host *ch;
+ int in_fd;
+
+ in_fd = accept(listen_fd, (struct sockaddr *) addr, &socklen);
+ if (in_fd < 0) {
+ perror("accept");
+ return;
+ }
+
+ ch = net_find_client_host(addr->sin_addr);
+ if (!ch) {
+ if (cl_hosts == NET_MAX_CL_HOSTS) {
+ fprintf(stderr, "server: no more clients allowed\n");
+ return;
+ }
+ ch = malloc(sizeof(struct cl_host));
+ memset(ch, 0, sizeof(*ch));
+ ch->cl_in_addr = addr->sin_addr;
+ net_add_client_host(ch);
+
+ printf("server: connection from %s\n", inet_ntoa(addr->sin_addr));
+ }
+
+ ch->net_connections = realloc(ch->net_connections, (ch->nconn + 1) * sizeof(*nc));
+ nc = &ch->net_connections[ch->nconn++];
+ memset(nc, 0, sizeof(*nc));
+
+ time(&nc->connect_time);
+ nc->ch = ch;
+ nc->in_fd = in_fd;
+ nc->ncpus = -1;
+ net_connects++;
+}
+
+/*
+ * event driven loop, handle new incoming connections and data from
+ * existing connections
+ */
+static void net_server_handle_connections(int listen_fd,
+ struct sockaddr_in *addr)
+{
+ struct pollfd *pfds = NULL;
+ struct net_connection **ncs = NULL;
+ int max_connects = 0;
+ int i, nconns, events;
+ struct cl_host *ch;
+ struct net_connection *nc;
+
+ printf("server: waiting for connections...\n");
+
+ while (!is_done()) {
+ if (net_connects >= max_connects) {
+ pfds = realloc(pfds, (net_connects + 1) * sizeof(*pfds));
+ ncs = realloc(ncs, (net_connects + 1) * sizeof(*ncs));
+ max_connects = net_connects + 1;
+ }
+ /*
+ * the zero entry is for incoming connections, remaining
+ * entries for clients
+ */
+ pfds[0].fd = listen_fd;
+ pfds[0].events = POLLIN;
+ nconns = 0;
+ for_each_cl_host(ch) {
+ for (i = 0; i < ch->nconn; i++) {
+ nc = &ch->net_connections[i];
+ pfds[nconns + 1].fd = nc->in_fd;
+ pfds[nconns + 1].events = POLLIN;
+ ncs[nconns++] = nc;
+ }
+ }
+
+ events = poll(pfds, 1 + nconns, -1);
+ if (events < 0) {
+ if (errno == EINTR)
+ continue;
+
+ perror("poll");
+ break;
+ } else if (!events)
+ continue;
+
+ if (pfds[0].revents & POLLIN) {
+ net_add_connection(listen_fd, addr);
+ events--;
+ }
+
+ for (i = 0; events && i < nconns; i++) {
+ if (pfds[i + 1].revents & POLLIN) {
+ net_client_data(ncs[i]);
+ events--;
+ }
+ }
+ }
+}
+
+/*
+ * Start here when we are in server mode - just fetch data from the network
+ * and dump to files
+ */
+static int net_server(void)
+{
+ struct sockaddr_in addr;
+ int fd, opt;
+
+ fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd < 0) {
+ perror("server: socket");
+ return 1;
+ }
+
+ opt = 1;
+ if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
+ perror("setsockopt");
+ return 1;
+ }
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_addr.s_addr = htonl(INADDR_ANY);
+ addr.sin_port = htons(net_port);
+
+ if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
+ perror("bind");
+ return 1;
+ }
+
+ if (listen(fd, 1) < 0) {
+ perror("listen");
+ return 1;
+ }
+
+ net_server_handle_connections(fd, &addr);
+ return 0;
+}
+
+/*
+ * Setup outgoing network connection where we will transmit data
+ */
+static int net_setup_client_cpu(int i, struct sockaddr_in *addr)
+{
+ int fd;
+
+ fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd < 0) {
+ perror("client: socket");
+ return 1;
+ }
+
+ if (connect(fd, (struct sockaddr *) addr, sizeof(*addr)) < 0) {
+ perror("client: connect");
+ return 1;
+ }
+
+ net_out_fd[i] = fd;
+ return 0;
+}
+
+static int net_setup_client(void)
+{
+ struct sockaddr_in addr;
+ int i;
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(net_port);
+
+ if (inet_aton(hostname, &addr.sin_addr) != 1) {
+ struct hostent *hent = gethostbyname(hostname);
+ if (!hent) {
+ perror("gethostbyname");
+ return 1;
+ }
+
+ memcpy(&addr.sin_addr, hent->h_addr, 4);
+ strcpy(hostname, hent->h_name);
+ }
+
+ printf("blktrace: connecting to %s\n", hostname);
+
+ net_out_fd = malloc(ncpus * sizeof(*net_out_fd));
+ for (i = 0; i < ncpus; i++) {
+ if (net_setup_client_cpu(i, &addr))
+ return 1;
+ }
+
+ printf("blktrace: connected!\n");
+
+ return 0;
+}
+
static char usage_str[] = \
- "-d <dev> [ -r relay path ] [ -o <output> ] [-k ] [ -w time ]\n" \
+ "-d <dev> [ -r debugfs path ] [ -o <output> ] [-k ] [ -w time ]\n" \
"[ -a action ] [ -A action mask ] [ -v ]\n\n" \
"\t-d Use specified device. May also be given last after options\n" \
- "\t-r Path to mounted relayfs, defaults to /relay\n" \
+ "\t-r Path to mounted debugfs, defaults to /debug\n" \
"\t-o File(s) to send output to\n" \
"\t-D Directory to prepend to output file names\n" \
"\t-k Kill a running trace\n" \
"\t-A Give trace mask as a single value. See documentation\n" \
"\t-b Sub buffer size in KiB\n" \
"\t-n Number of sub buffers\n" \
- "\t-v Print program version info\n\n";
+ "\t-l Run in network listen mode (blktrace server)\n" \
+ "\t-h Run in network client mode, connecting to the given host\n" \
+ "\t-p Network port to use (default 8462)\n" \
+ "\t-s Make the network client NOT use sendfile() to transfer data\n" \
+ "\t-V Print program version info\n\n";
static void show_usage(char *program)
{
fprintf(stderr, "Usage: %s %s %s",program, blktrace_version, usage_str);
}
-static void handle_sigint(__attribute__((__unused__)) int sig)
-{
- done = 1;
- if (!is_trace_stopped()) {
- trace_stopped = 1;
- stop_all_threads();
- stop_all_traces();
- }
-
- show_stats();
-}
int main(int argc, char *argv[])
{
- static char default_relay_path[] = "/relay";
+ static char default_debugfs_path[] = "/sys/kernel/debug";
struct statfs st;
int i, c;
int stop_watch = 0;
break;
case 'r':
- relay_path = optarg;
+ debugfs_path = optarg;
break;
case 'o':
case 'D':
output_dir = optarg;
break;
+ case 'h':
+ net_mode = Net_client;
+ strcpy(hostname, optarg);
+ break;
+ case 'l':
+ net_mode = Net_server;
+ break;
+ case 'p':
+ net_port = atoi(optarg);
+ break;
+ case 's':
+ net_use_sendfile = 0;
+ break;
default:
show_usage(argv[0]);
return 1;
}
}
+ setlocale(LC_NUMERIC, "en_US");
+
+ page_size = getpagesize();
+
+ if (net_mode == Net_server) {
+ if (output_name) {
+ fprintf(stderr, "-o ignored in server mode\n");
+ output_name = NULL;
+ }
+
+ return net_server();
+ }
+
while (optind < argc) {
if (resize_devices(argv[optind++]) != 0)
return 1;
return 1;
}
- if (!relay_path)
- relay_path = default_relay_path;
-
if (act_mask_tmp != 0)
act_mask = act_mask_tmp;
- if (statfs(relay_path, &st) < 0) {
+ if (!debugfs_path)
+ debugfs_path = default_debugfs_path;
+
+ if (statfs(debugfs_path, &st) < 0) {
perror("statfs");
fprintf(stderr,"%s does not appear to be a valid path\n",
- relay_path);
+ debugfs_path);
return 1;
- } else if (st.f_type != (long) RELAYFS_TYPE) {
- fprintf(stderr,"%s does not appear to be a relay filesystem\n",
- relay_path);
+ } else if (st.f_type != (long) DEBUGFS_TYPE) {
+ fprintf(stderr,"%s does not appear to be a debug filesystem\n",
+ debugfs_path);
return 1;
}
return 0;
}
- setlocale(LC_NUMERIC, "en_US");
-
ncpus = sysconf(_SC_NPROCESSORS_ONLN);
if (ncpus < 0) {
fprintf(stderr, "sysconf(_SC_NPROCESSORS_ONLN) failed\n");
return 1;
}
- if (start_devices() != 0)
- return 1;
-
signal(SIGINT, handle_sigint);
signal(SIGHUP, handle_sigint);
signal(SIGTERM, handle_sigint);
signal(SIGALRM, handle_sigint);
+ signal(SIGPIPE, SIG_IGN);
+
+ if (net_mode == Net_client && net_setup_client())
+ return 1;
+
+ if (start_devices() != 0)
+ return 1;
atexit(stop_all_tracing);
if (stop_watch)
alarm(stop_watch);
- while (!is_done())
- sleep(1);
+ wait_for_threads();
if (!is_trace_stopped()) {
trace_stopped = 1;
stop_all_traces();
}
- show_stats();
+ show_stats(device_information, ndevs, ncpus);
return 0;
}
projects / blktrace.git / blobdiff