ZOOKEEPER-319. add locking around auth info in zhandle_t. (chris darroch via mahadev)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@748653 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -11,6 +11,9 @@ BUGFIXES:
   
   ZOOKEEPER-330. zookeeper standalone server does not startup with just a port
 and datadir. (chris darroch and mahadev)
+
+  ZOOKEEPER-319. add locking around auth info in zhandle_t. (chris darroch via
+mahadev)
  
 IMPROVEMENTS:
   ZOOKEEPER-308. improve the atomic broadcast performance 3x. (breed via

src/c/src/mt_adaptor.c

@@ -41,6 +41,14 @@
 #include <assert.h>
 #include <unistd.h>
 
+void zoo_lock_auth(zhandle_t *zh)
+{
+    pthread_mutex_lock(&zh->auth.lock);
+}
+void zoo_unlock_auth(zhandle_t *zh)
+{
+    pthread_mutex_unlock(&zh->auth.lock);
+}
 void lock_buffer_list(buffer_head_t *l)
 {
     pthread_mutex_lock(&l->lock);
@@ -167,6 +175,8 @@ int adaptor_init(zhandle_t *zh)
     set_nonblock(adaptor_threads->self_pipe[1]);
     set_nonblock(adaptor_threads->self_pipe[0]);
 
+    pthread_mutex_init(&zh->auth.lock,0);
+
     zh->adaptor_priv = adaptor_threads;
     pthread_mutex_init(&zh->to_process.lock,0);
     pthread_mutex_init(&adaptor_threads->zh_lock,0);
@@ -226,6 +236,9 @@ void adaptor_destroy(zhandle_t *zh)
     pthread_mutex_destroy(&zh->completions_to_process.lock);
     pthread_cond_destroy(&zh->completions_to_process.cond);
     pthread_mutex_destroy(&adaptor->zh_lock);
+
+    pthread_mutex_destroy(&zh->auth.lock);
+
     close(adaptor->self_pipe[0]);
     close(adaptor->self_pipe[1]);
     free(adaptor);

src/c/src/st_adaptor.c

@@ -24,6 +24,12 @@
 #include <stdlib.h>
 #include <time.h>
 
+void zoo_lock_auth(zhandle_t *zh)
+{
+}
+void zoo_unlock_auth(zhandle_t *zh)
+{
+}
 void lock_buffer_list(buffer_head_t *l)
 {
 }

src/c/src/zk_adaptor.h

@@ -107,6 +107,9 @@ typedef struct _auth_info {
     struct buffer auth;
     void_completion_t completion;
     const char* data;
+#ifdef THREADED
+    pthread_mutex_t lock;
+#endif
 } auth_info;
 
 /**
@@ -213,6 +216,9 @@ int process_async(int outstanding_sync);
 void process_completions(zhandle_t *zh);
 int flush_send_queue(zhandle_t*zh, int timeout);
 
+void zoo_lock_auth(zhandle_t *zh);
+void zoo_unlock_auth(zhandle_t *zh);
+
 // critical section guards
 void enter_critical(zhandle_t* zh);
 void leave_critical(zhandle_t* zh);

src/c/src/zookeeper.c

@@ -856,21 +856,39 @@ static int handle_socket_error_msg(zhandle_t *zh, int line, int rc,
 
 static void auth_completion_func(int rc, zhandle_t* zh)
 {
+    void_completion_t auth_completion = NULL;
+    const char *auth_data = NULL;
+
     if(zh==NULL)
         return;
+
+    zoo_lock_auth(zh);
     
     if(rc!=0){
-        LOG_ERROR(("Authentication scheme %s failed. Connection closed.",
-                zh->auth.scheme));
         zh->state=ZOO_AUTH_FAILED_STATE;
     }else{
         zh->auth.state=1;  // active
+    }
+
+    if (zh->auth.completion) {
+        auth_completion = zh->auth.completion;
+        auth_data = zh->auth.data;
+        zh->auth.completion=0;
+    }
+
+    zoo_unlock_auth(zh);
+
+    if (rc) {
+        LOG_ERROR(("Authentication scheme %s failed. Connection closed.",
+                   zh->auth.scheme));
+    }
+    else {
         LOG_INFO(("Authentication scheme %s succeeded", zh->auth.scheme));
     }
+
     // chain call user's completion function
-    if(zh->auth.completion!=0){
-        zh->auth.completion(rc,zh->auth.data);
-        zh->auth.completion=0;
+    if (auth_completion) {
+        auth_completion(rc, auth_data);
     }
 }
 
@@ -881,15 +899,23 @@ static int send_auth_info(zhandle_t *zh)
     struct AuthPacket req;
     int rc;
 
-    if(zh->auth.scheme==NULL)
+    zoo_lock_auth(zh);
+
+    if(zh->auth.scheme==NULL) {
+      zoo_unlock_auth(zh);
       return ZOK; // there is nothing to send
+    }
 
     oa = create_buffer_oarchive();
+    rc = serialize_RequestHeader(oa, "header", &h);
+
     req.type=0;   // ignored by the server
     req.scheme = zh->auth.scheme;
     req.auth = zh->auth.auth;
-    rc = serialize_RequestHeader(oa, "header", &h);
     rc = rc < 0 ? rc : serialize_AuthPacket(oa, "req", &req);
+
+    zoo_unlock_auth(zh);
+
     /* add this buffer to the head of the send queue */
     rc = rc < 0 ? rc : queue_front_buffer_bytes(&zh->to_send, get_buffer(oa),
             get_buffer_len(oa));
@@ -2334,24 +2360,34 @@ const char* zerror(int c)
 int zoo_add_auth(zhandle_t *zh,const char* scheme,const char* cert, 
         int certLen,void_completion_t completion, const void *data)
 {
+    struct buffer auth;
+
     if(scheme==NULL || zh==NULL)
         return ZBADARGUMENTS;
     
     if (is_unrecoverable(zh))
         return ZINVALIDSTATE;
     
-    free_auth_info(&zh->auth);
-    zh->auth.scheme=strdup(scheme);
     if(cert!=NULL && certLen!=0){
-        zh->auth.auth.buff=calloc(1,certLen);
-        if(zh->auth.auth.buff==0)
+        auth.buff=calloc(1,certLen);
+        if(auth.buff==0) {
             return ZSYSTEMERROR;
-        memcpy(zh->auth.auth.buff,cert,certLen);
-        zh->auth.auth.len=certLen;
+        }
+        memcpy(auth.buff,cert,certLen);
+        auth.len=certLen;
     }
-    
+
+    zoo_lock_auth(zh);
+
+    free_auth_info(&zh->auth);
+    zh->auth.scheme=strdup(scheme);
+    if(cert!=NULL && certLen!=0)
+        zh->auth.auth=auth;
     zh->auth.completion=completion;
     zh->auth.data=data;
+
+    zoo_unlock_auth(zh);
+
     if(zh->state == ZOO_CONNECTED_STATE || zh->state == ZOO_ASSOCIATING_STATE)
         return send_auth_info(zh);