server-helpers: cleanup connection only if there are no active transports.

- thanks to Ioannis Aslanidis <iaslanidis@flumotion.com> for reporting.
  - breakup the server_connection_cleanup into smaller procedures.
  - do following operations in a single atomic operation.
      1. conn->active_transports--
      2. collecting pointer to lock table and all fds if there are no active transports
    this will avoid any race conditions.

Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

3 files changed, 194 insertions, 129 deletions

diff --git a/xlators/protocol/server/src/server-helpers.c b/xlators/protocol/server/src/server-helpers.c
index a4e0b2081..f49673546 100644
--- a/xlators/protocol/server/src/server-helpers.c
+++ b/xlators/protocol/server/src/server-helpers.c
@@ -391,6 +391,110 @@ out:
 }
 
 
+int 
+do_lock_table_cleanup (xlator_t *this, server_connection_t *conn,
+                       call_frame_t *frame, struct _lock_table *ltable)
+{
+        struct list_head  file_lockers, dir_lockers;
+        call_frame_t     *tmp_frame = NULL;
+        struct flock      flock = {0, };
+        xlator_t         *bound_xl = NULL;
+	struct _locker   *locker = NULL, *tmp = NULL;
+        int               ret = -1;
+        
+        bound_xl = conn->bound_xl;        
+        INIT_LIST_HEAD (&file_lockers);
+        INIT_LIST_HEAD (&dir_lockers);
+        
+        LOCK (&ltable->lock);
+        {
+                list_splice_init (&ltable->file_lockers, 
+                                  &file_lockers);
+                
+                list_splice_init (&ltable->dir_lockers, &dir_lockers);
+        }
+        UNLOCK (&ltable->lock);
+
+        free (ltable);
+
+        flock.l_type  = F_UNLCK;
+        flock.l_start = 0;
+        flock.l_len   = 0;
+        list_for_each_entry_safe (locker, 
+                                  tmp, &file_lockers, lockers) {
+                tmp_frame = copy_frame (frame);
+                if (tmp_frame == NULL) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "out of memory");
+                        goto out;
+                }
+                /* 
+                   pid = 0 is a special case that tells posix-locks
+                   to release all locks from this transport
+                */
+                tmp_frame->root->pid = 0;
+                tmp_frame->root->trans = conn;
+
+                if (locker->fd) {
+                        STACK_WIND (tmp_frame, server_nop_cbk,
+                                    bound_xl,
+                                    bound_xl->fops->finodelk,
+                                    locker->volume,
+                                    locker->fd, F_SETLK, &flock);
+                        fd_unref (locker->fd);
+                } else {
+                        STACK_WIND (tmp_frame, server_nop_cbk,
+                                    bound_xl,
+                                    bound_xl->fops->inodelk,
+                                    locker->volume,
+                                    &(locker->loc), F_SETLK, &flock);
+                        loc_wipe (&locker->loc);
+                }
+
+                free (locker->volume);
+                
+                list_del_init (&locker->lockers);
+                free (locker);
+        }
+
+        tmp = NULL;
+        locker = NULL;
+        list_for_each_entry_safe (locker, tmp, &dir_lockers, lockers) {
+                tmp_frame = copy_frame (frame);
+                
+                tmp_frame->root->pid = 0;
+                tmp_frame->root->trans = conn;
+
+                if (locker->fd) {
+                        STACK_WIND (tmp_frame, server_nop_cbk,
+                                    bound_xl,
+                                    bound_xl->fops->fentrylk,
+                                    locker->volume,
+                                    locker->fd, NULL, 
+                                    ENTRYLK_UNLOCK, ENTRYLK_WRLCK);
+                        fd_unref (locker->fd);
+                } else {
+                        STACK_WIND (tmp_frame, server_nop_cbk,
+                                    bound_xl,
+                                    bound_xl->fops->entrylk,
+                                    locker->volume,
+                                    &(locker->loc), NULL, 
+                                    ENTRYLK_UNLOCK, ENTRYLK_WRLCK);
+                        loc_wipe (&locker->loc);
+                }
+
+                free (locker->volume);
+                
+                list_del_init (&locker->lockers);
+                free (locker);
+        }
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
 static int32_t
 server_connection_cleanup_flush_cbk (call_frame_t *frame,
                                      void *cookie,
@@ -399,6 +503,7 @@ server_connection_cleanup_flush_cbk (call_frame_t *frame,
                                      int32_t op_errno)
 {
         fd_t *fd = NULL;
+
         fd = frame->local;
 
         fd_unref (fd);
@@ -410,159 +515,113 @@ server_connection_cleanup_flush_cbk (call_frame_t *frame,
 
 
 int
-server_connection_cleanup (xlator_t *this, server_connection_t *conn)
+do_fd_cleanup (xlator_t *this, server_connection_t *conn, call_frame_t *frame,
+               fd_t **fds, int fd_count)
 {
-	call_frame_t       *frame = NULL, *tmp_frame = NULL;
-	xlator_t           *bound_xl = NULL;
-	server_state_t     *state = NULL;
-	struct list_head    file_lockers;
-	struct list_head    dir_lockers;
-	struct _lock_table *ltable = NULL;
-	struct _locker     *locker = NULL, *tmp = NULL;
-	struct flock        flock = {0,};
         fd_t               *fd = NULL;
-        uint32_t            fd_count = 0;
-        fd_t              **fds = NULL;
-        int32_t             i = 0;
-
-        if (conn == NULL) {
-                goto out;
+        int                 i = 0, ret = -1;
+        call_frame_t       *tmp_frame = NULL;
+        xlator_t           *bound_xl = NULL;
+        
+        bound_xl = conn->bound_xl;
+        for (i = 0;i < fd_count; i++) {
+                fd = fds[i];
+                
+                if (fd != NULL) {
+                        tmp_frame = copy_frame (frame);
+                        if (tmp_frame == NULL) {
+                                gf_log (this->name, GF_LOG_ERROR,
+                                        "out of memory");
+                                goto out;
+                        }
+                        tmp_frame->local = fd;
+                        
+                        tmp_frame->root->pid = 0;
+                        tmp_frame->root->trans = conn;
+                        STACK_WIND (tmp_frame,
+                                    server_connection_cleanup_flush_cbk,
+                                    bound_xl,
+                                    bound_xl->fops->flush,
+                                    fd);
+                }
         }
+        FREE (fds);
+        ret = 0;
 
-	bound_xl = (xlator_t *) (conn->bound_xl);
-
-	if (bound_xl) {
-		/* trans will have ref_count = 1 after this call, but its 
-		   ok since this function is called in 
-		   GF_EVENT_TRANSPORT_CLEANUP */
-		frame = create_frame (this, this->ctx->pool);
-
-		pthread_mutex_lock (&(conn->lock));
-		{
-			if (conn->ltable) {
-				ltable = conn->ltable;
-				conn->ltable = gf_lock_table_new ();
-			}
-		}
-		pthread_mutex_unlock (&conn->lock);
+out:
+        return ret;
+}
 
-		INIT_LIST_HEAD (&file_lockers);
-		INIT_LIST_HEAD (&dir_lockers);
+int
+do_connection_cleanup (xlator_t *this, server_connection_t *conn,
+                       struct _lock_table *ltable, fd_t **fds, int fd_count)
+{
+        int32_t       ret = 0, saved_ret = 0;
+	call_frame_t *frame = NULL;
+        server_state_t *state = NULL;
 
-		LOCK (&ltable->lock);
-		{
-			list_splice_init (&ltable->file_lockers, 
-					  &file_lockers);
+        frame = create_frame (this, this->ctx->pool);
+        if (frame == NULL) {
+                gf_log (this->name, GF_LOG_ERROR, "out of memory");
+                goto out;
+        }
 
-			list_splice_init (&ltable->dir_lockers, &dir_lockers);
-		}
-		UNLOCK (&ltable->lock);
-		free (ltable);
+        saved_ret = do_lock_table_cleanup (this, conn, frame, ltable);
 
-		flock.l_type  = F_UNLCK;
-		flock.l_start = 0;
-		flock.l_len   = 0;
-		list_for_each_entry_safe (locker, 
-					  tmp, &file_lockers, lockers) {
-			tmp_frame = copy_frame (frame);
-			/* 
-			   pid = 0 is a special case that tells posix-locks
-			   to release all locks from this transport
-			*/
-			tmp_frame->root->pid = 0;
-			tmp_frame->root->trans = conn;
+        if (fds != NULL) {
+                ret = do_fd_cleanup (this, conn, frame, fds, fd_count);
+        }
 
-			if (locker->fd) {
-				STACK_WIND (tmp_frame, server_nop_cbk,
-					    bound_xl,
-					    bound_xl->fops->finodelk,
-                                            locker->volume,
-					    locker->fd, F_SETLK, &flock);
-				fd_unref (locker->fd);
-			} else {
-				STACK_WIND (tmp_frame, server_nop_cbk,
-					    bound_xl,
-					    bound_xl->fops->inodelk,
-                                            locker->volume,
-					    &(locker->loc), F_SETLK, &flock);
-				loc_wipe (&locker->loc);
-			}
+        state = CALL_STATE (frame);
+        if (state)
+                free (state);
 
-                        free (locker->volume);
+        STACK_DESTROY (frame->root);
 
-			list_del_init (&locker->lockers);
-			free (locker);
-		}
+        if (saved_ret || ret) {
+                ret = -1;
+        }
+        
+out:
+        return ret;
+}
 
-		tmp = NULL;
-		locker = NULL;
-		list_for_each_entry_safe (locker, tmp, &dir_lockers, lockers) {
-			tmp_frame = copy_frame (frame);
+int
+server_connection_cleanup (xlator_t *this, server_connection_t *conn)
+{
+        char                do_cleanup = 0;
+	struct _lock_table *ltable = NULL;
+        fd_t              **fds = NULL;
+        uint32_t            fd_count = 0;
+        int                 ret = 0; 
 
-			tmp_frame->root->pid = 0;
-			tmp_frame->root->trans = conn;
+        if (conn == NULL) {
+                goto out;
+        }
 
-			if (locker->fd) {
-				STACK_WIND (tmp_frame, server_nop_cbk,
-					    bound_xl,
-					    bound_xl->fops->fentrylk,
-                                            locker->volume,
-					    locker->fd, NULL, 
-					    ENTRYLK_UNLOCK, ENTRYLK_WRLCK);
-				fd_unref (locker->fd);
-			} else {
-				STACK_WIND (tmp_frame, server_nop_cbk,
-					    bound_xl,
-					    bound_xl->fops->entrylk,
-                                            locker->volume,
-					    &(locker->loc), NULL, 
-					    ENTRYLK_UNLOCK, ENTRYLK_WRLCK);
-				loc_wipe (&locker->loc);
+        pthread_mutex_lock (&conn->lock);
+        {
+                conn->active_transports--;
+                if (conn->active_transports == 0) {
+			if (conn->ltable) {
+				ltable = conn->ltable;
+				conn->ltable = gf_lock_table_new ();
 			}
 
-                        free (locker->volume);
-
-			list_del_init (&locker->lockers);
-			free (locker);
-		}
-
-                pthread_mutex_lock (&conn->lock);
-                {
                         if (conn->fdtable) {
                                 fds = gf_fd_fdtable_get_all_fds (conn->fdtable,
                                                                  &fd_count);
                         }
+                        do_cleanup = 1;
                 }
-                pthread_mutex_unlock (&conn->lock);
-
-                if (fds != NULL) {
-                        for (i = 0;i < fd_count; i++) {
-                                fd = fds[i];
-
-                                if (fd != NULL) {
-                                        tmp_frame = copy_frame (frame);
-                                        tmp_frame->local = fd;
-                                        
-                                        tmp_frame->root->pid = 0;
-                                        tmp_frame->root->trans = conn;
-                                        STACK_WIND (tmp_frame,
-                                                    server_connection_cleanup_flush_cbk,
-                                                    bound_xl,
-                                                    bound_xl->fops->flush,
-                                                    fd);
-                                }
-                        }
-                        FREE (fds);
-                }
-
-		state = CALL_STATE (frame);
-		if (state)
-			free (state);
-		STACK_DESTROY (frame->root);
         }
+        pthread_mutex_unlock (&conn->lock);
+
+        if (do_cleanup && conn->bound_xl)
+                ret = do_connection_cleanup (this, conn, ltable, fds, fd_count);
 
 out:
-        return 0;
+        return ret;
 }
 
 
@@ -761,6 +820,7 @@ server_connection_get (xlator_t *this, const char *id)
 		}
 
 		conn->ref++;
+                conn->active_transports++;
 	}
 	pthread_mutex_unlock (&conf->mutex);
 
diff --git a/xlators/protocol/server/src/server-protocol.c b/xlators/protocol/server/src/server-protocol.c
index 7fc379efb..47bc2ae96 100644
--- a/xlators/protocol/server/src/server-protocol.c
+++ b/xlators/protocol/server/src/server-protocol.c
@@ -7803,6 +7803,10 @@ notify (xlator_t *this, int32_t event, void *data, ...)
                                 "handshake with (%s) is successful",
                                 myinfo->identifier, peerinfo->identifier);
                 } else {
+                        /*
+                         * FIXME: shouldn't we check for return value?
+                         * what should be done if cleanup fails?
+                         */
                         server_connection_cleanup (this, trans->xl_private);
                 }
 	}
diff --git a/xlators/protocol/server/src/server-protocol.h b/xlators/protocol/server/src/server-protocol.h
index 1ea30cc6f..dabe6927b 100644
--- a/xlators/protocol/server/src/server-protocol.h
+++ b/xlators/protocol/server/src/server-protocol.h
@@ -63,6 +63,7 @@ struct _server_connection {
 	struct list_head    list;
 	char               *id;
 	int                 ref;
+        int                 active_transports;
 	pthread_mutex_t     lock;
 	char                disconnected;
 	fdtable_t          *fdtable;