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