Merge pull request #50 from zk-ruby/zoomonkey

Harden against connection and server failures
zk-ruby · Sep 9, 2013 · ebdfa7899fbcadadca8faa015479c22edc6a81f8 · ebdfa78
2 parents 00afd18 + f4dfdc5
commit ebdfa7899fbcadadca8faa015479c22edc6a81f8
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,3 +1,19 @@
+
+v1.4.6
+
+  * Fix two issues with dealing with unresponsive zookeeper servers
+    that would prevent the client from properly recovering the session
+    (including one in the Apache ZooKeeper C library — see ZOOKEEPER-1756
+    for more details)
+  * Reduce the chances of seeing Zookeeper::Exceptions::NotConnected
+    exceptions by only submitting new commands if we are connected
+  * Prevent commands from being queued to be sent to the server if the
+    session has expired which will solve most cases involving the
+    Zookeeper::Exceptions::ContinuationTimeoutError exception
+  * Upgrade the Apache ZooKeeper C library to 3.4.5 (client is backward
+    compatible with 3.3 servers). The Java library has not been updated.
+  * Cleanup complaints from compiler for uninitialized variable access
+
 v1.4.5
 
   * Allow passing :session_id and :session_password options #42 (thanks to avalanche123)
diff --git a/Gemfile b/Gemfile
@@ -1,4 +1,4 @@
-source :rubygems
+source 'https://rubygems.org'
 
 gemspec
 
@@ -8,7 +8,7 @@ group :test do
   gem "rspec" , "~> 2.11"
   gem 'eventmachine', '1.0.0'
   gem 'evented-spec', '~> 0.9.0'
-  gem 'zk-server', '~> 1.0'
+  gem 'zk-server', '~> 1.0', :git => 'https://github.com/zk-ruby/zk-server.git'
 end
 
 # ffs, :platform appears to be COMLETELY BROKEN so we just DO THAT HERE
diff --git a/Rakefile b/Rakefile
@@ -90,6 +90,14 @@ namespace :build do
 
     Rake::Task['build'].invoke
   end
+
+  task :clobber do
+    cd 'ext' do
+      sh 'rake clobber'
+    end
+
+    Rake::Task['build'].invoke
+  end
 end
 
 task :build do
diff --git a/ext/Rakefile b/ext/Rakefile
@@ -1,10 +1,6 @@
 require 'rbconfig'
 
-LIB_ZK_SO = 'lib/libzookeeper_mt_gem.la'
-
-TARBALL = FileList['zkc-*.tar.gz'].first
-
-raise "Where is the zkc tarball!?" unless TARBALL
+ZKRB_WRAPPER = %w[zkrb_wrapper.c zkrb_wrapper.h]
 
 namespace :zkrb do
   task :clean do
@@ -17,49 +13,29 @@ namespace :zkrb do
   end
 
   task :clobber => :clean do
-    rm_rf %w[Makefile c lib bin include ._c]
+    rm_rf %w[Makefile c lib bin include ._c] + ZKRB_WRAPPER
   end
 end
 
-task :clean   => 'zkrb:clean'
-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'
+task :clean    => 'zkrb:clean'
+task :clobber  => 'zkrb:clobber'
+task :wrappers => ZKRB_WRAPPER
+task :default  => :build
 
-file 'zkrb_wrapper.c' => GENERATE_GVL_CODE_RB do
+file 'zkrb_wrapper.c' => 'generate_gvl_code.rb' do
   sh "ruby generate_gvl_code.rb code"
 end
 
-file 'zkrb_wrapper.h' => GENERATE_GVL_CODE_RB do
+file 'zkrb_wrapper.h' => 'generate_gvl_code.rb' do
   sh "ruby generate_gvl_code.rb headers"
 end
 
-ZKRB_WRAPPER = %w[zkrb_wrapper.c zkrb_wrapper.h]
-
-task :wrappers => ZKRB_WRAPPER
-
 file 'Makefile' do
   sh "ruby extconf.rb"
 end
 
-task :build => [ZKRB_WRAPPER, 'Makefile'].flatten do
+task :build => [ 'Makefile', :wrappers ] do
   sh 'make'
 end
 
-task :default => :build
 
diff --git a/ext/c_zookeeper.rb b/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
@@ -269,7 +280,7 @@ def event_thread_body
       if @_shutting_down and not (@_closed or is_unrecoverable)
         logger.debug { "we're in shutting down state, there are #{@reg.in_flight.length} in_flight completions" }
 
-        until @reg.in_flight.empty? or is_unrecoverable or @_closed
+        until @reg.in_flight.empty? or @_closed or is_unrecoverable
           zkrb_iterate_event_loop
           iterate_event_delivery
           logger.debug { "there are #{@reg.in_flight} in_flight completions left" }
diff --git a/ext/event_lib.c b/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;
diff --git a/ext/event_lib.h b/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>
diff --git a/ext/extconf.rb b/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