3 * DEBUG: section 72 Peer Digest Routines
4 * AUTHOR: Alex Rousskov
6 * SQUID Web Proxy Cache http://www.squid-cache.org/
7 * ----------------------------------------------------------
9 * Squid is the result of efforts by numerous individuals from
10 * the Internet community; see the CONTRIBUTORS file for full
11 * details. Many organizations have provided support for Squid's
12 * development; see the SPONSORS file for full details. Squid is
13 * Copyrighted (C) 2001 by the Regents of the University of
14 * California; see the COPYRIGHT file for full details. Squid
15 * incorporates software developed and/or copyrighted by other
16 * sources; see the CREDITS file for full details.
18 * This program is free software; you can redistribute it and/or modify
19 * it under the terms of the GNU General Public License as published by
20 * the Free Software Foundation; either version 2 of the License, or
21 * (at your option) any later version.
23 * This program is distributed in the hope that it will be useful,
24 * but WITHOUT ANY WARRANTY; without even the implied warranty of
25 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
26 * GNU General Public License for more details.
28 * You should have received a copy of the GNU General Public License
29 * along with this program; if not, write to the Free Software
30 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
36 #include "CacheDigest.h"
40 #include "HttpReply.h"
41 #include "HttpRequest.h"
43 #include "MemObject.h"
44 #include "neighbors.h"
45 #include "mime_header.h"
46 #include "PeerDigest.h"
47 #include "SquidTime.h"
49 #include "store_key_md5.h"
50 #include "StoreClient.h"
55 /* local prototypes */
56 static time_t peerDigestIncDelay(const PeerDigest
* pd
);
57 static time_t peerDigestNewDelay(const StoreEntry
* e
);
58 static void peerDigestSetCheck(PeerDigest
* pd
, time_t delay
);
59 static void peerDigestClean(PeerDigest
*);
60 static EVH peerDigestCheck
;
61 static void peerDigestRequest(PeerDigest
* pd
);
62 static STCB peerDigestHandleReply
;
63 static int peerDigestFetchReply(void *, char *, ssize_t
);
64 int peerDigestSwapInHeaders(void *, char *, ssize_t
);
65 int peerDigestSwapInCBlock(void *, char *, ssize_t
);
66 int peerDigestSwapInMask(void *, char *, ssize_t
);
67 static int peerDigestFetchedEnough(DigestFetchState
* fetch
, char *buf
, ssize_t size
, const char *step_name
);
68 static void peerDigestFetchStop(DigestFetchState
* fetch
, char *buf
, const char *reason
);
69 static void peerDigestFetchAbort(DigestFetchState
* fetch
, char *buf
, const char *reason
);
70 static void peerDigestReqFinish(DigestFetchState
* fetch
, char *buf
, int, int, int, const char *reason
, int err
);
71 static void peerDigestPDFinish(DigestFetchState
* fetch
, int pcb_valid
, int err
);
72 static void peerDigestFetchFinish(DigestFetchState
* fetch
, int err
);
73 static void peerDigestFetchSetStats(DigestFetchState
* fetch
);
74 static int peerDigestSetCBlock(PeerDigest
* pd
, const char *buf
);
75 static int peerDigestUseful(const PeerDigest
* pd
);
78 Version
const CacheDigestVer
= { 5, 3 };
80 #define StoreDigestCBlockSize sizeof(StoreDigestCBlock)
82 /* min interval for requesting digests from a given peer */
83 static const time_t PeerDigestReqMinGap
= 5 * 60; /* seconds */
84 /* min interval for requesting digests (cumulative request stream) */
85 static const time_t GlobDigestReqMinGap
= 1 * 60; /* seconds */
89 static time_t pd_last_req_time
= 0; /* last call to Check */
91 /* initialize peer digest */
93 peerDigestInit(PeerDigest
* pd
, peer
* p
)
97 memset(pd
, 0, sizeof(*pd
));
100 * Lock on to the peer here. The corresponding cbdataReferenceDone()
101 * is in peerDigestDestroy().
103 pd
->peer
= cbdataReference(p
);
104 /* if peer disappears, we will know it's name */
107 pd
->times
.initialized
= squid_curtime
;
111 peerDigestClean(PeerDigest
* pd
)
116 cacheDigestDestroy(pd
->cd
);
121 CBDATA_CLASS_INIT(PeerDigest
);
124 PeerDigest::operator new (size_t)
126 CBDATA_INIT_TYPE(PeerDigest
);
127 PeerDigest
*result
= cbdataAlloc(PeerDigest
);
132 PeerDigest::operator delete (void *address
)
134 PeerDigest
*t
= static_cast<PeerDigest
*>(address
);
138 /* allocate new peer digest, call Init, and lock everything */
140 peerDigestCreate(peer
* p
)
146 peerDigestInit(pd
, p
);
148 /* XXX This does not look right, and the same thing again in the caller */
149 return cbdataReference(pd
);
152 /* call Clean and free/unlock everything */
154 peerDigestDestroy(PeerDigest
* pd
)
158 void * peerTmp
= pd
->peer
;
162 * We locked the peer in peerDigestInit(), this is
163 * where we unlock it. If the peer is still valid,
164 * tell it that the digest is gone.
166 if (cbdataReferenceValidDone(peerTmp
, &p
))
167 peerNoteDigestGone((peer
*)p
);
174 /* called by peer to indicate that somebody actually needs this digest */
176 peerDigestNeeded(PeerDigest
* pd
)
179 assert(!pd
->flags
.needed
);
182 pd
->flags
.needed
= 1;
183 pd
->times
.needed
= squid_curtime
;
184 peerDigestSetCheck(pd
, 0); /* check asap */
187 /* currently we do not have a reason to disable without destroying */
189 /* disables peer for good */
191 peerDigestDisable(PeerDigest
* pd
)
193 debugs(72, 2, "peerDigestDisable: peer " << pd
->host
.buf() << " disabled for good");
194 pd
->times
.disabled
= squid_curtime
;
195 pd
->times
.next_check
= -1; /* never */
196 pd
->flags
.usable
= 0;
199 cacheDigestDestroy(pd
->cd
);
203 /* we do not destroy the pd itself to preserve its "history" and stats */
208 /* increment retry delay [after an unsuccessful attempt] */
210 peerDigestIncDelay(const PeerDigest
* pd
)
213 return pd
->times
.retry_delay
> 0 ?
214 2 * pd
->times
.retry_delay
: /* exponential backoff */
215 PeerDigestReqMinGap
; /* minimal delay */
218 /* artificially increases Expires: setting to avoid race conditions
219 * returns the delay till that [increased] expiration time */
221 peerDigestNewDelay(const StoreEntry
* e
)
226 return e
->expires
+ PeerDigestReqMinGap
- squid_curtime
;
228 return PeerDigestReqMinGap
;
231 /* registers next digest verification */
233 peerDigestSetCheck(PeerDigest
* pd
, time_t delay
)
235 eventAdd("peerDigestCheck", peerDigestCheck
, pd
, (double) delay
, 1);
236 pd
->times
.next_check
= squid_curtime
+ delay
;
237 debugs(72, 3, "peerDigestSetCheck: will check peer " << pd
->host
<< " in " << delay
<< " secs");
241 * called when peer is about to disappear or have already disappeared
244 peerDigestNotePeerGone(PeerDigest
* pd
)
246 if (pd
->flags
.requested
) {
247 debugs(72, 2, "peerDigest: peer " << pd
->host
<< " gone, will destroy after fetch.");
248 /* do nothing now, the fetching chain will notice and take action */
250 debugs(72, 2, "peerDigest: peer " << pd
->host
<< " is gone, destroying now.");
251 peerDigestDestroy(pd
);
255 /* callback for eventAdd() (with peer digest locked)
256 * request new digest if our copy is too old or if we lack one;
257 * schedule next check otherwise */
259 peerDigestCheck(void *data
)
261 PeerDigest
*pd
= (PeerDigest
*)data
;
264 assert(!pd
->flags
.requested
);
266 pd
->times
.next_check
= 0; /* unknown */
268 if (!cbdataReferenceValid(pd
->peer
)) {
269 peerDigestNotePeerGone(pd
);
273 debugs(72, 3, "peerDigestCheck: peer " << pd
->peer
->host
<< ":" << pd
->peer
->http_port
);
274 debugs(72, 3, "peerDigestCheck: time: " << squid_curtime
<<
275 ", last received: " << (long int) pd
->times
.received
<< " (" <<
276 std::showpos
<< (int) (squid_curtime
- pd
->times
.received
) << ")");
278 /* decide when we should send the request:
279 * request now unless too close to other requests */
280 req_time
= squid_curtime
;
284 if (req_time
- pd
->times
.received
< PeerDigestReqMinGap
) {
285 debugs(72, 2, "peerDigestCheck: " << pd
->host
<<
286 ", avoiding close peer requests (" <<
287 (int) (req_time
- pd
->times
.received
) << " < " <<
288 (int) PeerDigestReqMinGap
<< " secs).");
290 req_time
= pd
->times
.received
+ PeerDigestReqMinGap
;
294 if (req_time
- pd_last_req_time
< GlobDigestReqMinGap
) {
295 debugs(72, 2, "peerDigestCheck: " << pd
->host
<<
296 ", avoiding close requests (" <<
297 (int) (req_time
- pd_last_req_time
) << " < " <<
298 (int) GlobDigestReqMinGap
<< " secs).");
300 req_time
= pd_last_req_time
+ GlobDigestReqMinGap
;
303 if (req_time
<= squid_curtime
)
304 peerDigestRequest(pd
); /* will set pd->flags.requested */
306 peerDigestSetCheck(pd
, req_time
- squid_curtime
);
309 CBDATA_TYPE(DigestFetchState
);
311 /* ask store for a digest */
313 peerDigestRequest(PeerDigest
* pd
)
316 StoreEntry
*e
, *old_e
;
318 const cache_key
*key
;
320 DigestFetchState
*fetch
= NULL
;
321 StoreIOBuffer tempBuffer
;
323 pd
->req_result
= NULL
;
324 pd
->flags
.requested
= 1;
326 /* compute future request components */
329 url
= xstrdup(p
->digest_url
);
331 url
= internalRemoteUri(p
->host
, p
->http_port
,
332 "/squid-internal-periodic/", StoreDigestFileName
);
334 req
= HttpRequest::CreateFromUrl(url
);
338 key
= storeKeyPublicByRequest(req
);
340 debugs(72, 2, "peerDigestRequest: " << url
<< " key: " << storeKeyText(key
));
342 /* add custom headers */
343 assert(!req
->header
.len
);
345 req
->header
.putStr(HDR_ACCEPT
, StoreDigestMimeStr
);
347 req
->header
.putStr(HDR_ACCEPT
, "text/html");
350 xstrncpy(req
->login
, p
->login
, MAX_LOGIN_SZ
);
352 /* create fetch state structure */
353 CBDATA_INIT_TYPE(DigestFetchState
);
355 fetch
= cbdataAlloc(DigestFetchState
);
357 fetch
->request
= HTTPMSGLOCK(req
);
359 fetch
->pd
= cbdataReference(pd
);
363 fetch
->state
= DIGEST_READ_REPLY
;
365 /* update timestamps */
366 fetch
->start_time
= squid_curtime
;
368 pd
->times
.requested
= squid_curtime
;
370 pd_last_req_time
= squid_curtime
;
372 req
->flags
.cachable
= 1;
374 /* the rest is based on clientProcessExpired() */
375 req
->flags
.refresh
= 1;
377 old_e
= fetch
->old_entry
= Store::Root().get(key
);
380 debugs(72, 5, "peerDigestRequest: found old entry");
383 old_e
->createMemObject(url
, url
);
385 fetch
->old_sc
= storeClientListAdd(old_e
, fetch
);
388 e
= fetch
->entry
= storeCreateEntry(url
, url
, req
->flags
, req
->method
);
389 assert(EBIT_TEST(e
->flags
, KEY_PRIVATE
));
390 fetch
->sc
= storeClientListAdd(e
, fetch
);
391 /* set lastmod to trigger IMS request if possible */
394 e
->lastmod
= old_e
->lastmod
;
396 /* push towards peer cache */
397 debugs(72, 3, "peerDigestRequest: forwarding to fwdStart...");
399 FwdState::fwdStart(Comm::ConnectionPointer(), e
, req
);
401 tempBuffer
.offset
= 0;
403 tempBuffer
.length
= SM_PAGE_SIZE
;
405 tempBuffer
.data
= fetch
->buf
;
407 storeClientCopy(fetch
->sc
, e
, tempBuffer
,
408 peerDigestHandleReply
, fetch
);
411 /* Handle the data copying .. */
414 * This routine handles the copy data and then redirects the
415 * copy to a bunch of subfunctions depending upon the copy state.
416 * It also tracks the buffer offset and "seen", since I'm actually
417 * not interested in rewriting everything to suit my little idea.
420 peerDigestHandleReply(void *data
, StoreIOBuffer receivedData
)
422 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
424 digest_read_state_t prevstate
;
427 assert(fetch
->pd
&& receivedData
.data
);
428 /* The existing code assumes that the received pointer is
429 * where we asked the data to be put
431 assert(fetch
->buf
+ fetch
->bufofs
== receivedData
.data
);
433 /* Update the buffer size */
434 fetch
->bufofs
+= receivedData
.length
;
436 assert(fetch
->bufofs
<= SM_PAGE_SIZE
);
438 /* If we've fetched enough, return */
440 if (peerDigestFetchedEnough(fetch
, fetch
->buf
, fetch
->bufofs
, "peerDigestHandleReply"))
443 /* Call the right function based on the state */
444 /* (Those functions will update the state if needed) */
446 /* Give us a temporary reference. Some of the calls we make may
447 * try to destroy the fetch structure, and we like to know if they
450 fetch
= cbdataReference(fetch
);
452 /* Repeat this loop until we're out of data OR the state changes */
453 /* (So keep going if the state has changed and we still have data */
455 prevstate
= fetch
->state
;
457 switch (fetch
->state
) {
459 case DIGEST_READ_REPLY
:
460 retsize
= peerDigestFetchReply(fetch
, fetch
->buf
, fetch
->bufofs
);
463 case DIGEST_READ_HEADERS
:
464 retsize
= peerDigestSwapInHeaders(fetch
, fetch
->buf
, fetch
->bufofs
);
467 case DIGEST_READ_CBLOCK
:
468 retsize
= peerDigestSwapInCBlock(fetch
, fetch
->buf
, fetch
->bufofs
);
471 case DIGEST_READ_MASK
:
472 retsize
= peerDigestSwapInMask(fetch
, fetch
->buf
, fetch
->bufofs
);
475 case DIGEST_READ_NONE
:
478 case DIGEST_READ_DONE
:
483 fatal("Bad digest transfer mode!\n");
490 * The returned size indicates how much of the buffer was read -
491 * so move the remainder of the buffer to the beginning
492 * and update the bufofs / bufsize
494 newsize
= fetch
->bufofs
- retsize
;
496 memmove(fetch
->buf
, fetch
->buf
+ retsize
, fetch
->bufofs
- newsize
);
498 fetch
->bufofs
= newsize
;
500 } while (cbdataReferenceValid(fetch
) && prevstate
!= fetch
->state
&& fetch
->bufofs
> 0);
502 /* Update the copy offset */
503 fetch
->offset
+= receivedData
.length
;
505 /* Schedule another copy */
506 if (cbdataReferenceValid(fetch
)) {
507 StoreIOBuffer tempBuffer
;
508 tempBuffer
.offset
= fetch
->offset
;
509 tempBuffer
.length
= SM_PAGE_SIZE
- fetch
->bufofs
;
510 tempBuffer
.data
= fetch
->buf
+ fetch
->bufofs
;
511 storeClientCopy(fetch
->sc
, fetch
->entry
, tempBuffer
,
512 peerDigestHandleReply
, fetch
);
516 /* Get rid of our reference, we've finished with it for now */
517 cbdataReferenceDone(fetch
);
520 /* wait for full http headers to be received then parse them */
522 * This routine handles parsing the reply line.
523 * If the reply line indicates an OK, the same data is thrown
524 * to SwapInHeaders(). If the reply line is a NOT_MODIFIED,
525 * we simply stop parsing.
528 peerDigestFetchReply(void *data
, char *buf
, ssize_t size
)
530 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
531 PeerDigest
*pd
= fetch
->pd
;
534 assert(!fetch
->offset
);
536 assert(fetch
->state
== DIGEST_READ_REPLY
);
538 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestFetchReply"))
541 if ((hdr_size
= headersEnd(buf
, size
))) {
543 HttpReply
const *reply
= fetch
->entry
->getReply();
545 assert (reply
->sline
.status
!= 0);
546 status
= reply
->sline
.status
;
547 debugs(72, 3, "peerDigestFetchReply: " << pd
->host
<< " status: " << status
<<
548 ", expires: " << (long int) reply
->expires
<< " (" << std::showpos
<<
549 (int) (reply
->expires
- squid_curtime
) << ")");
551 /* this "if" is based on clientHandleIMSReply() */
553 if (status
== HTTP_NOT_MODIFIED
) {
554 HttpRequest
*r
= NULL
;
555 /* our old entry is fine */
556 assert(fetch
->old_entry
);
558 if (!fetch
->old_entry
->mem_obj
->request
)
559 fetch
->old_entry
->mem_obj
->request
= r
=
560 HTTPMSGLOCK(fetch
->entry
->mem_obj
->request
);
562 assert(fetch
->old_entry
->mem_obj
->request
);
564 HttpReply
*old_rep
= (HttpReply
*) fetch
->old_entry
->getReply();
566 old_rep
->updateOnNotModified(reply
);
568 fetch
->old_entry
->timestampsSet();
570 /* get rid of 304 reply */
571 storeUnregister(fetch
->sc
, fetch
->entry
, fetch
);
573 fetch
->entry
->unlock();
575 fetch
->entry
= fetch
->old_entry
;
577 fetch
->old_entry
= NULL
;
579 /* preserve request -- we need its size to update counters */
580 /* requestUnlink(r); */
581 /* fetch->entry->mem_obj->request = NULL; */
582 } else if (status
== HTTP_OK
) {
583 /* get rid of old entry if any */
585 if (fetch
->old_entry
) {
586 debugs(72, 3, "peerDigestFetchReply: got new digest, releasing old one");
587 storeUnregister(fetch
->old_sc
, fetch
->old_entry
, fetch
);
588 fetch
->old_entry
->releaseRequest();
589 fetch
->old_entry
->unlock();
590 fetch
->old_entry
= NULL
;
593 /* some kind of a bug */
594 peerDigestFetchAbort(fetch
, buf
, httpStatusLineReason(&reply
->sline
));
595 return -1; /* XXX -1 will abort stuff in ReadReply! */
598 /* must have a ready-to-use store entry if we got here */
599 /* can we stay with the old in-memory digest? */
600 if (status
== HTTP_NOT_MODIFIED
&& fetch
->pd
->cd
) {
601 peerDigestFetchStop(fetch
, buf
, "Not modified");
602 fetch
->state
= DIGEST_READ_DONE
;
604 fetch
->state
= DIGEST_READ_HEADERS
;
607 /* need more data, do we have space? */
609 if (size
>= SM_PAGE_SIZE
)
610 peerDigestFetchAbort(fetch
, buf
, "reply header too big");
613 /* We don't want to actually ack that we've handled anything,
614 * otherwise SwapInHeaders() won't get the reply line .. */
618 /* fetch headers from disk, pass on to SwapInCBlock */
620 peerDigestSwapInHeaders(void *data
, char *buf
, ssize_t size
)
622 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
625 assert(fetch
->state
== DIGEST_READ_HEADERS
);
627 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestSwapInHeaders"))
630 assert(!fetch
->offset
);
632 if ((hdr_size
= headersEnd(buf
, size
))) {
633 assert(fetch
->entry
->getReply());
634 assert (fetch
->entry
->getReply()->sline
.status
!= 0);
636 if (fetch
->entry
->getReply()->sline
.status
!= HTTP_OK
) {
637 debugs(72, DBG_IMPORTANT
, "peerDigestSwapInHeaders: " << fetch
->pd
->host
<<
638 " status " << fetch
->entry
->getReply()->sline
.status
<<
641 peerDigestFetchAbort(fetch
, buf
, "internal status error");
645 fetch
->state
= DIGEST_READ_CBLOCK
;
646 return hdr_size
; /* Say how much data we read */
648 /* need more data, do we have space? */
650 if (size
>= SM_PAGE_SIZE
) {
651 peerDigestFetchAbort(fetch
, buf
, "stored header too big");
654 return 0; /* We need to read more to parse .. */
658 fatal("peerDigestSwapInHeaders() - shouldn't get here!\n");
659 return 0; /* keep gcc happy */
663 peerDigestSwapInCBlock(void *data
, char *buf
, ssize_t size
)
665 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
667 assert(fetch
->state
== DIGEST_READ_CBLOCK
);
669 if (peerDigestFetchedEnough(fetch
, buf
, size
, "peerDigestSwapInCBlock"))
672 if (size
>= (ssize_t
)StoreDigestCBlockSize
) {
673 PeerDigest
*pd
= fetch
->pd
;
675 assert(pd
&& fetch
->entry
->getReply());
677 if (peerDigestSetCBlock(pd
, buf
)) {
678 /* XXX: soon we will have variable header size */
679 /* switch to CD buffer and fetch digest guts */
681 assert(pd
->cd
->mask
);
682 fetch
->state
= DIGEST_READ_MASK
;
683 return StoreDigestCBlockSize
;
685 peerDigestFetchAbort(fetch
, buf
, "invalid digest cblock");
689 /* need more data, do we have space? */
691 if (size
>= SM_PAGE_SIZE
) {
692 peerDigestFetchAbort(fetch
, buf
, "digest cblock too big");
695 return 0; /* We need more data */
699 fatal("peerDigestSwapInCBlock(): shouldn't get here!\n");
700 return 0; /* keep gcc happy */
704 peerDigestSwapInMask(void *data
, char *buf
, ssize_t size
)
706 DigestFetchState
*fetch
= (DigestFetchState
*)data
;
710 assert(pd
->cd
&& pd
->cd
->mask
);
713 * NOTENOTENOTENOTENOTE: buf doesn't point to pd->cd->mask anymore!
714 * we need to do the copy ourselves!
716 memcpy(pd
->cd
->mask
+ fetch
->mask_offset
, buf
, size
);
718 /* NOTE! buf points to the middle of pd->cd->mask! */
720 if (peerDigestFetchedEnough(fetch
, NULL
, size
, "peerDigestSwapInMask"))
723 fetch
->mask_offset
+= size
;
725 if (fetch
->mask_offset
>= pd
->cd
->mask_size
) {
726 debugs(72, 2, "peerDigestSwapInMask: Done! Got " <<
727 fetch
->mask_offset
<< ", expected " << pd
->cd
->mask_size
);
728 assert(fetch
->mask_offset
== pd
->cd
->mask_size
);
729 assert(peerDigestFetchedEnough(fetch
, NULL
, 0, "peerDigestSwapInMask"));
730 return -1; /* XXX! */
732 /* We always read everything, so return so */
736 fatal("peerDigestSwapInMask(): shouldn't get here!\n");
737 return 0; /* keep gcc happy */
741 peerDigestFetchedEnough(DigestFetchState
* fetch
, char *buf
, ssize_t size
, const char *step_name
)
743 PeerDigest
*pd
= NULL
;
744 const char *host
= "<unknown>"; /* peer host */
745 const char *reason
= NULL
; /* reason for completion */
746 const char *no_bug
= NULL
; /* successful completion if set */
747 const int pdcb_valid
= cbdataReferenceValid(fetch
->pd
);
748 const int pcb_valid
= cbdataReferenceValid(fetch
->pd
->peer
);
750 /* test possible exiting conditions (the same for most steps!)
751 * cases marked with '?!' should not happen */
754 if (!(pd
= fetch
->pd
))
755 reason
= "peer digest disappeared?!";
757 #if DONT /* WHY NOT? /HNO */
759 else if (!cbdataReferenceValid(pd
))
760 reason
= "invalidated peer digest?!";
765 host
= pd
->host
.termedBuf();
768 debugs(72, 6, step_name
<< ": peer " << host
<< ", offset: " <<
769 fetch
->offset
<< " size: " << size
<< ".");
771 /* continue checking (with pd and host known and valid) */
774 if (!cbdataReferenceValid(pd
->peer
))
775 reason
= "peer disappeared";
777 reason
= "swap failure";
778 else if (!fetch
->entry
)
779 reason
= "swap aborted?!";
780 else if (EBIT_TEST(fetch
->entry
->flags
, ENTRY_ABORTED
))
781 reason
= "swap aborted";
784 /* continue checking (maybe-successful eof case) */
785 if (!reason
&& !size
) {
787 reason
= "null digest?!";
788 else if (fetch
->mask_offset
!= (int)pd
->cd
->mask_size
)
789 reason
= "premature end of digest?!";
790 else if (!peerDigestUseful(pd
))
791 reason
= "useless digest";
793 reason
= no_bug
= "success";
796 /* finish if we have a reason */
798 const int level
= strstr(reason
, "?!") ? 1 : 3;
799 debugs(72, level
, "" << step_name
<< ": peer " << host
<< ", exiting after '" << reason
<< "'");
800 peerDigestReqFinish(fetch
, buf
,
801 1, pdcb_valid
, pcb_valid
, reason
, !no_bug
);
804 assert(pdcb_valid
&& pcb_valid
);
807 return reason
!= NULL
;
810 /* call this when all callback data is valid and fetch must be stopped but
811 * no error has occurred (e.g. we received 304 reply and reuse old digest) */
813 peerDigestFetchStop(DigestFetchState
* fetch
, char *buf
, const char *reason
)
816 debugs(72, 2, "peerDigestFetchStop: peer " << fetch
->pd
->host
<< ", reason: " << reason
);
817 peerDigestReqFinish(fetch
, buf
, 1, 1, 1, reason
, 0);
820 /* call this when all callback data is valid but something bad happened */
822 peerDigestFetchAbort(DigestFetchState
* fetch
, char *buf
, const char *reason
)
825 debugs(72, 2, "peerDigestFetchAbort: peer " << fetch
->pd
->host
<< ", reason: " << reason
);
826 peerDigestReqFinish(fetch
, buf
, 1, 1, 1, reason
, 1);
829 /* complete the digest transfer, update stats, unlock/release everything */
831 peerDigestReqFinish(DigestFetchState
* fetch
, char *buf
,
832 int fcb_valid
, int pdcb_valid
, int pcb_valid
,
833 const char *reason
, int err
)
837 /* must go before peerDigestPDFinish */
840 fetch
->pd
->flags
.requested
= 0;
841 fetch
->pd
->req_result
= reason
;
844 /* schedule next check if peer is still out there */
846 PeerDigest
*pd
= fetch
->pd
;
849 pd
->times
.retry_delay
= peerDigestIncDelay(pd
);
850 peerDigestSetCheck(pd
, pd
->times
.retry_delay
);
852 pd
->times
.retry_delay
= 0;
853 peerDigestSetCheck(pd
, peerDigestNewDelay(fetch
->entry
));
857 /* note: order is significant */
859 peerDigestFetchSetStats(fetch
);
862 peerDigestPDFinish(fetch
, pcb_valid
, err
);
865 peerDigestFetchFinish(fetch
, err
);
868 /* destroys digest if peer disappeared
869 * must be called only when fetch and pd cbdata are valid */
871 peerDigestPDFinish(DigestFetchState
* fetch
, int pcb_valid
, int err
)
873 PeerDigest
*pd
= fetch
->pd
;
874 const char *host
= pd
->host
.termedBuf();
876 pd
->times
.received
= squid_curtime
;
877 pd
->times
.req_delay
= fetch
->resp_time
;
878 kb_incr(&pd
->stats
.sent
.kbytes
, (size_t) fetch
->sent
.bytes
);
879 kb_incr(&pd
->stats
.recv
.kbytes
, (size_t) fetch
->recv
.bytes
);
880 pd
->stats
.sent
.msgs
+= fetch
->sent
.msg
;
881 pd
->stats
.recv
.msgs
+= fetch
->recv
.msg
;
884 debugs(72, DBG_IMPORTANT
, "" << (pcb_valid
? "temporary " : "" ) << "disabling (" << pd
->req_result
<< ") digest from " << host
);
887 cacheDigestDestroy(pd
->cd
);
891 pd
->flags
.usable
= 0;
894 peerDigestNotePeerGone(pd
);
898 pd
->flags
.usable
= 1;
900 /* XXX: ugly condition, but how? */
902 if (fetch
->entry
->store_status
== STORE_OK
)
903 debugs(72, 2, "re-used old digest from " << host
);
905 debugs(72, 2, "received valid digest from " << host
);
908 cbdataReferenceDone(fetch
->pd
);
911 /* free fetch state structures
912 * must be called only when fetch cbdata is valid */
914 peerDigestFetchFinish(DigestFetchState
* fetch
, int err
)
916 assert(fetch
->entry
&& fetch
->request
);
918 if (fetch
->old_entry
) {
919 debugs(72, 3, "peerDigestFetchFinish: deleting old entry");
920 storeUnregister(fetch
->old_sc
, fetch
->old_entry
, fetch
);
921 fetch
->old_entry
->releaseRequest();
922 fetch
->old_entry
->unlock();
923 fetch
->old_entry
= NULL
;
926 /* update global stats */
927 kb_incr(&statCounter
.cd
.kbytes_sent
, (size_t) fetch
->sent
.bytes
);
929 kb_incr(&statCounter
.cd
.kbytes_recv
, (size_t) fetch
->recv
.bytes
);
931 statCounter
.cd
.msgs_sent
+= fetch
->sent
.msg
;
933 statCounter
.cd
.msgs_recv
+= fetch
->recv
.msg
;
935 /* unlock everything */
936 storeUnregister(fetch
->sc
, fetch
->entry
, fetch
);
938 fetch
->entry
->unlock();
940 HTTPMSGUNLOCK(fetch
->request
);
944 assert(fetch
->pd
== NULL
);
949 /* calculate fetch stats after completion */
951 peerDigestFetchSetStats(DigestFetchState
* fetch
)
954 assert(fetch
->entry
&& fetch
->request
);
956 mem
= fetch
->entry
->mem_obj
;
959 /* XXX: outgoing numbers are not precise */
960 /* XXX: we must distinguish between 304 hits and misses here */
961 fetch
->sent
.bytes
= fetch
->request
->prefixLen();
962 /* XXX: this is slightly wrong: we don't KNOW that the entire memobject
963 * was fetched. We only know how big it is
965 fetch
->recv
.bytes
= mem
->size();
966 fetch
->sent
.msg
= fetch
->recv
.msg
= 1;
967 fetch
->expires
= fetch
->entry
->expires
;
968 fetch
->resp_time
= squid_curtime
- fetch
->start_time
;
970 debugs(72, 3, "peerDigestFetchFinish: recv " << fetch
->recv
.bytes
<<
971 " bytes in " << (int) fetch
->resp_time
<< " secs");
973 debugs(72, 3, "peerDigestFetchFinish: expires: " <<
974 (long int) fetch
->expires
<< " (" << std::showpos
<<
975 (int) (fetch
->expires
- squid_curtime
) << "), lmt: " <<
976 std::noshowpos
<< (long int) fetch
->entry
->lastmod
<< " (" <<
977 std::showpos
<< (int) (fetch
->entry
->lastmod
- squid_curtime
) <<
983 peerDigestSetCBlock(PeerDigest
* pd
, const char *buf
)
985 StoreDigestCBlock cblock
;
987 const char *host
= pd
->host
.termedBuf();
989 memcpy(&cblock
, buf
, sizeof(cblock
));
990 /* network -> host conversions */
991 cblock
.ver
.current
= ntohs(cblock
.ver
.current
);
992 cblock
.ver
.required
= ntohs(cblock
.ver
.required
);
993 cblock
.capacity
= ntohl(cblock
.capacity
);
994 cblock
.count
= ntohl(cblock
.count
);
995 cblock
.del_count
= ntohl(cblock
.del_count
);
996 cblock
.mask_size
= ntohl(cblock
.mask_size
);
997 debugs(72, 2, "got digest cblock from " << host
<< "; ver: " <<
998 (int) cblock
.ver
.current
<< " (req: " << (int) cblock
.ver
.required
<<
1001 debugs(72, 2, "\t size: " <<
1002 cblock
.mask_size
<< " bytes, e-cnt: " <<
1003 cblock
.count
<< ", e-util: " <<
1004 xpercentInt(cblock
.count
, cblock
.capacity
) << "%" );
1005 /* check version requirements (both ways) */
1007 if (cblock
.ver
.required
> CacheDigestVer
.current
) {
1008 debugs(72, DBG_IMPORTANT
, "" << host
<< " digest requires version " <<
1009 cblock
.ver
.required
<< "; have: " << CacheDigestVer
.current
);
1014 if (cblock
.ver
.current
< CacheDigestVer
.required
) {
1015 debugs(72, DBG_IMPORTANT
, "" << host
<< " digest is version " <<
1016 cblock
.ver
.current
<< "; we require: " <<
1017 CacheDigestVer
.required
);
1022 /* check consistency */
1023 if (cblock
.ver
.required
> cblock
.ver
.current
||
1024 cblock
.mask_size
<= 0 || cblock
.capacity
<= 0 ||
1025 cblock
.bits_per_entry
<= 0 || cblock
.hash_func_count
<= 0) {
1026 debugs(72, DBG_CRITICAL
, "" << host
<< " digest cblock is corrupted.");
1030 /* check consistency further */
1031 if ((size_t)cblock
.mask_size
!= cacheDigestCalcMaskSize(cblock
.capacity
, cblock
.bits_per_entry
)) {
1032 debugs(72, DBG_CRITICAL
, host
<< " digest cblock is corrupted " <<
1033 "(mask size mismatch: " << cblock
.mask_size
<< " ? " <<
1034 cacheDigestCalcMaskSize(cblock
.capacity
, cblock
.bits_per_entry
)
1039 /* there are some things we cannot do yet */
1040 if (cblock
.hash_func_count
!= CacheDigestHashFuncCount
) {
1041 debugs(72, DBG_CRITICAL
, "" << host
<< " digest: unsupported #hash functions: " <<
1042 cblock
.hash_func_count
<< " ? " << CacheDigestHashFuncCount
<< ".");
1047 * no cblock bugs below this point
1049 /* check size changes */
1050 if (pd
->cd
&& cblock
.mask_size
!= (ssize_t
)pd
->cd
->mask_size
) {
1051 debugs(72, 2, host
<< " digest changed size: " << cblock
.mask_size
<<
1052 " -> " << pd
->cd
->mask_size
);
1053 freed_size
= pd
->cd
->mask_size
;
1054 cacheDigestDestroy(pd
->cd
);
1059 debugs(72, 2, "creating " << host
<< " digest; size: " << cblock
.mask_size
<< " (" <<
1060 std::showpos
<< (int) (cblock
.mask_size
- freed_size
) << ") bytes");
1061 pd
->cd
= cacheDigestCreate(cblock
.capacity
, cblock
.bits_per_entry
);
1063 if (cblock
.mask_size
>= freed_size
)
1064 kb_incr(&statCounter
.cd
.memory
, cblock
.mask_size
- freed_size
);
1068 /* these assignments leave us in an inconsistent state until we finish reading the digest */
1069 pd
->cd
->count
= cblock
.count
;
1070 pd
->cd
->del_count
= cblock
.del_count
;
1075 peerDigestUseful(const PeerDigest
* pd
)
1077 /* TODO: we should calculate the prob of a false hit instead of bit util */
1078 const int bit_util
= cacheDigestBitUtil(pd
->cd
);
1080 if (bit_util
> 65) {
1081 debugs(72, DBG_CRITICAL
, "Warning: " << pd
->host
<<
1082 " peer digest has too many bits on (" << bit_util
<< "%%).");
1091 saneDiff(time_t diff
)
1093 return abs((int) diff
) > squid_curtime
/ 2 ? 0 : diff
;
1097 peerDigestStatsReport(const PeerDigest
* pd
, StoreEntry
* e
)
1099 #define f2s(flag) (pd->flags.flag ? "yes" : "no")
1100 #define appendTime(tm) storeAppendPrintf(e, "%s\t %10ld\t %+d\t %+d\n", \
1101 ""#tm, (long int)pd->times.tm, \
1102 saneDiff(pd->times.tm - squid_curtime), \
1103 saneDiff(pd->times.tm - pd->times.initialized))
1107 const char *host
= pd
->host
.termedBuf();
1108 storeAppendPrintf(e
, "\npeer digest from %s\n", host
);
1110 cacheDigestGuessStatsReport(&pd
->stats
.guess
, e
, host
);
1112 storeAppendPrintf(e
, "\nevent\t timestamp\t secs from now\t secs from init\n");
1113 appendTime(initialized
);
1115 appendTime(requested
);
1116 appendTime(received
);
1117 appendTime(next_check
);
1119 storeAppendPrintf(e
, "peer digest state:\n");
1120 storeAppendPrintf(e
, "\tneeded: %3s, usable: %3s, requested: %3s\n",
1121 f2s(needed
), f2s(usable
), f2s(requested
));
1122 storeAppendPrintf(e
, "\n\tlast retry delay: %d secs\n",
1123 (int) pd
->times
.retry_delay
);
1124 storeAppendPrintf(e
, "\tlast request response time: %d secs\n",
1125 (int) pd
->times
.req_delay
);
1126 storeAppendPrintf(e
, "\tlast request result: %s\n",
1127 pd
->req_result
? pd
->req_result
: "(none)");
1129 storeAppendPrintf(e
, "\npeer digest traffic:\n");
1130 storeAppendPrintf(e
, "\trequests sent: %d, volume: %d KB\n",
1131 pd
->stats
.sent
.msgs
, (int) pd
->stats
.sent
.kbytes
.kb
);
1132 storeAppendPrintf(e
, "\treplies recv: %d, volume: %d KB\n",
1133 pd
->stats
.recv
.msgs
, (int) pd
->stats
.recv
.kbytes
.kb
);
1135 storeAppendPrintf(e
, "\npeer digest structure:\n");
1138 cacheDigestReport(pd
->cd
, host
, e
);
1140 storeAppendPrintf(e
, "\tno in-memory copy\n");