binlog.c

   1 /* binlog.c - binary log implementation */
   2
   3 /* Copyright (C) 2008 Graham Barr
   4
   5  * This program is free software: you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation, either version 3 of the License, or
   8  * (at your option) any later version.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include "config.h"
  20
  21 #if HAVE_STDINT_H
  22 # include <stdint.h>
  23 #endif /* else we get int types from config.h */
  24
  25 #include <stdlib.h>
  26 #include <stdio.h>
  27 #include <fcntl.h>
  28 #include <unistd.h>
  29 #include <string.h>
  30 #include <errno.h>
  31 #include <dirent.h>
  32 #include <sys/resource.h>
  33 #include <sys/param.h>
  34 #include <sys/uio.h>
  35 #include <sys/stat.h>
  36 #include <stdarg.h>
  37 #include <limits.h>
  38 #include <stddef.h>
  39
  40 #include "tube.h"
  41 #include "job.h"
  42 #include "binlog.h"
  43 #include "util.h"
  44 #include "config.h"
  45
  46 typedef struct binlog *binlog;
  47
  48 struct binlog {
  49   binlog next;
  50   unsigned int refs;
  51   int fd;
  52   size_t free;
  53   size_t reserved;
  54   char path[];
  55 };
  56
  57 /* max size we will create a log file */
  58 size_t binlog_size_limit = BINLOG_SIZE_LIMIT_DEFAULT;
  59
  60 int enable_fsync = 0;
  61 size_t fsync_throttle_ms = 0;
  62 uint64_t last_fsync = 0;
  63
  64 char *binlog_dir = NULL;
  65 static int binlog_index = 0;
  66 static int binlog_version = 3;
  67 static int lock_fd;
  68
  69 static binlog oldest_binlog = 0,
  70               current_binlog = 0,
  71               newest_binlog = 0;
  72
  73 static const size_t job_record_size = offsetof(struct job, pad);
  74
  75 static int
  76 binlog_scan_dir()
  77 {
  78     DIR *dirp;
  79     struct dirent *dp;
  80     long min = 0;
  81     long max = 0;
  82     long val;
  83     char *endptr;
  84     size_t name_len;
  85
  86     dirp = opendir(binlog_dir);
  87     if (!dirp) return 0;
  88
  89     while ((dp = readdir(dirp)) != NULL) {
  90         name_len = strlen(dp->d_name);
  91         if (name_len > 7 && !strncmp("binlog.", dp->d_name, 7)) {
  92             val = strtol(dp->d_name + 7, &endptr, 10);
  93             if (endptr && *endptr == 0) {
  94                 if (max == 0 || val > max) max = val;
  95                 if (min == 0 || val < min) min = val;
  96             }
  97         }
  98     }
  99
 100     closedir(dirp);
 101     binlog_index = (int) max;
 102     return (int) min;
 103 }
 104
 105 static void
 106 binlog_remove_oldest()
 107 {
 108     binlog b = oldest_binlog;
 109
 110     if (!b) return;
 111
 112     oldest_binlog = b->next;
 113
 114     unlink(b->path);
 115     free(b);
 116 }
 117
 118 static binlog
 119 binlog_iref(binlog b)
 120 {
 121     if (b) b->refs++;
 122     return b;
 123 }
 124
 125 static void
 126 binlog_dref(binlog b)
 127 {
 128     if (!b) return;
 129     if (b->refs < 1) return twarnx("refs is zero for binlog: %s", b->path);
 130
 131     --b->refs;
 132     if (b->refs < 1) {
 133         while (oldest_binlog && oldest_binlog->refs == 0) {
 134             binlog_remove_oldest();
 135         }
 136     }
 137 }
 138
 139 /*
 140 static void
 141 binlog_warn(int fd, const char* path, const char *msg)
 142 {
 143     warnx("WARNING, %s at %s:%u.\n%s", msg, path, lseek(fd, 0, SEEK_CUR),
 144           "  Continuing. You may be missing data.");
 145 }
 146 */
 147
 148 #define binlog_warn(b, fmt, args...) \
 149     warnx("WARNING, " fmt " at %s:%u. %s: ", \
 150           ##args, b->path, lseek(b->fd, 0, SEEK_CUR), \
 151           "Continuing. You may be missing data.")
 152
 153 static void
 154 binlog_read_log_file(binlog b, job binlog_jobs)
 155 {
 156     struct job js;
 157     tube t;
 158     job j;
 159     char tubename[MAX_TUBE_NAME_LEN];
 160     size_t namelen;
 161     ssize_t r;
 162     int version;
 163
 164     r = read(b->fd, &version, sizeof(version));
 165     if (r == -1) return twarn("read()");
 166     if (r < sizeof(version)) {
 167         return binlog_warn(b, "EOF while reading version record");
 168     }
 169
 170     if (version != binlog_version) {
 171         return warnx("%s: binlog version mismatch %d %d", b->path, version,
 172                      binlog_version);
 173     }
 174
 175     while (read(b->fd, &namelen, sizeof(size_t)) == sizeof(size_t)) {
 176         if (namelen > 0) {
 177             r = read(b->fd, tubename, namelen);
 178             if (r == -1) return twarn("read()");
 179             if (r < namelen) {
 180                 lseek(b->fd, SEEK_CUR, 0);
 181                 return binlog_warn(b, "EOF while reading tube name");
 182             }
 183         }
 184
 185         tubename[namelen] = '\0';
 186         r = read(b->fd, &js, job_record_size);
 187         if (r == -1) return twarn("read()");
 188         if (r < job_record_size) {
 189           return binlog_warn(b, "EOF while reading job record");
 190         }
 191
 192         if (!js.id) break;
 193
 194         j = job_find(js.id);
 195         switch (js.state) {
 196         case JOB_STATE_INVALID:
 197             if (j) {
 198                 job_remove(j);
 199                 binlog_dref(j->binlog);
 200                 job_free(j);
 201                 j = NULL;
 202             }
 203             break;
 204         case JOB_STATE_READY:
 205         case JOB_STATE_DELAYED:
 206             if (!j && namelen > 0) {
 207                 t = tube_find_or_make(tubename);
 208                 j = make_job_with_id(js.pri, js.delay, js.ttr, js.body_size,
 209                                      t, js.id);
 210                 j->next = j->prev = j;
 211                 j->creation = js.creation;
 212                 job_insert(binlog_jobs, j);
 213             }
 214             if (js.body_size) {
 215                 if (js.body_size > j->body_size) {
 216                     warnx("job size increased from %zu to %zu", j->body_size,
 217                           js.body_size);
 218                     job_remove(j);
 219                     binlog_dref(j->binlog);
 220                     job_free(j);
 221                     return binlog_warn(b, "EOF while reading job body");
 222                 }
 223                 r = read(b->fd, j->body, js.body_size);
 224                 if (r == -1) return twarn("read()");
 225                 if (r < js.body_size) {
 226                     warnx("dropping incomplete job %llu", j->id);
 227                     job_remove(j);
 228                     binlog_dref(j->binlog);
 229                     job_free(j);
 230                     return binlog_warn(b, "EOF while reading job body");
 231                 }
 232             }
 233             break;
 234         }
 235         if (j) {
 236             j->state = js.state;
 237             j->deadline = js.deadline;
 238             j->pri = js.pri;
 239             j->delay = js.delay;
 240             j->ttr = js.ttr;
 241             j->timeout_ct = js.timeout_ct;
 242             j->release_ct = js.release_ct;
 243             j->bury_ct = js.bury_ct;
 244             j->kick_ct = js.kick_ct;
 245
 246             /* this is a complete record, so we can move the binlog ref */
 247             if (namelen && js.body_size) {
 248                 binlog_dref(j->binlog);
 249                 j->binlog = binlog_iref(b);
 250             }
 251         }
 252     }
 253 }
 254
 255 static void
 256 binlog_close(binlog b)
 257 {
 258     if (!b) return;
 259     if (b->fd < 0) return;
 260     close(b->fd);
 261     b->fd = -1;
 262     binlog_dref(b);
 263 }
 264
 265 static binlog
 266 make_binlog(char *path)
 267 {
 268     binlog b;
 269
 270     b = (binlog) malloc(sizeof(struct binlog) + strlen(path) + 1);
 271     if (!b) return twarnx("OOM"), (binlog) 0;
 272     strcpy(b->path, path);
 273     b->refs = 0;
 274     b->next = NULL;
 275     b->fd = -1;
 276     b->free = 0;
 277     b->reserved = 0;
 278     return b;
 279 }
 280
 281 static binlog
 282 make_next_binlog()
 283 {
 284     int r;
 285     char path[PATH_MAX];
 286
 287     if (!binlog_dir) return NULL;
 288
 289     r = snprintf(path, PATH_MAX, "%s/binlog.%d", binlog_dir, ++binlog_index);
 290     if (r > PATH_MAX) return twarnx("path too long: %s", binlog_dir), (binlog)0;
 291
 292     return make_binlog(path);
 293 }
 294
 295 static binlog
 296 add_binlog(binlog b)
 297 {
 298     if (newest_binlog) newest_binlog->next = b;
 299     newest_binlog = b;
 300     if (!oldest_binlog) oldest_binlog = b;
 301
 302     return b;
 303 }
 304
 305 static void
 306 binlog_open(binlog log, size_t *written)
 307 {
 308     int fd;
 309     size_t bytes_written;
 310
 311     if (written) *written = 0;
 312
 313     if (!binlog_iref(log)) return;
 314
 315     fd = open(log->path, O_WRONLY | O_CREAT, 0400);
 316
 317     if (fd < 0) return twarn("Cannot open binlog %s", log->path);
 318
 319 #ifdef HAVE_POSIX_FALLOCATE
 320     {
 321         int r;
 322         r = posix_fallocate(fd, 0, binlog_size_limit);
 323         if (r) {
 324             close(fd);
 325             binlog_dref(log);
 326             errno = r;
 327             return twarn("Cannot allocate space for binlog %s", log->path);
 328         }
 329     }
 330 #else
 331     /* Allocate space in a slow but portable way. */
 332     {
 333         size_t i;
 334         ssize_t w;
 335         off_t p;
 336         #define ZERO_BUF_SIZE 512
 337         char buf[ZERO_BUF_SIZE] = {}; /* initialize to zero */
 338
 339         for (i = 0; i < binlog_size_limit; i += w) {
 340             w = write(fd, &buf, ZERO_BUF_SIZE);
 341             if (w == -1) {
 342                 twarn("Cannot allocate space for binlog %s", log->path);
 343                 close(fd);
 344                 binlog_dref(log);
 345                 return;
 346             }
 347         }
 348
 349         p = lseek(fd, 0, SEEK_SET);
 350         if (p == -1) {
 351             twarn("lseek");
 352             close(fd);
 353             binlog_dref(log);
 354             return;
 355         }
 356     }
 357 #endif
 358
 359     bytes_written = write(fd, &binlog_version, sizeof(int));
 360     if (written) *written = bytes_written;
 361
 362     if (bytes_written < sizeof(int)) {
 363         twarn("Cannot write to binlog");
 364         close(fd);
 365         binlog_dref(log);
 366         return;
 367     }
 368
 369     log->fd = fd;
 370 }
 371
 372 /* returns 1 on success, 0 on error. */
 373 static int
 374 binlog_use_next()
 375 {
 376     binlog next;
 377
 378     if (!current_binlog) return 0;
 379
 380     next = current_binlog->next;
 381
 382     if (!next) return 0;
 383
 384     /* assert(current_binlog->reserved == 0); */
 385
 386     binlog_close(current_binlog);
 387     current_binlog = next;
 388
 389     return 1;
 390 }
 391
 392 void
 393 binlog_shutdown()
 394 {
 395     binlog_use_next();
 396     binlog_close(current_binlog);
 397 }
 398
 399 /* Returns the number of jobs successfully written (either 0 or 1).
 400
 401    If this fails, something is seriously wrong. It should never fail because of
 402    a full disk. (The binlog_reserve_space_* functions, on the other hand, can
 403    fail because of a full disk.)
 404
 405    If we are not using the binlog at all (!current_binlog), then we pretend to
 406    have made a successful write and return 1. */
 407 int
 408 binlog_write_job(job j)
 409 {
 410     ssize_t written;
 411     size_t tube_namelen, to_write = 0;
 412     struct iovec vec[4], *vptr;
 413     int vcnt = 3, r;
 414     uint64_t now;
 415
 416     if (!current_binlog) return 1;
 417     tube_namelen = 0;
 418
 419     vec[0].iov_base = (char *) &tube_namelen;
 420     to_write += vec[0].iov_len = sizeof(size_t);
 421
 422     vec[1].iov_base = j->tube->name;
 423     vec[1].iov_len = 0;
 424
 425     vec[2].iov_base = (char *) j;
 426     to_write += vec[2].iov_len = job_record_size;
 427
 428     if (j->state == JOB_STATE_READY || j->state == JOB_STATE_DELAYED) {
 429         if (!j->binlog) {
 430             tube_namelen = strlen(j->tube->name);
 431             to_write += vec[1].iov_len = tube_namelen;
 432             vcnt = 4;
 433             vec[3].iov_base = j->body;
 434             to_write += vec[3].iov_len = j->body_size;
 435         }
 436     } else if (j->state == JOB_STATE_INVALID) {
 437         if (j->binlog) binlog_dref(j->binlog);
 438         j->binlog = NULL;
 439     } else {
 440         return twarnx("unserializable job state: %d", j->state), 0;
 441     }
 442
 443     if (to_write > current_binlog->reserved) {
 444         r = binlog_use_next();
 445         if (!r) return twarnx("failed to use next binlog"), 0;
 446     }
 447
 448     if (j->state && !j->binlog) j->binlog = binlog_iref(current_binlog);
 449
 450     while (to_write > 0) {
 451         written = writev(current_binlog->fd, vec, vcnt);
 452
 453         if (written < 0) {
 454             if (errno == EAGAIN) continue;
 455             if (errno == EINTR) continue;
 456
 457             twarn("writev");
 458             binlog_close(current_binlog);
 459             current_binlog = 0;
 460             return 0;
 461         }
 462
 463         to_write -= written;
 464         if (to_write > 0 && written > 0) {
 465             for (vptr = vec; written >= vptr->iov_len; vptr++) {
 466                 written -= vptr->iov_len;
 467                 vptr->iov_len = 0;
 468             }
 469             vptr->iov_base = (char *) vptr->iov_base + written;
 470             vptr->iov_len -= written;
 471         }
 472         current_binlog->reserved -= written;
 473         j->reserved_binlog_space -= written;
 474     }
 475
 476     now = now_usec() / 1000; /* usec -> msec */
 477     if (enable_fsync && now - last_fsync >= fsync_throttle_ms) {
 478         r = fdatasync(current_binlog->fd);
 479         if (r == -1) return twarn("fdatasync"), 0;
 480         last_fsync = now;
 481     }
 482
 483     return 1;
 484 }
 485
 486 static binlog
 487 make_future_binlog()
 488 {
 489     binlog b;
 490     size_t header;
 491
 492     /* open a new binlog with more space to reserve */
 493     b = make_next_binlog();
 494     if (!b) return twarnx("error making next binlog"), (binlog) 0;
 495     binlog_open(b, &header);
 496
 497     /* open failed, so we can't reserve any space */
 498     if (b->fd < 0) {
 499         free(b);
 500         return 0;
 501     }
 502
 503     b->free = binlog_size_limit - header;
 504     b->reserved = 0;
 505     return b;
 506 }
 507
 508 static int
 509 can_move_reserved(size_t n, binlog from, binlog to)
 510 {
 511     return from->reserved >= n && to->free >= n;
 512 }
 513
 514 static void
 515 move_reserved(size_t n, binlog from, binlog to)
 516 {
 517     from->reserved -= n;
 518     from->free += n;
 519     to->reserved += n;
 520     to->free -= n;
 521 }
 522
 523 static size_t
 524 ensure_free_space(size_t n)
 525 {
 526     binlog fb;
 527
 528     if (newest_binlog && newest_binlog->free >= n) return n;
 529
 530     /* open a new binlog */
 531     fb = make_future_binlog();
 532     if (!fb) return twarnx("make_future_binlog"), 0;
 533
 534     add_binlog(fb);
 535     return n;
 536 }
 537
 538 /* Preserve some invariants immediately after any space reservation.
 539  * Invariant 1: current_binlog->reserved >= n.
 540  * Invariant 2: current_binlog->reserved is congruent to n (mod z), where z
 541  * is the size of a delete record in the binlog. */
 542 static size_t
 543 maintain_invariant(size_t n)
 544 {
 545     size_t reserved_later, remainder, complement, z, r;
 546
 547     /* In this function, reserved bytes are conserved (they are neither created
 548      * nor destroyed). We just move them around to preserve the invariant. We
 549      * might have to create new free space (i.e. allocate a new binlog file),
 550      * though. */
 551
 552     /* Invariant 1. */
 553     /* This is a loop, but it's guaranteed to run at most once. The proof is
 554      * left as an exercise for the reader. */
 555     while (current_binlog->reserved < n) {
 556         size_t to_move = current_binlog->reserved;
 557
 558         r = ensure_free_space(to_move);
 559         if (r != to_move) {
 560             twarnx("ensure_free_space");
 561             if (newest_binlog->reserved >= n) {
 562                 newest_binlog->reserved -= n;
 563             } else {
 564                 twarnx("failed to unreserve %zd bytes", n); /* can't happen */
 565             }
 566             return 0;
 567         }
 568
 569         move_reserved(to_move, current_binlog, newest_binlog);
 570         binlog_use_next();
 571     }
 572
 573
 574     /* Invariant 2. */
 575
 576     z = sizeof(size_t) + job_record_size;
 577     reserved_later = current_binlog->reserved - n;
 578     remainder = reserved_later % z;
 579     if (remainder == 0) return n;
 580     complement = z - remainder;
 581     if (can_move_reserved(complement, newest_binlog, current_binlog)) {
 582         move_reserved(complement, newest_binlog, current_binlog);
 583         return n;
 584     }
 585
 586     r = ensure_free_space(remainder);
 587     if (r != remainder) {
 588         twarnx("ensure_free_space");
 589         if (newest_binlog->reserved >= n) {
 590             newest_binlog->reserved -= n;
 591         } else {
 592             twarnx("failed to unreserve %zd bytes", n); /* can't happen */
 593         }
 594         return 0;
 595     }
 596     move_reserved(remainder, current_binlog, newest_binlog);
 597
 598     return n;
 599 }
 600
 601 /* Returns the number of bytes successfully reserved: either 0 or n. */
 602 static size_t
 603 binlog_reserve_space(size_t n)
 604 {
 605     size_t r;
 606
 607     /* This return value must be nonzero but is otherwise ignored. */
 608     if (!current_binlog) return 1;
 609
 610     if (current_binlog->free >= n) {
 611         current_binlog->free -= n;
 612         current_binlog->reserved += n;
 613         return maintain_invariant(n);
 614     }
 615
 616     r = ensure_free_space(n);
 617     if (r != n) return twarnx("ensure_free_space"), 0;
 618
 619     newest_binlog->free -= n;
 620     newest_binlog->reserved += n;
 621     return maintain_invariant(n);
 622 }
 623
 624 /* Returns the number of bytes reserved. */
 625 size_t
 626 binlog_reserve_space_put(job j)
 627 {
 628     size_t z = 0;
 629
 630     /* reserve space for the initial job record */
 631     z += sizeof(size_t);
 632     z += strlen(j->tube->name);
 633     z += job_record_size;
 634     z += j->body_size;
 635
 636     /* plus space for a delete to come later */
 637     z += sizeof(size_t);
 638     z += job_record_size;
 639
 640     return binlog_reserve_space(z);
 641 }
 642
 643 size_t
 644 binlog_reserve_space_update(job j)
 645 {
 646     size_t z = 0;
 647
 648     z += sizeof(size_t);
 649     z += job_record_size;
 650     return binlog_reserve_space(z);
 651 }
 652
 653 int
 654 binlog_lock()
 655 {
 656     int r;
 657     struct flock lock;
 658     char path[PATH_MAX];
 659
 660     r = snprintf(path, PATH_MAX, "%s/lock", binlog_dir);
 661     if (r > PATH_MAX) return twarnx("path too long: %s", binlog_dir), 0;
 662
 663     lock_fd = open(path, O_WRONLY|O_CREAT, 0600);
 664     if (lock_fd == -1) return twarn("open"), 0;
 665
 666     lock.l_type = F_WRLCK;
 667     lock.l_whence = SEEK_SET;
 668     lock.l_start = 0;
 669     lock.l_len = 0;
 670     r = fcntl(lock_fd, F_SETLK, &lock);
 671     if (r) return twarn("fcntl"), 0;
 672
 673     return 1;
 674 }
 675
 676 void
 677 binlog_init(job binlog_jobs)
 678 {
 679     int binlog_index_min;
 680     struct stat sbuf;
 681     int fd, idx, r;
 682     size_t n;
 683     char path[PATH_MAX];
 684     binlog b;
 685
 686     if (!binlog_dir) return;
 687
 688     /* Recover any jobs in old binlogs */
 689
 690     if (stat(binlog_dir, &sbuf) < 0) {
 691         if (mkdir(binlog_dir, 0700) < 0) return twarn("%s", binlog_dir);
 692     } else if (!(sbuf.st_mode & S_IFDIR)) {
 693         twarnx("%s", binlog_dir);
 694         return;
 695     }
 696
 697     binlog_index_min = binlog_scan_dir();
 698
 699     if (binlog_index_min) {
 700         for (idx = binlog_index_min; idx <= binlog_index; idx++) {
 701             r = snprintf(path, PATH_MAX, "%s/binlog.%d", binlog_dir, idx);
 702             if (r > PATH_MAX) return twarnx("path too long: %s", binlog_dir);
 703
 704             fd = open(path, O_RDONLY);
 705
 706             if (fd < 0) {
 707                 twarn("%s", path);
 708             } else {
 709                 b = binlog_iref(add_binlog(make_binlog(path)));
 710                 b->fd = fd;
 711                 binlog_read_log_file(b, binlog_jobs);
 712                 close(fd);
 713                 b->fd = -1;
 714                 binlog_dref(b);
 715             }
 716         }
 717
 718     }
 719
 720
 721     /* Set up for writing out new jobs */
 722     n = ensure_free_space(1);
 723     if (!n) return twarnx("error making first writable binlog");
 724
 725     current_binlog = newest_binlog;
 726 }
 727
 728 const char *
 729 binlog_oldest_index()
 730 {
 731     if (!oldest_binlog) return "0";
 732
 733     return strrchr(oldest_binlog->path, '.') + 1;
 734 }
 735
 736 const char *
 737 binlog_current_index()
 738 {
 739     if (!newest_binlog) return "0";
 740
 741     return strrchr(newest_binlog->path, '.') + 1;
 742 }