1 /*
2 * Copyright (C) 1996-2021 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 "CacheDigest.h"
14 #include "CachePeer.h"
15 #include "event.h"
16 #include "FwdState.h"
17 #include "globals.h"
18 #include "HttpReply.h"
19 #include "HttpRequest.h"
20 #include "internal.h"
21 #include "MemObject.h"
22 #include "mime_header.h"
23 #include "neighbors.h"
24 #include "PeerDigest.h"
25 #include "SquidTime.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 peerDigestSwapInHeaders(void *, char *, ssize_t);
43 int peerDigestSwapInCBlock(void *, char *, ssize_t);
44 int peerDigestSwapInMask(void *, char *, ssize_t);
45 static int peerDigestFetchedEnough(DigestFetchState * fetch, char *buf, ssize_t size, const char *step_name);
46 static void peerDigestFetchStop(DigestFetchState * fetch, char *buf, const char *reason);
47 static void peerDigestFetchAbort(DigestFetchState * fetch, char *buf, const char *reason);
48 static void peerDigestReqFinish(DigestFetchState * fetch, char *buf, int, int, int, const char *reason, int err);
49 static void peerDigestPDFinish(DigestFetchState * fetch, int pcb_valid, int err);
50 static void peerDigestFetchFinish(DigestFetchState * fetch, int err);
51 static void peerDigestFetchSetStats(DigestFetchState * fetch);
52 static int peerDigestSetCBlock(PeerDigest * pd, const char *buf);
53 static int peerDigestUseful(const PeerDigest * pd);
54
55 /* local constants */
56 Version const CacheDigestVer = { 5, 3 };
57
58 #define StoreDigestCBlockSize sizeof(StoreDigestCBlock)
59
60 /* min interval for requesting digests from a given peer */
61 static const time_t PeerDigestReqMinGap = 5 * 60; /* seconds */
62 /* min interval for requesting digests (cumulative request stream) */
63 static const time_t GlobDigestReqMinGap = 1 * 60; /* seconds */
64
65 /* local vars */
66
67 static time_t pd_last_req_time = 0; /* last call to Check */
68
PeerDigest(CachePeer * p)69 PeerDigest::PeerDigest(CachePeer * p)
70 {
71 assert(p);
72
73 /*
74 * DPW 2007-04-12
75 * Lock on to the peer here. The corresponding cbdataReferenceDone()
76 * is in peerDigestDestroy().
77 */
78 peer = cbdataReference(p);
79 /* if peer disappears, we will know it's name */
80 host = p->host;
81
82 times.initialized = squid_curtime;
83 }
84
85 CBDATA_CLASS_INIT(PeerDigest);
86
87 CBDATA_CLASS_INIT(DigestFetchState);
88
DigestFetchState(PeerDigest * aPd,HttpRequest * req)89 DigestFetchState::DigestFetchState(PeerDigest *aPd, HttpRequest *req) :
90 pd(cbdataReference(aPd)),
91 entry(NULL),
92 old_entry(NULL),
93 sc(NULL),
94 old_sc(NULL),
95 request(req),
96 offset(0),
97 mask_offset(0),
98 start_time(squid_curtime),
99 resp_time(0),
100 expires(0),
101 bufofs(0),
102 state(DIGEST_READ_REPLY)
103 {
104 HTTPMSGLOCK(request);
105
106 sent.msg = 0;
107 sent.bytes = 0;
108
109 recv.msg = 0;
110 recv.bytes = 0;
111
112 *buf = 0;
113 }
114
~DigestFetchState()115 DigestFetchState::~DigestFetchState()
116 {
117 /* unlock everything */
118 storeUnregister(sc, entry, this);
119
120 entry->unlock("DigestFetchState destructed");
121 entry = NULL;
122
123 HTTPMSGUNLOCK(request);
124
125 assert(pd == NULL);
126 }
127
128 /* allocate new peer digest, call Init, and lock everything */
129 void
peerDigestCreate(CachePeer * p)130 peerDigestCreate(CachePeer * p)
131 {
132 assert(p);
133
134 PeerDigest *pd = new PeerDigest(p);
135
136 // TODO: make CachePeer member a CbcPointer
137 p->digest = cbdataReference(pd);
138
139 // lock a reference to pd again to prevent the PeerDigest
140 // disappearing during peerDigestDestroy() when
141 // cbdataReferenceValidDone is called.
142 // TODO test if it can be moved into peerDigestDestroy() or
143 // if things can break earlier (eg CachePeer death).
144 (void)cbdataReference(pd);
145 }
146
147 /* call Clean and free/unlock everything */
148 static void
peerDigestDestroy(PeerDigest * pd)149 peerDigestDestroy(PeerDigest * pd)
150 {
151 void *p;
152 assert(pd);
153 void * peerTmp = pd->peer;
154
155 /*
156 * DPW 2007-04-12
157 * We locked the peer in PeerDigest constructor, this is
158 * where we unlock it.
159 */
160 if (cbdataReferenceValidDone(peerTmp, &p)) {
161 // we locked the p->digest in peerDigestCreate()
162 // this is where we unlock that
163 cbdataReferenceDone(static_cast<CachePeer *>(p)->digest);
164 }
165
166 delete pd;
167 }
168
~PeerDigest()169 PeerDigest::~PeerDigest()
170 {
171 delete cd;
172 // req_result pointer is not owned by us
173 }
174
175 /* called by peer to indicate that somebody actually needs this digest */
176 void
peerDigestNeeded(PeerDigest * pd)177 peerDigestNeeded(PeerDigest * pd)
178 {
179 assert(pd);
180 assert(!pd->flags.needed);
181 assert(!pd->cd);
182
183 pd->flags.needed = true;
184 pd->times.needed = squid_curtime;
185 peerDigestSetCheck(pd, 0); /* check asap */
186 }
187
188 /* currently we do not have a reason to disable without destroying */
189 #if FUTURE_CODE
190 /* disables peer for good */
191 static void
peerDigestDisable(PeerDigest * pd)192 peerDigestDisable(PeerDigest * pd)
193 {
194 debugs(72, 2, "peerDigestDisable: peer " << pd->host.buf() << " disabled for good");
195 pd->times.disabled = squid_curtime;
196 pd->times.next_check = -1; /* never */
197 pd->flags.usable = 0;
198
199 delete pd->cd
200 pd->cd = nullptr;
201
202 /* we do not destroy the pd itself to preserve its "history" and stats */
203 }
204
205 #endif
206
207 /* increment retry delay [after an unsuccessful attempt] */
208 static time_t
peerDigestIncDelay(const PeerDigest * pd)209 peerDigestIncDelay(const PeerDigest * pd)
210 {
211 assert(pd);
212 return pd->times.retry_delay > 0 ?
213 2 * pd->times.retry_delay : /* exponential backoff */
214 PeerDigestReqMinGap; /* minimal delay */
215 }
216
217 /* artificially increases Expires: setting to avoid race conditions
218 * returns the delay till that [increased] expiration time */
219 static time_t
peerDigestNewDelay(const StoreEntry * e)220 peerDigestNewDelay(const StoreEntry * e)
221 {
222 assert(e);
223
224 if (e->expires > 0)
225 return e->expires + PeerDigestReqMinGap - squid_curtime;
226
227 return PeerDigestReqMinGap;
228 }
229
230 /* registers next digest verification */
231 static void
peerDigestSetCheck(PeerDigest * pd,time_t delay)232 peerDigestSetCheck(PeerDigest * pd, time_t delay)
233 {
234 eventAdd("peerDigestCheck", peerDigestCheck, pd, (double) delay, 1);
235 pd->times.next_check = squid_curtime + delay;
236 debugs(72, 3, "peerDigestSetCheck: will check peer " << pd->host << " in " << delay << " secs");
237 }
238
239 /*
240 * called when peer is about to disappear or have already disappeared
241 */
242 void
peerDigestNotePeerGone(PeerDigest * pd)243 peerDigestNotePeerGone(PeerDigest * pd)
244 {
245 if (pd->flags.requested) {
246 debugs(72, 2, "peerDigest: peer " << pd->host << " gone, will destroy after fetch.");
247 /* do nothing now, the fetching chain will notice and take action */
248 } else {
249 debugs(72, 2, "peerDigest: peer " << pd->host << " is gone, destroying now.");
250 peerDigestDestroy(pd);
251 }
252 }
253
254 /* callback for eventAdd() (with peer digest locked)
255 * request new digest if our copy is too old or if we lack one;
256 * schedule next check otherwise */
257 static void
peerDigestCheck(void * data)258 peerDigestCheck(void *data)
259 {
260 PeerDigest *pd = (PeerDigest *)data;
261 time_t req_time;
262
263 assert(!pd->flags.requested);
264
265 pd->times.next_check = 0; /* unknown */
266
267 if (!cbdataReferenceValid(pd->peer)) {
268 peerDigestNotePeerGone(pd);
269 return;
270 }
271
272 debugs(72, 3, "peerDigestCheck: peer " << pd->peer->host << ":" << pd->peer->http_port);
273 debugs(72, 3, "peerDigestCheck: time: " << squid_curtime <<
274 ", last received: " << (long int) pd->times.received << " (" <<
275 std::showpos << (int) (squid_curtime - pd->times.received) << ")");
276
277 /* decide when we should send the request:
278 * request now unless too close to other requests */
279 req_time = squid_curtime;
280
281 /* per-peer limit */
282
283 if (req_time - pd->times.received < PeerDigestReqMinGap) {
284 debugs(72, 2, "peerDigestCheck: " << pd->host <<
285 ", avoiding close peer requests (" <<
286 (int) (req_time - pd->times.received) << " < " <<
287 (int) PeerDigestReqMinGap << " secs).");
288
289 req_time = pd->times.received + PeerDigestReqMinGap;
290 }
291
292 /* global limit */
293 if (req_time - pd_last_req_time < GlobDigestReqMinGap) {
294 debugs(72, 2, "peerDigestCheck: " << pd->host <<
295 ", avoiding close requests (" <<
296 (int) (req_time - pd_last_req_time) << " < " <<
297 (int) GlobDigestReqMinGap << " secs).");
298
299 req_time = pd_last_req_time + GlobDigestReqMinGap;
300 }
301
302 if (req_time <= squid_curtime)
303 peerDigestRequest(pd); /* will set pd->flags.requested */
304 else
305 peerDigestSetCheck(pd, req_time - squid_curtime);
306 }
307
308 /* ask store for a digest */
309 static void
peerDigestRequest(PeerDigest * pd)310 peerDigestRequest(PeerDigest * pd)
311 {
312 CachePeer *p = pd->peer;
313 StoreEntry *e, *old_e;
314 char *url = NULL;
315 HttpRequest *req;
316 StoreIOBuffer tempBuffer;
317
318 pd->req_result = NULL;
319 pd->flags.requested = true;
320
321 /* compute future request components */
322
323 if (p->digest_url)
324 url = xstrdup(p->digest_url);
325 else
326 url = xstrdup(internalRemoteUri(p->secure.encryptTransport, p->host, p->http_port, "/squid-internal-periodic/", SBuf(StoreDigestFileName)));
327 debugs(72, 2, url);
328
329 const MasterXaction::Pointer mx = new MasterXaction(XactionInitiator::initCacheDigest);
330 req = HttpRequest::FromUrlXXX(url, mx);
331
332 assert(req);
333
334 /* add custom headers */
335 assert(!req->header.len);
336
337 req->header.putStr(Http::HdrType::ACCEPT, StoreDigestMimeStr);
338
339 req->header.putStr(Http::HdrType::ACCEPT, "text/html");
340
341 if (p->login &&
342 p->login[0] != '*' &&
343 strcmp(p->login, "PASS") != 0 &&
344 strcmp(p->login, "PASSTHRU") != 0 &&
345 strncmp(p->login, "NEGOTIATE",9) != 0 &&
346 strcmp(p->login, "PROXYPASS") != 0) {
347 req->url.userInfo(SBuf(p->login)); // XXX: performance regression make peer login SBuf as well.
348 }
349 /* create fetch state structure */
350 DigestFetchState *fetch = new DigestFetchState(pd, req);
351
352 /* update timestamps */
353 pd->times.requested = squid_curtime;
354 pd_last_req_time = squid_curtime;
355 req->flags.cachable = true;
356
357 /* the rest is based on clientReplyContext::processExpired() */
358 req->flags.refresh = true;
359
360 old_e = fetch->old_entry = storeGetPublicByRequest(req);
361
362 if (old_e) {
363 debugs(72, 5, "found old " << *old_e);
364
365 old_e->lock("peerDigestRequest");
366 old_e->ensureMemObject(url, url, req->method);
367
368 fetch->old_sc = storeClientListAdd(old_e, fetch);
369 }
370
371 e = fetch->entry = storeCreateEntry(url, url, req->flags, req->method);
372 debugs(72, 5, "created " << *e);
373 assert(EBIT_TEST(e->flags, KEY_PRIVATE));
374 fetch->sc = storeClientListAdd(e, fetch);
375 /* set lastmod to trigger IMS request if possible */
376
377 if (old_e)
378 e->lastModified(old_e->lastModified());
379
380 /* push towards peer cache */
381 FwdState::fwdStart(Comm::ConnectionPointer(), e, req);
382
383 tempBuffer.offset = 0;
384
385 tempBuffer.length = SM_PAGE_SIZE;
386
387 tempBuffer.data = fetch->buf;
388
389 storeClientCopy(fetch->sc, e, tempBuffer,
390 peerDigestHandleReply, fetch);
391
392 safe_free(url);
393 }
394
395 /* Handle the data copying .. */
396
397 /*
398 * This routine handles the copy data and then redirects the
399 * copy to a bunch of subfunctions depending upon the copy state.
400 * It also tracks the buffer offset and "seen", since I'm actually
401 * not interested in rewriting everything to suit my little idea.
402 */
403 static void
peerDigestHandleReply(void * data,StoreIOBuffer receivedData)404 peerDigestHandleReply(void *data, StoreIOBuffer receivedData)
405 {
406 DigestFetchState *fetch = (DigestFetchState *)data;
407 int retsize = -1;
408 digest_read_state_t prevstate;
409 int newsize;
410
411 assert(fetch->pd && receivedData.data);
412 /* The existing code assumes that the received pointer is
413 * where we asked the data to be put
414 */
415 assert(fetch->buf + fetch->bufofs == receivedData.data);
416
417 /* Update the buffer size */
418 fetch->bufofs += receivedData.length;
419
420 assert(fetch->bufofs <= SM_PAGE_SIZE);
421
422 /* If we've fetched enough, return */
423
424 if (peerDigestFetchedEnough(fetch, fetch->buf, fetch->bufofs, "peerDigestHandleReply"))
425 return;
426
427 /* Call the right function based on the state */
428 /* (Those functions will update the state if needed) */
429
430 /* Give us a temporary reference. Some of the calls we make may
431 * try to destroy the fetch structure, and we like to know if they
432 * do
433 */
434 CbcPointer<DigestFetchState> tmpLock = fetch;
435
436 /* Repeat this loop until we're out of data OR the state changes */
437 /* (So keep going if the state has changed and we still have data */
438 do {
439 prevstate = fetch->state;
440
441 switch (fetch->state) {
442
443 case DIGEST_READ_REPLY:
444 retsize = peerDigestFetchReply(fetch, fetch->buf, fetch->bufofs);
445 break;
446
447 case DIGEST_READ_HEADERS:
448 retsize = peerDigestSwapInHeaders(fetch, fetch->buf, fetch->bufofs);
449 break;
450
451 case DIGEST_READ_CBLOCK:
452 retsize = peerDigestSwapInCBlock(fetch, fetch->buf, fetch->bufofs);
453 break;
454
455 case DIGEST_READ_MASK:
456 retsize = peerDigestSwapInMask(fetch, fetch->buf, fetch->bufofs);
457 break;
458
459 case DIGEST_READ_NONE:
460 break;
461
462 case DIGEST_READ_DONE:
463 return;
464 break;
465
466 default:
467 fatal("Bad digest transfer mode!\n");
468 }
469
470 if (retsize < 0)
471 return;
472
473 /*
474 * The returned size indicates how much of the buffer was read -
475 * so move the remainder of the buffer to the beginning
476 * and update the bufofs / bufsize
477 */
478 newsize = fetch->bufofs - retsize;
479
480 memmove(fetch->buf, fetch->buf + retsize, fetch->bufofs - newsize);
481
482 fetch->bufofs = newsize;
483
484 } while (cbdataReferenceValid(fetch) && prevstate != fetch->state && fetch->bufofs > 0);
485
486 // Check for EOF here, thus giving the parser one extra run. We could avoid this overhead by
487 // checking at the beginning of this function. However, in this case, we would have to require
488 // that the parser does not regard EOF as a special condition (it is true now but may change
489 // in the future).
490 if (!receivedData.length) { // EOF
491 peerDigestFetchAbort(fetch, fetch->buf, "premature end of digest reply");
492 return;
493 }
494
495 /* Update the copy offset */
496 fetch->offset += receivedData.length;
497
498 /* Schedule another copy */
499 if (cbdataReferenceValid(fetch)) {
500 StoreIOBuffer tempBuffer;
501 tempBuffer.offset = fetch->offset;
502 tempBuffer.length = SM_PAGE_SIZE - fetch->bufofs;
503 tempBuffer.data = fetch->buf + fetch->bufofs;
504 storeClientCopy(fetch->sc, fetch->entry, tempBuffer,
505 peerDigestHandleReply, fetch);
506 }
507 }
508
509 /* wait for full http headers to be received then parse them */
510 /*
511 * This routine handles parsing the reply line.
512 * If the reply line indicates an OK, the same data is thrown
513 * to SwapInHeaders(). If the reply line is a NOT_MODIFIED,
514 * we simply stop parsing.
515 */
516 static int
peerDigestFetchReply(void * data,char * buf,ssize_t size)517 peerDigestFetchReply(void *data, char *buf, ssize_t size)
518 {
519 DigestFetchState *fetch = (DigestFetchState *)data;
520 PeerDigest *pd = fetch->pd;
521 size_t hdr_size;
522 assert(pd && buf);
523 assert(!fetch->offset);
524
525 assert(fetch->state == DIGEST_READ_REPLY);
526
527 if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestFetchReply"))
528 return -1;
529
530 if ((hdr_size = headersEnd(buf, size))) {
531 HttpReply const *reply = fetch->entry->getReply();
532 assert(reply);
533 assert(reply->sline.status() != Http::scNone);
534 const Http::StatusCode status = reply->sline.status();
535 debugs(72, 3, "peerDigestFetchReply: " << pd->host << " status: " << status <<
536 ", expires: " << (long int) reply->expires << " (" << std::showpos <<
537 (int) (reply->expires - squid_curtime) << ")");
538
539 /* this "if" is based on clientHandleIMSReply() */
540
541 if (status == Http::scNotModified) {
542 /* our old entry is fine */
543 assert(fetch->old_entry);
544
545 if (!fetch->old_entry->mem_obj->request) {
546 fetch->old_entry->mem_obj->request = fetch->entry->mem_obj->request;
547 HTTPMSGLOCK(fetch->old_entry->mem_obj->request);
548 }
549
550 assert(fetch->old_entry->mem_obj->request);
551
552 Store::Root().updateOnNotModified(fetch->old_entry, *fetch->entry);
553
554 /* get rid of 304 reply */
555 storeUnregister(fetch->sc, fetch->entry, fetch);
556
557 fetch->entry->unlock("peerDigestFetchReply 304");
558
559 fetch->entry = fetch->old_entry;
560
561 fetch->old_entry = NULL;
562
563 /* preserve request -- we need its size to update counters */
564 /* requestUnlink(r); */
565 /* fetch->entry->mem_obj->request = NULL; */
566 } else if (status == Http::scOkay) {
567 /* get rid of old entry if any */
568
569 if (fetch->old_entry) {
570 debugs(72, 3, "peerDigestFetchReply: got new digest, releasing old one");
571 storeUnregister(fetch->old_sc, fetch->old_entry, fetch);
572 fetch->old_entry->releaseRequest();
573 fetch->old_entry->unlock("peerDigestFetchReply 200");
574 fetch->old_entry = NULL;
575 }
576 } else {
577 /* some kind of a bug */
578 peerDigestFetchAbort(fetch, buf, reply->sline.reason());
579 return -1; /* XXX -1 will abort stuff in ReadReply! */
580 }
581
582 /* must have a ready-to-use store entry if we got here */
583 /* can we stay with the old in-memory digest? */
584 if (status == Http::scNotModified && fetch->pd->cd) {
585 peerDigestFetchStop(fetch, buf, "Not modified");
586 fetch->state = DIGEST_READ_DONE;
587 } else {
588 fetch->state = DIGEST_READ_HEADERS;
589 }
590 } else {
591 /* need more data, do we have space? */
592
593 if (size >= SM_PAGE_SIZE)
594 peerDigestFetchAbort(fetch, buf, "reply header too big");
595 }
596
597 /* We don't want to actually ack that we've handled anything,
598 * otherwise SwapInHeaders() won't get the reply line .. */
599 return 0;
600 }
601
602 /* fetch headers from disk, pass on to SwapInCBlock */
603 int
peerDigestSwapInHeaders(void * data,char * buf,ssize_t size)604 peerDigestSwapInHeaders(void *data, char *buf, ssize_t size)
605 {
606 DigestFetchState *fetch = (DigestFetchState *)data;
607 size_t hdr_size;
608
609 assert(fetch->state == DIGEST_READ_HEADERS);
610
611 if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestSwapInHeaders"))
612 return -1;
613
614 assert(!fetch->offset);
615
616 if ((hdr_size = headersEnd(buf, size))) {
617 assert(fetch->entry->getReply());
618 assert(fetch->entry->getReply()->sline.status() != Http::scNone);
619
620 if (fetch->entry->getReply()->sline.status() != Http::scOkay) {
621 debugs(72, DBG_IMPORTANT, "peerDigestSwapInHeaders: " << fetch->pd->host <<
622 " status " << fetch->entry->getReply()->sline.status() <<
623 " got cached!");
624
625 peerDigestFetchAbort(fetch, buf, "internal status error");
626 return -1;
627 }
628
629 fetch->state = DIGEST_READ_CBLOCK;
630 return hdr_size; /* Say how much data we read */
631 }
632
633 /* need more data, do we have space? */
634 if (size >= SM_PAGE_SIZE) {
635 peerDigestFetchAbort(fetch, buf, "stored header too big");
636 return -1;
637 }
638
639 return 0; /* We need to read more to parse .. */
640 }
641
642 int
peerDigestSwapInCBlock(void * data,char * buf,ssize_t size)643 peerDigestSwapInCBlock(void *data, char *buf, ssize_t size)
644 {
645 DigestFetchState *fetch = (DigestFetchState *)data;
646
647 assert(fetch->state == DIGEST_READ_CBLOCK);
648
649 if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestSwapInCBlock"))
650 return -1;
651
652 if (size >= (ssize_t)StoreDigestCBlockSize) {
653 PeerDigest *pd = fetch->pd;
654
655 assert(pd && fetch->entry->getReply());
656
657 if (peerDigestSetCBlock(pd, buf)) {
658 /* XXX: soon we will have variable header size */
659 /* switch to CD buffer and fetch digest guts */
660 buf = NULL;
661 assert(pd->cd->mask);
662 fetch->state = DIGEST_READ_MASK;
663 return StoreDigestCBlockSize;
664 } else {
665 peerDigestFetchAbort(fetch, buf, "invalid digest cblock");
666 return -1;
667 }
668 }
669
670 /* need more data, do we have space? */
671 if (size >= SM_PAGE_SIZE) {
672 peerDigestFetchAbort(fetch, buf, "digest cblock too big");
673 return -1;
674 }
675
676 return 0; /* We need more data */
677 }
678
679 int
peerDigestSwapInMask(void * data,char * buf,ssize_t size)680 peerDigestSwapInMask(void *data, char *buf, ssize_t size)
681 {
682 DigestFetchState *fetch = (DigestFetchState *)data;
683 PeerDigest *pd;
684
685 pd = fetch->pd;
686 assert(pd->cd && pd->cd->mask);
687
688 /*
689 * NOTENOTENOTENOTENOTE: buf doesn't point to pd->cd->mask anymore!
690 * we need to do the copy ourselves!
691 */
692 memcpy(pd->cd->mask + fetch->mask_offset, buf, size);
693
694 /* NOTE! buf points to the middle of pd->cd->mask! */
695
696 if (peerDigestFetchedEnough(fetch, NULL, size, "peerDigestSwapInMask"))
697 return -1;
698
699 fetch->mask_offset += size;
700
701 if (fetch->mask_offset >= pd->cd->mask_size) {
702 debugs(72, 2, "peerDigestSwapInMask: Done! Got " <<
703 fetch->mask_offset << ", expected " << pd->cd->mask_size);
704 assert(fetch->mask_offset == pd->cd->mask_size);
705 assert(peerDigestFetchedEnough(fetch, NULL, 0, "peerDigestSwapInMask"));
706 return -1; /* XXX! */
707 }
708
709 /* We always read everything, so return size */
710 return size;
711 }
712
713 static int
peerDigestFetchedEnough(DigestFetchState * fetch,char * buf,ssize_t size,const char * step_name)714 peerDigestFetchedEnough(DigestFetchState * fetch, char *buf, ssize_t size, const char *step_name)
715 {
716 PeerDigest *pd = NULL;
717 const char *host = "<unknown>"; /* peer host */
718 const char *reason = NULL; /* reason for completion */
719 const char *no_bug = NULL; /* successful completion if set */
720 const int pdcb_valid = cbdataReferenceValid(fetch->pd);
721 const int pcb_valid = cbdataReferenceValid(fetch->pd->peer);
722
723 /* test possible exiting conditions (the same for most steps!)
724 * cases marked with '?!' should not happen */
725
726 if (!reason) {
727 if (!(pd = fetch->pd))
728 reason = "peer digest disappeared?!";
729
730 #if DONT /* WHY NOT? /HNO */
731
732 else if (!cbdataReferenceValid(pd))
733 reason = "invalidated peer digest?!";
734
735 #endif
736
737 else
738 host = pd->host.termedBuf();
739 }
740
741 debugs(72, 6, step_name << ": peer " << host << ", offset: " <<
742 fetch->offset << " size: " << size << ".");
743
744 /* continue checking (with pd and host known and valid) */
745
746 if (!reason) {
747 if (!cbdataReferenceValid(pd->peer))
748 reason = "peer disappeared";
749 else if (size < 0)
750 reason = "swap failure";
751 else if (!fetch->entry)
752 reason = "swap aborted?!";
753 else if (EBIT_TEST(fetch->entry->flags, ENTRY_ABORTED))
754 reason = "swap aborted";
755 }
756
757 /* continue checking (maybe-successful eof case) */
758 if (!reason && !size) {
759 if (!pd->cd)
760 reason = "null digest?!";
761 else if (fetch->mask_offset != pd->cd->mask_size)
762 reason = "premature end of digest?!";
763 else if (!peerDigestUseful(pd))
764 reason = "useless digest";
765 else
766 reason = no_bug = "success";
767 }
768
769 /* finish if we have a reason */
770 if (reason) {
771 const int level = strstr(reason, "?!") ? 1 : 3;
772 debugs(72, level, "" << step_name << ": peer " << host << ", exiting after '" << reason << "'");
773 peerDigestReqFinish(fetch, buf,
774 1, pdcb_valid, pcb_valid, reason, !no_bug);
775 } else {
776 /* paranoid check */
777 assert(pdcb_valid && pcb_valid);
778 }
779
780 return reason != NULL;
781 }
782
783 /* call this when all callback data is valid and fetch must be stopped but
784 * no error has occurred (e.g. we received 304 reply and reuse old digest) */
785 static void
peerDigestFetchStop(DigestFetchState * fetch,char * buf,const char * reason)786 peerDigestFetchStop(DigestFetchState * fetch, char *buf, const char *reason)
787 {
788 assert(reason);
789 debugs(72, 2, "peerDigestFetchStop: peer " << fetch->pd->host << ", reason: " << reason);
790 peerDigestReqFinish(fetch, buf, 1, 1, 1, reason, 0);
791 }
792
793 /* call this when all callback data is valid but something bad happened */
794 static void
peerDigestFetchAbort(DigestFetchState * fetch,char * buf,const char * reason)795 peerDigestFetchAbort(DigestFetchState * fetch, char *buf, const char *reason)
796 {
797 assert(reason);
798 debugs(72, 2, "peerDigestFetchAbort: peer " << fetch->pd->host << ", reason: " << reason);
799 peerDigestReqFinish(fetch, buf, 1, 1, 1, reason, 1);
800 }
801
802 /* complete the digest transfer, update stats, unlock/release everything */
803 static void
peerDigestReqFinish(DigestFetchState * fetch,char * buf,int fcb_valid,int pdcb_valid,int pcb_valid,const char * reason,int err)804 peerDigestReqFinish(DigestFetchState * fetch, char *buf,
805 int fcb_valid, int pdcb_valid, int pcb_valid,
806 const char *reason, int err)
807 {
808 assert(reason);
809
810 /* must go before peerDigestPDFinish */
811
812 if (pdcb_valid) {
813 fetch->pd->flags.requested = false;
814 fetch->pd->req_result = reason;
815 }
816
817 /* schedule next check if peer is still out there */
818 if (pcb_valid) {
819 PeerDigest *pd = fetch->pd;
820
821 if (err) {
822 pd->times.retry_delay = peerDigestIncDelay(pd);
823 peerDigestSetCheck(pd, pd->times.retry_delay);
824 } else {
825 pd->times.retry_delay = 0;
826 peerDigestSetCheck(pd, peerDigestNewDelay(fetch->entry));
827 }
828 }
829
830 /* note: order is significant */
831 if (fcb_valid)
832 peerDigestFetchSetStats(fetch);
833
834 if (pdcb_valid)
835 peerDigestPDFinish(fetch, pcb_valid, err);
836
837 if (fcb_valid)
838 peerDigestFetchFinish(fetch, err);
839 }
840
841 /* destroys digest if peer disappeared
842 * must be called only when fetch and pd cbdata are valid */
843 static void
peerDigestPDFinish(DigestFetchState * fetch,int pcb_valid,int err)844 peerDigestPDFinish(DigestFetchState * fetch, int pcb_valid, int err)
845 {
846 PeerDigest *pd = fetch->pd;
847 const char *host = pd->host.termedBuf();
848
849 pd->times.received = squid_curtime;
850 pd->times.req_delay = fetch->resp_time;
851 pd->stats.sent.kbytes += fetch->sent.bytes;
852 pd->stats.recv.kbytes += fetch->recv.bytes;
853 pd->stats.sent.msgs += fetch->sent.msg;
854 pd->stats.recv.msgs += fetch->recv.msg;
855
856 if (err) {
857 debugs(72, DBG_IMPORTANT, "" << (pcb_valid ? "temporary " : "" ) << "disabling (" << pd->req_result << ") digest from " << host);
858
859 delete pd->cd;
860 pd->cd = nullptr;
861
862 pd->flags.usable = false;
863
864 if (!pcb_valid)
865 peerDigestNotePeerGone(pd);
866 } else {
867 assert(pcb_valid);
868
869 pd->flags.usable = true;
870
871 /* XXX: ugly condition, but how? */
872
873 if (fetch->entry->store_status == STORE_OK)
874 debugs(72, 2, "re-used old digest from " << host);
875 else
876 debugs(72, 2, "received valid digest from " << host);
877 }
878
879 cbdataReferenceDone(fetch->pd);
880 }
881
882 /* free fetch state structures
883 * must be called only when fetch cbdata is valid */
884 static void
peerDigestFetchFinish(DigestFetchState * fetch,int err)885 peerDigestFetchFinish(DigestFetchState * fetch, int err)
886 {
887 assert(fetch->entry && fetch->request);
888
889 if (fetch->old_entry) {
890 debugs(72, 3, "peerDigestFetchFinish: deleting old entry");
891 storeUnregister(fetch->old_sc, fetch->old_entry, fetch);
892 fetch->old_entry->releaseRequest();
893 fetch->old_entry->unlock("peerDigestFetchFinish old");
894 fetch->old_entry = NULL;
895 }
896
897 /* update global stats */
898 statCounter.cd.kbytes_sent += fetch->sent.bytes;
899 statCounter.cd.kbytes_recv += fetch->recv.bytes;
900 statCounter.cd.msgs_sent += fetch->sent.msg;
901 statCounter.cd.msgs_recv += fetch->recv.msg;
902
903 delete fetch;
904 }
905
906 /* calculate fetch stats after completion */
907 static void
peerDigestFetchSetStats(DigestFetchState * fetch)908 peerDigestFetchSetStats(DigestFetchState * fetch)
909 {
910 MemObject *mem;
911 assert(fetch->entry && fetch->request);
912
913 mem = fetch->entry->mem_obj;
914 assert(mem);
915
916 /* XXX: outgoing numbers are not precise */
917 /* XXX: we must distinguish between 304 hits and misses here */
918 fetch->sent.bytes = fetch->request->prefixLen();
919 /* XXX: this is slightly wrong: we don't KNOW that the entire memobject
920 * was fetched. We only know how big it is
921 */
922 fetch->recv.bytes = mem->size();
923 fetch->sent.msg = fetch->recv.msg = 1;
924 fetch->expires = fetch->entry->expires;
925 fetch->resp_time = squid_curtime - fetch->start_time;
926
927 debugs(72, 3, "peerDigestFetchFinish: recv " << fetch->recv.bytes <<
928 " bytes in " << (int) fetch->resp_time << " secs");
929
930 debugs(72, 3, "peerDigestFetchFinish: expires: " <<
931 (long int) fetch->expires << " (" << std::showpos <<
932 (int) (fetch->expires - squid_curtime) << "), lmt: " <<
933 std::noshowpos << (long int) fetch->entry->lastModified() << " (" <<
934 std::showpos << (int) (fetch->entry->lastModified() - squid_curtime) <<
935 ")");
936
937 }
938
939 static int
peerDigestSetCBlock(PeerDigest * pd,const char * buf)940 peerDigestSetCBlock(PeerDigest * pd, const char *buf)
941 {
942 StoreDigestCBlock cblock;
943 int freed_size = 0;
944 const char *host = pd->host.termedBuf();
945
946 memcpy(&cblock, buf, sizeof(cblock));
947 /* network -> host conversions */
948 cblock.ver.current = ntohs(cblock.ver.current);
949 cblock.ver.required = ntohs(cblock.ver.required);
950 cblock.capacity = ntohl(cblock.capacity);
951 cblock.count = ntohl(cblock.count);
952 cblock.del_count = ntohl(cblock.del_count);
953 cblock.mask_size = ntohl(cblock.mask_size);
954 debugs(72, 2, "got digest cblock from " << host << "; ver: " <<
955 (int) cblock.ver.current << " (req: " << (int) cblock.ver.required <<
956 ")");
957
958 debugs(72, 2, "\t size: " <<
959 cblock.mask_size << " bytes, e-cnt: " <<
960 cblock.count << ", e-util: " <<
961 xpercentInt(cblock.count, cblock.capacity) << "%" );
962 /* check version requirements (both ways) */
963
964 if (cblock.ver.required > CacheDigestVer.current) {
965 debugs(72, DBG_IMPORTANT, "" << host << " digest requires version " <<
966 cblock.ver.required << "; have: " << CacheDigestVer.current);
967
968 return 0;
969 }
970
971 if (cblock.ver.current < CacheDigestVer.required) {
972 debugs(72, DBG_IMPORTANT, "" << host << " digest is version " <<
973 cblock.ver.current << "; we require: " <<
974 CacheDigestVer.required);
975
976 return 0;
977 }
978
979 /* check consistency */
980 if (cblock.ver.required > cblock.ver.current ||
981 cblock.mask_size <= 0 || cblock.capacity <= 0 ||
982 cblock.bits_per_entry <= 0 || cblock.hash_func_count <= 0) {
983 debugs(72, DBG_CRITICAL, "" << host << " digest cblock is corrupted.");
984 return 0;
985 }
986
987 /* check consistency further */
988 if ((size_t)cblock.mask_size != CacheDigest::CalcMaskSize(cblock.capacity, cblock.bits_per_entry)) {
989 debugs(72, DBG_CRITICAL, host << " digest cblock is corrupted " <<
990 "(mask size mismatch: " << cblock.mask_size << " ? " <<
991 CacheDigest::CalcMaskSize(cblock.capacity, cblock.bits_per_entry)
992 << ").");
993 return 0;
994 }
995
996 /* there are some things we cannot do yet */
997 if (cblock.hash_func_count != CacheDigestHashFuncCount) {
998 debugs(72, DBG_CRITICAL, "" << host << " digest: unsupported #hash functions: " <<
999 cblock.hash_func_count << " ? " << CacheDigestHashFuncCount << ".");
1000 return 0;
1001 }
1002
1003 /*
1004 * no cblock bugs below this point
1005 */
1006 /* check size changes */
1007 if (pd->cd && cblock.mask_size != (ssize_t)pd->cd->mask_size) {
1008 debugs(72, 2, host << " digest changed size: " << cblock.mask_size <<
1009 " -> " << pd->cd->mask_size);
1010 freed_size = pd->cd->mask_size;
1011 delete pd->cd;
1012 pd->cd = nullptr;
1013 }
1014
1015 if (!pd->cd) {
1016 debugs(72, 2, "creating " << host << " digest; size: " << cblock.mask_size << " (" <<
1017 std::showpos << (int) (cblock.mask_size - freed_size) << ") bytes");
1018 pd->cd = new CacheDigest(cblock.capacity, cblock.bits_per_entry);
1019
1020 if (cblock.mask_size >= freed_size)
1021 statCounter.cd.memory += (cblock.mask_size - freed_size);
1022 }
1023
1024 assert(pd->cd);
1025 /* these assignments leave us in an inconsistent state until we finish reading the digest */
1026 pd->cd->count = cblock.count;
1027 pd->cd->del_count = cblock.del_count;
1028 return 1;
1029 }
1030
1031 static int
peerDigestUseful(const PeerDigest * pd)1032 peerDigestUseful(const PeerDigest * pd)
1033 {
1034 /* TODO: we should calculate the prob of a false hit instead of bit util */
1035 const auto bit_util = pd->cd->usedMaskPercent();
1036
1037 if (bit_util > 65.0) {
1038 debugs(72, DBG_CRITICAL, "Warning: " << pd->host <<
1039 " peer digest has too many bits on (" << bit_util << "%).");
1040 return 0;
1041 }
1042
1043 return 1;
1044 }
1045
1046 static int
saneDiff(time_t diff)1047 saneDiff(time_t diff)
1048 {
1049 return abs((int) diff) > squid_curtime / 2 ? 0 : diff;
1050 }
1051
1052 void
peerDigestStatsReport(const PeerDigest * pd,StoreEntry * e)1053 peerDigestStatsReport(const PeerDigest * pd, StoreEntry * e)
1054 {
1055 #define f2s(flag) (pd->flags.flag ? "yes" : "no")
1056 #define appendTime(tm) storeAppendPrintf(e, "%s\t %10ld\t %+d\t %+d\n", \
1057 ""#tm, (long int)pd->times.tm, \
1058 saneDiff(pd->times.tm - squid_curtime), \
1059 saneDiff(pd->times.tm - pd->times.initialized))
1060
1061 assert(pd);
1062
1063 const char *host = pd->host.termedBuf();
1064 storeAppendPrintf(e, "\npeer digest from %s\n", host);
1065
1066 cacheDigestGuessStatsReport(&pd->stats.guess, e, host);
1067
1068 storeAppendPrintf(e, "\nevent\t timestamp\t secs from now\t secs from init\n");
1069 appendTime(initialized);
1070 appendTime(needed);
1071 appendTime(requested);
1072 appendTime(received);
1073 appendTime(next_check);
1074
1075 storeAppendPrintf(e, "peer digest state:\n");
1076 storeAppendPrintf(e, "\tneeded: %3s, usable: %3s, requested: %3s\n",
1077 f2s(needed), f2s(usable), f2s(requested));
1078 storeAppendPrintf(e, "\n\tlast retry delay: %d secs\n",
1079 (int) pd->times.retry_delay);
1080 storeAppendPrintf(e, "\tlast request response time: %d secs\n",
1081 (int) pd->times.req_delay);
1082 storeAppendPrintf(e, "\tlast request result: %s\n",
1083 pd->req_result ? pd->req_result : "(none)");
1084
1085 storeAppendPrintf(e, "\npeer digest traffic:\n");
1086 storeAppendPrintf(e, "\trequests sent: %d, volume: %d KB\n",
1087 pd->stats.sent.msgs, (int) pd->stats.sent.kbytes.kb);
1088 storeAppendPrintf(e, "\treplies recv: %d, volume: %d KB\n",
1089 pd->stats.recv.msgs, (int) pd->stats.recv.kbytes.kb);
1090
1091 storeAppendPrintf(e, "\npeer digest structure:\n");
1092
1093 if (pd->cd)
1094 cacheDigestReport(pd->cd, host, e);
1095 else
1096 storeAppendPrintf(e, "\tno in-memory copy\n");
1097 }
1098
1099 #endif
1100
1101