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

/*
 * Copyright (C) 1996-2025 The Squid Software Foundation and contributors
 *
 * Squid software is distributed under GPLv2+ license and includes
 * contributions from numerous individuals and organizations.
 * Please see the COPYING and CONTRIBUTORS files for details.
 */

/* DEBUG: section 77    Delay Pools */

/**
 \defgroup DelayPoolsInternal Delay Pools Internal
 \ingroup DelayPoolsAPI
 */

#include "squid.h"

#if USE_DELAY_POOLS
#include "client_side_request.h"
#include "comm/Connection.h"
#include "CommonPool.h"
#include "CompositePoolNode.h"
#include "ConfigParser.h"
#include "DelayBucket.h"
#include "DelayId.h"
#include "DelayPool.h"
#include "DelayPools.h"
#include "DelaySpec.h"
#include "DelayTagged.h"
#include "DelayUser.h"
#include "DelayVector.h"
#include "event.h"
#include "http/Stream.h"
#include "ip/Address.h"
#include "MemObject.h"
#include "mgr/Registration.h"
#include "NullDelayId.h"
#include "sbuf/SBuf.h"
#include "Store.h"
#include "StoreClient.h"

/// \ingroup DelayPoolsInternal
class Aggregate : public CompositePoolNode
{
    MEMPROXY_CLASS(Aggregate);

public:
    typedef RefCount<Aggregate> Pointer;
    Aggregate();
    ~Aggregate() override;
    virtual DelaySpec *rate() {return &spec;}

    virtual DelaySpec const *rate() const {return &spec;}

    void stats(StoreEntry * sentry) override;
    void dump(StoreEntry *entry) const override;
    void update(int incr) override;
    void parse() override;

    DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;

private:

    /// \ingroup DelayPoolsInternal
    class AggregateId:public DelayIdComposite
    {
        MEMPROXY_CLASS(Aggregate::AggregateId);

    public:
        AggregateId (RefCount<Aggregate>);
        int bytesWanted (int min, int max) const override;
        void bytesIn(int qty) override;
        void delayRead(const AsyncCallPointer &) override;

    private:
        RefCount<Aggregate> theAggregate;
    };

    friend class AggregateId;

    DelayBucket theBucket;
    DelaySpec spec;
};

/// \ingroup DelayPoolsInternal
template <class Key, class Value>
class VectorMap
{

public:
    VectorMap();
    unsigned int size() const;
    unsigned char findKeyIndex (Key const key) const;
    bool indexUsed (unsigned char const index) const;
    unsigned int insert (Key const key);

#define IND_MAP_SZ 256

    Key key_map[IND_MAP_SZ];
    Value values[IND_MAP_SZ];

private:
    unsigned int nextMapPosition;
};

/// \ingroup DelayPoolsInternal
class VectorPool : public CompositePoolNode
{
    MEMPROXY_CLASS(VectorPool);

public:
    typedef RefCount<VectorPool> Pointer;
    void dump(StoreEntry *entry) const override;
    void parse() override;
    void update(int incr) override;
    void stats(StoreEntry * sentry) override;

    DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;
    VectorMap<unsigned char, DelayBucket> buckets;
    VectorPool();
    ~VectorPool() override;

protected:
    bool keyAllocated (unsigned char const key) const;
    virtual DelaySpec *rate() {return &spec;}

    virtual DelaySpec const *rate() const {return &spec;}

    virtual char const *label() const = 0;

    virtual unsigned int makeKey(Ip::Address &src_addr) const = 0;

    DelaySpec spec;

    /// \ingroup DelayPoolsInternal
    class Id:public DelayIdComposite
    {
        MEMPROXY_CLASS(VectorPool::Id);

    public:
        Id (RefCount<VectorPool>, int);
        int bytesWanted (int min, int max) const override;
        void bytesIn(int qty) override;

    private:
        RefCount<VectorPool> theVector;
        int theIndex;
    };
};

/// \ingroup DelayPoolsInternal
class IndividualPool : public VectorPool
{
    MEMPROXY_CLASS(IndividualPool);

protected:
    char const *label() const override {return "Individual";}
    unsigned int makeKey(Ip::Address &src_addr) const override;
};

/// \ingroup DelayPoolsInternal
class ClassCNetPool : public VectorPool
{
    MEMPROXY_CLASS(ClassCNetPool);

protected:
    char const *label() const override {return "Network";}
    unsigned int makeKey (Ip::Address &src_addr) const override;
};

/* don't use remote storage for these */
/// \ingroup DelayPoolsInternal
class ClassCBucket
{

public:
    bool individualUsed (unsigned int index)const;
    unsigned char findHostMapPosition (unsigned char const host) const;
    bool individualAllocated (unsigned char host) const;
    unsigned char hostPosition (DelaySpec &rate, unsigned char const host);
    void initHostIndex (DelaySpec &rate, unsigned char index, unsigned char host);
    void update (DelaySpec const &, int incr);
    void stats(StoreEntry *)const;

    DelayBucket net;
    VectorMap<unsigned char, DelayBucket> individuals;
};

/// \ingroup DelayPoolsInternal
class ClassCHostPool : public CompositePoolNode
{
    MEMPROXY_CLASS(ClassCHostPool);

public:
    typedef RefCount<ClassCHostPool> Pointer;
    void dump(StoreEntry *entry) const override;
    void parse() override;
    void update(int incr) override;
    void stats(StoreEntry * sentry) override;

    DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;
    ClassCHostPool();
    ~ClassCHostPool() override;

protected:
    bool keyAllocated (unsigned char const key) const;
    virtual DelaySpec *rate() {return &spec;}

    virtual DelaySpec const *rate() const {return &spec;}

    virtual char const *label() const {return "Individual";}

    virtual unsigned int makeKey(Ip::Address &src_addr) const;

    unsigned char makeHostKey(Ip::Address &src_addr) const;

    DelaySpec spec;
    VectorMap<unsigned char, ClassCBucket> buckets;

    class Id;

    friend class ClassCHostPool::Id;

    /// \ingroup DelayPoolsInternal
    class Id:public DelayIdComposite
    {
        MEMPROXY_CLASS(ClassCHostPool::Id);

    public:
        Id (RefCount<ClassCHostPool>, unsigned char, unsigned char);
        int bytesWanted (int min, int max) const override;
        void bytesIn(int qty) override;

    private:
        RefCount<ClassCHostPool> theClassCHost;
        unsigned char theNet;
        unsigned char theHost;
    };
};

void
Aggregate::AggregateId::delayRead(const AsyncCall::Pointer &aRead)
{
    theAggregate->delayRead(aRead);
}

CommonPool *
CommonPool::Factory(unsigned char _class, CompositePoolNode::Pointer& compositeCopy)
{
    CommonPool *result = new CommonPool;

    switch (_class) {

    case 0:
        break;

    case 1:
        compositeCopy = new Aggregate;
        result->typeLabel = SBuf("1");
        break;

    case 2:
        result->typeLabel = SBuf("2");
        {
            DelayVector::Pointer temp = new DelayVector;
            compositeCopy = temp.getRaw();
            temp->push_back (new Aggregate);
            temp->push_back(new IndividualPool);
        }
        break;

    case 3:
        result->typeLabel = SBuf("3");
        {
            DelayVector::Pointer temp = new DelayVector;
            compositeCopy = temp.getRaw();
            temp->push_back (new Aggregate);
            temp->push_back (new ClassCNetPool);
            temp->push_back (new ClassCHostPool);
        }
        break;

    case 4:
        result->typeLabel = SBuf("4");
        {
            DelayVector::Pointer temp = new DelayVector;
            compositeCopy = temp.getRaw();
            temp->push_back (new Aggregate);
            temp->push_back (new ClassCNetPool);
            temp->push_back (new ClassCHostPool);
#if USE_AUTH
            temp->push_back (new DelayUser);
#endif
        }
        break;

    case 5:
        result->typeLabel = SBuf("5");
        compositeCopy = new DelayTagged;
        break;

    default:
        fatal ("unknown delay pool class");
        return nullptr;
    };

    return result;
}

CommonPool::CommonPool()
{}

void
ClassCBucket::update (DelaySpec const &rate, int incr)
{
    /* If we aren't active, don't try to update us ! */
    assert (rate.restore_bps != -1);

    for (unsigned int j = 0; j < individuals.size(); ++j)
        individuals.values[j].update (rate, incr);
}

void
ClassCBucket::stats(StoreEntry *sentry)const
{
    for (unsigned int j = 0; j < individuals.size(); ++j) {
        assert (individualUsed (j));
        storeAppendPrintf(sentry, " %d:",individuals.key_map[j]);
        individuals.values[j].stats (sentry);
    }
}

unsigned char
ClassCBucket::findHostMapPosition (unsigned char const host) const
{
    return individuals.findKeyIndex(host);
}

bool
ClassCBucket::individualUsed (unsigned int index)const
{
    return individuals.indexUsed(index);
}

bool
ClassCBucket::individualAllocated (unsigned char host) const
{
    return individualUsed(findHostMapPosition (host));
}

unsigned char
ClassCBucket::hostPosition (DelaySpec &rate, unsigned char const host)
{
    if (individualAllocated (host))
        return findHostMapPosition(host);

    assert (!individualUsed (findHostMapPosition(host)));

    unsigned char result = findHostMapPosition(host);

    initHostIndex (rate, result, host);

    return result;
}

void
ClassCBucket::initHostIndex (DelaySpec &rate, unsigned char index, unsigned char host)
{
    assert (!individualUsed(index));

    unsigned char const newIndex = individuals.insert (host);

    /* give the bucket a default value */
    individuals.values[newIndex].init (rate);
}

Aggregate::Aggregate()
{
    theBucket.init (*rate());
    DelayPools::registerForUpdates (this);
}

Aggregate::~Aggregate()
{
    DelayPools::deregisterForUpdates (this);
}

void
Aggregate::stats(StoreEntry * sentry)
{
    rate()->stats (sentry, "Aggregate");

    if (rate()->restore_bps == -1)
        return;

    storeAppendPrintf(sentry, "\t\tCurrent: ");

    theBucket.stats(sentry);

    storeAppendPrintf(sentry, "\n\n");
}

void
Aggregate::dump(StoreEntry *entry) const
{
    rate()->dump (entry);
}

void
Aggregate::update(int incr)
{
    theBucket.update(*rate(), incr);
    kickReads();
}

void
Aggregate::parse()
{
    rate()->parse();
}

DelayIdComposite::Pointer
Aggregate::id(CompositeSelectionDetails &)
{
    if (rate()->restore_bps != -1)
        return new AggregateId (this);
    else
        return new NullDelayId;
}

Aggregate::AggregateId::AggregateId(RefCount<Aggregate> anAggregate) : theAggregate(anAggregate)
{}

int
Aggregate::AggregateId::bytesWanted (int min, int max) const
{
    return theAggregate->theBucket.bytesWanted(min, max);
}

void
Aggregate::AggregateId::bytesIn(int qty)
{
    theAggregate->theBucket.bytesIn(qty);
    theAggregate->kickReads();
}

DelayPool *DelayPools::delay_data = nullptr;
time_t DelayPools::LastUpdate = 0;
unsigned short DelayPools::pools_ (0);

void
DelayPools::RegisterWithCacheManager(void)
{
    Mgr::RegisterAction("delay", "Delay Pool Levels", Stats, 0, 1);
}

void
DelayPools::Init()
{
    LastUpdate = getCurrentTime();
    RegisterWithCacheManager();
}

void
DelayPools::InitDelayData()
{
    if (!pools())
        return;

    DelayPools::delay_data = new DelayPool[pools()];

    eventAdd("DelayPools::Update", DelayPools::Update, nullptr, 1.0, 1);
}

void
DelayPools::FreeDelayData()
{
    delete[] DelayPools::delay_data;
    pools_ = 0;
}

void
DelayPools::Update(void *)
{
    // To prevent stuck transactions, stop updates only after no new transactions can
    // register (because the pools were disabled) and the last registered transaction is gone.
    if (!pools() && toUpdate.empty())
        return;

    eventAdd("DelayPools::Update", Update, nullptr, 1.0, 1);

    int incr = squid_curtime - LastUpdate;

    if (incr < 1)
        return;

    LastUpdate = squid_curtime;

    std::vector<Updateable *>::iterator pos = toUpdate.begin();

    while (pos != toUpdate.end()) {
        (*pos)->update(incr);
        ++pos;
    }
}

void
DelayPools::registerForUpdates(Updateable *anObject)
{
    /* Assume no doubles */
    toUpdate.push_back(anObject);
}

void
DelayPools::deregisterForUpdates (Updateable *anObject)
{
    std::vector<Updateable *>::iterator pos = toUpdate.begin();

    while (pos != toUpdate.end() && *pos != anObject) {
        ++pos;
    }

    if (pos != toUpdate.end()) {
        /* move all objects down one */
        std::vector<Updateable *>::iterator temp = pos;
        ++pos;

        while (pos != toUpdate.end()) {
            *temp = *pos;
            ++temp;
            ++pos;
        }

        toUpdate.pop_back();
    }
}

std::vector<Updateable *> DelayPools::toUpdate;

void
DelayPools::Stats(StoreEntry * sentry)
{
    storeAppendPrintf(sentry, "Delay pools configured: %d\n\n", DelayPools::pools());

    for (unsigned short i = 0; i < DelayPools::pools(); ++i) {
        if (DelayPools::delay_data[i].theComposite().getRaw()) {
            storeAppendPrintf(sentry, "Pool: %d\n\tClass: " SQUIDSBUFPH "\n\n", i + 1, SQUIDSBUFPRINT(DelayPools::delay_data[i].pool->classTypeLabel()));
            DelayPools::delay_data[i].theComposite()->stats (sentry);
        } else
            storeAppendPrintf(sentry, "\tMisconfigured pool.\n\n");
    }
}

void
DelayPools::FreePools()
{
    if (!DelayPools::pools())
        return;

    FreeDelayData();
}

unsigned short
DelayPools::pools()
{
    return pools_;
}

void
DelayPools::pools(unsigned short newPools)
{
    if (pools()) {
        debugs(3, DBG_CRITICAL, "parse_delay_pool_count: multiple delay_pools lines, aborting all previous delay_pools config");
        FreePools();
    }

    pools_ = newPools;

    if (pools())
        InitDelayData();
}

template <class Key, class Value>
VectorMap<Key,Value>::VectorMap() : nextMapPosition(0)
{}

template <class Key, class Value>
unsigned int
VectorMap<Key,Value>::size() const
{
    return nextMapPosition;
}

template <class Key, class Value>
unsigned int
VectorMap<Key,Value>::insert (Key const key)
{
    unsigned char index = findKeyIndex (key);
    assert (!indexUsed(index));

    key_map[index] = key;

    ++nextMapPosition;

    return index;
}

VectorPool::VectorPool()
{
    DelayPools::registerForUpdates (this);
}

VectorPool::~VectorPool()
{
    DelayPools::deregisterForUpdates (this);
}

void
VectorPool::stats(StoreEntry * sentry)
{
    rate()->stats (sentry, label());

    if (rate()->restore_bps == -1) {
        storeAppendPrintf(sentry, "\n\n");
        return;
    }

    storeAppendPrintf(sentry, "\t\tCurrent:");

    for (unsigned int i = 0; i < buckets.size(); ++i) {
        storeAppendPrintf(sentry, " %d:", buckets.key_map[i]);
        buckets.values[i].stats(sentry);
    }

    if (!buckets.size())
        storeAppendPrintf(sentry, " Not used yet.");

    storeAppendPrintf(sentry, "\n\n");
}

void
VectorPool::dump(StoreEntry *entry) const
{
    rate()->dump (entry);
}

void
VectorPool::update(int incr)
{
    if (rate()->restore_bps == -1)
        return;

    for (unsigned int i = 0; i< buckets.size(); ++i)
        buckets.values[i].update (*rate(), incr);
}

void
VectorPool::parse()
{
    rate()->parse();
}

bool
VectorPool::keyAllocated (unsigned char const key) const
{
    return buckets.indexUsed(buckets.findKeyIndex (key));
}

template <class Key, class Value>
bool
VectorMap<Key,Value>::indexUsed (unsigned char const index) const
{
    return index < size();
}

/** returns the used position, or the position to allocate */
template <class Key, class Value>
unsigned char
VectorMap<Key,Value>::findKeyIndex (Key const key) const
{
    for (unsigned int index = 0; index < size(); ++index) {
        assert(indexUsed(index));

        if (key_map[index] == key)
            return index;
    }

    /* not in map */
    return size();
}

DelayIdComposite::Pointer
VectorPool::id(CompositeSelectionDetails &details)
{
    if (rate()->restore_bps == -1)
        return new NullDelayId;

    /* non-IPv4 are not able to provide IPv4-bitmask for this pool type key. */
    if ( !details.src_addr.isIPv4() )
        return new NullDelayId;

    unsigned int key = makeKey(details.src_addr);

    if (keyAllocated(key))
        return new Id(this, buckets.findKeyIndex(key));

    unsigned char const resultIndex = buckets.insert(key);

    buckets.values[resultIndex].init(*rate());

    return new Id(this, resultIndex);
}

VectorPool::Id::Id(VectorPool::Pointer aPool, int anIndex) : theVector (aPool), theIndex (anIndex)
{}

int
VectorPool::Id::bytesWanted (int min, int max) const
{
    return theVector->buckets.values[theIndex].bytesWanted (min, max);
}

void
VectorPool::Id::bytesIn(int qty)
{
    theVector->buckets.values[theIndex].bytesIn (qty);
}

unsigned int
IndividualPool::makeKey(Ip::Address &src_addr) const
{
    /* IPv4 required for this pool */
    if ( !src_addr.isIPv4() )
        return 1;

    struct in_addr host;
    src_addr.getInAddr(host);
    return (ntohl(host.s_addr) & 0xff);
}

unsigned int
ClassCNetPool::makeKey(Ip::Address &src_addr) const
{
    /* IPv4 required for this pool */
    if ( !src_addr.isIPv4() )
        return 1;

    struct in_addr net;
    src_addr.getInAddr(net);
    return ( (ntohl(net.s_addr) >> 8) & 0xff);
}

ClassCHostPool::ClassCHostPool()
{
    DelayPools::registerForUpdates (this);
}

ClassCHostPool::~ClassCHostPool()
{
    DelayPools::deregisterForUpdates (this);
}

void
ClassCHostPool::stats(StoreEntry * sentry)
{
    rate()->stats (sentry, label());

    if (rate()->restore_bps == -1) {
        storeAppendPrintf(sentry, "\n\n");
        return;
    }

    for (unsigned int index = 0; index < buckets.size(); ++index) {
        storeAppendPrintf(sentry, "\t\tCurrent [Network %d]:", buckets.key_map[index]);
        buckets.values[index].stats (sentry);
        storeAppendPrintf(sentry, "\n");
    }

    if (!buckets.size())
        storeAppendPrintf(sentry, "\t\tCurrent [All networks]: Not used yet.\n");

    storeAppendPrintf(sentry, "\n\n");
}

void
ClassCHostPool::dump(StoreEntry *entry) const
{
    rate()->dump (entry);
}

void
ClassCHostPool::update(int incr)
{
    if (rate()->restore_bps == -1)
        return;

    for (unsigned int i = 0; i< buckets.size(); ++i)
        buckets.values[i].update (*rate(), incr);
}

void
ClassCHostPool::parse()
{
    rate()->parse();
}

bool
ClassCHostPool::keyAllocated (unsigned char const key) const
{
    return buckets.indexUsed(buckets.findKeyIndex (key));
}

unsigned char
ClassCHostPool::makeHostKey(Ip::Address &src_addr) const
{
    /* IPv4 required for this pool */
    if ( !src_addr.isIPv4() )
        return 1;

    /* Temporary bypass for IPv4-only */
    struct in_addr host;
    src_addr.getInAddr(host);
    return (ntohl(host.s_addr) & 0xff);
}

unsigned int
ClassCHostPool::makeKey(Ip::Address &src_addr) const
{
    /* IPv4 required for this pool */
    if ( !src_addr.isIPv4() )
        return 1;

    struct in_addr net;
    src_addr.getInAddr(net);
    return ( (ntohl(net.s_addr) >> 8) & 0xff);
}

DelayIdComposite::Pointer
ClassCHostPool::id(CompositeSelectionDetails &details)
{
    if (rate()->restore_bps == -1)
        return new NullDelayId;

    /* non-IPv4 are not able to provide IPv4-bitmask for this pool type key. */
    if ( !details.src_addr.isIPv4() )
        return new NullDelayId;

    unsigned int key = makeKey (details.src_addr);

    unsigned char host = makeHostKey (details.src_addr);

    unsigned char hostIndex;

    unsigned char netIndex;

    if (keyAllocated (key))
        netIndex = buckets.findKeyIndex(key);
    else
        netIndex = buckets.insert (key);

    hostIndex = buckets.values[netIndex].hostPosition (*rate(), host);

    return new Id (this, netIndex, hostIndex);
}

ClassCHostPool::Id::Id (ClassCHostPool::Pointer aPool, unsigned char aNet, unsigned char aHost) : theClassCHost (aPool), theNet (aNet), theHost (aHost)
{}

int
ClassCHostPool::Id::bytesWanted (int min, int max) const
{
    return theClassCHost->buckets.values[theNet].individuals.values[theHost].bytesWanted (min, max);
}

void
ClassCHostPool::Id::bytesIn(int qty)
{
    theClassCHost->buckets.values[theNet].individuals.values[theHost].bytesIn (qty);
}

#endif /* USE_DELAY_POOLS */
Commit	Line	Data
	1	/*
	2	* Copyright (C) 1996-2025 The Squid Software Foundation and contributors
	3	*
	4	* Squid software is distributed under GPLv2+ license and includes
	5	* contributions from numerous individuals and organizations.
	6	* Please see the COPYING and CONTRIBUTORS files for details.
	7	*/
	8
	9	/* DEBUG: section 77 Delay Pools */
	10
	11	/**
	12	\defgroup DelayPoolsInternal Delay Pools Internal
	13	\ingroup DelayPoolsAPI
	14	*/
	15
	16	#include "squid.h"
	17
	18	#if USE_DELAY_POOLS
	19	#include "client_side_request.h"
	20	#include "comm/Connection.h"
	21	#include "CommonPool.h"
	22	#include "CompositePoolNode.h"
	23	#include "ConfigParser.h"
	24	#include "DelayBucket.h"
	25	#include "DelayId.h"
	26	#include "DelayPool.h"
	27	#include "DelayPools.h"
	28	#include "DelaySpec.h"
	29	#include "DelayTagged.h"
	30	#include "DelayUser.h"
	31	#include "DelayVector.h"
	32	#include "event.h"
	33	#include "http/Stream.h"
	34	#include "ip/Address.h"
	35	#include "MemObject.h"
	36	#include "mgr/Registration.h"
	37	#include "NullDelayId.h"
	38	#include "sbuf/SBuf.h"
	39	#include "Store.h"
	40	#include "StoreClient.h"
	41
	42	/// \ingroup DelayPoolsInternal
	43	class Aggregate : public CompositePoolNode
	44	{
	45	MEMPROXY_CLASS(Aggregate);
	46
	47	public:
	48	typedef RefCount<Aggregate> Pointer;
	49	Aggregate();
	50	~Aggregate() override;
	51	virtual DelaySpec *rate() {return &spec;}
	52
	53	virtual DelaySpec const *rate() const {return &spec;}
	54
	55	void stats(StoreEntry * sentry) override;
	56	void dump(StoreEntry *entry) const override;
	57	void update(int incr) override;
	58	void parse() override;
	59
	60	DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;
	61
	62	private:
	63
	64	/// \ingroup DelayPoolsInternal
	65	class AggregateId:public DelayIdComposite
	66	{
	67	MEMPROXY_CLASS(Aggregate::AggregateId);
	68
	69	public:
	70	AggregateId (RefCount<Aggregate>);
	71	int bytesWanted (int min, int max) const override;
	72	void bytesIn(int qty) override;
	73	void delayRead(const AsyncCallPointer &) override;
	74
	75	private:
	76	RefCount<Aggregate> theAggregate;
	77	};
	78
	79	friend class AggregateId;
	80
	81	DelayBucket theBucket;
	82	DelaySpec spec;
	83	};
	84
	85	/// \ingroup DelayPoolsInternal
	86	template <class Key, class Value>
	87	class VectorMap
	88	{
	89
	90	public:
	91	VectorMap();
	92	unsigned int size() const;
	93	unsigned char findKeyIndex (Key const key) const;
	94	bool indexUsed (unsigned char const index) const;
	95	unsigned int insert (Key const key);
	96
	97	#define IND_MAP_SZ 256
	98
	99	Key key_map[IND_MAP_SZ];
	100	Value values[IND_MAP_SZ];
	101
	102	private:
	103	unsigned int nextMapPosition;
	104	};
	105
	106	/// \ingroup DelayPoolsInternal
	107	class VectorPool : public CompositePoolNode
	108	{
	109	MEMPROXY_CLASS(VectorPool);
	110
	111	public:
	112	typedef RefCount<VectorPool> Pointer;
	113	void dump(StoreEntry *entry) const override;
	114	void parse() override;
	115	void update(int incr) override;
	116	void stats(StoreEntry * sentry) override;
	117
	118	DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;
	119	VectorMap<unsigned char, DelayBucket> buckets;
	120	VectorPool();
	121	~VectorPool() override;
	122
	123	protected:
	124	bool keyAllocated (unsigned char const key) const;
	125	virtual DelaySpec *rate() {return &spec;}
	126
	127	virtual DelaySpec const *rate() const {return &spec;}
	128
	129	virtual char const *label() const = 0;
	130
	131	virtual unsigned int makeKey(Ip::Address &src_addr) const = 0;
	132
	133	DelaySpec spec;
	134
	135	/// \ingroup DelayPoolsInternal
	136	class Id:public DelayIdComposite
	137	{
	138	MEMPROXY_CLASS(VectorPool::Id);
	139
	140	public:
	141	Id (RefCount<VectorPool>, int);
	142	int bytesWanted (int min, int max) const override;
	143	void bytesIn(int qty) override;
	144
	145	private:
	146	RefCount<VectorPool> theVector;
	147	int theIndex;
	148	};
	149	};
	150
	151	/// \ingroup DelayPoolsInternal
	152	class IndividualPool : public VectorPool
	153	{
	154	MEMPROXY_CLASS(IndividualPool);
	155
	156	protected:
	157	char const *label() const override {return "Individual";}
	158	unsigned int makeKey(Ip::Address &src_addr) const override;
	159	};
	160
	161	/// \ingroup DelayPoolsInternal
	162	class ClassCNetPool : public VectorPool
	163	{
	164	MEMPROXY_CLASS(ClassCNetPool);
	165
	166	protected:
	167	char const *label() const override {return "Network";}
	168	unsigned int makeKey (Ip::Address &src_addr) const override;
	169	};
	170
	171	/* don't use remote storage for these */
	172	/// \ingroup DelayPoolsInternal
	173	class ClassCBucket
	174	{
	175
	176	public:
	177	bool individualUsed (unsigned int index)const;
	178	unsigned char findHostMapPosition (unsigned char const host) const;
	179	bool individualAllocated (unsigned char host) const;
	180	unsigned char hostPosition (DelaySpec &rate, unsigned char const host);
	181	void initHostIndex (DelaySpec &rate, unsigned char index, unsigned char host);
	182	void update (DelaySpec const &, int incr);
	183	void stats(StoreEntry *)const;
	184
	185	DelayBucket net;
	186	VectorMap<unsigned char, DelayBucket> individuals;
	187	};
	188
	189	/// \ingroup DelayPoolsInternal
	190	class ClassCHostPool : public CompositePoolNode
	191	{
	192	MEMPROXY_CLASS(ClassCHostPool);
	193
	194	public:
	195	typedef RefCount<ClassCHostPool> Pointer;
	196	void dump(StoreEntry *entry) const override;
	197	void parse() override;
	198	void update(int incr) override;
	199	void stats(StoreEntry * sentry) override;
	200
	201	DelayIdComposite::Pointer id(CompositeSelectionDetails &) override;
	202	ClassCHostPool();
	203	~ClassCHostPool() override;
	204
	205	protected:
	206	bool keyAllocated (unsigned char const key) const;
	207	virtual DelaySpec *rate() {return &spec;}
	208
	209	virtual DelaySpec const *rate() const {return &spec;}
	210
	211	virtual char const *label() const {return "Individual";}
	212
	213	virtual unsigned int makeKey(Ip::Address &src_addr) const;
	214
	215	unsigned char makeHostKey(Ip::Address &src_addr) const;
	216
	217	DelaySpec spec;
	218	VectorMap<unsigned char, ClassCBucket> buckets;
	219
	220	class Id;
	221
	222	friend class ClassCHostPool::Id;
	223
	224	/// \ingroup DelayPoolsInternal
	225	class Id:public DelayIdComposite
	226	{
	227	MEMPROXY_CLASS(ClassCHostPool::Id);
	228
	229	public:
	230	Id (RefCount<ClassCHostPool>, unsigned char, unsigned char);
	231	int bytesWanted (int min, int max) const override;
	232	void bytesIn(int qty) override;
	233
	234	private:
	235	RefCount<ClassCHostPool> theClassCHost;
	236	unsigned char theNet;
	237	unsigned char theHost;
	238	};
	239	};
	240
	241	void
	242	Aggregate::AggregateId::delayRead(const AsyncCall::Pointer &aRead)
	243	{
	244	theAggregate->delayRead(aRead);
	245	}
	246
	247	CommonPool *
	248	CommonPool::Factory(unsigned char _class, CompositePoolNode::Pointer& compositeCopy)
	249	{
	250	CommonPool *result = new CommonPool;
	251
	252	switch (_class) {
	253
	254	case 0:
	255	break;
	256
	257	case 1:
	258	compositeCopy = new Aggregate;
	259	result->typeLabel = SBuf("1");
	260	break;
	261
	262	case 2:
	263	result->typeLabel = SBuf("2");
	264	{
	265	DelayVector::Pointer temp = new DelayVector;
	266	compositeCopy = temp.getRaw();
	267	temp->push_back (new Aggregate);
	268	temp->push_back(new IndividualPool);
	269	}
	270	break;
	271
	272	case 3:
	273	result->typeLabel = SBuf("3");
	274	{
	275	DelayVector::Pointer temp = new DelayVector;
	276	compositeCopy = temp.getRaw();
	277	temp->push_back (new Aggregate);
	278	temp->push_back (new ClassCNetPool);
	279	temp->push_back (new ClassCHostPool);
	280	}
	281	break;
	282
	283	case 4:
	284	result->typeLabel = SBuf("4");
	285	{
	286	DelayVector::Pointer temp = new DelayVector;
	287	compositeCopy = temp.getRaw();
	288	temp->push_back (new Aggregate);
	289	temp->push_back (new ClassCNetPool);
	290	temp->push_back (new ClassCHostPool);
	291	#if USE_AUTH
	292	temp->push_back (new DelayUser);
	293	#endif
	294	}
	295	break;
	296
	297	case 5:
	298	result->typeLabel = SBuf("5");
	299	compositeCopy = new DelayTagged;
	300	break;
	301
	302	default:
	303	fatal ("unknown delay pool class");
	304	return nullptr;
	305	};
	306
	307	return result;
	308	}
	309
	310	CommonPool::CommonPool()
	311	{}
	312
	313	void
	314	ClassCBucket::update (DelaySpec const &rate, int incr)
	315	{
	316	/* If we aren't active, don't try to update us ! */
	317	assert (rate.restore_bps != -1);
	318
	319	for (unsigned int j = 0; j < individuals.size(); ++j)
	320	individuals.values[j].update (rate, incr);
	321	}
	322
	323	void
	324	ClassCBucket::stats(StoreEntry *sentry)const
	325	{
	326	for (unsigned int j = 0; j < individuals.size(); ++j) {
	327	assert (individualUsed (j));
	328	storeAppendPrintf(sentry, " %d:",individuals.key_map[j]);
	329	individuals.values[j].stats (sentry);
	330	}
	331	}
	332
	333	unsigned char
	334	ClassCBucket::findHostMapPosition (unsigned char const host) const
	335	{
	336	return individuals.findKeyIndex(host);
	337	}
	338
	339	bool
	340	ClassCBucket::individualUsed (unsigned int index)const
	341	{
	342	return individuals.indexUsed(index);
	343	}
	344
	345	bool
	346	ClassCBucket::individualAllocated (unsigned char host) const
	347	{
	348	return individualUsed(findHostMapPosition (host));
	349	}
	350
	351	unsigned char
	352	ClassCBucket::hostPosition (DelaySpec &rate, unsigned char const host)
	353	{
	354	if (individualAllocated (host))
	355	return findHostMapPosition(host);
	356
	357	assert (!individualUsed (findHostMapPosition(host)));
	358
	359	unsigned char result = findHostMapPosition(host);
	360
	361	initHostIndex (rate, result, host);
	362
	363	return result;
	364	}
	365
	366	void
	367	ClassCBucket::initHostIndex (DelaySpec &rate, unsigned char index, unsigned char host)
	368	{
	369	assert (!individualUsed(index));
	370
	371	unsigned char const newIndex = individuals.insert (host);
	372
	373	/* give the bucket a default value */
	374	individuals.values[newIndex].init (rate);
	375	}
	376
	377	Aggregate::Aggregate()
	378	{
	379	theBucket.init (*rate());
	380	DelayPools::registerForUpdates (this);
	381	}
	382
	383	Aggregate::~Aggregate()
	384	{
	385	DelayPools::deregisterForUpdates (this);
	386	}
	387
	388	void
	389	Aggregate::stats(StoreEntry * sentry)
	390	{
	391	rate()->stats (sentry, "Aggregate");
	392
	393	if (rate()->restore_bps == -1)
	394	return;
	395
	396	storeAppendPrintf(sentry, "\t\tCurrent: ");
	397
	398	theBucket.stats(sentry);
	399
	400	storeAppendPrintf(sentry, "\n\n");
	401	}
	402
	403	void
	404	Aggregate::dump(StoreEntry *entry) const
	405	{
	406	rate()->dump (entry);
	407	}
	408
	409	void
	410	Aggregate::update(int incr)
	411	{
	412	theBucket.update(*rate(), incr);
	413	kickReads();
	414	}
	415
	416	void
	417	Aggregate::parse()
	418	{
	419	rate()->parse();
	420	}
	421
	422	DelayIdComposite::Pointer
	423	Aggregate::id(CompositeSelectionDetails &)
	424	{
	425	if (rate()->restore_bps != -1)
	426	return new AggregateId (this);
	427	else
	428	return new NullDelayId;
	429	}
	430
	431	Aggregate::AggregateId::AggregateId(RefCount<Aggregate> anAggregate) : theAggregate(anAggregate)
	432	{}
	433
	434	int
	435	Aggregate::AggregateId::bytesWanted (int min, int max) const
	436	{
	437	return theAggregate->theBucket.bytesWanted(min, max);
	438	}
	439
	440	void
	441	Aggregate::AggregateId::bytesIn(int qty)
	442	{
	443	theAggregate->theBucket.bytesIn(qty);
	444	theAggregate->kickReads();
	445	}
	446
	447	DelayPool *DelayPools::delay_data = nullptr;
	448	time_t DelayPools::LastUpdate = 0;
	449	unsigned short DelayPools::pools_ (0);
	450
	451	void
	452	DelayPools::RegisterWithCacheManager(void)
	453	{
	454	Mgr::RegisterAction("delay", "Delay Pool Levels", Stats, 0, 1);
	455	}
	456
	457	void
	458	DelayPools::Init()
	459	{
	460	LastUpdate = getCurrentTime();
	461	RegisterWithCacheManager();
	462	}
	463
	464	void
	465	DelayPools::InitDelayData()
	466	{
	467	if (!pools())
	468	return;
	469
	470	DelayPools::delay_data = new DelayPool[pools()];
	471
	472	eventAdd("DelayPools::Update", DelayPools::Update, nullptr, 1.0, 1);
	473	}
	474
	475	void
	476	DelayPools::FreeDelayData()
	477	{
	478	delete[] DelayPools::delay_data;
	479	pools_ = 0;
	480	}
	481
	482	void
	483	DelayPools::Update(void *)
	484	{
	485	// To prevent stuck transactions, stop updates only after no new transactions can
	486	// register (because the pools were disabled) and the last registered transaction is gone.
	487	if (!pools() && toUpdate.empty())
	488	return;
	489
	490	eventAdd("DelayPools::Update", Update, nullptr, 1.0, 1);
	491
	492	int incr = squid_curtime - LastUpdate;
	493
	494	if (incr < 1)
	495	return;
	496
	497	LastUpdate = squid_curtime;
	498
	499	std::vector<Updateable *>::iterator pos = toUpdate.begin();
	500
	501	while (pos != toUpdate.end()) {
	502	(*pos)->update(incr);
	503	++pos;
	504	}
	505	}
	506
	507	void
	508	DelayPools::registerForUpdates(Updateable *anObject)
	509	{
	510	/* Assume no doubles */
	511	toUpdate.push_back(anObject);
	512	}
	513
	514	void
	515	DelayPools::deregisterForUpdates (Updateable *anObject)
	516	{
	517	std::vector<Updateable *>::iterator pos = toUpdate.begin();
	518
	519	while (pos != toUpdate.end() && *pos != anObject) {
	520	++pos;
	521	}
	522
	523	if (pos != toUpdate.end()) {
	524	/* move all objects down one */
	525	std::vector<Updateable *>::iterator temp = pos;
	526	++pos;
	527
	528	while (pos != toUpdate.end()) {
	529	temp = pos;
	530	++temp;
	531	++pos;
	532	}
	533
	534	toUpdate.pop_back();
	535	}
	536	}
	537
	538	std::vector<Updateable *> DelayPools::toUpdate;
	539
	540	void
	541	DelayPools::Stats(StoreEntry * sentry)
	542	{
	543	storeAppendPrintf(sentry, "Delay pools configured: %d\n\n", DelayPools::pools());
	544
	545	for (unsigned short i = 0; i < DelayPools::pools(); ++i) {
	546	if (DelayPools::delay_data[i].theComposite().getRaw()) {
	547	storeAppendPrintf(sentry, "Pool: %d\n\tClass: " SQUIDSBUFPH "\n\n", i + 1, SQUIDSBUFPRINT(DelayPools::delay_data[i].pool->classTypeLabel()));
	548	DelayPools::delay_data[i].theComposite()->stats (sentry);
	549	} else
	550	storeAppendPrintf(sentry, "\tMisconfigured pool.\n\n");
	551	}
	552	}
	553
	554	void
	555	DelayPools::FreePools()
	556	{
	557	if (!DelayPools::pools())
	558	return;
	559
	560	FreeDelayData();
	561	}
	562
	563	unsigned short
	564	DelayPools::pools()
	565	{
	566	return pools_;
	567	}
	568
	569	void
	570	DelayPools::pools(unsigned short newPools)
	571	{
	572	if (pools()) {
	573	debugs(3, DBG_CRITICAL, "parse_delay_pool_count: multiple delay_pools lines, aborting all previous delay_pools config");
	574	FreePools();
	575	}
	576
	577	pools_ = newPools;
	578
	579	if (pools())
	580	InitDelayData();
	581	}
	582
	583	template <class Key, class Value>
	584	VectorMap<Key,Value>::VectorMap() : nextMapPosition(0)
	585	{}
	586
	587	template <class Key, class Value>
	588	unsigned int
	589	VectorMap<Key,Value>::size() const
	590	{
	591	return nextMapPosition;
	592	}
	593
	594	template <class Key, class Value>
	595	unsigned int
	596	VectorMap<Key,Value>::insert (Key const key)
	597	{
	598	unsigned char index = findKeyIndex (key);
	599	assert (!indexUsed(index));
	600
	601	key_map[index] = key;
	602
	603	++nextMapPosition;
	604
	605	return index;
	606	}
	607
	608	VectorPool::VectorPool()
	609	{
	610	DelayPools::registerForUpdates (this);
	611	}
	612
	613	VectorPool::~VectorPool()
	614	{
	615	DelayPools::deregisterForUpdates (this);
	616	}
	617
	618	void
	619	VectorPool::stats(StoreEntry * sentry)
	620	{
	621	rate()->stats (sentry, label());
	622
	623	if (rate()->restore_bps == -1) {
	624	storeAppendPrintf(sentry, "\n\n");
	625	return;
	626	}
	627
	628	storeAppendPrintf(sentry, "\t\tCurrent:");
	629
	630	for (unsigned int i = 0; i < buckets.size(); ++i) {
	631	storeAppendPrintf(sentry, " %d:", buckets.key_map[i]);
	632	buckets.values[i].stats(sentry);
	633	}
	634
	635	if (!buckets.size())
	636	storeAppendPrintf(sentry, " Not used yet.");
	637
	638	storeAppendPrintf(sentry, "\n\n");
	639	}
	640
	641	void
	642	VectorPool::dump(StoreEntry *entry) const
	643	{
	644	rate()->dump (entry);
	645	}
	646
	647	void
	648	VectorPool::update(int incr)
	649	{
	650	if (rate()->restore_bps == -1)
	651	return;
	652
	653	for (unsigned int i = 0; i< buckets.size(); ++i)
	654	buckets.values[i].update (*rate(), incr);
	655	}
	656
	657	void
	658	VectorPool::parse()
	659	{
	660	rate()->parse();
	661	}
	662
	663	bool
	664	VectorPool::keyAllocated (unsigned char const key) const
	665	{
	666	return buckets.indexUsed(buckets.findKeyIndex (key));
	667	}
	668
	669	template <class Key, class Value>
	670	bool
	671	VectorMap<Key,Value>::indexUsed (unsigned char const index) const
	672	{
	673	return index < size();
	674	}
	675
	676	/** returns the used position, or the position to allocate */
	677	template <class Key, class Value>
	678	unsigned char
	679	VectorMap<Key,Value>::findKeyIndex (Key const key) const
	680	{
	681	for (unsigned int index = 0; index < size(); ++index) {
	682	assert(indexUsed(index));
	683
	684	if (key_map[index] == key)
	685	return index;
	686	}
	687
	688	/* not in map */
	689	return size();
	690	}
	691
	692	DelayIdComposite::Pointer
	693	VectorPool::id(CompositeSelectionDetails &details)
	694	{
	695	if (rate()->restore_bps == -1)
	696	return new NullDelayId;
	697
	698	/* non-IPv4 are not able to provide IPv4-bitmask for this pool type key. */
	699	if ( !details.src_addr.isIPv4() )
	700	return new NullDelayId;
	701
	702	unsigned int key = makeKey(details.src_addr);
	703
	704	if (keyAllocated(key))
	705	return new Id(this, buckets.findKeyIndex(key));
	706
	707	unsigned char const resultIndex = buckets.insert(key);
	708
	709	buckets.values[resultIndex].init(*rate());
	710
	711	return new Id(this, resultIndex);
	712	}
	713
	714	VectorPool::Id::Id(VectorPool::Pointer aPool, int anIndex) : theVector (aPool), theIndex (anIndex)
	715	{}
	716
	717	int
	718	VectorPool::Id::bytesWanted (int min, int max) const
	719	{
	720	return theVector->buckets.values[theIndex].bytesWanted (min, max);
	721	}
	722
	723	void
	724	VectorPool::Id::bytesIn(int qty)
	725	{
	726	theVector->buckets.values[theIndex].bytesIn (qty);
	727	}
	728
	729	unsigned int
	730	IndividualPool::makeKey(Ip::Address &src_addr) const
	731	{
	732	/* IPv4 required for this pool */
	733	if ( !src_addr.isIPv4() )
	734	return 1;
	735
	736	struct in_addr host;
	737	src_addr.getInAddr(host);
	738	return (ntohl(host.s_addr) & 0xff);
	739	}
	740
	741	unsigned int
	742	ClassCNetPool::makeKey(Ip::Address &src_addr) const
	743	{
	744	/* IPv4 required for this pool */
	745	if ( !src_addr.isIPv4() )
	746	return 1;
	747
	748	struct in_addr net;
	749	src_addr.getInAddr(net);
	750	return ( (ntohl(net.s_addr) >> 8) & 0xff);
	751	}
	752
	753	ClassCHostPool::ClassCHostPool()
	754	{
	755	DelayPools::registerForUpdates (this);
	756	}
	757
	758	ClassCHostPool::~ClassCHostPool()
	759	{
	760	DelayPools::deregisterForUpdates (this);
	761	}
	762
	763	void
	764	ClassCHostPool::stats(StoreEntry * sentry)
	765	{
	766	rate()->stats (sentry, label());
	767
	768	if (rate()->restore_bps == -1) {
	769	storeAppendPrintf(sentry, "\n\n");
	770	return;
	771	}
	772
	773	for (unsigned int index = 0; index < buckets.size(); ++index) {
	774	storeAppendPrintf(sentry, "\t\tCurrent [Network %d]:", buckets.key_map[index]);
	775	buckets.values[index].stats (sentry);
	776	storeAppendPrintf(sentry, "\n");
	777	}
	778
	779	if (!buckets.size())
	780	storeAppendPrintf(sentry, "\t\tCurrent [All networks]: Not used yet.\n");
	781
	782	storeAppendPrintf(sentry, "\n\n");
	783	}
	784
	785	void
	786	ClassCHostPool::dump(StoreEntry *entry) const
	787	{
	788	rate()->dump (entry);
	789	}
	790
	791	void
	792	ClassCHostPool::update(int incr)
	793	{
	794	if (rate()->restore_bps == -1)
	795	return;
	796
	797	for (unsigned int i = 0; i< buckets.size(); ++i)
	798	buckets.values[i].update (*rate(), incr);
	799	}
	800
	801	void
	802	ClassCHostPool::parse()
	803	{
	804	rate()->parse();
	805	}
	806
	807	bool
	808	ClassCHostPool::keyAllocated (unsigned char const key) const
	809	{
	810	return buckets.indexUsed(buckets.findKeyIndex (key));
	811	}
	812
	813	unsigned char
	814	ClassCHostPool::makeHostKey(Ip::Address &src_addr) const
	815	{
	816	/* IPv4 required for this pool */
	817	if ( !src_addr.isIPv4() )
	818	return 1;
	819
	820	/* Temporary bypass for IPv4-only */
	821	struct in_addr host;
	822	src_addr.getInAddr(host);
	823	return (ntohl(host.s_addr) & 0xff);
	824	}
	825
	826	unsigned int
	827	ClassCHostPool::makeKey(Ip::Address &src_addr) const
	828	{
	829	/* IPv4 required for this pool */
	830	if ( !src_addr.isIPv4() )
	831	return 1;
	832
	833	struct in_addr net;
	834	src_addr.getInAddr(net);
	835	return ( (ntohl(net.s_addr) >> 8) & 0xff);
	836	}
	837
	838	DelayIdComposite::Pointer
	839	ClassCHostPool::id(CompositeSelectionDetails &details)
	840	{
	841	if (rate()->restore_bps == -1)
	842	return new NullDelayId;
	843
	844	/* non-IPv4 are not able to provide IPv4-bitmask for this pool type key. */
	845	if ( !details.src_addr.isIPv4() )
	846	return new NullDelayId;
	847
	848	unsigned int key = makeKey (details.src_addr);
	849
	850	unsigned char host = makeHostKey (details.src_addr);
	851
	852	unsigned char hostIndex;
	853
	854	unsigned char netIndex;
	855
	856	if (keyAllocated (key))
	857	netIndex = buckets.findKeyIndex(key);
	858	else
	859	netIndex = buckets.insert (key);
	860
	861	hostIndex = buckets.values[netIndex].hostPosition (*rate(), host);
	862
	863	return new Id (this, netIndex, hostIndex);
	864	}
	865
	866	ClassCHostPool::Id::Id (ClassCHostPool::Pointer aPool, unsigned char aNet, unsigned char aHost) : theClassCHost (aPool), theNet (aNet), theHost (aHost)
	867	{}
	868
	869	int
	870	ClassCHostPool::Id::bytesWanted (int min, int max) const
	871	{
	872	return theClassCHost->buckets.values[theNet].individuals.values[theHost].bytesWanted (min, max);
	873	}
	874
	875	void
	876	ClassCHostPool::Id::bytesIn(int qty)
	877	{
	878	theClassCHost->buckets.values[theNet].individuals.values[theHost].bytesIn (qty);
	879	}
	880
	881	#endif /* USE_DELAY_POOLS */
	882