ZOOKEEPER-3162: Broken lock semantics in C client lock-recipe.

This PR fixes a few issues with the C client lock-recipe, as documented in more detailed in [ZOOKEEPER-3162](https://issues.apache.org/jira/browse/ZOOKEEPER-3162) on JIRA.

Details are also provided in the individual commits, but in short:
- Fix a bug in the choice of the predecessor node while trying to acquire the lock
- Fix a possible deadlock in zkr_lock_operation
- Fix the return value of zkr_lock_lock to abide to the prescribed semantics.

Author: Andrea Reale <realean2@ie.ibm.com>

Reviewers: andor@apache.org

Closes #662 from andreareale/ZOOKEEPER-3162 and squashes the following commits:

09f7ff4ed [Andrea Reale] Fixes deadlock in zoo_lock_operation
670d25a8e [Andrea Reale] Fix return semantics of zkr_lock_lock
2a1d66c4b [Andrea Reale] Bugfix on zookeeper-recipes-lock C implementation
a9b6a1a09 [Andrea Reale] Fix wrong include path for C recipes

zookeeper-recipes/zookeeper-recipes-lock/build.xml

@@ -52,7 +52,7 @@
 
 	<target name="compile-test" depends="compile">
   		<property name="target.jdk" value="${ant.java.version}" />	
-		<property name="src.test.local" location="${basedir}/test" />
+		<property name="src.test.local" location="${basedir}/src/test/java" />
 		<mkdir dir="${build.test}"/>
 		<javac srcdir="${src.test.local}" 
 			destdir="${build.test}" 

zookeeper-recipes/zookeeper-recipes-lock/src/main/c/configure.ac

@@ -48,8 +48,8 @@ DX_PS_FEATURE(OFF)
 DX_INIT_DOXYGEN([zookeeper-locks],[c-doc.Doxyfile],[docs])
 
   
-ZOOKEEPER_PATH=${BUILD_PATH}/../../../../zookeeper-client/zookeeper-client-c
-ZOOKEEPER_LD=-L${BUILD_PATH}/../../../../zookeeper-client/zookeeper-client-c\ -lzookeeper_mt
+ZOOKEEPER_PATH=${BUILD_PATH}/../../../../../zookeeper-client/zookeeper-client-c
+ZOOKEEPER_LD=-L${BUILD_PATH}/../../../../../zookeeper-client/zookeeper-client-c\ -lzookeeper_mt
 
 AC_SUBST(ZOOKEEPER_PATH)
 AC_SUBST(ZOOKEEPER_LD)

zookeeper-recipes/zookeeper-recipes-lock/src/main/c/src/zoo_lock.c

@@ -74,11 +74,7 @@ ZOOAPI int zkr_lock_init_cb(zkr_lock_mutex_t *mutex, zhandle_t* zh,
     return 0;
 }
 
-/**
- * unlock the mutex
- */
-ZOOAPI int zkr_lock_unlock(zkr_lock_mutex_t *mutex) {
-    pthread_mutex_lock(&(mutex->pmutex));
+static int _zkr_lock_unlock_nolock(zkr_lock_mutex_t *mutex) {
     zhandle_t *zh = mutex->zh;
     if (mutex->id != NULL) {
         int len = strlen(mutex->path) + strlen(mutex->id) + 2;
@@ -106,15 +102,23 @@ ZOOAPI int zkr_lock_unlock(zkr_lock_mutex_t *mutex) {
 
             free(mutex->id);
             mutex->id = NULL;
-            pthread_mutex_unlock(&(mutex->pmutex));
             return 0;
         }
         LOG_WARN(LOGCALLBACK(zh), ("not able to connect to server - giving up"));
-        pthread_mutex_unlock(&(mutex->pmutex));
         return ZCONNECTIONLOSS;
     }
+
+	return ZSYSTEMERROR;
+}
+/**
+ * unlock the mutex
+ */
+ZOOAPI int zkr_lock_unlock(zkr_lock_mutex_t *mutex) {
+	int ret = 0;
+    pthread_mutex_lock(&(mutex->pmutex));
+    ret = _zkr_lock_unlock_nolock(mutex);
     pthread_mutex_unlock(&(mutex->pmutex));
-    return ZSYSTEMERROR;
+    return ret;
 }
 
 static void free_String_vector(struct String_vector *v) {
@@ -128,10 +132,14 @@ static void free_String_vector(struct String_vector *v) {
     }
 }
 
+static int strcmp_suffix(const char *str1, const char *str2) {
+    return strcmp(strrchr(str1, '-')+1, strrchr(str2, '-')+1);
+}
+
 static int vstrcmp(const void* str1, const void* str2) {
     const char **a = (const char**)str1;
     const char **b = (const char**) str2;
-    return strcmp(strrchr(*a, '-')+1, strrchr(*b, '-')+1); 
+    return strcmp_suffix(*a, *b);
 } 
 
 static void sort_children(struct String_vector *vector) {
@@ -140,12 +148,24 @@ static void sort_children(struct String_vector *vector) {
         
 static char* child_floor(char **sorted_data, int len, char *element) {
     char* ret = NULL;
-    int i =0;
-    for (i=0; i < len; i++) {
-        if (strcmp(sorted_data[i], element) < 0) {
-            ret = sorted_data[i];
-        }
-    }
+    int targetpos = -1, s = 0, e = len -1;
+
+    while ( targetpos < 0 && s <= e ) {
+		int const i = s + (e - s) / 2;
+		int const cmp = strcmp_suffix(sorted_data[i], element);
+		if (cmp < 0) {
+			s = i + 1;
+		} else if (cmp == 0) {
+			targetpos = i;
+		} else {
+			e = i - 1;
+		}
+	}
+
+	if (targetpos > 0) {
+		ret = sorted_data[targetpos - 1];
+	}
+
     return ret;
 }
 
@@ -267,7 +287,7 @@ static int zkr_lock_operation(zkr_lock_mutex_t *mutex, struct timespec *ts) {
             // do not want to retry the create since 
             // we would end up creating more than one child
             if (ret != ZOK) {
-                LOG_WARN(LOGCALLBACK(zh), ("could not create zoo node %s", buf));
+                LOG_WARN(LOGCALLBACK(zh), "could not create zoo node %s", buf);
                 return ret;
             }
             mutex->id = getName(retbuf);
@@ -300,11 +320,11 @@ static int zkr_lock_operation(zkr_lock_mutex_t *mutex, struct timespec *ts) {
                 if (ret != ZOK) {
                     free_String_vector(vector);
                     LOG_WARN(LOGCALLBACK(zh), ("unable to watch my predecessor"));
-                    ret = zkr_lock_unlock(mutex);
+                    ret = _zkr_lock_unlock_nolock(mutex);
                     while (ret == 0) {
                         //we have to give up our leadership
                         // since we cannot watch out predecessor
-                        ret = zkr_lock_unlock(mutex);
+                        ret = _zkr_lock_unlock_nolock(mutex);
                     }
                     return ret;
                 }
@@ -315,7 +335,7 @@ static int zkr_lock_operation(zkr_lock_mutex_t *mutex, struct timespec *ts) {
                 // this is the case when we are the owner 
                 // of the lock
                 if (strcmp(mutex->id, owner_id) == 0) {
-                    LOG_DEBUG(LOGCALLBACK(zh), ("got the zoo lock owner - %s", mutex->id));
+                    LOG_DEBUG(LOGCALLBACK(zh), "got the zoo lock owner - %s", mutex->id);
                     mutex->isOwner = 1;
                     if (mutex->completion != NULL) {
                         mutex->completion(0, mutex->cbdata);
@@ -364,7 +384,7 @@ ZOOAPI int zkr_lock_lock(zkr_lock_mutex_t *mutex) {
         }
     }
     pthread_mutex_unlock(&(mutex->pmutex));
-    return zkr_lock_isowner(mutex);
+    return 0;
 }
 
                     

zookeeper-recipes/zookeeper-recipes-lock/src/main/c/tests/TestClient.cc

@@ -28,6 +28,7 @@ using namespace std;
 #include <cstring>
 #include <list>
 
+#include <unistd.h>
 #include <zookeeper.h>
 #include <zoo_lock.h>
 

zookeeper-recipes/zookeeper-recipes-queue/build.xml

@@ -52,7 +52,7 @@
 
 	<target name="compile-test" depends="compile">
   		<property name="target.jdk" value="${ant.java.version}" />	
-		<property name="src.test.local" location="${basedir}/test" />
+		<property name="src.test.local" location="${basedir}/src/test/java" />
 		<mkdir dir="${build.test}"/>
 		<javac srcdir="${src.test.local}" 
 			destdir="${build.test}" 

zookeeper-recipes/zookeeper-recipes-queue/src/main/c/configure.ac

@@ -48,8 +48,8 @@ DX_PS_FEATURE(OFF)
 DX_INIT_DOXYGEN([zookeeper-queues],[c-doc.Doxyfile],[docs])
 
   
-ZOOKEEPER_PATH=${BUILD_PATH}/../../../../zookeeper-client/zookeeper-client-c
-ZOOKEEPER_LD=-L${BUILD_PATH}/../../../../zookeeper-client/zookeeper-client-c\ -lzookeeper_mt
+ZOOKEEPER_PATH=${BUILD_PATH}/../../../../../zookeeper-client/zookeeper-client-c
+ZOOKEEPER_LD=-L${BUILD_PATH}/../../../../../zookeeper-client/zookeeper-client-c\ -lzookeeper_mt
 
 AC_SUBST(ZOOKEEPER_PATH)
 AC_SUBST(ZOOKEEPER_LD)