src/wccp.cc

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