void *buf;
unsigned int len;
unsigned int max_len;
- off_t offset;
};
#define FIFO_SIZE (1024) /* should be plenty big! */
void *fd_buf;
char fn[MAXPATHLEN + 64];
- int pfd;
- size_t *pfd_buf;
-
struct in_addr cl_in_addr;
FILE *ofile;
off_t ofile_offset;
int ofile_stdout;
int ofile_mmap;
+ volatile int sendfile_pending;
int (*get_subbuf)(struct thread_information *, unsigned int);
int (*flush_subbuf)(struct thread_information *, struct tip_subbuf *);
unsigned long events_processed;
unsigned long long data_read;
+ unsigned long long data_queued;
struct device_information *device;
int exited;
return atoi(tmp);
}
-static size_t get_subbuf_padding(struct thread_information *tip,
- unsigned subbuf)
-{
- size_t padding_size = buf_nr * sizeof(size_t);
- size_t ret;
-
- if (read(tip->pfd, tip->pfd_buf, padding_size) < 0) {
- perror("tip pad read");
- ret = -1;
- } else
- ret = tip->pfd_buf[subbuf];
-
- return ret;
-}
-
static int start_trace(struct device_information *dip)
{
struct blk_user_trace_setup buts;
{
struct tip_subbuf *ts;
struct stat sb;
- unsigned int ready, this_size, total;
+ unsigned int ready;
wait_for_data(tip);
if (is_done())
return get_subbuf(tip, maxlen);
+ if (tip->sendfile_pending) {
+ usleep(100);
+ return 0;
+ }
+
if (fstat(tip->fd, &sb) < 0) {
perror("trace stat");
return -1;
}
-
- ready = sb.st_size - tip->ofile_offset;
- if (!ready) {
- /*
- * delay a little, since poll() will return data available
- * until sendfile() is run
- */
- usleep(100);
+ ready = sb.st_size - tip->data_queued;
+ if (!ready)
return 0;
- }
- this_size = buf_size;
- total = ready;
- while (ready) {
- if (this_size > ready)
- this_size = ready;
+ ts = malloc(sizeof(*ts));
+ ts->buf = NULL;
+ ts->max_len = 0;
+ ts->len = ready;
+ tip->data_queued += ready;
- ts = malloc(sizeof(*ts));
-
- ts->buf = NULL;
- ts->max_len = 0;
-
- ts->len = this_size;
- ts->offset = tip->ofile_offset;
- tip->ofile_offset += ts->len;
-
- if (subbuf_fifo_queue(tip, ts))
- return -1;
-
- ready -= this_size;
- }
+ if (subbuf_fifo_queue(tip, ts))
+ return -1;
- return total;
+ tip->sendfile_pending++;
+ return ready;
}
static void close_thread(struct thread_information *tip)
{
if (tip->fd != -1)
close(tip->fd);
- if (tip->pfd != -1)
- close(tip->pfd);
if (tip->ofile)
fclose(tip->ofile);
if (tip->ofile_buffer)
free(tip->ofile_buffer);
if (tip->fd_buf)
free(tip->fd_buf);
- if (tip->pfd_buf)
- free(tip->pfd_buf);
tip->fd = -1;
- tip->pfd = -1;
tip->ofile = NULL;
tip->ofile_buffer = NULL;
tip->fd_buf = NULL;
exit_trace(1);
}
- if (net_mode == Net_client && net_use_sendfile) {
- char tmp[MAXPATHLEN + 64];
-
- snprintf(tmp, sizeof(tmp), "%s/block/%s/trace%d.padding",
- relay_path, tip->device->buts_name, tip->cpu);
-
- tip->pfd = open(tmp, O_RDONLY);
- if (tip->pfd < 0) {
- fprintf(stderr, "Couldn't open padding file %s\n", tmp);
- exit_trace(1);
- }
-
- tip->pfd_buf = malloc(buf_nr * sizeof(size_t));
- }
-
while (!is_done()) {
if (tip->get_subbuf(tip, buf_size) < 0)
break;
static int net_sendfile(struct thread_information *tip, struct tip_subbuf *ts)
{
- unsigned int bytes_left = ts->len;
- int ret;
+ int ret = sendfile(net_out_fd, tip->fd, NULL, ts->len);
- while (bytes_left && !is_done()) {
- ret = sendfile(net_out_fd, tip->fd, &ts->offset, bytes_left);
- if (ret < 0) {
- perror("sendfile");
- break;
- } else if (!ret) {
- usleep(100);
- continue;
- }
-
- ts->offset += ret;
- bytes_left -= ret;
+ 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 bytes_left;
+ return 0;
}
static int flush_subbuf_sendfile(struct thread_information *tip,
struct tip_subbuf *ts)
{
- size_t padding;
- unsigned subbuf;
+ int ret = 1;
/*
* currently we cannot use sendfile() on the last bytes read, as they
if (ts->buf)
return flush_subbuf_net(tip, ts);
- subbuf = (ts->offset / buf_size) % buf_nr;
- padding = get_subbuf_padding(tip, subbuf);
- ts->len -= padding;
-
if (net_send_header(tip, ts->len))
- return 1;
+ goto err;
if (net_sendfile(tip, ts))
- return 1;
+ goto err;
tip->data_read += ts->len;
+ tip->ofile_offset += buf_size;
+ ret = 0;
+err:
+ tip->sendfile_pending--;
free(ts);
- return 0;
+ return ret;
}
static int write_data(struct thread_information *tip, void *buf,
tip->device = dip;
tip->events_processed = 0;
tip->fd = -1;
- tip->pfd = -1;
memset(&tip->fifo, 0, sizeof(tip->fifo));
tip->leftover_ts = NULL;
tip->cpu = i;
tip->device = dip;
tip->fd = -1;
- tip->pfd = -1;
tip->cl_in_addr = *cl_in_addr;
if (tip_open_output(dip, tip))
return 0;
}
+static int get_connection(int fd, struct sockaddr_in *addr)
+{
+ struct pollfd pfd = { .fd = fd, .events = POLLIN };
+ socklen_t socklen;
+
+ printf("blktrace: waiting for incoming connection...\n");
+
+ if (poll(&pfd, 1, -1) < 0) {
+ perror("poll for connection");
+ return 1;
+ }
+ if ((pfd.revents & POLLIN) == 0)
+ return 1;
+
+ socklen = sizeof(*addr);
+ net_in_fd = accept(fd, (struct sockaddr *) addr, &socklen);
+ if (net_in_fd < 0) {
+ perror("accept");
+ return 1;
+ }
+
+ printf("blktrace: connection from %s\n", inet_ntoa(addr->sin_addr));
+ return 0;
+}
+
/*
* Start here when we are in server mode - just fetch data from the network
* and dump to files
struct device_information *dip;
struct thread_information *tip;
struct sockaddr_in addr;
- socklen_t socklen;
int fd, opt, i, j;
fd = socket(AF_INET, SOCK_STREAM, 0);
}
repeat:
- signal(SIGINT, NULL);
- signal(SIGHUP, NULL);
- signal(SIGTERM, NULL);
- signal(SIGALRM, NULL);
-
- printf("blktrace: waiting for incoming connection...\n");
-
- socklen = sizeof(addr);
- net_in_fd = accept(fd, (struct sockaddr *) &addr, &socklen);
- if (net_in_fd < 0) {
- perror("accept");
- return 1;
- }
-
- signal(SIGINT, handle_sigint);
- signal(SIGHUP, handle_sigint);
- signal(SIGTERM, handle_sigint);
- signal(SIGALRM, handle_sigint);
-
- printf("blktrace: connection from %s\n", inet_ntoa(addr.sin_addr));
+ if (get_connection(fd, &addr))
+ return 0;
while (!is_done()) {
if (net_server_loop(&addr.sin_addr))
projects / blktrace.git / blobdiff