2 * Copyright (C) 1996-2015 The Squid Software Foundation and contributors
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.
9 /* DEBUG: section 72 Peer Digest Routines */
13 #include "CacheDigest.h"
14 #include "CachePeer.h"
18 #include "HttpReply.h"
19 #include "HttpRequest.h"
21 #include "MemObject.h"
22 #include "mime_header.h"
23 #include "neighbors.h"
24 #include "PeerDigest.h"
25 #include "SquidTime.h"
27 #include "store_key_md5.h"
28 #include "StoreClient.h"
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
);
56 Version
const CacheDigestVer
= { 5, 3 };
58 #define StoreDigestCBlockSize sizeof(StoreDigestCBlock)
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 */
67 static time_t pd_last_req_time
= 0; /* last call to Check */
69 /* initialize peer digest */
71 peerDigestInit(PeerDigest
* pd
, CachePeer
* p
)
75 memset(pd
, 0, sizeof(*pd
));
78 * Lock on to the peer here. The corresponding cbdataReferenceDone()
79 * is in peerDigestDestroy().
81 pd
->peer
= cbdataReference(p
);
82 /* if peer disappears, we will know it's name */
85 pd
->times
.initialized
= squid_curtime
;
88 CBDATA_CLASS_INIT(PeerDigest
);
90 CBDATA_CLASS_INIT(DigestFetchState
);
92 DigestFetchState::DigestFetchState(PeerDigest
*aPd
, HttpRequest
*req
) :
93 pd(cbdataReference(aPd
)),
101 start_time(squid_curtime
),
105 state(DIGEST_READ_REPLY
)
107 HTTPMSGLOCK(request
);
118 DigestFetchState::~DigestFetchState()
120 /* unlock everything */
121 storeUnregister(sc
, entry
, this);
123 entry
->unlock("DigestFetchState destructed");
126 HTTPMSGUNLOCK(request
);
131 /* allocate new peer digest, call Init, and lock everything */
133 peerDigestCreate(CachePeer
* p
)
139 peerDigestInit(pd
, p
);
141 /* XXX This does not look right, and the same thing again in the caller */
142 return cbdataReference(pd
);
145 /* call Clean and free/unlock everything */
147 peerDigestDestroy(PeerDigest
* pd
)
151 void * peerTmp
= pd
->peer
;
155 * We locked the peer in peerDigestInit(), this is
156 * where we unlock it. If the peer is still valid,
157 * tell it that the digest is gone.
159 if (cbdataReferenceValidDone(peerTmp
, &p
))
160 peerNoteDigestGone((CachePeer
*)p
);
168 /* called by peer to indicate that somebody actually needs this digest */
170 peerDigestNeeded(PeerDigest
* pd
)
173 assert(!pd
->flags
.needed
);
176 pd
->flags
.needed
= true;
177 pd
->times
.needed
= squid_curtime
;
178 peerDigestSetCheck(pd
, 0); /* check asap */
181 /* currently we do not have a reason to disable without destroying */
183 /* disables peer for good */
185 peerDigestDisable(PeerDigest
* pd
)
187 debugs(72, 2, "peerDigestDisable: peer " << pd
->host
.buf() << " disabled for good");
188 pd
->times
.disabled
= squid_curtime
;
189 pd
->times
.next_check
= -1; /* never */
190 pd
->flags
.usable
= 0;
195 /* we do not destroy the pd itself to preserve its "history" and stats */
200 /* increment retry delay [after an unsuccessful attempt] */
202 peerDigestIncDelay(const PeerDigest
* pd
)
205 return pd
->times
.retry_delay
> 0 ?
206 2 * pd
->times
.retry_delay
: /* exponential backoff */
207 PeerDigestReqMinGap
; /* minimal delay */
210 /* artificially increases Expires: setting to avoid race conditions
211 * returns the delay till that [increased] expiration time */
213 peerDigestNewDelay(const StoreEntry
* e
)
218 return e
->expires
+ PeerDigestReqMinGap
- squid_curtime
;
220 return PeerDigestReqMinGap
;
223 /* registers next digest verification */
225 peerDigestSetCheck(PeerDigest
* pd
, time_t delay
)
227 eventAdd("peerDigestCheck", peerDigestCheck
, pd
, (double) delay
, 1);
228 pd
->times
.next_check
= squid_curtime
+ delay
;
229 debugs(72, 3, "peerDigestSetCheck: will check peer " << pd
->host
<< " in " << delay
<< " secs");
233 * called when peer is about to disappear or have already disappeared
236 peerDigestNotePeerGone(PeerDigest
* pd
)
238 if (pd
->flags
.requested
) {
239 debugs(72, 2, "peerDigest: peer " << pd
->host
<< " gone, will destroy after fetch.");
240 /* do nothing now, the fetching chain will notice and take action */
242 debugs(72, 2, "peerDigest: peer " << pd
->host
<< " is gone, destroying now.");
243 peerDigestDestroy(pd
);
247 /* callback for eventAdd() (with peer digest locked)
248 * request new digest if our copy is too old or if we lack one;
249 * schedule next check otherwise */
251 peerDigestCheck(void *data
)
253 PeerDigest
*pd
= (PeerDigest
*)data
;
256 assert(!pd
->flags
.requested
);
258 pd
->times
.next_check
= 0; /* unknown */
260 if (!cbdataReferenceValid(pd
->peer
)) {
261 peerDigestNotePeerGone(pd
);
265 debugs(72, 3, "peerDigestCheck: peer " << pd
->peer
->host
<< ":" << pd
->peer
->http_port
);
266 debugs(72, 3, "peerDigestCheck: time: " << squid_curtime
<<
267 ", last received: " << (long int) pd
->times
.received
<< " (" <<
268 std::showpos
<< (int) (squid_curtime
- pd
->times
.received
) << ")");
270 /* decide when we should send the request:
271 * request now unless too close to other requests */
272 req_time
= squid_curtime
;
276 if (req_time
- pd
->times
.received
< PeerDigestReqMinGap
) {
277 debugs(72, 2, "peerDigestCheck: " << pd
->host
<<
278 ", avoiding close peer requests (" <<
279 (int) (req_time
- pd
->times
.received
) << " < " <<
280 (int) PeerDigestReqMinGap
<< " secs).");
282 req_time
= pd
->times
.received
+ PeerDigestReqMinGap
;
286 if (req_time
- pd_last_req_time
< GlobDigestReqMinGap
) {
287 debugs(72, 2, "peerDigestCheck: " << pd
->host
<<
288 ", avoiding close requests (" <<
289 (int) (req_time
- pd_last_req_time
) << " < " <<
290 (int) GlobDigestReqMinGap
<< " secs).");
292 req_time
= pd_last_req_time
+ GlobDigestReqMinGap
;
295 if (req_time
<= squid_curtime
)
296 peerDigestRequest(pd
); /* will set pd->flags.requested */
298 peerDigestSetCheck(pd
, req_time
- squid_curtime
);
301 /* ask store for a digest */
303 peerDigestRequest(PeerDigest
* pd
)
305 CachePeer
*p
= pd
->peer
;
306 StoreEntry
*e
, *old_e
;
308 const cache_key
*key
;
310 StoreIOBuffer tempBuffer
;
312 pd
->req_result
= NULL
;
313 pd
->flags
.requested
= true;
315 /* compute future request components */
318 url
= xstrdup(p
->digest_url
);
320 url
= xstrdup(internalRemoteUri(p
->host
, p
->http_port
, "/squid-internal-periodic/", SBuf(StoreDigestFileName
)));
322 req
= HttpRequest::CreateFromUrl(url
);
326 key
= storeKeyPublicByRequest(req
);
328 debugs(72, 2, "peerDigestRequest: " << url
<< " key: " << storeKeyText(key
));
330 /* add custom headers */
331 assert(!req
->header
.len
);
333 req
->header
.putStr(Http::HdrType::ACCEPT
, StoreDigestMimeStr
);
335 req
->header
.putStr(Http::HdrType::ACCEPT
, "text/html");
338 p
->login
[0] != '*' &&
339 strcmp(p
->login
, "PASS") != 0 &&
340 strcmp(p
->login
, "PASSTHRU") != 0 &&
341 strncmp(p
->login
, "NEGOTIATE",9) != 0 &&
342 strcmp(p
->login
, "PROXYPASS") != 0) {
343 req
->url
.userInfo(SBuf(p
->login
)); // XXX: performance regression make peer login SBuf as well.
345 /* create fetch state structure */
346 DigestFetchState
*fetch
= new DigestFetchState(pd
, req
);
348 /* update timestamps */
349 pd
->times
.requested
= squid_curtime
;
350 pd_last_req_time
= squid_curtime
;
351 req
->flags
.cachable
= true;
353 /* the rest is based on clientProcessExpired() */
354 req
->flags
.refresh
= true;
356 old_e
= fetch
->old_entry
= Store::Root().get(key
);
359 debugs(72, 5, "peerDigestRequest: found old entry");
361 old_e
->lock("peerDigestRequest");
362 old_e
->createMemObject(url
, url
, req
->method
);
364 fetch
->old_sc
= storeClientListAdd(old_e
, fetch
);
367 e
= fetch
->entry
= storeCreateEntry(url
, url
, req
->flags
, req
->method
);
368 assert(EBIT_TEST(e
->flags
, KEY_PRIVATE
));
369 fetch
->sc
= storeClientListAdd(e
, fetch
);
370 /* set lastmod to trigger IMS request if possible */
373 e
->lastmod
= old_e
->lastmod
;
375 /* push towards peer cache */
376 debugs(72, 3, "peerDigestRequest: forwarding to fwdStart...");
378 FwdState::fwdStart(Comm::ConnectionPointer(), e
, req
);
380 tempBuffer
.offset
= 0;
382 tempBuffer
.length
= SM_PAGE_SIZE
;
384 tempBuffer
.data
= fetch
->buf
;
386 storeClientCopy(fetch
->sc
, e
, tempBuffer
,
387 peerDigestHandleReply
, fetch
);
392 /* Handle the data copying .. */
395 * This routine handles the copy data and then redirects the
396 * copy to a bunch of subfunctions depending upon the copy state.
397 * It also tracks the buffer offset and "seen", since I'm actually
398 * not interested in rewriting everything to suit my little idea.
401 peerDigestHandleReply(void *data
, StoreIOBuffer receivedData
)
403 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
405 digest_read_state_t prevstate
;
408 assert(fetch
->pd
&& receivedData
.data
);
409 /* The existing code assumes that the received pointer is
410 * where we asked the data to be put
412 assert(fetch
->buf
+ fetch
->bufofs
== receivedData
.data
);
414 /* Update the buffer size */
415 fetch
->bufofs
+= receivedData
.length
;
417 assert(fetch
->bufofs
<= SM_PAGE_SIZE
);
419 /* If we've fetched enough, return */
421 if (peerDigestFetchedEnough(fetch
, fetch
->buf
, fetch
->bufofs
, "peerDigestHandleReply"))
424 /* Call the right function based on the state */
425 /* (Those functions will update the state if needed) */
427 /* Give us a temporary reference. Some of the calls we make may
428 * try to destroy the fetch structure, and we like to know if they
431 CbcPointer
<DigestFetchState
> tmpLock
= fetch
;
433 /* Repeat this loop until we're out of data OR the state changes */
434 /* (So keep going if the state has changed and we still have data */
436 prevstate
= fetch
->state
;
438 switch (fetch
->state
) {
440 case DIGEST_READ_REPLY
:
441 retsize
= peerDigestFetchReply(fetch
, fetch
->buf
, fetch
->bufofs
);
444 case DIGEST_READ_HEADERS
:
445 retsize
= peerDigestSwapInHeaders(fetch
, fetch
->buf
, fetch
->bufofs
);
448 case DIGEST_READ_CBLOCK
:
449 retsize
= peerDigestSwapInCBlock(fetch
, fetch
->buf
, fetch
->bufofs
);
452 case DIGEST_READ_MASK
:
453 retsize
= peerDigestSwapInMask(fetch
, fetch
->buf
, fetch
->bufofs
);
456 case DIGEST_READ_NONE
:
459 case DIGEST_READ_DONE
:
464 fatal("Bad digest transfer mode!\n");
471 * The returned size indicates how much of the buffer was read -
472 * so move the remainder of the buffer to the beginning
473 * and update the bufofs / bufsize
475 newsize
= fetch
->bufofs
- retsize
;
477 memmove(fetch
->buf
, fetch
->buf
+ retsize
, fetch
->bufofs
- newsize
);
479 fetch
->bufofs
= newsize
;
481 } while (cbdataReferenceValid(fetch
) && prevstate
!= fetch
->state
&& fetch
->bufofs
> 0);
483 /* Update the copy offset */
484 fetch
->offset
+= receivedData
.length
;
486 /* Schedule another copy */
487 if (cbdataReferenceValid(fetch
)) {
488 StoreIOBuffer tempBuffer
;
489 tempBuffer
.offset
= fetch
->offset
;
490 tempBuffer
.length
= SM_PAGE_SIZE
- fetch
->bufofs
;
491 tempBuffer
.data
= fetch
->buf
+ fetch
->bufofs
;
492 storeClientCopy(fetch
->sc
, fetch
->entry
, tempBuffer
,
493 peerDigestHandleReply
, fetch
);
497 /* wait for full http headers to be received then parse them */
499 * This routine handles parsing the reply line.
500 * If the reply line indicates an OK, the same data is thrown
501 * to SwapInHeaders(). If the reply line is a NOT_MODIFIED,
502 * we simply stop parsing.
505 peerDigestFetchReply(void *data
, char *buf
, ssize_t size
)
507 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
508 PeerDigest
*pd
= fetch
->pd
;
511 assert(!fetch
->offset
);
513 assert(fetch
->state
== DIGEST_READ_REPLY
);
515 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestFetchReply"))
518 if ((hdr_size
= headersEnd(buf
, size
))) {
519 HttpReply
const *reply
= fetch
->entry
->getReply();
521 assert(reply
->sline
.status() != Http::scNone
);
522 const Http::StatusCode status
= reply
->sline
.status();
523 debugs(72, 3, "peerDigestFetchReply: " << pd
->host
<< " status: " << status
<<
524 ", expires: " << (long int) reply
->expires
<< " (" << std::showpos
<<
525 (int) (reply
->expires
- squid_curtime
) << ")");
527 /* this "if" is based on clientHandleIMSReply() */
529 if (status
== Http::scNotModified
) {
530 /* our old entry is fine */
531 assert(fetch
->old_entry
);
533 if (!fetch
->old_entry
->mem_obj
->request
) {
534 fetch
->old_entry
->mem_obj
->request
= fetch
->entry
->mem_obj
->request
;
535 HTTPMSGLOCK(fetch
->old_entry
->mem_obj
->request
);
538 assert(fetch
->old_entry
->mem_obj
->request
);
540 HttpReply
*old_rep
= (HttpReply
*) fetch
->old_entry
->getReply();
542 old_rep
->updateOnNotModified(reply
);
544 fetch
->old_entry
->timestampsSet();
546 /* get rid of 304 reply */
547 storeUnregister(fetch
->sc
, fetch
->entry
, fetch
);
549 fetch
->entry
->unlock("peerDigestFetchReply 304");
551 fetch
->entry
= fetch
->old_entry
;
553 fetch
->old_entry
= NULL
;
555 /* preserve request -- we need its size to update counters */
556 /* requestUnlink(r); */
557 /* fetch->entry->mem_obj->request = NULL; */
558 } else if (status
== Http::scOkay
) {
559 /* get rid of old entry if any */
561 if (fetch
->old_entry
) {
562 debugs(72, 3, "peerDigestFetchReply: got new digest, releasing old one");
563 storeUnregister(fetch
->old_sc
, fetch
->old_entry
, fetch
);
564 fetch
->old_entry
->releaseRequest();
565 fetch
->old_entry
->unlock("peerDigestFetchReply 200");
566 fetch
->old_entry
= NULL
;
569 /* some kind of a bug */
570 peerDigestFetchAbort(fetch
, buf
, reply
->sline
.reason());
571 return -1; /* XXX -1 will abort stuff in ReadReply! */
574 /* must have a ready-to-use store entry if we got here */
575 /* can we stay with the old in-memory digest? */
576 if (status
== Http::scNotModified
&& fetch
->pd
->cd
) {
577 peerDigestFetchStop(fetch
, buf
, "Not modified");
578 fetch
->state
= DIGEST_READ_DONE
;
580 fetch
->state
= DIGEST_READ_HEADERS
;
583 /* need more data, do we have space? */
585 if (size
>= SM_PAGE_SIZE
)
586 peerDigestFetchAbort(fetch
, buf
, "reply header too big");
589 /* We don't want to actually ack that we've handled anything,
590 * otherwise SwapInHeaders() won't get the reply line .. */
594 /* fetch headers from disk, pass on to SwapInCBlock */
596 peerDigestSwapInHeaders(void *data
, char *buf
, ssize_t size
)
598 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
601 assert(fetch
->state
== DIGEST_READ_HEADERS
);
603 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestSwapInHeaders"))
606 assert(!fetch
->offset
);
608 if ((hdr_size
= headersEnd(buf
, size
))) {
609 assert(fetch
->entry
->getReply());
610 assert(fetch
->entry
->getReply()->sline
.status() != Http::scNone
);
612 if (fetch
->entry
->getReply()->sline
.status() != Http::scOkay
) {
613 debugs(72, DBG_IMPORTANT
, "peerDigestSwapInHeaders: " << fetch
->pd
->host
<<
614 " status " << fetch
->entry
->getReply()->sline
.status() <<
617 peerDigestFetchAbort(fetch
, buf
, "internal status error");
621 fetch
->state
= DIGEST_READ_CBLOCK
;
622 return hdr_size
; /* Say how much data we read */
625 /* need more data, do we have space? */
626 if (size
>= SM_PAGE_SIZE
) {
627 peerDigestFetchAbort(fetch
, buf
, "stored header too big");
631 return 0; /* We need to read more to parse .. */
635 peerDigestSwapInCBlock(void *data
, char *buf
, ssize_t size
)
637 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
639 assert(fetch
->state
== DIGEST_READ_CBLOCK
);
641 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestSwapInCBlock"))
644 if (size
>= (ssize_t
)StoreDigestCBlockSize
) {
645 PeerDigest
*pd
= fetch
->pd
;
647 assert(pd
&& fetch
->entry
->getReply());
649 if (peerDigestSetCBlock(pd
, buf
)) {
650 /* XXX: soon we will have variable header size */
651 /* switch to CD buffer and fetch digest guts */
653 assert(pd
->cd
->mask
);
654 fetch
->state
= DIGEST_READ_MASK
;
655 return StoreDigestCBlockSize
;
657 peerDigestFetchAbort(fetch
, buf
, "invalid digest cblock");
662 /* need more data, do we have space? */
663 if (size
>= SM_PAGE_SIZE
) {
664 peerDigestFetchAbort(fetch
, buf
, "digest cblock too big");
668 return 0; /* We need more data */
672 peerDigestSwapInMask(void *data
, char *buf
, ssize_t size
)
674 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
678 assert(pd
->cd
&& pd
->cd
->mask
);
681 * NOTENOTENOTENOTENOTE: buf doesn't point to pd->cd->mask anymore!
682 * we need to do the copy ourselves!
684 memcpy(pd
->cd
->mask
+ fetch
->mask_offset
, buf
, size
);
686 /* NOTE! buf points to the middle of pd->cd->mask! */
688 if (peerDigestFetchedEnough(fetch
, NULL
, size
, "peerDigestSwapInMask"))
691 fetch
->mask_offset
+= size
;
693 if (fetch
->mask_offset
>= pd
->cd
->mask_size
) {
694 debugs(72, 2, "peerDigestSwapInMask: Done! Got " <<
695 fetch
->mask_offset
<< ", expected " << pd
->cd
->mask_size
);
696 assert(fetch
->mask_offset
== pd
->cd
->mask_size
);
697 assert(peerDigestFetchedEnough(fetch
, NULL
, 0, "peerDigestSwapInMask"));
698 return -1; /* XXX! */
701 /* We always read everything, so return size */
706 peerDigestFetchedEnough(DigestFetchState
* fetch
, char *buf
, ssize_t size
, const char *step_name
)
708 PeerDigest
*pd
= NULL
;
709 const char *host
= "<unknown>"; /* peer host */
710 const char *reason
= NULL
; /* reason for completion */
711 const char *no_bug
= NULL
; /* successful completion if set */
712 const int pdcb_valid
= cbdataReferenceValid(fetch
->pd
);
713 const int pcb_valid
= cbdataReferenceValid(fetch
->pd
->peer
);
715 /* test possible exiting conditions (the same for most steps!)
716 * cases marked with '?!' should not happen */
719 if (!(pd
= fetch
->pd
))
720 reason
= "peer digest disappeared?!";
722 #if DONT /* WHY NOT? /HNO */
724 else if (!cbdataReferenceValid(pd
))
725 reason
= "invalidated peer digest?!";
730 host
= pd
->host
.termedBuf();
733 debugs(72, 6, step_name
<< ": peer " << host
<< ", offset: " <<
734 fetch
->offset
<< " size: " << size
<< ".");
736 /* continue checking (with pd and host known and valid) */
739 if (!cbdataReferenceValid(pd
->peer
))
740 reason
= "peer disappeared";
742 reason
= "swap failure";
743 else if (!fetch
->entry
)
744 reason
= "swap aborted?!";
745 else if (EBIT_TEST(fetch
->entry
->flags
, ENTRY_ABORTED
))
746 reason
= "swap aborted";
749 /* continue checking (maybe-successful eof case) */
750 if (!reason
&& !size
) {
752 reason
= "null digest?!";
753 else if (fetch
->mask_offset
!= (int)pd
->cd
->mask_size
)
754 reason
= "premature end of digest?!";
755 else if (!peerDigestUseful(pd
))
756 reason
= "useless digest";
758 reason
= no_bug
= "success";
761 /* finish if we have a reason */
763 const int level
= strstr(reason
, "?!") ? 1 : 3;
764 debugs(72, level
, "" << step_name
<< ": peer " << host
<< ", exiting after '" << reason
<< "'");
765 peerDigestReqFinish(fetch
, buf
,
766 1, pdcb_valid
, pcb_valid
, reason
, !no_bug
);
769 assert(pdcb_valid
&& pcb_valid
);
772 return reason
!= NULL
;
775 /* call this when all callback data is valid and fetch must be stopped but
776 * no error has occurred (e.g. we received 304 reply and reuse old digest) */
778 peerDigestFetchStop(DigestFetchState
* fetch
, char *buf
, const char *reason
)
781 debugs(72, 2, "peerDigestFetchStop: peer " << fetch
->pd
->host
<< ", reason: " << reason
);
782 peerDigestReqFinish(fetch
, buf
, 1, 1, 1, reason
, 0);
785 /* call this when all callback data is valid but something bad happened */
787 peerDigestFetchAbort(DigestFetchState
* fetch
, char *buf
, const char *reason
)
790 debugs(72, 2, "peerDigestFetchAbort: peer " << fetch
->pd
->host
<< ", reason: " << reason
);
791 peerDigestReqFinish(fetch
, buf
, 1, 1, 1, reason
, 1);
794 /* complete the digest transfer, update stats, unlock/release everything */
796 peerDigestReqFinish(DigestFetchState
* fetch
, char *buf
,
797 int fcb_valid
, int pdcb_valid
, int pcb_valid
,
798 const char *reason
, int err
)
802 /* must go before peerDigestPDFinish */
805 fetch
->pd
->flags
.requested
= false;
806 fetch
->pd
->req_result
= reason
;
809 /* schedule next check if peer is still out there */
811 PeerDigest
*pd
= fetch
->pd
;
814 pd
->times
.retry_delay
= peerDigestIncDelay(pd
);
815 peerDigestSetCheck(pd
, pd
->times
.retry_delay
);
817 pd
->times
.retry_delay
= 0;
818 peerDigestSetCheck(pd
, peerDigestNewDelay(fetch
->entry
));
822 /* note: order is significant */
824 peerDigestFetchSetStats(fetch
);
827 peerDigestPDFinish(fetch
, pcb_valid
, err
);
830 peerDigestFetchFinish(fetch
, err
);
833 /* destroys digest if peer disappeared
834 * must be called only when fetch and pd cbdata are valid */
836 peerDigestPDFinish(DigestFetchState
* fetch
, int pcb_valid
, int err
)
838 PeerDigest
*pd
= fetch
->pd
;
839 const char *host
= pd
->host
.termedBuf();
841 pd
->times
.received
= squid_curtime
;
842 pd
->times
.req_delay
= fetch
->resp_time
;
843 pd
->stats
.sent
.kbytes
+= fetch
->sent
.bytes
;
844 pd
->stats
.recv
.kbytes
+= fetch
->recv
.bytes
;
845 pd
->stats
.sent
.msgs
+= fetch
->sent
.msg
;
846 pd
->stats
.recv
.msgs
+= fetch
->recv
.msg
;
849 debugs(72, DBG_IMPORTANT
, "" << (pcb_valid
? "temporary " : "" ) << "disabling (" << pd
->req_result
<< ") digest from " << host
);
854 pd
->flags
.usable
= false;
857 peerDigestNotePeerGone(pd
);
861 pd
->flags
.usable
= true;
863 /* XXX: ugly condition, but how? */
865 if (fetch
->entry
->store_status
== STORE_OK
)
866 debugs(72, 2, "re-used old digest from " << host
);
868 debugs(72, 2, "received valid digest from " << host
);
871 cbdataReferenceDone(fetch
->pd
);
874 /* free fetch state structures
875 * must be called only when fetch cbdata is valid */
877 peerDigestFetchFinish(DigestFetchState
* fetch
, int err
)
879 assert(fetch
->entry
&& fetch
->request
);
881 if (fetch
->old_entry
) {
882 debugs(72, 3, "peerDigestFetchFinish: deleting old entry");
883 storeUnregister(fetch
->old_sc
, fetch
->old_entry
, fetch
);
884 fetch
->old_entry
->releaseRequest();
885 fetch
->old_entry
->unlock("peerDigestFetchFinish old");
886 fetch
->old_entry
= NULL
;
889 /* update global stats */
890 statCounter
.cd
.kbytes_sent
+= fetch
->sent
.bytes
;
891 statCounter
.cd
.kbytes_recv
+= fetch
->recv
.bytes
;
892 statCounter
.cd
.msgs_sent
+= fetch
->sent
.msg
;
893 statCounter
.cd
.msgs_recv
+= fetch
->recv
.msg
;
898 /* calculate fetch stats after completion */
900 peerDigestFetchSetStats(DigestFetchState
* fetch
)
903 assert(fetch
->entry
&& fetch
->request
);
905 mem
= fetch
->entry
->mem_obj
;
908 /* XXX: outgoing numbers are not precise */
909 /* XXX: we must distinguish between 304 hits and misses here */
910 fetch
->sent
.bytes
= fetch
->request
->prefixLen();
911 /* XXX: this is slightly wrong: we don't KNOW that the entire memobject
912 * was fetched. We only know how big it is
914 fetch
->recv
.bytes
= mem
->size();
915 fetch
->sent
.msg
= fetch
->recv
.msg
= 1;
916 fetch
->expires
= fetch
->entry
->expires
;
917 fetch
->resp_time
= squid_curtime
- fetch
->start_time
;
919 debugs(72, 3, "peerDigestFetchFinish: recv " << fetch
->recv
.bytes
<<
920 " bytes in " << (int) fetch
->resp_time
<< " secs");
922 debugs(72, 3, "peerDigestFetchFinish: expires: " <<
923 (long int) fetch
->expires
<< " (" << std::showpos
<<
924 (int) (fetch
->expires
- squid_curtime
) << "), lmt: " <<
925 std::noshowpos
<< (long int) fetch
->entry
->lastmod
<< " (" <<
926 std::showpos
<< (int) (fetch
->entry
->lastmod
- squid_curtime
) <<
932 peerDigestSetCBlock(PeerDigest
* pd
, const char *buf
)
934 StoreDigestCBlock cblock
;
936 const char *host
= pd
->host
.termedBuf();
938 memcpy(&cblock
, buf
, sizeof(cblock
));
939 /* network -> host conversions */
940 cblock
.ver
.current
= ntohs(cblock
.ver
.current
);
941 cblock
.ver
.required
= ntohs(cblock
.ver
.required
);
942 cblock
.capacity
= ntohl(cblock
.capacity
);
943 cblock
.count
= ntohl(cblock
.count
);
944 cblock
.del_count
= ntohl(cblock
.del_count
);
945 cblock
.mask_size
= ntohl(cblock
.mask_size
);
946 debugs(72, 2, "got digest cblock from " << host
<< "; ver: " <<
947 (int) cblock
.ver
.current
<< " (req: " << (int) cblock
.ver
.required
<<
950 debugs(72, 2, "\t size: " <<
951 cblock
.mask_size
<< " bytes, e-cnt: " <<
952 cblock
.count
<< ", e-util: " <<
953 xpercentInt(cblock
.count
, cblock
.capacity
) << "%" );
954 /* check version requirements (both ways) */
956 if (cblock
.ver
.required
> CacheDigestVer
.current
) {
957 debugs(72, DBG_IMPORTANT
, "" << host
<< " digest requires version " <<
958 cblock
.ver
.required
<< "; have: " << CacheDigestVer
.current
);
963 if (cblock
.ver
.current
< CacheDigestVer
.required
) {
964 debugs(72, DBG_IMPORTANT
, "" << host
<< " digest is version " <<
965 cblock
.ver
.current
<< "; we require: " <<
966 CacheDigestVer
.required
);
971 /* check consistency */
972 if (cblock
.ver
.required
> cblock
.ver
.current
||
973 cblock
.mask_size
<= 0 || cblock
.capacity
<= 0 ||
974 cblock
.bits_per_entry
<= 0 || cblock
.hash_func_count
<= 0) {
975 debugs(72, DBG_CRITICAL
, "" << host
<< " digest cblock is corrupted.");
979 /* check consistency further */
980 if ((size_t)cblock
.mask_size
!= CacheDigest::CalcMaskSize(cblock
.capacity
, cblock
.bits_per_entry
)) {
981 debugs(72, DBG_CRITICAL
, host
<< " digest cblock is corrupted " <<
982 "(mask size mismatch: " << cblock
.mask_size
<< " ? " <<
983 CacheDigest::CalcMaskSize(cblock
.capacity
, cblock
.bits_per_entry
)
988 /* there are some things we cannot do yet */
989 if (cblock
.hash_func_count
!= CacheDigestHashFuncCount
) {
990 debugs(72, DBG_CRITICAL
, "" << host
<< " digest: unsupported #hash functions: " <<
991 cblock
.hash_func_count
<< " ? " << CacheDigestHashFuncCount
<< ".");
996 * no cblock bugs below this point
998 /* check size changes */
999 if (pd
->cd
&& cblock
.mask_size
!= (ssize_t
)pd
->cd
->mask_size
) {
1000 debugs(72, 2, host
<< " digest changed size: " << cblock
.mask_size
<<
1001 " -> " << pd
->cd
->mask_size
);
1002 freed_size
= pd
->cd
->mask_size
;
1008 debugs(72, 2, "creating " << host
<< " digest; size: " << cblock
.mask_size
<< " (" <<
1009 std::showpos
<< (int) (cblock
.mask_size
- freed_size
) << ") bytes");
1010 pd
->cd
= new CacheDigest(cblock
.capacity
, cblock
.bits_per_entry
);
1012 if (cblock
.mask_size
>= freed_size
)
1013 statCounter
.cd
.memory
+= (cblock
.mask_size
- freed_size
);
1017 /* these assignments leave us in an inconsistent state until we finish reading the digest */
1018 pd
->cd
->count
= cblock
.count
;
1019 pd
->cd
->del_count
= cblock
.del_count
;
1024 peerDigestUseful(const PeerDigest
* pd
)
1026 /* TODO: we should calculate the prob of a false hit instead of bit util */
1027 const auto bit_util
= pd
->cd
->usedMaskPercent();
1029 if (bit_util
> 65.0) {
1030 debugs(72, DBG_CRITICAL
, "Warning: " << pd
->host
<<
1031 " peer digest has too many bits on (" << bit_util
<< "%).");
1039 saneDiff(time_t diff
)
1041 return abs((int) diff
) > squid_curtime
/ 2 ? 0 : diff
;
1045 peerDigestStatsReport(const PeerDigest
* pd
, StoreEntry
* e
)
1047 #define f2s(flag) (pd->flags.flag ? "yes" : "no")
1048 #define appendTime(tm) storeAppendPrintf(e, "%s\t %10ld\t %+d\t %+d\n", \
1049 ""#tm, (long int)pd->times.tm, \
1050 saneDiff(pd->times.tm - squid_curtime), \
1051 saneDiff(pd->times.tm - pd->times.initialized))
1055 const char *host
= pd
->host
.termedBuf();
1056 storeAppendPrintf(e
, "\npeer digest from %s\n", host
);
1058 cacheDigestGuessStatsReport(&pd
->stats
.guess
, e
, host
);
1060 storeAppendPrintf(e
, "\nevent\t timestamp\t secs from now\t secs from init\n");
1061 appendTime(initialized
);
1063 appendTime(requested
);
1064 appendTime(received
);
1065 appendTime(next_check
);
1067 storeAppendPrintf(e
, "peer digest state:\n");
1068 storeAppendPrintf(e
, "\tneeded: %3s, usable: %3s, requested: %3s\n",
1069 f2s(needed
), f2s(usable
), f2s(requested
));
1070 storeAppendPrintf(e
, "\n\tlast retry delay: %d secs\n",
1071 (int) pd
->times
.retry_delay
);
1072 storeAppendPrintf(e
, "\tlast request response time: %d secs\n",
1073 (int) pd
->times
.req_delay
);
1074 storeAppendPrintf(e
, "\tlast request result: %s\n",
1075 pd
->req_result
? pd
->req_result
: "(none)");
1077 storeAppendPrintf(e
, "\npeer digest traffic:\n");
1078 storeAppendPrintf(e
, "\trequests sent: %d, volume: %d KB\n",
1079 pd
->stats
.sent
.msgs
, (int) pd
->stats
.sent
.kbytes
.kb
);
1080 storeAppendPrintf(e
, "\treplies recv: %d, volume: %d KB\n",
1081 pd
->stats
.recv
.msgs
, (int) pd
->stats
.recv
.kbytes
.kb
);
1083 storeAppendPrintf(e
, "\npeer digest structure:\n");
1086 cacheDigestReport(pd
->cd
, host
, e
);
1088 storeAppendPrintf(e
, "\tno in-memory copy\n");