integ-test.c

   1 #include <stdint.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include <stdio.h>
   5 #include <unistd.h>
   6 #include <signal.h>
   7 #include <sys/stat.h>
   8 #include <sys/types.h>
   9 #include <sys/socket.h>
  10 #include <sys/select.h>
  11 #include <netdb.h>
  12 #include <netinet/ip.h>
  13 #include <netinet/in.h>
  14 #include <arpa/inet.h>
  15 #include <fcntl.h>
  16 #include <sys/wait.h>
  17 #include <errno.h>
  18 #include "ct/ct.h"
  19 #include "dat.h"
  20
  21 static int srvpid, port, fd, size;
  22 static int64 timeout = 5000000000; // 5s
  23 static char dir[] = "/tmp/beanstalkd.test.XXXXXX";
  24
  25 static byte fallocpat[3];
  26
  27
  28 static int
  29 wrapfalloc(int fd, int size)
  30 {
  31     static int c = 0;
  32
  33     printf("\nwrapfalloc: fd=%d size=%d\n", fd, size);
  34     if (c >= sizeof(fallocpat) || !fallocpat[c++]) {
  35         return ENOSPC;
  36     }
  37     return rawfalloc(fd, size);
  38 }
  39
  40
  41 static void
  42 muststart(char *a0, char *a1, char *a2, char *a3, char *a4)
  43 {
  44     srvpid = fork();
  45     if (srvpid < 0) {
  46         twarn("fork");
  47         exit(1);
  48     }
  49
  50     if (srvpid > 0) {
  51         printf("%s %s %s %s %s\n", a0, a1, a2, a3, a4);
  52         printf("start server pid=%d\n", srvpid);
  53         usleep(100000); // .1s; time for the child to bind to its port
  54         return;
  55     }
  56
  57     /* now in child */
  58
  59     execlp(a0, a0, a1, a2, a3, a4, NULL);
  60 }
  61
  62
  63 static int
  64 mustdiallocal(int port)
  65 {
  66     int r, fd;
  67     struct sockaddr_in addr = {};
  68
  69     addr.sin_family = AF_INET;
  70     addr.sin_port = htons(port);
  71     r = inet_aton("127.0.0.1", &addr.sin_addr);
  72     if (!r) {
  73         errno = EINVAL;
  74         twarn("inet_aton");
  75         exit(1);
  76     }
  77
  78     fd = socket(PF_INET, SOCK_STREAM, 0);
  79     if (fd == -1) {
  80         twarn("socket");
  81         exit(1);
  82     }
  83
  84     r = connect(fd, (struct sockaddr*)&addr, sizeof addr);
  85     if (r == -1) {
  86         twarn("connect");
  87         exit(1);
  88     }
  89
  90     return fd;
  91 }
  92
  93
  94 #define SERVER() (progname=__func__, mustforksrv())
  95
  96 static int
  97 mustforksrv()
  98 {
  99     int r, len, port, ok;
 100     struct sockaddr_in addr;
 101
 102     srv.sock.fd = make_server_socket("127.0.0.1", "0");
 103     if (srv.sock.fd == -1) {
 104         puts("mustforksrv failed");
 105         exit(1);
 106     }
 107
 108     len = sizeof(addr);
 109     r = getsockname(srv.sock.fd, (struct sockaddr*)&addr, (socklen_t*)&len);
 110     if (r == -1 || len > sizeof(addr)) {
 111         puts("mustforksrv failed");
 112         exit(1);
 113     }
 114
 115     port = ntohs(addr.sin_port);
 116     srvpid = fork();
 117     if (srvpid < 0) {
 118         twarn("fork");
 119         exit(1);
 120     }
 121
 122     if (srvpid > 0) {
 123         printf("start server port=%d pid=%d\n", port, srvpid);
 124         return port;
 125     }
 126
 127     /* now in child */
 128
 129     prot_init();
 130
 131     if (srv.wal.use) {
 132         struct job list = {};
 133         // We want to make sure that only one beanstalkd tries
 134         // to use the wal directory at a time. So acquire a lock
 135         // now and never release it.
 136         if (!waldirlock(&srv.wal)) {
 137             twarnx("failed to lock wal dir %s", srv.wal.dir);
 138             exit(10);
 139         }
 140
 141         list.prev = list.next = &list;
 142         walinit(&srv.wal, &list);
 143         ok = prot_replay(&srv, &list);
 144         if (!ok) {
 145             twarnx("failed to replay log");
 146             exit(11);
 147         }
 148     }
 149
 150     srvserve(&srv); /* does not return */
 151     exit(1); /* satisfy the compiler */
 152 }
 153
 154
 155 static char *
 156 readline(int fd)
 157 {
 158     int r, c = 0, p = 0, i = 0;
 159     static char buf[1024];
 160     fd_set rfd;
 161     struct timeval tv;
 162
 163     printf("<%d ", fd);
 164     fflush(stdout);
 165     for (;;) {
 166         FD_ZERO(&rfd);
 167         FD_SET(fd, &rfd);
 168         tv.tv_sec = timeout / 1000000000;
 169         tv.tv_usec = (timeout/1000) % 1000000;
 170         r = select(fd+1, &rfd, NULL, NULL, &tv);
 171         switch (r) {
 172         case 1:
 173             break;
 174         case 0:
 175             fputs("timeout", stderr);
 176             exit(8);
 177         case -1:
 178             perror("select");
 179             exit(1);
 180         default:
 181             fputs("unknown error", stderr);
 182             exit(3);
 183         }
 184
 185         r = read(fd, &c, 1);
 186         if (r == -1) {
 187             perror("write");
 188             exit(1);
 189         }
 190         if (i >= sizeof(buf)-1) {
 191             fputs("response too big", stderr);
 192             exit(4);
 193         }
 194         putc(c, stdout);
 195         fflush(stdout);
 196         buf[i++] = c;
 197         if (p == '\r' && c == '\n') {
 198             break;
 199         }
 200         p = c;
 201     }
 202     buf[i] = '\0';
 203     return buf;
 204 }
 205
 206
 207 static void
 208 ckresp(int fd, char *exp)
 209 {
 210     char *line;
 211
 212     line = readline(fd);
 213     assertf(strcmp(exp, line) == 0, "\"%s\" != \"%s\"", exp, line);
 214 }
 215
 216
 217 static void
 218 ckrespsub(int fd, char *sub)
 219 {
 220     char *line;
 221
 222     line = readline(fd);
 223     assertf(strstr(line, sub), "\"%s\" not in \"%s\"", sub, line);
 224 }
 225
 226
 227 static void
 228 writefull(int fd, char *s, int n)
 229 {
 230     int c;
 231     for (; n; n -= c) {
 232         c = write(fd, s, n);
 233         if (c == -1) {
 234             perror("write");
 235             exit(1);
 236         }
 237         s += c;
 238     }
 239 }
 240
 241
 242 static void
 243 mustsend(int fd, char *s)
 244 {
 245     writefull(fd, s, strlen(s));
 246     printf(">%d %s", fd, s);
 247     fflush(stdout);
 248 }
 249
 250
 251 static void
 252 killsrv(void)
 253 {
 254     if (srvpid > 1) {
 255         kill(srvpid, 9);
 256     }
 257 }
 258
 259
 260 static void
 261 sigabrt(int sig)
 262 {
 263     killsrv();
 264 }
 265
 266
 267 static void
 268 installabrt()
 269 {
 270     int r;
 271     struct sigaction sa = {};
 272
 273     sa.sa_handler = sigabrt;
 274     r = sigaction(SIGABRT, &sa, 0);
 275     if (r == -1) {
 276         twarn("sigaction SIGABRT");
 277         exit(1);
 278     }
 279 }
 280
 281
 282 static int
 283 filesize(char *path)
 284 {
 285     int r;
 286     struct stat s;
 287
 288     r = stat(path, &s);
 289     if (r == -1) {
 290         twarn("stat");
 291         exit(1);
 292     }
 293     return s.st_size;
 294 }
 295
 296
 297 static int
 298 exist(char *path)
 299 {
 300     int r;
 301     struct stat s;
 302
 303     r = stat(path, &s);
 304     return r != -1;
 305 }
 306
 307
 308 void
 309 cttestpause()
 310 {
 311     int64 s;
 312
 313     port = SERVER();
 314     fd = mustdiallocal(port);
 315     mustsend(fd, "put 0 0 0 1\r\n");
 316     mustsend(fd, "x\r\n");
 317     ckresp(fd, "INSERTED 1\r\n");
 318     s = nanoseconds();
 319     mustsend(fd, "pause-tube default 1\r\n");
 320     ckresp(fd, "PAUSED\r\n");
 321     mustsend(fd, "reserve\r\n");
 322     ckresp(fd, "RESERVED 1 1\r\n");
 323     ckresp(fd, "x\r\n");
 324     assert(nanoseconds() - s >= 1000000000); // 1s
 325
 326     killsrv();
 327 }
 328
 329
 330 void
 331 cttestunderscore()
 332 {
 333     port = SERVER();
 334     fd = mustdiallocal(port);
 335     mustsend(fd, "use x_y\r\n");
 336     ckresp(fd, "USING x_y\r\n");
 337
 338     killsrv();
 339 }
 340
 341
 342 void
 343 cttesttoobig()
 344 {
 345     job_data_size_limit = 10;
 346     port = SERVER();
 347     fd = mustdiallocal(port);
 348     mustsend(fd, "put 0 0 0 11\r\n");
 349     mustsend(fd, "delete 9999\r\n");
 350     mustsend(fd, "put 0 0 0 1\r\n");
 351     mustsend(fd, "x\r\n");
 352     ckresp(fd, "JOB_TOO_BIG\r\n");
 353     ckresp(fd, "INSERTED 1\r\n");
 354
 355     killsrv();
 356 }
 357
 358
 359 void
 360 cttestdeleteready()
 361 {
 362     port = SERVER();
 363     fd = mustdiallocal(port);
 364     mustsend(fd, "put 0 0 0 0\r\n");
 365     mustsend(fd, "\r\n");
 366     ckresp(fd, "INSERTED 1\r\n");
 367     mustsend(fd, "delete 1\r\n");
 368     ckresp(fd, "DELETED\r\n");
 369
 370     killsrv();
 371 }
 372
 373
 374 void
 375 cttestmultitube()
 376 {
 377     port = SERVER();
 378     fd = mustdiallocal(port);
 379     mustsend(fd, "use abc\r\n");
 380     ckresp(fd, "USING abc\r\n");
 381     mustsend(fd, "put 999999 0 0 0\r\n");
 382     mustsend(fd, "\r\n");
 383     ckresp(fd, "INSERTED 1\r\n");
 384     mustsend(fd, "use def\r\n");
 385     ckresp(fd, "USING def\r\n");
 386     mustsend(fd, "put 99 0 0 0\r\n");
 387     mustsend(fd, "\r\n");
 388     ckresp(fd, "INSERTED 2\r\n");
 389     mustsend(fd, "watch abc\r\n");
 390     ckresp(fd, "WATCHING 2\r\n");
 391     mustsend(fd, "watch def\r\n");
 392     ckresp(fd, "WATCHING 3\r\n");
 393     mustsend(fd, "reserve\r\n");
 394     ckresp(fd, "RESERVED 2 0\r\n");
 395
 396     killsrv();
 397 }
 398
 399
 400 void
 401 cttestnonegativedelay()
 402 {
 403     port = SERVER();
 404     fd = mustdiallocal(port);
 405     mustsend(fd, "put 512 -1 100 0\r\n");
 406     ckresp(fd, "BAD_FORMAT\r\n");
 407
 408     killsrv();
 409 }
 410
 411
 412 void
 413 cttestomittimeleft()
 414 {
 415     port = SERVER();
 416     fd = mustdiallocal(port);
 417     mustsend(fd, "put 0 0 5 1\r\n");
 418     mustsend(fd, "a\r\n");
 419     ckresp(fd, "INSERTED 1\r\n");
 420     mustsend(fd, "stats-job 1\r\n");
 421     ckrespsub(fd, "OK ");
 422     ckrespsub(fd, "\ntime-left: 0\n");
 423
 424     killsrv();
 425 }
 426
 427
 428 void
 429 cttestsmalldelay()
 430 {
 431     port = SERVER();
 432     fd = mustdiallocal(port);
 433     mustsend(fd, "put 0 1 1 0\r\n");
 434     mustsend(fd, "\r\n");
 435     ckresp(fd, "INSERTED 1\r\n");
 436
 437     killsrv();
 438 }
 439
 440
 441 void
 442 ctteststatstube()
 443 {
 444     port = SERVER();
 445     fd = mustdiallocal(port);
 446     mustsend(fd, "use tubea\r\n");
 447     ckresp(fd, "USING tubea\r\n");
 448     mustsend(fd, "put 0 0 0 1\r\n");
 449     mustsend(fd, "x\r\n");
 450     ckresp(fd, "INSERTED 1\r\n");
 451     mustsend(fd, "delete 1\r\n");
 452     ckresp(fd, "DELETED\r\n");
 453
 454     mustsend(fd, "stats-tube tubea\r\n");
 455     ckrespsub(fd, "OK ");
 456     ckrespsub(fd, "\nname: tubea\n");
 457     mustsend(fd, "stats-tube tubea\r\n");
 458     ckrespsub(fd, "OK ");
 459     ckrespsub(fd, "\ncurrent-jobs-urgent: 0\n");
 460     mustsend(fd, "stats-tube tubea\r\n");
 461     ckrespsub(fd, "OK ");
 462     ckrespsub(fd, "\ncurrent-jobs-ready: 0\n");
 463     mustsend(fd, "stats-tube tubea\r\n");
 464     ckrespsub(fd, "OK ");
 465     ckrespsub(fd, "\ncurrent-jobs-reserved: 0\n");
 466     mustsend(fd, "stats-tube tubea\r\n");
 467     ckrespsub(fd, "OK ");
 468     ckrespsub(fd, "\ncurrent-jobs-delayed: 0\n");
 469     mustsend(fd, "stats-tube tubea\r\n");
 470     ckrespsub(fd, "OK ");
 471     ckrespsub(fd, "\ncurrent-jobs-buried: 0\n");
 472     mustsend(fd, "stats-tube tubea\r\n");
 473     ckrespsub(fd, "OK ");
 474     ckrespsub(fd, "\ntotal-jobs: 1\n");
 475     mustsend(fd, "stats-tube tubea\r\n");
 476     ckrespsub(fd, "OK ");
 477     ckrespsub(fd, "\ncurrent-using: 1\n");
 478     mustsend(fd, "stats-tube tubea\r\n");
 479     ckrespsub(fd, "OK ");
 480     ckrespsub(fd, "\ncurrent-watching: 0\n");
 481     mustsend(fd, "stats-tube tubea\r\n");
 482     ckrespsub(fd, "OK ");
 483     ckrespsub(fd, "\ncurrent-waiting: 0\n");
 484     mustsend(fd, "stats-tube tubea\r\n");
 485     ckrespsub(fd, "OK ");
 486     ckrespsub(fd, "\ncmd-delete: 1\n");
 487     mustsend(fd, "stats-tube tubea\r\n");
 488     ckrespsub(fd, "OK ");
 489     ckrespsub(fd, "\ncmd-pause-tube: 0\n");
 490     mustsend(fd, "stats-tube tubea\r\n");
 491     ckrespsub(fd, "OK ");
 492     ckrespsub(fd, "\npause: 0\n");
 493     mustsend(fd, "stats-tube tubea\r\n");
 494     ckrespsub(fd, "OK ");
 495     ckrespsub(fd, "\npause-time-left: 0\n");
 496
 497     mustsend(fd, "stats-tube default\r\n");
 498     ckrespsub(fd, "OK ");
 499     ckrespsub(fd, "\nname: default\n");
 500     mustsend(fd, "stats-tube default\r\n");
 501     ckrespsub(fd, "OK ");
 502     ckrespsub(fd, "\ncurrent-jobs-urgent: 0\n");
 503     mustsend(fd, "stats-tube default\r\n");
 504     ckrespsub(fd, "OK ");
 505     ckrespsub(fd, "\ncurrent-jobs-ready: 0\n");
 506     mustsend(fd, "stats-tube default\r\n");
 507     ckrespsub(fd, "OK ");
 508     ckrespsub(fd, "\ncurrent-jobs-reserved: 0\n");
 509     mustsend(fd, "stats-tube default\r\n");
 510     ckrespsub(fd, "OK ");
 511     ckrespsub(fd, "\ncurrent-jobs-delayed: 0\n");
 512     mustsend(fd, "stats-tube default\r\n");
 513     ckrespsub(fd, "OK ");
 514     ckrespsub(fd, "\ncurrent-jobs-buried: 0\n");
 515     mustsend(fd, "stats-tube default\r\n");
 516     ckrespsub(fd, "OK ");
 517     ckrespsub(fd, "\ntotal-jobs: 0\n");
 518     mustsend(fd, "stats-tube default\r\n");
 519     ckrespsub(fd, "OK ");
 520     ckrespsub(fd, "\ncurrent-using: 0\n");
 521     mustsend(fd, "stats-tube default\r\n");
 522     ckrespsub(fd, "OK ");
 523     ckrespsub(fd, "\ncurrent-watching: 1\n");
 524     mustsend(fd, "stats-tube default\r\n");
 525     ckrespsub(fd, "OK ");
 526     ckrespsub(fd, "\ncurrent-waiting: 0\n");
 527     mustsend(fd, "stats-tube default\r\n");
 528     ckrespsub(fd, "OK ");
 529     ckrespsub(fd, "\ncmd-delete: 0\n");
 530     mustsend(fd, "stats-tube default\r\n");
 531     ckrespsub(fd, "OK ");
 532     ckrespsub(fd, "\ncmd-pause-tube: 0\n");
 533     mustsend(fd, "stats-tube default\r\n");
 534     ckrespsub(fd, "OK ");
 535     ckrespsub(fd, "\npause: 0\n");
 536     mustsend(fd, "stats-tube default\r\n");
 537     ckrespsub(fd, "OK ");
 538     ckrespsub(fd, "\npause-time-left: 0\n");
 539
 540     killsrv();
 541 }
 542
 543
 544 void
 545 cttestttrlarge()
 546 {
 547     port = SERVER();
 548     fd = mustdiallocal(port);
 549     mustsend(fd, "put 0 0 120 1\r\n");
 550     mustsend(fd, "a\r\n");
 551     ckresp(fd, "INSERTED 1\r\n");
 552     mustsend(fd, "put 0 0 4294 1\r\n");
 553     mustsend(fd, "a\r\n");
 554     ckresp(fd, "INSERTED 2\r\n");
 555     mustsend(fd, "put 0 0 4295 1\r\n");
 556     mustsend(fd, "a\r\n");
 557     ckresp(fd, "INSERTED 3\r\n");
 558     mustsend(fd, "put 0 0 4296 1\r\n");
 559     mustsend(fd, "a\r\n");
 560     ckresp(fd, "INSERTED 4\r\n");
 561     mustsend(fd, "put 0 0 4297 1\r\n");
 562     mustsend(fd, "a\r\n");
 563     ckresp(fd, "INSERTED 5\r\n");
 564     mustsend(fd, "put 0 0 5000 1\r\n");
 565     mustsend(fd, "a\r\n");
 566     ckresp(fd, "INSERTED 6\r\n");
 567     mustsend(fd, "put 0 0 21600 1\r\n");
 568     mustsend(fd, "a\r\n");
 569     ckresp(fd, "INSERTED 7\r\n");
 570     mustsend(fd, "stats-job 1\r\n");
 571     ckrespsub(fd, "OK ");
 572     ckrespsub(fd, "\nttr: 120\n");
 573     mustsend(fd, "stats-job 2\r\n");
 574     ckrespsub(fd, "OK ");
 575     ckrespsub(fd, "\nttr: 4294\n");
 576     mustsend(fd, "stats-job 3\r\n");
 577     ckrespsub(fd, "OK ");
 578     ckrespsub(fd, "\nttr: 4295\n");
 579     mustsend(fd, "stats-job 4\r\n");
 580     ckrespsub(fd, "OK ");
 581     ckrespsub(fd, "\nttr: 4296\n");
 582     mustsend(fd, "stats-job 5\r\n");
 583     ckrespsub(fd, "OK ");
 584     ckrespsub(fd, "\nttr: 4297\n");
 585     mustsend(fd, "stats-job 6\r\n");
 586     ckrespsub(fd, "OK ");
 587     ckrespsub(fd, "\nttr: 5000\n");
 588     mustsend(fd, "stats-job 7\r\n");
 589     ckrespsub(fd, "OK ");
 590     ckrespsub(fd, "\nttr: 21600\n");
 591
 592     killsrv();
 593 }
 594
 595
 596 void
 597 cttestttrsmall()
 598 {
 599     port = SERVER();
 600     fd = mustdiallocal(port);
 601     mustsend(fd, "put 0 0 0 1\r\n");
 602     mustsend(fd, "a\r\n");
 603     ckresp(fd, "INSERTED 1\r\n");
 604     mustsend(fd, "stats-job 1\r\n");
 605     ckrespsub(fd, "OK ");
 606     ckrespsub(fd, "\nttr: 1\n");
 607
 608     killsrv();
 609 }
 610
 611
 612 void
 613 cttestzerodelay()
 614 {
 615     port = SERVER();
 616     fd = mustdiallocal(port);
 617     mustsend(fd, "put 0 0 1 0\r\n");
 618     mustsend(fd, "\r\n");
 619     ckresp(fd, "INSERTED 1\r\n");
 620
 621     killsrv();
 622 }
 623
 624
 625 void
 626 cttestreservewithtimeout2conn()
 627 {
 628     int fd0, fd1;
 629
 630     job_data_size_limit = 10;
 631
 632     port = SERVER();
 633     fd0 = mustdiallocal(port);
 634     fd1 = mustdiallocal(port);
 635     mustsend(fd0, "watch foo\r\n");
 636     ckresp(fd0, "WATCHING 2\r\n");
 637     mustsend(fd0, "reserve-with-timeout 1\r\n");
 638     mustsend(fd1, "watch foo\r\n");
 639     ckresp(fd1, "WATCHING 2\r\n");
 640     timeout = 1100000000; // 1.1s
 641     ckresp(fd0, "TIMED_OUT\r\n");
 642
 643     killsrv();
 644 }
 645
 646
 647 void
 648 cttestunpausetube()
 649 {
 650     int fd0, fd1;
 651
 652     port = SERVER();
 653     fd0 = mustdiallocal(port);
 654     fd1 = mustdiallocal(port);
 655
 656     mustsend(fd0, "put 0 0 0 0\r\n");
 657     mustsend(fd0, "\r\n");
 658     ckresp(fd0, "INSERTED 1\r\n");
 659
 660     mustsend(fd0, "pause-tube default 86400\r\n");
 661     ckresp(fd0, "PAUSED\r\n");
 662
 663     mustsend(fd1, "reserve\r\n");
 664
 665     mustsend(fd0, "pause-tube default 0\r\n");
 666     ckresp(fd0, "PAUSED\r\n");
 667
 668     // ckresp will time out if this takes too long, so the
 669     // test will not pass.
 670     ckresp(fd1, "RESERVED 1 0\r\n");
 671     ckresp(fd1, "\r\n");
 672
 673     killsrv();
 674 }
 675
 676
 677 void
 678 cttestbinlogemptyexit()
 679 {
 680     mkdtemp(dir);
 681     srv.wal.dir = dir;
 682     srv.wal.use = 1;
 683     job_data_size_limit = 10;
 684
 685     port = SERVER();
 686
 687     killsrv();
 688
 689     port = SERVER();
 690     fd = mustdiallocal(port);
 691     mustsend(fd, "put 0 0 0 0\r\n");
 692     mustsend(fd, "\r\n");
 693     ckresp(fd, "INSERTED 1\r\n");
 694
 695     killsrv();
 696     execlp("rm", "rm", "-rf", dir, NULL);
 697 }
 698
 699
 700 void
 701 cttestbinlogbury()
 702 {
 703     mkdtemp(dir);
 704     srv.wal.dir = dir;
 705     srv.wal.use = 1;
 706     job_data_size_limit = 10;
 707
 708     port = SERVER();
 709     fd = mustdiallocal(port);
 710     mustsend(fd, "put 0 0 100 0\r\n");
 711     mustsend(fd, "\r\n");
 712     ckresp(fd, "INSERTED 1\r\n");
 713     mustsend(fd, "reserve\r\n");
 714     ckresp(fd, "RESERVED 1 0\r\n");
 715     ckresp(fd, "\r\n");
 716     mustsend(fd, "bury 1 0\r\n");
 717     ckresp(fd, "BURIED\r\n");
 718
 719     killsrv();
 720     execlp("rm", "rm", "-rf", dir, NULL);
 721 }
 722
 723
 724 void
 725 cttestbinlogbasic()
 726 {
 727     mkdtemp(dir);
 728     srv.wal.dir = dir;
 729     srv.wal.use = 1;
 730     job_data_size_limit = 10;
 731
 732     port = SERVER();
 733     fd = mustdiallocal(port);
 734     mustsend(fd, "put 0 0 100 0\r\n");
 735     mustsend(fd, "\r\n");
 736     ckresp(fd, "INSERTED 1\r\n");
 737
 738     killsrv();
 739     waitpid(srvpid, NULL, 0);
 740
 741     port = SERVER();
 742     fd = mustdiallocal(port);
 743     mustsend(fd, "delete 1\r\n");
 744     ckresp(fd, "DELETED\r\n");
 745
 746     killsrv();
 747     execlp("rm", "rm", "-rf", dir, NULL);
 748 }
 749
 750
 751 void
 752 cttestbinlogsizelimit()
 753 {
 754     int i = 0;
 755     char *b2;
 756     int gotsize;
 757
 758     size = 1024;
 759     mkdtemp(dir);
 760     srv.wal.dir = dir;
 761     srv.wal.use = 1;
 762     srv.wal.filesize = size;
 763     srv.wal.syncrate = 0;
 764     srv.wal.wantsync = 1;
 765
 766     port = SERVER();
 767     fd = mustdiallocal(port);
 768     b2 = fmtalloc("%s/binlog.2", dir);
 769     while (!exist(b2)) {
 770         mustsend(fd, "put 0 0 100 50\r\n");
 771         mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 772         ckresp(fd, fmtalloc("INSERTED %d\r\n", ++i));
 773     }
 774
 775     gotsize = filesize(fmtalloc("%s/binlog.1", dir));
 776     assertf(gotsize == size, "binlog.1 %d != %d", gotsize, size);
 777     gotsize = filesize(b2);
 778     assertf(gotsize == size, "binlog.2 %d != %d", gotsize, size);
 779
 780     killsrv();
 781     execlp("rm", "rm", "-rf", dir, NULL);
 782 }
 783
 784
 785 void
 786 cttestbinlogallocation()
 787 {
 788     int i = 0;
 789
 790     size = 601;
 791     mkdtemp(dir);
 792     srv.wal.dir = dir;
 793     srv.wal.use = 1;
 794     srv.wal.filesize = size;
 795     srv.wal.syncrate = 0;
 796     srv.wal.wantsync = 1;
 797
 798     port = SERVER();
 799     fd = mustdiallocal(port);
 800     for (i = 1; i <= 96; i++) {
 801         mustsend(fd, "put 0 0 120 22\r\n");
 802         mustsend(fd, "job payload xxxxxxxxxx\r\n");
 803         ckresp(fd, fmtalloc("INSERTED %d\r\n", i));
 804     }
 805     for (i = 1; i <= 96; i++) {
 806         mustsend(fd, fmtalloc("delete %d\r\n", i));
 807         ckresp(fd, "DELETED\r\n");
 808     }
 809
 810     killsrv();
 811     execlp("rm", "rm", "-rf", dir, NULL);
 812 }
 813
 814
 815 void
 816 cttestbinlogread()
 817 {
 818     mkdtemp(dir);
 819     srv.wal.dir = dir;
 820     srv.wal.use = 1;
 821     srv.wal.syncrate = 0;
 822     srv.wal.wantsync = 1;
 823
 824     port = SERVER();
 825     fd = mustdiallocal(port);
 826     mustsend(fd, "use test\r\n");
 827     ckresp(fd, "USING test\r\n");
 828     mustsend(fd, "put 0 0 120 4\r\n");
 829     mustsend(fd, "test\r\n");
 830     ckresp(fd, "INSERTED 1\r\n");
 831     mustsend(fd, "put 0 0 120 4\r\n");
 832     mustsend(fd, "tes1\r\n");
 833     ckresp(fd, "INSERTED 2\r\n");
 834     mustsend(fd, "watch test\r\n");
 835     ckresp(fd, "WATCHING 2\r\n");
 836     mustsend(fd, "reserve\r\n");
 837     ckresp(fd, "RESERVED 1 4\r\n");
 838     ckresp(fd, "test\r\n");
 839     mustsend(fd, "release 1 1 1\r\n");
 840     ckresp(fd, "RELEASED\r\n");
 841     mustsend(fd, "reserve\r\n");
 842     ckresp(fd, "RESERVED 2 4\r\n");
 843     ckresp(fd, "tes1\r\n");
 844     mustsend(fd, "delete 2\r\n");
 845     ckresp(fd, "DELETED\r\n");
 846
 847     killsrv();
 848     waitpid(srvpid, NULL, 0);
 849
 850     port = SERVER();
 851     fd = mustdiallocal(port);
 852     mustsend(fd, "watch test\r\n");
 853     ckresp(fd, "WATCHING 2\r\n");
 854     mustsend(fd, "reserve\r\n");
 855     ckresp(fd, "RESERVED 1 4\r\n");
 856     ckresp(fd, "test\r\n");
 857     mustsend(fd, "delete 1\r\n");
 858     ckresp(fd, "DELETED\r\n");
 859     mustsend(fd, "delete 2\r\n");
 860     ckresp(fd, "NOT_FOUND\r\n");
 861
 862     killsrv();
 863     execlp("rm", "rm", "-rf", dir, NULL);
 864 }
 865
 866
 867 void
 868 cttestbinlogdiskfull()
 869 {
 870     size = 1000;
 871     falloc = &wrapfalloc;
 872     fallocpat[0] = 1;
 873     fallocpat[2] = 1;
 874
 875     mkdtemp(dir);
 876     srv.wal.dir = dir;
 877     srv.wal.use = 1;
 878     srv.wal.filesize = size;
 879     srv.wal.syncrate = 0;
 880     srv.wal.wantsync = 1;
 881
 882     port = SERVER();
 883     fd = mustdiallocal(port);
 884     mustsend(fd, "put 0 0 100 50\r\n");
 885     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 886     ckresp(fd, "INSERTED 1\r\n");
 887     mustsend(fd, "put 0 0 100 50\r\n");
 888     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 889     ckresp(fd, "INSERTED 2\r\n");
 890     mustsend(fd, "put 0 0 100 50\r\n");
 891     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 892     ckresp(fd, "INSERTED 3\r\n");
 893     mustsend(fd, "put 0 0 100 50\r\n");
 894     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 895     ckresp(fd, "INSERTED 4\r\n");
 896
 897     mustsend(fd, "put 0 0 100 50\r\n");
 898     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 899     ckresp(fd, "OUT_OF_MEMORY\r\n");
 900
 901     mustsend(fd, "put 0 0 100 50\r\n");
 902     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 903     ckresp(fd, "INSERTED 6\r\n");
 904     mustsend(fd, "put 0 0 100 50\r\n");
 905     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 906     ckresp(fd, "INSERTED 7\r\n");
 907     mustsend(fd, "put 0 0 100 50\r\n");
 908     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 909     ckresp(fd, "INSERTED 8\r\n");
 910     mustsend(fd, "put 0 0 100 50\r\n");
 911     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 912     ckresp(fd, "INSERTED 9\r\n");
 913
 914     mustsend(fd, "delete 1\r\n");
 915     ckresp(fd, "DELETED\r\n");
 916     mustsend(fd, "delete 2\r\n");
 917     ckresp(fd, "DELETED\r\n");
 918     mustsend(fd, "delete 3\r\n");
 919     ckresp(fd, "DELETED\r\n");
 920     mustsend(fd, "delete 4\r\n");
 921     ckresp(fd, "DELETED\r\n");
 922     mustsend(fd, "delete 6\r\n");
 923     ckresp(fd, "DELETED\r\n");
 924     mustsend(fd, "delete 7\r\n");
 925     ckresp(fd, "DELETED\r\n");
 926     mustsend(fd, "delete 8\r\n");
 927     ckresp(fd, "DELETED\r\n");
 928     mustsend(fd, "delete 9\r\n");
 929     ckresp(fd, "DELETED\r\n");
 930
 931     killsrv();
 932     execlp("rm", "rm", "-rf", dir, NULL);
 933 }
 934
 935
 936 void
 937 cttestbinlogdiskfulldelete()
 938 {
 939     size = 1000;
 940     falloc = &wrapfalloc;
 941     fallocpat[0] = 1;
 942     fallocpat[1] = 1;
 943
 944     mkdtemp(dir);
 945     srv.wal.dir = dir;
 946     srv.wal.use = 1;
 947     srv.wal.filesize = size;
 948     srv.wal.syncrate = 0;
 949     srv.wal.wantsync = 1;
 950
 951     port = SERVER();
 952     fd = mustdiallocal(port);
 953     mustsend(fd, "put 0 0 100 50\r\n");
 954     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 955     ckresp(fd, "INSERTED 1\r\n");
 956     mustsend(fd, "put 0 0 100 50\r\n");
 957     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 958     ckresp(fd, "INSERTED 2\r\n");
 959     mustsend(fd, "put 0 0 100 50\r\n");
 960     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 961     ckresp(fd, "INSERTED 3\r\n");
 962     mustsend(fd, "put 0 0 100 50\r\n");
 963     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 964     ckresp(fd, "INSERTED 4\r\n");
 965     mustsend(fd, "put 0 0 100 50\r\n");
 966     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 967     ckresp(fd, "INSERTED 5\r\n");
 968
 969     mustsend(fd, "put 0 0 100 50\r\n");
 970     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 971     ckresp(fd, "INSERTED 6\r\n");
 972     mustsend(fd, "put 0 0 100 50\r\n");
 973     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 974     ckresp(fd, "INSERTED 7\r\n");
 975     mustsend(fd, "put 0 0 100 50\r\n");
 976     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 977     ckresp(fd, "INSERTED 8\r\n");
 978
 979     mustsend(fd, "put 0 0 100 50\r\n");
 980     mustsend(fd, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n");
 981     ckresp(fd, "OUT_OF_MEMORY\r\n");
 982
 983     assert(exist(fmtalloc("%s/binlog.1", dir)));
 984
 985     mustsend(fd, "delete 1\r\n");
 986     ckresp(fd, "DELETED\r\n");
 987     mustsend(fd, "delete 2\r\n");
 988     ckresp(fd, "DELETED\r\n");
 989     mustsend(fd, "delete 3\r\n");
 990     ckresp(fd, "DELETED\r\n");
 991     mustsend(fd, "delete 4\r\n");
 992     ckresp(fd, "DELETED\r\n");
 993     mustsend(fd, "delete 5\r\n");
 994     ckresp(fd, "DELETED\r\n");
 995     mustsend(fd, "delete 6\r\n");
 996     ckresp(fd, "DELETED\r\n");
 997     mustsend(fd, "delete 7\r\n");
 998     ckresp(fd, "DELETED\r\n");
 999     mustsend(fd, "delete 8\r\n");
1000     ckresp(fd, "DELETED\r\n");
1001
1002     killsrv();
1003     execlp("rm", "rm", "-rf", dir, NULL);
1004 }
1005
1006
1007 void
1008 cttestbinlogv5()
1009 {
1010     char portstr[10];
1011
1012     if (system("which beanstalkd-1.4.6") != 0) {
1013         puts("beanstalkd 1.4.6 not found, skipping");
1014         exit(0);
1015     }
1016
1017     installabrt();
1018     mkdtemp(dir);
1019
1020     progname=__func__;
1021     port = (rand()&0xfbff) + 1024;
1022     sprintf(portstr, "%d", port);
1023     muststart("beanstalkd-1.4.6", "-b", dir, "-p", portstr);
1024     fd = mustdiallocal(port);
1025     mustsend(fd, "use test\r\n");
1026     ckresp(fd, "USING test\r\n");
1027     mustsend(fd, "put 1 2 3 4\r\n");
1028     mustsend(fd, "test\r\n");
1029     ckresp(fd, "INSERTED 1\r\n");
1030     mustsend(fd, "put 4 3 2 1\r\n");
1031     mustsend(fd, "x\r\n");
1032     ckresp(fd, "INSERTED 2\r\n");
1033
1034     mustsend(fd, "stats-job 1\r\n");
1035     ckrespsub(fd, "OK ");
1036     ckrespsub(fd, "\nid: 1\n");
1037     mustsend(fd, "stats-job 1\r\n");
1038     ckrespsub(fd, "OK ");
1039     ckrespsub(fd, "\ntube: test\n");
1040     mustsend(fd, "stats-job 1\r\n");
1041     ckrespsub(fd, "OK ");
1042     ckrespsub(fd, "\nstate: delayed\n");
1043     mustsend(fd, "stats-job 1\r\n");
1044     ckrespsub(fd, "OK ");
1045     ckrespsub(fd, "\npri: 1\n");
1046     mustsend(fd, "stats-job 1\r\n");
1047     ckrespsub(fd, "OK ");
1048     ckrespsub(fd, "\ndelay: 2\n");
1049     mustsend(fd, "stats-job 1\r\n");
1050     ckrespsub(fd, "OK ");
1051     ckrespsub(fd, "\nttr: 3\n");
1052     mustsend(fd, "stats-job 1\r\n");
1053     ckrespsub(fd, "OK ");
1054     ckrespsub(fd, "\nreserves: 0\n");
1055     mustsend(fd, "stats-job 1\r\n");
1056     ckrespsub(fd, "OK ");
1057     ckrespsub(fd, "\ntimeouts: 0\n");
1058     mustsend(fd, "stats-job 1\r\n");
1059     ckrespsub(fd, "OK ");
1060     ckrespsub(fd, "\nreleases: 0\n");
1061     mustsend(fd, "stats-job 1\r\n");
1062     ckrespsub(fd, "OK ");
1063     ckrespsub(fd, "\nburies: 0\n");
1064     mustsend(fd, "stats-job 1\r\n");
1065     ckrespsub(fd, "OK ");
1066     ckrespsub(fd, "\nkicks: 0\n");
1067
1068     mustsend(fd, "stats-job 2\r\n");
1069     ckrespsub(fd, "OK ");
1070     ckrespsub(fd, "\nid: 2\n");
1071     mustsend(fd, "stats-job 2\r\n");
1072     ckrespsub(fd, "OK ");
1073     ckrespsub(fd, "\ntube: test\n");
1074     mustsend(fd, "stats-job 2\r\n");
1075     ckrespsub(fd, "OK ");
1076     ckrespsub(fd, "\nstate: delayed\n");
1077     mustsend(fd, "stats-job 2\r\n");
1078     ckrespsub(fd, "OK ");
1079     ckrespsub(fd, "\npri: 4\n");
1080     mustsend(fd, "stats-job 2\r\n");
1081     ckrespsub(fd, "OK ");
1082     ckrespsub(fd, "\ndelay: 3\n");
1083     mustsend(fd, "stats-job 2\r\n");
1084     ckrespsub(fd, "OK ");
1085     ckrespsub(fd, "\nttr: 2\n");
1086     mustsend(fd, "stats-job 2\r\n");
1087     ckrespsub(fd, "OK ");
1088     ckrespsub(fd, "\nreserves: 0\n");
1089     mustsend(fd, "stats-job 2\r\n");
1090     ckrespsub(fd, "OK ");
1091     ckrespsub(fd, "\ntimeouts: 0\n");
1092     mustsend(fd, "stats-job 2\r\n");
1093     ckrespsub(fd, "OK ");
1094     ckrespsub(fd, "\nreleases: 0\n");
1095     mustsend(fd, "stats-job 2\r\n");
1096     ckrespsub(fd, "OK ");
1097     ckrespsub(fd, "\nburies: 0\n");
1098     mustsend(fd, "stats-job 2\r\n");
1099     ckrespsub(fd, "OK ");
1100     ckrespsub(fd, "\nkicks: 0\n");
1101
1102     killsrv();
1103
1104     srv.wal.dir = dir;
1105     srv.wal.use = 1;
1106     srv.wal.syncrate = 0;
1107     srv.wal.wantsync = 1;
1108
1109     port = SERVER();
1110     fd = mustdiallocal(port);
1111
1112     mustsend(fd, "stats-job 1\r\n");
1113     ckrespsub(fd, "OK ");
1114     ckrespsub(fd, "\nid: 1\n");
1115     mustsend(fd, "stats-job 1\r\n");
1116     ckrespsub(fd, "OK ");
1117     ckrespsub(fd, "\ntube: test\n");
1118     mustsend(fd, "stats-job 1\r\n");
1119     ckrespsub(fd, "OK ");
1120     ckrespsub(fd, "\nstate: delayed\n");
1121     mustsend(fd, "stats-job 1\r\n");
1122     ckrespsub(fd, "OK ");
1123     ckrespsub(fd, "\npri: 1\n");
1124     mustsend(fd, "stats-job 1\r\n");
1125     ckrespsub(fd, "OK ");
1126     ckrespsub(fd, "\ndelay: 2\n");
1127     mustsend(fd, "stats-job 1\r\n");
1128     ckrespsub(fd, "OK ");
1129     ckrespsub(fd, "\nttr: 3\n");
1130     mustsend(fd, "stats-job 1\r\n");
1131     ckrespsub(fd, "OK ");
1132     ckrespsub(fd, "\nreserves: 0\n");
1133     mustsend(fd, "stats-job 1\r\n");
1134     ckrespsub(fd, "OK ");
1135     ckrespsub(fd, "\ntimeouts: 0\n");
1136     mustsend(fd, "stats-job 1\r\n");
1137     ckrespsub(fd, "OK ");
1138     ckrespsub(fd, "\nreleases: 0\n");
1139     mustsend(fd, "stats-job 1\r\n");
1140     ckrespsub(fd, "OK ");
1141     ckrespsub(fd, "\nburies: 0\n");
1142     mustsend(fd, "stats-job 1\r\n");
1143     ckrespsub(fd, "OK ");
1144     ckrespsub(fd, "\nkicks: 0\n");
1145
1146     mustsend(fd, "stats-job 2\r\n");
1147     ckrespsub(fd, "OK ");
1148     ckrespsub(fd, "\nid: 2\n");
1149     mustsend(fd, "stats-job 2\r\n");
1150     ckrespsub(fd, "OK ");
1151     ckrespsub(fd, "\ntube: test\n");
1152     mustsend(fd, "stats-job 2\r\n");
1153     ckrespsub(fd, "OK ");
1154     ckrespsub(fd, "\nstate: delayed\n");
1155     mustsend(fd, "stats-job 2\r\n");
1156     ckrespsub(fd, "OK ");
1157     ckrespsub(fd, "\npri: 4\n");
1158     mustsend(fd, "stats-job 2\r\n");
1159     ckrespsub(fd, "OK ");
1160     ckrespsub(fd, "\ndelay: 3\n");
1161     mustsend(fd, "stats-job 2\r\n");
1162     ckrespsub(fd, "OK ");
1163     ckrespsub(fd, "\nttr: 2\n");
1164     mustsend(fd, "stats-job 2\r\n");
1165     ckrespsub(fd, "OK ");
1166     ckrespsub(fd, "\nreserves: 0\n");
1167     mustsend(fd, "stats-job 2\r\n");
1168     ckrespsub(fd, "OK ");
1169     ckrespsub(fd, "\ntimeouts: 0\n");
1170     mustsend(fd, "stats-job 2\r\n");
1171     ckrespsub(fd, "OK ");
1172     ckrespsub(fd, "\nreleases: 0\n");
1173     mustsend(fd, "stats-job 2\r\n");
1174     ckrespsub(fd, "OK ");
1175     ckrespsub(fd, "\nburies: 0\n");
1176     mustsend(fd, "stats-job 2\r\n");
1177     ckrespsub(fd, "OK ");
1178     ckrespsub(fd, "\nkicks: 0\n");
1179
1180     killsrv();
1181     execlp("rm", "rm", "-rf", dir, NULL);
1182 }