开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > 代理服务器 > 查看源码
peer_digest.c
资源名称：squid-2.2.devel3-src.tar.gz　[点击查看]
上传用户：liugui
上传日期：2007-01-04
资源大小：822k
文件大小：27k
源码类别：

代理服务器

开发平台：

Unix_Linux

peer_digest.c：源码内容
  2. /*
  3.  * $Id: peer_digest.c,v 1.70 1999/01/29 21:28:17 wessels Exp $
  4.  *
  5.  * DEBUG: section 72    Peer Digest Routines
  6.  * AUTHOR: Alex Rousskov
  7.  *
  8.  * SQUID Internet Object Cache  http://squid.nlanr.net/Squid/
  9.  * ----------------------------------------------------------
  10.  *
  11.  *  Squid is the result of efforts by numerous individuals from the
  12.  *  Internet community.  Development is led by Duane Wessels of the
  13.  *  National Laboratory for Applied Network Research and funded by the
  14.  *  National Science Foundation.  Squid is Copyrighted (C) 1998 by
  15.  *  Duane Wessels and the University of California San Diego.  Please
  16.  *  see the COPYRIGHT file for full details.  Squid incorporates
  17.  *  software developed and/or copyrighted by other sources.  Please see
  18.  *  the CREDITS file for full details.
  19.  *
  20.  *  This program is free software; you can redistribute it and/or modify
  21.  *  it under the terms of the GNU General Public License as published by
  22.  *  the Free Software Foundation; either version 2 of the License, or
  23.  *  (at your option) any later version.
  24.  *  
  25.  *  This program is distributed in the hope that it will be useful,
  26.  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  27.  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  28.  *  GNU General Public License for more details.
  29.  *  
  30.  *  You should have received a copy of the GNU General Public License
  31.  *  along with this program; if not, write to the Free Software
  32.  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  33.  *
  34.  */
  36. #include "squid.h"
  38. #if USE_CACHE_DIGESTS
  40. /* local types */
  42. /* local prototypes */
  43. static time_t peerDigestIncDelay(const PeerDigest * pd);
  44. static time_t peerDigestNewDelay(const StoreEntry * e);
  45. static void peerDigestSetCheck(PeerDigest * pd, time_t delay);
  46. static void peerDigestClean(PeerDigest *);
  47. static EVH peerDigestCheck;
  48. static void peerDigestRequest(PeerDigest * pd);
  49. static STCB peerDigestFetchReply;
  50. static STCB peerDigestSwapInHeaders;
  51. static STCB peerDigestSwapInCBlock;
  52. static STCB peerDigestSwapInMask;
  53. static int peerDigestFetchedEnough(DigestFetchState * fetch, char *buf, ssize_t size, const char *step_name);
  54. static void peerDigestFetchStop(DigestFetchState * fetch, char *buf, const char *reason);
  55. static void peerDigestFetchAbort(DigestFetchState * fetch, char *buf, const char *reason);
  56. static void peerDigestReqFinish(DigestFetchState * fetch, char *buf, int, int, int, const char *reason, int err);
  57. static void peerDigestPDFinish(DigestFetchState * fetch, int pcb_valid, int err);
  58. static void peerDigestFetchFinish(DigestFetchState * fetch, int err);
  59. static void peerDigestFetchSetStats(DigestFetchState * fetch);
  60. static int peerDigestSetCBlock(PeerDigest * pd, const char *buf);
  61. static int peerDigestUseful(const PeerDigest * pd);
  64. /* local constants */
  66. #define StoreDigestCBlockSize sizeof(StoreDigestCBlock)
  68. /* min interval for requesting digests from a given peer */
  69. static const time_t PeerDigestReqMinGap = 5 * 60; /* seconds */
  70. /* min interval for requesting digests (cumulative request stream) */
  71. static const time_t GlobDigestReqMinGap = 1 * 60; /* seconds */
  73. /* local vars */
  75. static time_t pd_last_req_time = 0; /* last call to Check */
  77. /* initialize peer digest */
  78. static void
  79. peerDigestInit(PeerDigest * pd, peer * p)
  80. {
  81.     assert(pd && p);
  83.     memset(pd, 0, sizeof(*pd));
  84.     pd->peer = p;
  85.     /* if peer disappears, we will know it's name */
  86.     stringInit(&pd->host, p->host);
  88.     pd->times.initialized = squid_curtime;
  89. }
  91. static void
  92. peerDigestClean(PeerDigest * pd)
  93. {
  94.     assert(pd);
  95.     if (pd->cd)
  96. cacheDigestDestroy(pd->cd);
  97.     stringClean(&pd->host);
  98. }
  100. /* allocate new peer digest, call Init, and lock everything */
  101. PeerDigest *
  102. peerDigestCreate(peer * p)
  103. {
  104.     PeerDigest *pd;
  105.     assert(p);
  107.     pd = memAllocate(MEM_PEER_DIGEST);
  108.     cbdataAdd(pd, memFree, MEM_PEER_DIGEST);
  109.     peerDigestInit(pd, p);
  110.     cbdataLock(pd->peer); /* we will use the peer */
  112.     return pd;
  113. }
  115. /* call Clean and free/unlock everything */
  116. void
  117. peerDigestDestroy(PeerDigest * pd)
  118. {
  119.     peer *p;
  120.     assert(pd);
  122.     p = pd->peer;
  123.     pd->peer = NULL;
  124.     /* inform peer (if any) that we are gone */
  125.     if (cbdataValid(p))
  126. peerNoteDigestGone(p);
  127.     cbdataUnlock(p); /* must unlock, valid or not */
  129.     peerDigestClean(pd);
  130.     cbdataFree(pd);
  131. }
  133. /* called by peer to indicate that somebody actually needs this digest */
  134. void
  135. peerDigestNeeded(PeerDigest * pd)
  136. {
  137.     assert(pd);
  138.     assert(!pd->flags.needed);
  139.     assert(!pd->cd);
  141.     pd->flags.needed = 1;
  142.     pd->times.needed = squid_curtime;
  143.     peerDigestSetCheck(pd, 0); /* check asap */
  144. }
  146. /* currently we do not have a reason to disable without destroying */
  147. #if FUTURE_CODE
  148. /* disables peer for good */
  149. static void
  150. peerDigestDisable(PeerDigest * pd)
  151. {
  152.     debug(72, 2) ("peerDigestDisable: peer %s disabled for goodn",
  153. strBuf(pd->host));
  154.     pd->times.disabled = squid_curtime;
  155.     pd->times.next_check = -1; /* never */
  156.     pd->flags.usable = 0;
  158.     if (pd->cd) {
  159. cacheDigestDestroy(pd->cd);
  160. pd->cd = NULL;
  161.     }
  162.     /* we do not destroy the pd itself to preserve its "history" and stats */
  163. }
  164. #endif
  166. /* increment retry delay [after an unsuccessful attempt] */
  167. static time_t
  168. peerDigestIncDelay(const PeerDigest * pd)
  169. {
  170.     assert(pd);
  171.     return pd->times.retry_delay > 0 ?
  172. 2 * pd->times.retry_delay : /* exponential backoff */
  173. PeerDigestReqMinGap; /* minimal delay */
  174. }
  176. /* artificially increases Expires: setting to avoid race conditions 
  177.  * returns the delay till that [increased] expiration time */
  178. static time_t
  179. peerDigestNewDelay(const StoreEntry * e)
  180. {
  181.     assert(e);
  182.     if (e->expires > 0)
  183. return e->expires + PeerDigestReqMinGap - squid_curtime;
  184.     return PeerDigestReqMinGap;
  185. }
  187. /* registers next digest verification */
  188. static void
  189. peerDigestSetCheck(PeerDigest * pd, time_t delay)
  190. {
  191.     eventAdd("peerDigestCheck", peerDigestCheck, pd, (double) delay, 1);
  192.     pd->times.next_check = squid_curtime + delay;
  193.     debug(72, 3) ("peerDigestSetCheck: will check peer %s in %d secsn",
  194. strBuf(pd->host), delay);
  195. }
  197. /*
  198.  * called when peer is about to disappear or have already disappeared
  199.  */
  200. void
  201. peerDigestNotePeerGone(PeerDigest * pd)
  202. {
  203.     if (pd->flags.requested) {
  204. debug(72, 2) ("peerDigest: peer %s gone, will destroy after fetch.n",
  205.     strBuf(pd->host));
  206. /* do nothing now, the fetching chain will notice and take action */
  207.     } else {
  208. debug(72, 2) ("peerDigest: peer %s is gone, destroying now.n",
  209.     strBuf(pd->host));
  210. peerDigestDestroy(pd);
  211.     }
  212. }
  214. /* callback for eventAdd() (with peer digest locked)
  215.  * request new digest if our copy is too old or if we lack one; 
  216.  * schedule next check otherwise */
  217. static void
  218. peerDigestCheck(void *data)
  219. {
  220.     PeerDigest *pd = data;
  221.     time_t req_time;
  223.     /*
  224.      * you can't assert(cbdataValid(pd)) -- if its not valid this
  225.      * function never gets called
  226.      */
  227.     assert(!pd->flags.requested);
  229.     pd->times.next_check = 0; /* unknown */
  231.     if (!cbdataValid(pd->peer)) {
  232. peerDigestNotePeerGone(pd);
  233. return;
  234.     }
  235.     debug(72, 3) ("peerDigestCheck: peer %s:%dn", pd->peer->host, pd->peer->http_port);
  236.     debug(72, 3) ("peerDigestCheck: time: %d, last received: %d (%+d)n",
  237. squid_curtime, pd->times.received, (squid_curtime - pd->times.received));
  239.     /* decide when we should send the request:
  240.      * request now unless too close to other requests */
  241.     req_time = squid_curtime;
  243.     /* per-peer limit */
  244.     if (req_time - pd->times.received < PeerDigestReqMinGap) {
  245. debug(72, 2) ("peerDigestCheck: %s, avoiding close peer requests (%d < %d secs).n",
  246.     strBuf(pd->host), req_time - pd->times.received,
  247.     PeerDigestReqMinGap);
  248. req_time = pd->times.received + PeerDigestReqMinGap;
  249.     }
  250.     /* global limit */
  251.     if (req_time - pd_last_req_time < GlobDigestReqMinGap) {
  252. debug(72, 2) ("peerDigestCheck: %s, avoiding close requests (%d < %d secs).n",
  253.     strBuf(pd->host), req_time - pd_last_req_time,
  254.     GlobDigestReqMinGap);
  255. req_time = pd_last_req_time + GlobDigestReqMinGap;
  256.     }
  257.     if (req_time <= squid_curtime)
  258. peerDigestRequest(pd); /* will set pd->flags.requested */
  259.     else
  260. peerDigestSetCheck(pd, req_time - squid_curtime);
  261. }
  263. /* ask store for a digest */
  264. static void
  265. peerDigestRequest(PeerDigest * pd)
  266. {
  267.     peer *p = pd->peer;
  268.     StoreEntry *e, *old_e;
  269.     char *url;
  270.     const cache_key *key;
  271.     request_t *req;
  272.     DigestFetchState *fetch = NULL;
  274.     pd->req_result = NULL;
  275.     pd->flags.requested = 1;
  277.     /* compute future request components */
  278.     url = internalRemoteUri(p->host, p->http_port, "/squid-internal-periodic/", StoreDigestFileName);
  279.     key = storeKeyPublic(url, METHOD_GET);
  280.     debug(72, 2) ("peerDigestRequest: %s key: %sn", url, storeKeyText(key));
  281.     req = urlParse(METHOD_GET, url);
  282.     assert(req);
  284.     /* add custom headers */
  285.     assert(!req->header.len);
  286.     httpHeaderPutStr(&req->header, HDR_ACCEPT, StoreDigestMimeStr);
  287.     httpHeaderPutStr(&req->header, HDR_ACCEPT, "text/html");
  289.     /* create fetch state structure */
  290.     fetch = memAllocate(MEM_DIGEST_FETCH_STATE);
  291.     cbdataAdd(fetch, memFree, MEM_DIGEST_FETCH_STATE);
  292.     fetch->request = requestLink(req);
  293.     fetch->pd = pd;
  294.     fetch->offset = 0;
  296.     /* update timestamps */
  297.     fetch->start_time = squid_curtime;
  298.     pd->times.requested = squid_curtime;
  299.     pd_last_req_time = squid_curtime;
  301.     req->flags.cachable = 1;
  302.     /* the rest is based on clientProcessExpired() */
  303.     req->flags.refresh = 1;
  304.     old_e = fetch->old_entry = storeGet(key);
  305.     if (old_e) {
  306. debug(72, 5) ("peerDigestRequest: found old entryn");
  307. storeLockObject(old_e);
  308. storeCreateMemObject(old_e, url, url);
  309. storeClientListAdd(old_e, fetch);
  310.     }
  311.     e = fetch->entry = storeCreateEntry(url, url, req->flags, req->method);
  312.     assert(EBIT_TEST(e->flags, KEY_PRIVATE));
  313.     storeClientListAdd(e, fetch);
  314.     /* set lastmod to trigger IMS request if possible */
  315.     if (old_e)
  316. e->lastmod = old_e->lastmod;
  318.     /* push towards peer cache */
  319.     debug(72, 3) ("peerDigestRequest: forwarding to fwdStart...n");
  320.     fwdStart(-1, e, req, no_addr, no_addr);
  321.     cbdataLock(fetch);
  322.     cbdataLock(fetch->pd);
  323.     storeClientCopy(e, 0, 0, 4096, memAllocate(MEM_4K_BUF),
  324. peerDigestFetchReply, fetch);
  325. }
  327. /* wait for full http headers to be received then parse them */
  328. static void
  329. peerDigestFetchReply(void *data, char *buf, ssize_t size)
  330. {
  331.     DigestFetchState *fetch = data;
  332.     PeerDigest *pd = fetch->pd;
  333.     assert(pd && buf);
  334.     assert(!fetch->offset);
  336.     if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestFetchReply"))
  337. return;
  339.     if (headersEnd(buf, size)) {
  340. http_status status;
  341. HttpReply *reply = fetch->entry->mem_obj->reply;
  342. assert(reply);
  343. httpReplyParse(reply, buf);
  344. status = reply->sline.status;
  345. debug(72, 3) ("peerDigestFetchReply: %s status: %d, expires: %d (%+d)n",
  346.     strBuf(pd->host), status,
  347.     reply->expires, reply->expires - squid_curtime);
  349. /* this "if" is based on clientHandleIMSReply() */
  350. if (status == HTTP_NOT_MODIFIED) {
  351.     request_t *r = NULL;
  352.     /* our old entry is fine */
  353.     assert(fetch->old_entry);
  354.     if (!fetch->old_entry->mem_obj->request)
  355. fetch->old_entry->mem_obj->request = r =
  356.     requestLink(fetch->entry->mem_obj->request);
  357.     assert(fetch->old_entry->mem_obj->request);
  358.     httpReplyUpdateOnNotModified(fetch->old_entry->mem_obj->reply, reply);
  359.     storeTimestampsSet(fetch->old_entry);
  360.     /* get rid of 304 reply */
  361.     storeUnregister(fetch->entry, fetch);
  362.     storeUnlockObject(fetch->entry);
  363.     fetch->entry = fetch->old_entry;
  364.     fetch->old_entry = NULL;
  365.     /* preserve request -- we need its size to update counters */
  366.     /* requestUnlink(r); */
  367.     /* fetch->entry->mem_obj->request = NULL; */
  368. } else if (status == HTTP_OK) {
  369.     /* get rid of old entry if any */
  370.     if (fetch->old_entry) {
  371. debug(72, 3) ("peerDigestFetchReply: got new digest, releasing old onen");
  372. storeUnregister(fetch->old_entry, fetch);
  373. storeReleaseRequest(fetch->old_entry);
  374. storeUnlockObject(fetch->old_entry);
  375. fetch->old_entry = NULL;
  376.     }
  377. } else {
  378.     /* some kind of a bug */
  379.     peerDigestFetchAbort(fetch, buf, httpStatusLineReason(&reply->sline));
  380.     return;
  381. }
  382. /* must have a ready-to-use store entry if we got here */
  383. /* can we stay with the old in-memory digest? */
  384. if (status == HTTP_NOT_MODIFIED && fetch->pd->cd)
  385.     peerDigestFetchStop(fetch, buf, "Not modified");
  386. else
  387.     storeClientCopy(fetch->entry, /* have to swap in */
  388. 0, 0, SM_PAGE_SIZE, buf, peerDigestSwapInHeaders, fetch);
  389.     } else {
  390. /* need more data, do we have space? */
  391. if (size >= SM_PAGE_SIZE)
  392.     peerDigestFetchAbort(fetch, buf, "reply header too big");
  393. else
  394.     storeClientCopy(fetch->entry, size, 0, SM_PAGE_SIZE, buf,
  395. peerDigestFetchReply, fetch);
  396.     }
  397. }
  399. /* fetch headers from disk, pass on to SwapInCBlock */
  400. static void
  401. peerDigestSwapInHeaders(void *data, char *buf, ssize_t size)
  402. {
  403.     DigestFetchState *fetch = data;
  404.     size_t hdr_size;
  406.     if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestSwapInHeaders"))
  407. return;
  409.     assert(!fetch->offset);
  410.     if ((hdr_size = headersEnd(buf, size))) {
  411. assert(fetch->entry->mem_obj->reply);
  412. if (!fetch->entry->mem_obj->reply->sline.status)
  413.     httpReplyParse(fetch->entry->mem_obj->reply, buf);
  414. if (fetch->entry->mem_obj->reply->sline.status != HTTP_OK) {
  415.     debug(72, 1) ("peerDigestSwapInHeaders: %s status %d got cached!n",
  416. strBuf(fetch->pd->host), fetch->entry->mem_obj->reply->sline.status);
  417.     peerDigestFetchAbort(fetch, buf, "internal status error");
  418.     return;
  419. }
  420. fetch->offset += hdr_size;
  421. storeClientCopy(fetch->entry, size, fetch->offset,
  422.     SM_PAGE_SIZE, buf,
  423.     peerDigestSwapInCBlock, fetch);
  424.     } else {
  425. /* need more data, do we have space? */
  426. if (size >= SM_PAGE_SIZE)
  427.     peerDigestFetchAbort(fetch, buf, "stored header too big");
  428. else
  429.     storeClientCopy(fetch->entry, size, 0, SM_PAGE_SIZE, buf,
  430. peerDigestSwapInHeaders, fetch);
  431.     }
  432. }
  434. static void
  435. peerDigestSwapInCBlock(void *data, char *buf, ssize_t size)
  436. {
  437.     DigestFetchState *fetch = data;
  439.     if (peerDigestFetchedEnough(fetch, buf, size, "peerDigestSwapInCBlock"))
  440. return;
  442.     if (size >= StoreDigestCBlockSize) {
  443. PeerDigest *pd = fetch->pd;
  444. HttpReply *rep = fetch->entry->mem_obj->reply;
  445. const int seen = fetch->offset + size;
  447. assert(pd && rep);
  448. if (peerDigestSetCBlock(pd, buf)) {
  449.     /* XXX: soon we will have variable header size */
  450.     fetch->offset += StoreDigestCBlockSize;
  451.     /* switch to CD buffer and fetch digest guts */
  452.     memFree(buf, MEM_4K_BUF);
  453.     buf = NULL;
  454.     assert(pd->cd->mask);
  455.     storeClientCopy(fetch->entry,
  456. seen,
  457. fetch->offset,
  458. pd->cd->mask_size,
  459. pd->cd->mask,
  460. peerDigestSwapInMask, fetch);
  461. } else {
  462.     peerDigestFetchAbort(fetch, buf, "invalid digest cblock");
  463. }
  464.     } else {
  465. /* need more data, do we have space? */
  466. if (size >= SM_PAGE_SIZE)
  467.     peerDigestFetchAbort(fetch, buf, "digest cblock too big");
  468. else
  469.     storeClientCopy(fetch->entry, size, 0, SM_PAGE_SIZE, buf,
  470. peerDigestSwapInCBlock, fetch);
  471.     }
  472. }
  474. static void
  475. peerDigestSwapInMask(void *data, char *buf, ssize_t size)
  476. {
  477.     DigestFetchState *fetch = data;
  478.     PeerDigest *pd;
  480.     /* NOTE! buf points to the middle of pd->cd->mask! */
  481.     if (peerDigestFetchedEnough(fetch, NULL, size, "peerDigestSwapInMask"))
  482. return;
  484.     pd = fetch->pd;
  485.     assert(pd->cd && pd->cd->mask);
  487.     fetch->offset += size;
  488.     fetch->mask_offset += size;
  489.     if (fetch->mask_offset >= pd->cd->mask_size) {
  490. debug(72, 2) ("peerDigestSwapInMask: Done! Got %d, expected %dn",
  491.     fetch->mask_offset, pd->cd->mask_size);
  492. assert(fetch->mask_offset == pd->cd->mask_size);
  493. assert(peerDigestFetchedEnough(fetch, NULL, 0, "peerDigestSwapInMask"));
  494.     } else {
  495. const size_t buf_sz = pd->cd->mask_size - fetch->mask_offset;
  496. assert(buf_sz > 0);
  497. storeClientCopy(fetch->entry,
  498.     fetch->offset,
  499.     fetch->offset,
  500.     buf_sz,
  501.     pd->cd->mask + fetch->mask_offset,
  502.     peerDigestSwapInMask, fetch);
  503.     }
  504. }
  506. static int
  507. peerDigestFetchedEnough(DigestFetchState * fetch, char *buf, ssize_t size, const char *step_name)
  508. {
  509.     PeerDigest *pd = NULL;
  510.     const char *host = "<unknown>"; /* peer host */
  511.     const char *reason = NULL; /* reason for completion */
  512.     const char *no_bug = NULL; /* successful completion if set */
  513.     const int fcb_valid = cbdataValid(fetch);
  514.     const int pdcb_valid = fcb_valid && cbdataValid(fetch->pd);
  515.     const int pcb_valid = pdcb_valid && cbdataValid(fetch->pd->peer);
  517.     /* test possible exiting conditions (the same for most steps!)
  518.      * cases marked with '?!' should not happen */
  520.     if (!reason) {
  521. if (!fcb_valid)
  522.     reason = "fetch aborted?!";
  523. else if (!(pd = fetch->pd))
  524.     reason = "peer digest disappeared?!";
  525. #if DONT
  526. else if (!cbdataValid(pd))
  527.     reason = "invalidated peer digest?!";
  528. #endif
  529. else
  530.     host = strBuf(pd->host);
  531.     }
  532.     debug(72, 6) ("%s: peer %s, offset: %d size: %d.n",
  533. step_name, host, fcb_valid ? fetch->offset : -1, size);
  535.     /* continue checking (with pd and host known and valid) */
  536.     if (!reason) {
  537. if (!cbdataValid(pd->peer))
  538.     reason = "peer disappeared";
  539. else if (size < 0)
  540.     reason = "swap failure";
  541. else if (!fetch->entry)
  542.     reason = "swap aborted?!";
  543. else if (EBIT_TEST(fetch->entry->flags, ENTRY_ABORTED))
  544.     reason = "swap aborted";
  545.     }
  546.     /* continue checking (maybe-successful eof case) */
  547.     if (!reason && !size) {
  548. if (!pd->cd)
  549.     reason = "null digest?!";
  550. else if (fetch->mask_offset != pd->cd->mask_size)
  551.     reason = "premature end of digest?!";
  552. else if (!peerDigestUseful(pd))
  553.     reason = "useless digest";
  554. else
  555.     reason = no_bug = "success";
  556.     }
  557.     /* finish if we have a reason */
  558.     if (reason) {
  559. const int level = strstr(reason, "?!") ? 1 : 3;
  560. debug(72, level) ("%s: peer %s, exiting after '%s'n",
  561.     step_name, host, reason);
  562. peerDigestReqFinish(fetch, buf,
  563.     fcb_valid, pdcb_valid, pcb_valid, reason, !no_bug);
  564.     } else {
  565. /* paranoid check */
  566. assert(fcb_valid && pdcb_valid && pcb_valid);
  567.     }
  568.     return reason != NULL;
  569. }
  571. /* call this when all callback data is valid and fetch must be stopped but
  572.  * no error has occurred (e.g. we received 304 reply and reuse old digest) */
  573. static void
  574. peerDigestFetchStop(DigestFetchState * fetch, char *buf, const char *reason)
  575. {
  576.     assert(reason);
  577.     debug(72, 2) ("peerDigestFetchStop: peer %s, reason: %sn",
  578. strBuf(fetch->pd->host), reason);
  579.     peerDigestReqFinish(fetch, buf, 1, 1, 1, reason, 0);
  580. }
  582. /* call this when all callback data is valid but something bad happened */
  583. static void
  584. peerDigestFetchAbort(DigestFetchState * fetch, char *buf, const char *reason)
  585. {
  586.     assert(reason);
  587.     debug(72, 2) ("peerDigestFetchAbort: peer %s, reason: %sn",
  588. strBuf(fetch->pd->host), reason);
  589.     peerDigestReqFinish(fetch, buf, 1, 1, 1, reason, 1);
  590. }
  592. /* complete the digest transfer, update stats, unlock/release everything */
  593. static void
  594. peerDigestReqFinish(DigestFetchState * fetch, char *buf,
  595.     int fcb_valid, int pdcb_valid, int pcb_valid,
  596.     const char *reason, int err)
  597. {
  598.     assert(reason);
  600.     /* must go before peerDigestPDFinish */
  601.     if (pdcb_valid) {
  602. fetch->pd->flags.requested = 0;
  603. fetch->pd->req_result = reason;
  604.     }
  605.     /* schedule next check if peer is still out there */
  606.     if (pcb_valid) {
  607. PeerDigest *pd = fetch->pd;
  608. if (err) {
  609.     pd->times.retry_delay = peerDigestIncDelay(pd);
  610.     peerDigestSetCheck(pd, pd->times.retry_delay);
  611. } else {
  612.     pd->times.retry_delay = 0;
  613.     peerDigestSetCheck(pd, peerDigestNewDelay(fetch->entry));
  614. }
  615.     }
  616.     /* note: order is significant */
  617.     if (fcb_valid)
  618. peerDigestFetchSetStats(fetch);
  619.     if (pdcb_valid)
  620. peerDigestPDFinish(fetch, pcb_valid, err);
  621.     if (fcb_valid)
  622. peerDigestFetchFinish(fetch, err);
  623.     if (buf)
  624. memFree(buf, MEM_4K_BUF);
  625. }
  628. /* destroys digest if peer disappeared
  629.  * must be called only when fetch and pd cbdata are valid */
  630. static void
  631. peerDigestPDFinish(DigestFetchState * fetch, int pcb_valid, int err)
  632. {
  633.     PeerDigest *pd = fetch->pd;
  634.     const char *host = strBuf(pd->host);
  636.     pd->times.received = squid_curtime;
  637.     pd->times.req_delay = fetch->resp_time;
  638.     kb_incr(&pd->stats.sent.kbytes, (size_t) fetch->sent.bytes);
  639.     kb_incr(&pd->stats.recv.kbytes, (size_t) fetch->recv.bytes);
  640.     pd->stats.sent.msgs += fetch->sent.msg;
  641.     pd->stats.recv.msgs += fetch->recv.msg;
  643.     if (err) {
  644. debug(72, 1) ("%sdisabling (%s) digest from %sn",
  645.     pcb_valid ? "temporary " : "",
  646.     pd->req_result, host);
  648. if (pd->cd) {
  649.     cacheDigestDestroy(pd->cd);
  650.     pd->cd = NULL;
  651. }
  652. pd->flags.usable = 0;
  654. if (!pcb_valid)
  655.     peerDigestNotePeerGone(pd);
  656.     } else {
  657. assert(pcb_valid);
  659. pd->flags.usable = 1;
  661. /* XXX: ugly condition, but how? */
  662. if (fetch->entry->store_status == STORE_OK)
  663.     debug(72, 2) ("re-used old digest from %sn", host);
  664. else
  665.     debug(72, 2) ("received valid digest from %sn", host);
  666.     }
  667.     fetch->pd = NULL;
  668.     cbdataUnlock(pd);
  669. }
  671. /* free fetch state structures
  672.  * must be called only when fetch cbdata is valid */
  673. static void
  674. peerDigestFetchFinish(DigestFetchState * fetch, int err)
  675. {
  676.     assert(fetch->entry && fetch->request);
  678.     if (fetch->old_entry) {
  679. debug(72, 2) ("peerDigestFetchFinish: deleting old entryn");
  680. storeUnregister(fetch->old_entry, fetch);
  681. storeReleaseRequest(fetch->old_entry);
  682. storeUnlockObject(fetch->old_entry);
  683. fetch->old_entry = NULL;
  684.     }
  685.     /* update global stats */
  686.     kb_incr(&Counter.cd.kbytes_sent, (size_t) fetch->sent.bytes);
  687.     kb_incr(&Counter.cd.kbytes_recv, (size_t) fetch->recv.bytes);
  688.     Counter.cd.msgs_sent += fetch->sent.msg;
  689.     Counter.cd.msgs_recv += fetch->recv.msg;
  691.     /* unlock everything */
  692.     storeUnregister(fetch->entry, fetch);
  693.     storeUnlockObject(fetch->entry);
  694.     requestUnlink(fetch->request);
  695.     fetch->entry = NULL;
  696.     fetch->request = NULL;
  697.     assert(fetch->pd == NULL);
  698.     cbdataUnlock(fetch);
  699.     cbdataFree(fetch);
  700. }
  702. /* calculate fetch stats after completion */
  703. static void
  704. peerDigestFetchSetStats(DigestFetchState * fetch)
  705. {
  706.     MemObject *mem;
  707.     assert(fetch->entry && fetch->request);
  709.     mem = fetch->entry->mem_obj;
  710.     assert(mem);
  712.     /* XXX: outgoing numbers are not precise */
  713.     /* XXX: we must distinguish between 304 hits and misses here */
  714.     fetch->sent.bytes = httpRequestPrefixLen(fetch->request);
  715.     fetch->recv.bytes = fetch->entry->store_status == STORE_PENDING ?
  716. mem->inmem_hi : mem->object_sz;
  717.     fetch->sent.msg = fetch->recv.msg = 1;
  718.     fetch->expires = fetch->entry->expires;
  719.     fetch->resp_time = squid_curtime - fetch->start_time;
  721.     debug(72, 3) ("peerDigestFetchFinish: recv %d bytes in %d secsn",
  722. fetch->recv.bytes, fetch->resp_time);
  723.     debug(72, 3) ("peerDigestFetchFinish: expires: %d (%+d), lmt: %d (%+d)n",
  724. fetch->expires, fetch->expires - squid_curtime,
  725. fetch->entry->lastmod, fetch->entry->lastmod - squid_curtime);
  726. }
  729. static int
  730. peerDigestSetCBlock(PeerDigest * pd, const char *buf)
  731. {
  732.     StoreDigestCBlock cblock;
  733.     int freed_size = 0;
  734.     const char *host = strBuf(pd->host);
  736.     xmemcpy(&cblock, buf, sizeof(cblock));
  737.     /* network -> host conversions */
  738.     cblock.ver.current = ntohs(cblock.ver.current);
  739.     cblock.ver.required = ntohs(cblock.ver.required);
  740.     cblock.capacity = ntohl(cblock.capacity);
  741.     cblock.count = ntohl(cblock.count);
  742.     cblock.del_count = ntohl(cblock.del_count);
  743.     cblock.mask_size = ntohl(cblock.mask_size);
  744.     debug(72, 2) ("got digest cblock from %s; ver: %d (req: %d)n",
  745. host, (int) cblock.ver.current, (int) cblock.ver.required);
  746.     debug(72, 2) ("t size: %d bytes, e-cnt: %d, e-util: %d%%n",
  747. cblock.mask_size, cblock.count,
  748. xpercentInt(cblock.count, cblock.capacity));
  749.     /* check version requirements (both ways) */
  750.     if (cblock.ver.required > CacheDigestVer.current) {
  751. debug(72, 1) ("%s digest requires version %d; have: %dn",
  752.     host, cblock.ver.required, CacheDigestVer.current);
  753. return 0;
  754.     }
  755.     if (cblock.ver.current < CacheDigestVer.required) {
  756. debug(72, 1) ("%s digest is version %d; we require: %dn",
  757.     host, cblock.ver.current, CacheDigestVer.required);
  758. return 0;
  759.     }
  760.     /* check consistency */
  761.     if (cblock.ver.required > cblock.ver.current ||
  762. cblock.mask_size <= 0 || cblock.capacity <= 0 ||
  763. cblock.bits_per_entry <= 0 || cblock.hash_func_count <= 0) {
  764. debug(72, 0) ("%s digest cblock is corrupted.n", host);
  765. return 0;
  766.     }
  767.     /* check consistency further */
  768.     if (cblock.mask_size != cacheDigestCalcMaskSize(cblock.capacity, cblock.bits_per_entry)) {
  769. debug(72, 0) ("%s digest cblock is corrupted (mask size mismatch: %d ? %d).n",
  770.     host, cblock.mask_size, cacheDigestCalcMaskSize(cblock.capacity, cblock.bits_per_entry));
  771. return 0;
  772.     }
  773.     /* there are some things we cannot do yet */
  774.     if (cblock.hash_func_count != CacheDigestHashFuncCount) {
  775. debug(72, 0) ("%s digest: unsupported #hash functions: %d ? %d.n",
  776.     host, cblock.hash_func_count, CacheDigestHashFuncCount);
  777. return 0;
  778.     }
  779.     /*
  780.      * no cblock bugs below this point
  781.      */
  782.     /* check size changes */
  783.     if (pd->cd && cblock.mask_size != pd->cd->mask_size) {
  784. debug(72, 2) ("%s digest changed size: %d -> %dn",
  785.     host, cblock.mask_size, pd->cd->mask_size);
  786. freed_size = pd->cd->mask_size;
  787. cacheDigestDestroy(pd->cd);
  788. pd->cd = NULL;
  789.     }
  790.     if (!pd->cd) {
  791. debug(72, 2) ("creating %s digest; size: %d (%+d) bytesn",
  792.     host, cblock.mask_size, (int) (cblock.mask_size - freed_size));
  793. pd->cd = cacheDigestCreate(cblock.capacity, cblock.bits_per_entry);
  794. if (cblock.mask_size >= freed_size)
  795.     kb_incr(&Counter.cd.memory, cblock.mask_size - freed_size);
  796.     }
  797.     assert(pd->cd);
  798.     /* these assignments leave us in an inconsistent state until we finish reading the digest */
  799.     pd->cd->count = cblock.count;
  800.     pd->cd->del_count = cblock.del_count;
  801.     return 1;
  802. }
  804. static int
  805. peerDigestUseful(const PeerDigest * pd)
  806. {
  807.     /* TODO: we should calculate the prob of a false hit instead of bit util */
  808.     const int bit_util = cacheDigestBitUtil(pd->cd);
  809.     if (bit_util > 65) {
  810. debug(72, 0) ("Warning: %s peer digest has too many bits on (%d%%).n",
  811.     strBuf(pd->host), bit_util);
  812. return 0;
  813.     }
  814.     return 1;
  815. }
  817. static int
  818. saneDiff(time_t diff)
  819. {
  820.     return abs(diff) > squid_curtime / 2 ? 0 : diff;
  821. }
  823. void
  824. peerDigestStatsReport(const PeerDigest * pd, StoreEntry * e)
  825. {
  826. #define f2s(flag) (pd->flags.flag ? "yes" : "no")
  827. #define appendTime(tm) storeAppendPrintf(e, "%st %10dt %+dt %+dn", 
  828.     ""#tm, pd->times.tm, 
  829.     saneDiff(pd->times.tm - squid_curtime), 
  830.     saneDiff(pd->times.tm - pd->times.initialized))
  832.     const char *host = pd ? strBuf(pd->host) : NULL;
  833.     assert(pd);
  835.     storeAppendPrintf(e, "npeer digest from %sn", host);
  837.     cacheDigestGuessStatsReport(&pd->stats.guess, e, host);
  839.     storeAppendPrintf(e, "neventt timestampt secs from nowt secs from initn");
  840.     appendTime(initialized);
  841.     appendTime(needed);
  842.     appendTime(requested);
  843.     appendTime(received);
  844.     appendTime(next_check);
  846.     storeAppendPrintf(e, "peer digest state:n");
  847.     storeAppendPrintf(e, "tneeded: %3s, usable: %3s, requested: %3sn",
  848. f2s(needed), f2s(usable), f2s(requested));
  849.     storeAppendPrintf(e, "ntlast retry delay: %d secsn",
  850. pd->times.retry_delay);
  851.     storeAppendPrintf(e, "tlast request response time: %d secsn",
  852. pd->times.req_delay);
  853.     storeAppendPrintf(e, "tlast request result: %sn",
  854. pd->req_result ? pd->req_result : "(none)");
  856.     storeAppendPrintf(e, "npeer digest traffic:n");
  857.     storeAppendPrintf(e, "trequests sent: %d, volume: %d KBn",
  858. pd->stats.sent.msgs, (int) pd->stats.sent.kbytes.kb);
  859.     storeAppendPrintf(e, "treplies recv:  %d, volume: %d KBn",
  860. pd->stats.recv.msgs, (int) pd->stats.recv.kbytes.kb);
  862.     storeAppendPrintf(e, "npeer digest structure:n");
  863.     if (pd->cd)
  864. cacheDigestReport(pd->cd, host, e);
  865.     else
  866. storeAppendPrintf(e, "tno in-memory copyn");
  867. }
  869. #endif