src/peer_digest.cc

   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 72    Peer Digest Routines */
  10
  11 #include "squid.h"
  12 #if USE_CACHE_DIGESTS
  13 #include "base/IoManip.h"
  14 #include "CacheDigest.h"
  15 #include "CachePeer.h"
  16 #include "event.h"
  17 #include "FwdState.h"
  18 #include "globals.h"
  19 #include "HttpReply.h"
  20 #include "HttpRequest.h"
  21 #include "internal.h"
  22 #include "MemObject.h"
  23 #include "mime_header.h"
  24 #include "neighbors.h"
  25 #include "PeerDigest.h"
  26 #include "Store.h"
  27 #include "store_key_md5.h"
  28 #include "StoreClient.h"
  29 #include "tools.h"
  30 #include "util.h"
  31
  32 /* local types */
  33
  34 /* local prototypes */
  35 static time_t peerDigestIncDelay(const PeerDigest * pd);
  36 static time_t peerDigestNewDelay(const StoreEntry * e);
  37 static void peerDigestSetCheck(PeerDigest * pd, time_t delay);
  38 static EVH peerDigestCheck;
  39 static void peerDigestRequest(PeerDigest * pd);
  40 static STCB peerDigestHandleReply;
  41 static int peerDigestFetchReply(void *, char *, ssize_t);
  42 int peerDigestSwapInCBlock(void *, char *, ssize_t);
  43 int peerDigestSwapInMask(void *, char *, ssize_t);
  44 static int peerDigestFetchedEnough(DigestFetchState * fetch, char *buf, ssize_t size, const char *step_name);
  45 static void finishAndDeleteFetch(DigestFetchState *, const char *reason, bool sawError);
  46 static void peerDigestFetchSetStats(DigestFetchState * fetch);
  47 static int peerDigestSetCBlock(PeerDigest * pd, const char *buf);
  48 static int peerDigestUseful(const PeerDigest * pd);
  49
  50 /* local constants */
  51 Version const CacheDigestVer = { 5, 3 };
  52
  53 #define StoreDigestCBlockSize sizeof(StoreDigestCBlock)
  54
  55 /* min interval for requesting digests from a given peer */
  56 static const time_t PeerDigestReqMinGap = 5 * 60;   /* seconds */
  57 /* min interval for requesting digests (cumulative request stream) */
  58 static const time_t GlobDigestReqMinGap = 1 * 60;   /* seconds */
  59
  60 /* local vars */
  61
  62 static time_t pd_last_req_time = 0; /* last call to Check */
  63
  64 PeerDigest::PeerDigest(CachePeer * const p):
  65     peer(p),
  66     host(peer->host) // if peer disappears, we will know its name
  67 {
  68     times.initialized = squid_curtime;
  69 }
  70
  71 CBDATA_CLASS_INIT(PeerDigest);
  72
  73 CBDATA_CLASS_INIT(DigestFetchState);
  74
  75 DigestFetchState::DigestFetchState(PeerDigest *aPd, HttpRequest *req) :
  76     pd(aPd),
  77     entry(nullptr),
  78     old_entry(nullptr),
  79     sc(nullptr),
  80     old_sc(nullptr),
  81     request(req),
  82     offset(0),
  83     mask_offset(0),
  84     start_time(squid_curtime),
  85     resp_time(0),
  86     expires(0),
  87     bufofs(0),
  88     state(DIGEST_READ_REPLY)
  89 {
  90     HTTPMSGLOCK(request);
  91
  92     sent.msg = 0;
  93     sent.bytes = 0;
  94
  95     recv.msg = 0;
  96     recv.bytes = 0;
  97
  98     *buf = 0;
  99 }
 100
 101 DigestFetchState::~DigestFetchState()
 102 {
 103     if (old_entry) {
 104         debugs(72, 3, "deleting old entry");
 105         storeUnregister(old_sc, old_entry, this);
 106         old_entry->releaseRequest();
 107         old_entry->unlock("DigestFetchState destructed old");
 108         old_entry = nullptr;
 109     }
 110
 111     /* unlock everything */
 112     storeUnregister(sc, entry, this);
 113
 114     entry->unlock("DigestFetchState destructed");
 115     entry = nullptr;
 116
 117     HTTPMSGUNLOCK(request);
 118 }
 119
 120 PeerDigest::~PeerDigest()
 121 {
 122     if (times.next_check && eventFind(peerDigestCheck, this))
 123         eventDelete(peerDigestCheck, this);
 124     delete cd;
 125     // req_result pointer is not owned by us
 126 }
 127
 128 /* called by peer to indicate that somebody actually needs this digest */
 129 void
 130 peerDigestNeeded(PeerDigest * pd)
 131 {
 132     assert(pd);
 133     assert(!pd->flags.needed);
 134     assert(!pd->cd);
 135
 136     pd->flags.needed = true;
 137     pd->times.needed = squid_curtime;
 138     peerDigestSetCheck(pd, 0);  /* check asap */
 139 }
 140
 141 /* increment retry delay [after an unsuccessful attempt] */
 142 static time_t
 143 peerDigestIncDelay(const PeerDigest * pd)
 144 {
 145     assert(pd);
 146     return pd->times.retry_delay > 0 ?
 147            2 * pd->times.retry_delay :  /* exponential backoff */
 148            PeerDigestReqMinGap; /* minimal delay */
 149 }
 150
 151 /* artificially increases Expires: setting to avoid race conditions
 152  * returns the delay till that [increased] expiration time */
 153 static time_t
 154 peerDigestNewDelay(const StoreEntry * e)
 155 {
 156     assert(e);
 157
 158     if (e->expires > 0)
 159         return e->expires + PeerDigestReqMinGap - squid_curtime;
 160
 161     return PeerDigestReqMinGap;
 162 }
 163
 164 /* registers next digest verification */
 165 static void
 166 peerDigestSetCheck(PeerDigest * pd, time_t delay)
 167 {
 168     eventAdd("peerDigestCheck", peerDigestCheck, pd, (double) delay, 1);
 169     pd->times.next_check = squid_curtime + delay;
 170     debugs(72, 3, "peerDigestSetCheck: will check peer " << pd->host << " in " << delay << " secs");
 171 }
 172
 173 /* callback for eventAdd() (with peer digest locked)
 174  * request new digest if our copy is too old or if we lack one;
 175  * schedule next check otherwise */
 176 static void
 177 peerDigestCheck(void *data)
 178 {
 179     PeerDigest *pd = (PeerDigest *)data;
 180     time_t req_time;
 181
 182     assert(!pd->flags.requested);
 183
 184     pd->times.next_check = 0;   /* unknown */
 185
 186     Assure(pd->peer.valid());
 187
 188     debugs(72, 3, "cache_peer " << RawPointer(pd->peer).orNil());
 189     debugs(72, 3, "peerDigestCheck: time: " << squid_curtime <<
 190            ", last received: " << (long int) pd->times.received << "  (" <<
 191            std::showpos << (int) (squid_curtime - pd->times.received) << ")");
 192
 193     /* decide when we should send the request:
 194      * request now unless too close to other requests */
 195     req_time = squid_curtime;
 196
 197     /* per-peer limit */
 198
 199     if (req_time - pd->times.received < PeerDigestReqMinGap) {
 200         debugs(72, 2, "peerDigestCheck: " << pd->host <<
 201                ", avoiding close peer requests (" <<
 202                (int) (req_time - pd->times.received) << " < " <<
 203                (int) PeerDigestReqMinGap << " secs).");
 204
 205         req_time = pd->times.received + PeerDigestReqMinGap;
 206     }
 207
 208     /* global limit */
 209     if (req_time - pd_last_req_time < GlobDigestReqMinGap) {
 210         debugs(72, 2, "peerDigestCheck: " << pd->host <<
 211                ", avoiding close requests (" <<
 212                (int) (req_time - pd_last_req_time) << " < " <<
 213                (int) GlobDigestReqMinGap << " secs).");
 214
 215         req_time = pd_last_req_time + GlobDigestReqMinGap;
 216     }
 217
 218     if (req_time <= squid_curtime)
 219         peerDigestRequest(pd);  /* will set pd->flags.requested */
 220     else
 221         peerDigestSetCheck(pd, req_time - squid_curtime);
 222 }
 223
 224 /* ask store for a digest */
 225 static void
 226 peerDigestRequest(PeerDigest * pd)
 227 {
 228     const auto p = pd->peer.get(); // TODO: Replace with a reference.
 229     StoreEntry *e, *old_e;
 230     char *url = nullptr;
 231     HttpRequest *req;
 232     StoreIOBuffer tempBuffer;
 233
 234     pd->req_result = nullptr;
 235     pd->flags.requested = true;
 236
 237     /* compute future request components */
 238
 239     if (p->digest_url)
 240         url = xstrdup(p->digest_url);
 241     else
 242         url = xstrdup(internalRemoteUri(p->secure.encryptTransport, p->host, p->http_port, "/squid-internal-periodic/", SBuf(StoreDigestFileName)));
 243     debugs(72, 2, url);
 244
 245     const auto mx = MasterXaction::MakePortless<XactionInitiator::initCacheDigest>();
 246     req = HttpRequest::FromUrlXXX(url, mx);
 247
 248     assert(req);
 249
 250     /* add custom headers */
 251     assert(!req->header.len);
 252
 253     req->header.putStr(Http::HdrType::ACCEPT, StoreDigestMimeStr);
 254
 255     req->header.putStr(Http::HdrType::ACCEPT, "text/html");
 256
 257     if (p->login &&
 258             p->login[0] != '*' &&
 259             strcmp(p->login, "PASS") != 0 &&
 260             strcmp(p->login, "PASSTHRU") != 0 &&
 261             strncmp(p->login, "NEGOTIATE",9) != 0 &&
 262             strcmp(p->login, "PROXYPASS") != 0) {
 263         req->url.userInfo(SBuf(p->login)); // XXX: performance regression make peer login SBuf as well.
 264     }
 265     /* create fetch state structure */
 266     DigestFetchState *fetch = new DigestFetchState(pd, req);
 267
 268     /* update timestamps */
 269     pd->times.requested = squid_curtime;
 270     pd_last_req_time = squid_curtime;
 271     req->flags.cachable.support(); // prevent RELEASE_REQUEST in storeCreateEntry()
 272
 273     /* the rest is based on clientReplyContext::processExpired() */
 274     req->flags.refresh = true;
 275
 276     old_e = fetch->old_entry = storeGetPublicByRequest(req);
 277
 278     // XXX: Missing a hittingRequiresCollapsing() && startCollapsingOn() check.
 279     if (old_e) {
 280         debugs(72, 5, "found old " << *old_e);
 281
 282         old_e->lock("peerDigestRequest");
 283         old_e->ensureMemObject(url, url, req->method);
 284
 285         fetch->old_sc = storeClientListAdd(old_e, fetch);
 286     }
 287
 288     e = fetch->entry = storeCreateEntry(url, url, req->flags, req->method);
 289     debugs(72, 5, "created " << *e);
 290     assert(EBIT_TEST(e->flags, KEY_PRIVATE));
 291     fetch->sc = storeClientListAdd(e, fetch);
 292     /* set lastmod to trigger IMS request if possible */
 293
 294     // TODO: Also check for fetch->pd->cd presence as a precondition for sending
 295     // IMS requests because peerDigestFetchReply() does not accept 304 responses
 296     // without an in-memory cache digest.
 297     if (old_e)
 298         e->lastModified(old_e->lastModified());
 299
 300     /* push towards peer cache */
 301     FwdState::fwdStart(Comm::ConnectionPointer(), e, req);
 302
 303     tempBuffer.offset = 0;
 304
 305     tempBuffer.length = SM_PAGE_SIZE;
 306
 307     tempBuffer.data = fetch->buf;
 308
 309     storeClientCopy(fetch->sc, e, tempBuffer,
 310                     peerDigestHandleReply, fetch);
 311
 312     safe_free(url);
 313 }
 314
 315 /* Handle the data copying .. */
 316
 317 /*
 318  * This routine handles the copy data and then redirects the
 319  * copy to a bunch of subfunctions depending upon the copy state.
 320  * It also tracks the buffer offset and "seen", since I'm actually
 321  * not interested in rewriting everything to suit my little idea.
 322  */
 323 static void
 324 peerDigestHandleReply(void *data, StoreIOBuffer receivedData)
 325 {
 326     DigestFetchState *fetch = (DigestFetchState *)data;
 327     int retsize = -1;
 328     digest_read_state_t prevstate;
 329     int newsize;
 330
 331     if (receivedData.flags.error) {
 332         finishAndDeleteFetch(fetch, "failure loading digest reply from Store", true);
 333         return;
 334     }
 335
 336     if (!fetch->pd) {
 337         finishAndDeleteFetch(fetch, "digest disappeared while loading digest reply from Store", true);
 338         return;
 339     }
 340
 341     /* The existing code assumes that the received pointer is
 342      * where we asked the data to be put
 343      */
 344     assert(!receivedData.data || fetch->buf + fetch->bufofs == receivedData.data);
 345
 346     /* Update the buffer size */
 347     fetch->bufofs += receivedData.length;
 348
 349     assert(fetch->bufofs <= SM_PAGE_SIZE);
 350
 351     /* If we've fetched enough, return */
 352
 353     if (peerDigestFetchedEnough(fetch, fetch->buf, fetch->bufofs, "peerDigestHandleReply"))
 354         return;
 355
 356     /* Call the right function based on the state */
 357     /* (Those functions will update the state if needed) */
 358
 359     /* Give us a temporary reference. Some of the calls we make may
 360      * try to destroy the fetch structure, and we like to know if they
 361      * do
 362      */
 363     CbcPointer<DigestFetchState> tmpLock = fetch;
 364
 365     /* Repeat this loop until we're out of data OR the state changes */
 366     /* (So keep going if the state has changed and we still have data */
 367     do {
 368         prevstate = fetch->state;
 369
 370         switch (fetch->state) {
 371
 372         case DIGEST_READ_REPLY:
 373             retsize = peerDigestFetchReply(fetch, fetch->buf, fetch->bufofs);
 374             break;
 375
 376         case DIGEST_READ_CBLOCK:
 377             retsize = peerDigestSwapInCBlock(fetch, fetch->buf, fetch->bufofs);
 378             break;
 379
 380         case DIGEST_READ_MASK:
 381             retsize = peerDigestSwapInMask(fetch, fetch->buf, fetch->bufofs);
 382             break;
 383
 384         case DIGEST_READ_NONE:
 385             break;
 386
 387         default:
 388             fatal("Bad digest transfer mode!\n");
 389         }
 390
 391         if (retsize < 0)
 392             return;
 393
 394         /*
 395          * The returned size indicates how much of the buffer was read -
 396          * so move the remainder of the buffer to the beginning
 397          * and update the bufofs / bufsize
 398          */
 399         newsize = fetch->bufofs - retsize;
 400
 401         memmove(fetch->buf, fetch->buf + retsize, fetch->bufofs - newsize);
 402
 403         fetch->bufofs = newsize;
 404
 405     } while (cbdataReferenceValid(fetch) && prevstate != fetch->state && fetch->bufofs > 0);
 406
 407     // Check for EOF here, thus giving the parser one extra run. We could avoid this overhead by
 408     // checking at the beginning of this function. However, in this case, we would have to require
 409     // that the parser does not regard EOF as a special condition (it is true now but may change
 410     // in the future).
 411     if (fetch->sc->atEof()) {
 412         finishAndDeleteFetch(fetch, "premature end of digest reply", true);
 413         return;
 414     }
 415
 416     /* Update the copy offset */
 417     fetch->offset += receivedData.length;
 418
 419     /* Schedule another copy */
 420     if (cbdataReferenceValid(fetch)) {
 421         StoreIOBuffer tempBuffer;
 422         tempBuffer.offset = fetch->offset;
 423         tempBuffer.length = SM_PAGE_SIZE - fetch->bufofs;
 424         tempBuffer.data = fetch->buf + fetch->bufofs;
 425         storeClientCopy(fetch->sc, fetch->entry, tempBuffer,
 426                         peerDigestHandleReply, fetch);
 427     }
 428 }
 429
 430 /// handle HTTP response headers in the initial storeClientCopy() response
 431 static int
 432 peerDigestFetchReply(void *data, char *buf, ssize_t size)
 433 {
 434     DigestFetchState *fetch = (DigestFetchState *)data;
 435     const auto pd = fetch->pd.get();
 436     assert(pd && buf);
 437     assert(!fetch->offset);
 438
 439     assert(fetch->state == DIGEST_READ_REPLY);
 440
 441     if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestFetchReply"))
 442         return -1;
 443
 444     {
 445         const auto &reply = fetch->entry->mem().freshestReply();
 446         const auto status = reply.sline.status();
 447         assert(status != Http::scNone);
 448         debugs(72, 3, "peerDigestFetchReply: " << pd->host << " status: " << status <<
 449                ", expires: " << (long int) reply.expires << " (" << std::showpos <<
 450                (int) (reply.expires - squid_curtime) << ")");
 451
 452         /* this "if" is based on clientHandleIMSReply() */
 453
 454         if (status == Http::scNotModified) {
 455             /* our old entry is fine */
 456             assert(fetch->old_entry);
 457
 458             if (!fetch->old_entry->mem_obj->request)
 459                 fetch->old_entry->mem_obj->request = fetch->entry->mem_obj->request;
 460
 461             assert(fetch->old_entry->mem_obj->request);
 462
 463             if (!Store::Root().updateOnNotModified(fetch->old_entry, *fetch->entry)) {
 464                 finishAndDeleteFetch(fetch, "header update failure after a 304 response", true);
 465                 return -1;
 466             }
 467
 468             /* get rid of 304 reply */
 469             storeUnregister(fetch->sc, fetch->entry, fetch);
 470
 471             fetch->entry->unlock("peerDigestFetchReply 304");
 472
 473             fetch->entry = fetch->old_entry;
 474
 475             fetch->old_entry = nullptr;
 476
 477             /* preserve request -- we need its size to update counters */
 478             /* requestUnlink(r); */
 479             /* fetch->entry->mem_obj->request = nullptr; */
 480
 481             if (!fetch->pd->cd) {
 482                 finishAndDeleteFetch(fetch, "304 without the old in-memory digest", true);
 483                 return -1;
 484             }
 485
 486             // stay with the old in-memory digest
 487             finishAndDeleteFetch(fetch, "Not modified", false);
 488             return -1;
 489         } else if (status == Http::scOkay) {
 490             /* get rid of old entry if any */
 491
 492             if (fetch->old_entry) {
 493                 debugs(72, 3, "peerDigestFetchReply: got new digest, releasing old one");
 494                 storeUnregister(fetch->old_sc, fetch->old_entry, fetch);
 495                 fetch->old_entry->releaseRequest();
 496                 fetch->old_entry->unlock("peerDigestFetchReply 200");
 497                 fetch->old_entry = nullptr;
 498             }
 499
 500             fetch->state = DIGEST_READ_CBLOCK;
 501         } else {
 502             /* some kind of a bug */
 503             finishAndDeleteFetch(fetch, reply.sline.reason(), true);
 504             return -1;      /* XXX -1 will abort stuff in ReadReply! */
 505         }
 506     }
 507
 508     return 0; // we consumed/used no buffered bytes
 509 }
 510
 511 int
 512 peerDigestSwapInCBlock(void *data, char *buf, ssize_t size)
 513 {
 514     DigestFetchState *fetch = (DigestFetchState *)data;
 515
 516     assert(fetch->state == DIGEST_READ_CBLOCK);
 517
 518     if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestSwapInCBlock"))
 519         return -1;
 520
 521     if (size >= (ssize_t)StoreDigestCBlockSize) {
 522         const auto pd = fetch->pd.get();
 523
 524         assert(pd);
 525         assert(fetch->entry->mem_obj);
 526
 527         if (peerDigestSetCBlock(pd, buf)) {
 528             /* XXX: soon we will have variable header size */
 529             /* switch to CD buffer and fetch digest guts */
 530             buf = nullptr;
 531             assert(pd->cd->mask);
 532             fetch->state = DIGEST_READ_MASK;
 533             return StoreDigestCBlockSize;
 534         } else {
 535             finishAndDeleteFetch(fetch, "invalid digest cblock", true);
 536             return -1;
 537         }
 538     }
 539
 540     /* need more data, do we have space? */
 541     if (size >= SM_PAGE_SIZE) {
 542         finishAndDeleteFetch(fetch, "digest cblock too big", true);
 543         return -1;
 544     }
 545
 546     return 0;       /* We need more data */
 547 }
 548
 549 int
 550 peerDigestSwapInMask(void *data, char *buf, ssize_t size)
 551 {
 552     DigestFetchState *fetch = (DigestFetchState *)data;
 553     const auto pd = fetch->pd.get();
 554     assert(pd);
 555     assert(pd->cd && pd->cd->mask);
 556
 557     /*
 558      * NOTENOTENOTENOTENOTE: buf doesn't point to pd->cd->mask anymore!
 559      * we need to do the copy ourselves!
 560      */
 561     memcpy(pd->cd->mask + fetch->mask_offset, buf, size);
 562
 563     /* NOTE! buf points to the middle of pd->cd->mask! */
 564
 565     if (peerDigestFetchedEnough(fetch, nullptr, size, "peerDigestSwapInMask"))
 566         return -1;
 567
 568     fetch->mask_offset += size;
 569
 570     if (fetch->mask_offset >= pd->cd->mask_size) {
 571         debugs(72, 2, "peerDigestSwapInMask: Done! Got " <<
 572                fetch->mask_offset << ", expected " << pd->cd->mask_size);
 573         assert(fetch->mask_offset == pd->cd->mask_size);
 574         assert(peerDigestFetchedEnough(fetch, nullptr, 0, "peerDigestSwapInMask"));
 575         return -1;      /* XXX! */
 576     }
 577
 578     /* We always read everything, so return size */
 579     return size;
 580 }
 581
 582 static int
 583 peerDigestFetchedEnough(DigestFetchState * fetch, char *, ssize_t size, const char *step_name)
 584 {
 585     static const SBuf hostUnknown("<unknown>"); // peer host (if any)
 586     SBuf host = hostUnknown;
 587
 588     const auto pd = fetch->pd.get();
 589     Assure(pd);
 590     const char *reason = nullptr;  /* reason for completion */
 591     const char *no_bug = nullptr;  /* successful completion if set */
 592
 593     /* test possible exiting conditions (the same for most steps!)
 594      * cases marked with '?!' should not happen */
 595
 596     if (!reason) {
 597         host = pd->host;
 598     }
 599
 600     debugs(72, 6, step_name << ": peer " << host << ", offset: " <<
 601            fetch->offset << " size: " << size << ".");
 602
 603     /* continue checking (with pd and host known and valid) */
 604
 605     if (!reason) {
 606         if (size < 0)
 607             reason = "swap failure";
 608         else if (!fetch->entry)
 609             reason = "swap aborted?!";
 610         else if (EBIT_TEST(fetch->entry->flags, ENTRY_ABORTED))
 611             reason = "swap aborted";
 612     }
 613
 614     /* continue checking (maybe-successful eof case) */
 615     if (!reason && !size && fetch->state != DIGEST_READ_REPLY) {
 616         if (!pd->cd)
 617             reason = "null digest?!";
 618         else if (fetch->mask_offset != pd->cd->mask_size)
 619             reason = "premature end of digest?!";
 620         else if (!peerDigestUseful(pd))
 621             reason = "useless digest";
 622         else
 623             reason = no_bug = "success";
 624     }
 625
 626     /* finish if we have a reason */
 627     if (reason) {
 628         const int level = strstr(reason, "?!") ? 1 : 3;
 629         debugs(72, level, "" << step_name << ": peer " << host << ", exiting after '" << reason << "'");
 630         finishAndDeleteFetch(fetch, reason, !no_bug);
 631     }
 632
 633     return reason != nullptr;
 634 }
 635
 636 /* complete the digest transfer, update stats, unlock/release everything */
 637 static void
 638 finishAndDeleteFetch(DigestFetchState * const fetch, const char * const reason, const bool err)
 639 {
 640     assert(reason);
 641
 642     debugs(72, 2, "peer: " << RawPointer(fetch->pd.valid() ? fetch->pd->peer : nullptr).orNil() << ", reason: " << reason << ", err: " << err);
 643
 644     /* note: order is significant */
 645     peerDigestFetchSetStats(fetch);
 646     if (const auto pd = fetch->pd.get())
 647         pd->noteFetchFinished(*fetch, reason, err);
 648
 649     delete fetch;
 650 }
 651
 652 void
 653 PeerDigest::noteFetchFinished(const DigestFetchState &finishedFetch, const char * const outcomeDescription, const bool sawError)
 654 {
 655     const auto pd = this; // TODO: remove this diff reducer
 656     const auto fetch = &finishedFetch; // TODO: remove this diff reducer
 657
 658     pd->flags.requested = false;
 659     pd->req_result = outcomeDescription;
 660
 661     pd->times.received = squid_curtime;
 662     pd->times.req_delay = fetch->resp_time;
 663     pd->stats.sent.kbytes += fetch->sent.bytes;
 664     pd->stats.recv.kbytes += fetch->recv.bytes;
 665     pd->stats.sent.msgs += fetch->sent.msg;
 666     pd->stats.recv.msgs += fetch->recv.msg;
 667
 668     if (sawError) {
 669         debugs(72, DBG_IMPORTANT, "disabling (" << outcomeDescription << ") digest from " << host);
 670
 671         pd->times.retry_delay = peerDigestIncDelay(pd);
 672         peerDigestSetCheck(pd, pd->times.retry_delay);
 673         delete pd->cd;
 674         pd->cd = nullptr;
 675
 676         pd->flags.usable = false;
 677     } else {
 678         pd->flags.usable = true;
 679         pd->times.retry_delay = 0;
 680         peerDigestSetCheck(pd, peerDigestNewDelay(fetch->entry));
 681
 682         /* XXX: ugly condition, but how? */
 683
 684         if (fetch->entry->store_status == STORE_OK)
 685             debugs(72, 2, "re-used old digest from " << host);
 686         else
 687             debugs(72, 2, "received valid digest from " << host);
 688     }
 689 }
 690
 691 /* calculate fetch stats after completion */
 692 static void
 693 peerDigestFetchSetStats(DigestFetchState * fetch)
 694 {
 695     MemObject *mem;
 696     assert(fetch->entry && fetch->request);
 697
 698     mem = fetch->entry->mem_obj;
 699     assert(mem);
 700
 701     /* XXX: outgoing numbers are not precise */
 702     /* XXX: we must distinguish between 304 hits and misses here */
 703     fetch->sent.bytes = fetch->request->prefixLen();
 704     /* XXX: this is slightly wrong: we don't KNOW that the entire memobject
 705      * was fetched. We only know how big it is
 706      */
 707     fetch->recv.bytes = mem->size();
 708     fetch->sent.msg = fetch->recv.msg = 1;
 709     fetch->expires = fetch->entry->expires;
 710     fetch->resp_time = squid_curtime - fetch->start_time;
 711
 712     debugs(72, 3, "peerDigestFetchFinish: recv " << fetch->recv.bytes <<
 713            " bytes in " << (int) fetch->resp_time << " secs");
 714
 715     debugs(72, 3, "peerDigestFetchFinish: expires: " <<
 716            (long int) fetch->expires << " (" << std::showpos <<
 717            (int) (fetch->expires - squid_curtime) << "), lmt: " <<
 718            std::noshowpos << (long int) fetch->entry->lastModified() << " (" <<
 719            std::showpos << (int) (fetch->entry->lastModified() - squid_curtime) <<
 720            ")");
 721
 722     statCounter.cd.kbytes_sent += fetch->sent.bytes;
 723     statCounter.cd.kbytes_recv += fetch->recv.bytes;
 724     statCounter.cd.msgs_sent += fetch->sent.msg;
 725     statCounter.cd.msgs_recv += fetch->recv.msg;
 726 }
 727
 728 static int
 729 peerDigestSetCBlock(PeerDigest * pd, const char *buf)
 730 {
 731     StoreDigestCBlock cblock;
 732     int freed_size = 0;
 733     const auto host = pd->host;
 734
 735     memcpy(&cblock, buf, sizeof(cblock));
 736     /* network -> host conversions */
 737     cblock.ver.current = ntohs(cblock.ver.current);
 738     cblock.ver.required = ntohs(cblock.ver.required);
 739     cblock.capacity = ntohl(cblock.capacity);
 740     cblock.count = ntohl(cblock.count);
 741     cblock.del_count = ntohl(cblock.del_count);
 742     cblock.mask_size = ntohl(cblock.mask_size);
 743     debugs(72, 2, "got digest cblock from " << host << "; ver: " <<
 744            (int) cblock.ver.current << " (req: " << (int) cblock.ver.required <<
 745            ")");
 746
 747     debugs(72, 2, "\t size: " <<
 748            cblock.mask_size << " bytes, e-cnt: " <<
 749            cblock.count << ", e-util: " <<
 750            xpercentInt(cblock.count, cblock.capacity) << "%" );
 751     /* check version requirements (both ways) */
 752
 753     if (cblock.ver.required > CacheDigestVer.current) {
 754         debugs(72, DBG_IMPORTANT, "" << host << " digest requires version " <<
 755                cblock.ver.required << "; have: " << CacheDigestVer.current);
 756
 757         return 0;
 758     }
 759
 760     if (cblock.ver.current < CacheDigestVer.required) {
 761         debugs(72, DBG_IMPORTANT, "" << host << " digest is version " <<
 762                cblock.ver.current << "; we require: " <<
 763                CacheDigestVer.required);
 764
 765         return 0;
 766     }
 767
 768     /* check consistency */
 769     if (cblock.ver.required > cblock.ver.current ||
 770             cblock.mask_size <= 0 || cblock.capacity <= 0 ||
 771             cblock.bits_per_entry <= 0 || cblock.hash_func_count <= 0) {
 772         debugs(72, DBG_CRITICAL, "" << host << " digest cblock is corrupted.");
 773         return 0;
 774     }
 775
 776     /* check consistency further */
 777     if ((size_t)cblock.mask_size != CacheDigest::CalcMaskSize(cblock.capacity, cblock.bits_per_entry)) {
 778         debugs(72, DBG_CRITICAL, host << " digest cblock is corrupted " <<
 779                "(mask size mismatch: " << cblock.mask_size << " ? " <<
 780                CacheDigest::CalcMaskSize(cblock.capacity, cblock.bits_per_entry)
 781                << ").");
 782         return 0;
 783     }
 784
 785     /* there are some things we cannot do yet */
 786     if (cblock.hash_func_count != CacheDigestHashFuncCount) {
 787         debugs(72, DBG_CRITICAL, "ERROR: " << host << " digest: unsupported #hash functions: " <<
 788                cblock.hash_func_count << " ? " << CacheDigestHashFuncCount << ".");
 789         return 0;
 790     }
 791
 792     /*
 793      * no cblock bugs below this point
 794      */
 795     /* check size changes */
 796     if (pd->cd && cblock.mask_size != (ssize_t)pd->cd->mask_size) {
 797         debugs(72, 2, host << " digest changed size: " << cblock.mask_size <<
 798                " -> " << pd->cd->mask_size);
 799         freed_size = pd->cd->mask_size;
 800         delete pd->cd;
 801         pd->cd = nullptr;
 802     }
 803
 804     if (!pd->cd) {
 805         debugs(72, 2, "creating " << host << " digest; size: " << cblock.mask_size << " (" <<
 806                std::showpos <<  (int) (cblock.mask_size - freed_size) << ") bytes");
 807         pd->cd = new CacheDigest(cblock.capacity, cblock.bits_per_entry);
 808
 809         if (cblock.mask_size >= freed_size)
 810             statCounter.cd.memory += (cblock.mask_size - freed_size);
 811     }
 812
 813     assert(pd->cd);
 814     /* these assignments leave us in an inconsistent state until we finish reading the digest */
 815     pd->cd->count = cblock.count;
 816     pd->cd->del_count = cblock.del_count;
 817     return 1;
 818 }
 819
 820 static int
 821 peerDigestUseful(const PeerDigest * pd)
 822 {
 823     /* TODO: we should calculate the prob of a false hit instead of bit util */
 824     const auto bit_util = pd->cd->usedMaskPercent();
 825
 826     if (bit_util > 65.0) {
 827         debugs(72, DBG_CRITICAL, "WARNING: " << pd->host <<
 828                " peer digest has too many bits on (" << bit_util << "%).");
 829         return 0;
 830     }
 831
 832     return 1;
 833 }
 834
 835 static int
 836 saneDiff(time_t diff)
 837 {
 838     return abs((int) diff) > squid_curtime / 2 ? 0 : diff;
 839 }
 840
 841 void
 842 peerDigestStatsReport(const PeerDigest * pd, StoreEntry * e)
 843 {
 844 #define f2s(flag) (pd->flags.flag ? "yes" : "no")
 845 #define appendTime(tm) storeAppendPrintf(e, "%s\t %10ld\t %+d\t %+d\n", \
 846     ""#tm, (long int)pd->times.tm, \
 847     saneDiff(pd->times.tm - squid_curtime), \
 848     saneDiff(pd->times.tm - pd->times.initialized))
 849
 850     assert(pd);
 851
 852     auto host = pd->host;
 853     storeAppendPrintf(e, "\npeer digest from " SQUIDSBUFPH "\n", SQUIDSBUFPRINT(host));
 854
 855     cacheDigestGuessStatsReport(&pd->stats.guess, e, host);
 856
 857     storeAppendPrintf(e, "\nevent\t timestamp\t secs from now\t secs from init\n");
 858     appendTime(initialized);
 859     appendTime(needed);
 860     appendTime(requested);
 861     appendTime(received);
 862     appendTime(next_check);
 863
 864     storeAppendPrintf(e, "peer digest state:\n");
 865     storeAppendPrintf(e, "\tneeded: %3s, usable: %3s, requested: %3s\n",
 866                       f2s(needed), f2s(usable), f2s(requested));
 867     storeAppendPrintf(e, "\n\tlast retry delay: %d secs\n",
 868                       (int) pd->times.retry_delay);
 869     storeAppendPrintf(e, "\tlast request response time: %d secs\n",
 870                       (int) pd->times.req_delay);
 871     storeAppendPrintf(e, "\tlast request result: %s\n",
 872                       pd->req_result ? pd->req_result : "(none)");
 873
 874     storeAppendPrintf(e, "\npeer digest traffic:\n");
 875     storeAppendPrintf(e, "\trequests sent: %d, volume: %d KB\n",
 876                       pd->stats.sent.msgs, (int) pd->stats.sent.kbytes.kb);
 877     storeAppendPrintf(e, "\treplies recv:  %d, volume: %d KB\n",
 878                       pd->stats.recv.msgs, (int) pd->stats.recv.kbytes.kb);
 879
 880     storeAppendPrintf(e, "\npeer digest structure:\n");
 881
 882     if (pd->cd)
 883         cacheDigestReport(pd->cd, host, e);
 884     else
 885         storeAppendPrintf(e, "\tno in-memory copy\n");
 886 }
 887
 888 #endif
 889