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         return ret;
 147 }
 148
 149 void fio_net_cmd_crc(struct fio_net_cmd *cmd)
 150 {
 151         uint32_t pdu_len;
 152
 153         cmd->cmd_crc32 = cpu_to_le32(crc32(cmd,
 154                                         sizeof(*cmd) - 2 * sizeof(uint32_t)));
 155
 156         pdu_len = le32_to_cpu(cmd->pdu_len);
 157         if (pdu_len)
 158                 cmd->pdu_crc32 = cpu_to_le32(crc32(cmd->payload, pdu_len));
 159 }
 160
 161 static int send_simple_command(int sk, uint16_t opcode, uint64_t serial)
 162 {
 163         struct fio_net_cmd cmd = {
 164                 .version        = cpu_to_le16(FIO_SERVER_VER1),
 165                 .opcode         = cpu_to_le16(opcode),
 166                 .serial         = cpu_to_le16(serial),
 167         };
 168
 169         fio_net_cmd_crc(&cmd);
 170
 171         return fio_send_data(sk, &cmd, sizeof(cmd));
 172 }
 173
 174 /*
 175  * Send an ack for this command
 176  */
 177 static int ack_command(int sk, struct fio_net_cmd *cmd)
 178 {
 179         return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
 180 }
 181
 182 #if 0
 183 static int nak_command(int sk, struct fio_net_cmd *cmd)
 184 {
 185         return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial);
 186 }
 187 #endif
 188
 189 static int handle_cur_job(struct fio_net_cmd *cmd, int done)
 190 {
 191         unsigned int left = job_max_len - job_cur_len;
 192         int ret = 0;
 193
 194         if (left < cmd->pdu_len) {
 195                 job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len);
 196                 job_max_len += 2 * cmd->pdu_len;
 197         }
 198
 199         memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len);
 200         job_cur_len += cmd->pdu_len;
 201
 202         if (done) {
 203                 parse_jobs_ini(job_buf, 1, 0);
 204                 ret = exec_run();
 205                 reset_fio_state();
 206                 free(job_buf);
 207                 job_buf = NULL;
 208                 job_cur_len = job_max_len = 0;
 209         }
 210
 211         return ret;
 212 }
 213
 214 static int handle_command(struct fio_net_cmd *cmd)
 215 {
 216         int ret;
 217
 218         switch (cmd->opcode) {
 219         case FIO_NET_CMD_QUIT:
 220                 exit_backend = 1;
 221                 return 1;
 222         case FIO_NET_CMD_ACK:
 223                 return 0;
 224         case FIO_NET_CMD_NAK:
 225                 return 1;
 226         case FIO_NET_CMD_JOB:
 227                 ret = handle_cur_job(cmd, 0);
 228                 break;
 229         case FIO_NET_CMD_JOB_END:
 230                 ret = handle_cur_job(cmd, 1);
 231                 break;
 232         default:
 233                 log_err("fio: unknown opcode: %d\n", cmd->opcode);
 234                 ret = 1;
 235         }
 236
 237         return ret;
 238 }
 239
 240 static int handle_connection(int sk)
 241 {
 242         struct fio_net_cmd *cmd = NULL;
 243         int ret = 0;
 244
 245         /* read forever */
 246         while (!exit_backend) {
 247                 cmd = fio_net_cmd_read(sk);
 248                 if (!cmd) {
 249                         ret = 1;
 250                         break;
 251                 }
 252
 253                 ret = ack_command(sk, cmd);
 254                 if (ret)
 255                         break;
 256
 257                 ret = handle_command(cmd);
 258                 if (ret)
 259                         break;
 260
 261                 free(cmd);
 262         }
 263
 264         if (cmd)
 265                 free(cmd);
 266
 267         return ret;
 268 }
 269
 270 static int accept_loop(int listen_sk)
 271 {
 272         struct sockaddr addr;
 273         unsigned int len = sizeof(addr);
 274         struct pollfd pfd;
 275         int ret, sk, flags, exitval = 0;
 276
 277         flags = fcntl(listen_sk, F_GETFL);
 278         flags |= O_NONBLOCK;
 279         fcntl(listen_sk, F_SETFL, flags);
 280 again:
 281         pfd.fd = listen_sk;
 282         pfd.events = POLLIN;
 283         do {
 284                 ret = poll(&pfd, 1, 100);
 285                 if (ret < 0) {
 286                         if (errno == EINTR)
 287                                 break;
 288                         perror("poll");
 289                         goto out;
 290                 } else if (!ret)
 291                         continue;
 292
 293                 if (pfd.revents & POLLIN)
 294                         break;
 295         } while (!exit_backend);
 296
 297         if (exit_backend)
 298                 goto out;
 299
 300         sk = accept(listen_sk, &addr, &len);
 301         if (sk < 0) {
 302                 log_err("fio: accept: %s\n", strerror(errno));
 303                 return -1;
 304         }
 305
 306         server_fd = sk;
 307
 308         exitval = handle_connection(sk);
 309
 310         server_fd = -1;
 311         close(sk);
 312
 313         if (!exit_backend)
 314                 goto again;
 315
 316 out:
 317         return exitval;
 318 }
 319
 320 int fio_server(void)
 321 {
 322         struct sockaddr_in saddr_in;
 323         struct sockaddr addr;
 324         unsigned int len;
 325         int sk, opt, ret;
 326
 327         sk = socket(AF_INET, SOCK_STREAM, 0);
 328         if (sk < 0) {
 329                 log_err("fio: socket: %s\n", strerror(errno));
 330                 return -1;
 331         }
 332
 333         opt = 1;
 334         if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
 335                 log_err("fio: setsockopt: %s\n", strerror(errno));
 336                 return -1;
 337         }
 338 #ifdef SO_REUSEPORT
 339         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
 340                 log_err("fio: setsockopt: %s\n", strerror(errno));
 341                 return 1;
 342         }
 343 #endif
 344
 345         saddr_in.sin_family = AF_INET;
 346         saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
 347         saddr_in.sin_port = htons(fio_net_port);
 348
 349         if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
 350                 log_err("fio: bind: %s\n", strerror(errno));
 351                 return -1;
 352         }
 353
 354         if (listen(sk, 1) < 0) {
 355                 log_err("fio: listen: %s\n", strerror(errno));
 356                 return -1;
 357         }
 358
 359         len = sizeof(addr);
 360         if (getsockname(sk, &addr, &len) < 0) {
 361                 log_err("fio: getsockname: %s\n", strerror(errno));
 362                 return -1;
 363         }
 364
 365         ret = accept_loop(sk);
 366         close(sk);
 367         return ret;
 368 }
 369
 370 int fio_server_text_output(const char *buf, unsigned int len)
 371 {
 372         struct fio_net_cmd *cmd;
 373         int size = sizeof(*cmd) + len;
 374
 375         cmd = malloc(size);
 376         fio_init_net_cmd(cmd, FIO_NET_CMD_TEXT, buf, len);
 377         fio_net_cmd_crc(cmd);
 378
 379         fio_send_data(server_fd, cmd, size);
 380         free(cmd);
 381         return size;
 382 }
 383
 384 int fio_server_log(const char *format, ...)
 385 {
 386         char buffer[1024];
 387         va_list args;
 388
 389         va_start(args, format);
 390         snprintf(buffer, sizeof(buffer), format, args);
 391         va_end(args);
 392
 393         return fio_server_text_output(buffer, strlen(buffer));
 394 }