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