Merge pull request #18 from JasonAlt/master

Memory and hang issues found while testing

Author: bbockelm

src/gridftp_hdfs.c

@@ -36,8 +36,9 @@ char gridftp_file_name[PATH_MAX];
 char gridftp_transfer_type[10];
 
 static globus_mutex_t g_hdfs_mutex;
-static pthread_t g_thread_id;
-static int g_thread_pipe_fd;
+static pthread_t g_thread_id = 0;
+static int g_thread_pipe_fd = -1;
+static int g_thread_saved_stderr_fd = -1;
 
 static globus_result_t check_connection_limits(const hdfs_handle_t *, int, int);
 static int dumb_sem_open(const char *fname, int flags, mode_t mode, int value);
@@ -54,15 +55,6 @@ void
 hdfs_destroy(
     void *                              user_arg);
 
-void
-hdfs_start(
-    globus_gfs_operation_t              op,
-    globus_gfs_session_info_t *         session_info);
-
-void
-hdfs_destroy(
-    void *                              user_arg);
-
 
 /*
  *  Interface definitions for HDFS
@@ -231,14 +223,22 @@ setup_hdfs_logging()
         globus_mutex_unlock(&g_hdfs_mutex);
         return;
     }
-    if (-1 == dup3(pipe_fds[1], 2, O_CLOEXEC))
+
+    if ((g_thread_saved_stderr_fd = dup(STDERR_FILENO)) == -1)
+    {
+        globus_gfs_log_message(GLOBUS_GFS_LOG_ERR, "Failed to save stderr for HDFS logging: (errno=%d, %s).\n", errno, strerror(errno));
+        globus_mutex_unlock(&g_hdfs_mutex);
+        return;
+    }
+
+    if (-1 == dup3(pipe_fds[1], STDERR_FILENO, O_CLOEXEC))
     {
         globus_gfs_log_message(GLOBUS_GFS_LOG_ERR, "Failed to reopen stderr for HDFS logging: (errno=%d, %s).\n", errno, strerror(errno));
         globus_mutex_unlock(&g_hdfs_mutex);
         return;
     }
     close(pipe_fds[1]);
-    g_thread_pipe_fd = 2;
+    g_thread_pipe_fd = STDERR_FILENO;
 
     int *pipe_ptr = malloc(sizeof(int));
     if (pipe_ptr == NULL)
@@ -266,13 +266,6 @@ setup_hdfs_logging()
 int
 hdfs_activate(void)
 {
-    if (globus_mutex_init(&g_hdfs_mutex, GLOBUS_NULL)) {
-        globus_gfs_log_message(GLOBUS_GFS_LOG_ERR, "Unable to initialize global mutex");
-        return 1;
-    }
-    g_thread_id = 0;
-    g_thread_pipe_fd = -1;
-
     globus_extension_registry_add(
         GLOBUS_GFS_DSI_REGISTRY,
         "hdfs",
@@ -289,23 +282,6 @@ hdfs_activate(void)
 int
 hdfs_deactivate(void)
 {
-    if (g_thread_id > 0)
-    {
-        if (g_thread_pipe_fd >= 0)
-        {
-            fflush(stderr);
-            close(g_thread_pipe_fd);
-        }
-        void *retval;
-        pthread_join(g_thread_id, &retval);
-        g_thread_id = 0;
-        g_thread_pipe_fd = -1;
-    }
-
-    globus_mutex_destroy(&g_hdfs_mutex);
-    g_thread_id = 0;
-    g_thread_pipe_fd = -1;
-
     globus_extension_registry_remove(
         GLOBUS_GFS_DSI_REGISTRY, "hdfs");
 
@@ -568,7 +544,7 @@ hdfs_start(
     globus_gfs_operation_t              op,
     globus_gfs_session_info_t *         session_info)
 {
-    hdfs_handle_t*       hdfs_handle;
+    hdfs_handle_t*       hdfs_handle = NULL;
     globus_gfs_finished_info_t          finished_info;
     GlobusGFSName(hdfs_start);
     globus_result_t rc;
@@ -581,6 +557,12 @@ hdfs_start(
     int user_transfer_limit = -1;
     int transfer_limit = -1;
 
+    if (globus_mutex_init(&g_hdfs_mutex, GLOBUS_NULL)) {
+        SystemError(hdfs_handle, "Unable to initialize mutex", rc);
+        globus_gridftp_server_operation_finished(op, rc, &finished_info);
+        return;
+    }
+
     hdfs_handle = (hdfs_handle_t *)globus_malloc(sizeof(hdfs_handle_t));
     memset(hdfs_handle, 0, sizeof(hdfs_handle_t));
 
@@ -605,6 +587,7 @@ hdfs_start(
         globus_gridftp_server_operation_finished(op, rc, &finished_info);
         return;
     }
+
     if (globus_mutex_init(hdfs_handle->mutex, GLOBUS_NULL)) {
         SystemError(hdfs_handle, "Unable to initialize mutex", rc);
         globus_gridftp_server_operation_finished(op, rc, &finished_info);
@@ -846,9 +829,32 @@ hdfs_destroy(
             globus_mutex_destroy(hdfs_handle->mutex);
             globus_free(hdfs_handle->mutex);
         }
+        if (hdfs_handle->cvmfs_graft)
+            free(hdfs_handle->cvmfs_graft);
         globus_free(hdfs_handle);
-        free(hdfs_handle->cvmfs_graft);
     }
+
+    if (g_thread_id > 0)
+    {
+        if (g_thread_saved_stderr_fd >= 0)
+        {
+            dup2(g_thread_saved_stderr_fd, STDERR_FILENO);
+            close(g_thread_saved_stderr_fd);
+        }
+
+        if (g_thread_pipe_fd >= 0)
+        {
+            close(g_thread_pipe_fd);
+        }
+
+        void *retval;
+        pthread_join(g_thread_id, &retval);
+    }
+
+    g_thread_id = 0;
+    g_thread_pipe_fd = -1;
+    g_thread_saved_stderr_fd = -1;
+
     closelog();
 }
 
@@ -915,18 +921,6 @@ set_close_done(
     if ((hdfs_handle->done_status == GLOBUS_SUCCESS) && (rc != GLOBUS_SUCCESS)) {
         hdfs_handle->done_status = rc;
     }
-    if (g_thread_id > 0)
-    {
-        if (g_thread_pipe_fd >= 0)
-        {
-            fflush(stderr);
-            close(g_thread_pipe_fd);
-        }
-        void *retval;
-        pthread_join(g_thread_id, &retval);
-        g_thread_id = 0;
-        g_thread_pipe_fd = -1;
-    }
 }
 
 /*************************************************************************

src/gridftp_hdfs_cksm.c

@@ -394,15 +394,15 @@ globus_result_t hdfs_save_checksum(hdfs_handle_t *hdfs_handle) {
 }
 
 static globus_result_t
-hdfs_calculate_checksum(hdfs_handle_t *hdfs_handle, hdfsFS fs, const char *type)
+hdfs_calculate_checksum(hdfs_handle_t *hdfs_handle, const char *type)
 {
     globus_result_t rc = GLOBUS_SUCCESS;
 
     GlobusGFSName(hdfs_calculate_checksum);
 
     hdfs_initialize_checksums(hdfs_handle);
 
-    hdfsFile fd = hdfsOpenFile(fs, hdfs_handle->pathname, O_RDONLY, 0, 1, 0);
+    hdfsFile fd = hdfsOpenFile(hdfs_handle->fs, hdfs_handle->pathname, O_RDONLY, 0, 1, 0);
     if (fd == NULL) {
         SystemError(hdfs_handle, "Failed to open file for checksumming", rc)
         return rc;
@@ -412,23 +412,23 @@ hdfs_calculate_checksum(hdfs_handle_t *hdfs_handle, hdfsFS fs, const char *type)
     void *buffer = malloc(cksum_buffer_size);
     if (buffer == NULL) {
         MemoryError(hdfs_handle, "Unable to allocate checksum temp buffer", rc);
-        hdfsCloseFile(fs, fd);
+        hdfsCloseFile(hdfs_handle->fs, fd);
         return rc;
     }
     ssize_t retval = 0;
     globus_off_t offset = 0;
     do {
         hdfs_update_checksums(hdfs_handle, buffer, retval);
         errno = 0;  // older versions of libhdfs sometimes fails to reset errno.
-        retval = hdfsRead(fs, fd, buffer, cksum_buffer_size);
+        retval = hdfsRead(hdfs_handle->fs, fd, buffer, cksum_buffer_size);
         if ((retval == -1) && (errno == EINTR)) {continue;}
         offset += retval;
     } while (retval > 0);
     if (retval == -1) {
         SystemError(hdfs_handle, "Failed to read from file for checksumming", rc);
         // Fall-through
     }
-    hdfsCloseFile(fs, fd);
+    hdfsCloseFile(hdfs_handle->fs, fd);
 
     if (rc == GLOBUS_SUCCESS) {
         hdfs_handle->offset = offset;
@@ -490,7 +490,7 @@ globus_result_t hdfs_get_checksum_internal(hdfs_handle_t *hdfs_handle, const cha
 
     hdfsFile fh = hdfsOpenFile(fs, filename, O_RDONLY, 0, 0, 0);
     if (fh == NULL) {
-        rc = hdfs_calculate_checksum(hdfs_handle, fs, requested_cksm);
+        rc = hdfs_calculate_checksum(hdfs_handle, requested_cksm);
         if (rc != GLOBUS_SUCCESS) {return rc;}
         fh = hdfsOpenFile(fs, filename, O_RDONLY, 0, 0, 0);
         if (fh == NULL) {

src/gridftp_hdfs_error.h

@@ -19,9 +19,10 @@
 
 
 #define SystemError(hdfs_handle, msg, rc) \
+    int system_errno = errno; \
     SomeError(hdfs_handle, msg) \
-    rc = GlobusGFSErrorSystemError(formatted_msg, errno); \
-    globus_free(formatted_msg);
+    rc = GlobusGFSErrorSystemError(formatted_msg, system_errno); \
+    globus_free(formatted_msg); \
 
 
 #define MemoryError(hdfs_handle, msg, rc) \