sbin/ggate/ggated/ggated.c

   1 /*-
   2  * Copyright (c) 2004 Pawel Jakub Dawidek <pjd@FreeBSD.org>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice, this list of conditions and the following disclaimer.
  10  * 2. Redistributions in binary form must reproduce the above copyright
  11  *    notice, this list of conditions and the following disclaimer in the
  12  *    documentation and/or other materials provided with the distribution.
  13  *
  14  * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
  15  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  16  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  17  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
  18  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  19  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  20  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  21  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  22  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  23  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  24  * SUCH DAMAGE.
  25  *
  26  * $FreeBSD$
  27  */
  28
  29 #include <stdio.h>
  30 #include <stdlib.h>
  31 #include <stdint.h>
  32 #include <unistd.h>
  33 #include <fcntl.h>
  34 #include <pthread.h>
  35 #include <sys/param.h>
  36 #include <sys/queue.h>
  37 #include <sys/endian.h>
  38 #include <sys/socket.h>
  39 #include <sys/ioctl.h>
  40 #include <sys/stat.h>
  41 #include <sys/time.h>
  42 #include <sys/disk.h>
  43 #include <sys/bio.h>
  44 #include <netinet/in.h>
  45 #include <netinet/tcp.h>
  46 #include <arpa/inet.h>
  47 #include <signal.h>
  48 #include <assert.h>
  49 #include <err.h>
  50 #include <errno.h>
  51 #include <string.h>
  52 #include <libgen.h>
  53 #include <syslog.h>
  54 #include <stdarg.h>
  55
  56 #include "ggate.h"
  57
  58
  59 #define GGATED_EXPORT_FILE      "/etc/gg.exports"
  60
  61 struct ggd_connection {
  62         off_t            c_mediasize;
  63         unsigned         c_sectorsize;
  64         unsigned         c_flags;       /* flags (RO/RW) */
  65         int              c_diskfd;
  66         int              c_sendfd;
  67         int              c_recvfd;
  68         time_t           c_birthtime;
  69         char            *c_path;
  70         uint64_t         c_token;
  71         in_addr_t        c_srcip;
  72         LIST_ENTRY(ggd_connection) c_next;
  73 };
  74
  75 struct ggd_request {
  76         struct g_gate_hdr        r_hdr;
  77         char                    *r_data;
  78         TAILQ_ENTRY(ggd_request) r_next;
  79 };
  80 #define r_cmd           r_hdr.gh_cmd
  81 #define r_offset        r_hdr.gh_offset
  82 #define r_length        r_hdr.gh_length
  83 #define r_error         r_hdr.gh_error
  84
  85 struct ggd_export {
  86         char            *e_path;        /* path to device/file */
  87         in_addr_t        e_ip;          /* remote IP address */
  88         in_addr_t        e_mask;        /* IP mask */
  89         unsigned         e_flags;       /* flags (RO/RW) */
  90         SLIST_ENTRY(ggd_export) e_next;
  91 };
  92
  93 static const char *exports_file = GGATED_EXPORT_FILE;
  94 static int got_sighup = 0;
  95 static in_addr_t bindaddr;
  96
  97 static TAILQ_HEAD(, ggd_request) inqueue = TAILQ_HEAD_INITIALIZER(inqueue);
  98 static TAILQ_HEAD(, ggd_request) outqueue = TAILQ_HEAD_INITIALIZER(outqueue);
  99 static pthread_mutex_t inqueue_mtx, outqueue_mtx;
 100 static pthread_cond_t inqueue_cond, outqueue_cond;
 101
 102 static SLIST_HEAD(, ggd_export) exports = SLIST_HEAD_INITIALIZER(exports);
 103 static LIST_HEAD(, ggd_connection) connections = LIST_HEAD_INITIALIZER(connections);
 104
 105 static void *recv_thread(void *arg);
 106 static void *disk_thread(void *arg);
 107 static void *send_thread(void *arg);
 108
 109 static void
 110 usage(void)
 111 {
 112
 113         fprintf(stderr, "usage: %s [-nv] [-a address] [-p port] [-R rcvbuf] "
 114             "[-S sndbuf] [exports file]\n", getprogname());
 115         exit(EXIT_FAILURE);
 116 }
 117
 118 static char *
 119 ip2str(in_addr_t ip)
 120 {
 121         static char sip[16];
 122
 123         snprintf(sip, sizeof(sip), "%u.%u.%u.%u",
 124             ((ip >> 24) & 0xff),
 125             ((ip >> 16) & 0xff),
 126             ((ip >> 8) & 0xff),
 127             (ip & 0xff));
 128         return (sip);
 129 }
 130
 131 static in_addr_t
 132 countmask(unsigned m)
 133 {
 134         in_addr_t mask;
 135
 136         if (m == 0) {
 137                 mask = 0x0;
 138         } else {
 139                 mask = 1 << (32 - m);
 140                 mask--;
 141                 mask = ~mask;
 142         }
 143         return (mask);
 144 }
 145
 146 static void
 147 line_parse(char *line, unsigned lineno)
 148 {
 149         struct ggd_export *ex;
 150         char *word, *path, *sflags;
 151         unsigned flags, i, vmask;
 152         in_addr_t ip, mask;
 153
 154         ip = mask = flags = vmask = 0;
 155         path = NULL;
 156         sflags = NULL;
 157
 158         for (i = 0, word = strtok(line, " \t"); word != NULL;
 159             i++, word = strtok(NULL, " \t")) {
 160                 switch (i) {
 161                 case 0: /* IP address or host name */
 162                         ip = g_gate_str2ip(strsep(&word, "/"));
 163                         if (ip == INADDR_NONE) {
 164                                 g_gate_xlog("Invalid IP/host name at line %u.",
 165                                     lineno);
 166                         }
 167                         ip = ntohl(ip);
 168                         if (word == NULL)
 169                                 vmask = 32;
 170                         else {
 171                                 errno = 0;
 172                                 vmask = strtoul(word, NULL, 10);
 173                                 if (vmask == 0 && errno != 0) {
 174                                         g_gate_xlog("Invalid IP mask value at "
 175                                             "line %u.", lineno);
 176                                 }
 177                                 if ((unsigned)vmask > 32) {
 178                                         g_gate_xlog("Invalid IP mask value at line %u.",
 179                                             lineno);
 180                                 }
 181                         }
 182                         mask = countmask(vmask);
 183                         break;
 184                 case 1: /* flags */
 185                         if (strcasecmp("rd", word) == 0 ||
 186                             strcasecmp("ro", word) == 0) {
 187                                 flags = O_RDONLY;
 188                         } else if (strcasecmp("wo", word) == 0) {
 189                                 flags = O_WRONLY;
 190                         } else if (strcasecmp("rw", word) == 0) {
 191                                 flags = O_RDWR;
 192                         } else {
 193                                 g_gate_xlog("Invalid value in flags field at "
 194                                     "line %u.", lineno);
 195                         }
 196                         sflags = word;
 197                         break;
 198                 case 2: /* path */
 199                         if (strlen(word) >= MAXPATHLEN) {
 200                                 g_gate_xlog("Path too long at line %u. ",
 201                                     lineno);
 202                         }
 203                         path = word;
 204                         break;
 205                 default:
 206                         g_gate_xlog("Too many arguments at line %u. ", lineno);
 207                 }
 208         }
 209         if (i != 3)
 210                 g_gate_xlog("Too few arguments at line %u.", lineno);
 211
 212         ex = malloc(sizeof(*ex));
 213         if (ex == NULL)
 214                 g_gate_xlog("Not enough memory.");
 215         ex->e_path = strdup(path);
 216         if (ex->e_path == NULL)
 217                 g_gate_xlog("Not enough memory.");
 218
 219         /* Made 'and' here. */
 220         ex->e_ip = (ip & mask);
 221         ex->e_mask = mask;
 222         ex->e_flags = flags;
 223
 224         SLIST_INSERT_HEAD(&exports, ex, e_next);
 225
 226         g_gate_log(LOG_DEBUG, "Added %s/%u %s %s to exports list.",
 227             ip2str(ex->e_ip), vmask, path, sflags);
 228 }
 229
 230 static void
 231 exports_clear(void)
 232 {
 233         struct ggd_export *ex;
 234
 235         while (!SLIST_EMPTY(&exports)) {
 236                 ex = SLIST_FIRST(&exports);
 237                 SLIST_REMOVE_HEAD(&exports, e_next);
 238                 free(ex);
 239         }
 240 }
 241
 242 #define EXPORTS_LINE_SIZE       2048
 243 static void
 244 exports_get(void)
 245 {
 246         char buf[EXPORTS_LINE_SIZE], *line;
 247         unsigned lineno = 0, objs = 0, len;
 248         FILE *fd;
 249
 250         exports_clear();
 251
 252         fd = fopen(exports_file, "r");
 253         if (fd == NULL) {
 254                 g_gate_xlog("Cannot open exports file (%s): %s.", exports_file,
 255                     strerror(errno));
 256         }
 257
 258         g_gate_log(LOG_INFO, "Reading exports file (%s).", exports_file);
 259
 260         for (;;) {
 261                 if (fgets(buf, sizeof(buf), fd) == NULL) {
 262                         if (feof(fd))
 263                                 break;
 264
 265                         g_gate_xlog("Error while reading exports file: %s.",
 266                             strerror(errno));
 267                 }
 268
 269                 /* Increase line count. */
 270                 lineno++;
 271
 272                 /* Skip spaces and tabs. */
 273                 for (line = buf; *line == ' ' || *line == '\t'; ++line)
 274                         ;
 275
 276                 /* Empty line, comment or empty line at the end of file. */
 277                 if (*line == '\n' || *line == '#' || *line == '\0')
 278                         continue;
 279
 280                 len = strlen(line);
 281                 if (line[len - 1] == '\n') {
 282                         /* Remove new line char. */
 283                         line[len - 1] = '\0';
 284                 } else {
 285                         if (!feof(fd))
 286                                 g_gate_xlog("Line %u too long.", lineno);
 287                 }
 288
 289                 line_parse(line, lineno);
 290                 objs++;
 291         }
 292
 293         fclose(fd);
 294
 295         if (objs == 0)
 296                 g_gate_xlog("There are no objects to export.");
 297
 298         g_gate_log(LOG_INFO, "Exporting %u object(s).", objs);
 299 }
 300
 301 static int
 302 exports_check(struct ggd_export *ex, struct g_gate_cinit *cinit,
 303     struct ggd_connection *conn)
 304 {
 305         char ipmask[32]; /* 32 == strlen("xxx.xxx.xxx.xxx/xxx.xxx.xxx.xxx")+1 */
 306         int error = 0, flags;
 307
 308         strlcpy(ipmask, ip2str(ex->e_ip), sizeof(ipmask));
 309         strlcat(ipmask, "/", sizeof(ipmask));
 310         strlcat(ipmask, ip2str(ex->e_mask), sizeof(ipmask));
 311         if ((cinit->gc_flags & GGATE_FLAG_RDONLY) != 0) {
 312                 if (ex->e_flags == O_WRONLY) {
 313                         g_gate_log(LOG_WARNING, "Read-only access requested, "
 314                             "but %s (%s) is exported write-only.", ex->e_path,
 315                             ipmask);
 316                         return (EPERM);
 317                 } else {
 318                         conn->c_flags |= GGATE_FLAG_RDONLY;
 319                 }
 320         } else if ((cinit->gc_flags & GGATE_FLAG_WRONLY) != 0) {
 321                 if (ex->e_flags == O_RDONLY) {
 322                         g_gate_log(LOG_WARNING, "Write-only access requested, "
 323                             "but %s (%s) is exported read-only.", ex->e_path,
 324                             ipmask);
 325                         return (EPERM);
 326                 } else {
 327                         conn->c_flags |= GGATE_FLAG_WRONLY;
 328                 }
 329         } else {
 330                 if (ex->e_flags == O_RDONLY) {
 331                         g_gate_log(LOG_WARNING, "Read-write access requested, "
 332                             "but %s (%s) is exported read-only.", ex->e_path,
 333                             ipmask);
 334                         return (EPERM);
 335                 } else if (ex->e_flags == O_WRONLY) {
 336                         g_gate_log(LOG_WARNING, "Read-write access requested, "
 337                             "but %s (%s) is exported write-only.", ex->e_path,
 338                             ipmask);
 339                         return (EPERM);
 340                 }
 341         }
 342         if ((conn->c_flags & GGATE_FLAG_RDONLY) != 0)
 343                 flags = O_RDONLY;
 344         else if ((conn->c_flags & GGATE_FLAG_WRONLY) != 0)
 345                 flags = O_WRONLY;
 346         else
 347                 flags = O_RDWR;
 348         conn->c_diskfd = open(ex->e_path, flags);
 349         if (conn->c_diskfd == -1) {
 350                 error = errno;
 351                 g_gate_log(LOG_ERR, "Cannot open %s: %s.", ex->e_path,
 352                     strerror(error));
 353                 return (error);
 354         }
 355         return (0);
 356 }
 357
 358 static struct ggd_export *
 359 exports_find(struct sockaddr *s, struct g_gate_cinit *cinit,
 360     struct ggd_connection *conn)
 361 {
 362         struct ggd_export *ex;
 363         in_addr_t ip;
 364         int error;
 365
 366         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 367         SLIST_FOREACH(ex, &exports, e_next) {
 368                 if ((ip & ex->e_mask) != ex->e_ip) {
 369                         g_gate_log(LOG_DEBUG, "exports[%s]: IP mismatch.",
 370                             ex->e_path);
 371                         continue;
 372                 }
 373                 if (strcmp(cinit->gc_path, ex->e_path) != 0) {
 374                         g_gate_log(LOG_DEBUG, "exports[%s]: Path mismatch.",
 375                             ex->e_path);
 376                         continue;
 377                 }
 378                 error = exports_check(ex, cinit, conn);
 379                 if (error == 0)
 380                         return (ex);
 381                 else {
 382                         errno = error;
 383                         return (NULL);
 384                 }
 385         }
 386         g_gate_log(LOG_WARNING, "Unauthorized connection from: %s.",
 387             ip2str(ip));
 388         errno = EPERM;
 389         return (NULL);
 390 }
 391
 392 /*
 393  * Remove timed out connections.
 394  */
 395 static void
 396 connection_cleanups(void)
 397 {
 398         struct ggd_connection *conn, *tconn;
 399         time_t now;
 400
 401         time(&now);
 402         LIST_FOREACH_SAFE(conn, &connections, c_next, tconn) {
 403                 if (now - conn->c_birthtime > 10) {
 404                         LIST_REMOVE(conn, c_next);
 405                         g_gate_log(LOG_NOTICE,
 406                             "Connection from %s [%s] removed.",
 407                             ip2str(conn->c_srcip), conn->c_path);
 408                         close(conn->c_diskfd);
 409                         close(conn->c_sendfd);
 410                         close(conn->c_recvfd);
 411                         free(conn->c_path);
 412                         free(conn);
 413                 }
 414         }
 415 }
 416
 417 static struct ggd_connection *
 418 connection_find(struct g_gate_cinit *cinit)
 419 {
 420         struct ggd_connection *conn;
 421
 422         LIST_FOREACH(conn, &connections, c_next) {
 423                 if (conn->c_token == cinit->gc_token)
 424                         break;
 425         }
 426         return (conn);
 427 }
 428
 429 static struct ggd_connection *
 430 connection_new(struct g_gate_cinit *cinit, struct sockaddr *s, int sfd)
 431 {
 432         struct ggd_connection *conn;
 433         in_addr_t ip;
 434
 435         /*
 436          * First, look for old connections.
 437          * We probably should do it every X seconds, but what for?
 438          * It is only dangerous if an attacker wants to overload connections
 439          * queue, so here is a good place to do the cleanups.
 440          */
 441         connection_cleanups();
 442
 443         conn = malloc(sizeof(*conn));
 444         if (conn == NULL)
 445                 return (NULL);
 446         conn->c_path = strdup(cinit->gc_path);
 447         if (conn->c_path == NULL) {
 448                 free(conn);
 449                 return (NULL);
 450         }
 451         conn->c_token = cinit->gc_token;
 452         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 453         conn->c_srcip = ip;
 454         conn->c_sendfd = conn->c_recvfd = -1;
 455         if ((cinit->gc_flags & GGATE_FLAG_SEND) != 0)
 456                 conn->c_sendfd = sfd;
 457         else
 458                 conn->c_recvfd = sfd;
 459         conn->c_mediasize = 0;
 460         conn->c_sectorsize = 0;
 461         time(&conn->c_birthtime);
 462         conn->c_flags = cinit->gc_flags;
 463         LIST_INSERT_HEAD(&connections, conn, c_next);
 464         g_gate_log(LOG_DEBUG, "Connection created [%s, %s].", ip2str(ip),
 465             conn->c_path);
 466         return (conn);
 467 }
 468
 469 static int
 470 connection_add(struct ggd_connection *conn, struct g_gate_cinit *cinit,
 471     struct sockaddr *s, int sfd)
 472 {
 473         in_addr_t ip;
 474
 475         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 476         if ((cinit->gc_flags & GGATE_FLAG_SEND) != 0) {
 477                 if (conn->c_sendfd != -1) {
 478                         g_gate_log(LOG_WARNING,
 479                             "Send socket already exists [%s, %s].", ip2str(ip),
 480                             conn->c_path);
 481                         return (EEXIST);
 482                 }
 483                 conn->c_sendfd = sfd;
 484         } else {
 485                 if (conn->c_recvfd != -1) {
 486                         g_gate_log(LOG_WARNING,
 487                             "Receive socket already exists [%s, %s].",
 488                             ip2str(ip), conn->c_path);
 489                         return (EEXIST);
 490                 }
 491                 conn->c_recvfd = sfd;
 492         }
 493         g_gate_log(LOG_DEBUG, "Connection added [%s, %s].", ip2str(ip),
 494             conn->c_path);
 495         return (0);
 496 }
 497
 498 /*
 499  * Remove one socket from the given connection or the whole
 500  * connection if sfd == -1.
 501  */
 502 static void
 503 connection_remove(struct ggd_connection *conn)
 504 {
 505
 506         LIST_REMOVE(conn, c_next);
 507         g_gate_log(LOG_DEBUG, "Connection removed [%s %s].",
 508             ip2str(conn->c_srcip), conn->c_path);
 509         if (conn->c_sendfd != -1)
 510                 close(conn->c_sendfd);
 511         if (conn->c_recvfd != -1)
 512                 close(conn->c_recvfd);
 513         free(conn->c_path);
 514         free(conn);
 515 }
 516
 517 static int
 518 connection_ready(struct ggd_connection *conn)
 519 {
 520
 521         return (conn->c_sendfd != -1 && conn->c_recvfd != -1);
 522 }
 523
 524 static void
 525 connection_launch(struct ggd_connection *conn)
 526 {
 527         pthread_t td;
 528         int error, pid;
 529
 530         pid = fork();
 531         if (pid > 0)
 532                 return;
 533         else if (pid == -1) {
 534                 g_gate_log(LOG_ERR, "Cannot fork: %s.", strerror(errno));
 535                 return;
 536         }
 537         g_gate_log(LOG_DEBUG, "Process created [%s].", conn->c_path);
 538
 539         /*
 540          * Create condition variables and mutexes for in-queue and out-queue
 541          * synchronization.
 542          */
 543         error = pthread_mutex_init(&inqueue_mtx, NULL);
 544         if (error != 0) {
 545                 g_gate_xlog("pthread_mutex_init(inqueue_mtx): %s.",
 546                     strerror(error));
 547         }
 548         error = pthread_cond_init(&inqueue_cond, NULL);
 549         if (error != 0) {
 550                 g_gate_xlog("pthread_cond_init(inqueue_cond): %s.",
 551                     strerror(error));
 552         }
 553         error = pthread_mutex_init(&outqueue_mtx, NULL);
 554         if (error != 0) {
 555                 g_gate_xlog("pthread_mutex_init(outqueue_mtx): %s.",
 556                     strerror(error));
 557         }
 558         error = pthread_cond_init(&outqueue_cond, NULL);
 559         if (error != 0) {
 560                 g_gate_xlog("pthread_cond_init(outqueue_cond): %s.",
 561                     strerror(error));
 562         }
 563
 564         /*
 565          * Create threads:
 566          * recvtd - thread for receiving I/O request
 567          * diskio - thread for doing I/O request
 568          * sendtd - thread for sending I/O requests back
 569          */
 570         error = pthread_create(&td, NULL, send_thread, conn);
 571         if (error != 0) {
 572                 g_gate_xlog("pthread_create(send_thread): %s.",
 573                     strerror(error));
 574         }
 575         error = pthread_create(&td, NULL, recv_thread, conn);
 576         if (error != 0) {
 577                 g_gate_xlog("pthread_create(recv_thread): %s.",
 578                     strerror(error));
 579         }
 580         disk_thread(conn);
 581 }
 582
 583 static void
 584 sendfail(int sfd, int error, const char *fmt, ...)
 585 {
 586         struct g_gate_sinit sinit;
 587         va_list ap;
 588         ssize_t data;
 589
 590         sinit.gs_error = error;
 591         g_gate_swap2n_sinit(&sinit);
 592         data = g_gate_send(sfd, &sinit, sizeof(sinit), 0);
 593         g_gate_swap2h_sinit(&sinit);
 594         if (data != sizeof(sinit)) {
 595                 g_gate_log(LOG_WARNING, "Cannot send initial packet: %s.",
 596                     strerror(errno));
 597                 return;
 598         }
 599         if (fmt != NULL) {
 600                 va_start(ap, fmt);
 601                 g_gate_vlog(LOG_WARNING, fmt, ap);
 602                 va_end(ap);
 603         }
 604 }
 605
 606 static void *
 607 malloc_waitok(size_t size)
 608 {
 609         void *p;
 610
 611         while ((p = malloc(size)) == NULL) {
 612                 g_gate_log(LOG_DEBUG, "Cannot allocate %zu bytes.", size);
 613                 sleep(1);
 614         }
 615         return (p);
 616 }
 617
 618 static void *
 619 recv_thread(void *arg)
 620 {
 621         struct ggd_connection *conn;
 622         struct ggd_request *req;
 623         ssize_t data;
 624         int error, fd;
 625
 626         conn = arg;
 627         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 628         fd = conn->c_recvfd;
 629         for (;;) {
 630                 /*
 631                  * Get header packet.
 632                  */
 633                 req = malloc_waitok(sizeof(*req));
 634                 data = g_gate_recv(fd, &req->r_hdr, sizeof(req->r_hdr),
 635                     MSG_WAITALL);
 636                 if (data == 0) {
 637                         g_gate_log(LOG_DEBUG, "Process %u exiting.", getpid());
 638                         exit(EXIT_SUCCESS);
 639                 } else if (data == -1) {
 640                         g_gate_xlog("Error while receiving hdr packet: %s.",
 641                             strerror(errno));
 642                 } else if (data != sizeof(req->r_hdr)) {
 643                         g_gate_xlog("Malformed hdr packet received.");
 644                 }
 645                 g_gate_log(LOG_DEBUG, "Received hdr packet.");
 646                 g_gate_swap2h_hdr(&req->r_hdr);
 647
 648                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 649                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 650
 651                 /*
 652                  * Allocate memory for data.
 653                  */
 654                 req->r_data = malloc_waitok(req->r_length);
 655
 656                 /*
 657                  * Receive data to write for WRITE request.
 658                  */
 659                 if (req->r_cmd == GGATE_CMD_WRITE) {
 660                         g_gate_log(LOG_DEBUG, "Waiting for %u bytes of data...",
 661                             req->r_length);
 662                         data = g_gate_recv(fd, req->r_data, req->r_length,
 663                             MSG_WAITALL);
 664                         if (data == -1) {
 665                                 g_gate_xlog("Error while receiving data: %s.",
 666                                     strerror(errno));
 667                         }
 668                 }
 669
 670                 /*
 671                  * Put the request onto the incoming queue.
 672                  */
 673                 error = pthread_mutex_lock(&inqueue_mtx);
 674                 assert(error == 0);
 675                 TAILQ_INSERT_TAIL(&inqueue, req, r_next);
 676                 error = pthread_cond_signal(&inqueue_cond);
 677                 assert(error == 0);
 678                 error = pthread_mutex_unlock(&inqueue_mtx);
 679                 assert(error == 0);
 680         }
 681 }
 682
 683 static void *
 684 disk_thread(void *arg)
 685 {
 686         struct ggd_connection *conn;
 687         struct ggd_request *req;
 688         ssize_t data;
 689         int error, fd;
 690
 691         conn = arg;
 692         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 693         fd = conn->c_diskfd;
 694         for (;;) {
 695                 /*
 696                  * Get a request from the incoming queue.
 697                  */
 698                 error = pthread_mutex_lock(&inqueue_mtx);
 699                 assert(error == 0);
 700                 while ((req = TAILQ_FIRST(&inqueue)) == NULL) {
 701                         error = pthread_cond_wait(&inqueue_cond, &inqueue_mtx);
 702                         assert(error == 0);
 703                 }
 704                 TAILQ_REMOVE(&inqueue, req, r_next);
 705                 error = pthread_mutex_unlock(&inqueue_mtx);
 706                 assert(error == 0);
 707
 708                 /*
 709                  * Check the request.
 710                  */
 711                 assert(req->r_cmd == GGATE_CMD_READ || req->r_cmd == GGATE_CMD_WRITE);
 712                 assert(req->r_offset + req->r_length <= (uintmax_t)conn->c_mediasize);
 713                 assert((req->r_offset % conn->c_sectorsize) == 0);
 714                 assert((req->r_length % conn->c_sectorsize) == 0);
 715
 716                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 717                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 718
 719                 /*
 720                  * Do the request.
 721                  */
 722                 data = 0;
 723                 switch (req->r_cmd) {
 724                 case GGATE_CMD_READ:
 725                         data = pread(fd, req->r_data, req->r_length,
 726                             req->r_offset);
 727                         break;
 728                 case GGATE_CMD_WRITE:
 729                         data = pwrite(fd, req->r_data, req->r_length,
 730                             req->r_offset);
 731                         /* Free data memory here - better sooner. */
 732                         free(req->r_data);
 733                         req->r_data = NULL;
 734                         break;
 735                 }
 736                 if (data != (ssize_t)req->r_length) {
 737                         /* Report short reads/writes as I/O errors. */
 738                         if (errno == 0)
 739                                 errno = EIO;
 740                         g_gate_log(LOG_ERR, "Disk error: %s", strerror(errno));
 741                         req->r_error = errno;
 742                         if (req->r_data != NULL) {
 743                                 free(req->r_data);
 744                                 req->r_data = NULL;
 745                         }
 746                 }
 747
 748                 /*
 749                  * Put the request onto the outgoing queue.
 750                  */
 751                 error = pthread_mutex_lock(&outqueue_mtx);
 752                 assert(error == 0);
 753                 TAILQ_INSERT_TAIL(&outqueue, req, r_next);
 754                 error = pthread_cond_signal(&outqueue_cond);
 755                 assert(error == 0);
 756                 error = pthread_mutex_unlock(&outqueue_mtx);
 757                 assert(error == 0);
 758         }
 759
 760         /* NOTREACHED */
 761         return (NULL);
 762 }
 763
 764 static void *
 765 send_thread(void *arg)
 766 {
 767         struct ggd_connection *conn;
 768         struct ggd_request *req;
 769         ssize_t data;
 770         int error, fd;
 771
 772         conn = arg;
 773         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 774         fd = conn->c_sendfd;
 775         for (;;) {
 776                 /*
 777                  * Get a request from the outgoing queue.
 778                  */
 779                 error = pthread_mutex_lock(&outqueue_mtx);
 780                 assert(error == 0);
 781                 while ((req = TAILQ_FIRST(&outqueue)) == NULL) {
 782                         error = pthread_cond_wait(&outqueue_cond,
 783                             &outqueue_mtx);
 784                         assert(error == 0);
 785                 }
 786                 TAILQ_REMOVE(&outqueue, req, r_next);
 787                 error = pthread_mutex_unlock(&outqueue_mtx);
 788                 assert(error == 0);
 789
 790                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 791                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 792
 793                 /*
 794                  * Send the request.
 795                  */
 796                 g_gate_swap2n_hdr(&req->r_hdr);
 797                 if (g_gate_send(fd, &req->r_hdr, sizeof(req->r_hdr), 0) == -1) {
 798                         g_gate_xlog("Error while sending hdr packet: %s.",
 799                             strerror(errno));
 800                 }
 801                 g_gate_log(LOG_DEBUG, "Sent hdr packet.");
 802                 g_gate_swap2h_hdr(&req->r_hdr);
 803                 if (req->r_data != NULL) {
 804                         data = g_gate_send(fd, req->r_data, req->r_length, 0);
 805                         if (data != (ssize_t)req->r_length) {
 806                                 g_gate_xlog("Error while sending data: %s.",
 807                                     strerror(errno));
 808                         }
 809                         g_gate_log(LOG_DEBUG,
 810                             "Sent %zd bytes (offset=%ju, size=%zu).", data,
 811                             (uintmax_t)req->r_offset, (size_t)req->r_length);
 812                         free(req->r_data);
 813                 }
 814                 free(req);
 815         }
 816
 817         /* NOTREACHED */
 818         return (NULL);
 819 }
 820
 821 static void
 822 log_connection(struct sockaddr *from)
 823 {
 824         in_addr_t ip;
 825
 826         ip = htonl(((struct sockaddr_in *)(void *)from)->sin_addr.s_addr);
 827         g_gate_log(LOG_INFO, "Connection from: %s.", ip2str(ip));
 828 }
 829
 830 static int
 831 handshake(struct sockaddr *from, int sfd)
 832 {
 833         struct g_gate_version ver;
 834         struct g_gate_cinit cinit;
 835         struct g_gate_sinit sinit;
 836         struct ggd_connection *conn;
 837         struct ggd_export *ex;
 838         ssize_t data;
 839
 840         log_connection(from);
 841         /*
 842          * Phase 1: Version verification.
 843          */
 844         g_gate_log(LOG_DEBUG, "Receiving version packet.");
 845         data = g_gate_recv(sfd, &ver, sizeof(ver), MSG_WAITALL);
 846         g_gate_swap2h_version(&ver);
 847         if (data != sizeof(ver)) {
 848                 g_gate_log(LOG_WARNING, "Malformed version packet.");
 849                 return (0);
 850         }
 851         g_gate_log(LOG_DEBUG, "Version packet received.");
 852         if (memcmp(ver.gv_magic, GGATE_MAGIC, strlen(GGATE_MAGIC)) != 0) {
 853                 g_gate_log(LOG_WARNING, "Invalid magic field.");
 854                 return (0);
 855         }
 856         if (ver.gv_version != GGATE_VERSION) {
 857                 g_gate_log(LOG_WARNING, "Version %u is not supported.",
 858                     ver.gv_version);
 859                 return (0);
 860         }
 861         ver.gv_error = 0;
 862         g_gate_swap2n_version(&ver);
 863         data = g_gate_send(sfd, &ver, sizeof(ver), 0);
 864         g_gate_swap2h_version(&ver);
 865         if (data == -1) {
 866                 sendfail(sfd, errno, "Error while sending version packet: %s.",
 867                     strerror(errno));
 868                 return (0);
 869         }
 870
 871         /*
 872          * Phase 2: Request verification.
 873          */
 874         g_gate_log(LOG_DEBUG, "Receiving initial packet.");
 875         data = g_gate_recv(sfd, &cinit, sizeof(cinit), MSG_WAITALL);
 876         g_gate_swap2h_cinit(&cinit);
 877         if (data != sizeof(cinit)) {
 878                 g_gate_log(LOG_WARNING, "Malformed initial packet.");
 879                 return (0);
 880         }
 881         g_gate_log(LOG_DEBUG, "Initial packet received.");
 882         conn = connection_find(&cinit);
 883         if (conn != NULL) {
 884                 /*
 885                  * Connection should already exists.
 886                  */
 887                 g_gate_log(LOG_DEBUG, "Found existing connection (token=%lu).",
 888                     (unsigned long)conn->c_token);
 889                 if (connection_add(conn, &cinit, from, sfd) == -1) {
 890                         connection_remove(conn);
 891                         return (0);
 892                 }
 893         } else {
 894                 /*
 895                  * New connection, allocate space.
 896                  */
 897                 conn = connection_new(&cinit, from, sfd);
 898                 if (conn == NULL) {
 899                         sendfail(sfd, ENOMEM,
 900                             "Cannot allocate new connection.");
 901                         return (0);
 902                 }
 903                 g_gate_log(LOG_DEBUG, "New connection created (token=%lu).",
 904                     (unsigned long)conn->c_token);
 905         }
 906
 907         ex = exports_find(from, &cinit, conn);
 908         if (ex == NULL) {
 909                 connection_remove(conn);
 910                 sendfail(sfd, errno, NULL);
 911                 return (0);
 912         }
 913         if (conn->c_mediasize == 0) {
 914                 conn->c_mediasize = g_gate_mediasize(conn->c_diskfd);
 915                 conn->c_sectorsize = g_gate_sectorsize(conn->c_diskfd);
 916         }
 917         sinit.gs_mediasize = conn->c_mediasize;
 918         sinit.gs_sectorsize = conn->c_sectorsize;
 919         sinit.gs_error = 0;
 920
 921         g_gate_log(LOG_DEBUG, "Sending initial packet.");
 922
 923         g_gate_swap2n_sinit(&sinit);
 924         data = g_gate_send(sfd, &sinit, sizeof(sinit), 0);
 925         g_gate_swap2h_sinit(&sinit);
 926         if (data == -1) {
 927                 sendfail(sfd, errno, "Error while sending initial packet: %s.",
 928                     strerror(errno));
 929                 return (0);
 930         }
 931
 932         if (connection_ready(conn)) {
 933                 connection_launch(conn);
 934                 connection_remove(conn);
 935         }
 936         return (1);
 937 }
 938
 939 static void
 940 huphandler(int sig __unused)
 941 {
 942
 943         got_sighup = 1;
 944 }
 945
 946 int
 947 main(int argc, char *argv[])
 948 {
 949         struct sockaddr_in serv;
 950         struct sockaddr from;
 951         socklen_t fromlen;
 952         int sfd, tmpsfd;
 953         unsigned port;
 954
 955         bindaddr = htonl(INADDR_ANY);
 956         port = G_GATE_PORT;
 957         for (;;) {
 958                 int ch;
 959
 960                 ch = getopt(argc, argv, "a:hnp:R:S:v");
 961                 if (ch == -1)
 962                         break;
 963                 switch (ch) {
 964                 case 'a':
 965                         bindaddr = g_gate_str2ip(optarg);
 966                         if (bindaddr == INADDR_NONE) {
 967                                 errx(EXIT_FAILURE,
 968                                     "Invalid IP/host name to bind to.");
 969                         }
 970                         break;
 971                 case 'n':
 972                         nagle = 0;
 973                         break;
 974                 case 'p':
 975                         errno = 0;
 976                         port = strtoul(optarg, NULL, 10);
 977                         if (port == 0 && errno != 0)
 978                                 errx(EXIT_FAILURE, "Invalid port.");
 979                         break;
 980                 case 'R':
 981                         errno = 0;
 982                         rcvbuf = strtoul(optarg, NULL, 10);
 983                         if (rcvbuf == 0 && errno != 0)
 984                                 errx(EXIT_FAILURE, "Invalid rcvbuf.");
 985                         break;
 986                 case 'S':
 987                         errno = 0;
 988                         sndbuf = strtoul(optarg, NULL, 10);
 989                         if (sndbuf == 0 && errno != 0)
 990                                 errx(EXIT_FAILURE, "Invalid sndbuf.");
 991                         break;
 992                 case 'v':
 993                         g_gate_verbose++;
 994                         break;
 995                 case 'h':
 996                 default:
 997                         usage();
 998                 }
 999         }
1000         argc -= optind;
1001         argv += optind;
1002
1003         if (argv[0] != NULL)
1004                 exports_file = argv[0];
1005         exports_get();
1006
1007         if (!g_gate_verbose) {
1008                 /* Run in daemon mode. */
1009                 if (daemon(0, 0) == -1)
1010                         g_gate_xlog("Cannot daemonize: %s", strerror(errno));
1011         }
1012
1013         signal(SIGCHLD, SIG_IGN);
1014
1015         sfd = socket(AF_INET, SOCK_STREAM, 0);
1016         if (sfd == -1)
1017                 g_gate_xlog("Cannot open stream socket: %s.", strerror(errno));
1018         bzero(&serv, sizeof(serv));
1019         serv.sin_family = AF_INET;
1020         serv.sin_addr.s_addr = bindaddr;
1021         serv.sin_port = htons(port);
1022
1023         g_gate_socket_settings(sfd);
1024
1025         if (bind(sfd, (struct sockaddr *)&serv, sizeof(serv)) == -1)
1026                 g_gate_xlog("bind(): %s.", strerror(errno));
1027         if (listen(sfd, 5) == -1)
1028                 g_gate_xlog("listen(): %s.", strerror(errno));
1029
1030         g_gate_log(LOG_INFO, "Listen on port: %d.", port);
1031
1032         signal(SIGHUP, huphandler);
1033
1034         for (;;) {
1035                 fromlen = sizeof(from);
1036                 tmpsfd = accept(sfd, &from, &fromlen);
1037                 if (tmpsfd == -1)
1038                         g_gate_xlog("accept(): %s.", strerror(errno));
1039
1040                 if (got_sighup) {
1041                         got_sighup = 0;
1042                         exports_get();
1043                 }
1044
1045                 if (!handshake(&from, tmpsfd))
1046                         close(tmpsfd);
1047         }
1048         close(sfd);
1049         exit(EXIT_SUCCESS);
1050 }