src/wccp.cc

   1 /*
   2  * Copyright (C) 1996-2018 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 80    WCCP Support */
  10
  11 #include "squid.h"
  12
  13 #if USE_WCCP
  14 #include "comm.h"
  15 #include "comm/Connection.h"
  16 #include "comm/Loops.h"
  17 #include "event.h"
  18 #include "fatal.h"
  19 #include "SquidConfig.h"
  20
  21 #define WCCP_PORT 2048
  22 #define WCCP_REVISION 0
  23 #define WCCP_ACTIVE_CACHES 32
  24 #define WCCP_HASH_SIZE 32
  25 #define WCCP_BUCKETS 256
  26 #define WCCP_CACHE_LEN 4
  27
  28 #define WCCP_HERE_I_AM 7
  29 #define WCCP_I_SEE_YOU 8
  30 #define WCCP_ASSIGN_BUCKET 9
  31
  32 struct wccp_here_i_am_t {
  33     int type;
  34     int version;
  35     int revision;
  36     char hash[WCCP_HASH_SIZE];
  37     int reserved;
  38     int id;
  39 };
  40
  41 struct wccp_cache_entry_t {
  42     struct in_addr ip_addr;  // WCCP on-the-wire in 32-bit IPv4-only.
  43     int revision;
  44     char hash[WCCP_HASH_SIZE];
  45     int reserved;
  46 };
  47
  48 struct wccp_i_see_you_t {
  49     int32_t type;
  50     int32_t version;
  51     int32_t change;
  52     int32_t id;
  53     int32_t number;
  54
  55     struct wccp_cache_entry_t wccp_cache_entry[WCCP_ACTIVE_CACHES];
  56 };
  57
  58 struct wccp_assign_bucket_t {
  59     int type;
  60     int id;
  61     int number;
  62 };
  63
  64 static int theWccpConnection = -1;
  65
  66 static struct wccp_here_i_am_t wccp_here_i_am;
  67
  68 static struct wccp_i_see_you_t wccp_i_see_you;
  69 static int last_change;
  70 static int last_id;
  71 static int last_assign_buckets_change;
  72 static unsigned int number_caches;
  73
  74 static Ip::Address local_ip;
  75
  76 static PF wccpHandleUdp;
  77 static int wccpLowestIP(void);
  78 static EVH wccpHereIam;
  79 static void wccpAssignBuckets(void);
  80
  81 /*
  82  * The functions used during startup:
  83  * wccpInit
  84  * wccpConnectionOpen
  85  * wccpConnectionShutdown
  86  * wccpConnectionClose
  87  */
  88
  89 void
  90 wccpInit(void)
  91 {
  92     debugs(80, 5, "wccpInit: Called");
  93     memset(&wccp_here_i_am, '\0', sizeof(wccp_here_i_am));
  94     wccp_here_i_am.type = htonl(WCCP_HERE_I_AM);
  95     wccp_here_i_am.version = htonl(Config.Wccp.version);
  96     wccp_here_i_am.revision = htonl(WCCP_REVISION);
  97     last_change = 0;
  98     last_id = 0;
  99     last_assign_buckets_change = 0;
 100     number_caches = 0;
 101
 102     if (!Config.Wccp.router.isAnyAddr())
 103         if (!eventFind(wccpHereIam, NULL))
 104             eventAdd("wccpHereIam", wccpHereIam, NULL, 5.0, 1);
 105 }
 106
 107 void
 108 wccpConnectionOpen(void)
 109 {
 110     debugs(80, 5, "wccpConnectionOpen: Called");
 111
 112     if (Config.Wccp.router.isAnyAddr()) {
 113         debugs(80, 2, "WCCPv1 disabled.");
 114         return;
 115     }
 116
 117     if ( !Config.Wccp.router.setIPv4() ) {
 118         debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Router " << Config.Wccp.router << " is not an IPv4 address.");
 119         return;
 120     }
 121
 122     if ( !Config.Wccp.address.setIPv4() ) {
 123         debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Local address " << Config.Wccp.address << " is not an IPv4 address.");
 124         return;
 125     }
 126
 127     Config.Wccp.address.port(WCCP_PORT);
 128     Config.Wccp.router.port(WCCP_PORT);
 129
 130     theWccpConnection = comm_open_listener(SOCK_DGRAM,
 131                                            IPPROTO_UDP,
 132                                            Config.Wccp.address,
 133                                            COMM_NONBLOCKING,
 134                                            "WCCP Socket");
 135
 136     if (theWccpConnection < 0)
 137         fatal("Cannot open WCCP Port");
 138
 139     Comm::SetSelect(theWccpConnection, COMM_SELECT_READ, wccpHandleUdp, NULL, 0);
 140
 141     debugs(80, DBG_IMPORTANT, "Accepting WCCPv1 messages on " << Config.Wccp.address << ", FD " << theWccpConnection << ".");
 142
 143     // Sadly WCCP only does IPv4
 144
 145     struct sockaddr_in router;
 146     Config.Wccp.router.getSockAddr(router);
 147     if (connect(theWccpConnection, (struct sockaddr*)&router, sizeof(router)))
 148         fatal("Unable to connect WCCP out socket");
 149
 150     struct sockaddr_in local;
 151     memset(&local, '\0', sizeof(local));
 152     socklen_t slen = sizeof(local);
 153     if (getsockname(theWccpConnection, (struct sockaddr*)&local, &slen))
 154         fatal("Unable to getsockname on WCCP out socket");
 155
 156     local_ip = local;
 157 }
 158
 159 void
 160 wccpConnectionClose(void)
 161 {
 162     if (theWccpConnection > -1) {
 163         debugs(80, DBG_IMPORTANT, "FD " << theWccpConnection << " Closing WCCPv1 socket");
 164         comm_close(theWccpConnection);
 165         theWccpConnection = -1;
 166     }
 167 }
 168
 169 /*
 170  * Functions for handling the requests.
 171  */
 172
 173 /*
 174  * Accept the UDP packet
 175  */
 176 static void
 177 wccpHandleUdp(int sock, void *)
 178 {
 179     Ip::Address from;
 180     int len;
 181
 182     debugs(80, 6, "wccpHandleUdp: Called.");
 183
 184     Comm::SetSelect(sock, COMM_SELECT_READ, wccpHandleUdp, NULL, 0);
 185
 186     memset(&wccp_i_see_you, '\0', sizeof(wccp_i_see_you));
 187
 188     len = comm_udp_recvfrom(sock,
 189                             (void *) &wccp_i_see_you,
 190                             sizeof(wccp_i_see_you),
 191                             0,
 192                             from);
 193     debugs(80, 3, "wccpHandleUdp: " << len << " bytes WCCP pkt from " << from <<
 194            ": type=" <<
 195            (unsigned) ntohl(wccp_i_see_you.type) << ", version=" <<
 196            (unsigned) ntohl(wccp_i_see_you.version) << ", change=" <<
 197            (unsigned) ntohl(wccp_i_see_you.change) << ", id=" <<
 198            (unsigned) ntohl(wccp_i_see_you.id) << ", number=" <<
 199            (unsigned) ntohl(wccp_i_see_you.number));
 200
 201     if (len < 0)
 202         return;
 203
 204     if (from != Config.Wccp.router)
 205         return;
 206
 207     if ((unsigned) ntohl(wccp_i_see_you.version) != (unsigned) Config.Wccp.version)
 208         return;
 209
 210     if (ntohl(wccp_i_see_you.type) != WCCP_I_SEE_YOU)
 211         return;
 212
 213     if (ntohl(wccp_i_see_you.number) > WCCP_ACTIVE_CACHES) {
 214         debugs(80, DBG_IMPORTANT, "Ignoring WCCP_I_SEE_YOU from " <<
 215                from << " with number of caches set to " <<
 216                (int) ntohl(wccp_i_see_you.number));
 217
 218         return;
 219     }
 220
 221     last_id = wccp_i_see_you.id;
 222
 223     if ((0 == last_change) && (number_caches == (unsigned) ntohl(wccp_i_see_you.number))) {
 224         if (last_assign_buckets_change == wccp_i_see_you.change) {
 225             /*
 226              * After a WCCP_ASSIGN_BUCKET message, the router should
 227              * update the change value.  If not, maybe the route didn't
 228              * receive our WCCP_ASSIGN_BUCKET message, so send it again.
 229              *
 230              * Don't update change here.  Instead, fall through to
 231              * the next block to call wccpAssignBuckets() again.
 232              */
 233             (void) 0;
 234         } else {
 235             last_change = wccp_i_see_you.change;
 236             return;
 237         }
 238     }
 239
 240     if (last_change != wccp_i_see_you.change) {
 241         last_change = wccp_i_see_you.change;
 242
 243         if (wccpLowestIP() && wccp_i_see_you.number) {
 244             last_assign_buckets_change = last_change;
 245             wccpAssignBuckets();
 246         }
 247     }
 248 }
 249
 250 static int
 251 wccpLowestIP(void)
 252 {
 253     unsigned int loop;
 254     int found = 0;
 255
 256     /*
 257      * We sanity checked wccp_i_see_you.number back in wccpHandleUdp()
 258      */
 259
 260     for (loop = 0; loop < (unsigned) ntohl(wccp_i_see_you.number); ++loop) {
 261         assert(loop < WCCP_ACTIVE_CACHES);
 262
 263         if (local_ip > wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
 264             return 0;
 265
 266         if (local_ip == wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
 267             found = 1;
 268     }
 269
 270     return found;
 271 }
 272
 273 static void
 274 wccpHereIam(void *)
 275 {
 276     debugs(80, 6, "wccpHereIam: Called");
 277
 278     wccp_here_i_am.id = last_id;
 279     double interval = 10.0; // TODO: make this configurable, possibly negotiate with the router.
 280     ssize_t sent = comm_udp_send(theWccpConnection, &wccp_here_i_am, sizeof(wccp_here_i_am), 0);
 281
 282     // if we failed to send the whole lot, try again at a shorter interval (20%)
 283     if (sent != sizeof(wccp_here_i_am)) {
 284         int xerrno = errno;
 285         debugs(80, 2, "ERROR: failed to send WCCP HERE_I_AM packet: " << xstrerr(xerrno));
 286         interval = 2.0;
 287     }
 288
 289     if (!eventFind(wccpHereIam, NULL))
 290         eventAdd("wccpHereIam", wccpHereIam, NULL, interval, 1);
 291 }
 292
 293 static void
 294 wccpAssignBuckets(void)
 295 {
 296
 297     struct wccp_assign_bucket_t *wccp_assign_bucket;
 298     int wab_len;
 299     char *buckets;
 300     int buckets_per_cache;
 301     unsigned int loop;
 302     int bucket = 0;
 303     int *caches;
 304     int cache_len;
 305     char *buf;
 306
 307     debugs(80, 6, "wccpAssignBuckets: Called");
 308     number_caches = ntohl(wccp_i_see_you.number);
 309
 310     assert(number_caches > 0);
 311     assert(number_caches <= WCCP_ACTIVE_CACHES);
 312
 313     wab_len = sizeof(struct wccp_assign_bucket_t);
 314
 315     cache_len = WCCP_CACHE_LEN * number_caches;
 316
 317     buf = (char *)xmalloc(wab_len +
 318                           WCCP_BUCKETS +
 319                           cache_len);
 320
 321     wccp_assign_bucket = (struct wccp_assign_bucket_t *) buf;
 322
 323     caches = (int *) (buf + wab_len);
 324
 325     buckets = buf + wab_len + cache_len;
 326
 327     memset(wccp_assign_bucket, '\0', sizeof(*wccp_assign_bucket));
 328
 329     memset(buckets, 0xFF, WCCP_BUCKETS);
 330
 331     buckets_per_cache = WCCP_BUCKETS / number_caches;
 332
 333     for (loop = 0; loop < number_caches; ++loop) {
 334         int i;
 335         memcpy(&caches[loop],
 336                &wccp_i_see_you.wccp_cache_entry[loop].ip_addr,
 337                sizeof(*caches));
 338
 339         for (i = 0; i < buckets_per_cache; ++i) {
 340             assert(bucket < WCCP_BUCKETS);
 341             buckets[bucket] = loop;
 342             ++bucket;
 343         }
 344     }
 345
 346     while (bucket < WCCP_BUCKETS) {
 347         buckets[bucket] = number_caches - 1;
 348         ++bucket;
 349     }
 350
 351     wccp_assign_bucket->type = htonl(WCCP_ASSIGN_BUCKET);
 352     wccp_assign_bucket->id = wccp_i_see_you.id;
 353     wccp_assign_bucket->number = wccp_i_see_you.number;
 354
 355     comm_udp_send(theWccpConnection,
 356                   buf,
 357                   wab_len + WCCP_BUCKETS + cache_len,
 358                   0);
 359     last_change = 0;
 360     xfree(buf);
 361 }
 362
 363 #endif /* USE_WCCP */
 364