Merge pull request #4926 from eisenhauer/get-query-routing

Various HTTP/XROOTD changes

Author: eisenhauer

source/adios2/engine/campaign/CampaignReader.cpp

@@ -121,7 +121,7 @@ void CampaignReader::PerformGets()
         return true;
     };
 
-    size_t nThreads = std::min(nEngines, (size_t)16);
+    size_t nThreads = std::min(nEngines, (size_t)64);
     std::vector<std::future<bool>> futures(nThreads - 1);
 
     // launch Threads-1 threads to process subsets of handles,

source/adios2/toolkit/remote/XrootdHttpRemote.cpp

@@ -7,9 +7,22 @@
 #include "XrootdHttpRemote.h"
 #include "adios2/helper/adiosLog.h"
 
+#include <chrono>
+#include <cstdlib>
 #include <cstring>
 #include <sstream>
 #include <stdexcept>
+#include <thread>
+
+#ifdef _WIN32
+#include <direct.h>
+#include <stdlib.h>
+#define getcwd _getcwd
+#define PATH_MAX _MAX_PATH
+#else
+#include <climits>
+#include <unistd.h>
+#endif
 
 #ifdef ADIOS2_HAVE_CURL
 #include <curl/curl.h>
@@ -219,9 +232,18 @@ void CurlMultiPool::WorkerLoop()
             continue;
         }
 
-        // Wait for socket activity (up to 100ms)
+        // Skip waiting if new requests are queued -- submit them immediately
+        {
+            std::lock_guard<std::mutex> lock(m_QueueMutex);
+            if (!m_PendingQueue.empty())
+            {
+                continue;
+            }
+        }
+
+        // Wait for socket activity (up to 10ms)
         int numfds;
-        CURLMcode mc = curl_multi_wait(m_MultiHandle, nullptr, 0, 100, &numfds);
+        CURLMcode mc = curl_multi_wait(m_MultiHandle, nullptr, 0, 10, &numfds);
         if (mc != CURLM_OK)
         {
             helper::Log("Remote", "CurlMultiPool", "WorkerLoop",
@@ -269,7 +291,23 @@ void XrootdHttpRemote::Open(const std::string hostname, const int32_t port,
                             const std::string filename, const Mode mode, bool RowMajorOrdering,
                             const Params &params)
 {
-    m_Filename = filename;
+    // Ensure filename is absolute so the URL path is clean (no double-slash ambiguity)
+    if (!filename.empty() && filename[0] != '/')
+    {
+        char cwd[PATH_MAX];
+        if (getcwd(cwd, sizeof(cwd)))
+        {
+            m_Filename = std::string(cwd) + "/" + filename;
+        }
+        else
+        {
+            m_Filename = filename;
+        }
+    }
+    else
+    {
+        m_Filename = filename;
+    }
     m_Mode = mode;
     m_RowMajorOrdering = RowMajorOrdering;
 
@@ -310,7 +348,8 @@ void XrootdHttpRemote::Open(const std::string hostname, const int32_t port,
     }
 
     std::ostringstream urlStream;
-    urlStream << (m_UseHttps ? "https" : "http") << "://" << hostname << ":" << port << "/ssi";
+    urlStream << (m_UseHttps ? "https" : "http") << "://" << hostname << ":" << port << "/adios/"
+              << UrlEncode(m_Filename);
     m_BaseUrl = urlStream.str();
 
     m_OpenSuccess = true;
@@ -323,26 +362,47 @@ std::string XrootdHttpRemote::BuildRequestString(const char *VarName, size_t Ste
                                                  const Dims &Start, const Accuracy &accuracy)
 {
     std::ostringstream reqStream;
-    std::string encodedFilename = UrlEncode(m_Filename);
     std::string encodedVarName = UrlEncode(std::string(VarName));
 
-    reqStream << "get Filename=" << encodedFilename;
-    reqStream << "&RMOrder=" << (m_RowMajorOrdering ? 1 : 0);
-    reqStream << "&Varname=" << encodedVarName;
-    reqStream << "&StepStart=" << Step;
-    reqStream << "&StepCount=" << StepCount;
-    reqStream << "&Block=" << BlockID;
-    reqStream << "&Dims=" << Count.size();
-
-    for (const auto &c : Count)
-        reqStream << "&Count=" << c;
-    for (const auto &s : Start)
-        reqStream << "&Start=" << s;
-
-    reqStream << "&AccuracyError=" << accuracy.error;
-    reqStream << "&AccuracyNorm=" << accuracy.norm;
-    reqStream << "&AccuracyRelative=" << (accuracy.relative ? 1 : 0);
-
+    reqStream << "get&Varname=" << encodedVarName;
+    if (m_RowMajorOrdering)
+    {
+        reqStream << "&RMOrder=1";
+    }
+    if (Step != 0)
+    {
+        reqStream << "&StepStart=" << Step;
+    }
+    if (StepCount != 1)
+    {
+        reqStream << "&StepCount=" << StepCount;
+    }
+    if (BlockID != static_cast<size_t>(-1))
+    {
+        reqStream << "&Block=" << BlockID;
+    }
+    if (!Count.empty())
+    {
+        reqStream << "&Count=" << Count[0];
+        for (size_t i = 1; i < Count.size(); i++)
+        {
+            reqStream << "," << Count[i];
+        }
+    }
+    if (!Start.empty())
+    {
+        reqStream << "&Start=" << Start[0];
+        for (size_t i = 1; i < Start.size(); i++)
+        {
+            reqStream << "," << Start[i];
+        }
+    }
+    if (accuracy.error != 0.0 || accuracy.norm != 0.0 || accuracy.relative)
+    {
+        reqStream << "&AccuracyError=" << accuracy.error;
+        reqStream << "&AccuracyNorm=" << accuracy.norm;
+        reqStream << "&AccuracyRelative=" << (accuracy.relative ? 1 : 0);
+    }
     if (!m_EngineParams.empty())
     {
         reqStream << "&EngineParams=" << UrlEncode(m_EngineParams);
@@ -352,7 +412,7 @@ std::string XrootdHttpRemote::BuildRequestString(const char *VarName, size_t Ste
 }
 
 CURL *XrootdHttpRemote::CreateEasyHandle(AsyncGet *asyncOp, const std::string &url,
-                                         const std::string &postData)
+                                         const std::string &queryData)
 {
 #ifdef ADIOS2_HAVE_CURL
     CURL *easy = curl_easy_init();
@@ -362,8 +422,9 @@ CURL *XrootdHttpRemote::CreateEasyHandle(AsyncGet *asyncOp, const std::string &u
     curl_easy_setopt(easy, CURLOPT_PRIVATE, asyncOp);
     asyncOp->easyHandle = easy;
 
-    curl_easy_setopt(easy, CURLOPT_URL, url.c_str());
-    curl_easy_setopt(easy, CURLOPT_COPYPOSTFIELDS, postData.c_str());
+    std::string fullUrl = url + "?" + queryData;
+    curl_easy_setopt(easy, CURLOPT_URL, fullUrl.c_str());
+    curl_easy_setopt(easy, CURLOPT_HTTPGET, 1L);
     curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, WriteCallback);
     curl_easy_setopt(easy, CURLOPT_WRITEDATA, &asyncOp->responseData);
     curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT, m_ConnectTimeout);
@@ -385,9 +446,6 @@ CURL *XrootdHttpRemote::CreateEasyHandle(AsyncGet *asyncOp, const std::string &u
         }
     }
 
-    asyncOp->headers =
-        curl_slist_append(nullptr, "Content-Type: application/x-www-form-urlencoded");
-    curl_easy_setopt(easy, CURLOPT_HTTPHEADER, asyncOp->headers);
     curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
 
     return easy;
@@ -399,11 +457,12 @@ CURL *XrootdHttpRemote::CreateEasyHandle(AsyncGet *asyncOp, const std::string &u
 std::string XrootdHttpRemote::BuildBatchRequestString(const std::vector<BatchGetRequest> &requests)
 {
     std::ostringstream reqStream;
-    std::string encodedFilename = UrlEncode(m_Filename);
 
-    reqStream << "batchget Filename=" << encodedFilename;
-    reqStream << "&RMOrder=" << (m_RowMajorOrdering ? 1 : 0);
-    reqStream << "&NVars=" << requests.size();
+    reqStream << "batchget&NVars=" << requests.size();
+    if (m_RowMajorOrdering)
+    {
+        reqStream << "&RMOrder=1";
+    }
     if (!m_EngineParams.empty())
     {
         reqStream << "&EngineParams=" << UrlEncode(m_EngineParams);
@@ -412,18 +471,41 @@ std::string XrootdHttpRemote::BuildBatchRequestString(const std::vector<BatchGet
     for (const auto &req : requests)
     {
         std::string encodedVarName = UrlEncode(std::string(req.VarName));
-        reqStream << "|Varname=" << encodedVarName;
-        reqStream << "&StepStart=" << req.Step;
-        reqStream << "&StepCount=" << req.StepCount;
-        reqStream << "&Block=" << req.BlockID;
-        reqStream << "&Dims=" << req.Count.size();
-        for (const auto &c : req.Count)
-            reqStream << "&Count=" << c;
-        for (const auto &s : req.Start)
-            reqStream << "&Start=" << s;
-        reqStream << "&AccuracyError=" << req.accuracy.error;
-        reqStream << "&AccuracyNorm=" << req.accuracy.norm;
-        reqStream << "&AccuracyRelative=" << (req.accuracy.relative ? 1 : 0);
+        reqStream << "&Varname=" << encodedVarName;
+        if (req.Step != 0)
+        {
+            reqStream << "&StepStart=" << req.Step;
+        }
+        if (req.StepCount != 1)
+        {
+            reqStream << "&StepCount=" << req.StepCount;
+        }
+        if (req.BlockID != static_cast<size_t>(-1))
+        {
+            reqStream << "&Block=" << req.BlockID;
+        }
+        if (!req.Count.empty())
+        {
+            reqStream << "&Count=" << req.Count[0];
+            for (size_t i = 1; i < req.Count.size(); i++)
+            {
+                reqStream << "," << req.Count[i];
+            }
+        }
+        if (!req.Start.empty())
+        {
+            reqStream << "&Start=" << req.Start[0];
+            for (size_t i = 1; i < req.Start.size(); i++)
+            {
+                reqStream << "," << req.Start[i];
+            }
+        }
+        if (req.accuracy.error != 0.0 || req.accuracy.norm != 0.0 || req.accuracy.relative)
+        {
+            reqStream << "&AccuracyError=" << req.accuracy.error;
+            reqStream << "&AccuracyNorm=" << req.accuracy.norm;
+            reqStream << "&AccuracyRelative=" << (req.accuracy.relative ? 1 : 0);
+        }
     }
 
     return reqStream.str();
@@ -489,30 +571,58 @@ bool XrootdHttpRemote::BatchGet(const std::vector<BatchGetRequest> &requests)
         subBatches.push_back({startIdx, count, asyncOp});
     }
 
-    // Wait for all sub-batches and parse responses
+    // Wait for all sub-batches, retry transient failures
+    const int maxRetries = 3;
     bool allOk = true;
     for (auto &sb : subBatches)
     {
         bool success = sb.asyncOp->promise.get_future().get();
+        std::string lastErr;
 
-        if (!success || !allOk)
+        if (!success)
         {
-            if (!success && allOk)
+            lastErr = sb.asyncOp->errorMsg;
+            delete sb.asyncOp;
+            sb.asyncOp = nullptr;
+
+            // Retry this sub-batch
+            for (int retry = 1; retry <= maxRetries; retry++)
             {
-                std::string errMsg = sb.asyncOp->errorMsg;
-                // Clean up remaining sub-batches before throwing
-                delete sb.asyncOp;
-                for (size_t j = (&sb - &subBatches[0]) + 1; j < subBatches.size(); j++)
+                helper::Log("Remote", "XrootdHttpRemote", "BatchGet",
+                            "Sub-batch failed (" + lastErr + "), retry " + std::to_string(retry) +
+                                "/" + std::to_string(maxRetries),
+                            helper::LogMode::WARNING);
+
+                std::this_thread::sleep_for(std::chrono::milliseconds(100 * retry));
+
+                std::vector<BatchGetRequest> subRequests(requests.begin() + sb.startIdx,
+                                                         requests.begin() + sb.startIdx + sb.count);
+                std::string postData = BuildBatchRequestString(subRequests);
+                AsyncGet *retryOp = new AsyncGet();
+                CURL *easy = CreateEasyHandle(retryOp, m_BaseUrl, postData);
+                if (!easy)
                 {
-                    subBatches[j].asyncOp->promise.get_future().get();
-                    delete subBatches[j].asyncOp;
+                    delete retryOp;
+                    continue;
                 }
+                pool.Submit(easy, retryOp);
+                success = retryOp->promise.get_future().get();
+                if (success)
+                {
+                    sb.asyncOp = retryOp;
+                    break;
+                }
+                lastErr = retryOp->errorMsg;
+                delete retryOp;
+            }
+
+            if (!success)
+            {
                 helper::Throw<std::runtime_error>("Remote", "XrootdHttpRemote", "BatchGet",
-                                                  "Sub-batch failed: " + errMsg);
+                                                  "Sub-batch failed after " +
+                                                      std::to_string(maxRetries) +
+                                                      " retries: " + lastErr);
             }
-            allOk = false;
-            delete sb.asyncOp;
-            continue;
         }
 
         // Parse binary response: