Implement per-user and global transfer limits.

Author: bbockelm

src/Makefile.am

@@ -5,7 +5,7 @@ JNIHDIR=@JNIHDIR@
 INCLUDES= -I$(JNIHDIR) -I$(JNIHDIR)/linux @INCLUDE@
 
 #libglobus_gridftp_server_hdfs_la_SOURCES = globus_gridftp_server_hdfs.c
-libglobus_gridftp_server_hdfs_la_LDFLAGS = -lhdfs -lcrypto -lz
+libglobus_gridftp_server_hdfs_la_LDFLAGS = -lhdfs -lcrypto -lz -lrt
 libglobus_gridftp_server_hdfs_la_SOURCES = \
     gridftp_hdfs.h \
     gridftp_hdfs_error.h \

src/gridftp_hdfs.c

@@ -10,6 +10,8 @@
 #include <sys/syscall.h>
 #include <signal.h>
 #include <execinfo.h>
+#include <stdio.h>
+#include <fcntl.h>
 
 #include "gridftp_hdfs.h"
 
@@ -32,6 +34,10 @@ static globus_mutex_t g_hdfs_mutex;
 static pthread_t g_thread_id;
 static int g_thread_pipe_fd;
 
+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);
+static int dumb_sem_timedwait(int fd, int value, int secs);
+
 static void hdfs_trev(globus_gfs_event_info_t *, void *);
 inline void set_done(hdfs_handle_t *, globus_result_t);
 static int  hdfs_activate(void);
@@ -504,6 +510,8 @@ hdfs_start(
     int load_limit = 20;
     int replicas;
     int port;
+    int user_transfer_limit = -1;
+    int transfer_limit = -1;
 
     hdfs_handle = (hdfs_handle_t *)globus_malloc(sizeof(hdfs_handle_t));
     memset(hdfs_handle, 0, sizeof(hdfs_handle_t));
@@ -560,6 +568,34 @@ hdfs_start(
     char * port_char = getenv("GRIDFTP_HDFS_PORT");
     char * mount_point_char = getenv("GRIDFTP_HDFS_MOUNT_POINT");
     char * load_limit_char = getenv("GRIDFTP_LOAD_LIMIT");
+    char * global_transfer_limit_char = getenv("GRIDFTP_TRANSFER_LIMIT");
+    char * default_user_limit_char = getenv("GRIDFTP_DEFAULT_USER_TRANSFER_LIMIT");
+
+    char specific_limit_env_var[256];
+    snprintf(specific_limit_env_var, 255, "GRIDFTP_%s_USER_TRANSFER_LIMIT", hdfs_handle->username);
+    specific_limit_env_var[255] = '\0';
+    int idx;
+    for (idx=0; idx<256; idx++) {
+        if (specific_limit_env_var[idx] == '\0') {break;}
+        specific_limit_env_var[idx] = toupper(specific_limit_env_var[idx]);
+    }
+    char * specific_user_limit_char = getenv(specific_limit_env_var);
+
+    if (!specific_user_limit_char) {
+        specific_user_limit_char = default_user_limit_char;
+    }
+    if (specific_user_limit_char) {
+        user_transfer_limit = atoi(specific_user_limit_char);
+    }
+    if (global_transfer_limit_char) {
+        transfer_limit = atoi(global_transfer_limit_char);
+    }
+
+    if (load_limit_char != NULL) {
+        load_limit = atoi(load_limit_char);
+        if (load_limit < 1)
+            load_limit = 20;
+    }
 
     // Get our hostname
     hdfs_handle->local_host = globus_malloc(256);
@@ -645,7 +681,7 @@ hdfs_start(
     double load;
     int ctr = 0;
     while (fd >= 0) {
-        if (ctr == 120)
+        if (ctr == 10)
             break;
         ctr += 1;
         nbytes = read(fd, buf, bufsize);
@@ -658,13 +694,26 @@ hdfs_start(
         globus_gfs_log_message(GLOBUS_GFS_LOG_DUMP, "Detected system load %.2f.\n", load);
         if ((load >= load_limit) && (load < 1000)) {
             globus_gfs_log_message(GLOBUS_GFS_LOG_DUMP, "Preventing gridftp transfer startup due to system load of %.2f.\n", load);
-            sleep(5);
+            sleep(5*ctr);
         } else {
             break;
         }
         close(fd);
         fd = open("/proc/loadavg", O_RDONLY);
     }
+    if (load > load_limit) {
+        globus_gfs_log_message(GLOBUS_GFS_LOG_INFO, "Failing transfer due to load %.2f over limit %d\n", load, load_limit);
+        GenericError(hdfs_handle, "Server is cancelling transfer due to over-load limit", rc);
+        finished_info.result = rc;
+        globus_gridftp_server_operation_finished(op, rc, &finished_info);
+        return;
+    }
+
+    if ((rc = check_connection_limits(hdfs_handle, user_transfer_limit, transfer_limit)) != GLOBUS_SUCCESS) {
+        finished_info.result = rc;
+        globus_gridftp_server_operation_finished(op, rc, &finished_info);
+        return;
+    }
 
     globus_gfs_log_message(GLOBUS_GFS_LOG_INFO,
         "Start gridftp server; hadoop nameserver %s, port %i, replicas %i.\n",
@@ -808,3 +857,133 @@ set_close_done(
     }
 }
 
+/*************************************************************************
+ * check_connection_limits
+ * -----------------------
+ * Make sure the number of concurrent connections to HDFS is below a certain
+ * threshold.  If we are over-threshold, wait for a fixed amount of time (1 
+ * minute) and fail the transfer.
+ * Implementation baed on named POSIX semaphores.
+ *************************************************************************/
+globus_result_t
+check_connection_limits(const hdfs_handle_t *hdfs_handle, int user_transfer_limit, int transfer_limit)
+{
+    GlobusGFSName(check_connection_limit);
+    globus_result_t result = GLOBUS_SUCCESS;
+
+    int user_lock_count = 0;
+    if (user_transfer_limit > 0) {
+        char user_sem_name[256];
+        snprintf(user_sem_name, 255, "/dev/shm/gridftp-hdfs-%s-%d", hdfs_handle->username, user_transfer_limit);
+        user_sem_name[255] = '\0';
+        int usem = dumb_sem_open(user_sem_name, O_CREAT, 0600, user_transfer_limit);
+        if (usem == -1) {
+            SystemError(hdfs_handle, "Failure when determining user connection limit", result);
+            return result;
+        }
+        if (-1 == (user_lock_count = dumb_sem_timedwait(usem, user_transfer_limit, 60))) {
+            if (errno == ETIMEDOUT) {
+                globus_gfs_log_message(GLOBUS_GFS_LOG_INFO, "Failing transfer for %s due to user connection limit of %d.\n", hdfs_handle->username, user_transfer_limit);
+                char * failure_msg = (char *)globus_malloc(1024);
+                snprintf(failure_msg, 1024, "Server over the user connection limit of %d", user_transfer_limit);
+                failure_msg[1023] = '\0';
+                GenericError(hdfs_handle, failure_msg, result);
+                globus_free(failure_msg);
+            } else {
+                SystemError(hdfs_handle, "Failed to check user connection semaphore", result);
+            }
+            return result;
+        }
+        // NOTE: We now purposely leak the semaphore.  It will be automatically closed when
+        // the server process finishes this connection.
+    }
+
+    int global_lock_count = 0;
+    if (transfer_limit > 0) {
+        char global_sem_name[256];
+        snprintf(global_sem_name, 255, "/dev/shm//gridftp-hdfs-overall-%d", transfer_limit);
+        global_sem_name[255] = '\0';
+        int gsem = dumb_sem_open(global_sem_name, O_CREAT, 0666, transfer_limit);
+        if (gsem == -1) {
+            SystemError(hdfs_handle, "Failure when determining global connection limit", result);
+            return result;
+        }
+        if (-1 == (global_lock_count=dumb_sem_timedwait(gsem, transfer_limit, 60))) {
+            if (errno == ETIMEDOUT) {
+                globus_gfs_log_message(GLOBUS_GFS_LOG_INFO, "Failing transfer for %s due to global connection limit of %d (user has %d transfers).\n", hdfs_handle->username, transfer_limit, user_lock_count);
+                char * failure_msg = (char *)globus_malloc(1024);
+                snprintf(failure_msg, 1024, "Server over the global connection limit of %d (user has %d transfers)", transfer_limit, user_lock_count);
+                failure_msg[1023] = '\0';
+                GenericError(hdfs_handle, failure_msg, result);
+                globus_free(failure_msg);
+            } else {
+                SystemError(hdfs_handle, "Failed to check global connection semaphore", result);
+            }
+            return result;
+        }
+        // NOTE: We now purposely leak the semaphore.  It will be automatically closed when
+        // the server process finishes this connection.
+    }
+    if ((transfer_limit > 0) || (user_transfer_limit > 0)) {
+        globus_gfs_log_message(GLOBUS_GFS_LOG_INFO, "Proceeding with transfer; user %s has %d active transfers (limit %d); server has %d active transfers (limit %d).\n", hdfs_handle->username, user_lock_count, user_transfer_limit, global_lock_count, transfer_limit);
+    }
+
+    return result;
+}
+
+int
+dumb_sem_open(const char *fname, int flags, mode_t mode, int value) {
+    int fd = open(fname, flags | O_RDWR, mode);
+    if (-1 == fd) {
+        return fd;
+    }
+    if (-1 == posix_fallocate(fd, 0, value)) {
+        return -1;
+    }
+    fchmod(fd, mode);
+    return fd;
+}
+
+int
+dumb_sem_timedwait(int fd, int value, int secs) {
+    struct timespec start, now, sleeptime;
+    clock_gettime(CLOCK_MONOTONIC, &start);
+    sleeptime.tv_sec = 0;
+    sleeptime.tv_nsec = 500*1e6;
+    while (1) {
+        int idx = 0;
+        int lock_count = 0;
+        int need_lock = 1;
+        for (idx=0; idx<value; idx++) {
+            struct flock mylock; memset(&mylock, '\0', sizeof(mylock));
+            mylock.l_type = F_WRLCK;
+            mylock.l_whence = SEEK_SET;
+            mylock.l_start = idx;
+            mylock.l_len = 1;
+            if (0 == fcntl(fd, need_lock ? F_SETLK : F_GETLK, &mylock)) {
+                if (need_lock) {  // We now have the lock.
+                    need_lock = 0;
+                    lock_count++;
+                } else if (mylock.l_type != F_UNLCK) {  // We're just seeing how many locks are taken.
+                    lock_count++;
+                }
+                continue;
+            }
+            if (errno == EAGAIN || errno == EACCES || errno == EINTR) {
+                lock_count++;
+                continue;
+            }
+            return -1;
+        }
+        if (!need_lock) {  // we were able to take a lock.
+            return lock_count;
+        }
+        nanosleep(&sleeptime, NULL);
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        if (now.tv_sec > start.tv_sec + secs) {
+            errno = ETIMEDOUT;
+            return -1;
+        }
+    }
+}
+