wccp.cc

Go to the documentation of this file.

/*
 * 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 80    WCCP Support */
 
#include "squid.h"
 
#if USE_WCCP
#include "base/RunnersRegistry.h"
#include "comm.h"
#include "comm/Connection.h"
#include "comm/Loops.h"
#include "compat/socket.h"
#include "event.h"
#include "fatal.h"
#include "SquidConfig.h"
#include "tools.h"
 
#define WCCP_PORT 2048
#define WCCP_REVISION 0
#define WCCP_ACTIVE_CACHES 32
#define WCCP_HASH_SIZE 32
#define WCCP_BUCKETS 256
#define WCCP_CACHE_LEN 4
 
#define WCCP_HERE_I_AM 7
#define WCCP_I_SEE_YOU 8
#define WCCP_ASSIGN_BUCKET 9
 
struct wccp_here_i_am_t {
    int type;
    int version;
    int revision;
    char hash[WCCP_HASH_SIZE];
    int reserved;
    int id;
};
 
struct wccp_cache_entry_t {
    struct in_addr ip_addr;  // WCCP on-the-wire in 32-bit IPv4-only.
    int revision;
    char hash[WCCP_HASH_SIZE];
    int reserved;
};
 
struct wccp_i_see_you_t {
    int32_t type;
    int32_t version;
    int32_t change;
    int32_t id;
    int32_t number;
 
    struct wccp_cache_entry_t wccp_cache_entry[WCCP_ACTIVE_CACHES];
};
 
struct wccp_assign_bucket_t {
    int type;
    int id;
    int number;
};
 
static int theWccpConnection = -1;
 
static struct wccp_here_i_am_t wccp_here_i_am;
 
static struct wccp_i_see_you_t wccp_i_see_you;
static int last_change;
static int last_id;
static int last_assign_buckets_change;
static unsigned int number_caches;
 
static Ip::Address local_ip;
 
static PF wccpHandleUdp;
static int wccpLowestIP(void);
static EVH wccpHereIam;
static void wccpAssignBuckets(void);
 
static void
wccpInit(void)
{
    if (!IamPrimaryProcess())
        return;
 
    debugs(80, 5, "wccpInit: Called");
    memset(&wccp_here_i_am, '\0', sizeof(wccp_here_i_am));
    wccp_here_i_am.type = htonl(WCCP_HERE_I_AM);
    wccp_here_i_am.version = htonl(Config.Wccp.version);
    wccp_here_i_am.revision = htonl(WCCP_REVISION);
    last_change = 0;
    last_id = 0;
    last_assign_buckets_change = 0;
    number_caches = 0;
 
    if (!Config.Wccp.router.isAnyAddr())
        if (!eventFind(wccpHereIam, nullptr))
            eventAdd("wccpHereIam", wccpHereIam, nullptr, 5.0, 1);
}
 
static void
wccpConnectionOpen(void)
{
    if (!IamPrimaryProcess())
        return;
 
    debugs(80, 5, "wccpConnectionOpen: Called");
 
    if (Config.Wccp.router.isAnyAddr()) {
        debugs(80, 2, "WCCPv1 disabled.");
        return;
    }
 
    if ( !Config.Wccp.router.setIPv4() ) {
        debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Router " << Config.Wccp.router << " is not an IPv4 address.");
        return;
    }
 
    if ( !Config.Wccp.address.setIPv4() ) {
        debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Local address " << Config.Wccp.address << " is not an IPv4 address.");
        return;
    }
 
    Config.Wccp.address.port(WCCP_PORT);
    Config.Wccp.router.port(WCCP_PORT);
 
    theWccpConnection = comm_open_listener(SOCK_DGRAM,
                                           IPPROTO_UDP,
                                           Config.Wccp.address,
                                           COMM_NONBLOCKING,
                                           "WCCP Socket");
 
    if (theWccpConnection < 0)
        fatal("Cannot open WCCP Port");
 
    Comm::SetSelect(theWccpConnection, COMM_SELECT_READ, wccpHandleUdp, nullptr, 0);
 
    debugs(80, DBG_IMPORTANT, "Accepting WCCPv1 messages on " << Config.Wccp.address << ", FD " << theWccpConnection << ".");
 
    // Sadly WCCP only does IPv4
 
    struct sockaddr_in router;
    Config.Wccp.router.getSockAddr(router);
    if (xconnect(theWccpConnection, (struct sockaddr*)&router, sizeof(router)))
        fatal("Unable to connect WCCP out socket");
 
    struct sockaddr_in local;
    memset(&local, '\0', sizeof(local));
    socklen_t slen = sizeof(local);
    if (xgetsockname(theWccpConnection, (struct sockaddr*)&local, &slen))
        fatal("Unable to getsockname on WCCP out socket");
 
    local_ip = local;
}
 
static void
wccpConnectionClose(void)
{
    if (!IamPrimaryProcess())
        return;
 
    if (theWccpConnection > -1) {
        debugs(80, DBG_IMPORTANT, "FD " << theWccpConnection << " Closing WCCPv1 socket");
        comm_close(theWccpConnection);
        theWccpConnection = -1;
    }
}
 
class WccpRr : public RegisteredRunner
{
public:
    void useConfig() override { wccpInit(); wccpConnectionOpen(); }
    void startReconfigure() override { wccpConnectionClose(); }
    void syncConfig() override { wccpConnectionOpen(); }
    void startShutdown() override { wccpConnectionClose(); }
};
DefineRunnerRegistrator(WccpRr);
 
/*
 * Functions for handling the requests.
 */
 
/*
 * Accept the UDP packet
 */
static void
wccpHandleUdp(int sock, void *)
{
    Ip::Address from;
    int len;
 
    debugs(80, 6, "wccpHandleUdp: Called.");
 
    Comm::SetSelect(sock, COMM_SELECT_READ, wccpHandleUdp, nullptr, 0);
 
    memset(&wccp_i_see_you, '\0', sizeof(wccp_i_see_you));
 
    len = comm_udp_recvfrom(sock,
                            (void *) &wccp_i_see_you,
                            sizeof(wccp_i_see_you),
                            0,
                            from);
    debugs(80, 3, "wccpHandleUdp: " << len << " bytes WCCP pkt from " << from <<
           ": type=" <<
           (unsigned) ntohl(wccp_i_see_you.type) << ", version=" <<
           (unsigned) ntohl(wccp_i_see_you.version) << ", change=" <<
           (unsigned) ntohl(wccp_i_see_you.change) << ", id=" <<
           (unsigned) ntohl(wccp_i_see_you.id) << ", number=" <<
           (unsigned) ntohl(wccp_i_see_you.number));
 
    if (len < 0)
        return;
 
    if (from != Config.Wccp.router)
        return;
 
    if ((unsigned) ntohl(wccp_i_see_you.version) != (unsigned) Config.Wccp.version)
        return;
 
    if (ntohl(wccp_i_see_you.type) != WCCP_I_SEE_YOU)
        return;
 
    if (ntohl(wccp_i_see_you.number) > WCCP_ACTIVE_CACHES) {
        debugs(80, DBG_IMPORTANT, "Ignoring WCCP_I_SEE_YOU from " <<
               from << " with number of caches set to " <<
               (int) ntohl(wccp_i_see_you.number));
 
        return;
    }
 
    last_id = wccp_i_see_you.id;
 
    if ((0 == last_change) && (number_caches == (unsigned) ntohl(wccp_i_see_you.number))) {
        if (last_assign_buckets_change == wccp_i_see_you.change) {
            /*
             * After a WCCP_ASSIGN_BUCKET message, the router should
             * update the change value.  If not, maybe the route didn't
             * receive our WCCP_ASSIGN_BUCKET message, so send it again.
             *
             * Don't update change here.  Instead, fall through to
             * the next block to call wccpAssignBuckets() again.
             */
            (void) 0;
        } else {
            last_change = wccp_i_see_you.change;
            return;
        }
    }
 
    if (last_change != wccp_i_see_you.change) {
        last_change = wccp_i_see_you.change;
 
        if (wccpLowestIP() && wccp_i_see_you.number) {
            last_assign_buckets_change = last_change;
            wccpAssignBuckets();
        }
    }
}
 
static int
wccpLowestIP(void)
{
    unsigned int loop;
    int found = 0;
 
    /*
     * We sanity checked wccp_i_see_you.number back in wccpHandleUdp()
     */
 
    for (loop = 0; loop < (unsigned) ntohl(wccp_i_see_you.number); ++loop) {
        assert(loop < WCCP_ACTIVE_CACHES);
 
        if (local_ip > wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
            return 0;
 
        if (local_ip == wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
            found = 1;
    }
 
    return found;
}
 
static void
wccpHereIam(void *)
{
    debugs(80, 6, "wccpHereIam: Called");
 
    wccp_here_i_am.id = last_id;
    double interval = 10.0; // TODO: make this configurable, possibly negotiate with the router.
    ssize_t sent = comm_udp_send(theWccpConnection, &wccp_here_i_am, sizeof(wccp_here_i_am), 0);
 
    // if we failed to send the whole lot, try again at a shorter interval (20%)
    if (sent != sizeof(wccp_here_i_am)) {
        int xerrno = errno;
        debugs(80, 2, "ERROR: failed to send WCCP HERE_I_AM packet: " << xstrerr(xerrno));
        interval = 2.0;
    }
 
    if (!eventFind(wccpHereIam, nullptr))
        eventAdd("wccpHereIam", wccpHereIam, nullptr, interval, 1);
}
 
static void
wccpAssignBuckets(void)
{
 
    struct wccp_assign_bucket_t *wccp_assign_bucket;
    int wab_len;
    char *buckets;
    int buckets_per_cache;
    unsigned int loop;
    int bucket = 0;
    int *caches;
    int cache_len;
    char *buf;
 
    debugs(80, 6, "wccpAssignBuckets: Called");
    number_caches = ntohl(wccp_i_see_you.number);
 
    assert(number_caches > 0);
    assert(number_caches <= WCCP_ACTIVE_CACHES);
 
    wab_len = sizeof(struct wccp_assign_bucket_t);
 
    cache_len = WCCP_CACHE_LEN * number_caches;
 
    buf = (char *)xmalloc(wab_len +
                          WCCP_BUCKETS +
                          cache_len);
 
    wccp_assign_bucket = (struct wccp_assign_bucket_t *) buf;
 
    caches = (int *) (buf + wab_len);
 
    buckets = buf + wab_len + cache_len;
 
    memset(wccp_assign_bucket, '\0', sizeof(*wccp_assign_bucket));
 
    memset(buckets, 0xFF, WCCP_BUCKETS);
 
    buckets_per_cache = WCCP_BUCKETS / number_caches;
 
    for (loop = 0; loop < number_caches; ++loop) {
        int i;
        memcpy(&caches[loop],
               &wccp_i_see_you.wccp_cache_entry[loop].ip_addr,
               sizeof(*caches));
 
        for (i = 0; i < buckets_per_cache; ++i) {
            assert(bucket < WCCP_BUCKETS);
            buckets[bucket] = loop;
            ++bucket;
        }
    }
 
    while (bucket < WCCP_BUCKETS) {
        buckets[bucket] = number_caches - 1;
        ++bucket;
    }
 
    wccp_assign_bucket->type = htonl(WCCP_ASSIGN_BUCKET);
    wccp_assign_bucket->id = wccp_i_see_you.id;
    wccp_assign_bucket->number = wccp_i_see_you.number;
 
    comm_udp_send(theWccpConnection,
                  buf,
                  wab_len + WCCP_BUCKETS + cache_len,
                  0);
    last_change = 0;
    xfree(buf);
}
 
#endif /* USE_WCCP */