Harden against connection and server failures

Gemfile

@@ -1,4 +1,4 @@
-source :rubygems
+source 'https://rubygems.org'
 
 gemspec
 

Rakefile

@@ -94,6 +94,7 @@ end
 
 task :build do
   cd 'ext' do
+    ENV['ZK_DEV'] = 'true'
     sh "rake"
   end
 end

ext/Rakefile

@@ -1,11 +1,5 @@
 require 'rbconfig'
 
-LIB_ZK_SO = 'lib/libzookeeper_mt_gem.la'
-
-TARBALL = FileList['zkc-*.tar.gz'].first
-
-raise "Where is the zkc tarball!?" unless TARBALL
-
 namespace :zkrb do
   task :clean do
     if File.exists?('Makefile')
@@ -26,21 +20,6 @@ task :clobber => 'zkrb:clobber'
 
 GENERATE_GVL_CODE_RB = 'generate_gvl_code.rb'
 
-# file 'c' do
-#   if tarball = Dir['zkc-*.tar.gz'].first
-#     sh "tar -zxf #{tarball}"
-#   else
-#     raise "couldn't find the tarball! wtf?!"
-#   end
-# end
-
-file 'c' => TARBALL do
-  sh "tar -zxf #{TARBALL}"
-  sh "patch -p0 < patch-zookeeper"
-end
-
-file GENERATE_GVL_CODE_RB => 'c'
-
 file 'zkrb_wrapper.c' => GENERATE_GVL_CODE_RB do
   sh "ruby generate_gvl_code.rb code"
 end

ext/c_zookeeper.rb

@@ -17,7 +17,7 @@ class CZookeeper
   include Exceptions
   include Logger
 
-  DEFAULT_SESSION_TIMEOUT_MSEC = 10000
+  DEFAULT_RECEIVE_TIMEOUT_MSEC = 10000
 
   class GotNilEventException < StandardError; end
 
@@ -65,7 +65,7 @@ def initialize(host, event_queue, opts={})
     # the actual C data is stashed in this ivar. never *ever* touch this
     @_data = nil
 
-    @_session_timeout_msec = DEFAULT_SESSION_TIMEOUT_MSEC
+    @_receive_timeout_msec = opts[:receive_timeout_msec] || DEFAULT_RECEIVE_TIMEOUT_MSEC
 
     @mutex = Monitor.new
 
@@ -120,6 +120,14 @@ def associating?
     state == ZOO_ASSOCIATING_STATE
   end
 
+  def unhealthy?
+    @_closed || @_shutting_down || is_unrecoverable
+  end
+
+  def healthy?
+    !unhealthy?
+  end
+
   def close
     return if closed?
 
@@ -183,11 +191,11 @@ def wait_until_connected(timeout=10)
       while true 
         if timeout
           now = Time.now
-          break if (@state == ZOO_CONNECTED_STATE) || @_shutting_down || @_closed || (now > time_to_stop)
+          break if (@state == ZOO_CONNECTED_STATE) || unhealthy? || (now > time_to_stop)
           delay = time_to_stop.to_f - now.to_f
           @state_cond.wait(delay)
         else
-          break if (@state == ZOO_CONNECTED_STATE) || @_shutting_down || @_closed
+          break if (@state == ZOO_CONNECTED_STATE) || unhealthy?
           @state_cond.wait
         end
       end
@@ -201,7 +209,7 @@ def wait_until_connected(timeout=10)
     # blocks the caller until result has returned
     def submit_and_block(meth, *args)
       @mutex.synchronize do
-        raise Exceptions::NotConnected if @_shutting_down
+        raise Exceptions::NotConnected if unhealthy?
       end
 
       cnt = Continuation.new(meth, *args)
@@ -257,8 +265,11 @@ def event_thread_body
       event_thread_await_running
 
       # this is the main loop
-      until (@_shutting_down or @_closed or is_unrecoverable)
-        submit_pending_calls    if @reg.anything_to_do?
+      while healthy?
+        if @reg.anything_to_do? && connected?
+          submit_pending_calls
+        end
+
         zkrb_iterate_event_loop 
         iterate_event_delivery
       end

ext/event_lib.c

@@ -21,7 +21,7 @@ slyphon@gmail.com
 */
 
 #include "ruby.h"
-#include "c-client-src/zookeeper.h"
+#include "zookeeper/zookeeper.h"
 #include <errno.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -321,47 +321,47 @@ VALUE zkrb_event_to_ruby(zkrb_event_t *event) {
 
   switch (event->type) {
     case ZKRB_DATA: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_DATA\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_DATA");
       struct zkrb_data_completion *data_ctx = event->completion.data_completion;
       if (ZKRBDebugging) zkrb_print_stat(data_ctx->stat);
       rb_hash_aset(hash, GET_SYM("data"), data_ctx->data ? rb_str_new(data_ctx->data, data_ctx->data_len) : Qnil);
       rb_hash_aset(hash, GET_SYM("stat"), data_ctx->stat ? zkrb_stat_to_rarray(data_ctx->stat) : Qnil);
       break;
     }
     case ZKRB_STAT: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_STAT\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_STAT");
       struct zkrb_stat_completion *stat_ctx = event->completion.stat_completion;
       rb_hash_aset(hash, GET_SYM("stat"), stat_ctx->stat ? zkrb_stat_to_rarray(stat_ctx->stat) : Qnil);
       break;
     }
     case ZKRB_STRING: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_STRING\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_STRING");
       struct zkrb_string_completion *string_ctx = event->completion.string_completion;
       rb_hash_aset(hash, GET_SYM("string"), string_ctx->value ? rb_str_new2(string_ctx->value) : Qnil);
       break;
     }
     case ZKRB_STRINGS: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_STRINGS\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_STRINGS");
       struct zkrb_strings_completion *strings_ctx = event->completion.strings_completion;
       rb_hash_aset(hash, GET_SYM("strings"), strings_ctx->values ? zkrb_string_vector_to_ruby(strings_ctx->values) : Qnil);
       break;
     }
     case ZKRB_STRINGS_STAT: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_STRINGS_STAT\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_STRINGS_STAT");
       struct zkrb_strings_stat_completion *strings_stat_ctx = event->completion.strings_stat_completion;
       rb_hash_aset(hash, GET_SYM("strings"), strings_stat_ctx->values ? zkrb_string_vector_to_ruby(strings_stat_ctx->values) : Qnil);
       rb_hash_aset(hash, GET_SYM("stat"), strings_stat_ctx->stat ? zkrb_stat_to_rarray(strings_stat_ctx->stat) : Qnil);
       break;
     }
     case ZKRB_ACL: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_ACL\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_ACL");
       struct zkrb_acl_completion *acl_ctx = event->completion.acl_completion;
       rb_hash_aset(hash, GET_SYM("acl"), acl_ctx->acl ? zkrb_acl_vector_to_ruby(acl_ctx->acl) : Qnil);
       rb_hash_aset(hash, GET_SYM("stat"), acl_ctx->stat ? zkrb_stat_to_rarray(acl_ctx->stat) : Qnil);
       break;
     }
     case ZKRB_WATCHER: {
-      zkrb_debug("zkrb_event_to_ruby ZKRB_WATCHER\n");
+      zkrb_debug("zkrb_event_to_ruby ZKRB_WATCHER");
       struct zkrb_watcher_completion *watcher_ctx = event->completion.watcher_completion;
       rb_hash_aset(hash, GET_SYM("type"), INT2FIX(watcher_ctx->type));
       rb_hash_aset(hash, GET_SYM("state"), INT2FIX(watcher_ctx->state));
@@ -377,8 +377,8 @@ VALUE zkrb_event_to_ruby(zkrb_event_t *event) {
 }
 
 void zkrb_print_stat(const struct Stat *s) {
-  fprintf(stderr,  "stat {\n");
   if (s != NULL) {
+    fprintf(stderr,  "stat {\n");
     fprintf(stderr,  "\t          czxid: %"PRId64"\n", s->czxid);   // PRId64 defined in inttypes.h
     fprintf(stderr,  "\t          mzxid: %"PRId64"\n", s->mzxid);
     fprintf(stderr,  "\t          ctime: %"PRId64"\n", s->ctime);
@@ -390,10 +390,10 @@ void zkrb_print_stat(const struct Stat *s) {
     fprintf(stderr,  "\t     dataLength: %d\n",        s->dataLength);
     fprintf(stderr,  "\t    numChildren: %d\n",        s->numChildren);
     fprintf(stderr,  "\t          pzxid: %"PRId64"\n", s->pzxid);
+    fprintf(stderr,  "}\n");
   } else {
-    fprintf(stderr, "\tNULL\n");
+    fprintf(stderr,  "stat { NULL }\n");
   }
-  fprintf(stderr,  "}\n");
 }
 
 zkrb_calling_context *zkrb_calling_context_alloc(int64_t req_id, zkrb_queue_t *queue) {
@@ -436,7 +436,7 @@ void zkrb_state_callback(
     zhandle_t *zh, int type, int state, const char *path, void *calling_ctx) {
 
   zkrb_debug("ZOOKEEPER_C_STATE WATCHER "
-                    "type = %d, state = %d, path = %p, value = %s\n",
+                    "type = %d, state = %d, path = %p, value = %s",
       type, state, (void *) path, path ? path : "NULL");
 
   /* save callback context */
@@ -466,7 +466,7 @@ void zkrb_data_callback(
     int rc, const char *value, int value_len, const struct Stat *stat, const void *calling_ctx) {
 
   zkrb_debug("ZOOKEEPER_C_DATA WATCHER "
-                "rc = %d (%s), value = %s, len = %d\n",
+                "rc = %d (%s), value = %s, len = %d",
                 rc, zerror(rc), value ? value : "NULL", value_len);
 
   /* copy data completion */
@@ -494,7 +494,7 @@ void zkrb_data_callback(
 void zkrb_stat_callback(
     int rc, const struct Stat *stat, const void *calling_ctx) {
   zkrb_debug("ZOOKEEPER_C_STAT WATCHER "
-                    "rc = %d (%s)\n", rc, zerror(rc));
+                    "rc = %d (%s)", rc, zerror(rc));
 
   struct zkrb_stat_completion *sc = zk_malloc(sizeof(struct zkrb_stat_completion));
   sc->stat = NULL;
@@ -512,7 +512,7 @@ void zkrb_string_callback(
     int rc, const char *string, const void *calling_ctx) {
 
   zkrb_debug("ZOOKEEPER_C_STRING WATCHER "
-                    "rc = %d (%s)\n", rc, zerror(rc));
+                    "rc = %d (%s)", rc, zerror(rc));
 
   struct zkrb_string_completion *sc = zk_malloc(sizeof(struct zkrb_string_completion));
   sc->value = NULL;
@@ -530,7 +530,7 @@ void zkrb_string_callback(
 void zkrb_strings_callback(
     int rc, const struct String_vector *strings, const void *calling_ctx) {
   zkrb_debug("ZOOKEEPER_C_STRINGS WATCHER "
-                    "rc = %d (%s), calling_ctx = %p\n", rc, zerror(rc), calling_ctx);
+                    "rc = %d (%s), calling_ctx = %p", rc, zerror(rc), calling_ctx);
 
   /* copy string vector */
   struct zkrb_strings_completion *sc = zk_malloc(sizeof(struct zkrb_strings_completion));
@@ -547,7 +547,7 @@ void zkrb_strings_callback(
 void zkrb_strings_stat_callback(
     int rc, const struct String_vector *strings, const struct Stat *stat, const void *calling_ctx) {
   zkrb_debug("ZOOKEEPER_C_STRINGS_STAT WATCHER "
-                    "rc = %d (%s), calling_ctx = %p\n", rc, zerror(rc), calling_ctx);
+                    "rc = %d (%s), calling_ctx = %p", rc, zerror(rc), calling_ctx);
 
   struct zkrb_strings_stat_completion *sc = zk_malloc(sizeof(struct zkrb_strings_stat_completion));
   sc->stat = NULL;
@@ -565,7 +565,7 @@ void zkrb_strings_stat_callback(
 
 void zkrb_void_callback(int rc, const void *calling_ctx) {
   zkrb_debug("ZOOKEEPER_C_VOID WATCHER "
-                    "rc = %d (%s)\n", rc, zerror(rc));
+                    "rc = %d (%s)", rc, zerror(rc));
 
   ZKH_SETUP_EVENT(queue, event);
   event->rc = rc;
@@ -577,7 +577,7 @@ void zkrb_void_callback(int rc, const void *calling_ctx) {
 
 void zkrb_acl_callback(
     int rc, struct ACL_vector *acls, struct Stat *stat, const void *calling_ctx) {
-  zkrb_debug("ZOOKEEPER_C_ACL WATCHER rc = %d (%s)\n", rc, zerror(rc));
+  zkrb_debug("ZOOKEEPER_C_ACL WATCHER rc = %d (%s)", rc, zerror(rc));
 
   struct zkrb_acl_completion *ac = zk_malloc(sizeof(struct zkrb_acl_completion));
   ac->acl = NULL;

ext/event_lib.h

@@ -2,7 +2,7 @@
 #define ZKRB_EVENT_LIB_H
 
 #include "ruby.h"
-#include "c-client-src/zookeeper.h"
+#include "zookeeper/zookeeper.h"
 #include <errno.h>
 #include <stdio.h>
 #include <stdlib.h>

ext/extconf.rb

@@ -3,10 +3,11 @@
 require 'fileutils'
 
 HERE = File.expand_path(File.dirname(__FILE__))
-BUNDLE = Dir.glob("zkc-*.tar.gz").first
-ZKPATCH = "patch-zookeeper"
+BUNDLE = Dir.glob("zkc-*.tar.gz").sort.last
+ZKC_VERSION = BUNDLE[/(zkc-.*?)\.tar.gz$/, 1]
+PATCHES = Dir.glob("patches/#{ZKC_VERSION}*.patch")
 
-BUNDLE_PATH = File.join(HERE, 'c')
+BUNDLE_PATH = File.join(HERE, ZKC_VERSION, 'c')
 
 $EXTRA_CONF = ''
 
@@ -58,8 +59,10 @@ def safe_sh(cmd)
     puts "Building zkc."
 
     unless File.exists?('c')
-      puts(cmd = "tar xzf #{BUNDLE} 2>&1 && patch -p0 < #{ZKPATCH} 2>&1")
-      raise "'#{cmd}' failed" unless system(cmd)
+      safe_sh "tar xzf #{BUNDLE} 2>&1"
+      PATCHES.each do |patch|
+        safe_sh "patch -p0 < #{patch} 2>&1"
+      end
     end
 
     # clean up stupid apple rsrc fork bullshit

ext/generate_gvl_code.rb

@@ -233,7 +233,7 @@ def render_header_file(code)
 #endif
 
 #include "ruby.h"
-#include "c-client-src/zookeeper.h"
+#include "zookeeper/zookeeper.h"
 #include "zkrb_wrapper_compat.h"
 #include "dbg.h"
 

ext/patches/zkc-3.4.5-logging.patch

@@ -0,0 +1,41 @@
+diff -ur zkc-3.4.5-orig/c/src/zookeeper.c zkc-3.4.5/c/src/zookeeper.c
+--- zkc-3.4.5-orig/c/src/zookeeper.c	2012-09-30 10:53:32.000000000 -0700
++++ zkc-3.4.5/c/src/zookeeper.c	2013-09-07 21:25:24.000000000 -0700
+@@ -1650,14 +1650,16 @@
+         // a PING
+         if (zh->state==ZOO_CONNECTED_STATE) {
+             send_to = zh->recv_timeout/3 - idle_send;
+-            if (send_to <= 0 && zh->sent_requests.head==0) {
+-//                LOG_DEBUG(("Sending PING to %s (exceeded idle by %dms)",
+-//                                format_current_endpoint_info(zh),-send_to));
+-                int rc=send_ping(zh);
+-                if (rc < 0){
+-                    LOG_ERROR(("failed to send PING request (zk retcode=%d)",rc));
+-                    return api_epilog(zh,rc);
+-                }
++            if (send_to <= 0) {
++                if (zh->sent_requests.head==0) {
++                    LOG_DEBUG(("Sending PING to %s (exceeded idle by %dms)",
++                                   format_current_endpoint_info(zh),-send_to));
++                    int rc=send_ping(zh);
++                    if (rc < 0){
++                        LOG_ERROR(("failed to send PING request (zk retcode=%d)",rc));
++                        return api_epilog(zh,rc);
++                    }
++                }
+                 send_to = zh->recv_timeout/3;
+             }
+         }
+@@ -1669,6 +1671,12 @@
+             zh->next_deadline.tv_sec += zh->next_deadline.tv_usec / 1000000;
+             zh->next_deadline.tv_usec = zh->next_deadline.tv_usec % 1000000;
+         }
++        
++        if (tv->tv_sec == 0 && tv->tv_usec == 0) {
++          LOG_DEBUG(("Returning a 0.0 timeval: state=%d idle_recv=%d idle_send=%d recv_to=%d send_to=%d  send_requests=%s",
++            zh->state, idle_recv, idle_send, recv_to, send_to, zh->sent_requests.head==0 ? "false" : "true"));
++        }
++        
+         *interest = ZOOKEEPER_READ;
+         /* we are interested in a write if we are connected and have something
+          * to send, or we are waiting for a connect to finish. */