repcached.patch

   1 diff -urN --exclude *.m4 memcached-1.4.4/Makefile.am repcached-2.2-1.4.4/Makefile.am
   2 --- memcached-1.4.4/Makefile.am Fri Oct 30 04:24:52 2009
   3 +++ repcached-2.2-1.4.4/Makefile.am     Tue Feb  9 23:02:45 2010
   4 @@ -31,6 +31,10 @@
   5  memcached_SOURCES += sasl_defs.c
   6  endif
   7
   8 +if ENABLE_REPLICATION
   9 +memcached_SOURCES += replication.h replication.c
  10 +endif
  11 +
  12  memcached_debug_SOURCES = $(memcached_SOURCES)
  13  memcached_CPPFLAGS = -DNDEBUG
  14  memcached_debug_LDADD = @PROFILER_LDFLAGS@
  15 diff -urN --exclude *.m4 memcached-1.4.4/Makefile.in repcached-2.2-1.4.4/Makefile.in
  16 diff -urN --exclude *.m4 memcached-1.4.4/assoc.c repcached-2.2-1.4.4/assoc.c
  17 --- memcached-1.4.4/assoc.c     Sat Oct 24 00:38:01 2009
  18 +++ repcached-2.2-1.4.4/assoc.c Tue Feb  9 23:02:45 2010
  19 @@ -258,3 +258,51 @@
  20  }
  21
  22
  23 +#ifdef USE_REPLICATION
  24 +char *assoc_key_snap(int *n)
  25 +{
  26 +    char *p = NULL;
  27 +    char *b = NULL;
  28 +    item *i = NULL;
  29 +    int  co = 0;
  30 +    int  sz = 1;
  31 +    int  hs = 0;
  32 +    int  hm = hashsize(hashpower);
  33 +
  34 +    hs = hm;
  35 +    while(hs--){
  36 +        if(expanding && hs < hashsize(hashpower - 1) && hs >= expand_bucket){
  37 +            i = old_hashtable[hs];
  38 +        }else{
  39 +            i = primary_hashtable[hs];
  40 +        }
  41 +        while(i){
  42 +            sz += i->nkey + 1;
  43 +            co++;
  44 +            i = i->h_next;
  45 +        }
  46 +    }
  47 +
  48 +    if(co){
  49 +        if((p = b = malloc(sz))){
  50 +            hs = hm;
  51 +            while(hs--){
  52 +                if(expanding && hs < hashsize(hashpower - 1) && hs >= expand_bucket){
  53 +                    i = old_hashtable[hs];
  54 +                }else{
  55 +                    i = primary_hashtable[hs];
  56 +                }
  57 +                while(i){
  58 +                    memcpy(p, ITEM_key(i), i->nkey);
  59 +                    p += i->nkey;
  60 +                    *(p++) = 0;
  61 +                    i = i->h_next;
  62 +                }
  63 +            }
  64 +            *(p++) = 0;
  65 +        }
  66 +    }
  67 +    if(n) *n = co;
  68 +    return(b);
  69 +}
  70 +#endif /* USE_REPLICATION */
  71 diff -urN --exclude *.m4 memcached-1.4.4/assoc.h repcached-2.2-1.4.4/assoc.h
  72 --- memcached-1.4.4/assoc.h     Sun Aug 30 03:00:58 2009
  73 +++ repcached-2.2-1.4.4/assoc.h Tue Feb  9 23:02:45 2010
  74 @@ -7,3 +7,6 @@
  75  int start_assoc_maintenance_thread(void);
  76  void stop_assoc_maintenance_thread(void);
  77
  78 +#ifdef USE_REPLICATION
  79 +char *assoc_key_snap(int *n);
  80 +#endif /*USE_REPLICATION*/
  81 diff -urN --exclude *.m4 memcached-1.4.4/config.guess repcached-2.2-1.4.4/config.guess
  82 diff -urN --exclude *.m4 memcached-1.4.4/config.h.in repcached-2.2-1.4.4/config.h.in
  83 --- memcached-1.4.4/config.h.in Fri Nov 27 09:34:56 2009
  84 +++ repcached-2.2-1.4.4/config.h.in     Wed Feb 10 19:12:46 2010
  85 @@ -99,6 +99,9 @@
  86  /* Define to 1 if you have the ANSI C header files. */
  87  #undef STDC_HEADERS
  88
  89 +/* Define this if you want to use replication */
  90 +#undef USE_REPLICATION
  91 +
  92  /* Version number of package */
  93  #undef VERSION
  94
  95 diff -urN --exclude *.m4 memcached-1.4.4/config.sub repcached-2.2-1.4.4/config.sub
  96 diff -urN --exclude *.m4 memcached-1.4.4/configure repcached-2.2-1.4.4/configure
  97 diff -urN --exclude *.m4 memcached-1.4.4/configure.ac repcached-2.2-1.4.4/configure.ac
  98 --- memcached-1.4.4/configure.ac        Wed Nov 25 03:40:29 2009
  99 +++ repcached-2.2-1.4.4/configure.ac    Tue Feb  9 23:02:45 2010
 100 @@ -382,6 +382,18 @@
 101    AC_MSG_ERROR([Can't enable threads without the POSIX thread library.])
 102  fi
 103
 104 +dnl Check whether the user wants replication or not
 105 +AC_ARG_ENABLE(replication,
 106 +  [AS_HELP_STRING([--enable-replication],[support replication])],
 107 +  [if test "x$enable_threads" = "xyes"; then
 108 +     AC_MSG_ERROR([Can't enable threads and replication together.])
 109 +   else
 110 +     AC_DEFINE([USE_REPLICATION],,[Define this if you want to use replication])
 111 +   fi
 112 +  ])
 113 +
 114 +AM_CONDITIONAL(ENABLE_REPLICATION, test "x$enable_replication" = "xyes")
 115 +
 116  AC_CHECK_FUNCS(mlockall)
 117  AC_CHECK_FUNCS(getpagesizes)
 118  AC_CHECK_FUNCS(memcntl)
 119 diff -urN --exclude *.m4 memcached-1.4.4/doc/Makefile repcached-2.2-1.4.4/doc/Makefile
 120 diff -urN --exclude *.m4 memcached-1.4.4/items.c repcached-2.2-1.4.4/items.c
 121 --- memcached-1.4.4/items.c     Sat Oct 24 00:38:01 2009
 122 +++ repcached-2.2-1.4.4/items.c Tue Feb  9 23:02:45 2010
 123 @@ -155,6 +155,9 @@
 124                      STATS_LOCK();
 125                      stats.evictions++;
 126                      STATS_UNLOCK();
 127 +#ifdef USE_REPLICATION
 128 +                    replication_call_del(ITEM_key(search), search->nkey);
 129 +#endif /* USE_REPLICATION */
 130                  }
 131                  do_item_unlink(search);
 132                  break;
 133 @@ -288,8 +291,14 @@
 134      stats.total_items += 1;
 135      STATS_UNLOCK();
 136
 137 +#ifdef USE_REPLICATION
 138 +    /* Allocate a new CAS ID on link. */
 139 +    if(!(it->it_flags & ITEM_REPDATA))
 140 +        ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0);
 141 +#else
 142      /* Allocate a new CAS ID on link. */
 143      ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0);
 144 +#endif /* USE_REPLICATION */
 145
 146      item_link_q(it);
 147
 148 diff -urN --exclude *.m4 memcached-1.4.4/memcached.c repcached-2.2-1.4.4/memcached.c
 149 --- memcached-1.4.4/memcached.c Fri Nov 27 08:45:13 2009
 150 +++ repcached-2.2-1.4.4/memcached.c     Wed Feb 10 16:08:37 2010
 151 @@ -102,6 +102,30 @@
 152
 153  static void conn_free(conn *c);
 154
 155 +#ifdef USE_REPLICATION
 156 +static int   rep_exit = 0;
 157 +static conn *rep_recv = NULL;
 158 +static conn *rep_send = NULL;
 159 +static conn *rep_conn = NULL;
 160 +static conn *rep_serv = NULL;
 161 +static int  server_socket_replication(const int);
 162 +static void server_close_replication(void);
 163 +static int  replication_init(void);
 164 +static int  replication_server_init(void);
 165 +static int  replication_client_init(void);
 166 +static int  replication_start(void);
 167 +static int  replication_connect(void);
 168 +static int  replication_close(void);
 169 +static void replication_dispatch_close(void);
 170 +static int  replication_marugoto(int);
 171 +static int  replication_send(conn *);
 172 +static int  replication_pop(void);
 173 +static int  replication_push(void);
 174 +static int  replication_exit(void);
 175 +static int  replication_item(Q_ITEM *);
 176 +static pthread_mutex_t replication_pipe_lock = PTHREAD_MUTEX_INITIALIZER;
 177 +#endif /* USE_REPLICATION */
 178 +
 179  /** exported globals **/
 180  struct stats stats;
 181  struct settings settings;
 182 @@ -194,6 +218,11 @@
 183      settings.backlog = 1024;
 184      settings.binding_protocol = negotiating_prot;
 185      settings.item_size_max = 1024 * 1024; /* The famous 1MB upper limit. */
 186 +#ifdef USE_REPLICATION
 187 +    settings.rep_addr.s_addr = htonl(INADDR_ANY);
 188 +    settings.rep_port = 11212;
 189 +    settings.rep_qmax = 8192;
 190 +#endif /* USE_REPLICATION */
 191  }
 192
 193  /*
 194 @@ -382,6 +411,10 @@
 195                  prot_text(c->protocol));
 196          } else if (IS_UDP(transport)) {
 197              fprintf(stderr, "<%d server listening (udp)\n", sfd);
 198 +#ifdef USE_REPLICATION
 199 +        } else if (init_state == conn_rep_listen) {
 200 +            fprintf(stderr, "<%d server listening (replication)\n", sfd);
 201 +#endif /* USE_REPLICATION */
 202          } else if (c->protocol == negotiating_prot) {
 203              fprintf(stderr, "<%d new auto-negotiating client connection\n",
 204                      sfd);
 205 @@ -593,7 +626,11 @@
 206                                         "conn_nread",
 207                                         "conn_swallow",
 208                                         "conn_closing",
 209 -                                       "conn_mwrite" };
 210 +                                       "conn_mwrite",
 211 +                                       "conn_repconnect",
 212 +                                       "conn_rep_listen",
 213 +                                       "conn_pipe_recv",
 214 +                                       "conn_pipe_send" };
 215      return statenames[state];
 216  }
 217
 218 @@ -752,6 +789,14 @@
 219
 220      assert(c != NULL);
 221
 222 +#ifdef USE_REPLICATION
 223 +    if (c == rep_conn){
 224 +        if (settings.verbose > 1)
 225 +            fprintf(stderr, "REP>%d %s\n", c->sfd, str);
 226 +        conn_set_state(c, conn_new_cmd);
 227 +        return;
 228 +    }
 229 +#endif /* USE_REPLICATION */
 230      if (c->noreply) {
 231          if (settings.verbose > 1)
 232              fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);
 233 @@ -791,9 +836,11 @@
 234      int comm = c->cmd;
 235      enum store_item_type ret;
 236
 237 -    pthread_mutex_lock(&c->thread->stats.mutex);
 238 -    c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++;
 239 -    pthread_mutex_unlock(&c->thread->stats.mutex);
 240 +    if (c->thread) {
 241 +        pthread_mutex_lock(&c->thread->stats.mutex);
 242 +        c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++;
 243 +        pthread_mutex_unlock(&c->thread->stats.mutex);
 244 +    }
 245
 246      if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) {
 247          out_string(c, "CLIENT_ERROR bad data chunk");
 248 @@ -832,6 +879,11 @@
 249
 250        switch (ret) {
 251        case STORED:
 252 +#ifdef USE_REPLICATION
 253 +          if( c != rep_conn ){
 254 +            replication_call_rep(ITEM_key(it), it->nkey);
 255 +          }
 256 +#endif /* USE_REPLICATION */
 257            out_string(c, "STORED");
 258            break;
 259        case EXISTS:
 260 @@ -2410,6 +2462,11 @@
 261      APPEND_STAT("listen_disabled_num", "%llu", (unsigned long long)stats.listen_disabled_num);
 262      APPEND_STAT("threads", "%d", settings.num_threads);
 263      APPEND_STAT("conn_yields", "%llu", (unsigned long long)thread_stats.conn_yields);
 264 +#ifdef USE_REPLICATION
 265 +    APPEND_STAT("replication", "MASTER", 0);
 266 +    APPEND_STAT("repcached_version", "%s", REPCACHED_VERSION);
 267 +    APPEND_STAT("repcached_qi_free", "%u", settings.rep_qmax - get_qi_count());
 268 +#endif /*USE_REPLICATION*/
 269      STATS_UNLOCK();
 270  }
 271
 272 @@ -2797,6 +2854,11 @@
 273      switch(add_delta(c, it, incr, delta, temp)) {
 274      case OK:
 275          out_string(c, temp);
 276 +#ifdef USE_REPLICATION
 277 +        if( c != rep_conn){
 278 +            replication_call_rep(ITEM_key(it), it->nkey);
 279 +        }
 280 +#endif /* USE_REPLICATION */
 281          break;
 282      case NON_NUMERIC:
 283          out_string(c, "CLIENT_ERROR cannot increment or decrement non-numeric value");
 284 @@ -2911,17 +2973,25 @@
 285      if (it) {
 286          MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey);
 287
 288 -        pthread_mutex_lock(&c->thread->stats.mutex);
 289 -        c->thread->stats.slab_stats[it->slabs_clsid].delete_hits++;
 290 -        pthread_mutex_unlock(&c->thread->stats.mutex);
 291 +        if (c->thread) {
 292 +            pthread_mutex_lock(&c->thread->stats.mutex);
 293 +            c->thread->stats.slab_stats[it->slabs_clsid].delete_hits++;
 294 +            pthread_mutex_unlock(&c->thread->stats.mutex);
 295 +        }
 296
 297          item_unlink(it);
 298          item_remove(it);      /* release our reference */
 299 +#ifdef USE_REPLICATION
 300 +        if( c != rep_conn )
 301 +            replication_call_del(key, nkey);
 302 +#endif /* USE_REPLICATION */
 303          out_string(c, "DELETED");
 304      } else {
 305 -        pthread_mutex_lock(&c->thread->stats.mutex);
 306 -        c->thread->stats.delete_misses++;
 307 -        pthread_mutex_unlock(&c->thread->stats.mutex);
 308 +        if (c->thread) {
 309 +            pthread_mutex_lock(&c->thread->stats.mutex);
 310 +            c->thread->stats.delete_misses++;
 311 +            pthread_mutex_unlock(&c->thread->stats.mutex);
 312 +        }
 313
 314          out_string(c, "NOT_FOUND");
 315      }
 316 @@ -2986,6 +3056,22 @@
 317
 318          process_update_command(c, tokens, ntokens, comm, true);
 319
 320 +#ifdef USE_REPLICATION
 321 +    } else if ((ntokens == 7) && (strcmp(tokens[COMMAND_TOKEN].value, "rep") == 0 && (comm = NREAD_SET)) && (c == rep_conn)) {
 322 +
 323 +        process_update_command(c, tokens, ntokens, comm, true);
 324 +        if(c->item)
 325 +            ((item *)(c->item))->it_flags |= ITEM_REPDATA;
 326 +
 327 +    } else if ((ntokens == 2) && (strcmp(tokens[COMMAND_TOKEN].value, "marugoto_end") == 0) && (c == rep_conn)) {
 328 +        if(replication_start() == -1)
 329 +            exit(EXIT_FAILURE);
 330 +        if (settings.verbose > 0)
 331 +            fprintf(stderr,"replication: start\n");
 332 +        out_string(c, "OK");
 333 +        return;
 334 +
 335 +#endif /* USE_REPLICATION */
 336      } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) {
 337
 338          process_arithmetic_command(c, tokens, ntokens, 1);
 339 @@ -3012,11 +3098,17 @@
 340
 341          set_noreply_maybe(c, tokens, ntokens);
 342
 343 -        pthread_mutex_lock(&c->thread->stats.mutex);
 344 -        c->thread->stats.flush_cmds++;
 345 -        pthread_mutex_unlock(&c->thread->stats.mutex);
 346 +        if (c->thread) {
 347 +            pthread_mutex_lock(&c->thread->stats.mutex);
 348 +            c->thread->stats.flush_cmds++;
 349 +            pthread_mutex_unlock(&c->thread->stats.mutex);
 350 +        }
 351
 352          if(ntokens == (c->noreply ? 3 : 2)) {
 353 +#ifdef USE_REPLICATION
 354 +            if( c != rep_conn )
 355 +                replication_call_flush_all();
 356 +#endif
 357              settings.oldest_live = current_time - 1;
 358              item_flush_expired();
 359              out_string(c, "OK");
 360 @@ -3029,6 +3121,11 @@
 361              return;
 362          }
 363
 364 +#ifdef USE_REPLICATION
 365 +        if( c != rep_conn )
 366 +            replication_call_defer_flush_all(realtime(exptime) + process_started);
 367 +#endif
 368 +        settings.oldest_live = realtime(exptime) - 1;
 369          /*
 370            If exptime is zero realtime() would return zero too, and
 371            realtime(exptime) - 1 would overflow to the max unsigned
 372 @@ -3275,9 +3372,11 @@
 373          int avail = c->rsize - c->rbytes;
 374          res = read(c->sfd, c->rbuf + c->rbytes, avail);
 375          if (res > 0) {
 376 -            pthread_mutex_lock(&c->thread->stats.mutex);
 377 -            c->thread->stats.bytes_read += res;
 378 -            pthread_mutex_unlock(&c->thread->stats.mutex);
 379 +            if (c->thread) {
 380 +                pthread_mutex_lock(&c->thread->stats.mutex);
 381 +                c->thread->stats.bytes_read += res;
 382 +                pthread_mutex_unlock(&c->thread->stats.mutex);
 383 +            }
 384              gotdata = READ_DATA_RECEIVED;
 385              c->rbytes += res;
 386              if (res == avail) {
 387 @@ -3423,6 +3522,12 @@
 388
 389      assert(c != NULL);
 390
 391 +#ifdef USE_REPLICATION
 392 +    if(rep_exit && (c->state != conn_pipe_recv)){
 393 +        return;
 394 +    }
 395 +#endif /* USE_REPLICATION */
 396 +
 397      while (!stop) {
 398
 399          switch(c->state) {
 400 @@ -3502,9 +3607,11 @@
 401              if (nreqs >= 0) {
 402                  reset_cmd_handler(c);
 403              } else {
 404 -                pthread_mutex_lock(&c->thread->stats.mutex);
 405 -                c->thread->stats.conn_yields++;
 406 -                pthread_mutex_unlock(&c->thread->stats.mutex);
 407 +                if (c->thread) {
 408 +                    pthread_mutex_lock(&c->thread->stats.mutex);
 409 +                    c->thread->stats.conn_yields++;
 410 +                    pthread_mutex_unlock(&c->thread->stats.mutex);
 411 +                }
 412                  if (c->rbytes > 0) {
 413                      /* We have already read in data into the input buffer,
 414                         so libevent will most likely not signal read events
 415 @@ -3545,9 +3652,11 @@
 416              /*  now try reading from the socket */
 417              res = read(c->sfd, c->ritem, c->rlbytes);
 418              if (res > 0) {
 419 -                pthread_mutex_lock(&c->thread->stats.mutex);
 420 -                c->thread->stats.bytes_read += res;
 421 -                pthread_mutex_unlock(&c->thread->stats.mutex);
 422 +                if (c->thread) {
 423 +                    pthread_mutex_lock(&c->thread->stats.mutex);
 424 +                    c->thread->stats.bytes_read += res;
 425 +                    pthread_mutex_unlock(&c->thread->stats.mutex);
 426 +                }
 427                  if (c->rcurr == c->ritem) {
 428                      c->rcurr += res;
 429                  }
 430 @@ -3600,9 +3709,11 @@
 431              /*  now try reading from the socket */
 432              res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize);
 433              if (res > 0) {
 434 -                pthread_mutex_lock(&c->thread->stats.mutex);
 435 -                c->thread->stats.bytes_read += res;
 436 -                pthread_mutex_unlock(&c->thread->stats.mutex);
 437 +                if (c->thread) {
 438 +                    pthread_mutex_lock(&c->thread->stats.mutex);
 439 +                    c->thread->stats.bytes_read += res;
 440 +                    pthread_mutex_unlock(&c->thread->stats.mutex);
 441 +                }
 442                  c->sbytes -= res;
 443                  break;
 444              }
 445 @@ -3698,6 +3809,10 @@
 446          case conn_closing:
 447              if (IS_UDP(c->transport))
 448                  conn_cleanup(c);
 449 +#ifdef USE_REPLICATION
 450 +            else if(c == rep_conn)
 451 +                replication_close();
 452 +#endif /*USE_REPLICATION*/
 453              else
 454                  conn_close(c);
 455              stop = true;
 456 @@ -3706,6 +3821,70 @@
 457          case conn_max_state:
 458              assert(false);
 459              break;
 460 +
 461 +#ifdef USE_REPLICATION
 462 +        case conn_pipe_recv:
 463 +            if(replication_pop()){
 464 +                replication_close();
 465 +            }else{
 466 +                replication_send(rep_conn);
 467 +            }
 468 +            stop = true;
 469 +            break;
 470 +
 471 +        case conn_rep_listen:
 472 +            if (settings.verbose > 0)
 473 +                fprintf(stderr,"replication: accept\n");
 474 +            addrlen = sizeof(addr);
 475 +            res = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
 476 +            if(res == -1){
 477 +                if(errno == EAGAIN || errno == EWOULDBLOCK) {
 478 +                } else if (errno == EMFILE) {
 479 +                    fprintf(stderr, "replication: Too many opened connections\n");
 480 +                } else {
 481 +                    fprintf(stderr, "replication: accept error\n");
 482 +                }
 483 +            }else{
 484 +                if(rep_conn){
 485 +                    close(res);
 486 +                    fprintf(stderr,"replication: already connected\n");
 487 +                }else{
 488 +                    if((flags = fcntl(res, F_GETFL, 0)) < 0 || fcntl(res, F_SETFL, flags | O_NONBLOCK) < 0){
 489 +                        close(res);
 490 +                        fprintf(stderr, "replication: Can't Setting O_NONBLOCK\n");
 491 +                    }else{
 492 +                        server_close_replication();
 493 +                        rep_conn = conn_new(res, conn_read, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, tcp_transport, main_base);
 494 +                        rep_conn->item   = NULL;
 495 +                        rep_conn->rbytes = 0;
 496 +                        rep_conn->rcurr  = rep_conn->rbuf;
 497 +                        replication_connect();
 498 +                        replication_marugoto(1);
 499 +                        replication_marugoto(0);
 500 +                    }
 501 +                }
 502 +            }
 503 +            stop = true;
 504 +            break;
 505 +
 506 +        case conn_repconnect:
 507 +            rep_conn = c;
 508 +            replication_connect();
 509 +            conn_set_state(c, conn_read);
 510 +            if (settings.verbose > 0)
 511 +                fprintf(stderr,"replication: marugoto copying\n");
 512 +            if(!update_event(c, EV_READ | EV_PERSIST)){
 513 +                fprintf(stderr, "replication: Couldn't update event\n");
 514 +                conn_set_state(c, conn_closing);
 515 +            }
 516 +            stop = true;
 517 +            break;
 518 +
 519 +        case conn_pipe_send:
 520 +            /* should not happen */
 521 +            fprintf(stderr, "replication: unexpected conn_pipe_send state\n");
 522 +            break;
 523 +#endif /* USE_REPLICATION */
 524          }
 525      }
 526
 527 @@ -4002,6 +4181,89 @@
 528      return 0;
 529  }
 530
 531 +#ifdef USE_REPLICATION
 532 +static int server_socket_replication(const int port) {
 533 +    int sfd;
 534 +    struct linger ling = {0, 0};
 535 +    struct addrinfo *ai;
 536 +    struct addrinfo *next;
 537 +    struct addrinfo hints;
 538 +    char port_buf[NI_MAXSERV];
 539 +    int error;
 540 +    int success = 0;
 541 +
 542 +    int flags =1;
 543 +
 544 +    memset(&hints, 0, sizeof (hints));
 545 +    hints.ai_flags = AI_PASSIVE|AI_ADDRCONFIG;
 546 +    hints.ai_family = AF_UNSPEC;
 547 +    hints.ai_protocol = IPPROTO_TCP;
 548 +    hints.ai_socktype = SOCK_STREAM;
 549 +    snprintf(port_buf, NI_MAXSERV, "%d", port);
 550 +    error= getaddrinfo(settings.inter, port_buf, &hints, &ai);
 551 +    if (error != 0) {
 552 +      if (error != EAI_SYSTEM)
 553 +        fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));
 554 +      else
 555 +        perror("getaddrinfo()");
 556 +
 557 +      return 1;
 558 +    }
 559 +
 560 +    for (next= ai; next; next= next->ai_next) {
 561 +        conn *rep_serv_add;
 562 +        if ((sfd = new_socket(next)) == -1) {
 563 +            freeaddrinfo(ai);
 564 +            return 1;
 565 +        }
 566 +        setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
 567 +        setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));
 568 +        setsockopt(sfd, SOL_SOCKET, SO_LINGER,    (void *)&ling,  sizeof(ling));
 569 +        setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));
 570 +
 571 +        if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {
 572 +            if (errno != EADDRINUSE) {
 573 +                perror("bind()");
 574 +                close(sfd);
 575 +                freeaddrinfo(ai);
 576 +                return 1;
 577 +            }
 578 +            close(sfd);
 579 +            continue;
 580 +        } else {
 581 +            success++;
 582 +            if (listen(sfd, 1024) == -1) {
 583 +                perror("listen()");
 584 +                close(sfd);
 585 +                freeaddrinfo(ai);
 586 +                return 1;
 587 +            }
 588 +        }
 589 +
 590 +        if (!(rep_serv_add = conn_new(sfd, conn_rep_listen,
 591 +                                       EV_READ | EV_PERSIST, 1, tcp_transport, main_base))) {
 592 +            fprintf(stderr, "failed to create replication server connection\n");
 593 +            exit(EXIT_FAILURE);
 594 +        }
 595 +
 596 +        rep_serv_add->next = rep_serv;
 597 +        rep_serv = rep_serv_add;
 598 +    }
 599 +
 600 +    freeaddrinfo(ai);
 601 +
 602 +    /* Return zero iff we detected no errors in starting up connections */
 603 +    return success == 0;
 604 +}
 605 +
 606 +static void server_close_replication(void) {
 607 +  while(rep_serv){
 608 +      conn_close(rep_serv);
 609 +      rep_serv = rep_serv->next;
 610 +  }
 611 +}
 612 +#endif /* USE_REPLICATION */
 613 +
 614  /*
 615   * We keep the current time of day in a global variable that's updated by a
 616   * timer event. This saves us a bunch of time() system calls (we really only
 617 @@ -4041,6 +4303,9 @@
 618
 619  static void usage(void) {
 620      printf(PACKAGE " " VERSION "\n");
 621 +#ifdef USE_REPLICATION
 622 +    printf("repcached %s\n",REPCACHED_VERSION);
 623 +#endif /* USE_REPLICATION */
 624      printf("-p <num>      TCP port number to listen on (default: 11211)\n"
 625             "-U <num>      UDP port number to listen on (default: 11211, 0 is off)\n"
 626             "-s <file>     UNIX socket path to listen on (disables network support)\n"
 627 @@ -4088,6 +4353,10 @@
 628  #ifdef ENABLE_SASL
 629      printf("-S            Turn on Sasl authentication\n");
 630  #endif
 631 +#ifdef USE_REPLICATION
 632 +    printf("-x <ip_addr>  hostname or IP address of peer repcached\n");
 633 +    printf("-X <num>      TCP port number for replication (default: 11212)\n");
 634 +#endif /* USE_REPLICATION */
 635      return;
 636  }
 637
 638 @@ -4194,6 +4463,26 @@
 639      exit(EXIT_SUCCESS);
 640  }
 641
 642 +#ifdef USE_REPLICATION
 643 +static void sig_handler_cb(int fd, short event, void *arg)
 644 +{
 645 +    struct event *signal = arg;
 646 +
 647 +    if (settings.verbose)
 648 +        fprintf(stderr, "got signal %d\n", EVENT_SIGNAL(signal));
 649 +
 650 +    if (replication_exit()) {
 651 +        exit(EXIT_FAILURE);
 652 +    }
 653 +
 654 +    pthread_mutex_lock(&replication_pipe_lock);
 655 +    if (!rep_send) {
 656 +        exit(EXIT_SUCCESS);
 657 +    }
 658 +    pthread_mutex_unlock(&replication_pipe_lock);
 659 +}
 660 +#endif /* USE_REPLICATION */
 661 +
 662  #ifndef HAVE_SIGIGNORE
 663  static int sigignore(int sig) {
 664      struct sigaction sa = { .sa_handler = SIG_IGN, .sa_flags = 0 };
 665 @@ -4249,6 +4538,57 @@
 666  #endif
 667  }
 668
 669 +static void create_listening_sockets(void)
 670 +{
 671 +    /* create unix mode sockets after dropping privileges */
 672 +    if (settings.socketpath != NULL) {
 673 +        errno = 0;
 674 +        if (server_socket_unix(settings.socketpath,settings.access)) {
 675 +            vperror("failed to listen on UNIX socket: %s", settings.socketpath);
 676 +            exit(EX_OSERR);
 677 +        }
 678 +    }
 679 +
 680 +    /* create the listening socket, bind it, and init */
 681 +    if (settings.socketpath == NULL) {
 682 +        const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME");
 683 +        char temp_portnumber_filename[PATH_MAX];
 684 +        FILE *portnumber_file = NULL;
 685 +
 686 +        if (portnumber_filename != NULL) {
 687 +            snprintf(temp_portnumber_filename,
 688 +                     sizeof(temp_portnumber_filename),
 689 +                     "%s.lck", portnumber_filename);
 690 +
 691 +            portnumber_file = fopen(temp_portnumber_filename, "a");
 692 +            if (portnumber_file == NULL) {
 693 +                fprintf(stderr, "Failed to open \"%s\": %s\n",
 694 +                        temp_portnumber_filename, strerror(errno));
 695 +            }
 696 +        }
 697 +
 698 +        errno = 0;
 699 +        if (settings.port && server_socket(settings.port, tcp_transport,
 700 +                                           portnumber_file)) {
 701 +            vperror("failed to listen on TCP port %d", settings.port);
 702 +            exit(EX_OSERR);
 703 +        }
 704 +
 705 +        /* create the UDP listening socket and bind it */
 706 +        errno = 0;
 707 +        if (settings.udpport && server_socket(settings.udpport, udp_transport,
 708 +                                              portnumber_file)) {
 709 +            vperror("failed to listen on UDP port %d", settings.udpport);
 710 +            exit(EX_OSERR);
 711 +        }
 712 +
 713 +        if (portnumber_file) {
 714 +            fclose(portnumber_file);
 715 +            rename(temp_portnumber_filename, portnumber_filename);
 716 +        }
 717 +    }
 718 +}
 719 +
 720  int main (int argc, char **argv) {
 721      int c;
 722      bool lock_memory = false;
 723 @@ -4261,6 +4601,11 @@
 724      struct rlimit rlim;
 725      char unit = '\0';
 726      int size_max = 0;
 727 +#ifdef USE_REPLICATION
 728 +    struct in_addr   addr;
 729 +    struct addrinfo  master_hint;
 730 +    struct addrinfo *master_addr;
 731 +#endif /* USE_REPLICATION */
 732      /* listening sockets */
 733      static int *l_socket = NULL;
 734
 735 @@ -4307,6 +4652,11 @@
 736            "B:"  /* Binding protocol */
 737            "I:"  /* Max item size */
 738            "S"   /* Sasl ON */
 739 +#ifdef USE_REPLICATION
 740 +          "X:"  /* replication port */
 741 +          "x:"  /* replication master */
 742 +          "q:"  /* replication queue length */
 743 +#endif /* USE_REPLICATION */
 744          ))) {
 745          switch (c) {
 746          case 'a':
 747 @@ -4462,6 +4812,31 @@
 748                  );
 749              }
 750              break;
 751 +#ifdef USE_REPLICATION
 752 +        case 'x':
 753 +            if (inet_pton(AF_INET, optarg, &addr) <= 0) {
 754 +                memset(&master_hint, 0, sizeof(master_hint));
 755 +                master_hint.ai_flags    = 0;
 756 +                master_hint.ai_socktype = 0;
 757 +                master_hint.ai_protocol = 0;
 758 +                if(!getaddrinfo(optarg, NULL, &master_hint, &master_addr)){
 759 +                    settings.rep_addr = ((struct sockaddr_in *)(master_addr->ai_addr)) -> sin_addr;
 760 +                    freeaddrinfo(master_addr);
 761 +                }else{
 762 +                    fprintf(stderr, "Illegal address: %s\n", optarg);
 763 +                    return 1;
 764 +                }
 765 +            } else {
 766 +                settings.rep_addr = addr;
 767 +            }
 768 +            break;
 769 +        case 'X':
 770 +            settings.rep_port = atoi(optarg);
 771 +            break;
 772 +        case 'q':
 773 +            settings.rep_qmax = atoi(optarg);
 774 +            break;
 775 +#endif /* USE_REPLICATION */
 776          case 'S': /* set Sasl authentication to true. Default is false */
 777  #ifndef ENABLE_SASL
 778              fprintf(stderr, "This server is not built with SASL support.\n");
 779 @@ -4587,6 +4962,17 @@
 780      /* initialize main thread libevent instance */
 781      main_base = event_init();
 782
 783 +#ifdef USE_REPLICATION
 784 +    /* register events for SIGINT and SIGTERM to handle them in main thread */
 785 +    struct event signal_int, signal_term;
 786 +    event_set(&signal_int, SIGINT, EV_SIGNAL|EV_PERSIST, sig_handler_cb,
 787 +              &signal_int);
 788 +    event_add(&signal_int, NULL);
 789 +    event_set(&signal_term, SIGTERM, EV_SIGNAL|EV_PERSIST, sig_handler_cb,
 790 +              &signal_term);
 791 +    event_add(&signal_term, NULL);
 792 +#endif
 793 +
 794      /* initialize other stuff */
 795      stats_init();
 796      assoc_init();
 797 @@ -4615,63 +5001,21 @@
 798      /* initialise clock event */
 799      clock_handler(0, 0, 0);
 800
 801 -    /* create unix mode sockets after dropping privileges */
 802 -    if (settings.socketpath != NULL) {
 803 -        errno = 0;
 804 -        if (server_socket_unix(settings.socketpath,settings.access)) {
 805 -            vperror("failed to listen on UNIX socket: %s", settings.socketpath);
 806 -            exit(EX_OSERR);
 807 -        }
 808 -    }
 809 -
 810 -    /* create the listening socket, bind it, and init */
 811 -    if (settings.socketpath == NULL) {
 812 -        int udp_port;
 813 -
 814 -        const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME");
 815 -        char temp_portnumber_filename[PATH_MAX];
 816 -        FILE *portnumber_file = NULL;
 817 -
 818 -        if (portnumber_filename != NULL) {
 819 -            snprintf(temp_portnumber_filename,
 820 -                     sizeof(temp_portnumber_filename),
 821 -                     "%s.lck", portnumber_filename);
 822 -
 823 -            portnumber_file = fopen(temp_portnumber_filename, "a");
 824 -            if (portnumber_file == NULL) {
 825 -                fprintf(stderr, "Failed to open \"%s\": %s\n",
 826 -                        temp_portnumber_filename, strerror(errno));
 827 -            }
 828 -        }
 829 -
 830 -        errno = 0;
 831 -        if (settings.port && server_socket(settings.port, tcp_transport,
 832 -                                           portnumber_file)) {
 833 -            vperror("failed to listen on TCP port %d", settings.port);
 834 -            exit(EX_OSERR);
 835 -        }
 836 -
 837 -        /*
 838 -         * initialization order: first create the listening sockets
 839 -         * (may need root on low ports), then drop root if needed,
 840 -         * then daemonise if needed, then init libevent (in some cases
 841 -         * descriptors created by libevent wouldn't survive forking).
 842 -         */
 843 -        udp_port = settings.udpport ? settings.udpport : settings.port;
 844 -
 845 -        /* create the UDP listening socket and bind it */
 846 -        errno = 0;
 847 -        if (settings.udpport && server_socket(settings.udpport, udp_transport,
 848 -                                              portnumber_file)) {
 849 -            vperror("failed to listen on UDP port %d", settings.udpport);
 850 -            exit(EX_OSERR);
 851 -        }
 852 +    /*
 853 +     * initialization order: first create the listening sockets
 854 +     * (may need root on low ports), then drop root if needed,
 855 +     * then daemonise if needed, then init libevent (in some cases
 856 +     * descriptors created by libevent wouldn't survive forking).
 857 +     */
 858
 859 -        if (portnumber_file) {
 860 -            fclose(portnumber_file);
 861 -            rename(temp_portnumber_filename, portnumber_filename);
 862 -        }
 863 +#ifdef USE_REPLICATION
 864 +    if(replication_init() == -1){
 865 +        fprintf(stderr, "faild to replication init\n");
 866 +        exit(EXIT_FAILURE);
 867      }
 868 +#else
 869 +    create_listening_sockets();
 870 +#endif
 871
 872      /* Drop privileges no longer needed */
 873      drop_privileges();
 874 @@ -4694,3 +5038,401 @@
 875
 876      return EXIT_SUCCESS;
 877  }
 878 +
 879 +#ifdef USE_REPLICATION
 880 +static int replication_start(void)
 881 +{
 882 +    static int start = 0;
 883 +    if(start)
 884 +        return(0);
 885 +
 886 +    create_listening_sockets();
 887 +
 888 +    start = 1;
 889 +    return(0);
 890 +}
 891 +
 892 +static int replication_server_init(void)
 893 +{
 894 +    rep_recv = NULL;
 895 +    rep_send = NULL;
 896 +    rep_conn = NULL;
 897 +    if(server_socket_replication(settings.rep_port)){
 898 +        fprintf(stderr, "replication: failed to initialize replication server socket\n");
 899 +        return(-1);
 900 +    }
 901 +    if (settings.verbose > 0)
 902 +        fprintf(stderr, "replication: listen\n");
 903 +    return(replication_start());
 904 +}
 905 +
 906 +static int replication_client_init(void)
 907 +{
 908 +    int s;
 909 +    conn *c;
 910 +    struct addrinfo    ai;
 911 +    struct sockaddr_in server;
 912 +
 913 +    rep_recv  = NULL;
 914 +    rep_send  = NULL;
 915 +    rep_conn  = NULL;
 916 +
 917 +    memset(&ai,0,sizeof(ai));
 918 +    ai.ai_family   = AF_INET;
 919 +    ai.ai_socktype = SOCK_STREAM;
 920 +    s = new_socket(&ai);
 921 +
 922 +    if(s == -1) {
 923 +        fprintf(stderr, "replication: failed to replication client socket\n");
 924 +        return(-1);
 925 +    }else{
 926 +        /* connect */
 927 +        memset((char *)&server, 0, sizeof(server));
 928 +        server.sin_family = AF_INET;
 929 +        server.sin_addr   = settings.rep_addr;
 930 +        server.sin_port   = htons(settings.rep_port);
 931 +        if (settings.verbose > 0)
 932 +            fprintf(stderr,"replication: connect (peer=%s:%d)\n", inet_ntoa(settings.rep_addr), settings.rep_port);
 933 +        if(connect(s,(struct sockaddr *)&server, sizeof(server)) == 0){
 934 +            c = conn_new(s, conn_repconnect, EV_WRITE | EV_PERSIST, DATA_BUFFER_SIZE, false, main_base);
 935 +            if(c == NULL){
 936 +                fprintf(stderr, "replication: failed to create client conn");
 937 +                close(s);
 938 +                return(-1);
 939 +            }
 940 +            drive_machine(c);
 941 +        }else{
 942 +            if(errno == EINPROGRESS){
 943 +                c = conn_new(s, conn_repconnect, EV_WRITE | EV_PERSIST, DATA_BUFFER_SIZE, false, main_base);
 944 +                if(c == NULL){
 945 +                    fprintf(stderr, "replication: failed to create client conn");
 946 +                    close(s);
 947 +                    return(-1);
 948 +                }
 949 +            }else{
 950 +                fprintf(stdout,"replication: can't connect %s:%d\n", inet_ntoa(server.sin_addr), ntohs(server.sin_port));
 951 +                close(s);
 952 +                return(-1);
 953 +            }
 954 +        }
 955 +    }
 956 +    return(0);
 957 +}
 958 +
 959 +static int replication_init(void)
 960 +{
 961 +    if(settings.rep_addr.s_addr != htonl(INADDR_ANY)){
 962 +        if(replication_client_init() != -1){
 963 +            return(0);
 964 +        }
 965 +    }
 966 +    return(replication_server_init());
 967 +}
 968 +
 969 +static int replication_connect(void)
 970 +{
 971 +    int f;
 972 +    int p[2];
 973 +
 974 +    if(pipe(p) == -1){
 975 +        fprintf(stderr, "replication: can't create pipe\n");
 976 +        return(-1);
 977 +    }else{
 978 +        if((f = fcntl(p[0], F_GETFL, 0)) < 0 || fcntl(p[0], F_SETFL, f | O_NONBLOCK) < 0) {
 979 +            fprintf(stderr, "replication: can't setting O_NONBLOCK pipe[0]\n");
 980 +            return(-1);
 981 +        }
 982 +        if((f = fcntl(p[1], F_GETFL, 0)) < 0 || fcntl(p[1], F_SETFL, f | O_NONBLOCK) < 0) {
 983 +            fprintf(stderr, "replication: can't setting O_NONBLOCK pipe[0]\n");
 984 +            return(-1);
 985 +        }
 986 +        pthread_mutex_lock(&replication_pipe_lock);
 987 +        rep_recv = conn_new(p[0], conn_pipe_recv, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, tcp_transport, main_base);
 988 +        rep_send = conn_new(p[1], conn_pipe_send, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, tcp_transport, main_base);
 989 +        event_del(&rep_send->event);
 990 +        pthread_mutex_unlock(&replication_pipe_lock);
 991 +    }
 992 +    return(0);
 993 +}
 994 +
 995 +static int replication_close(void)
 996 +{
 997 +    int     c;
 998 +    int     r;
 999 +    Q_ITEM *q;
1000 +
1001 +    if(settings.verbose > 0)
1002 +        fprintf(stderr,"replication: close\n");
1003 +    if(rep_recv){
1004 +        rep_recv->rbytes = sizeof(q);
1005 +        rep_recv->rcurr  = rep_recv->rbuf;
1006 +        c = 0;
1007 +        do{
1008 +            r = read(rep_recv->sfd, rep_recv->rcurr, rep_recv->rbytes);
1009 +            if(r == -1){
1010 +                break;
1011 +            }
1012 +            rep_recv->rbytes -= r;
1013 +            rep_recv->rcurr  += r;
1014 +            if(!rep_recv->rbytes){
1015 +                memcpy(&q, rep_recv->rbuf, sizeof(q));
1016 +                rep_recv->rbytes = sizeof(q);
1017 +                rep_recv->rcurr  = rep_recv->rbuf;
1018 +                qi_free(q);
1019 +                c++;
1020 +            }
1021 +        }while(r);
1022 +        conn_close(rep_recv);
1023 +        rep_recv = NULL;
1024 +        if (settings.verbose > 1) {
1025 +            fprintf(stderr, "replication: qitem free %d items\n", qi_free_list());
1026 +            fprintf(stderr, "replication: close recv %d items\n", c);
1027 +        }
1028 +    }
1029 +    pthread_mutex_lock(&replication_pipe_lock);
1030 +    if(rep_send){
1031 +        conn_close(rep_send);
1032 +        rep_send = NULL;
1033 +        if (settings.verbose > 1)
1034 +            fprintf(stderr,"replication: close send\n");
1035 +    }
1036 +    pthread_mutex_unlock(&replication_pipe_lock);
1037 +    if(rep_conn){
1038 +        conn_close(rep_conn);
1039 +        rep_conn = NULL;
1040 +        if (settings.verbose > 1)
1041 +            fprintf(stderr,"replication: close conn\n");
1042 +    }
1043 +    if(!rep_exit)
1044 +        replication_server_init();
1045 +    return(0);
1046 +}
1047 +
1048 +static void replication_dispatch_close(void)
1049 +{
1050 +    if (settings.verbose > 1)
1051 +        fprintf(stderr, "replication: dispatch close\n");
1052 +    pthread_mutex_lock(&replication_pipe_lock);
1053 +    if (rep_send) {
1054 +        conn_close(rep_send);
1055 +        rep_send = NULL;
1056 +    }
1057 +    pthread_mutex_unlock(&replication_pipe_lock);
1058 +}
1059 +
1060 +static int replication_marugoto(int f)
1061 +{
1062 +    static int   keysend  = 0;
1063 +    static int   keycount = 0;
1064 +    static char *keylist  = NULL;
1065 +    static char *keyptr   = NULL;
1066 +
1067 +    if(f){
1068 +        if(keylist){
1069 +            free(keylist);
1070 +            keylist  = NULL;
1071 +            keyptr   = NULL;
1072 +            keycount = 0;
1073 +            keysend  = 0;
1074 +        }
1075 +        pthread_mutex_lock(&cache_lock);
1076 +        keylist = (char *)assoc_key_snap((int *)&keycount);
1077 +        pthread_mutex_unlock(&cache_lock);
1078 +        keyptr  = keylist;
1079 +        if (!keyptr){
1080 +            replication_call_marugoto_end();
1081 +        }else{
1082 +        if (settings.verbose > 0)
1083 +            fprintf(stderr,"replication: marugoto start\n");
1084 +        }
1085 +    }else{
1086 +        if(keyptr){
1087 +            while(*keyptr){
1088 +                item *it = item_get(keyptr, strlen(keyptr));
1089 +                if(it){
1090 +                    item_remove(it);
1091 +                    if(replication_call_rep(keyptr, strlen(keyptr)) == -1){
1092 +                        return(-1);
1093 +                    }else{
1094 +                        keysend++;
1095 +                        keyptr += strlen(keyptr) + 1;
1096 +                        return(0);
1097 +                    }
1098 +                }
1099 +                keyptr += strlen(keyptr) + 1;
1100 +            }
1101 +            if(settings.verbose > 0)
1102 +                fprintf(stderr,"replication: marugoto %d\n", keysend);
1103 +            replication_call_marugoto_end();
1104 +            if(settings.verbose > 0)
1105 +                fprintf(stderr,"replication: marugoto owari\n");
1106 +            free(keylist);
1107 +            keylist  = NULL;
1108 +            keyptr   = NULL;
1109 +            keycount = 0;
1110 +            keysend  = 0;
1111 +        }
1112 +    }
1113 +    return(0);
1114 +}
1115 +
1116 +static int replication_send(conn *c)
1117 +{
1118 +    while(c->wbytes){
1119 +        int w = write(c->sfd, c->wcurr, c->wbytes);
1120 +        if(w == -1){
1121 +            if(errno == EAGAIN || errno == EINTR){
1122 +            }else{
1123 +                fprintf(stderr,"replication: send error\n");
1124 +                replication_close();
1125 +                break;
1126 +            }
1127 +        }else{
1128 +            c->wbytes -= w;
1129 +            c->wcurr  += w;
1130 +        }
1131 +    }
1132 +    return(c->wbytes);
1133 +}
1134 +
1135 +static int replication_pop(void)
1136 +{
1137 +    int      r;
1138 +    int      c;
1139 +    int      m;
1140 +    Q_ITEM **q;
1141 +
1142 +    if(settings.verbose > 1)
1143 +        fprintf(stderr, "replication: pop\n");
1144 +
1145 +    if(!rep_recv)
1146 +        return(0);
1147 +
1148 +    r = read(rep_recv->sfd, rep_recv->rbuf, rep_recv->rsize);
1149 +    if(r == -1){
1150 +        if(errno == EAGAIN || errno == EINTR){
1151 +        }else{
1152 +            fprintf(stderr,"replication: pop error %d\n", errno);
1153 +            return(-1);
1154 +        }
1155 +    }if(r == 0){
1156 +        /* other end closed, trigger replication_close() */
1157 +        return(-1);
1158 +    }else{
1159 +        c = r / sizeof(Q_ITEM *);
1160 +        m = r % sizeof(Q_ITEM *);
1161 +        q = (Q_ITEM **)(rep_recv->rbuf);
1162 +        while(c--){
1163 +            if(q[c]){
1164 +                if(rep_conn && replication_cmd(rep_conn, q[c])){
1165 +                    replication_item(q[c]); /* error retry */
1166 +                }else{
1167 +                    qi_free(q[c]);
1168 +                }
1169 +            }else{
1170 +                if(!rep_exit){
1171 +                    if (settings.verbose)
1172 +                        fprintf(stderr,"replication: cleanup start\n");
1173 +                    rep_exit = 1;
1174 +                }
1175 +            }
1176 +        }
1177 +    }
1178 +    if(rep_exit){
1179 +        if(rep_conn->wbytes){
1180 +            /* retry */
1181 +            if(replication_exit()){
1182 +                replication_close();
1183 +                fprintf(stderr,"replication: cleanup error\n");
1184 +                exit(EXIT_FAILURE);
1185 +            }
1186 +        }else{
1187 +            /* finish */
1188 +            replication_close();
1189 +            if (settings.verbose)
1190 +                fprintf(stderr,"replication: cleanup complete\n");
1191 +            exit(EXIT_SUCCESS);
1192 +        }
1193 +    }
1194 +    replication_marugoto(0);
1195 +    return(0);
1196 +}
1197 +
1198 +static int replication_push(void)
1199 +{
1200 +    int w;
1201 +
1202 +    while(rep_send->wbytes){
1203 +        w = write(rep_send->sfd, rep_send->wcurr, rep_send->wbytes);
1204 +        if(w == -1){
1205 +            if(errno == EAGAIN || errno == EINTR){
1206 +                fprintf(stderr,"replication: push EAGAIN or EINTR\n");
1207 +            }else{
1208 +                return(-1);
1209 +            }
1210 +        }else{
1211 +            rep_send->wbytes -= w;
1212 +            rep_send->wcurr  += w;
1213 +        }
1214 +    }
1215 +    rep_send->wcurr = rep_send->wbuf;
1216 +    return(0);
1217 +}
1218 +
1219 +static int replication_exit(void)
1220 +{
1221 +    return(replication_item(NULL));
1222 +}
1223 +
1224 +static int replication_item(Q_ITEM *q)
1225 +{
1226 +    pthread_mutex_lock(&replication_pipe_lock);
1227 +    if (!rep_send) {
1228 +        pthread_mutex_unlock(&replication_pipe_lock);
1229 +        return 0;
1230 +    }
1231 +    if(rep_send->wcurr + rep_send->wbytes + sizeof(q) > rep_send->wbuf + rep_send->wsize){
1232 +        fprintf(stderr,"replication: buffer over fllow\n");
1233 +        if(q){
1234 +            qi_free(q);
1235 +        }
1236 +        pthread_mutex_unlock(&replication_pipe_lock);
1237 +        replication_dispatch_close();
1238 +        return(-1);
1239 +    }
1240 +    memcpy(rep_send->wcurr + rep_send->wbytes, &q, sizeof(q));
1241 +    rep_send->wbytes += sizeof(q);
1242 +    if(replication_push()){
1243 +        fprintf(stderr, "replication: push error\n");
1244 +        if(q){
1245 +            qi_free(q);
1246 +        }
1247 +        pthread_mutex_unlock(&replication_pipe_lock);
1248 +        replication_dispatch_close();
1249 +        return(-1);
1250 +    }
1251 +    pthread_mutex_unlock(&replication_pipe_lock);
1252 +    return(0);
1253 +}
1254 +
1255 +int replication(enum CMD_TYPE type, R_CMD *cmd)
1256 +{
1257 +    Q_ITEM *q;
1258 +
1259 +    pthread_mutex_lock(&replication_pipe_lock);
1260 +    if (!rep_send) {
1261 +        pthread_mutex_unlock(&replication_pipe_lock);
1262 +        return 0;
1263 +    }
1264 +    pthread_mutex_unlock(&replication_pipe_lock);
1265 +
1266 +    if((q = qi_new(type, cmd, false))) {
1267 +        replication_item(q);
1268 +    }else{
1269 +        fprintf(stderr,"replication: can't create Q_ITEM\n");
1270 +        replication_dispatch_close();
1271 +        return(-1);
1272 +    }
1273 +    return(0);
1274 +}
1275 +#endif /* USE_REPLICATION */
1276 diff -urN --exclude *.m4 memcached-1.4.4/memcached.h repcached-2.2-1.4.4/memcached.h
1277 --- memcached-1.4.4/memcached.h Thu Nov 26 03:37:49 2009
1278 +++ repcached-2.2-1.4.4/memcached.h     Tue Feb  9 23:02:45 2010
1279 @@ -144,7 +144,13 @@
1280      conn_swallow,    /**< swallowing unnecessary bytes w/o storing */
1281      conn_closing,    /**< closing this connection */
1282      conn_mwrite,     /**< writing out many items sequentially */
1283 -    conn_max_state   /**< Max state value (used for assertion) */
1284 +#ifdef USE_REPLICATION
1285 +    conn_repconnect, /**< replication connecting to master */
1286 +    conn_rep_listen, /**< replication listening socket */
1287 +    conn_pipe_recv,  /**< replication command pipe recv */
1288 +    conn_pipe_send,  /**< replication command pipe send */
1289 +#endif /* USE_REPLICATION */
1290 +    conn_max_state,  /**< Max state value (used for assertion) */
1291  };
1292
1293  enum bin_substates {
1294 @@ -240,7 +246,9 @@
1295      uint64_t      get_hits;
1296      uint64_t      get_misses;
1297      uint64_t      evictions;
1298 +#if 0
1299      time_t        started;          /* when the process was started */
1300 +#endif
1301      bool          accepting_conns;  /* whether we are currently accepting */
1302      uint64_t      listen_disabled_num;
1303  };
1304 @@ -274,6 +282,11 @@
1305      int backlog;
1306      int item_size_max;        /* Maximum item size, and upper end for slabs */
1307      bool sasl;              /* SASL on/off */
1308 +#ifdef USE_REPLICATION
1309 +    struct in_addr rep_addr;    /* replication addr */
1310 +    int rep_port;               /* replication port */
1311 +    int rep_qmax;               /* replication QITEM max */
1312 +#endif /*USE_REPLICATION*/
1313  };
1314
1315  extern struct stats stats;
1316 @@ -286,6 +299,10 @@
1317  /* temp */
1318  #define ITEM_SLABBED 4
1319
1320 +#ifdef USE_REPLICATION
1321 +#define ITEM_REPDATA 128
1322 +#endif /*USE_REPLICATION*/
1323 +
1324  /**
1325   * Structure for storing items within memcached.
1326   */
1327 @@ -438,6 +455,10 @@
1328  #include "trace.h"
1329  #include "hash.h"
1330  #include "util.h"
1331 +
1332 +#ifdef USE_REPLICATION
1333 +#include "replication.h"
1334 +#endif /* USE_REPLICATION */
1335
1336  /*
1337   * Functions such as the libevent-related calls that need to do cross-thread
1338 diff -urN --exclude *.m4 memcached-1.4.4/memcached.spec repcached-2.2-1.4.4/memcached.spec
1339 diff -urN --exclude *.m4 memcached-1.4.4/replication.c repcached-2.2-1.4.4/replication.c
1340 --- memcached-1.4.4/replication.c       Thu Jan  1 03:00:00 1970
1341 +++ repcached-2.2-1.4.4/replication.c   Wed Feb 10 18:40:48 2010
1342 @@ -0,0 +1,355 @@
1343 +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
1344 +/*
1345 + *
1346 + */
1347 +#include "memcached.h"
1348 +#include "replication.h"
1349 +#include <stdlib.h>
1350 +#include <stdio.h>
1351 +#include <unistd.h>
1352 +#include <string.h>
1353 +#include <errno.h>
1354 +
1355 +static Q_ITEM *q_freelist  = NULL;
1356 +static int     q_itemcount = 0;
1357 +static pthread_mutex_t replication_queue_lock = PTHREAD_MUTEX_INITIALIZER;
1358 +
1359 +int get_qi_count(void)
1360 +{
1361 +    int c;
1362 +    pthread_mutex_lock(&replication_queue_lock);
1363 +    c = q_itemcount;
1364 +    pthread_mutex_unlock(&replication_queue_lock);
1365 +    return(c);
1366 +}
1367 +
1368 +Q_ITEM *qi_new(enum CMD_TYPE type, R_CMD *cmd, bool reuse)
1369 +{
1370 +    Q_ITEM     *q      = NULL;
1371 +    char       *key    = NULL;
1372 +    uint32_t    keylen = 0;
1373 +    rel_time_t  time   = 0;
1374 +
1375 +    pthread_mutex_lock(&replication_queue_lock);
1376 +    if(q_freelist){
1377 +        q = q_freelist;
1378 +        q_freelist = q->next;
1379 +    }
1380 +
1381 +    if(NULL == q){
1382 +        if(reuse) {
1383 +            pthread_mutex_unlock(&replication_queue_lock);
1384 +            return(NULL);
1385 +        }
1386 +        if(q_itemcount >= settings.rep_qmax) {
1387 +            pthread_mutex_unlock(&replication_queue_lock);
1388 +            return(NULL);
1389 +        }
1390 +        q = malloc(sizeof(Q_ITEM));
1391 +        if (NULL == q){
1392 +            fprintf(stderr,"replication: qi_new out of memory\n");
1393 +            pthread_mutex_unlock(&replication_queue_lock);
1394 +            return(NULL);
1395 +        }
1396 +        q_itemcount++;
1397 +        if (settings.verbose > 2)
1398 +            fprintf(stderr,"replication: alloc c=%d\n", q_itemcount);
1399 +    }
1400 +
1401 +    pthread_mutex_unlock(&replication_queue_lock);
1402 +
1403 +    switch (type) {
1404 +    case REPLICATION_REP:
1405 +    case REPLICATION_DEL:
1406 +        key    = cmd->key;
1407 +        keylen = cmd->keylen;
1408 +        break;
1409 +    case REPLICATION_FLUSH_ALL:
1410 +        break;
1411 +    case REPLICATION_DEFER_FLUSH_ALL:
1412 +        time   = cmd->time;
1413 +        break;
1414 +    case REPLICATION_MARUGOTO_END:
1415 +        break;
1416 +    default:
1417 +        fprintf(stderr,"replication: got unknown command: %d\n", type);
1418 +        return(NULL);
1419 +    }
1420 +
1421 +    q->key  = NULL;
1422 +    q->type = type;
1423 +    q->time = time;
1424 +    q->next = NULL;
1425 +    if (keylen) {
1426 +        q->key = malloc(keylen + 1);
1427 +        if(NULL == q->key){
1428 +            qi_free(q);
1429 +            q = NULL;
1430 +        }else{
1431 +            memcpy(q->key, key, keylen);
1432 +            *(q->key + keylen) = 0;
1433 +        }
1434 +    }
1435 +
1436 +    return(q);
1437 +}
1438 +
1439 +void qi_free(Q_ITEM *q)
1440 +{
1441 +    if(q){
1442 +        if(q->key){
1443 +            free(q->key);
1444 +            q->key = NULL;
1445 +        }
1446 +        pthread_mutex_lock(&replication_queue_lock);
1447 +        q->next = q_freelist;
1448 +        q_freelist = q;
1449 +        pthread_mutex_unlock(&replication_queue_lock);
1450 +    }
1451 +}
1452 +
1453 +int qi_free_list()
1454 +{
1455 +    int     c = 0;
1456 +    Q_ITEM *q = NULL;
1457 +
1458 +    pthread_mutex_lock(&replication_queue_lock);
1459 +    while((q = q_freelist)){
1460 +        q_itemcount--;
1461 +        c++;
1462 +        q_freelist = q->next;
1463 +        free(q);
1464 +    }
1465 +    pthread_mutex_unlock(&replication_queue_lock);
1466 +    return(c);
1467 +}
1468 +
1469 +static int replication_get_num(char *p, int n)
1470 +{
1471 +    int  l;
1472 +    char b[64];
1473 +    if(p)
1474 +        l = sprintf(p, "%u", n);
1475 +    else
1476 +        l = sprintf(b, "%u", n);
1477 +    return(l);
1478 +}
1479 +
1480 +int replication_call_rep(char *key, size_t keylen)
1481 +{
1482 +    R_CMD r;
1483 +    r.key    = key;
1484 +    r.keylen = keylen;
1485 +    return(replication(REPLICATION_REP, &r));
1486 +}
1487 +
1488 +int replication_call_del(char *key, size_t keylen)
1489 +{
1490 +    R_CMD r;
1491 +    r.key    = key;
1492 +    r.keylen = keylen;
1493 +    return(replication(REPLICATION_DEL, &r));
1494 +}
1495 +
1496 +int replication_call_flush_all()
1497 +{
1498 +    R_CMD r;
1499 +    r.key = NULL;
1500 +    return(replication(REPLICATION_FLUSH_ALL, &r));
1501 +}
1502 +
1503 +int replication_call_defer_flush_all(const rel_time_t time)
1504 +{
1505 +    R_CMD r;
1506 +    r.key  = NULL;
1507 +    r.time = time;
1508 +    return(replication(REPLICATION_DEFER_FLUSH_ALL, &r));
1509 +}
1510 +
1511 +int replication_call_marugoto_end()
1512 +{
1513 +    R_CMD r;
1514 +    r.key = NULL;
1515 +    return(replication(REPLICATION_MARUGOTO_END, &r));
1516 +}
1517 +
1518 +static int replication_alloc(conn *c, int s)
1519 +{
1520 +    char *p;
1521 +    s += c->wbytes;
1522 +    if(c->wsize < s){
1523 +        while(c->wsize < s)
1524 +            c->wsize += 4096;
1525 +        if((p = malloc(c->wsize))){
1526 +            memcpy(p, c->wbuf, c->wbytes);
1527 +            free(c->wbuf);
1528 +            c->wbuf = p;
1529 +        }else{
1530 +            return(-1);
1531 +        }
1532 +    }
1533 +    return(0);
1534 +}
1535 +
1536 +static int replication_del(conn *c, char *k)
1537 +{
1538 +    int   l = 0;
1539 +    char *s = "delete ";
1540 +    char *n = "\r\n";
1541 +    char *p = NULL;
1542 +
1543 +    l += strlen(s);
1544 +    l += strlen(k);
1545 +    l += strlen(n);
1546 +    if(replication_alloc(c,l) == -1){
1547 +        fprintf(stderr, "replication: del malloc error\n");
1548 +        return(-1);
1549 +    }
1550 +    p = c->wbuf + c->wbytes;
1551 +    memcpy(p, s, strlen(s));
1552 +    p += strlen(s);
1553 +    memcpy(p, k, strlen(k));
1554 +    p += strlen(k);
1555 +    memcpy(p, n, strlen(n));
1556 +    p += strlen(n);
1557 +    c->wbytes = p - c->wbuf;
1558 +    c->wcurr  = c->wbuf;
1559 +    return(0);
1560 +}
1561 +
1562 +static int replication_rep(conn *c, item *it)
1563 +{
1564 +    int exp = 0;
1565 +    int len = 0;
1566 +    char *s = "rep ";
1567 +    char *n = "\r\n";
1568 +    char *p = NULL;
1569 +    char flag[40];
1570 +
1571 +    if(it->exptime)
1572 +        exp = it->exptime + process_started;
1573 +    flag[0]=0;
1574 +    if((p=ITEM_suffix(it))){
1575 +        int i;
1576 +        memcpy(flag, p, it->nsuffix - 2);
1577 +        flag[it->nsuffix - 2] = 0;
1578 +        for(i=0;i<strlen(flag);i++){
1579 +            if(flag[i] > ' ')
1580 +                break;
1581 +        }
1582 +        memmove(flag,&flag[i],strlen(flag)-i);
1583 +        for(p=flag;*p>' ';p++);
1584 +        *p=0;
1585 +    }
1586 +    len += strlen(s);
1587 +    len += it->nkey;
1588 +    len += 1;
1589 +    len += strlen(flag);
1590 +    len += 1;
1591 +    len += replication_get_num(NULL, exp);
1592 +    len += 1;
1593 +    len += replication_get_num(NULL, it->nbytes - 2);
1594 +    len += 1;
1595 +    len += replication_get_num(NULL, ITEM_get_cas(it));
1596 +    len += strlen(n);
1597 +    len += it->nbytes;
1598 +    len += strlen(n);
1599 +    if(replication_alloc(c,len) == -1){
1600 +        fprintf(stderr, "replication: rep malloc error\n");
1601 +        return(-1);
1602 +    }
1603 +    p = c->wbuf + c->wbytes;
1604 +    memcpy(p, s, strlen(s));
1605 +    p += strlen(s);
1606 +    memcpy(p, ITEM_key(it), it->nkey);
1607 +    p += it->nkey;
1608 +    *(p++) = ' ';
1609 +    memcpy(p, flag, strlen(flag));
1610 +    p += strlen(flag);
1611 +    *(p++) = ' ';
1612 +    p += replication_get_num(p, exp);
1613 +    *(p++) = ' ';
1614 +    p += replication_get_num(p, it->nbytes - 2);
1615 +    *(p++) = ' ';
1616 +    p += replication_get_num(p, ITEM_get_cas(it));
1617 +    memcpy(p, n, strlen(n));
1618 +    p += strlen(n);
1619 +    memcpy(p, ITEM_data(it), it->nbytes);
1620 +    p += it->nbytes;
1621 +    c->wbytes = p - c->wbuf;
1622 +    c->wcurr  = c->wbuf;
1623 +    return(0);
1624 +}
1625 +
1626 +static int replication_flush_all(conn *c, rel_time_t exp)
1627 +{
1628 +    char *s = "flush_all ";
1629 +    char *n = "\r\n";
1630 +    char *p = NULL;
1631 +
1632 +    int l = strlen(s) + strlen(n);
1633 +    if (exp > 0)
1634 +        l += replication_get_num(NULL, exp);
1635 +    if(replication_alloc(c,l) == -1){
1636 +        fprintf(stderr, "replication: flush_all malloc error\n");
1637 +        return(-1);
1638 +    }
1639 +    p = c->wbuf + c->wbytes;
1640 +    memcpy(p, s, strlen(s));
1641 +    p += strlen(s);
1642 +    if (exp > 0)
1643 +        p += replication_get_num(p, exp);
1644 +    memcpy(p, n, strlen(n));
1645 +    p += strlen(n);
1646 +    c->wbytes = p - c->wbuf;
1647 +    c->wcurr  = c->wbuf;
1648 +    return(0);
1649 +}
1650 +
1651 +static int replication_marugoto_end(conn *c)
1652 +{
1653 +    char *s = "marugoto_end";
1654 +    char *n = "\r\n";
1655 +    char *p = NULL;
1656 +
1657 +    int l = strlen(s) + strlen(n);
1658 +    if(replication_alloc(c,l) == -1){
1659 +        fprintf(stderr, "replication: marugoto_end malloc error\n");
1660 +        return(-1);
1661 +    }
1662 +    p = c->wbuf + c->wbytes;
1663 +    memcpy(p, s, strlen(s));
1664 +    p += strlen(s);
1665 +    memcpy(p, n, strlen(n));
1666 +    p += strlen(n);
1667 +    c->wbytes = p - c->wbuf;
1668 +    c->wcurr  = c->wbuf;
1669 +    return(0);
1670 +}
1671 +
1672 +int replication_cmd(conn *c, Q_ITEM *q)
1673 +{
1674 +    item *it;
1675 +    int r;
1676 +
1677 +    switch (q->type) {
1678 +    case REPLICATION_REP:
1679 +        it = item_get(q->key, strlen(q->key));
1680 +        if (!it)
1681 +            return(replication_del(c, q->key));
1682 +        r = replication_rep(c, it);
1683 +        item_remove(it);
1684 +        return r;
1685 +    case REPLICATION_DEL:
1686 +        return(replication_del(c, q->key));
1687 +    case REPLICATION_FLUSH_ALL:
1688 +        return(replication_flush_all(c, 0));
1689 +    case REPLICATION_DEFER_FLUSH_ALL:
1690 +        return(replication_flush_all(c, q->time));
1691 +    case REPLICATION_MARUGOTO_END:
1692 +        return(replication_marugoto_end(c));
1693 +    default:
1694 +        fprintf(stderr,"replication: got unknown command:%d\n", q->type);
1695 +        return(0);
1696 +    }
1697 +}
1698 diff -urN --exclude *.m4 memcached-1.4.4/replication.h repcached-2.2-1.4.4/replication.h
1699 --- memcached-1.4.4/replication.h       Thu Jan  1 03:00:00 1970
1700 +++ repcached-2.2-1.4.4/replication.h   Wed Feb 10 18:40:31 2010
1701 @@ -0,0 +1,42 @@
1702 +#ifndef MEMCACHED_REPLICATION_H
1703 +#define MEMCACHED_REPLICATION_H
1704 +#define REPCACHED_VERSION "2.2"
1705 +#include <netdb.h>
1706 +
1707 +enum CMD_TYPE {
1708 +  REPLICATION_REP,
1709 +  REPLICATION_DEL,
1710 +  REPLICATION_FLUSH_ALL,
1711 +  REPLICATION_DEFER_FLUSH_ALL,
1712 +  REPLICATION_MARUGOTO_END,
1713 +};
1714 +
1715 +typedef struct queue_item_t Q_ITEM;
1716 +struct queue_item_t {
1717 +  enum CMD_TYPE  type;
1718 +  char          *key;
1719 +  rel_time_t     time;
1720 +  Q_ITEM        *next;
1721 +};
1722 +
1723 +typedef struct replication_cmd_t R_CMD;
1724 +struct replication_cmd_t {
1725 +  char       *key;
1726 +  int         keylen;
1727 +  rel_time_t  time;
1728 +};
1729 +
1730 +Q_ITEM *qi_new(enum CMD_TYPE type, R_CMD *cmd, bool);
1731 +void    qi_free(Q_ITEM *);
1732 +int     qi_free_list(void);
1733 +int     replication_cmd(conn *, Q_ITEM *);
1734 +int     get_qi_count(void);
1735 +
1736 +int replication_call_rep(char *key, size_t keylen);
1737 +int replication_call_del(char *key, size_t keylen);
1738 +int replication_call_flush_all(void);
1739 +int replication_call_defer_flush_all(const rel_time_t time);
1740 +int replication_call_marugoto_end(void);
1741 +int replication(enum CMD_TYPE type, R_CMD *cmd);
1742 +
1743 +#endif
1744 diff -urN --exclude *.m4 memcached-1.4.4/t/binary.t repcached-2.2-1.4.4/t/binary.t
1745 --- memcached-1.4.4/t/binary.t  Fri Nov 27 08:05:16 2009
1746 +++ repcached-2.2-1.4.4/t/binary.t      Wed Feb 10 17:04:01 2010
1747 @@ -2,10 +2,12 @@
1748
1749  use strict;
1750  use warnings;
1751 -use Test::More tests => 3349;
1752 +use Test::More;
1753  use FindBin qw($Bin);
1754  use lib "$Bin/lib";
1755  use MemcachedTest;
1756 +
1757 +Test::More::plan(tests => support_replication() ? 3385 : 3349);
1758
1759  my $server = new_memcached();
1760  ok($server, "started the server");
1761 diff -urN --exclude *.m4 memcached-1.4.4/t/issue_67.t repcached-2.2-1.4.4/t/issue_67.t
1762 --- memcached-1.4.4/t/issue_67.t        Sun Nov  1 01:44:09 2009
1763 +++ repcached-2.2-1.4.4/t/issue_67.t    Wed Feb 10 17:50:12 2010
1764 @@ -41,6 +41,10 @@
1765      my $exe = "$builddir/memcached-debug";
1766      croak("memcached binary doesn't exist.  Haven't run 'make' ?\n") unless -e $exe;
1767
1768 +    if (support_replication()) {
1769 +        $args .= ' -X 0';
1770 +    }
1771 +
1772      my $childpid = fork();
1773
1774      my $cmd = "$builddir/timedrun 10 $exe $args";
1775 diff -urN --exclude *.m4 memcached-1.4.4/t/lib/MemcachedTest.pm repcached-2.2-1.4.4/t/lib/MemcachedTest.pm
1776 --- memcached-1.4.4/t/lib/MemcachedTest.pm      Fri Oct 30 04:24:52 2009
1777 +++ repcached-2.2-1.4.4/t/lib/MemcachedTest.pm  Wed Feb 10 17:53:34 2010
1778 @@ -13,7 +13,8 @@
1779
1780
1781  @EXPORT = qw(new_memcached sleep mem_get_is mem_gets mem_gets_is mem_stats
1782 -             supports_sasl free_port);
1783 +             supports_sasl free_port support_replication memcached_version
1784 +             version2num);
1785
1786  sub sleep {
1787      my $n = shift;
1788 @@ -148,6 +149,23 @@
1789      return 0;
1790  }
1791
1792 +sub support_replication {
1793 +    my $output = `$builddir/memcached-debug -h`;
1794 +    return 1 if $output =~ /^-x <ip_addr>/m;
1795 +    return 0;
1796 +}
1797 +
1798 +sub memcached_version {
1799 +    my $output = `$builddir/memcached-debug -h`;
1800 +    return $1 if $output =~ /^memcached (\d[\d\.]+)/;
1801 +    return 0;
1802 +}
1803 +
1804 +sub version2num {
1805 +    my($major,$minor,$pl) = ($_[0] =~ /^(\d+)\.(\d+)\.(\d+)$/);
1806 +    return $major*100**2 + $minor*100 + $pl
1807 +}
1808 +
1809  sub new_memcached {
1810      my ($args, $passed_port) = @_;
1811      my $port = $passed_port || free_port();
1812 @@ -171,6 +189,9 @@
1813      }
1814      if ($< == 0) {
1815          $args .= " -u root";
1816 +    }
1817 +    if (support_replication() && $args !~ m/-X/) {
1818 +        $args .= ' -X 0';
1819      }
1820
1821      my $childpid = fork();
1822 diff -urN --exclude *.m4 memcached-1.4.4/t/stats.t repcached-2.2-1.4.4/t/stats.t
1823 --- memcached-1.4.4/t/stats.t   Thu Nov 26 03:37:49 2009
1824 +++ repcached-2.2-1.4.4/t/stats.t       Tue Feb  9 23:13:26 2010
1825 @@ -56,7 +56,11 @@
1826  my $stats = mem_stats($sock);
1827
1828  # Test number of keys
1829 -is(scalar(keys(%$stats)), 37, "37 stats values");
1830 +if (! support_replication()) {
1831 +    is(scalar(keys(%$stats)), 37, "37 stats values");
1832 +} else {
1833 +    is(scalar(keys(%$stats)), 40, "40 stats values");
1834 +}
1835
1836  # Test initial state
1837  foreach my $key (qw(curr_items total_items bytes cmd_get cmd_set get_hits evictions get_misses
1838 diff -urN --exclude *.m4 memcached-1.4.4/testapp.c repcached-2.2-1.4.4/testapp.c
1839 --- memcached-1.4.4/testapp.c   Wed Nov 25 03:40:29 2009
1840 +++ repcached-2.2-1.4.4/testapp.c       Wed Feb 10 17:52:05 2010
1841 @@ -300,6 +300,10 @@
1842          argv[arg++] = "-1";
1843          argv[arg++] = "-U";
1844          argv[arg++] = "0";
1845 +#ifdef USE_REPLICATION
1846 +        argv[arg++] = "-X";
1847 +        argv[arg++] = "0";
1848 +#endif
1849          /* Handle rpmbuild and the like doing this as root */
1850          if (getuid() == 0) {
1851              argv[arg++] = "-u";