XrdClHttp::CurlWorker Class Reference

#include <XrdClHttpWorker.hh>

Collaboration diagram for XrdClHttp::CurlWorker:

Public Member Functions
	CurlWorker (const CurlWorker &)=delete
	CurlWorker (std::shared_ptr< HandlerQueue > queue, VerbsCache &cache, XrdCl::Log *logger)
std::tuple< std::string, std::string >	ClientX509CertKeyFile () const
void	Run ()
void	Start (std::unique_ptr< XrdClHttp::CurlWorker > self, std::thread tid)

Static Public Member Functions
static std::string	GetMonitoringJson ()
static void	RunStatic (CurlWorker *myself)
static void	SetMaintenancePeriod (unsigned maint)

Detailed Description

Definition at line 52 of file XrdClHttpWorker.hh.

Constructor & Destructor Documentation

CurlWorker() [1/2]

CurlWorker::CurlWorker	(	std::shared_ptr< HandlerQueue >	queue,
		VerbsCache &	cache,
		XrdCl::Log *	logger )

Definition at line 1040 of file XrdClHttpUtil.cc.

                                                                                             :
    m_cache(cache),
    m_queue(queue),
    m_logger(logger)
{
    {
        std::unique_lock lk(m_worker_stats_mutex);
        m_stats_offset = m_workers_last_completed_cycle.size();
        m_workers_last_completed_cycle.push_back(&m_last_completed_cycle);
        m_workers_oldest_op.push_back(&m_oldest_op);
    }
    int pipeInfo[2];
    if ((pipe(pipeInfo) == -1) || (fcntl(pipeInfo[0], F_SETFD, FD_CLOEXEC)) || (fcntl(pipeInfo[1], F_SETFD, FD_CLOEXEC))) {
        throw std::runtime_error("Failed to create shutdown monitoring pipe for curl worker");
    }
    m_shutdown_pipe_r = pipeInfo[0];
    m_shutdown_pipe_w = pipeInfo[1];
 
    // Handle setup of the X509 authentication
    auto env = XrdCl::DefaultEnv::GetEnv();
    env->GetString("HttpClientCertFile", m_x509_client_cert_file);
    env->GetString("HttpClientKeyFile", m_x509_client_key_file);
}

References XrdCl::DefaultEnv::GetEnv().

Referenced by CurlWorker(), and RunStatic().

Here is the call graph for this function:

Here is the caller graph for this function:

CurlWorker() [2/2]

XrdClHttp::CurlWorker::CurlWorker ( const CurlWorker & )

delete

References CurlWorker(), and Start().

Here is the call graph for this function:

Member Function Documentation

ClientX509CertKeyFile()

std::tuple< std::string, std::string > CurlWorker::ClientX509CertKeyFile

(

)

const

Definition at line 1064 of file XrdClHttpUtil.cc.

{
    return std::make_tuple(m_x509_client_cert_file, m_x509_client_key_file);
}

Referenced by XrdClHttp::CurlOperation::Setup().

Here is the caller graph for this function:

GetMonitoringJson()

std::string CurlWorker::GetMonitoringJson ( )

static

Definition at line 1070 of file XrdClHttpUtil.cc.

{
    auto now = std::chrono::system_clock::now().time_since_epoch().count();
    auto oldest_op = now;
    auto oldest_cycle = now;
    {
        std::unique_lock lk(m_worker_stats_mutex);
        for (const auto &entry : m_workers_last_completed_cycle) {
            if (!entry) {continue;}
            auto cycle = entry->load(std::memory_order_relaxed);
            if (cycle < oldest_cycle) oldest_cycle = cycle;
        }
        for (const auto &entry : m_workers_oldest_op) {
            if (!entry) {continue;}
            auto op = entry->load(std::memory_order_relaxed);
            if (op < oldest_op) oldest_op = op;
        }
    }
    auto oldest_op_dbl = std::chrono::duration<double>(std::chrono::system_clock::time_point(std::chrono::system_clock::duration(oldest_op)).time_since_epoch()).count();
    auto oldest_cycle_dbl = std::chrono::duration<double>(std::chrono::system_clock::time_point(std::chrono::system_clock::duration(oldest_cycle)).time_since_epoch()).count();
    std::string retval = "{"
        "\"oldest_op\":" + std::to_string(oldest_op_dbl) + ","
        "\"oldest_cycle\":" + std::to_string(oldest_cycle_dbl) + ","
    ;
 
    for (size_t verb_idx = 0; verb_idx < static_cast<int>(XrdClHttp::CurlOperation::HttpVerb::Count); verb_idx++) {
        const auto &verb_str = XrdClHttp::CurlOperation::GetVerbString(static_cast<XrdClHttp::CurlOperation::HttpVerb>(verb_idx));
        for (size_t op_idx = 0; op_idx < 402; op_idx++) {
            if (op_idx == 401) continue;
 
            auto &op_stats = m_ops[verb_idx][op_idx];
            auto duration = op_stats.m_duration.load(std::memory_order_relaxed);
            if (duration == 0) continue;
 
            std::string prefix = "http_" + verb_str + "_" + ((op_idx == 402) ? "invalid" : std::to_string(200 + op_idx)) + "_";
 
            auto duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
            retval += "\"" + prefix + "duration\":" + std::to_string(duration_dbl) + ",";
 
            duration = op_stats.m_pause_duration.load(std::memory_order_relaxed);
            if (duration > 0) {
                duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
                retval += "\"" + prefix + "pause_duration\":" + std::to_string(duration_dbl) + ",";
            }
 
            auto count = op_stats.m_bytes.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "bytes\":" + std::to_string(count) + ",";
            count = op_stats.m_error.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "error\":" + std::to_string(count) + ",";
            count = op_stats.m_finished.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "finished\":" + std::to_string(count) + ",";
            count = op_stats.m_client_timeout.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "client_timeout\":" + std::to_string(count) + ",";
            count = op_stats.m_server_timeout.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "server_timeout\":" + std::to_string(count) + ",";
        }
        {
            auto &op_stats = m_ops[verb_idx][401];
            auto duration = op_stats.m_duration.load(std::memory_order_relaxed);
            if (duration == 0) continue;
 
            std::string prefix = "http_" + verb_str + "_";
 
            auto duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
            retval += "\"" + prefix + "preheader_duration\":" + std::to_string(duration_dbl) + ",";
 
            auto count = op_stats.m_started.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "started\":" + std::to_string(count) + ",";
            count = op_stats.m_error.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "preheader_error\":" + std::to_string(count) + ",";
            count = op_stats.m_finished.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "preheader_finished\":" + std::to_string(count) + ",";
            count = op_stats.m_server_timeout.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "preheader_timeout\":" + std::to_string(count) + ",";
            count = op_stats.m_conncall_timeout.load(std::memory_order_relaxed);
            if (count) retval += "\"" + prefix + "conncall_timeout\":" + std::to_string(count) + ",";
        }
    }
 
    retval +=
        "\"conncall_error\":" + std::to_string(m_conncall_errors.load(std::memory_order_relaxed)) + ","
        "\"conncall_started\":" + std::to_string(m_conncall_req.load(std::memory_order_relaxed)) + ","
        "\"conncall_success\":" + std::to_string(m_conncall_success.load(std::memory_order_relaxed)) + ","
        "\"conncall_timeout\":" + std::to_string(m_conncall_timeout.load(std::memory_order_relaxed)) +
        "}";
 
    return retval;
}

References XrdClHttp::CurlOperation::GetVerbString().

Here is the call graph for this function:

Run()

void CurlWorker::Run

(

)

Definition at line 1239 of file XrdClHttpUtil.cc.

                {
    int max_pending = 50;
    XrdCl::DefaultEnv::GetEnv()->GetInt("HttpMaxPendingOps", max_pending);
    m_continue_queue.reset(new HandlerQueue(max_pending));
    auto &queue = *m_queue.get();
    m_logger->Debug(kLogXrdClHttp, "Started a curl worker");
 
    CURLM *multi_handle = curl_multi_init();
    if (multi_handle == nullptr) {
        throw std::runtime_error("Failed to create curl multi-handle");
    }
 
    int running_handles = 0;
    time_t last_maintenance = time(NULL);
    CURLMcode mres = CURLM_OK;
 
    // Map from a file descriptor that has an outstanding broker request
    // to the corresponding CURL handle.
    std::unordered_map<int, WaitingForBroker> broker_reqs;
    std::vector<struct curl_waitfd> waitfds;
 
    // Detach the handle before clearing the operation's curl callbacks.  The
    // caller may still hold the operation after this worker stops using it.
    auto release_handle = [&](const std::shared_ptr<CurlOperation> &op, CURL *handle) {
        if (!handle) return;
        curl_multi_remove_handle(multi_handle, handle);
        if (op->GetCurlHandle()) op->ReleaseHandle();
        curl_easy_cleanup(handle);
    };
    auto remove_op = [&](CURL *handle) {
        auto iter = m_op_map.find(handle);
        if (iter == m_op_map.end()) return;
        const auto &[op, started_at] = iter->second;
        release_handle(op, handle);
        m_op_map.erase(iter);
        running_handles -= 1;
        for (auto req = broker_reqs.begin(); req != broker_reqs.end();) {
            if (req->second.curl == handle) {
                req = broker_reqs.erase(req);
            } else {
                ++req;
            }
        }
    };
    auto start_options = [&](CURL *parent_handle, const std::shared_ptr<CurlOperation> &parent_op,
                             const std::string &target) {
        std::shared_ptr<CurlOptionsOp> options_op;
        CURL *options_handle = nullptr;
        try {
            std::string modified_url;
            options_op = std::make_shared<CurlOptionsOp>(parent_handle, parent_op,
                std::string(VerbsCache::GetUrlKey(target, modified_url)),
                m_logger, parent_op->GetConnCalloutFunc());
            options_handle = queue.GetHandle();
            if (!options_handle || !options_op->Setup(options_handle, *this)) {
                throw std::runtime_error("Failed to setup the curl handle for OPTIONS");
            }
        } catch (...) {
            m_logger->Debug(kLogXrdClHttp, "Failed to setup the curl handle for OPTIONS");
            if (options_op) {
                options_op->Fail(XrdCl::errInternal, ENOMEM,
                    "Failed to setup the curl handle for the OPTIONS operation");
                release_handle(options_op, options_handle);
            }
            parent_op->Fail(XrdCl::errInternal, ENOMEM,
                "Failed to setup the curl handle for the OPTIONS operation");
            OpRecord(*parent_op, OpKind::Error);
            return false;
        }
        m_op_map[options_handle] = {options_op, std::chrono::system_clock::now()};
        OpRecord(*options_op, OpKind::Start);
        running_handles += 1;
        auto options_res = curl_multi_add_handle(multi_handle, options_handle);
        if (options_res == CURLM_OK) return true;
        options_op->Fail(XrdCl::errInternal, options_res,
            "Unable to add OPTIONS operation to the curl multi-handle");
        OpRecord(*options_op, OpKind::Error);
        parent_op->Fail(XrdCl::errInternal, options_res,
            "Unable to add OPTIONS operation to the curl multi-handle");
        OpRecord(*parent_op, OpKind::Error);
        remove_op(options_handle);
        return false;
    };
    auto reactivate_parent = [&](const std::shared_ptr<CurlOptionsOp> &options_op) {
        auto parent_op = options_op->GetOperation();
        auto parent_handle = options_op->GetParentCurlHandle();
        if (parent_op->IsDone()) {
            remove_op(parent_handle);
            return false;
        }
        // OPTIONS is advisory, including for redirect targets.  Resolve a
        // pending redirect before adding the parent's handle back to libcurl.
        if (parent_op->IsRedirect()) {
            std::string target;
            auto action = parent_op->Redirect(target);
            if (action == CurlOperation::RedirectAction::Fail) {
                OpRecord(*parent_op, OpKind::Error);
                remove_op(parent_handle);
                return false;
            }
            if (action == CurlOperation::RedirectAction::ReinvokeAfterAllow) {
                if (start_options(parent_handle, parent_op, target)) return true;
                remove_op(parent_handle);
                return false;
            }
        }
        OpRecord(*parent_op, OpKind::Start);
        auto callout_socket = parent_op->WaitSocket();
        if (callout_socket >= 0) {
            broker_reqs[callout_socket] = {parent_handle, time(nullptr) + 20};
            m_conncall_req.fetch_add(1, std::memory_order_relaxed);
            return true;
        }
        auto parent_res = curl_multi_add_handle(multi_handle, parent_handle);
        if (parent_res == CURLM_OK) return true;
        m_logger->Debug(kLogXrdClHttp,
            "Unable to re-add the parent operation to the curl multi-handle: %s",
            curl_multi_strerror(parent_res));
        parent_op->Fail(XrdCl::errInternal, parent_res,
            "Unable to re-add the parent operation to the curl multi-handle");
        OpRecord(*parent_op, OpKind::Error);
        remove_op(parent_handle);
        return false;
    };
 
    bool want_shutdown = false;
    while (!want_shutdown) {
        m_last_completed_cycle.store(std::chrono::system_clock::now().time_since_epoch().count());
        auto oldest_op = std::chrono::system_clock::now();
        for (const auto &[handle, entry] : m_op_map) {
            const auto &[op, started_at] = entry;
            OpRecord(*op, OpKind::Update);
            if (started_at < oldest_op) {
                oldest_op = started_at;
            }
        }
        m_oldest_op.store(oldest_op.time_since_epoch().count());
 
        // Try continuing any available handles that have more data
        while (true) {
            auto op = m_continue_queue->TryConsume();
            if (!op) {
                break;
            }
            // Avoid race condition where external thread added a continue operation to queue
            // while the curl worker thread failed the transfer.
            if (op->IsDone()) {
                m_logger->Debug(kLogXrdClHttp, "Ignoring continuation of operation that has already completed");
                continue;
            }
            m_logger->Debug(kLogXrdClHttp, "Continuing the curl handle from op %p on thread %d", op.get(), getthreadid());
            auto curl = op->GetCurlHandle();
            if (!op->ContinueHandle()) {
                op->Fail(XrdCl::errInternal, 0, "Failed to continue the curl handle for the operation");
                OpRecord(*op, OpKind::Error);
                remove_op(curl);
                continue;
            } else {
                auto iter = m_op_map.find(curl);
                if (iter != m_op_map.end()) {
                    auto &[operation, started_at] = iter->second;
                    started_at = std::chrono::system_clock::now();
                }
            }
                }
        // Consume from the shared new operation queue
        while (running_handles < static_cast<int>(m_max_ops)) {
            auto op = running_handles == 0 ? queue.Consume(std::chrono::seconds(1)) : queue.TryConsume();
            if (!op) {
                break;
            }
            auto curl = queue.GetHandle();
            if (curl == nullptr) {
                m_logger->Debug(kLogXrdClHttp, "Unable to allocate a curl handle");
                op->Fail(XrdCl::errInternal, ENOMEM, "Unable to get allocate a curl handle");
                continue;
            }
            try {
                auto rv = op->Setup(curl, *this);
                if (!rv) {
                    m_logger->Debug(kLogXrdClHttp, "Failed to setup the curl handle");
                    op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the operation");
                    release_handle(op, curl);
                    continue;
                }
                if (!op->FinishSetup(curl)) {
                    m_logger->Debug(kLogXrdClHttp, "Failed to finish setup of the curl handle");
                    op->Fail(XrdCl::errInternal, ENOMEM, "Failed to finish setup of the curl handle for the operation");
                    release_handle(op, curl);
                    continue;
                }
            } catch (...) {
                m_logger->Debug(kLogXrdClHttp, "Unable to setup the curl handle");
                op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the operation");
                release_handle(op, curl);
                continue;
            }
            op->SetContinueQueue(m_continue_queue);
 
            if (op->IsDone()) {
                op->ReleaseHandle();
                queue.RecycleHandle(curl);
                continue;
            }
            m_op_map[curl] = {op, std::chrono::system_clock::now()};
            running_handles += 1;
 
            // If the operation requires the result of the OPTIONS verb to function, then
            // we add that to the multi-handle instead, chaining the two calls together.
            if (op->RequiresOptions()) {
                if (!start_options(curl, op, op->GetUrl())) {
                    remove_op(curl);
                    continue;
                }
            } else {
                OpRecord(*op, OpKind::Start);
                auto add_res = curl_multi_add_handle(multi_handle, curl);
                if (add_res != CURLM_OK) {
                    m_logger->Debug(kLogXrdClHttp, "Unable to add operation to the curl multi-handle");
                    op->Fail(XrdCl::errInternal, add_res, "Unable to add operation to the curl multi-handle");
                    OpRecord(*op, OpKind::Error);
                    remove_op(curl);
                    continue;
                }
            }
            m_logger->Debug(kLogXrdClHttp, "Added request for URL %s to worker thread for processing", op->GetUrl().c_str());
        }
 
        // Maintain the periodic reporting of thread activity and fail any operations
        // that have expired / timed out.
        time_t now = time(NULL);
        time_t next_maintenance = last_maintenance + m_maintenance_period.load(std::memory_order_relaxed);
        if (now >= next_maintenance) {
            m_queue->Expire();
            m_continue_queue->Expire();
            m_logger->Debug(kLogXrdClHttp, "Curl worker thread %d is running %d operations",
                getthreadid(), running_handles);
            last_maintenance = now;
 
            // Timeout all the pending broker requests.
            std::vector<std::pair<int, CURL *>> expired_ops;
            for (const auto &[fd, request] : broker_reqs) {
                if (request.expiry < now) {
                    expired_ops.emplace_back(fd, request.curl);
                }
            }
            for (const auto &[fd, handle] : expired_ops) {
                auto iter = m_op_map.find(handle);
                if (iter == m_op_map.end()) {
                    m_logger->Warning(kLogXrdClHttp, "Found an expired curl handle with no corresponding operation!");
                } else {
                    auto [op, started_at] = iter->second;
                    op->Fail(XrdCl::errConnectionError, 1, "Timeout: connection never provided for request");
                    OpRecord(*op, OpKind::ConncallTimeout);
                    if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                        reactivate_parent(options_op);
                    remove_op(handle);
                }
                broker_reqs.erase(fd);
                m_conncall_timeout.fetch_add(1, std::memory_order_relaxed);
            }
 
            // Cleanup the fake connection cache entries.
            XrdClHttp::CurlOperation::CleanupDnsCache();
        }
 
        waitfds.clear();
        waitfds.resize(3 + broker_reqs.size());
 
        waitfds[0].fd = queue.PollFD();
        waitfds[0].events = CURL_WAIT_POLLIN;
        waitfds[0].revents = 0;
        waitfds[1].fd = m_continue_queue->PollFD();
        waitfds[1].events = CURL_WAIT_POLLIN;
        waitfds[1].revents = 0;
        waitfds[2].fd = m_shutdown_pipe_r;
        waitfds[2].revents = 0;
        waitfds[2].events = CURL_WAIT_POLLIN | CURL_WAIT_POLLPRI;
 
        int idx = 3;
        for (const auto &[fd, request] : broker_reqs) {
            waitfds[idx].fd = fd;
            waitfds[idx].events = CURL_WAIT_POLLIN|CURL_WAIT_POLLPRI;
            waitfds[idx].revents = 0;
            idx += 1;
        }
 
        long timeo;
        curl_multi_timeout(multi_handle, &timeo);
        // These commented-out lines are purposely left; will need to revisit after the 0.9.1 release;
        // for now, they are too verbose on RHEL7.
        //m_logger->Debug(kLogXrdClHttp, "Curl advises a timeout of %ld ms", timeo);
        if (running_handles && timeo == -1) {
            // Bug workaround: we've seen RHEL7 libcurl have a race condition where it'll not
            // set a timeout while doing the DNS lookup; assume that if there are running handles
            // but no timeout, we've hit this bug.
            //m_logger->Debug(kLogXrdClHttp, "Will sleep for up to 50ms");
            mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), 50, nullptr);
        } else {
            //m_logger->Debug(kLogXrdClHttp, "Will sleep for up to %d seconds", max_sleep_time);
            //mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), max_sleep_time*1000, nullptr);
            // Temporary test: we've been seeing DNS lookups timeout on additional platforms.  Switch to always
            // poll as curl_multi_wait doesn't seem to get notified when DNS lookups are done.
            mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), 50, nullptr);
        }
        if (mres != CURLM_OK) {
            m_logger->Warning(kLogXrdClHttp, "Failed to wait on multi-handle: %d", mres);
        }
 
        // Iterate through the waiting broker callbacks.
        for (const auto &entry : waitfds) {
            // Ignore the queue's poll fd.
            if (waitfds[0].fd == entry.fd || waitfds[1].fd == entry.fd) {
                continue;
            }
            // Handle shutdown requests
            if ((waitfds[2].fd == entry.fd) && entry.revents) {
                want_shutdown = true;
                break;
            }
            if ((entry.revents & CURL_WAIT_POLLIN) != CURL_WAIT_POLLIN) {
                continue;
            }
            auto request = broker_reqs.find(entry.fd);
            if (request == broker_reqs.end()) continue;
            auto handle = request->second.curl;
            auto iter = m_op_map.find(handle);
            if (iter == m_op_map.end()) {
                m_logger->Warning(kLogXrdClHttp, "Internal error: broker responded on FD %d but no corresponding curl operation", entry.fd);
                broker_reqs.erase(entry.fd);
                m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
                continue;
            }
            std::string err;
            auto [op, started_at] = iter->second;
            auto result = op->WaitSocketCallback(err);
            if (result == -1) {
                m_logger->Warning(kLogXrdClHttp, "Error when invoking the broker callback: %s", err.c_str());
                op->Fail(XrdCl::errErrorResponse, 1, err);
                OpRecord(*op, OpKind::Error);
                if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                    reactivate_parent(options_op);
                remove_op(handle);
                broker_reqs.erase(entry.fd);
                m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
            } else {
                broker_reqs.erase(entry.fd);
                auto add_res = curl_multi_add_handle(multi_handle, handle);
                if (add_res != CURLM_OK) {
                    op->Fail(XrdCl::errInternal, add_res,
                        "Unable to add operation to the curl multi-handle after the broker callback");
                    OpRecord(*op, OpKind::Error);
                    if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                        reactivate_parent(options_op);
                    remove_op(handle);
                    m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
                    continue;
                }
                m_conncall_success.fetch_add(1, std::memory_order_relaxed);
            }
        }
 
        // Do maintenance on the multi-handle
        int still_running;
        mres = curl_multi_perform(multi_handle, &still_running);
        if (mres == CURLM_CALL_MULTI_PERFORM) {
            mres = CURLM_OK;
            continue;
        } else if (mres != CURLM_OK) {
            m_logger->Warning(kLogXrdClHttp, "Failed to perform multi-handle operation: %d", mres);
            break;
        }
 
        CURLMsg *msg;
        do {
            int msgq = 0;
            msg = curl_multi_info_read(multi_handle, &msgq);
            if (msg && (msg->msg == CURLMSG_DONE)) {
                if (!msg->easy_handle) {
                    m_logger->Warning(kLogXrdClHttp, "Logic error: got a callback for a null handle");
                    mres = CURLM_BAD_EASY_HANDLE;
                    break;
                }
                CURL *const easy_handle = msg->easy_handle;
                auto iter = m_op_map.find(easy_handle);
                if (iter == m_op_map.end()) {
                    m_logger->Error(kLogXrdClHttp, "Logic error: got a callback for an entry that doesn't exist");
                    mres = CURLM_BAD_EASY_HANDLE;
                    break;
                }
                auto [op, started_at] = iter->second;
                auto res = msg->data.result;
                // Completion callbacks may reconfigure or release their handle.
                // Stop libcurl from owning it before invoking operation methods.
                curl_multi_remove_handle(multi_handle, easy_handle);
                bool keep_handle = false;
                bool waiting_on_callout = false;
                bool options_pending = false;
                if (res == CURLE_OK) {
                    auto sc = op->GetStatusCode();
                    OpRecord(*op, OpKind::Finish);
                    if (HTTPStatusIsError(sc)) {
                        auto [code, number] = HTTPStatusConvert(sc);
                        op->Fail(code, number, op->GetStatusMessage());
                        op->ReleaseHandle();
                        // If this was a failed CurlOptionsOp, then we re-activate the parent handle.
                        // If the parent handle was stopped at a redirect that now returns failure, then
                        // we'll clean it up.
                        if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                            reactivate_parent(options_op);
                        // The curl operation was successful, it's just the HTTP request failed; recycle the handle.
                        queue.RecycleHandle(easy_handle);
                    } else {
                        // The child and suspended parent own distinct handles.
                        // Complete the child, then resume or fail the parent without
                        // treating the parent's redirect as a child request.
                        if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op)) {
                            options_op->Success();
                            options_op->ReleaseHandle();
                            if (!options_op->GetOperation()->IsDone()) {
                                options_op->GetOperation()->OptionsDone();
                            }
                            reactivate_parent(options_op);
                            queue.RecycleHandle(easy_handle);
                        } else {
                            if (op->IsRedirect()) {
                                std::string target;
                                switch (op->Redirect(target)) {
                                    case CurlOperation::RedirectAction::Fail:
                                        keep_handle = false;
                                        break;
                                    case CurlOperation::RedirectAction::Reinvoke:
                                        keep_handle = true;
                                        OpRecord(*op, OpKind::Start);
                                        break;
                                    case CurlOperation::RedirectAction::ReinvokeAfterAllow:
                                    {
                                        // The redirect resulted in a new endpoint where the cache lookup failed;
                                        // we need to know what HTTP verbs are in the server's Allow list before this
                                        // operation can continue.  Inject a new CurlOptionsOp and chain it to the one
                                        // being processed.  Once the OPTIONS request is done, then we'll restart this
                                        // operation.
                                        if (!start_options(easy_handle, op, target)) {
                                            keep_handle = false;
                                            break;
                                        }
                                        m_logger->Debug(kLogXrdClHttp, "Invoking the OPTIONS operation before redirect to %s", target.c_str());
                                        // The original curl operation needs to be kept around without re-adding its handle
                                        // to the multi-handle until the OPTIONS operation completes.
                                        options_pending = true;
                                        keep_handle = true;
                                    }
                                }
                                int callout_socket = op->WaitSocket();
                                if ((waiting_on_callout = keep_handle && !options_pending && callout_socket >= 0)) {
                                    auto expiry = time(nullptr) + 20;
                                    m_logger->Debug(kLogXrdClHttp, "Creating a callout wait request on socket %d", callout_socket);
                                    broker_reqs[callout_socket] = {easy_handle, expiry};
                                    m_conncall_req.fetch_add(1, std::memory_order_relaxed);
                                }
                            }
                            if (keep_handle) {
                                if (!waiting_on_callout && !options_pending) {
                                    auto redirect_res = curl_multi_add_handle(multi_handle, easy_handle);
                                    if (redirect_res != CURLM_OK) {
                                        op->Fail(XrdCl::errInternal, redirect_res,
                                            "Unable to add redirected operation to the curl multi-handle");
                                        OpRecord(*op, OpKind::Error);
                                        op->ReleaseHandle();
                                        queue.RecycleHandle(easy_handle);
                                        keep_handle = false;
                                    }
                                }
                            } else {
                                if (!op->IsDone()) {
                                    op->Success();
                                }
                                if (op->IsDone()) {
                                    op->ReleaseHandle();
                                    // If the handle was successful, then we can recycle it.
                                    queue.RecycleHandle(easy_handle);
                                } else {
                                    // Multi-step operations may configure their easy handle
                                    // for another request from Success().  Keep ownership of
                                    // the handle and run the next request through this worker's
                                    // multi-handle like any other operation.
                                    auto next_res = curl_multi_add_handle(multi_handle, easy_handle);
                                    if (next_res == CURLM_OK) {
                                        keep_handle = true;
                                        OpRecord(*op, OpKind::Start);
                                    } else {
                                        op->Fail(XrdCl::errInternal, next_res,
                                            "Unable to add the next operation request to the curl multi-handle");
                                        OpRecord(*op, OpKind::Error);
                                        op->ReleaseHandle();
                                        queue.RecycleHandle(easy_handle);
                                    }
                                }
                            }
                        }
                    }
                } else if (res == CURLE_COULDNT_CONNECT && op->UseConnectionCallout() && !op->GetTriedBoker()) {
                    // In this case, we need to use the broker and the curl handle couldn't reuse
                    // an existing socket.
                    keep_handle = true;
                    op->SetTriedBoker(); // Flag to ensure we try a connection only once per operation.
                    std::string err;
                    int wait_socket = -1;
                    if (!op->StartConnectionCallout(err) || (wait_socket=op->WaitSocket()) == -1) {
                        m_logger->Error(kLogXrdClHttp, "Failed to start broker-based connection: %s", err.c_str());
                        if (!op->IsDone())
                            op->Fail(XrdCl::errConnectionError, 1, err);
                        OpRecord(*op, OpKind::Error);
                        if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                            reactivate_parent(options_op);
                        op->ReleaseHandle();
                        keep_handle = false;
                    } else {
                        curl_multi_remove_handle(multi_handle, easy_handle);
                        auto expiry = time(nullptr) + 20;
                        m_logger->Debug(kLogXrdClHttp, "Curl operation requires a new TCP socket; waiting for callout to respond on socket %d", wait_socket);
                        broker_reqs[wait_socket] = {easy_handle, expiry};
                        m_conncall_req.fetch_add(1, std::memory_order_relaxed);
                    }
                } else {
                    if (res == CURLE_ABORTED_BY_CALLBACK || res == CURLE_WRITE_ERROR) {
                        // We cannot invoke the failure from within a callback as the curl thread and
                        // original thread of execution may fight over the ownership of the handle memory.
                        switch (op->GetError()) {
                        case CurlOperation::OpError::ErrHeaderTimeout:
#ifdef HAVE_XPROTOCOL_TIMEREXPIRED
                            op->Fail(XrdCl::errOperationExpired, 0, "Origin did not respond with headers within timeout");
#else
                            op->Fail(XrdCl::errOperationExpired, 0, "Origin did not respond within timeout");
#endif
                            OpRecord(*op, OpKind::Error);
                            break;
                        case CurlOperation::OpError::ErrCallback: {
                            auto [ecode, emsg] = op->GetCallbackError();
                            op->Fail(XrdCl::errErrorResponse, ecode, emsg);
                            OpRecord(*op, OpKind::Error);
                            break;
                        }
                        case CurlOperation::OpError::ErrOperationTimeout:
                            op->Fail(XrdCl::errOperationExpired, 0, "Operation timed out");
                            OpRecord(*op, op->IsPaused() ? OpKind::ClientTimeout : OpKind::ServerTimeout);
                            break;
                        case CurlOperation::OpError::ErrTransferSlow:
                            op->Fail(XrdCl::errOperationExpired, 0, "Transfer speed below minimum threshold");
                            OpRecord(*op, OpKind::ServerTimeout);
                            break;
                        case CurlOperation::OpError::ErrTransferClientStall:
                            op->Fail(XrdCl::errOperationExpired, 0, "Transfer stalled for too long");
                            OpRecord(*op, OpKind::ClientTimeout);
                            break;
                        case CurlOperation::OpError::ErrTransferStall:
                            op->Fail(XrdCl::errOperationExpired, 0, "Transfer stalled for too long");
                            OpRecord(*op, OpKind::ServerTimeout);
                            break;
                        case CurlOperation::OpError::ErrNone:
                            op->Fail(XrdCl::errInternal, 0, "Operation was aborted without recording an abort reason");
                            OpRecord(*op, OpKind::Error);
                            break;
                        };
                        if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                            reactivate_parent(options_op);
                    } else {
                        auto [code, number] = CurlCodeConvert(res);
                        const auto curl_err = op->GetCurlErrorMessage();
                        const char *curl_easy_err = curl_easy_strerror(res);
                        const std::string fail_err = !curl_err.empty() ? curl_err : curl_easy_err;
                        m_logger->Debug(kLogXrdClHttp, "Curl generated an error: %s (%d)", fail_err.c_str(), res);
                        op->Fail(code, number, fail_err);
                        OpRecord(*op, OpKind::Error);
                        if (auto options_op = std::dynamic_pointer_cast<CurlOptionsOp>(op))
                            reactivate_parent(options_op);
                    }
                    op->ReleaseHandle();
                }
                if (!keep_handle) {
                    curl_multi_remove_handle(multi_handle, easy_handle);
                    if (res != CURLE_OK) {
                        curl_easy_cleanup(easy_handle);
                    }
                    for (auto req = broker_reqs.begin(); req != broker_reqs.end();) {
                        if (req->second.curl == easy_handle) {
                            m_logger->Warning(kLogXrdClHttp, "Curl handle finished while a broker operation was outstanding");
                            m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
                            req = broker_reqs.erase(req);
                        } else {
                            ++req;
                        }
                    }
                    m_op_map.erase(easy_handle);
                    running_handles -= 1;
                }
            }
        } while (msg);
        if (mres != CURLM_OK) break;
    }
 
    for (const auto &[handle, entry] : m_op_map) {
        const auto &[op, started_at] = entry;
        if (!op->IsDone()) {
            const auto err_num = mres == CURLM_OK ? ECANCELED : static_cast<int>(mres);
            const std::string err_msg = mres == CURLM_OK
                ? "Curl worker shut down before the operation completed"
                : curl_multi_strerror(mres);
            op->Fail(XrdCl::errInternal, err_num, err_msg);
            OpRecord(*op, OpKind::Error);
        }
        release_handle(op, handle);
    }
    m_op_map.clear();
 
    m_queue->ReleaseHandles();
    curl_multi_cleanup(multi_handle);
}

References XrdClHttp::CurlOperation::CleanupDnsCache(), CurlCodeConvert(), emsg(), XrdClHttp::CurlOperation::ErrCallback, XrdCl::errConnectionError, XrdCl::errErrorResponse, XrdClHttp::CurlOperation::ErrHeaderTimeout, XrdCl::errInternal, XrdClHttp::CurlOperation::ErrNone, XrdCl::errOperationExpired, XrdClHttp::CurlOperation::ErrOperationTimeout, XrdClHttp::CurlOperation::ErrTransferClientStall, XrdClHttp::CurlOperation::ErrTransferSlow, XrdClHttp::CurlOperation::ErrTransferStall, XrdClHttp::CurlOperation::Fail, XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdClHttp::VerbsCache::GetUrlKey(), XrdClHttp::HTTPStatusConvert(), XrdClHttp::HTTPStatusIsError(), XrdClHttp::kLogXrdClHttp, XrdClHttp::CurlOperation::Reinvoke, and XrdClHttp::CurlOperation::ReinvokeAfterAllow.

Referenced by RunStatic().

Here is the call graph for this function:

Here is the caller graph for this function:

RunStatic()

void CurlWorker::RunStatic ( CurlWorker * myself )

static

Definition at line 1220 of file XrdClHttpUtil.cc.

{
    {
        std::unique_lock lock(myself->m_start_lock);
        myself->m_start_complete_cv.wait(lock, [&]{return myself->m_start_complete;});
    }
    try {
        myself->Run();
    } catch (...) {
        myself->m_logger->Warning(kLogXrdClHttp, "Curl worker got an exception");
        {
            std::unique_lock lock(m_workers_mutex);
            auto iter = std::remove_if(m_workers.begin(), m_workers.end(), [&](std::unique_ptr<XrdClHttp::CurlWorker> &worker){return worker.get() == myself;});
            m_workers.erase(iter);
        }
    }
}

References CurlWorker(), XrdClHttp::kLogXrdClHttp, Run(), and XrdCl::Log::Warning().

Here is the call graph for this function:

SetMaintenancePeriod()

static void XrdClHttp::CurlWorker::SetMaintenancePeriod ( unsigned maint )

inlinestatic

Definition at line 70 of file XrdClHttpWorker.hh.

                                                     {
        m_maintenance_period.store(maint, std::memory_order_relaxed);
    }

Referenced by XrdClHttp::File::SetProperty().

Here is the caller graph for this function:

Start()

void CurlWorker::Start	(	std::unique_ptr< XrdClHttp::CurlWorker >	self,
		std::thread	tid )

Definition at line 1207 of file XrdClHttpUtil.cc.

{
    {
        std::unique_lock lock(m_workers_mutex);
        m_workers.emplace_back(std::move(self));
        m_self_tid = std::move(tid);
    }
    std::unique_lock lock(m_start_lock);
    m_start_complete = true;
    m_start_complete_cv.notify_one();
}

References Start().

Referenced by CurlWorker(), and Start().

Here is the call graph for this function:

Here is the caller graph for this function:

---

The documentation for this class was generated from the following files:

• XrdClHttpWorker.hh
• XrdClHttpUtil.cc