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