Merge branch 'develop'

CMakeLists.txt

@@ -45,7 +45,7 @@ set (OUTPUT_NAME_TRANSFER "gfal_transfer")
 
 set (VERSION_MAJOR  2)
 set (VERSION_MINOR  16)
-set (VERSION_PATCH  1)
+set (VERSION_PATCH  2)
 set (VERSION_STRING ${VERSION_MAJOR}.${VERSION_MINOR}.${VERSION_PATCH})
 
 add_definitions(-DVERSION="${VERSION_STRING}")

dist/etc/gfal2.d/http_plugin.conf

@@ -7,13 +7,18 @@ ENABLE_REMOTE_COPY=true
 ## Enable streamed copies (data passes via the client)
 ENABLE_STREAM_COPY=true
 
+## Enable fallback copies: in case of TPC we start first with DEFAULT_COPY_MODE
+## and we fallback in case of an error
+ENABLE_FALLBACK_TPC_COPY=true
+
 ## Default mode for http copy
 ## possible values are: 3rd pull, 3rd push or streamed 
 ## By default all methods will be tried in case of failures starting from 3rd pull
 ## if 3rd push is set as default only 3rd push and then streamed will be executed
 ## if streamed is set as detault only streamed transfer will be executed
 DEFAULT_COPY_MODE=3rd pull
 
+
 # Enable or disable the SSL CA check
 INSECURE=false
 

packaging/debian/changelog

@@ -1,3 +1,8 @@
+gfal2 (2.16.2-1) unstable; urgency=low
+  * New version
+
+ -- DMC Devel <dmc-devel@cern.ch> Tue, 27 Nov 2018 14:00:00 +0100
+
 gfal2 (2.16.1-1) unstable; urgency=low
   * New version
 

packaging/rpm/gfal2.spec

@@ -4,7 +4,7 @@
 %bcond_with tests
 
 Name:               gfal2
-Version:            2.16.1
+Version:            2.16.2
 Release:            1%{?dist}
 Summary:            Grid file access library 2.0
 Group:              Applications/Internet

src/plugins/gridftp/gridftp_filecopy.cpp

@@ -435,7 +435,10 @@ int gridftp_filecopy_copy_file_internal(GridFTPModule* module,
     if (nb_streams_from_conf !=0 ) {
         nbstream = nb_streams_from_conf;
     }
-
+    //always set streams = 0 if source or dest are ftp, this disables the  GLOBUS_FTP_CONTROL_MODE_EXTENDED_BLOCK
+    if (strncmp(src, "ftp:", 4) == 0 || strncmp(dst, "ftp:", 4) == 0) {
+    	nbstream = 0;
+    }
     handler.session->set_nb_streams(nbstream);
 
     gfal2_log(G_LOG_LEVEL_DEBUG,

src/plugins/gridftp/gridftpwrapper.cpp

@@ -212,10 +212,6 @@ void GridFTPSession::set_dcau(bool dcau)
 
 void GridFTPSession::set_nb_streams(unsigned int nbstream)
 {
-    // plain FTP urls do not support mode
-    if (baseurl.compare(0, 3, "ftp") == 0) {
-        return;
-    }
     if (nbstream == 0) {
         parallelism.fixed.size = 1;
         parallelism.mode = GLOBUS_FTP_CONTROL_PARALLELISM_NONE;

src/plugins/http/gfal_http_copy.cpp

@@ -40,12 +40,12 @@ const char* CopyModeStr[] = {
     GFAL_TRANSFER_TYPE_PULL, GFAL_TRANSFER_TYPE_PUSH, GFAL_TRANSFER_TYPE_STREAMED, NULL
 };
 
-static const CopyMode getCopyModeFromString(const char * copyModeStr) { 
-    if (!strcmp(copyModeStr, GFAL_TRANSFER_TYPE_PULL))  
+static const CopyMode get_copy_mode_from_string(const char * copy_mode_str) { 
+    if (!strcmp(copy_mode_str, GFAL_TRANSFER_TYPE_PULL))  
         return HTTP_COPY_PULL;
-    if (!strcmp(copyModeStr, GFAL_TRANSFER_TYPE_PUSH)) 
+    if (!strcmp(copy_mode_str, GFAL_TRANSFER_TYPE_PUSH)) 
         return HTTP_COPY_PUSH;
-     if (!strcmp(copyModeStr, GFAL_TRANSFER_TYPE_STREAMED))
+     if (!strcmp(copy_mode_str, GFAL_TRANSFER_TYPE_STREAMED))
         return HTTP_COPY_STREAM;
      return HTTP_COPY_PULL;
 }
@@ -62,6 +62,76 @@ struct PerfCallbackData {
     }
 };
 
+static bool set_copy_mode(gfal2_context_t context,const char * copy_mode) {
+    bool set = true;   
+    GError *error = NULL;
+    if (!strcmp(copy_mode,"push" )) {
+        gfal2_set_opt_boolean(context, "HTTP PLUGIN", "ENABLE_REMOTE_COPY", TRUE, &error);
+        gfal2_set_opt_string(context, "HTTP PLUGIN", "DEFAULT_COPY_MODE", GFAL_TRANSFER_TYPE_PUSH, &error);
+        gfal2_set_opt_boolean(context, "HTTP PLUGIN", "ENABLE_FALLBACK_TPC_COPY", FALSE, &error);
+        if (error != NULL) {
+            g_clear_error(&error);
+            set =false;
+        }
+    } else if (!strcmp(copy_mode,"pull") ) {
+        gfal2_set_opt_boolean(context, "HTTP PLUGIN", "ENABLE_REMOTE_COPY", TRUE, &error);
+        gfal2_set_opt_string(context, "HTTP PLUGIN", "DEFAULT_COPY_MODE", GFAL_TRANSFER_TYPE_PULL, &error);
+        gfal2_set_opt_boolean(context, "HTTP PLUGIN", "ENABLE_FALLBACK_TPC_COPY", FALSE, &error);
+        if (error != NULL) {
+            g_clear_error(&error);
+            set =false;
+        }
+    } else {
+        set =false;
+    }
+    return set;
+}
+
+static void extract_query_parameter(const char* url, const char *key, char *value, size_t val_size)
+{
+    value[0] = '\0';
+
+    const char *str = strchr(url, '?');
+    if (str == NULL) {
+        return;
+    }
+
+    size_t key_len = strlen(key);
+    char **args = g_strsplit(str + 1, "&", 0);
+    int i;
+    for (i = 0; args[i] != NULL; ++i) {
+        if (strncmp(args[i], key, key_len) == 0) {
+            char *p = strchr(args[i], '=');
+            if (p) {
+                g_strlcpy(value, p + 1, val_size);
+                break;
+            }
+        }
+    }
+
+    g_strfreev(args);
+}    
+
+static void set_copy_mode_from_urls(gfal2_context_t context, const char * src_url, const char * dst_url)
+{
+    GError *error = NULL;
+    char copy_mode[64] = {0};
+    bool done = false;
+    extract_query_parameter(src_url, "copy_mode", copy_mode,sizeof(copy_mode));
+    if (copy_mode[0] != '\0') {
+            gfal2_log(G_LOG_LEVEL_INFO,"Source copy mode is %s", copy_mode);
+            done= set_copy_mode(context, copy_mode); 
+    }
+
+    if (!done) {
+        extract_query_parameter(dst_url, "copy_mode", copy_mode,sizeof(copy_mode));
+        if (copy_mode[0] != '\0') {
+	    gfal2_log(G_LOG_LEVEL_INFO,"Destination copy mode is %s", copy_mode);
+            set_copy_mode(context,copy_mode);
+
+        }
+    }
+}
 
 static bool is_http_scheme(const char* url)
 {
@@ -90,7 +160,12 @@ static bool is_http_streamed_enabled(gfal2_context_t context)
 
 static CopyMode get_default_copy_mode(gfal2_context_t context)
 {
-    return getCopyModeFromString(gfal2_get_opt_string_with_default(context, "HTTP PLUGIN", "DEFAULT_COPY_MODE", GFAL_TRANSFER_TYPE_PULL));
+    return get_copy_mode_from_string(gfal2_get_opt_string_with_default(context, "HTTP PLUGIN", "DEFAULT_COPY_MODE", GFAL_TRANSFER_TYPE_PULL));
+}
+
+static bool is_http_3rdcopy_fallback_enabled(gfal2_context_t context)
+{
+    return gfal2_get_opt_boolean_with_default(context, "HTTP PLUGIN", "ENABLE_FALLBACK_TPC_COPY", TRUE);
 }
 
 static int gfal_http_exists(plugin_handle plugin_data,
@@ -567,24 +642,35 @@ int gfal_http_copy(plugin_handle plugin_data, gfal2_context_t context,
                          GFAL_EVENT_NONE, GFAL_EVENT_PREPARE_EXIT,
                          "%s => %s", src_full, dst_full);
 
+
+    set_copy_mode_from_urls(context,src_full, dst_full);
     // Initial copy mode
     CopyMode copy_mode = get_default_copy_mode(context);
-
+
+    bool only_streaming = false;
     // If source is not even http, go straight to streamed
     // or if third party copy is disabled, go straight to streamed
     if (!is_http_scheme(src) || !is_http_3rdcopy_enabled(context)) {
         copy_mode = HTTP_COPY_STREAM;
+        only_streaming = true;
     }
 
     // Re-try different approaches
     int ret = 0;
-    while (copy_mode < HTTP_COPY_END) {
+
+    CopyMode end_copy_mode = HTTP_COPY_END;
+
+    //if streaming is disabled stop the loop before
+    if (!is_http_streamed_enabled(context)) {
+        end_copy_mode = HTTP_COPY_STREAM;
+    }
+
+    do {
         // The real, actual, copy
         plugin_trigger_event(params, http_plugin_domain,
                                      GFAL_EVENT_NONE, GFAL_EVENT_TRANSFER_ENTER,
                                "%s => %s", src_full, dst_full);
-        gfal2_log(G_LOG_LEVEL_MESSAGE,
-            "Trying copying with mode %s",
+        gfal2_log(G_LOG_LEVEL_MESSAGE, "Trying copying with mode %s",
             CopyModeStr[copy_mode]);
         plugin_trigger_event(params, http_plugin_domain,
             GFAL_EVENT_NONE, GFAL_EVENT_TRANSFER_TYPE,
@@ -598,7 +684,8 @@ int gfal_http_copy(plugin_handle plugin_data, gfal2_context_t context,
                     checksum_mode, checksum_type, user_checksum,
                     params, &nested_error);
             }
-            else {
+            else if (only_streaming) {
+                ret = -1;
                 gfalt_set_error(&nested_error, http_plugin_domain, EIO, __func__,
                     GFALT_ERROR_TRANSFER, "STREAMED DISABLED",
                     "Trying to fallback to a streamed copy, but they are disabled");
@@ -614,16 +701,17 @@ int gfal_http_copy(plugin_handle plugin_data, gfal2_context_t context,
             break;
         }
         else if (ret < 0) {
-               plugin_trigger_event(params, http_plugin_domain,
-                         GFAL_EVENT_NONE, GFAL_EVENT_TRANSFER_EXIT,
-                        "ERROR: Copy failed with mode %s, will delete destination and retry with the next available mode: %s",
-                        CopyModeStr[copy_mode], nested_error->message);
-                // Delete any potential destination file.
-                gfal_http_copy_cleanup(plugin_data, dst, &nested_error);
+            g_prefix_error(&nested_error, "ERROR: Copy failed with mode %s, with error: ",CopyModeStr[copy_mode]);
+            plugin_trigger_event(params, http_plugin_domain,
+               GFAL_EVENT_NONE, GFAL_EVENT_TRANSFER_EXIT,
+                nested_error->message);
+               // Delete any potential destination file.
+            gfal_http_copy_cleanup(plugin_data, dst, &nested_error);
         }
 
         copy_mode = (CopyMode)((int)copy_mode + 1);
-    }
+
+    } while ((copy_mode < end_copy_mode) && is_http_3rdcopy_fallback_enabled(context));
 
 
     plugin_trigger_event(params, http_plugin_domain,

src/plugins/http/gfal_http_plugin.cpp

@@ -527,8 +527,11 @@ static int davix2errno(StatusCode::Code code)
 
 void davix2gliberr(const DavixError* daverr, GError** err)
 {
+
+    std::string error_string= g_utf8_validate(daverr->getErrMsg().c_str(), daverr->getErrMsg().length(),NULL) ? 
+	daverr->getErrMsg().c_str(): "Error string contains not valid UTF-8 chars";
     gfal2_set_error(err, http_plugin_domain, davix2errno(daverr->getStatus()), __func__,
-              "%s", daverr->getErrMsg().c_str());
+              "%s", error_string.c_str());
 }
 
 

src/plugins/xrootd/gfal_xrootd_plugin_3rd_copy.cpp

@@ -230,6 +230,7 @@ int gfal_xrootd_3rd_copy_bulk(plugin_handle plugin_data,
         if ((source_url.GetProtocol() == "root") && (dest_url.GetProtocol() == "root")) {
             job.Set("thirdParty", "only");
             isThirdParty = true;
+            job.Set("delegate", true);
         }
         else {
             job.Set("thirdParty", "first");