splice-bench.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <unistd.h>
   4 #include <limits.h>
   5 #include <string.h>
   6 #include <getopt.h>
   7 #include <errno.h>
   8 #include <fcntl.h>
   9 #include <sys/poll.h>
  10 #include <sys/stat.h>
  11 #include <sys/types.h>
  12 #include <sys/wait.h>
  13 #include <sys/time.h>
  14 #include <signal.h>
  15 #include <netinet/in.h>
  16 #include <arpa/inet.h>
  17 #include <netdb.h>
  18 #include <sched.h>
  19
  20 #include "splice.h"
  21
  22 static int nr_clients = 8;
  23 static int net_port = 8888;
  24 static int client_loops = 10;
  25 static int bind_cpu;
  26 static int write_to_null;
  27 static int same_file;
  28 static int splice_size = SPLICE_SIZE;
  29 static char *filename = "splice-file";
  30
  31 static int nr_cpus;
  32
  33 static int usage(char *name)
  34 {
  35         fprintf(stderr, "Usage %s [options] [filename]:\n", name);
  36         fprintf(stderr, "\t[-n] (number of clients]\n");
  37         fprintf(stderr, "\t[-p] (port number)\n");
  38         fprintf(stderr, "\t[-l] (number of loops)\n");
  39         fprintf(stderr, "\t[-z] (write to /dev/null)\n");
  40         fprintf(stderr, "\t[-s] (use 1 file for all)\n");
  41         fprintf(stderr, "\t[-a] (set CPU affinity)\n");
  42         fprintf(stderr, "\t[-b] (splice chunk size)\n");
  43         return 1;
  44 }
  45
  46 static int parse_options(int argc, char *argv[])
  47 {
  48         int c, index = 1;
  49
  50         while ((c = getopt(argc, argv, "n:p:l:azsb:")) != -1) {
  51                 switch (c) {
  52                 case 'n':
  53                         nr_clients = atoi(optarg);
  54                         index++;
  55                         break;
  56                 case 'p':
  57                         net_port = atoi(optarg);
  58                         index++;
  59                         break;
  60                 case 'l':
  61                         client_loops = atoi(optarg);
  62                         index++;
  63                         break;
  64                 case 'a':
  65                         bind_cpu = 1;
  66                         index++;
  67                         break;
  68                 case 'z':
  69                         write_to_null = 1;
  70                         index++;
  71                         break;
  72                 case 's':
  73                         same_file = 1;
  74                         index++;
  75                         break;
  76                 case 'b':
  77                         splice_size = atoi(optarg);
  78                         index++;
  79                         break;
  80                 default:
  81                         return -1;
  82                 }
  83         }
  84
  85         return index;
  86 }
  87
  88 int bind_to_cpu(int index)
  89 {
  90         cpu_set_t cpu_mask;
  91         pid_t pid;
  92         int cpu;
  93
  94         if (!bind_cpu || nr_cpus == 1)
  95                 return 0;
  96
  97         cpu = index % nr_cpus;
  98
  99         CPU_ZERO(&cpu_mask);
 100         CPU_SET((cpu), &cpu_mask);
 101
 102         pid = getpid();
 103         if (sched_setaffinity(pid, sizeof(cpu_mask), &cpu_mask) == -1)
 104                 return error("set affinity");
 105
 106         return 0;
 107 }
 108
 109 int accept_loop(int listen_sk)
 110 {
 111         unsigned long received;
 112         struct sockaddr addr;
 113         unsigned int len = sizeof(addr);
 114         int sk;
 115
 116 again:
 117         sk = accept(listen_sk, &addr, &len);
 118         if (sk < 0)
 119                 return error("accept");
 120
 121         /* read forever */
 122         received = 0;
 123         for (;;) {
 124                 int ret = recv(sk, NULL, 128*1024*1024, MSG_TRUNC);
 125                 if (ret > 0) {
 126                         received += ret;
 127                         continue;
 128                 }
 129                 if (!ret)
 130                         break;
 131                 if (errno == EAGAIN || errno == EINTR)
 132                         continue;
 133                 break;
 134         }
 135
 136         close(sk);
 137         goto again;
 138 }
 139
 140 int server(int offset)
 141 {
 142         struct sockaddr addr;
 143         struct sockaddr_in saddr_in;
 144         unsigned int len;
 145         int sk;
 146
 147         bind_to_cpu(offset);
 148         nice(-20);
 149
 150         sk = socket(PF_INET, SOCK_STREAM, 0);
 151         if (sk < 0)
 152                 return error("socket");
 153
 154         saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
 155         saddr_in.sin_port = htons(net_port + offset);
 156
 157         if (bind(sk, (struct sockaddr*)&saddr_in, sizeof(saddr_in)) < 0)
 158                 return error("bind");
 159
 160         if (listen(sk, 1) < 0)
 161                 return error("listen");
 162
 163         len = sizeof(addr);
 164         if (getsockname(sk, &addr, &len) < 0)
 165                 return error("getsockname");
 166
 167         return accept_loop(sk);
 168 }
 169
 170 static unsigned long mtime_since(struct timeval *s, struct timeval *e)
 171 {
 172         double sec, usec;
 173
 174         sec = e->tv_sec - s->tv_sec;
 175         usec = e->tv_usec - s->tv_usec;
 176         if (sec > 0 && usec < 0) {
 177                 sec--;
 178                 usec += 1000000;
 179         }
 180
 181         sec *= (double) 1000;
 182         usec /= (double) 1000;
 183
 184         return sec + usec;
 185 }
 186
 187 static unsigned long mtime_since_now(struct timeval *s)
 188 {
 189         struct timeval t;
 190
 191         gettimeofday(&t, NULL);
 192         return mtime_since(s, &t);
 193 }
 194
 195 int client_loop(int out_fd, int fd, int *pfd, int offset)
 196 {
 197         struct timeval start;
 198         unsigned long long size;
 199         unsigned long msecs;
 200         struct stat sb;
 201         int loops = client_loops;
 202         loff_t off;
 203
 204         if (fstat(fd, &sb) < 0)
 205                 return error("fstat");
 206
 207         gettimeofday(&start, NULL);
 208
 209 again:
 210         size = sb.st_size;
 211         off = 0;
 212
 213         do {
 214                 int ret = splice(fd, &off, pfd[1], NULL, min(size, (unsigned long long) splice_size), 0);
 215
 216                 if (ret <= 0)
 217                         return error("splice-in");
 218
 219                 size -= ret;
 220                 while (ret > 0) {
 221                         int flags = size ? SPLICE_F_MORE : 0;
 222                         int written = splice(pfd[0], NULL, out_fd, NULL, ret, flags);
 223
 224                         if (written <= 0)
 225                                 return error("splice-out");
 226
 227                         ret -= written;
 228                 }
 229         } while (size);
 230
 231         if (--loops)
 232                 goto again;
 233
 234         size = sb.st_size >> 10;
 235         size *= client_loops;
 236         msecs = mtime_since_now(&start);
 237         fprintf(stdout, "Client%d: %Lu MiB/sec\n", offset, size / msecs);
 238         return 0;
 239 }
 240
 241 static int client_open_net(int offset)
 242 {
 243         int sk = socket(PF_INET, SOCK_STREAM, 0);
 244         struct sockaddr_in s_to;
 245         struct hostent *hp;
 246
 247         hp = gethostbyname("localhost");
 248         if (!hp)
 249                 return error("gethostbyname");
 250
 251         bzero((char *) &s_to, sizeof (s_to));
 252         bcopy((char *) hp->h_addr, (char *) &(s_to.sin_addr), hp->h_length);
 253         s_to.sin_family = hp->h_addrtype;
 254         s_to.sin_port = htons(net_port + offset);
 255
 256         if (connect(sk, (struct sockaddr *)&s_to, sizeof(s_to)) < 0)
 257                 return error("connect");
 258
 259         return sk;
 260 }
 261
 262 int client(int offset)
 263 {
 264         int file_fd, out_fd;
 265         char fname[64];
 266         int pfd[2];
 267
 268         bind_to_cpu(offset);
 269         nice(-20);
 270
 271         if (!write_to_null)
 272                 out_fd = client_open_net(offset);
 273         else
 274                 out_fd = open("/dev/null", O_WRONLY);
 275
 276         if (out_fd < 0)
 277                 return error("socket");
 278
 279         sprintf(fname, "%s%d", filename, same_file ? 0 : offset);
 280         file_fd = open(filename, O_RDONLY);
 281         if (file_fd < 0)
 282                 return error("open");
 283
 284         if (pipe(pfd) < 0)
 285                 return error("pipe");
 286
 287         return client_loop(out_fd, file_fd, pfd, offset);
 288 }
 289
 290 int main(int argc, char *argv[])
 291 {
 292         pid_t *spids, *cpids;
 293         int i, index;
 294
 295         index = parse_options(argc, argv);
 296         if (index < 0)
 297                 return usage(argv[0]);
 298
 299         if (index < argc)
 300                 filename = argv[index];
 301
 302         spids = malloc(nr_clients * sizeof(pid_t));
 303         cpids = malloc(nr_clients * sizeof(pid_t));
 304         memset(spids, 0, nr_clients * sizeof(pid_t));
 305         memset(cpids, 0, nr_clients * sizeof(pid_t));
 306
 307         nr_cpus = sysconf(_SC_NPROCESSORS_ONLN);
 308         if (nr_cpus < 0)
 309                 return error("_SC_NPROCESSORS_ONLN");
 310
 311         fprintf(stdout,"Processors: %d\n", nr_cpus);
 312
 313         /*
 314          * fork servers
 315          */
 316         if (!write_to_null) {
 317                 for (i = 0; i < nr_clients; i++) {
 318                         pid_t pid = fork();
 319
 320                         if (pid)
 321                                 spids[i] = pid;
 322                         else {
 323                                 server(i);
 324                                 spids[i] = 0;
 325                                 exit(0);
 326                         }
 327                 }
 328                 sleep(1); /* should have servers started now */
 329         }
 330
 331         /*
 332          * fork clients
 333          */
 334         for (i = 0; i < nr_clients; i++) {
 335                 pid_t pid = fork();
 336
 337                 if (pid)
 338                         cpids[i] = pid;
 339                 else {
 340                         client(i);
 341                         cpids[i] = 0;
 342                         exit(0);
 343                 }
 344         }
 345
 346         /*
 347          * wait for clients to exit
 348          */
 349         fprintf(stdout, "Waiting for clients\n");
 350         for (i = 0; i < nr_clients; i++) {
 351                 if (cpids[i]) {
 352                         waitpid(cpids[i], NULL, 0);
 353                         cpids[i] = 0;
 354                 }
 355         }
 356
 357         /*
 358          * then kill servers
 359          */
 360         for (i = 0; i < nr_clients; i++) {
 361                 if (spids[i]) {
 362                         kill(spids[i], SIGKILL);
 363                         waitpid(spids[i], NULL, 0);
 364                         spids[i] = 0;
 365                 }
 366         }
 367
 368         return 0;
 369 }