Oops, forgot to attach the patch.
I am attaching here the v6 version
On 07/12/2011 02:41 AM, Alex Rousskov wrote:
On 07/01/2011 02:44 PM, Tsantilas Christos wrote:
- theIdleConns("ICAP Service",NULL),
Please initialize to NULL in case we throw before we reach the
allocation lines below.
Done.
+ theIdleConns = new IdleConnList("ICAP Service",NULL);
+ assert(theIdleConns);
Please remove the assert(). The new operator cannot return NULL.
OK.
+ if ((reused = Comm::IsConnOpen(connection))) {
debugs(93,3, HERE<< "reused pconn "<< connection);
- ++theBusyConns;
}
+ ++theBusyConns;
return connection;
}
Unless I am missing something, this can be simplified and optimized as:
++theBusyConns;
debugs(93,3, HERE<< "got connection: "<< connection);
return connection;
When the connection is printed, we will see whether it is open, right?
yes, we just need the "reused = Comm::IsConnOpen(connection)" line. Now
exist alone, without the "if" test.
The above changes are cosmetic and should not block the commit if the
patch passes your tests, of course.
OK.
If there is not any objection I will commit it to trunk.
Thank you,
Alex.
=== modified file 'src/adaptation/icap/ServiceRep.cc'
--- src/adaptation/icap/ServiceRep.cc 2011-06-17 10:41:10 +0000
+++ src/adaptation/icap/ServiceRep.cc 2011-07-12 08:04:12 +0000
@@ -25,17 +25,19 @@
theBusyConns(0),
theAllWaiters(0),
connOverloadReported(false),
- theIdleConns("ICAP Service",NULL),
+ theIdleConns(NULL),
isSuspended(0), notifying(false),
updateScheduled(false),
wasAnnouncedUp(true), // do not announce an "up" service at startup
isDetached(false)
{
setMaxConnections();
+ theIdleConns = new IdleConnList("ICAP Service", NULL);
}
Adaptation::Icap::ServiceRep::~ServiceRep()
{
+ delete theIdleConns;
Must(!theOptionsFetcher);
delete theOptions;
}
@@ -102,17 +104,13 @@
* In other words, (2) tells us to close one FD for each new one we open due to retriable.
*/
if (retriableXact)
- connection = theIdleConns.pop();
+ connection = theIdleConns->pop();
else
- theIdleConns.closeN(1);
-
- if (!(reused = Comm::IsConnOpen(connection)))
- connection = new Comm::Connection;
- else {
- debugs(93,3, HERE << "reused pconn " << connection);
- ++theBusyConns;
- }
-
+ theIdleConns->closeN(1);
+
+ reused = Comm::IsConnOpen(connection);
+ ++theBusyConns;
+ debugs(93,3, HERE << "got connection: " << connection);
return connection;
}
@@ -124,7 +122,7 @@
if (isReusable && excessConnections() == 0) {
debugs(93, 3, HERE << "pushing pconn" << comment);
commUnsetConnTimeout(conn);
- theIdleConns.push(conn);
+ theIdleConns->push(conn);
} else {
debugs(93, 3, HERE << "closing pconn" << comment);
// comm_close will clear timeout
@@ -144,6 +142,12 @@
fd_table[conn->fd].noteUse(NULL); // pconn re-use but not via PconnPool API
}
+void Adaptation::Icap::ServiceRep::noteConnectionFailed(const char *comment)
+{
+ debugs(93, 3, HERE << "Connection failed: " << comment);
+ --theBusyConns;
+}
+
void Adaptation::Icap::ServiceRep::setMaxConnections()
{
if (cfg().maxConn >= 0)
@@ -171,8 +175,8 @@
if (!available && !connOverloadReported) {
debugs(93, DBG_IMPORTANT, "WARNING: ICAP Max-Connections limit " <<
"exceeded for service " << cfg().uri << ". Open connections now: " <<
- theBusyConns + theIdleConns.count() << ", including " <<
- theIdleConns.count() << " idle persistent connections.");
+ theBusyConns + theIdleConns->count() << ", including " <<
+ theIdleConns->count() << " idle persistent connections.");
connOverloadReported = true;
}
@@ -191,7 +195,7 @@
// Waiters affect the number of needed connections but a needed
// connection may still be excessive from Max-Connections p.o.v.
// so we should not account for waiting transaction needs here.
- const int debt = theBusyConns + theIdleConns.count() - theMaxConnections;
+ const int debt = theBusyConns + theIdleConns->count() - theMaxConnections;
if (debt > 0)
return debt;
else
@@ -378,7 +382,7 @@
debugs(93,8, "ICAPServiceRep::callWhenAvailable");
Must(cb!=NULL);
Must(up());
- Must(!theIdleConns.count()); // or we should not be waiting
+ Must(!theIdleConns->count()); // or we should not be waiting
Client i;
i.service = Pointer(this);
@@ -560,11 +564,10 @@
setMaxConnections();
const int excess = excessConnections();
// if we owe connections and have idle pconns, close the latter
- // XXX: but ... idle pconn to *where*?
- if (excess && theIdleConns.count() > 0) {
- const int n = min(excess, theIdleConns.count());
+ if (excess && theIdleConns->count() > 0) {
+ const int n = min(excess, theIdleConns->count());
debugs(93,5, HERE << "closing " << n << " pconns to relief debt");
- theIdleConns.closeN(n);
+ theIdleConns->closeN(n);
}
scheduleNotification();
=== modified file 'src/adaptation/icap/ServiceRep.h'
--- src/adaptation/icap/ServiceRep.h 2011-06-17 06:04:05 +0000
+++ src/adaptation/icap/ServiceRep.h 2011-07-01 15:05:54 +0000
@@ -113,6 +113,7 @@
Comm::ConnectionPointer getConnection(bool isRetriable, bool &isReused);
void putConnection(const Comm::ConnectionPointer &conn, bool isReusable, const char *comment);
void noteConnectionUse(const Comm::ConnectionPointer &conn);
+ void noteConnectionFailed(const char *comment);
void noteFailure(); // called by transactions to report service failure
@@ -160,7 +161,7 @@
int theMaxConnections; ///< the maximum allowed connections to the service
// TODO: use a better type like the FadingCounter for connOverloadReported
mutable bool connOverloadReported; ///< whether we reported exceeding theMaxConnections
- IdleConnList theIdleConns; ///< idle persistent connection pool
+ IdleConnList *theIdleConns; ///< idle persistent connection pool
FadingCounter theSessionFailures;
const char *isSuspended; // also stores suspension reason for debugging
=== modified file 'src/adaptation/icap/Xaction.cc'
--- src/adaptation/icap/Xaction.cc 2011-06-17 10:41:10 +0000
+++ src/adaptation/icap/Xaction.cc 2011-07-11 14:08:29 +0000
@@ -16,6 +16,7 @@
#include "pconn.h"
#include "HttpRequest.h"
#include "HttpReply.h"
+#include "ipcache.h"
#include "acl/FilledChecklist.h"
#include "icap_log.h"
#include "fde.h"
@@ -85,6 +86,13 @@
Must(static_cast<size_t>(readBuf.potentialSpaceSize()) <= commBufSize);
}
+static void
+icapLookupDnsResults(const ipcache_addrs *ia, const DnsLookupDetails &, void *data)
+{
+ Adaptation::Icap::Xaction *xa = static_cast<Adaptation::Icap::Xaction *>(data);
+ xa->dnsLookupDone(ia);
+}
+
// TODO: obey service-specific, OPTIONS-reported connection limit
void
Adaptation::Icap::Xaction::openConnection()
@@ -101,11 +109,6 @@
if (wasReused && Comm::IsConnOpen(connection)) {
// Set comm Close handler
- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
- closer = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommClosed",
- CloseDialer(this,&Adaptation::Icap::Xaction::noteCommClosed));
- comm_add_close_handler(connection->fd, closer);
-
// fake the connect callback
// TODO: can we sync call Adaptation::Icap::Xaction::noteCommConnected here instead?
typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> Dialer;
@@ -124,23 +127,42 @@
// Attempt to open a new connection...
debugs(93,3, typeName << " opens connection to " << s.cfg().host.termedBuf() << ":" << s.cfg().port);
- // TODO: find the IPs and attempt each one if this is a named service.
- connection->remote = s.cfg().host.termedBuf();
+ // Locate the Service IP(s) to open
+ ipcache_nbgethostbyname(s.cfg().host.termedBuf(), icapLookupDnsResults, this);
+}
+
+void
+Adaptation::Icap::Xaction::dnsLookupDone(const ipcache_addrs *ia)
+{
+ Adaptation::Icap::ServiceRep &s = service();
+
+ if (ia == NULL) {
+ debugs(44, DBG_IMPORTANT, "ICAP: Unknown service host: " << s.cfg().host);
+
+#if WHEN_IPCACHE_NBGETHOSTBYNAME_USES_ASYNC_CALLS
+ dieOnConnectionFailure(); // throws
+#else // take a step back into protected Async call dialing.
+ // fake the connect callback
+ typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> Dialer;
+ CbcPointer<Xaction> self(this);
+ Dialer dialer(self, &Adaptation::Icap::Xaction::noteCommConnected);
+ dialer.params.conn = connection;
+ dialer.params.flag = COMM_ERROR;
+ // fake other parameters by copying from the existing connection
+ connector = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommConnected", dialer);
+ ScheduleCallHere(connector);
+#endif
+ return;
+ }
+
+ assert(ia->cur < ia->count);
+
+ connection = new Comm::Connection;
+ connection->remote = ia->in_addrs[ia->cur];
connection->remote.SetPort(s.cfg().port);
+ getOutgoingAddress(NULL, connection);
// TODO: service bypass status may differ from that of a transaction
- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
- AsyncCall::Pointer timeoutCall = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout",
- TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout));
-
- commSetTimeout(connection->fd, TheConfig.connect_timeout(
- service().cfg().bypass), timeoutCall);
-
- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
- closer = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommClosed",
- CloseDialer(this,&Adaptation::Icap::Xaction::noteCommClosed));
- comm_add_close_handler(connection->fd, closer);
-
typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> ConnectDialer;
connector = JobCallback(93,3, ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected);
Comm::ConnOpener *cs = new Comm::ConnOpener(connection, connector, TheConfig.connect_timeout(service().cfg().bypass));
@@ -206,6 +228,12 @@
if (io.flag != COMM_OK)
dieOnConnectionFailure(); // throws
+ typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
+ AsyncCall::Pointer timeoutCall = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout",
+ TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout));
+ commSetTimeout(io.conn->fd, TheConfig.connect_timeout(
+ service().cfg().bypass), timeoutCall);
+
typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
closer = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommClosed",
CloseDialer(this,&Adaptation::Icap::Xaction::noteCommClosed));
@@ -221,6 +249,7 @@
{
debugs(93, 2, HERE << typeName <<
" failed to connect to " << service().cfg().uri);
+ service().noteConnectionFailed("failure");
detailError(ERR_DETAIL_ICAP_XACT_START);
throw TexcHere("cannot connect to the ICAP service");
}
@@ -268,7 +297,11 @@
theService->cfg().uri << status());
reuseConnection = false;
const bool whileConnecting = connector != NULL;
- closeConnection(); // so that late Comm callbacks do not disturb bypass
+ if (whileConnecting) {
+ assert(!haveConnection());
+ theService->noteConnectionFailed("timedout");
+ } else
+ closeConnection(); // so that late Comm callbacks do not disturb bypass
throw TexcHere(whileConnecting ?
"timed out while connecting to the ICAP service" :
"timed out while talking to the ICAP service");
=== modified file 'src/adaptation/icap/Xaction.h'
--- src/adaptation/icap/Xaction.h 2011-06-04 12:48:45 +0000
+++ src/adaptation/icap/Xaction.h 2011-06-29 10:27:11 +0000
@@ -41,6 +41,7 @@
#include "adaptation/Initiate.h"
#include "AccessLogEntry.h"
#include "HttpReply.h"
+#include "ipcache.h"
class CommConnectCbParams;
@@ -133,6 +134,7 @@
// custom exception handling and end-of-call checks
virtual void callException(const std::exception &e);
virtual void callEnd();
+ void dnsLookupDone(const ipcache_addrs *ia);
protected:
// logging
=== modified file 'src/comm.cc'
--- src/comm.cc 2011-06-04 12:48:45 +0000
+++ src/comm.cc 2011-06-30 16:48:29 +0000
@@ -1156,7 +1156,7 @@
commCallCloseHandlers(fd);
- if (F->pconn.uses)
+ if (F->pconn.uses && F->pconn.pool)
F->pconn.pool->noteUses(F->pconn.uses);
comm_empty_os_read_buffers(fd);
