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 1027 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().

Here is the call graph for this function:

CurlWorker() [2/2]

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

delete

Member Function Documentation

ClientX509CertKeyFile()

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

(

)

const

Definition at line 1051 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 1057 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 1226 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;
 
    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 &entry : m_op_map) {
            OpRecord(*entry.second.first, OpKind::Update);
            if (entry.second.second < oldest_op) {
                oldest_op = entry.second.second;
            }
        }
        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);
                op->ReleaseHandle();
                if (curl) {
                    curl_multi_remove_handle(multi_handle, curl);
                    curl_easy_cleanup(curl);
                    m_op_map.erase(curl);
                }
                running_handles -= 1;
                continue;
            } else {
                auto iter = m_op_map.find(curl);
                if (iter != m_op_map.end()) iter->second.second = 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");
                    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");
                    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");
                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()};
 
            // 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()) {
                std::string modified_url;
                std::shared_ptr<CurlOptionsOp> options_op(
                    new CurlOptionsOp(
                        curl, op,
                        std::string(
                            VerbsCache::GetUrlKey(op->GetUrl(), modified_url)
                        ),
                        m_logger, op->GetConnCalloutFunc()
                    )
                );
                // Note this `curl` variable is not local to the conditional; it is the curl handle of the
                // CurlOptionsOp and will be added below to the multi-handle, causing it - not the parent's
                // curl handle - to be executed.
                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");
                    OpRecord(*op, OpKind::Error);
                    continue;
                }
                auto rv = options_op->Setup(curl, *this);
                if (!rv) {
                    m_logger->Debug(kLogXrdClHttp, "Failed to allocate a curl handle for OPTIONS");
                    continue;
                }
                m_op_map[curl] = {options_op, std::chrono::system_clock::now()};
                OpRecord(*options_op, OpKind::Start);
                running_handles += 1;
            } else {
                OpRecord(*op, OpKind::Start);
            }
 
            auto mres = curl_multi_add_handle(multi_handle, curl);
            if (mres != CURLM_OK) {
                m_logger->Debug(kLogXrdClHttp, "Unable to add operation to the curl multi-handle");
                op->Fail(XrdCl::errInternal, mres, "Unable to add operation to the curl multi-handle");
                OpRecord(*op, OpKind::Error);
                continue;
            }
            m_logger->Debug(kLogXrdClHttp, "Added request for URL %s to worker thread for processing", op->GetUrl().c_str());
            running_handles += 1;
        }
 
        // 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 &entry : broker_reqs) {
                if (entry.second.expiry < now) {
                    expired_ops.emplace_back(entry.first, entry.second.curl);
                }
            }
            for (const auto &entry : expired_ops) {
                auto iter = m_op_map.find(entry.second);
                if (iter == m_op_map.end()) {
                    m_logger->Warning(kLogXrdClHttp, "Found an expired curl handle with no corresponding operation!");
                } else {
 
                    CurlOptionsOp *options_op = nullptr;
                    if ((options_op = dynamic_cast<CurlOptionsOp*>(iter->second.first.get())) != nullptr) {
                        auto parent_op = options_op->GetOperation();
                        bool parent_op_failed = false;
                        if (parent_op->IsRedirect()) {
                            std::string target;
                            if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
                                auto iter = m_op_map.find(options_op->GetParentCurlHandle());
                                if (iter != m_op_map.end()) {
                                    OpRecord(*iter->second.first, OpKind::Error);
                                    iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
                                    m_op_map.erase(iter);
                                    running_handles -= 1;
                                }
                                parent_op_failed = true;
                            } else {
                                OpRecord(*parent_op, OpKind::Start);
                            }
                        } else {
                            OpRecord(*parent_op, OpKind::Start);
                        }
                        if (!parent_op_failed){
                            curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                        }
                    }
 
                    iter->second.first->Fail(XrdCl::errConnectionError, 1, "Timeout: connection never provided for request");
                    iter->second.first->ReleaseHandle();
                    OpRecord(*(iter->second.first), OpKind::ConncallTimeout);
                    m_op_map.erase(entry.second);
                    curl_easy_cleanup(entry.second);
                    running_handles -= 1;
                }
                broker_reqs.erase(entry.first);
                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 &entry : broker_reqs) {
            waitfds[idx].fd = entry.first;
            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 handle = broker_reqs[entry.fd].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 result = iter->second.first->WaitSocketCallback(err);
            if (result == -1) {
                m_logger->Warning(kLogXrdClHttp, "Error when invoking the broker callback: %s", err.c_str());
 
                CurlOptionsOp *options_op = nullptr;
                if ((options_op = dynamic_cast<CurlOptionsOp*>(iter->second.first.get())) != nullptr) {
                    auto parent_op = options_op->GetOperation();
                    bool parent_op_failed = false;
                    if (parent_op->IsRedirect()) {
                        std::string target;
                        if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
                            auto iter = m_op_map.find(options_op->GetParentCurlHandle());
                            if (iter != m_op_map.end()) {
                                OpRecord(*iter->second.first, OpKind::Error);
                                iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
                                m_op_map.erase(iter);
                                running_handles -= 1;
                            }
                            parent_op_failed = true;
                        } else {
                            OpRecord(*parent_op, OpKind::Start);
                        }
                    } else {
                        OpRecord(*parent_op, OpKind::Start);
                    }
                    if (!parent_op_failed){
                        curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                    }
                }
 
                iter->second.first->Fail(XrdCl::errErrorResponse, 1, err);
                OpRecord(*iter->second.first, OpKind::Error);
                m_op_map.erase(handle);
                broker_reqs.erase(entry.fd);
                m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
                running_handles -= 1;
            } else {
                broker_reqs.erase(entry.fd);
                curl_multi_add_handle(multi_handle, handle);
                m_conncall_success.fetch_add(1, std::memory_order_relaxed);
            }
        }
 
        // Do maintenance on the multi-handle
        int still_running;
        auto mres = curl_multi_perform(multi_handle, &still_running);
        if (mres == CURLM_CALL_MULTI_PERFORM) {
            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;
                }
                auto iter = m_op_map.find(msg->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 = iter->second.first;
                auto res = msg->data.result;
                bool keep_handle = false;
                bool waiting_on_callout = false;
                if (res == CURLE_OK) {
                    auto sc = op->GetStatusCode();
                    OpRecord(*op, OpKind::Finish);
                    if (HTTPStatusIsError(sc)) {
                        auto httpErr = HTTPStatusConvert(sc);
                        op->Fail(httpErr.first, httpErr.second, 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.
                        CurlOptionsOp *options_op = nullptr;
                        if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
                            auto parent_op = options_op->GetOperation();
                            bool parent_op_failed = false;
                            if (parent_op->IsRedirect()) {
                                std::string target;
                                if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
                                    OpRecord(*parent_op, OpKind::Error);
                                    m_op_map.erase(options_op->GetParentCurlHandle());
                                    running_handles -= 1;
                                    parent_op_failed = true;
                                } else {
                                    OpRecord(*parent_op, OpKind::Start);
                                }
                            } else {
                                OpRecord(*parent_op, OpKind::Start);
                            }
                            // Have curl execute the parent operation
                            if (!parent_op_failed) {
                                curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                            }
                        }
                        // The curl operation was successful, it's just the HTTP request failed; recycle the handle.
                        queue.RecycleHandle(iter->first);
                    } else {
                        CurlOptionsOp *options_op = nullptr;
                        // If this was a successful OPTIONS op, invoke the parent operation.
                        if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get()))) {
                            options_op->Success();
                            options_op->ReleaseHandle();
                            // Note: op is scoped external to the conditional block
                            op = options_op->GetOperation();
                            op->OptionsDone();
                            OpRecord(*op, OpKind::Start);
                            curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                            curl_multi_remove_handle(multi_handle, iter->first);
                            queue.RecycleHandle(iter->first);
                        }
                        // Check to see if the operation ended in a redirect (note: this might)
                        // be invoked a second time if this was the parent operation of an OPTIONS
                        // op.
                        if (op->IsRedirect()) {
                            std::string target;
                            switch (op->Redirect(target)) {
                                case CurlOperation::RedirectAction::Fail:
                                    if (options_op) {
                                        // In this case, we failed immediately after an OPTIONS finished.
                                        // Since there's a Start recorded after the OPTIONS processing, we
                                        // must record an error.
                                        // In the non-OPTIONS case, we never recorded a second start and
                                        // don't need a matching failure.
                                        OpRecord(*op, OpKind::Error);
                                    }
                                    keep_handle = false;
                                    break;
                                case CurlOperation::RedirectAction::Reinvoke:
                                    if (!options_op) {
                                        // In this case, the redirect occurred without any prior
                                        // OPTIONS call.  This implies that `op` is the original call
                                        // and we need to restart it later and record another op start.
                                        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.
                                    std::string modified_url;
                                    target = VerbsCache::GetUrlKey(target, modified_url);
                                    options_op = new CurlOptionsOp(iter->first, op, target, m_logger, op->GetConnCalloutFunc());
                                    std::shared_ptr<CurlOperation> new_op(options_op);
                                    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");
                                        keep_handle = false;
                                        options_op = nullptr;
                                        break;
                                    }
                                    OpRecord(*new_op, OpKind::Start);
                                    try {
                                        auto rv = new_op->Setup(curl, *this);
                                        if (!rv) {
                                            m_logger->Debug(kLogXrdClHttp,  "Unable to configure a curl handle for OPTIONS");
                                            keep_handle = false;
                                            options_op = nullptr;
                                            break;
                                        }
                                    } catch (...) {
                                        m_logger->Debug(kLogXrdClHttp, "Unable to setup the curl handle for the OPTIONS operation");
                                        new_op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the OPTIONS operation");
                                        OpRecord(*new_op, OpKind::Error);
                                        keep_handle = false;
                                        break;
                                    }
                                    new_op->SetContinueQueue(m_continue_queue);
                                    m_op_map[curl] = {new_op, std::chrono::system_clock::now()};
                                    auto mres = curl_multi_add_handle(multi_handle, curl);
                                    if (mres != CURLM_OK) {
                                        m_logger->Debug(kLogXrdClHttp, "Unable to add OPTIONS operation to the curl multi-handle: %s", curl_multi_strerror(mres));
                                        op->Fail(XrdCl::errInternal, mres, "Unable to add OPTIONS operation to the curl multi-handle");
                                        OpRecord(*new_op, OpKind::Error);
                                        break;
                                    }
                                    running_handles += 1;
                                    m_logger->Debug(kLogXrdClHttp, "Invoking the OPTIONS operation before redirect to %s", target.c_str());
                                    // The original curl operation needs to be kept around.  Note that because options_op
                                    // is non-nil, we won't re-add the handle to the multi-handle.
                                    keep_handle = true;
                                }
                            }
                            int callout_socket = op->WaitSocket();
                            if ((waiting_on_callout = 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] = {iter->first, expiry};
                                m_conncall_req.fetch_add(1, std::memory_order_relaxed);
                            }
                        } else if (options_op) {
                            // In this case, the OPTIONS call happened before the parent operation was started.
                            curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                        }
                        if (keep_handle) {
                            curl_multi_remove_handle(multi_handle, iter->first);
                            if (!waiting_on_callout && !options_op) {
                                curl_multi_add_handle(multi_handle, iter->first);
                            }
                        } else if (!options_op) {
                            // A multi-step operation may reset and reconfigure its
                            // easy handle from Success().  Remove the completed
                            // request before invoking it so libcurl no longer owns
                            // the request configuration being replaced.
                            curl_multi_remove_handle(multi_handle, iter->first);
                            op->Success();
                            if (op->IsDone()) {
                                op->ReleaseHandle();
                                // If the handle was successful, then we can recycle it.
                                queue.RecycleHandle(iter->first);
                            } 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, iter->first);
                                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(iter->first);
                                }
                            }
                        }
                    }
                } 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());
                        op->ReleaseHandle();
                        keep_handle = false;
                    } else {
                        curl_multi_remove_handle(multi_handle, iter->first);
                        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] = {iter->first, 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;
                        };
                        CurlOptionsOp *options_op = nullptr;
                        if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
                            auto parent_op = options_op->GetOperation();
                            bool parent_op_failed = false;
                            if (parent_op->IsRedirect()) {
                                std::string target;
                                if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
                                    auto iter = m_op_map.find(options_op->GetParentCurlHandle());
                                    if (iter != m_op_map.end()) {
                                        OpRecord(*iter->second.first, OpKind::Error);
                                        iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
                                        m_op_map.erase(iter);
                                        running_handles -= 1;
                                    }
                                    parent_op_failed = true;
                                } else {
                                    OpRecord(*parent_op, OpKind::Start);
                                }
                            } else {
                                OpRecord(*parent_op, OpKind::Start);
                            }
                            if (!parent_op_failed){
                                curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                            }
                        }
                    } else {
                        auto xrdCode = 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(xrdCode.first, xrdCode.second, fail_err);
                        OpRecord(*op, OpKind::Error);
                        CurlOptionsOp *options_op = nullptr;
                        if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
                            auto parent_op = options_op->GetOperation();
                            bool parent_op_failed = false;
                            if (parent_op->IsRedirect()) {
                                std::string target;
                                if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
                                    auto iter = m_op_map.find(options_op->GetParentCurlHandle());
                                    if (iter != m_op_map.end()) {
                                        OpRecord(*iter->second.first, OpKind::Error);
                                        iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
                                        m_op_map.erase(iter);
                                        running_handles -= 1;
                                    }
                                    parent_op_failed = true;
                                }
                            }
                            if (!parent_op_failed){
                                curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
                            }
                        }
                    }
                    op->ReleaseHandle();
                }
                if (!keep_handle) {
                    curl_multi_remove_handle(multi_handle, iter->first);
                    if (res != CURLE_OK) {
                        curl_easy_cleanup(iter->first);
                    }
                    for (auto &req : broker_reqs) {
                        if (req.second.curl == iter->first) {
                            m_logger->Warning(kLogXrdClHttp, "Curl handle finished while a broker operation was outstanding");
                            m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
                        }
                    }
                    m_op_map.erase(iter);
                    running_handles -= 1;
                }
            }
        } while (msg);
    }
 
    for (auto map_entry : m_op_map) {
        if (mres) {
            map_entry.second.first->Fail(XrdCl::errInternal, mres, curl_multi_strerror(mres));
            OpRecord(*map_entry.second.first, OpKind::Error);
        }
        if (multi_handle && map_entry.first) curl_multi_remove_handle(multi_handle, map_entry.first);
    }
 
    m_queue->ReleaseHandles();
    curl_multi_cleanup(multi_handle);
}

References XrdClHttp::CurlOperation::CleanupDnsCache(), XrdClHttp::CurlOperation::ContinueHandle(), CurlCodeConvert(), XrdCl::Log::Debug(), emsg(), XrdCl::errConnectionError, XrdCl::errErrorResponse, XrdCl::errInternal, XrdCl::errOperationExpired, XrdCl::Log::Error(), Error, XrdClHttp::CurlOperation::Fail(), XrdClHttp::CurlOperation::FinishSetup(), XrdClHttp::CurlOperation::GetCallbackError(), XrdClHttp::CurlOperation::GetConnCalloutFunc(), XrdClHttp::CurlOperation::GetCurlErrorMessage(), XrdClHttp::CurlOperation::GetCurlHandle(), XrdCl::DefaultEnv::GetEnv(), XrdClHttp::CurlOperation::GetError(), XrdCl::Env::GetInt(), XrdClHttp::CurlOptionsOp::GetOperation(), XrdClHttp::CurlOptionsOp::GetParentCurlHandle(), XrdClHttp::CurlOperation::GetStatusCode(), XrdClHttp::CurlOperation::GetStatusMessage(), XrdClHttp::CurlOperation::GetTriedBoker(), XrdClHttp::CurlOperation::GetUrl(), XrdClHttp::VerbsCache::GetUrlKey(), XrdClHttp::HTTPStatusConvert(), XrdClHttp::HTTPStatusIsError(), XrdClHttp::CurlOperation::IsDone(), XrdClHttp::CurlOperation::IsPaused(), XrdClHttp::CurlOperation::IsRedirect(), XrdClHttp::kLogXrdClHttp, XrdClHttp::CurlOperation::OptionsDone(), XrdClHttp::CurlOperation::Redirect(), XrdClHttp::CurlOperation::ReleaseHandle(), XrdClHttp::CurlOptionsOp::ReleaseHandle(), XrdClHttp::CurlOperation::RequiresOptions(), XrdClHttp::CurlOperation::SetContinueQueue(), XrdClHttp::CurlOperation::SetTriedBoker(), XrdClHttp::CurlOperation::Setup(), XrdClHttp::CurlOperation::StartConnectionCallout(), XrdClHttp::CurlOptionsOp::Success(), XrdClHttp::CurlOperation::Success(), XrdClHttp::CurlOperation::UseConnectionCallout(), XrdClHttp::CurlOperation::WaitSocket(), and XrdCl::Log::Warning().

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 1207 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 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);
    }

Start()

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

Definition at line 1194 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();
}

---

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

• XrdClHttpWorker.hh
• XrdClHttpUtil.cc