git.ozlabs.org Git - ccan/blob - ccan/tdb/tools/replay_trace.c

   1 #include <ccan/tdb/tdb.h>
   2 #include <ccan/grab_file/grab_file.h>
   3 #include <ccan/hash/hash.h>
   4 #include <ccan/talloc/talloc.h>
   5 #include <ccan/str_talloc/str_talloc.h>
   6 #include <ccan/str/str.h>
   7 #include <ccan/list/list.h>
   8 #include <err.h>
   9 #include <ctype.h>
  10 #include <string.h>
  11 #include <unistd.h>
  12 #include <sys/types.h>
  13 #include <sys/wait.h>
  14 #include <sys/time.h>
  15 #include <errno.h>
  16 #include <signal.h>
  17
  18 #define STRINGIFY2(x) #x
  19 #define STRINGIFY(x) STRINGIFY2(x)
  20
  21 /* Avoid mod by zero */
  22 static unsigned int total_keys = 1;
  23
  24 /* #define DEBUG_DEPS 1 */
  25
  26 /* Traversals block transactions in the current implementation. */
  27 #define TRAVERSALS_TAKE_TRANSACTION_LOCK 1
  28
  29 struct pipe {
  30         int fd[2];
  31 };
  32 static struct pipe *pipes;
  33
  34 static void __attribute__((noreturn)) fail(const char *filename,
  35                                            unsigned int line,
  36                                            const char *fmt, ...)
  37 {
  38         va_list ap;
  39
  40         va_start(ap, fmt);
  41         fprintf(stderr, "%s:%u: FAIL: ", filename, line);
  42         vfprintf(stderr, fmt, ap);
  43         fprintf(stderr, "\n");
  44         va_end(ap);
  45         exit(1);
  46 }
  47
  48 /* Try or die. */
  49 #define try(expr, expect)                                               \
  50         do {                                                            \
  51                 int ret = (expr);                                       \
  52                 if (ret != (expect))                                    \
  53                         fail(filename[file], i+1,                       \
  54                              STRINGIFY(expr) "= %i", ret);              \
  55         } while (0)
  56
  57 /* Try or imitate results. */
  58 #define unreliable(expr, expect, force, undo)                           \
  59         do {                                                            \
  60                 int ret = expr;                                         \
  61                 if (ret != expect) {                                    \
  62                         fprintf(stderr, "%s:%u: %s gave %i not %i",     \
  63                                 filename[file], i+1, STRINGIFY(expr),   \
  64                                 ret, expect);                           \
  65                         if (expect == 0)                                \
  66                                 force;                                  \
  67                         else                                            \
  68                                 undo;                                   \
  69                 }                                                       \
  70         } while (0)
  71
  72 static bool key_eq(TDB_DATA a, TDB_DATA b)
  73 {
  74         if (a.dsize != b.dsize)
  75                 return false;
  76         return memcmp(a.dptr, b.dptr, a.dsize) == 0;
  77 }
  78
  79 /* This is based on the hash algorithm from gdbm */
  80 static unsigned int hash_key(TDB_DATA *key)
  81 {
  82         uint32_t value; /* Used to compute the hash value.  */
  83         uint32_t   i;   /* Used to cycle through random values. */
  84
  85         /* Set the initial value from the key size. */
  86         for (value = 0x238F13AF ^ key->dsize, i=0; i < key->dsize; i++)
  87                 value = (value + (key->dptr[i] << (i*5 % 24)));
  88
  89         return (1103515243 * value + 12345);
  90 }
  91
  92 enum op_type {
  93         OP_TDB_LOCKALL,
  94         OP_TDB_LOCKALL_MARK,
  95         OP_TDB_LOCKALL_UNMARK,
  96         OP_TDB_LOCKALL_NONBLOCK,
  97         OP_TDB_UNLOCKALL,
  98         OP_TDB_LOCKALL_READ,
  99         OP_TDB_LOCKALL_READ_NONBLOCK,
 100         OP_TDB_UNLOCKALL_READ,
 101         OP_TDB_CHAINLOCK,
 102         OP_TDB_CHAINLOCK_NONBLOCK,
 103         OP_TDB_CHAINLOCK_MARK,
 104         OP_TDB_CHAINLOCK_UNMARK,
 105         OP_TDB_CHAINUNLOCK,
 106         OP_TDB_CHAINLOCK_READ,
 107         OP_TDB_CHAINUNLOCK_READ,
 108         OP_TDB_PARSE_RECORD,
 109         OP_TDB_EXISTS,
 110         OP_TDB_STORE,
 111         OP_TDB_APPEND,
 112         OP_TDB_GET_SEQNUM,
 113         OP_TDB_WIPE_ALL,
 114         OP_TDB_TRANSACTION_START,
 115         OP_TDB_TRANSACTION_CANCEL,
 116         OP_TDB_TRANSACTION_COMMIT,
 117         OP_TDB_TRAVERSE_READ_START,
 118         OP_TDB_TRAVERSE_START,
 119         OP_TDB_TRAVERSE_END,
 120         OP_TDB_TRAVERSE,
 121         OP_TDB_FIRSTKEY,
 122         OP_TDB_NEXTKEY,
 123         OP_TDB_FETCH,
 124         OP_TDB_DELETE,
 125 };
 126
 127 struct op {
 128         unsigned int serial;
 129         enum op_type op;
 130         TDB_DATA key;
 131         TDB_DATA data;
 132         int ret;
 133
 134         /* Who is waiting for us? */
 135         struct list_head post;
 136         /* What are we waiting for? */
 137         struct list_head pre;
 138
 139         /* If I'm part of a group (traverse/transaction) where is
 140          * start?  (Otherwise, 0) */
 141         unsigned int group_start;
 142
 143         union {
 144                 int flag; /* open and store */
 145                 struct traverse *trav; /* traverse start */
 146                 TDB_DATA pre_append; /* append */
 147                 unsigned int transaction_end; /* transaction start */
 148         };
 149 };
 150
 151 static unsigned char hex_char(const char *filename, unsigned int line, char c)
 152 {
 153         c = toupper(c);
 154         if (c >= 'A' && c <= 'F')
 155                 return c - 'A' + 10;
 156         if (c >= '0' && c <= '9')
 157                 return c - '0';
 158         fail(filename, line, "invalid hex character '%c'", c);
 159 }
 160
 161 /* TDB data is <size>:<%02x>* */
 162 static TDB_DATA make_tdb_data(const void *ctx,
 163                               const char *filename, unsigned int line,
 164                               const char *word)
 165 {
 166         TDB_DATA data;
 167         unsigned int i;
 168         const char *p;
 169
 170         if (streq(word, "NULL"))
 171                 return tdb_null;
 172
 173         data.dsize = atoi(word);
 174         data.dptr = talloc_array(ctx, unsigned char, data.dsize);
 175         p = strchr(word, ':');
 176         if (!p)
 177                 fail(filename, line, "invalid tdb data '%s'", word);
 178         p++;
 179         for (i = 0; i < data.dsize; i++)
 180                 data.dptr[i] = hex_char(filename, line, p[i*2])*16
 181                         + hex_char(filename, line, p[i*2+1]);
 182
 183         return data;
 184 }
 185
 186 static void add_op(const char *filename, struct op **op, unsigned int i,
 187                    unsigned int serial, enum op_type type)
 188 {
 189         struct op *new;
 190         *op = talloc_realloc(NULL, *op, struct op, i+1);
 191         new = (*op) + i;
 192         new->op = type;
 193         new->serial = serial;
 194         new->ret = 0;
 195         new->group_start = 0;
 196 }
 197
 198 static void op_add_nothing(const char *filename,
 199                            struct op op[], unsigned int op_num, char *words[])
 200 {
 201         if (words[2])
 202                 fail(filename, op_num+1, "Expected no arguments");
 203         op[op_num].key = tdb_null;
 204 }
 205
 206 static void op_add_key(const char *filename,
 207                        struct op op[], unsigned int op_num, char *words[])
 208 {
 209         if (words[2] == NULL || words[3])
 210                 fail(filename, op_num+1, "Expected just a key");
 211
 212         op[op_num].key = make_tdb_data(op, filename, op_num+1, words[2]);
 213         if (op[op_num].op != OP_TDB_TRAVERSE)
 214                 total_keys++;
 215 }
 216
 217 static void op_add_key_ret(const char *filename,
 218                            struct op op[], unsigned int op_num, char *words[])
 219 {
 220         if (!words[2] || !words[3] || !words[4] || words[5]
 221             || !streq(words[3], "="))
 222                 fail(filename, op_num+1, "Expected <key> = <ret>");
 223         op[op_num].ret = atoi(words[4]);
 224         op[op_num].key = make_tdb_data(op, filename, op_num+1, words[2]);
 225         /* May only be a unique key if it fails */
 226         if (op[op_num].ret != 0)
 227                 total_keys++;
 228 }
 229
 230 static void op_add_key_data(const char *filename,
 231                             struct op op[], unsigned int op_num, char *words[])
 232 {
 233         if (!words[2] || !words[3] || !words[4] || words[5]
 234             || !streq(words[3], "="))
 235                 fail(filename, op_num+1, "Expected <key> = <data>");
 236         op[op_num].key = make_tdb_data(op, filename, op_num+1, words[2]);
 237         op[op_num].data = make_tdb_data(op, filename, op_num+1, words[4]);
 238         /* May only be a unique key if it fails */
 239         if (!op[op_num].data.dptr)
 240                 total_keys++;
 241 }
 242
 243 /* <serial> tdb_store <rec> <rec> <flag> = <ret> */
 244 static void op_add_store(const char *filename,
 245                          struct op op[], unsigned int op_num, char *words[])
 246 {
 247         if (!words[2] || !words[3] || !words[4] || !words[5] || !words[6]
 248             || words[7] || !streq(words[5], "="))
 249                 fail(filename, op_num+1, "Expect <key> <data> <flag> = <ret>");
 250
 251         op[op_num].flag = strtoul(words[4], NULL, 0);
 252         op[op_num].ret = atoi(words[6]);
 253         op[op_num].key = make_tdb_data(op, filename, op_num+1, words[2]);
 254         op[op_num].data = make_tdb_data(op, filename, op_num+1, words[3]);
 255         total_keys++;
 256 }
 257
 258 /* <serial> tdb_append <rec> <rec> = <rec> */
 259 static void op_add_append(const char *filename,
 260                           struct op op[], unsigned int op_num, char *words[])
 261 {
 262         TDB_DATA post_append;
 263
 264         if (!words[2] || !words[3] || !words[4] || !words[5] || words[6]
 265             || !streq(words[4], "="))
 266                 fail(filename, op_num+1, "Expect <key> <data> = <rec>");
 267
 268         op[op_num].key = make_tdb_data(op, filename, op_num+1, words[2]);
 269         op[op_num].data = make_tdb_data(op, filename, op_num+1, words[3]);
 270
 271         post_append = make_tdb_data(op, filename, op_num+1, words[5]);
 272
 273         /* By subtraction, figure out what previous data was. */
 274         op[op_num].pre_append.dptr = post_append.dptr;
 275         op[op_num].pre_append.dsize = post_append.dsize - op[op_num].data.dsize;
 276         total_keys++;
 277 }
 278
 279 /* <serial> tdb_get_seqnum = <ret> */
 280 static void op_add_seqnum(const char *filename,
 281                           struct op op[], unsigned int op_num, char *words[])
 282 {
 283         if (!words[2] || !words[3] || words[4] || !streq(words[2], "="))
 284                 fail(filename, op_num+1, "Expect = <ret>");
 285
 286         op[op_num].key = tdb_null;
 287         op[op_num].ret = atoi(words[3]);
 288 }
 289
 290 static void op_add_traverse(const char *filename,
 291                             struct op op[], unsigned int op_num, char *words[])
 292 {
 293         if (words[2])
 294                 fail(filename, op_num+1, "Expect no arguments");
 295
 296         op[op_num].key = tdb_null;
 297         op[op_num].trav = NULL;
 298 }
 299
 300 static void op_add_transaction(const char *filename, struct op op[],
 301                                unsigned int op_num, char *words[])
 302 {
 303         if (words[2])
 304                 fail(filename, op_num+1, "Expect no arguments");
 305
 306         op[op_num].key = tdb_null;
 307         op[op_num].transaction_end = 0;
 308 }
 309
 310 static void op_analyze_transaction(const char *filename,
 311                                    struct op op[], unsigned int op_num,
 312                                    char *words[])
 313 {
 314         int i, start;
 315
 316         op[op_num].key = tdb_null;
 317
 318         if (words[2])
 319                 fail(filename, op_num+1, "Expect no arguments");
 320
 321         for (i = op_num-1; i >= 0; i--) {
 322                 if (op[i].op == OP_TDB_TRANSACTION_START &&
 323                     !op[i].transaction_end)
 324                         break;
 325         }
 326
 327         if (i < 0)
 328                 fail(filename, op_num+1, "no transaction start found");
 329
 330         start = i;
 331         op[start].transaction_end = op_num;
 332
 333         /* This rolls in nested transactions.  I think that's right. */
 334         for (i++; i <= op_num; i++)
 335                 op[i].group_start = start;
 336 }
 337
 338 struct traverse_hash {
 339         TDB_DATA key;
 340         unsigned int index;
 341 };
 342
 343 /* A traverse is a hash of keys, each one associated with ops. */
 344 struct traverse {
 345         /* How many traversal callouts should I do? */
 346         unsigned int num;
 347
 348         /* Where is traversal end op? */
 349         unsigned int end;
 350
 351         /* For trivial traversals. */
 352         struct traverse_hash *hash;
 353 };
 354
 355 /* A trivial traversal is one which doesn't terminate early and only
 356  * plays with its own record.  We can reliably replay these even if
 357  * traverse order changes. */
 358 static bool is_trivial_traverse(struct op op[], unsigned int end)
 359 {
 360 #if 0
 361         unsigned int i;
 362         TDB_DATA cur = tdb_null;
 363
 364         if (op[end].ret != 0)
 365                 return false;
 366
 367         for (i = 0; i < end; i++) {
 368                 if (!op[i].key.dptr)
 369                         continue;
 370                 if (op[i].op == OP_TDB_TRAVERSE)
 371                         cur = op[i].key;
 372                 if (!key_eq(cur, op[i].key))
 373                         return false;
 374         }
 375         return true;
 376 #endif
 377         /* With multiple things happening at once, no traverse is trivial. */
 378         return false;
 379 }
 380
 381 static void op_analyze_traverse(const char *filename,
 382                                 struct op op[], unsigned int op_num,
 383                                 char *words[])
 384 {
 385         int i, start;
 386         struct traverse *trav = talloc(op, struct traverse);
 387
 388         op[op_num].key = tdb_null;
 389
 390         /* = %u means traverse function terminated. */
 391         if (words[2]) {
 392                 if (!streq(words[2], "=") || !words[3] || words[4])
 393                         fail(filename, op_num+1, "expect = <num>");
 394                 op[op_num].ret = atoi(words[3]);
 395         } else
 396                 op[op_num].ret = 0;
 397
 398         trav->num = 0;
 399         trav->end = op_num;
 400         for (i = op_num-1; i >= 0; i--) {
 401                 if (op[i].op == OP_TDB_TRAVERSE)
 402                         trav->num++;
 403                 if (op[i].op != OP_TDB_TRAVERSE_READ_START
 404                     && op[i].op != OP_TDB_TRAVERSE_START)
 405                         continue;
 406                 if (op[i].trav)
 407                         continue;
 408                 break;
 409         }
 410
 411         if (i < 0)
 412                 fail(filename, op_num+1, "no traversal start found");
 413
 414         start = i;
 415         op[start].trav = trav;
 416
 417         for (i = start; i <= op_num; i++)
 418                 op[i].group_start = start;
 419
 420         if (is_trivial_traverse(op+i, op_num-i)) {
 421                 /* Fill in a plentiful hash table. */
 422                 op[start].trav->hash = talloc_zero_array(op[i].trav,
 423                                                          struct traverse_hash,
 424                                                          trav->num * 2);
 425                 for (i = start; i < op_num; i++) {
 426                         unsigned int h;
 427                         if (op[i].op != OP_TDB_TRAVERSE)
 428                                 continue;
 429                         h = hash_key(&op[i].key) % (trav->num * 2);
 430                         while (trav->hash[h].index)
 431                                 h = (h + 1) % (trav->num * 2);
 432                         trav->hash[h].index = i+1;
 433                         trav->hash[h].key = op[i].key;
 434                 }
 435         } else
 436                 trav->hash = NULL;
 437 }
 438
 439 /* Keep -Wmissing-declarations happy: */
 440 const struct op_table *
 441 find_keyword (register const char *str, register unsigned int len);
 442
 443 #include "keywords.c"
 444
 445 struct depend {
 446         /* We can have more than one */
 447         struct list_node list;
 448         unsigned int file;
 449         unsigned int op;
 450 };
 451
 452 struct depend_xmit {
 453         unsigned int dst_op;
 454         unsigned int src_file, src_op;
 455 };
 456
 457 static void remove_matching_dep(struct list_head *deps,
 458                                 unsigned int file, unsigned int op)
 459 {
 460         struct depend *dep;
 461
 462         list_for_each(deps, dep, list) {
 463                 if (dep->file == file && dep->op == op) {
 464                         list_del(&dep->list);
 465                         return;
 466                 }
 467         }
 468         errx(1, "Failed to find depend on file %u line %u\n", file, op+1);
 469 }
 470
 471 static void check_deps(const char *filename, struct op op[], unsigned int num)
 472 {
 473 #ifdef DEBUG_DEPS
 474         unsigned int i;
 475
 476         for (i = 1; i < num; i++)
 477                 if (!list_empty(&op[i].pre))
 478                         fail(filename, i+1, "Still has dependencies");
 479 #endif
 480 }
 481
 482 static void dump_pre(char *filename[], unsigned int file,
 483                      struct op op[], unsigned int i)
 484 {
 485         struct depend *dep;
 486
 487         printf("%s:%u still waiting for:\n", filename[file], i+1);
 488         list_for_each(&op[i].pre, dep, list)
 489                 printf("    %s:%u\n", filename[dep->file], dep->op+1);
 490         check_deps(filename[file], op, i);
 491 }
 492
 493 static void do_pre(char *filename[], unsigned int file, int pre_fd,
 494                    struct op op[], unsigned int i)
 495 {
 496         while (!list_empty(&op[i].pre)) {
 497                 struct depend_xmit dep;
 498
 499 #if DEBUG_DEPS
 500                 printf("%s:%u:waiting for pre\n", filename[file], i+1);
 501                 fflush(stdout);
 502 #endif
 503                 alarm(10);
 504                 while (read(pre_fd, &dep, sizeof(dep)) != sizeof(dep)) {
 505                         if (errno == EINTR) {
 506                                 dump_pre(filename, file, op, i);
 507                                 exit(1);
 508                         } else
 509                                 errx(1, "Reading from pipe");
 510                 }
 511                 alarm(0);
 512
 513 #if DEBUG_DEPS
 514                 printf("%s:%u:got pre %u from %s:%u\n", filename[file], i+1,
 515                        dep.dst_op+1, filename[dep.src_file], dep.src_op+1);
 516                 fflush(stdout);
 517 #endif
 518                 /* This could be any op, not just this one. */
 519                 remove_matching_dep(&op[dep.dst_op].pre,
 520                                     dep.src_file, dep.src_op);
 521         }
 522 }
 523
 524 static void do_post(char *filename[], unsigned int file,
 525                     const struct op op[], unsigned int i)
 526 {
 527         struct depend *dep;
 528
 529         list_for_each(&op[i].post, dep, list) {
 530                 struct depend_xmit dx;
 531
 532                 dx.src_file = file;
 533                 dx.src_op = i;
 534                 dx.dst_op = dep->op;
 535 #if DEBUG_DEPS
 536                 printf("%s:%u:sending to file %s:%u\n", filename[file], i+1,
 537                        filename[dep->file], dep->op+1);
 538 #endif
 539                 if (write(pipes[dep->file].fd[1], &dx, sizeof(dx))
 540                     != sizeof(dx))
 541                         err(1, "%s:%u failed to tell file %s",
 542                             filename[file], i+1, filename[dep->file]);
 543         }
 544 }
 545
 546 static int get_len(TDB_DATA key, TDB_DATA data, void *private_data)
 547 {
 548         return data.dsize;
 549 }
 550
 551 static unsigned run_ops(struct tdb_context *tdb,
 552                         int pre_fd,
 553                         char *filename[],
 554                         unsigned int file,
 555                         struct op op[],
 556                         unsigned int start, unsigned int stop);
 557
 558 struct traverse_info {
 559         struct op *op;
 560         char **filename;
 561         unsigned file;
 562         int pre_fd;
 563         unsigned int start;
 564         unsigned int i;
 565 };
 566
 567 /* Trivial case: do whatever they did for this key. */
 568 static int trivial_traverse(struct tdb_context *tdb,
 569                             TDB_DATA key, TDB_DATA data,
 570                             void *_tinfo)
 571 {
 572         struct traverse_info *tinfo = _tinfo;
 573         struct traverse *trav = tinfo->op[tinfo->start].trav;
 574         unsigned int h = hash_key(&key) % (trav->num * 2);
 575
 576         while (trav->hash[h].index) {
 577                 if (key_eq(trav->hash[h].key, key)) {
 578                         run_ops(tdb, tinfo->pre_fd, tinfo->filename,
 579                                 tinfo->file, tinfo->op, trav->hash[h].index,
 580                                 trav->end);
 581                         tinfo->i++;
 582                         return 0;
 583                 }
 584                 h = (h + 1) % (trav->num * 2);
 585         }
 586         fail(tinfo->filename[tinfo->file], tinfo->start + 1,
 587              "unexpected traverse key");
 588 }
 589
 590 /* More complex.  Just do whatever's they did at the n'th entry. */
 591 static int nontrivial_traverse(struct tdb_context *tdb,
 592                                TDB_DATA key, TDB_DATA data,
 593                                void *_tinfo)
 594 {
 595         struct traverse_info *tinfo = _tinfo;
 596         struct traverse *trav = tinfo->op[tinfo->start].trav;
 597
 598         if (tinfo->i == trav->end) {
 599                 /* This can happen if traverse expects to be empty. */
 600                 if (tinfo->start + 1 == trav->end)
 601                         return 1;
 602                 fail(tinfo->filename[tinfo->file], tinfo->start + 1,
 603                      "traverse did not terminate");
 604         }
 605
 606         if (tinfo->op[tinfo->i].op != OP_TDB_TRAVERSE)
 607                 fail(tinfo->filename[tinfo->file], tinfo->start + 1,
 608                      "%s:%u:traverse terminated early");
 609
 610         /* Run any normal ops. */
 611         tinfo->i = run_ops(tdb, tinfo->pre_fd, tinfo->filename, tinfo->file,
 612                            tinfo->op, tinfo->i+1, trav->end);
 613
 614         if (tinfo->i == trav->end)
 615                 return 1;
 616
 617         return 0;
 618 }
 619
 620 static unsigned op_traverse(struct tdb_context *tdb,
 621                             int pre_fd,
 622                             char *filename[],
 623                             unsigned int file,
 624                             int (*traversefn)(struct tdb_context *,
 625                                               tdb_traverse_func, void *),
 626                             struct op op[],
 627                             unsigned int start)
 628 {
 629         struct traverse *trav = op[start].trav;
 630         struct traverse_info tinfo = { op, filename, file, pre_fd,
 631                                        start, start+1 };
 632
 633         /* Trivial case. */
 634         if (trav->hash) {
 635                 int ret = traversefn(tdb, trivial_traverse, &tinfo);
 636                 if (ret != trav->num)
 637                         fail(filename[file], start+1,
 638                              "short traversal %i", ret);
 639                 return trav->end;
 640         }
 641
 642         traversefn(tdb, nontrivial_traverse, &tinfo);
 643
 644         /* Traversing in wrong order can have strange effects: eg. if
 645          * original traverse went A (delete A), B, we might do B
 646          * (delete A).  So if we have ops left over, we do it now. */
 647         while (tinfo.i != trav->end) {
 648                 if (op[tinfo.i].op == OP_TDB_TRAVERSE)
 649                         tinfo.i++;
 650                 else
 651                         tinfo.i = run_ops(tdb, pre_fd, filename, file, op,
 652                                           tinfo.i, trav->end);
 653         }
 654
 655         return trav->end;
 656 }
 657
 658 static void break_out(int sig)
 659 {
 660 }
 661
 662 static __attribute__((noinline))
 663 unsigned run_ops(struct tdb_context *tdb,
 664                  int pre_fd,
 665                  char *filename[],
 666                  unsigned int file,
 667                  struct op op[], unsigned int start, unsigned int stop)
 668 {
 669         unsigned int i;
 670         struct sigaction sa;
 671
 672         sa.sa_handler = break_out;
 673         sa.sa_flags = 0;
 674
 675         sigaction(SIGALRM, &sa, NULL);
 676         for (i = start; i < stop; i++) {
 677                 do_pre(filename, file, pre_fd, op, i);
 678
 679                 switch (op[i].op) {
 680                 case OP_TDB_LOCKALL:
 681                         try(tdb_lockall(tdb), op[i].ret);
 682                         break;
 683                 case OP_TDB_LOCKALL_MARK:
 684                         try(tdb_lockall_mark(tdb), op[i].ret);
 685                         break;
 686                 case OP_TDB_LOCKALL_UNMARK:
 687                         try(tdb_lockall_unmark(tdb), op[i].ret);
 688                         break;
 689                 case OP_TDB_LOCKALL_NONBLOCK:
 690                         unreliable(tdb_lockall_nonblock(tdb), op[i].ret,
 691                                    tdb_lockall(tdb), tdb_unlockall(tdb));
 692                         break;
 693                 case OP_TDB_UNLOCKALL:
 694                         try(tdb_unlockall(tdb), op[i].ret);
 695                         break;
 696                 case OP_TDB_LOCKALL_READ:
 697                         try(tdb_lockall_read(tdb), op[i].ret);
 698                         break;
 699                 case OP_TDB_LOCKALL_READ_NONBLOCK:
 700                         unreliable(tdb_lockall_read_nonblock(tdb), op[i].ret,
 701                                    tdb_lockall_read(tdb),
 702                                    tdb_unlockall_read(tdb));
 703                         break;
 704                 case OP_TDB_UNLOCKALL_READ:
 705                         try(tdb_unlockall_read(tdb), op[i].ret);
 706                         break;
 707                 case OP_TDB_CHAINLOCK:
 708                         try(tdb_chainlock(tdb, op[i].key), op[i].ret);
 709                         break;
 710                 case OP_TDB_CHAINLOCK_NONBLOCK:
 711                         unreliable(tdb_chainlock_nonblock(tdb, op[i].key),
 712                                    op[i].ret,
 713                                    tdb_chainlock(tdb, op[i].key),
 714                                    tdb_chainunlock(tdb, op[i].key));
 715                         break;
 716                 case OP_TDB_CHAINLOCK_MARK:
 717                         try(tdb_chainlock_mark(tdb, op[i].key), op[i].ret);
 718                         break;
 719                 case OP_TDB_CHAINLOCK_UNMARK:
 720                         try(tdb_chainlock_unmark(tdb, op[i].key), op[i].ret);
 721                         break;
 722                 case OP_TDB_CHAINUNLOCK:
 723                         try(tdb_chainunlock(tdb, op[i].key), op[i].ret);
 724                         break;
 725                 case OP_TDB_CHAINLOCK_READ:
 726                         try(tdb_chainlock_read(tdb, op[i].key), op[i].ret);
 727                         break;
 728                 case OP_TDB_CHAINUNLOCK_READ:
 729                         try(tdb_chainunlock_read(tdb, op[i].key), op[i].ret);
 730                         break;
 731                 case OP_TDB_PARSE_RECORD:
 732                         try(tdb_parse_record(tdb, op[i].key, get_len, NULL),
 733                             op[i].ret);
 734                         break;
 735                 case OP_TDB_EXISTS:
 736                         try(tdb_exists(tdb, op[i].key), op[i].ret);
 737                         break;
 738                 case OP_TDB_STORE:
 739                         try(tdb_store(tdb, op[i].key, op[i].data, op[i].flag),
 740                             op[i].ret < 0 ? op[i].ret : 0);
 741                         break;
 742                 case OP_TDB_APPEND:
 743                         try(tdb_append(tdb, op[i].key, op[i].data),
 744                             op[i].ret < 0 ? op[i].ret : 0);
 745                         break;
 746                 case OP_TDB_GET_SEQNUM:
 747                         try(tdb_get_seqnum(tdb), op[i].ret);
 748                         break;
 749                 case OP_TDB_WIPE_ALL:
 750                         try(tdb_wipe_all(tdb), op[i].ret);
 751                         break;
 752                 case OP_TDB_TRANSACTION_START:
 753                         try(tdb_transaction_start(tdb), op[i].ret);
 754                         break;
 755                 case OP_TDB_TRANSACTION_CANCEL:
 756                         try(tdb_transaction_cancel(tdb), op[i].ret);
 757                         break;
 758                 case OP_TDB_TRANSACTION_COMMIT:
 759                         try(tdb_transaction_commit(tdb), op[i].ret);
 760                         break;
 761                 case OP_TDB_TRAVERSE_READ_START:
 762                         i = op_traverse(tdb, pre_fd, filename, file,
 763                                         tdb_traverse_read, op, i);
 764                         break;
 765                 case OP_TDB_TRAVERSE_START:
 766                         i = op_traverse(tdb, pre_fd, filename, file,
 767                                         tdb_traverse, op, i);
 768                         break;
 769                 case OP_TDB_TRAVERSE:
 770                         /* Terminate: we're in a traverse, and we've
 771                          * done our ops. */
 772                         return i;
 773                 case OP_TDB_TRAVERSE_END:
 774                         fail(filename[file], i+1, "unepxected end traverse");
 775                 /* FIXME: These must be treated like traverse. */
 776                 case OP_TDB_FIRSTKEY:
 777                         if (!key_eq(tdb_firstkey(tdb), op[i].data))
 778                                 fail(filename[file], i+1, "bad firstkey");
 779                         break;
 780                 case OP_TDB_NEXTKEY:
 781                         if (!key_eq(tdb_nextkey(tdb, op[i].key), op[i].data))
 782                                 fail(filename[file], i+1, "bad nextkey");
 783                         break;
 784                 case OP_TDB_FETCH: {
 785                         TDB_DATA f = tdb_fetch(tdb, op[i].key);
 786                         if (!key_eq(f, op[i].data))
 787                                 fail(filename[file], i+1, "bad fetch %u",
 788                                      f.dsize);
 789                         break;
 790                 }
 791                 case OP_TDB_DELETE:
 792                         try(tdb_delete(tdb, op[i].key), op[i].ret);
 793                         break;
 794                 }
 795                 do_post(filename, file, op, i);
 796         }
 797         return i;
 798 }
 799
 800 static struct op *load_tracefile(const char *filename, unsigned int *num,
 801                                  unsigned int *hashsize,
 802                                  unsigned int *tdb_flags,
 803                                  unsigned int *open_flags)
 804 {
 805         unsigned int i;
 806         struct op *op = talloc_array(NULL, struct op, 1);
 807         char **words;
 808         char **lines;
 809         char *file;
 810
 811         file = grab_file(NULL, filename, NULL);
 812         if (!file)
 813                 err(1, "Reading %s", filename);
 814
 815         lines = strsplit(file, file, "\n", NULL);
 816         if (!lines[0])
 817                 errx(1, "%s is empty", filename);
 818
 819         words = strsplit(lines, lines[0], " ", NULL);
 820         if (!streq(words[1], "tdb_open"))
 821                 fail(filename, 1, "does not start with tdb_open");
 822
 823         *hashsize = atoi(words[2]);
 824         *tdb_flags = strtoul(words[3], NULL, 0);
 825         *open_flags = strtoul(words[4], NULL, 0);
 826
 827         for (i = 1; lines[i]; i++) {
 828                 const struct op_table *opt;
 829
 830                 words = strsplit(lines, lines[i], " ", NULL);
 831                 if (!words[0] || !words[1])
 832                         fail(filename, i+1, "Expected serial number and op");
 833
 834                 opt = find_keyword(words[1], strlen(words[1]));
 835                 if (!opt) {
 836                         if (streq(words[1], "tdb_close")) {
 837                                 if (lines[i+1])
 838                                         fail(filename, i+2,
 839                                              "lines after tdb_close");
 840                                 *num = i;
 841                                 talloc_free(lines);
 842                                 return op;
 843                         }
 844                         fail(filename, i+1, "Unknown operation '%s'", words[1]);
 845                 }
 846
 847                 add_op(filename, &op, i, atoi(words[0]), opt->type);
 848                 opt->enhance_op(filename, op, i, words);
 849         }
 850
 851         fprintf(stderr, "%s:%u:last operation is not tdb_close: incomplete?",
 852               filename, i);
 853         talloc_free(lines);
 854         *num = i - 1;
 855         return op;
 856 }
 857
 858 /* We remember all the keys we've ever seen, and who has them. */
 859 struct key_user {
 860         unsigned int file;
 861         unsigned int op_num;
 862 };
 863
 864 struct keyinfo {
 865         TDB_DATA key;
 866         unsigned int num_users;
 867         struct key_user *user;
 868 };
 869
 870 static const TDB_DATA must_not_exist;
 871 static const TDB_DATA must_exist;
 872 static const TDB_DATA not_exists_or_empty;
 873
 874 /* NULL means doesn't care if it exists or not, &must_exist means
 875  * it must exist but we don't care what, &must_not_exist means it must
 876  * not exist, otherwise the data it needs. */
 877 static const TDB_DATA *needs(const struct op *op)
 878 {
 879         switch (op->op) {
 880         /* FIXME: Pull forward deps, since we can deadlock */
 881         case OP_TDB_CHAINLOCK:
 882         case OP_TDB_CHAINLOCK_NONBLOCK:
 883         case OP_TDB_CHAINLOCK_MARK:
 884         case OP_TDB_CHAINLOCK_UNMARK:
 885         case OP_TDB_CHAINUNLOCK:
 886         case OP_TDB_CHAINLOCK_READ:
 887         case OP_TDB_CHAINUNLOCK_READ:
 888                 return NULL;
 889
 890         case OP_TDB_APPEND:
 891                 if (op->pre_append.dsize == 0)
 892                         return &not_exists_or_empty;
 893                 return &op->pre_append;
 894
 895         case OP_TDB_STORE:
 896                 if (op->flag == TDB_INSERT) {
 897                         if (op->ret < 0)
 898                                 return &must_exist;
 899                         else
 900                                 return &must_not_exist;
 901                 } else if (op->flag == TDB_MODIFY) {
 902                         if (op->ret < 0)
 903                                 return &must_not_exist;
 904                         else
 905                                 return &must_exist;
 906                 }
 907                 /* No flags?  Don't care */
 908                 return NULL;
 909
 910         case OP_TDB_EXISTS:
 911                 if (op->ret == 1)
 912                         return &must_exist;
 913                 else
 914                         return &must_not_exist;
 915
 916         case OP_TDB_PARSE_RECORD:
 917                 if (op->ret < 0)
 918                         return &must_not_exist;
 919                 return &must_exist;
 920
 921         /* FIXME: handle these. */
 922         case OP_TDB_WIPE_ALL:
 923         case OP_TDB_FIRSTKEY:
 924         case OP_TDB_NEXTKEY:
 925         case OP_TDB_GET_SEQNUM:
 926         case OP_TDB_TRAVERSE:
 927         case OP_TDB_TRANSACTION_COMMIT:
 928         case OP_TDB_TRANSACTION_CANCEL:
 929         case OP_TDB_TRANSACTION_START:
 930                 return NULL;
 931
 932         case OP_TDB_FETCH:
 933                 if (!op->data.dptr)
 934                         return &must_not_exist;
 935                 return &op->data;
 936
 937         case OP_TDB_DELETE:
 938                 if (op->ret < 0)
 939                         return &must_not_exist;
 940                 return &must_exist;
 941
 942         default:
 943                 errx(1, "Unexpected op %i", op->op);
 944         }
 945
 946 }
 947
 948 enum satisfaction {
 949         /* This op makes the other one possible. */
 950         SATISFIES,
 951         /* This op makes the other one impossible. */
 952         DISSATISFIES,
 953         /* This op makes no difference. */
 954         NO_CHANGE
 955 };
 956
 957 static enum satisfaction satisfies(const struct op *op, const TDB_DATA *need)
 958 {
 959         bool deletes, creates;
 960
 961         /* Failed ops don't change state of db. */
 962         if (op->ret < 0)
 963                 return NO_CHANGE;
 964
 965         deletes = (op->op == OP_TDB_DELETE || op->op == OP_TDB_WIPE_ALL);
 966         /* Append/store is creating the record if ret == 0 (1 means replaced) */
 967         if (op->op == OP_TDB_APPEND || op->op == OP_TDB_STORE)
 968                 creates = (op->ret == 0);
 969         else
 970                 creates = false;
 971
 972         if (need == &must_not_exist) {
 973                 if (deletes)
 974                         return SATISFIES;
 975                 if (creates)
 976                         return DISSATISFIES;
 977                 return NO_CHANGE;
 978         }
 979
 980         if (need == &must_exist) {
 981                 if (deletes)
 982                         return DISSATISFIES;
 983                 if (creates)
 984                         return SATISFIES;
 985                 return NO_CHANGE;
 986         }
 987
 988         if (need == &not_exists_or_empty) {
 989                 if (deletes)
 990                         return SATISFIES;
 991                 if (!creates)
 992                         return NO_CHANGE;
 993         }
 994
 995         /* OK, we need an exact match. */
 996         if (deletes)
 997                 return DISSATISFIES;
 998
 999         /* An append which results in the wrong data dissatisfies. */
1000         if (op->op == OP_TDB_APPEND) {
1001                 if (op->pre_append.dsize + op->data.dsize != need->dsize)
1002                         return DISSATISFIES;
1003                 if (memcmp(op->pre_append.dptr, need->dptr,
1004                            op->pre_append.dsize) != 0)
1005                         return DISSATISFIES;
1006                 if (memcmp(op->data.dptr, need->dptr + op->pre_append.dsize,
1007                            op->data.dsize) != 0)
1008                         return DISSATISFIES;
1009                 return SATISFIES;
1010         } else if (op->op == OP_TDB_STORE) {
1011                 if (key_eq(op->data, *need))
1012                         return SATISFIES;
1013                 return DISSATISFIES;
1014         }
1015         return NO_CHANGE;
1016 }
1017
1018 static struct keyinfo *hash_ops(struct op *op[], unsigned int num_ops[],
1019                                 unsigned int num)
1020 {
1021         unsigned int i, j, h;
1022         struct keyinfo *hash;
1023
1024         /* Gcc nexted function extension.  How cool is this? */
1025         int compare_user_serial(const void *_a, const void *_b)
1026         {
1027                 const struct key_user *a = _a, *b = _b;
1028                 int ret = op[a->file][a->op_num].serial
1029                         - op[b->file][b->op_num].serial;
1030
1031                 /* Serial is not completely reliable.  First, fetches don't
1032                  * inc serial, second we don't lock to get seq number.
1033                  * This smooths things a little for simple cases. */
1034                 if (ret == 0) {
1035                         const TDB_DATA *a_needs, *b_needs;
1036
1037                         b_needs = needs(&op[b->file][b->op_num]);
1038                         switch (satisfies(&op[a->file][a->op_num], b_needs)) {
1039                         case SATISFIES:
1040                                 /* A comes first: it satisfies B. */
1041                                 return -1;
1042                         case DISSATISFIES:
1043                                 /* A doesn't come first: it messes up B. */
1044                                 return 1;
1045                         default:
1046                                 break;
1047                         }
1048
1049                         a_needs = needs(&op[a->file][a->op_num]);
1050                         switch (satisfies(&op[b->file][b->op_num], a_needs)) {
1051                         case SATISFIES:
1052                                 /* B comes first: it satisfies A. */
1053                                 return 1;
1054                         case DISSATISFIES:
1055                                 /* B doesn't come first: it messes up A. */
1056                                 return -1;
1057                         default:
1058                                 break;
1059                         }
1060                 }
1061                 return ret;
1062         }
1063
1064         hash = talloc_zero_array(op[0], struct keyinfo, total_keys*2);
1065         for (i = 0; i < num; i++) {
1066                 for (j = 1; j < num_ops[i]; j++) {
1067                         /* We can't do this on allocation, due to realloc. */
1068                         list_head_init(&op[i][j].post);
1069                         list_head_init(&op[i][j].pre);
1070
1071                         if (!op[i][j].key.dptr)
1072                                 continue;
1073
1074                         /* We don't wait for traverse keys */
1075                         /* FIXME: We should, for trivial traversals. */
1076                         if (op[i][j].op == OP_TDB_TRAVERSE)
1077                                 continue;
1078
1079                         h = hash_key(&op[i][j].key) % (total_keys * 2);
1080                         while (!key_eq(hash[h].key, op[i][j].key)) {
1081                                 if (!hash[h].key.dptr) {
1082                                         hash[h].key = op[i][j].key;
1083                                         break;
1084                                 }
1085                                 h = (h + 1) % (total_keys * 2);
1086                         }
1087                         /* Might as well save some memory if we can. */
1088                         if (op[i][j].key.dptr != hash[h].key.dptr) {
1089                                 talloc_free(op[i][j].key.dptr);
1090                                 op[i][j].key.dptr = hash[h].key.dptr;
1091                         }
1092                         hash[h].user = talloc_realloc(hash, hash[h].user,
1093                                                      struct key_user,
1094                                                      hash[h].num_users+1);
1095                         hash[h].user[hash[h].num_users].op_num = j;
1096                         hash[h].user[hash[h].num_users].file = i;
1097                         hash[h].num_users++;
1098                 }
1099         }
1100
1101         /* Now sort into seqnum order. */
1102         for (h = 0; h < total_keys * 2; h++)
1103                 qsort(hash[h].user, hash[h].num_users, sizeof(hash[h].user[0]),
1104                       compare_user_serial);
1105
1106         return hash;
1107 }
1108
1109 static void add_dependency(void *ctx,
1110                            struct op *op[],
1111                            char *filename[],
1112                            unsigned int needs_file,
1113                            unsigned int needs_opnum,
1114                            unsigned int satisfies_file,
1115                            unsigned int satisfies_opnum)
1116 {
1117         struct depend *post, *pre;
1118         unsigned int needs_start, sat_start;
1119
1120         /* We don't depend on ourselves. */
1121         if (needs_file == satisfies_file)
1122                 return;
1123
1124 #if DEBUG_DEPS
1125         printf("%s:%u: depends on %s:%u\n",
1126                filename[needs_file], needs_opnum+1,
1127                filename[satisfies_file], satisfies_opnum+1);
1128 #endif
1129
1130         needs_start = op[needs_file][needs_opnum].group_start;
1131         sat_start = op[satisfies_file][satisfies_opnum].group_start;
1132
1133         /* If needs is in a transaction, we need it before start. */
1134         if (needs_start) {
1135                 switch (op[needs_file][needs_start].op) {
1136                 case OP_TDB_TRANSACTION_START:
1137                         needs_opnum = needs_start;
1138 #ifdef DEBUG_DEPS
1139                         printf("  -> Back to %u\n", needs_start+1);
1140                         fflush(stdout);
1141 #endif
1142                         break;
1143                 default:
1144                         break;
1145                 }
1146         }
1147
1148         /* If satisfies is in a transaction, we wait until after commit. */
1149         /* FIXME: If transaction is cancelled, don't need dependency. */
1150         if (sat_start) {
1151                 if (op[satisfies_file][sat_start].op
1152                     == OP_TDB_TRANSACTION_START) {
1153                         satisfies_opnum
1154                                 = op[satisfies_file][sat_start].transaction_end;
1155 #ifdef DEBUG_DEPS
1156                         printf("  -> Depends on %u\n", satisfies_opnum+1);
1157                         fflush(stdout);
1158 #endif
1159                 }
1160         }
1161
1162         post = talloc(ctx, struct depend);
1163         post->file = needs_file;
1164         post->op = needs_opnum;
1165         list_add(&op[satisfies_file][satisfies_opnum].post, &post->list);
1166
1167         pre = talloc(ctx, struct depend);
1168         pre->file = satisfies_file;
1169         pre->op = satisfies_opnum;
1170         list_add(&op[needs_file][needs_opnum].pre, &pre->list);
1171 }
1172
1173 #if TRAVERSALS_TAKE_TRANSACTION_LOCK
1174 struct traverse_dep {
1175         unsigned int file;
1176         unsigned int op_num;
1177         const struct op *op;
1178 };
1179
1180 /* Sort by which one runs first. */
1181 static int compare_traverse_dep(const void *_a, const void *_b)
1182 {
1183         const struct traverse_dep *a = _a, *b = _b;
1184         const struct traverse *trava = a->op->trav, *travb = b->op->trav;
1185
1186         if (a->op->serial != b->op->serial)
1187                 return a->op->serial - b->op->serial;
1188
1189         /* If they have same serial, it means one didn't make any changes.
1190          * Thus sort by end in that case. */
1191         return a->op[trava->end - a->op_num].serial
1192                 - b->op[travb->end - b->op_num].serial;
1193 }
1194
1195 /* Traversals can deadlock against each other.  Force order. */
1196 static void make_traverse_depends(char *filename[],
1197                                   struct op *op[], unsigned int num_ops[],
1198                                   unsigned int num)
1199 {
1200         unsigned int i, j, num_traversals = 0;
1201         struct traverse_dep *dep;
1202
1203         dep = talloc_array(NULL, struct traverse_dep, 1);
1204
1205         /* Count them. */
1206         for (i = 0; i < num; i++) {
1207                 for (j = 0; j < num_ops[i]; j++) {
1208                         if (op[i][j].op == OP_TDB_TRAVERSE_START
1209                             || op[i][j].op == OP_TDB_TRAVERSE_READ_START) {
1210                                 dep = talloc_realloc(NULL, dep,
1211                                                      struct traverse_dep,
1212                                                      num_traversals+1);
1213                                 dep[num_traversals].file = i;
1214                                 dep[num_traversals].op_num = j;
1215                                 dep[num_traversals].op = &op[i][j];
1216                                 num_traversals++;
1217                         }
1218                 }
1219         }
1220         qsort(dep, num_traversals, sizeof(dep[0]), compare_traverse_dep);
1221         for (i = 1; i < num_traversals; i++) {
1222                 /* i depends on end of traverse i-1. */
1223                 add_dependency(NULL, op, filename, dep[i].file, dep[i].op_num,
1224                                dep[i-1].file, dep[i-1].op->trav->end);
1225         }
1226         talloc_free(dep);
1227 }
1228 #endif /* TRAVERSALS_TAKE_TRANSACTION_LOCK */
1229
1230 static void derive_dependencies(char *filename[],
1231                                 struct op *op[], unsigned int num_ops[],
1232                                 unsigned int num)
1233 {
1234         struct keyinfo *hash;
1235         unsigned int i, j;
1236
1237         /* Create hash table for faster key lookup. */
1238         hash = hash_ops(op, num_ops, num);
1239
1240         /* We make the naive assumption that two ops on the same key
1241          * have to be ordered; it's overkill. */
1242         for (i = 0; i < total_keys * 2; i++) {
1243                 for (j = 1; j < hash[i].num_users; j++) {
1244                         add_dependency(hash, op, filename,
1245                                        hash[i].user[j].file,
1246                                        hash[i].user[j].op_num,
1247                                        hash[i].user[j-1].file,
1248                                        hash[i].user[j-1].op_num);
1249                 }
1250         }
1251
1252 #if TRAVERSALS_TAKE_TRANSACTION_LOCK
1253         make_traverse_depends(filename, op, num_ops, num);
1254 #endif
1255 }
1256
1257 int main(int argc, char *argv[])
1258 {
1259         struct timeval start, end;
1260         unsigned int i, num_ops[argc], hashsize[argc], tdb_flags[argc], open_flags[argc];
1261         struct op *op[argc];
1262         int fds[2];
1263         char c;
1264         bool ok = true;
1265
1266         if (argc < 3)
1267                 errx(1, "Usage: %s <tdbfile> <tracefile>...", argv[0]);
1268
1269         pipes = talloc_array(NULL, struct pipe, argc - 2);
1270         for (i = 0; i < argc - 2; i++) {
1271                 printf("Loading tracefile %s...", argv[2+i]);
1272                 fflush(stdout);
1273                 op[i] = load_tracefile(argv[2+i], &num_ops[i], &hashsize[i],
1274                                        &tdb_flags[i], &open_flags[i]);
1275                 if (pipe(pipes[i].fd) != 0)
1276                         err(1, "creating pipe");
1277                 printf("done\n");
1278         }
1279
1280         printf("Calculating inter-dependencies...");
1281         fflush(stdout);
1282         derive_dependencies(argv+2, op, num_ops, i);
1283         printf("done\n");
1284
1285         /* Don't fork for single arg case: simple debugging. */
1286         if (argc == 3) {
1287                 struct tdb_context *tdb;
1288                 tdb = tdb_open_ex(argv[1], hashsize[0], tdb_flags[0],
1289                                   open_flags[0], 0600,
1290                                   NULL, hash_key);
1291                 printf("Single threaded run...");
1292                 fflush(stdout);
1293
1294                 run_ops(tdb, pipes[0].fd[0], argv+2, 0, op[0], 1, num_ops[0]);
1295                 check_deps(argv[2], op[0], num_ops[0]);
1296
1297                 printf("done\n");
1298                 exit(0);
1299         }
1300
1301         if (pipe(fds) != 0)
1302                 err(1, "creating pipe");
1303
1304         for (i = 0; i < argc - 2; i++) {
1305                 struct tdb_context *tdb;
1306
1307                 switch (fork()) {
1308                 case -1:
1309                         err(1, "fork failed");
1310                 case 0:
1311                         close(fds[1]);
1312                         tdb = tdb_open_ex(argv[1], hashsize[i], tdb_flags[i],
1313                                           open_flags[i], 0600,
1314                                           NULL, hash_key);
1315                         if (!tdb)
1316                                 err(1, "Opening tdb %s", argv[1]);
1317
1318                         /* This catches parent exiting. */
1319                         if (read(fds[0], &c, 1) != 1)
1320                                 exit(1);
1321                         run_ops(tdb, pipes[i].fd[0], argv+2, i, op[i], 1,
1322                                 num_ops[i]);
1323                         check_deps(argv[2+i], op[i], num_ops[i]);
1324                         exit(0);
1325                 default:
1326                         break;
1327                 }
1328         }
1329
1330         /* Let everything settle. */
1331         sleep(1);
1332
1333         printf("Starting run...");
1334         fflush(stdout);
1335         gettimeofday(&start, NULL);
1336         /* Tell them all to go!  Any write of sufficient length will do. */
1337         if (write(fds[1], hashsize, i) != i)
1338                 err(1, "Writing to wakeup pipe");
1339
1340         for (i = 0; i < argc - 2; i++) {
1341                 int status;
1342                 wait(&status);
1343                 if (!WIFEXITED(status)) {
1344                         warnx("Child died with signal %i", WTERMSIG(status));
1345                         ok = false;
1346                 } else if (WEXITSTATUS(status) != 0)
1347                         /* Assume child spat out error. */
1348                         ok = false;
1349         }
1350         if (!ok)
1351                 exit(1);
1352
1353         gettimeofday(&end, NULL);
1354         printf("done\n");
1355
1356         end.tv_sec -= start.tv_sec;
1357         printf("Time replaying: %lu usec\n",
1358                end.tv_sec * 1000000UL + (end.tv_usec - start.tv_usec));
1359
1360         exit(0);
1361 }