blktrace.c

   1 /*
   2  * block queue tracing application
   3  *
   4  * TODO (in no particular order):
   5  *      - Add ability to specify capture mask instead of logging all events
   6  *      - Add option for relayfs mount point
   7  *
   8  */
   9 #include <pthread.h>
  10 #include <sys/types.h>
  11 #include <sys/stat.h>
  12 #include <unistd.h>
  13 #include <locale.h>
  14 #include <signal.h>
  15 #include <fcntl.h>
  16 #include <string.h>
  17 #include <sys/ioctl.h>
  18 #include <stdio.h>
  19 #include <stdlib.h>
  20 #include <sched.h>
  21
  22 #include "blktrace.h"
  23
  24 #define BUF_SIZE        (128 *1024)
  25 #define BUF_NR          (4)
  26
  27 struct thread_information {
  28         int cpu;
  29         pthread_t thread;
  30         unsigned long events_processed;
  31 };
  32
  33 static char relay_path[] = "/relay/";
  34
  35 #define is_done()       (*(volatile int *)(&done))
  36 static volatile int done;
  37
  38 static int devfd, ncpus;
  39 static struct thread_information *thread_information;
  40 static char *buts_name_p;
  41
  42 static int start_trace(char *dev)
  43 {
  44         struct blk_user_trace_setup buts;
  45
  46         devfd = open(dev, O_RDONLY);
  47         if (devfd < 0) {
  48                 perror(dev);
  49                 return 1;
  50         }
  51
  52         memset(&buts, sizeof(buts), 0);
  53         buts.buf_size = BUF_SIZE;
  54         buts.buf_nr = BUF_NR;
  55
  56         printf("Starting trace on %s\n", dev);
  57         if (ioctl(devfd, BLKSTARTTRACE, &buts) < 0) {
  58                 perror("BLKSTARTTRACE");
  59                 return 1;
  60         }
  61
  62         buts_name_p = strdup(buts.name);
  63         return 0;
  64 }
  65
  66 static void stop_trace(void)
  67 {
  68         if (ioctl(devfd, BLKSTOPTRACE) < 0)
  69                 perror("BLKSTOPTRACE");
  70
  71         close(devfd);
  72 }
  73
  74 static void extract_data(int cpu, char *ifn, int ifd, char *ofn, int ofd,
  75                          int nb)
  76 {
  77         int ret, bytes_left;
  78         unsigned char *buf, *p;
  79
  80         buf = malloc(nb);
  81         p = buf;
  82         bytes_left = nb;
  83         while (bytes_left > 0) {
  84                 ret = read(ifd, p, bytes_left);
  85                 if (!ret)
  86                         usleep(1000);
  87                 else if (ret < 0) {
  88                         perror(ifn);
  89                         fprintf(stderr, "Thread %d extract_data %s failed\n",
  90                                 cpu, ifn);
  91                         free(buf);
  92                         exit(1);
  93                 } else {
  94                         p += ret;
  95                         bytes_left -= ret;
  96                 }
  97         }
  98
  99         ret = write(ofd, buf, nb);
 100         if (ret != nb) {
 101                 perror(ofn);
 102                 fprintf(stderr,"Thread %d extract_data %s failed\n", cpu, ofn);
 103                 free(buf);
 104                 exit(1);
 105         }
 106
 107         free(buf);
 108 }
 109
 110 static void *extract(void *arg)
 111 {
 112         struct thread_information *tip = arg;
 113         int tracefd, ret, ofd;
 114         char ip[64], op[64], dp[64];
 115         struct blk_io_trace t;
 116         pid_t pid = getpid();
 117         cpu_set_t cpu_mask;
 118
 119         CPU_ZERO(&cpu_mask);
 120         CPU_SET(tip->cpu, &cpu_mask);
 121
 122         if (sched_setaffinity(pid, sizeof(cpu_mask), &cpu_mask) == -1) {
 123                 perror("sched_setaffinity");
 124                 exit(1);
 125         }
 126
 127         sprintf(op, "%s_out.%d", buts_name_p, tip->cpu);
 128         ofd = open(op, O_CREAT|O_TRUNC|O_WRONLY, 0644);
 129         if (ofd < 0) {
 130                 perror(op);
 131                 fprintf(stderr,"Thread %d failed creat of %s\n", tip->cpu, op);
 132                 exit(1);
 133         }
 134
 135         sprintf(ip, "%s%s%d", relay_path, buts_name_p, tip->cpu);
 136         tracefd = open(ip, O_RDONLY);
 137         if (tracefd < 0) {
 138                 perror(ip);
 139                 fprintf(stderr,"Thread %d failed open of %s\n", tip->cpu, ip);
 140                 exit(1);
 141         }
 142
 143         while (!is_done()) {
 144                 ret = read(tracefd, &t, sizeof(t));
 145                 if (ret != sizeof(t)) {
 146                         if (ret < 0) {
 147                                 perror(ip);
 148                                 fprintf(stderr,"Thread %d failed read of %s\n",
 149                                         tip->cpu, ip);
 150                                 exit(1);
 151                         } else if (ret > 0) {
 152                                 fprintf(stderr,"Thread %d misread %s %d,%d\n",
 153                                         tip->cpu, ip, ret, (int)sizeof(t));
 154                                 exit(1);
 155                         } else {
 156                                 usleep(10000);
 157                                 continue;
 158                         }
 159                 }
 160
 161                 if (verify_trace(&t))
 162                         exit(1);
 163
 164                 trace_to_be(&t);
 165
 166                 ret = write(ofd, &t, sizeof(t));
 167                 if (ret < 0) {
 168                         perror(op);
 169                         fprintf(stderr,"Thread %d failed write of %s\n",
 170                                 tip->cpu, op);
 171                         exit(1);
 172                 }
 173
 174                 if (t.pdu_len)
 175                         extract_data(tip->cpu, ip, tracefd, dp, ofd, t.pdu_len);
 176
 177                 tip->events_processed++;
 178         }
 179
 180         return NULL;
 181 }
 182
 183 static int start_threads(void)
 184 {
 185         struct thread_information *tip;
 186         int i;
 187
 188         ncpus = sysconf(_SC_NPROCESSORS_ONLN);
 189         if (ncpus < 0) {
 190                 fprintf(stderr, "sysconf(_SC_NPROCESSORS_ONLN) failed\n");
 191                 return 1;
 192         }
 193         printf("Processors online: %d\n", ncpus);
 194
 195         thread_information = malloc(ncpus * sizeof(struct thread_information));
 196         for (i = 0, tip = thread_information; i < ncpus; i++, tip++) {
 197                 tip->cpu = i;
 198                 tip->events_processed = 0;
 199
 200                 if (pthread_create(&tip->thread, NULL, extract, tip)) {
 201                         perror( "pthread_create");
 202                         return 0;
 203                 }
 204         }
 205
 206         return ncpus;
 207 }
 208
 209 static void stop_threads(void)
 210 {
 211         struct thread_information *tip = thread_information;
 212         int i;
 213
 214         for (i = 0; i < ncpus; i++, tip++) {
 215                 int ret;
 216
 217                 if (pthread_join(tip->thread, (void *) &ret))
 218                         perror("thread_join");
 219         }
 220 }
 221
 222 void show_stats(void)
 223 {
 224         int i;
 225         struct thread_information *tip;
 226         unsigned long events_processed = 0;
 227
 228         for (i = 0, tip = thread_information; i < ncpus; i++, tip++) {
 229                 printf("CPU%3d: %20ld events\n",
 230                        tip->cpu, tip->events_processed);
 231                 events_processed += tip->events_processed;
 232         }
 233
 234         printf("Total:  %20ld events\n", events_processed);
 235 }
 236
 237 void handle_sigint(int sig)
 238 {
 239         printf("exiting on signal %d\n", sig);
 240         done = 1;
 241 }
 242
 243 int main(int argc, char *argv[])
 244 {
 245         struct stat st;
 246         int i;
 247
 248         if (argc < 2) {
 249                 fprintf(stderr, "Usage: %s <dev>\n", argv[0]);
 250                 return 1;
 251         }
 252
 253         if (stat(relay_path, &st) < 0) {
 254                 fprintf(stderr,"%s does not appear to be mounted\n",
 255                         relay_path);
 256                 return 2;
 257         }
 258
 259         if (start_trace(argv[1])) {
 260                 fprintf(stderr, "Failed to start trace\n");
 261                 stop_trace();
 262                 return 3;
 263         }
 264
 265         setlocale(LC_NUMERIC, "en_US");
 266
 267         i = start_threads();
 268         if (!i) {
 269                 fprintf(stderr, "Failed to start worker threads\n");
 270                 stop_trace();
 271                 return 4;
 272         }
 273
 274         printf("Threads started  : %d\n", i);
 275
 276         signal(SIGINT, handle_sigint);
 277         signal(SIGHUP, handle_sigint);
 278         signal(SIGTERM, handle_sigint);
 279
 280         while (!is_done())
 281                 sleep(1);
 282
 283         stop_threads();
 284         stop_trace();
 285         close(devfd);
 286         show_stats();
 287
 288         return 0;
 289 }
 290