XrdClHttp::HandlerQueue Class Reference

#include <XrdClHttpUtil.hh>

Collaboration diagram for XrdClHttp::HandlerQueue:

Public Member Functions
	HandlerQueue (unsigned max_pending_ops)
std::shared_ptr< CurlOperation >	Consume (std::chrono::steady_clock::duration)
void	Expire ()
CURL *	GetHandle ()
int	PollFD () const
void	Produce (std::shared_ptr< CurlOperation > handler)
void	RecycleHandle (CURL *)
void	ReleaseHandles ()
void	Shutdown ()
std::shared_ptr< CurlOperation >	TryConsume ()

Static Public Member Functions
static unsigned	GetDefaultMaxPendingOps ()
static std::string	GetMonitoringJson ()

Detailed Description

HandlerQueue is a deque of curl operations that need to be performed. The object is thread safe and can be waited on via poll().

The fact that it's poll'able is necessary because the multi-curl driver thread is based on polling FD's

Definition at line 191 of file XrdClHttpUtil.hh.

Constructor & Destructor Documentation

HandlerQueue()

HandlerQueue::HandlerQueue

(

unsigned

max_pending_ops

)

Definition at line 684 of file XrdClHttpUtil.cc.

                                                   :
    m_max_pending_ops(max_pending_ops)
{
    int filedes[2];
    auto result = pipe(filedes);
    if (result == -1) {
        throw std::system_error(errno, std::generic_category(),
                                "failed to create HTTP worker pipe");
    }
    if (fcntl(filedes[0], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1 || fcntl(filedes[1], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1) {
        const int error = errno;
        close(filedes[0]);
        close(filedes[1]);
        throw std::system_error(error, std::generic_category(),
                                "failed to configure HTTP worker pipe");
    }
    m_read_fd = filedes[0];
    m_write_fd = filedes[1];
};

References close.

Referenced by XrdClHttp::CurlWorker::Run().

Here is the caller graph for this function:

Member Function Documentation

Consume()

std::shared_ptr< CurlOperation > HandlerQueue::Consume

(

std::chrono::steady_clock::duration

dur

)

Definition at line 926 of file XrdClHttpUtil.cc.

{
    std::unique_lock<std::mutex> lk(m_mutex);
    m_consumer_cv.wait_for(lk, dur, [&]{return m_ops.size() > 0 || m_shutdown;});
    if (m_shutdown || m_ops.empty()) {
        return {};
    }
 
    std::shared_ptr<CurlOperation> result = m_ops.front();
    m_ops.pop_front();
 
    char ready[1];
    while (true) {
        auto result = read(m_read_fd, ready, 1);
        if (result == -1) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // This should never happen, but if it does, just continue
                // as if we successfully read the byte.
                break;
            }
            throw std::system_error(errno, std::generic_category(),
                                    "failed to consume HTTP worker notification");
        }
        break;
    }
 
    lk.unlock();
    m_producer_cv.notify_one();
    m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
 
    return result;
}

References read.

Expire()

void HandlerQueue::Expire

(

)

Definition at line 824 of file XrdClHttpUtil.cc.

{
    std::unique_lock<std::mutex> lk(m_mutex);
    auto now = std::chrono::steady_clock::now();
 
    // Iterate through the paused transfers, checking if they are done.
    for (auto &op : m_ops) {
        if (!op->IsPaused()) continue;
 
        if (op->TransferStalled(0, now)) {
            op->ContinueHandle();
        }
    }
 
    std::vector<decltype(m_ops)::value_type> expired_ops;
    unsigned expired_count = 0;
    auto it = std::remove_if(m_ops.begin(), m_ops.end(),
        [&](const std::shared_ptr<CurlOperation> &handler) {
            auto expired = handler->GetOperationExpiry() < now;
            if (expired) {
                expired_ops.push_back(handler);
                expired_count++;
            }
            return expired;
        });
    m_ops.erase(it, m_ops.end());
 
    // The contents of our pipe and the in-memory queue are now off by expired_count.
    // Read exactly that many bytes from the pipe and throw them away.
    char throwaway[64];
    unsigned bytes_to_read = expired_count;
    while (bytes_to_read > 0) {
        size_t chunk = std::min<size_t>(sizeof(throwaway), bytes_to_read);
        ssize_t n = read(m_read_fd, throwaway, chunk);
        if (n > 0) {
            bytes_to_read -= n;
        } else if (n == -1) {
            if (errno == EINTR) {
                continue;
            } else {
                // EWOULDBLOCK is a possibility if there's a synchronization error;
                // for now, just continue on as if we were successful in reading out
                // the missing bytes
                break;
            }
        } else {
            break;
        }
    }
 
    // Note: the failure handler may trigger new operations submitted to the queue
    // (which requires the lock to be held) such as a prefetch operation that gets split
    // into multiple sub-operations.
    //
    // Thus, we must unlock the mutex protecting the queue and avoid touching the shared state of
    // m_ops.
    lk.unlock();
    for (auto &handler : expired_ops) {
        if (handler) handler->Fail(XrdCl::errOperationExpired, 0, "Operation expired while in queue");
    }
}

GetDefaultMaxPendingOps()

unsigned XrdClHttp::HandlerQueue::GetDefaultMaxPendingOps ( )

inlinestatic

Definition at line 218 of file XrdClHttpUtil.hh.

{return m_default_max_pending_ops;}

GetHandle()

CURL * HandlerQueue::GetHandle

(

)

Definition at line 808 of file XrdClHttpUtil.cc.

                        {
    if (m_handles.size()) {
        auto result = m_handles.back();
        m_handles.pop_back();
        return result;
    }
 
    return ::GetHandle(EnableCurlHeaderDump());
}

GetMonitoringJson()

std::string HandlerQueue::GetMonitoringJson ( )

static

Definition at line 962 of file XrdClHttpUtil.cc.

{
    auto consumed = m_ops_consumed.load(std::memory_order_relaxed);
    auto produced = m_ops_produced.load(std::memory_order_relaxed);
    return "{"
            "\"produced\":" + std::to_string(produced) + ","
            "\"consumed\":" + std::to_string(consumed) + ","
            "\"pending\":" + std::to_string(produced - consumed) + ","
            "\"rejected\":" + std::to_string(m_ops_rejected.load(std::memory_order_relaxed)) +
        "}";
}

PollFD()

int XrdClHttp::HandlerQueue::PollFD ( ) const

inline

Definition at line 200 of file XrdClHttpUtil.hh.

{return m_read_fd;}

Produce()

void HandlerQueue::Produce

(

std::shared_ptr< CurlOperation >

handler

)

Definition at line 887 of file XrdClHttpUtil.cc.

{
    auto handler_expiry = handler->GetOperationExpiry();
    std::unique_lock<std::mutex> lk{m_mutex};
    m_producer_cv.wait_until(lk,
        handler_expiry,
        [&]{return m_ops.size() < m_max_pending_ops;}
    );
    if (std::chrono::steady_clock::now() > handler_expiry) {
        lk.unlock();
        handler->Fail(XrdCl::errOperationExpired, 0, "Operation expired while waiting for worker");
        m_ops_rejected.fetch_add(1, std::memory_order_relaxed);
        return;
    }
 
    m_ops.push_back(handler);
    char ready[] = "1";
    while (true) {
        auto result = write(m_write_fd, ready, 1);
        if (result == -1) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // This should never happen, but if it does, just continue
                // as if we successfully wrote the notification to the pipe.
                break;
            }
            throw std::system_error(errno, std::generic_category(),
                                    "failed to notify HTTP worker");
        }
        break;
    }
 
    lk.unlock();
    m_consumer_cv.notify_one();
    m_ops_produced.fetch_add(1, std::memory_order_relaxed);
}

References XrdCl::errOperationExpired, and write.

Referenced by XrdClHttp::CurlReadOp::Continue(), and XrdClHttp::stat_size().

Here is the caller graph for this function:

RecycleHandle()

void HandlerQueue::RecycleHandle

(

CURL *

curl

)

Definition at line 819 of file XrdClHttpUtil.cc.

                                      {
    m_handles.push_back(curl);
}

ReleaseHandles()

void HandlerQueue::ReleaseHandles

(

)

Definition at line 1019 of file XrdClHttpUtil.cc.

{
    for (auto handle : m_handles) {
        curl_easy_cleanup(handle);
    }
    m_handles.clear();
}

Shutdown()

void HandlerQueue::Shutdown

(

)

Definition at line 1011 of file XrdClHttpUtil.cc.

{
    std::unique_lock lock(m_mutex);
    m_shutdown = true;
    m_consumer_cv.notify_all();
}

TryConsume()

std::shared_ptr< CurlOperation > HandlerQueue::TryConsume

(

)

Definition at line 975 of file XrdClHttpUtil.cc.

{
    std::unique_lock<std::mutex> lk(m_mutex);
    if (m_ops.size() == 0) {
        std::shared_ptr<CurlOperation> result;
        return result;
    }
 
    std::shared_ptr<CurlOperation> result = m_ops.front();
    m_ops.pop_front();
 
    char ready[1];
    while (true) {
        auto result = read(m_read_fd, ready, 1); 
        if (result == -1) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // This should never happen, but if it does, just continue
                // as if we successfully read the byte.
                break;
            }
            throw std::system_error(errno, std::generic_category(),
                                    "failed to consume HTTP worker notification");
        }   
        break;
    }
 
    lk.unlock();
    m_producer_cv.notify_one();
    m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
 
    return result;
}

References read.

---

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

• XrdClHttpUtil.hh
• XrdClHttpUtil.cc