server.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <stdarg.h>
   4 #include <unistd.h>
   5 #include <limits.h>
   6 #include <errno.h>
   7 #include <fcntl.h>
   8 #include <sys/poll.h>
   9 #include <sys/types.h>
  10 #include <sys/wait.h>
  11 #include <sys/mman.h>
  12 #include <netinet/in.h>
  13 #include <arpa/inet.h>
  14 #include <netdb.h>
  15
  16 #include "fio.h"
  17 #include "server.h"
  18 #include "crc/crc32.h"
  19
  20 int fio_net_port = 8765;
  21
  22 int exit_backend = 0;
  23
  24 static char *job_buf;
  25 static unsigned int job_cur_len;
  26 static unsigned int job_max_len;
  27
  28 static int server_fd;
  29
  30 int fio_send_data(int sk, const void *p, unsigned int len)
  31 {
  32         do {
  33                 int ret = send(sk, p, len, 0);
  34
  35                 if (ret > 0) {
  36                         len -= ret;
  37                         if (!len)
  38                                 break;
  39                         p += ret;
  40                         continue;
  41                 } else if (!ret)
  42                         break;
  43                 else if (errno == EAGAIN || errno == EINTR)
  44                         continue;
  45         } while (!exit_backend);
  46
  47         if (!len)
  48                 return 0;
  49
  50         return 1;
  51 }
  52
  53 int fio_recv_data(int sk, void *p, unsigned int len)
  54 {
  55         do {
  56                 int ret = recv(sk, p, len, MSG_WAITALL);
  57
  58                 if (ret > 0) {
  59                         len -= ret;
  60                         if (!len)
  61                                 break;
  62                         p += ret;
  63                         continue;
  64                 } else if (!ret)
  65                         break;
  66                 else if (errno == EAGAIN || errno == EINTR)
  67                         continue;
  68         } while (!exit_backend);
  69
  70         if (!len)
  71                 return 0;
  72
  73         return -1;
  74 }
  75
  76 static int verify_convert_cmd(struct fio_net_cmd *cmd)
  77 {
  78         uint32_t crc;
  79
  80         cmd->cmd_crc32 = le32_to_cpu(cmd->cmd_crc32);
  81         cmd->pdu_crc32 = le32_to_cpu(cmd->pdu_crc32);
  82
  83         crc = crc32(cmd, sizeof(*cmd) - 2 * sizeof(uint32_t));
  84         if (crc != cmd->cmd_crc32) {
  85                 log_err("fio: server bad crc on command (got %x, wanted %x)\n",
  86                                 cmd->cmd_crc32, crc);
  87                 return 1;
  88         }
  89
  90         cmd->version    = le16_to_cpu(cmd->version);
  91         cmd->opcode     = le16_to_cpu(cmd->opcode);
  92         cmd->flags      = le32_to_cpu(cmd->flags);
  93         cmd->serial     = le64_to_cpu(cmd->serial);
  94         cmd->pdu_len    = le32_to_cpu(cmd->pdu_len);
  95
  96         switch (cmd->version) {
  97         case FIO_SERVER_VER1:
  98                 break;
  99         default:
 100                 log_err("fio: bad server cmd version %d\n", cmd->version);
 101                 return 1;
 102         }
 103
 104         if (cmd->pdu_len > FIO_SERVER_MAX_PDU) {
 105                 log_err("fio: command payload too large: %u\n", cmd->pdu_len);
 106                 return 1;
 107         }
 108
 109         return 0;
 110 }
 111
 112 struct fio_net_cmd *fio_net_cmd_read(int sk)
 113 {
 114         struct fio_net_cmd cmd, *ret = NULL;
 115         uint32_t crc;
 116
 117         if (fio_recv_data(sk, &cmd, sizeof(cmd)))
 118                 return NULL;
 119
 120         /* We have a command, verify it and swap if need be */
 121         if (verify_convert_cmd(&cmd))
 122                 return NULL;
 123
 124         /* Command checks out, alloc real command and fill in */
 125         ret = malloc(sizeof(cmd) + cmd.pdu_len);
 126         memcpy(ret, &cmd, sizeof(cmd));
 127
 128         if (!ret->pdu_len)
 129                 return ret;
 130
 131         /* There's payload, get it */
 132         if (fio_recv_data(sk, (void *) ret + sizeof(*ret), ret->pdu_len)) {
 133                 free(ret);
 134                 return NULL;
 135         }
 136
 137         /* Verify payload crc */
 138         crc = crc32(ret->payload, ret->pdu_len);
 139         if (crc != ret->pdu_crc32) {
 140                 log_err("fio: server bad crc on payload (got %x, wanted %x)\n",
 141                                 ret->pdu_crc32, crc);
 142                 free(ret);
 143                 return NULL;
 144         }
 145
 146
 147         return ret;
 148 }
 149
 150 void fio_net_cmd_crc(struct fio_net_cmd *cmd)
 151 {
 152         uint32_t pdu_len;
 153
 154         cmd->cmd_crc32 = cpu_to_le32(crc32(cmd,
 155                                         sizeof(*cmd) - 2 * sizeof(uint32_t)));
 156
 157         pdu_len = le32_to_cpu(cmd->pdu_len);
 158         if (pdu_len)
 159                 cmd->pdu_crc32 = cpu_to_le32(crc32(cmd->payload, pdu_len));
 160 }
 161
 162 static int send_simple_command(int sk, uint16_t opcode, uint64_t serial)
 163 {
 164         struct fio_net_cmd cmd = {
 165                 .version        = cpu_to_le16(FIO_SERVER_VER1),
 166                 .opcode         = cpu_to_le16(opcode),
 167                 .serial         = cpu_to_le16(serial),
 168         };
 169
 170         fio_net_cmd_crc(&cmd);
 171
 172         return fio_send_data(sk, &cmd, sizeof(cmd));
 173 }
 174
 175 /*
 176  * Send an ack for this command
 177  */
 178 static int ack_command(int sk, struct fio_net_cmd *cmd)
 179 {
 180         return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
 181 }
 182
 183 #if 0
 184 static int nak_command(int sk, struct fio_net_cmd *cmd)
 185 {
 186         return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial);
 187 }
 188 #endif
 189
 190 static int handle_cur_job(struct fio_net_cmd *cmd, int done)
 191 {
 192         unsigned int left = job_max_len - job_cur_len;
 193         int ret = 0;
 194
 195         if (left < cmd->pdu_len) {
 196                 job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len);
 197                 job_max_len += 2 * cmd->pdu_len;
 198         }
 199
 200         memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len);
 201         job_cur_len += cmd->pdu_len;
 202
 203         if (done) {
 204                 parse_jobs_ini(job_buf, 1, 0);
 205                 ret = exec_run();
 206                 reset_fio_state();
 207                 free(job_buf);
 208                 job_buf = NULL;
 209                 job_cur_len = job_max_len = 0;
 210         }
 211
 212         return ret;
 213 }
 214
 215 static int handle_command(struct fio_net_cmd *cmd)
 216 {
 217         int ret;
 218
 219         switch (cmd->opcode) {
 220         case FIO_NET_CMD_QUIT:
 221                 exit_backend = 1;
 222                 return 1;
 223         case FIO_NET_CMD_ACK:
 224                 return 0;
 225         case FIO_NET_CMD_NAK:
 226                 return 1;
 227         case FIO_NET_CMD_JOB:
 228                 ret = handle_cur_job(cmd, 0);
 229                 break;
 230         case FIO_NET_CMD_JOB_END:
 231                 ret = handle_cur_job(cmd, 1);
 232                 break;
 233         default:
 234                 log_err("fio: unknown opcode: %d\n", cmd->opcode);
 235                 ret = 1;
 236         }
 237
 238         return ret;
 239 }
 240
 241 static int handle_connection(int sk)
 242 {
 243         struct fio_net_cmd *cmd = NULL;
 244         int ret = 0;
 245
 246         /* read forever */
 247         while (!exit_backend) {
 248                 cmd = fio_net_cmd_read(sk);
 249                 if (!cmd) {
 250                         ret = 1;
 251                         break;
 252                 }
 253
 254                 ret = ack_command(sk, cmd);
 255                 if (ret)
 256                         break;
 257
 258                 ret = handle_command(cmd);
 259                 if (ret)
 260                         break;
 261
 262                 free(cmd);
 263         }
 264
 265         if (cmd)
 266                 free(cmd);
 267
 268         return ret;
 269 }
 270
 271 static int accept_loop(int listen_sk)
 272 {
 273         struct sockaddr addr;
 274         unsigned int len = sizeof(addr);
 275         struct pollfd pfd;
 276         int ret, sk, flags, exitval = 0;
 277
 278         flags = fcntl(listen_sk, F_GETFL);
 279         flags |= O_NONBLOCK;
 280         fcntl(listen_sk, F_SETFL, flags);
 281 again:
 282         pfd.fd = listen_sk;
 283         pfd.events = POLLIN;
 284         do {
 285                 ret = poll(&pfd, 1, 100);
 286                 if (ret < 0) {
 287                         if (errno == EINTR)
 288                                 break;
 289                         perror("poll");
 290                         goto out;
 291                 } else if (!ret)
 292                         continue;
 293
 294                 if (pfd.revents & POLLIN)
 295                         break;
 296         } while (!exit_backend);
 297
 298         if (exit_backend)
 299                 goto out;
 300
 301         sk = accept(listen_sk, &addr, &len);
 302         if (sk < 0) {
 303                 log_err("fio: accept: %s\n", strerror(errno));
 304                 return -1;
 305         }
 306
 307         server_fd = sk;
 308
 309         exitval = handle_connection(sk);
 310
 311         server_fd = -1;
 312         close(sk);
 313
 314         if (!exit_backend)
 315                 goto again;
 316
 317 out:
 318         return exitval;
 319 }
 320
 321 int fio_server(void)
 322 {
 323         struct sockaddr_in saddr_in;
 324         struct sockaddr addr;
 325         unsigned int len;
 326         int sk, opt, ret;
 327
 328         sk = socket(AF_INET, SOCK_STREAM, 0);
 329         if (sk < 0) {
 330                 log_err("fio: socket: %s\n", strerror(errno));
 331                 return -1;
 332         }
 333
 334         opt = 1;
 335         if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
 336                 log_err("fio: setsockopt: %s\n", strerror(errno));
 337                 return -1;
 338         }
 339 #ifdef SO_REUSEPORT
 340         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
 341                 log_err("fio: setsockopt: %s\n", strerror(errno));
 342                 return 1;
 343         }
 344 #endif
 345
 346         saddr_in.sin_family = AF_INET;
 347         saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
 348         saddr_in.sin_port = htons(fio_net_port);
 349
 350         if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
 351                 log_err("fio: bind: %s\n", strerror(errno));
 352                 return -1;
 353         }
 354
 355         if (listen(sk, 1) < 0) {
 356                 log_err("fio: listen: %s\n", strerror(errno));
 357                 return -1;
 358         }
 359
 360         len = sizeof(addr);
 361         if (getsockname(sk, &addr, &len) < 0) {
 362                 log_err("fio: getsockname: %s\n", strerror(errno));
 363                 return -1;
 364         }
 365
 366         ret = accept_loop(sk);
 367         close(sk);
 368         return ret;
 369 }
 370
 371 int fio_server_text_output(const char *buf, unsigned int len)
 372 {
 373         struct fio_net_cmd *cmd;
 374         int size = sizeof(*cmd) + len;
 375
 376         cmd = malloc(size);
 377         fio_init_net_cmd(cmd);
 378         cmd->opcode     = cpu_to_le16(FIO_NET_CMD_TEXT);
 379         cmd->pdu_len    = cpu_to_le32(len);
 380         memcpy(&cmd->payload, buf, len);
 381
 382         fio_net_cmd_crc(cmd);
 383
 384         fio_send_data(server_fd, cmd, size);
 385         free(cmd);
 386         return size;
 387 }
 388
 389 int fio_server_log(const char *format, ...)
 390 {
 391         char buffer[1024];
 392         va_list args;
 393
 394         va_start(args, format);
 395         sprintf(buffer, format, args);
 396         va_end(args);
 397
 398         return fio_server_text_output(buffer, strlen(buffer));
 399 }