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 167 of file XrdClHttpUtil.hh.

Constructor & Destructor Documentation

HandlerQueue()

HandlerQueue::HandlerQueue

(

unsigned

max_pending_ops

)

Definition at line 559 of file XrdClHttpUtil.cc.

                                                   :
    m_max_pending_ops(max_pending_ops)
{
    int filedes[2];
    auto result = pipe(filedes);
    if (result == -1) {
        throw std::runtime_error(strerror(errno));
    }
    if (fcntl(filedes[0], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1 || fcntl(filedes[1], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1) {
        close(filedes[0]);
        close(filedes[1]);
        throw std::runtime_error(strerror(errno));
    }
    m_read_fd = filedes[0];
    m_write_fd = filedes[1];
};

References close.

Member Function Documentation

Consume()

std::shared_ptr< CurlOperation > HandlerQueue::Consume

(

std::chrono::steady_clock::duration

dur

)

Definition at line 792 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::runtime_error(strerror(errno));
        }
        break;
    }
 
    lk.unlock();
    m_producer_cv.notify_one();
    m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
 
    return result;
}

References read().

Here is the call graph for this function:

Expire()

void HandlerQueue::Expire

(

)

Definition at line 691 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()

static unsigned XrdClHttp::HandlerQueue::GetDefaultMaxPendingOps ( )

inlinestatic

Definition at line 194 of file XrdClHttpUtil.hh.

{return m_default_max_pending_ops;}

GetHandle()

CURL * HandlerQueue::GetHandle

(

)

Definition at line 675 of file XrdClHttpUtil.cc.

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

References XrdClHttp::GetHandle().

Here is the call graph for this function:

GetMonitoringJson()

std::string HandlerQueue::GetMonitoringJson ( )

static

Definition at line 827 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 176 of file XrdClHttpUtil.hh.

{return m_read_fd;}

Produce()

void HandlerQueue::Produce

(

std::shared_ptr< CurlOperation >

handler

)

Definition at line 754 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::runtime_error(strerror(errno));
        }
        break;
    }
 
    lk.unlock();
    m_consumer_cv.notify_one();
    m_ops_produced.fetch_add(1, std::memory_order_relaxed);
}

References XrdCl::errOperationExpired, and write.

RecycleHandle()

void HandlerQueue::RecycleHandle

(

CURL *

curl

)

Definition at line 686 of file XrdClHttpUtil.cc.

                                      {
    m_handles.push_back(curl);
}

ReleaseHandles()

void HandlerQueue::ReleaseHandles

(

)

Definition at line 883 of file XrdClHttpUtil.cc.

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

Shutdown()

void HandlerQueue::Shutdown

(

)

Definition at line 875 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 840 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::runtime_error(strerror(errno));
        }   
        break;
    }
 
    lk.unlock();
    m_producer_cv.notify_one();
    m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
 
    return result;
}

References read().

Here is the call graph for this function:

---

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

• XrdClHttpUtil.hh
• XrdClHttpUtil.cc