Various fixes related to overlapping and collapsed entry caching.

[thirdparty/squid.git] / src / CollapsedForwarding.cc

/*
 * DEBUG: section 17    Request Forwarding
 *
 */

#include "squid.h"
#include "ipc/mem/Segment.h"
#include "ipc/Messages.h"
#include "ipc/Port.h"
#include "ipc/TypedMsgHdr.h"
#include "CollapsedForwarding.h"
#include "globals.h"
#include "SquidConfig.h"
#include "Store.h"
#include "store_key_md5.h"
#include "tools.h"

/// shared memory segment path to use for CollapsedForwarding queue
static const char *const ShmLabel = "cf";
/// a single worker-to-worker queue capacity
// TODO: make configurable or compute from squid.conf settings if possible
static const int QueueCapacity = 1024;

std::auto_ptr<CollapsedForwarding::Queue> CollapsedForwarding::queue;

/// IPC queue message
class CollapsedForwardingMsg
{
public:
    CollapsedForwardingMsg(): sender(-1) { key[0] = key[1] = 0; }

public:
    int sender; /// kid ID of sending process
    uint64_t key[2]; ///< StoreEntry key
};

// CollapsedForwarding

void
CollapsedForwarding::Init()
{
    Must(!queue.get());
    if (UsingSmp() && IamWorkerProcess())
        queue.reset(new Queue(ShmLabel, KidIdentifier));
}

void
CollapsedForwarding::Broadcast(const StoreEntry &e)
{
    if (!queue.get())
        return;

    CollapsedForwardingMsg msg;
    msg.sender = KidIdentifier;
    memcpy(msg.key, e.key, sizeof(msg.key));

    debugs(17, 5, storeKeyText(static_cast<cache_key*>(e.key)) << " to " <<
           Config.workers << "-1 workers");

    // TODO: send only to workers who are waiting for data
    for (int workerId = 1; workerId <= Config.workers; ++workerId) {
        try {
            if (workerId != KidIdentifier && queue->push(workerId, msg))
                Notify(workerId);
        } catch (const Queue::Full &) {
            debugs(17, DBG_IMPORTANT, "ERROR: Collapsed forwarding " <<
                   "queue overflow for kid" << workerId <<
                   " at " << queue->outSize(workerId) << " items");
            // TODO: grow queue size
        }
    }
}

void
CollapsedForwarding::Notify(const int workerId)
{
    // TODO: Count and report the total number of notifications, pops, pushes.
    debugs(17, 7, "to kid" << workerId);
    Ipc::TypedMsgHdr msg;
    // TODO: add proper message type?
    msg.setType(Ipc::mtCollapsedForwardingNotification);
    msg.putInt(KidIdentifier);
    const String addr = Ipc::Port::MakeAddr(Ipc::strandAddrPfx, workerId);
    Ipc::SendMessage(addr, msg);
}

void
CollapsedForwarding::HandleNewData(const char *const when)
{
    debugs(17, 4, "popping all " << when);
    CollapsedForwardingMsg msg;
    int workerId;
    int poppedCount = 0;
    while (queue->pop(workerId, msg)) {
        debugs(17, 3, "message from kid" << workerId);
        if (workerId != msg.sender) {
            debugs(17, DBG_IMPORTANT, "mismatching kid IDs: " << workerId <<
                   " != " << msg.sender);
        }

        const cache_key *key = reinterpret_cast<const cache_key*>(msg.key);
        debugs(17, 7, "hadling " << storeKeyText(key));
        Store::Root().syncCollapsed(key);
        debugs(17, 7, "handled " << storeKeyText(key));

        // XXX: stop and schedule an async call to continue
        assert(++poppedCount < SQUID_MAXFD);
    }
}

void
CollapsedForwarding::HandleNotification(const Ipc::TypedMsgHdr &msg)
{
    const int from = msg.getInt();
    debugs(17, 7, HERE << "from " << from);
    assert(queue.get());
    queue->clearReaderSignal(from);
    HandleNewData("after notification");
}

/// initializes shared queue used by CollapsedForwarding
class CollapsedForwardingRr: public Ipc::Mem::RegisteredRunner
{
public:
    /* RegisteredRunner API */
    CollapsedForwardingRr(): owner(NULL) {}
    virtual ~CollapsedForwardingRr();

protected:
    virtual void create(const RunnerRegistry &);
    virtual void open(const RunnerRegistry &);

private:
    Ipc::MultiQueue::Owner *owner;
};

RunnerRegistrationEntry(rrAfterConfig, CollapsedForwardingRr);

void CollapsedForwardingRr::create(const RunnerRegistry &)
{
    Must(!owner);
    owner = Ipc::MultiQueue::Init(ShmLabel, Config.workers, 1,
                                  sizeof(CollapsedForwardingMsg),
                                  QueueCapacity);
}

void CollapsedForwardingRr::open(const RunnerRegistry &)
{
    CollapsedForwarding::Init();
}

CollapsedForwardingRr::~CollapsedForwardingRr()
{
    delete owner;
}