The attached patch implements aggregation of SNMP responses, similar to how we aggregate some cache manager stats. The code contains changes that allow us to share some of the classes between Cache Manager and SNMP code: * implement the following base classes under the ipc directory/module: - Ipc::Forwarder (ipc/Forwarder{.cc,.h} files) - Ipc::Inquirer (ipc/Inquirer{.cc,.h} files) - Ipc::Request (ipc/Request{.cc,.h} files) - Ipc::Response (ipc/Response{.cc,.h} files) * fix the Mgr::Forwarder, Mgr::Inquirer, Mgr::Request and Mgr::Response classes to be implemented as kid classes of the equivalent Icp::* classes. Also implements for the SNMP the same mechanism used for cache manager: The SNMP requests forwarder to coordinator which collects the statistics from kids and aggregate them. === modified file 'configure.ac' --- configure.ac 2011-01-12 12:54:10 +0000 +++ configure.ac 2011-01-31 13:50:53 +0000 @@ -976,43 +976,43 @@ AC_MSG_NOTICE([Web Cache Coordination V2 Protocol enabled: $enable_wccpv2]) AC_ARG_ENABLE(kill-parent-hack, AS_HELP_STRING([--enable-kill-parent-hack],[Kill parent on shutdown]), [ SQUID_YESNO([$enableval], [unrecognized argument to --enable-kill-parent-hack: $enableval]) ]) SQUID_DEFINE_BOOL(KILL_PARENT_OPT,${enable_kill_parent_hack:=no}, [A dangerous feature which causes Squid to kill its parent process (presumably the RunCache script) upon receipt of SIGTERM or SIGINT. Deprecated, Use with caution.]) AC_MSG_NOTICE([Kill parent on shutdown hack enabled: $enable_kill_parent_hack]) AC_ARG_ENABLE(snmp, AS_HELP_STRING([--disable-snmp],[Disable SNMP monitoring support]), [ SQUID_YESNO([$enableval], [unrecognized argument to --disable-snmp: $enableval]) ]) SQUID_DEFINE_BOOL(SQUID_SNMP,${enable_snmp:=yes}, [Define to enable SNMP monitoring of Squid]) -AM_CONDITIONAL(USE_SNMP, [test "x$enable_snmp" = "xyes"]) +AM_CONDITIONAL(ENABLE_SNMP, [test "x$enable_snmp" = "xyes"]) if test "x$enable_snmp" = "xyes"; then - SNMPLIB='../snmplib/libsnmp.a' + SNMPLIB='../snmplib/libsnmplib.a' makesnmplib=snmplib fi AC_MSG_NOTICE([SNMP support enabled: $enable_snmp]) AC_SUBST(SNMPLIB) AC_SUBST(makesnmplib) AC_ARG_ENABLE(cachemgr-hostname, AS_HELP_STRING([--enable-cachemgr-hostname=hostname], [Make cachemgr.cgi default to this host. If unspecified, uses the name of the build-host]), [ case $enableval in yes) AC_DEFINE(CACHEMGR_HOSTNAME,[getfullhostname()], [If you are upset that the cachemgr.cgi form comes up with the hostname field blank, then define this to getfullhostname()]) AC_MSG_NOTICE([Cachemgr default hostname == host where cachemgr runs]) ;; no) : # Nothing to do.. ;; *) @@ -3353,40 +3353,41 @@ scripts/Makefile \ src/Makefile \ src/base/Makefile \ src/acl/Makefile \ src/fs/Makefile \ src/repl/Makefile \ src/auth/Makefile \ src/adaptation/Makefile \ src/adaptation/icap/Makefile \ src/adaptation/ecap/Makefile \ src/comm/Makefile \ src/esi/Makefile \ src/eui/Makefile \ src/icmp/Makefile \ src/ident/Makefile \ src/ip/Makefile \ src/log/Makefile \ src/ipc/Makefile \ src/ssl/Makefile \ src/mgr/Makefile \ + src/snmp/Makefile \ contrib/Makefile \ snmplib/Makefile \ icons/Makefile \ errors/Makefile \ test-suite/Makefile \ doc/Makefile \ doc/manuals/Makefile \ helpers/Makefile \ helpers/basic_auth/Makefile \ helpers/basic_auth/DB/Makefile \ helpers/basic_auth/fake/Makefile \ helpers/basic_auth/getpwnam/Makefile \ helpers/basic_auth/LDAP/Makefile \ helpers/basic_auth/MSNT/Makefile \ helpers/basic_auth/MSNT-multi-domain/Makefile \ helpers/basic_auth/NCSA/Makefile \ helpers/basic_auth/NIS/Makefile \ helpers/basic_auth/PAM/Makefile \ helpers/basic_auth/POP3/Makefile \ helpers/basic_auth/RADIUS/Makefile \ === modified file 'snmplib/Makefile.am' --- snmplib/Makefile.am 2009-11-12 01:12:50 +0000 +++ snmplib/Makefile.am 2011-01-31 13:49:32 +0000 @@ -1,20 +1,20 @@ ## Process this file with automake to produce Makefile.in ## ## Makefile for libsnmp. ## AM_CFLAGS = $(SQUID_CFLAGS) AM_CXXFLAGS = $(SQUID_CXXFLAGS) -noinst_LIBRARIES = libsnmp.a -libsnmp_a_SOURCES = asn1.c parse.c snmp_vars.c \ +noinst_LIBRARIES = libsnmplib.a +libsnmplib_a_SOURCES = asn1.c parse.c snmp_vars.c \ coexistance.c snmp_api.c snmp_error.c \ mib.c snmp_api_error.c \ snmp_msg.c \ snmp_pdu.c snmplib_debug.c INCLUDES = \ -I$(top_srcdir) \ -I$(top_builddir)/include \ -I$(top_srcdir)/include VERSION = 3.4 DEFS = -DSQUID_SNMP=1 === modified file 'src/Makefile.am' --- src/Makefile.am 2011-01-10 09:43:43 +0000 +++ src/Makefile.am 2011-01-31 11:27:15 +0000 @@ -7,66 +7,69 @@ include $(top_srcdir)/src/Common.am AUTOMAKE_OPTIONS = subdir-objects if USE_DNSSERVER DNSSOURCE = dns.cc DNSSERVER = dnsserver else DNSSOURCE = dns_internal.cc DNSSERVER = endif DNSSOURCE += \ DnsLookupDetails.h \ DnsLookupDetails.cc SBUF_SOURCE= \ base/InstanceId.h \ MemBlob.h \ MemBlob.cc -SNMP_ALL_SOURCE = \ - snmp_core.cc \ - snmp_agent.cc -if USE_SNMP -SNMP_SOURCE = $(SNMP_ALL_SOURCE) -else -SNMP_SOURCE = -endif - LOADABLE_MODULES_SOURCES = \ LoadableModule.h \ LoadableModule.cc \ LoadableModules.h \ LoadableModules.cc SUBDIRS = base comm eui acl fs repl auth ip icmp ident log ipc mgr if ENABLE_SSL SUBDIRS += ssl SSL_LIBS = \ ssl/libsslutil.la \ ssl/libsslsquid.la else SSL_LOCAL_LIBS = endif +SNMP_ALL_SOURCE = \ + snmp_core.h \ + snmp_core.cc \ + snmp_agent.cc +if ENABLE_SNMP +SNMP_SOURCE = $(SNMP_ALL_SOURCE) +SUBDIRS += snmp +SNMP_LIBS = snmp/libsnmp.la +else +SNMP_SOURCE = +endif + if USE_ADAPTATION SUBDIRS += adaptation endif if USE_ESI SUBDIRS += esi ESI_LOCAL_LIBS = \ esi/libesi.la \ $(top_builddir)/lib/libTrie/src/libTrie.a ESI_LIBS = $(ESI_LOCAL_LIBS) \ $(XMLLIB) \ $(EXPATLIB) else ESI_LIBS = endif DELAY_POOL_ALL_SOURCE = \ CommonPool.h \ CompositePoolNode.h \ delay_pools.cc \ @@ -530,40 +533,41 @@ CLEANFILES += $(BUILT_SOURCES) nodist_squid_SOURCES = \ $(DISKIO_GEN_SOURCE) \ $(BUILT_SOURCES) squid_LDADD = \ $(COMMON_LIBS) \ comm/libcomm.la \ eui/libeui.la \ icmp/libicmp.la icmp/libicmp-core.la \ log/liblog.la \ $(XTRA_OBJS) \ $(DISK_LINKOBJS) \ $(REPL_OBJS) \ $(DISK_LIBS) \ $(DISK_OS_LIBS) \ $(CRYPTLIB) \ $(REGEXLIB) \ $(SNMPLIB) \ + $(SNMP_LIBS) \ ${ADAPTATION_LIBS} \ $(ESI_LIBS) \ $(SSL_LIBS) \ $(top_builddir)/lib/libmisccontainers.la \ $(top_builddir)/lib/libmiscencoding.la \ $(top_builddir)/lib/libmiscutil.la \ $(SSLLIB) \ $(EPOLL_LIBS) \ $(MINGW_LIBS) \ $(KRB5LIBS) \ $(COMPAT_LIB) \ $(XTRA_LIBS) squid_DEPENDENCIES = \ $(DISK_LIBS) \ $(DISK_LINKOBJS) \ $(REPL_OBJS) \ $(SNMPLIB) \ ${ADAPTATION_LIBS} \ $(ESI_LOCAL_LIBS) \ $(SSL_LIBS) \ === modified file 'src/ipc/Coordinator.cc' --- src/ipc/Coordinator.cc 2010-10-29 00:12:28 +0000 +++ src/ipc/Coordinator.cc 2011-01-31 11:32:55 +0000 @@ -1,40 +1,42 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #include "config.h" #include "base/TextException.h" #include "CacheManager.h" #include "comm.h" #include "ipc/Coordinator.h" -#include "ipc/FdNotes.h" #include "ipc/SharedListen.h" #include "mgr/Inquirer.h" #include "mgr/Request.h" #include "mgr/Response.h" -#include "mgr/StoreToCommWriter.h" - +#if SQUID_SNMP +#include "snmp/Inquirer.h" +#include "snmp/Request.h" +#include "snmp/Response.h" +#endif CBDATA_NAMESPACED_CLASS_INIT(Ipc, Coordinator); Ipc::Coordinator* Ipc::Coordinator::TheInstance = NULL; Ipc::Coordinator::Coordinator(): Port(coordinatorAddr) { } void Ipc::Coordinator::start() { Port::start(); } Ipc::StrandCoord* Ipc::Coordinator::findStrand(int kidId) { typedef StrandCoords::iterator SI; for (SI iter = strands_.begin(); iter != strands_.end(); ++iter) { if (iter->kidId == kidId) @@ -57,40 +59,52 @@ case mtRegistration: debugs(54, 6, HERE << "Registration request"); handleRegistrationRequest(StrandCoord(message)); break; case mtSharedListenRequest: debugs(54, 6, HERE << "Shared listen request"); handleSharedListenRequest(SharedListenRequest(message)); break; case mtCacheMgrRequest: debugs(54, 6, HERE << "Cache manager request"); handleCacheMgrRequest(Mgr::Request(message)); break; case mtCacheMgrResponse: debugs(54, 6, HERE << "Cache manager response"); handleCacheMgrResponse(Mgr::Response(message)); break; +#if SQUID_SNMP + case mtSnmpRequest: + debugs(54, 6, HERE << "SNMP request"); + handleSnmpRequest(Snmp::Request(message)); + break; + + case mtSnmpResponse: + debugs(54, 6, HERE << "SNMP response"); + handleSnmpResponse(Snmp::Response(message)); + break; +#endif + default: debugs(54, 1, HERE << "Unhandled message type: " << message.type()); break; } } void Ipc::Coordinator::handleRegistrationRequest(const StrandCoord& strand) { registerStrand(strand); // send back an acknowledgement; TODO: remove as not needed? TypedMsgHdr message; strand.pack(message); SendMessage(MakeAddr(strandAddrPfx, strand.kidId), message); } void Ipc::Coordinator::handleSharedListenRequest(const SharedListenRequest& request) { debugs(54, 4, HERE << "kid" << request.requestorId << @@ -106,50 +120,71 @@ SharedListenResponse response(sock, errNo, request.mapId); TypedMsgHdr message; response.pack(message); SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message); } void Ipc::Coordinator::handleCacheMgrRequest(const Mgr::Request& request) { debugs(54, 4, HERE); // Let the strand know that we are now responsible for handling the request Mgr::Response response(request.requestId); TypedMsgHdr message; response.pack(message); SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message); Mgr::Action::Pointer action = CacheManager::GetInstance()->createRequestedAction(request.params); - AsyncJob::Start(new Mgr::Inquirer(action, - Mgr::ImportHttpFdIntoComm(request.fd), request, strands_)); + AsyncJob::Start(new Mgr::Inquirer(action, request, strands_)); } void Ipc::Coordinator::handleCacheMgrResponse(const Mgr::Response& response) { Mgr::Inquirer::HandleRemoteAck(response); } +#if SQUID_SNMP +void +Ipc::Coordinator::handleSnmpRequest(const Snmp::Request& request) +{ + debugs(54, 4, HERE); + + Snmp::Response response(request.requestId); + TypedMsgHdr message; + response.pack(message); + SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message); + + AsyncJob::Start(new Snmp::Inquirer(request, strands_)); +} + +void +Ipc::Coordinator::handleSnmpResponse(const Snmp::Response& response) +{ + debugs(54, 4, HERE); + Snmp::Inquirer::HandleRemoteAck(response); +} +#endif + int Ipc::Coordinator::openListenSocket(const SharedListenRequest& request, int &errNo) { const OpenListenerParams &p = request.params; debugs(54, 6, HERE << "opening listen FD at " << p.addr << " for kid" << request.requestorId); Ip::Address addr = p.addr; // comm_open_listener may modify it enter_suid(); const int sock = comm_open_listener(p.sock_type, p.proto, addr, p.flags, FdNote(p.fdNote)); errNo = (sock >= 0) ? 0 : errno; leave_suid(); // cache positive results if (sock >= 0) listeners[request.params] = sock; === modified file 'src/ipc/Coordinator.h' --- src/ipc/Coordinator.h 2010-10-28 18:52:59 +0000 +++ src/ipc/Coordinator.h 2011-01-31 11:32:55 +0000 @@ -1,69 +1,74 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_COORDINATOR_H #define SQUID_IPC_COORDINATOR_H #include "Array.h" #include "ipc/Messages.h" #include "ipc/Port.h" #include "ipc/SharedListen.h" #include "ipc/StrandCoords.h" #include "mgr/forward.h" - +#if SQUID_SNMP +#include "snmp/forward.h" +#endif #include namespace Ipc { /// Coordinates shared activities of Strands (Squid processes or threads) class Coordinator: public Port { public: static Coordinator* Instance(); public: Coordinator(); void broadcastSignal(int sig) const; ///< send sig to registered strands const StrandCoords &strands() const; ///< currently registered strands protected: virtual void start(); // Port (AsyncJob) API virtual void receive(const TypedMsgHdr& message); // Port API StrandCoord* findStrand(int kidId); ///< registered strand or NULL void registerStrand(const StrandCoord &); ///< adds or updates existing void handleRegistrationRequest(const StrandCoord &); ///< register,ACK /// returns cached socket or calls openListenSocket() void handleSharedListenRequest(const SharedListenRequest& request); void handleCacheMgrRequest(const Mgr::Request& request); void handleCacheMgrResponse(const Mgr::Response& response); - +#if SQUID_SNMP + void handleSnmpRequest(const Snmp::Request& request); + void handleSnmpResponse(const Snmp::Response& response); +#endif /// calls comm_open_listener() int openListenSocket(const SharedListenRequest& request, int &errNo); private: StrandCoords strands_; ///< registered processes and threads typedef std::map Listeners; ///< params:fd map Listeners listeners; ///< cached comm_open_listener() results static Coordinator* TheInstance; ///< the only class instance in existence private: Coordinator(const Coordinator&); // not implemented Coordinator& operator =(const Coordinator&); // not implemented CBDATA_CLASS2(Coordinator); }; } // namespace Ipc === modified file 'src/ipc/FdNotes.cc' --- src/ipc/FdNotes.cc 2010-06-14 21:22:01 +0000 +++ src/ipc/FdNotes.cc 2011-01-31 11:32:55 +0000 @@ -1,31 +1,33 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #include "config.h" #include "Debug.h" #include "ipc/FdNotes.h" const char * Ipc::FdNote(int fdNoteId) { static const char *FdNotes[Ipc::fdnEnd] = { "None", // fdnNone "HTTP Socket", // fdnHttpSocket "HTTPS Socket", // fdnHttpsSocket +#if SQUID_SNMP "Incoming SNMP Socket", // fdnInSnmpSocket "Outgoing SNMP Socket", // fdnOutSnmpSocket +#endif "Incoming ICP Socket", // fdnInIcpSocket "Incoming HTCP Socket" // fdnInHtcpSocket }; if (fdnNone < fdNoteId && fdNoteId < fdnEnd) return FdNotes[fdNoteId]; debugs(54, 1, HERE << "salvaged bug: wrong fd_note ID: " << fdNoteId); return FdNotes[fdnNone]; } === modified file 'src/ipc/FdNotes.h' --- src/ipc/FdNotes.h 2010-07-06 18:58:38 +0000 +++ src/ipc/FdNotes.h 2011-01-31 11:32:55 +0000 @@ -1,27 +1,29 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_FD_NOTES_H #define SQUID_IPC_FD_NOTES_H namespace Ipc { /// We cannot send char* FD notes to other processes. Pass int IDs and convert. /// fd_note() label ID typedef enum { fdnNone, fdnHttpSocket, fdnHttpsSocket, +#if SQUID_SNMP fdnInSnmpSocket, fdnOutSnmpSocket, +#endif fdnInIcpSocket, fdnInHtcpSocket, fdnEnd } FdNoteId; extern const char *FdNote(int fdNodeId); ///< converts FdNoteId into a string } // namespace Ipc; #endif /* SQUID_IPC_FD_NOTES_H */ === renamed file 'src/mgr/Forwarder.cc' => 'src/ipc/Forwarder.cc' --- src/mgr/Forwarder.cc 2010-11-26 00:13:51 +0000 +++ src/ipc/Forwarder.cc 2011-01-31 11:32:55 +0000 @@ -1,240 +1,177 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ #include "config.h" #include "base/AsyncJobCalls.h" #include "base/TextException.h" -#include "CommCalls.h" -#include "errorpage.h" -#include "HttpReply.h" -#include "HttpRequest.h" +#include "ipc/Forwarder.h" #include "ipc/Port.h" -#include "mgr/Forwarder.h" -#include "mgr/Request.h" -#include "SquidTime.h" -#include "Store.h" +#include "ipc/TypedMsgHdr.h" -CBDATA_NAMESPACED_CLASS_INIT(Mgr, Forwarder); +CBDATA_NAMESPACED_CLASS_INIT(Ipc, Forwarder); -Mgr::Forwarder::RequestsMap Mgr::Forwarder::TheRequestsMap; -unsigned int Mgr::Forwarder::LastRequestId = 0; +Ipc::Forwarder::RequestsMap Ipc::Forwarder::TheRequestsMap; +unsigned int Ipc::Forwarder::LastRequestId = 0; -Mgr::Forwarder::Forwarder(int aFd, const ActionParams &aParams, - HttpRequest* aRequest, StoreEntry* anEntry): - AsyncJob("Mgr::Forwarder"), - params(aParams), - request(aRequest), entry(anEntry), fd(aFd), requestId(0), closer(NULL) +Ipc::Forwarder::Forwarder(Request::Pointer aRequest, double aTimeout): + AsyncJob("Ipc::Forwarder"), + request(aRequest), timeout(aTimeout) { - debugs(16, 5, HERE << "FD " << aFd); - Must(fd >= 0); - Must(request != NULL); - Must(entry != NULL); - - HTTPMSGLOCK(request); - entry->lock(); - EBIT_SET(entry->flags, ENTRY_FWD_HDR_WAIT); - - closer = asyncCall(16, 5, "Mgr::Forwarder::noteCommClosed", - CommCbMemFunT(this, &Forwarder::noteCommClosed)); - comm_add_close_handler(fd, closer); + debugs(54, 5, HERE); } -Mgr::Forwarder::~Forwarder() +Ipc::Forwarder::~Forwarder() { - debugs(16, 5, HERE); - Must(request != NULL); - Must(entry != NULL); - Must(requestId == 0); - - HTTPMSGUNLOCK(request); - entry->unregisterAbort(); - entry->unlock(); - close(); + debugs(54, 5, HERE); + Must(request->requestId == 0); + cleanup(); } -/// closes our copy of the client HTTP connection socket +/// perform cleanup actions void -Mgr::Forwarder::close() +Ipc::Forwarder::cleanup() { - if (fd >= 0) { - if (closer != NULL) { - comm_remove_close_handler(fd, closer); - closer = NULL; - } - comm_close(fd); - fd = -1; - } } void -Mgr::Forwarder::start() +Ipc::Forwarder::start() { - debugs(16, 3, HERE); - entry->registerAbort(&Forwarder::Abort, this); + debugs(54, 3, HERE); - typedef NullaryMemFunT Dialer; - AsyncCall::Pointer callback = JobCallback(16, 5, Dialer, this, - Forwarder::handleRemoteAck); - if (++LastRequestId == 0) // don't use zero value as requestId + typedef NullaryMemFunT Dialer; + AsyncCall::Pointer callback = JobCallback(54, 5, Dialer, this, Forwarder::handleRemoteAck); + if (++LastRequestId == 0) // don't use zero value as request->requestId ++LastRequestId; - requestId = LastRequestId; - TheRequestsMap[requestId] = callback; - Request mgrRequest(KidIdentifier, requestId, fd, params); - Ipc::TypedMsgHdr message; + request->requestId = LastRequestId; + TheRequestsMap[request->requestId] = callback; + TypedMsgHdr message; try { - mgrRequest.pack(message); + request->pack(message); } catch (...) { // assume the pack() call failed because the message did not fit // TODO: add a more specific exception? - debugs(16, DBG_CRITICAL, "ERROR: uri " << entry->url() << " exceeds buffer size"); - quitOnError("long URI", errorCon(ERR_INVALID_URL, HTTP_REQUEST_URI_TOO_LARGE, request)); + handleError(); } - Ipc::SendMessage(Ipc::coordinatorAddr, message); - const double timeout = 10; // in seconds - eventAdd("Mgr::Forwarder::requestTimedOut", &Forwarder::RequestTimedOut, + SendMessage(coordinatorAddr, message); + eventAdd("Ipc::Forwarder::requestTimedOut", &Forwarder::RequestTimedOut, this, timeout, 0, false); } void -Mgr::Forwarder::swanSong() +Ipc::Forwarder::swanSong() { - debugs(16, 5, HERE); + debugs(54, 5, HERE); removeTimeoutEvent(); - if (requestId > 0) { - DequeueRequest(requestId); - requestId = 0; + if (request->requestId > 0) { + DequeueRequest(request->requestId); + request->requestId = 0; } - close(); + cleanup(); } bool -Mgr::Forwarder::doneAll() const +Ipc::Forwarder::doneAll() const { - debugs(16, 5, HERE); - return requestId == 0; -} - -/// called when the client socket gets closed by some external force -void -Mgr::Forwarder::noteCommClosed(const CommCloseCbParams &io) -{ - debugs(16, 5, HERE); - Must(fd == io.fd); - fd = -1; - mustStop("commClosed"); + debugs(54, 5, HERE); + return request->requestId == 0; } /// called when Coordinator starts processing the request void -Mgr::Forwarder::handleRemoteAck() +Ipc::Forwarder::handleRemoteAck() { - debugs(16, 3, HERE); - Must(entry != NULL); - - requestId = 0; - EBIT_CLR(entry->flags, ENTRY_FWD_HDR_WAIT); - entry->complete(); + debugs(54, 3, HERE); + request->requestId = 0; } -/// Mgr::Forwarder::requestTimedOut wrapper +/// Ipc::Forwarder::requestTimedOut wrapper void -Mgr::Forwarder::RequestTimedOut(void* param) +Ipc::Forwarder::RequestTimedOut(void* param) { - debugs(16, 3, HERE); + debugs(54, 3, HERE); Must(param != NULL); - Forwarder* mgrFwdr = static_cast(param); + Forwarder* fwdr = static_cast(param); // use async call to enable job call protection that time events lack - CallJobHere(16, 5, mgrFwdr, Mgr::Forwarder, requestTimedOut); + CallJobHere(54, 5, fwdr, Forwarder, requestTimedOut); } /// called when Coordinator fails to start processing the request [in time] void -Mgr::Forwarder::requestTimedOut() +Ipc::Forwarder::requestTimedOut() { - debugs(16, 3, HERE); - quitOnError("timeout", errorCon(ERR_LIFETIME_EXP, HTTP_REQUEST_TIMEOUT, request)); + debugs(54, 3, HERE); + handleTimeout(); } -/// terminate with an error void -Mgr::Forwarder::quitOnError(const char *reason, ErrorState *error) +Ipc::Forwarder::handleError() { - debugs(16, 3, HERE); - Must(reason != NULL); - Must(error != NULL); - Must(entry != NULL); - Must(request != NULL); + mustStop("error"); +} - EBIT_CLR(entry->flags, ENTRY_FWD_HDR_WAIT); - entry->buffer(); - entry->replaceHttpReply(error->BuildHttpReply()); - entry->expires = squid_curtime; - errorStateFree(error); - entry->flush(); - entry->complete(); +void +Ipc::Forwarder::handleTimeout() +{ + mustStop("timeout"); +} - mustStop(reason); +/// terminate with an error +void +Ipc::Forwarder::handleException(const std::exception& e) +{ + debugs(54, 3, HERE << e.what()); + mustStop("exception"); } void -Mgr::Forwarder::callException(const std::exception& e) +Ipc::Forwarder::callException(const std::exception& e) { try { - if (entry != NULL && request != NULL && fd >= 0) - quitOnError("exception", errorCon(ERR_INVALID_RESP, HTTP_INTERNAL_SERVER_ERROR, request)); + handleException(e); } catch (const std::exception& ex) { - debugs(16, DBG_CRITICAL, HERE << ex.what()); + debugs(54, DBG_CRITICAL, HERE << ex.what()); } AsyncJob::callException(e); } /// returns and forgets the right Forwarder callback for the request AsyncCall::Pointer -Mgr::Forwarder::DequeueRequest(unsigned int requestId) +Ipc::Forwarder::DequeueRequest(unsigned int requestId) { - debugs(16, 3, HERE); + debugs(54, 3, HERE); Must(requestId != 0); AsyncCall::Pointer call; RequestsMap::iterator request = TheRequestsMap.find(requestId); if (request != TheRequestsMap.end()) { call = request->second; Must(call != NULL); TheRequestsMap.erase(request); } return call; } /// called when we are no longer waiting for Coordinator to respond void -Mgr::Forwarder::removeTimeoutEvent() +Ipc::Forwarder::removeTimeoutEvent() { if (eventFind(&Forwarder::RequestTimedOut, this)) eventDelete(&Forwarder::RequestTimedOut, this); } void -Mgr::Forwarder::HandleRemoteAck(unsigned int requestId) +Ipc::Forwarder::HandleRemoteAck(unsigned int requestId) { - debugs(16, 3, HERE); + debugs(54, 3, HERE); Must(requestId != 0); AsyncCall::Pointer call = DequeueRequest(requestId); if (call != NULL) ScheduleCallHere(call); } - -/// called when something goes wrong with the Store entry -void -Mgr::Forwarder::Abort(void* param) -{ - Forwarder* mgrFwdr = static_cast(param); - if (mgrFwdr->fd >= 0) - comm_close(mgrFwdr->fd); -} === renamed file 'src/mgr/Forwarder.h' => 'src/ipc/Forwarder.h' --- src/mgr/Forwarder.h 2010-10-29 00:12:28 +0000 +++ src/ipc/Forwarder.h 2011-01-31 11:32:55 +0000 @@ -1,77 +1,68 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ -#ifndef SQUID_MGR_FORWARDER_H -#define SQUID_MGR_FORWARDER_H +#ifndef SQUID_IPC_FORWARDER_H +#define SQUID_IPC_FORWARDER_H #include "base/AsyncJob.h" -#include "mgr/ActionParams.h" +#include "ipc/Request.h" #include -class CommCloseCbParams; -class HttpRequest; -class StoreEntry; -class ErrorState; - -namespace Mgr +namespace Ipc { -/** Forwards a single client cache manager request to Coordinator. - * Waits for an ACK from Coordinator while holding the Store entry. - * Fills the store entry with an error response if forwarding fails. +/** Forwards a worker request to coordinator. + * Waits for an ACK from Coordinator + * Send the data unit with an error response if forwarding fails. */ class Forwarder: public AsyncJob { public: - Forwarder(int aFd, const ActionParams &aParams, HttpRequest* aRequest, - StoreEntry* anEntry); + Forwarder(Request::Pointer aRequest, double aTimeout); virtual ~Forwarder(); /// finds and calls the right Forwarder upon Coordinator's response static void HandleRemoteAck(unsigned int requestId); /* has-to-be-public AsyncJob API */ virtual void callException(const std::exception& e); protected: /* AsyncJob API */ virtual void start(); virtual void swanSong(); virtual bool doneAll() const; + virtual void cleanup(); ///< perform cleanup actions + virtual void handleError(); + virtual void handleTimeout(); + virtual void handleException(const std::exception& e); + virtual void handleRemoteAck(); + private: - void handleRemoteAck(); static void RequestTimedOut(void* param); void requestTimedOut(); - void quitOnError(const char *reason, ErrorState *error); - void noteCommClosed(const CommCloseCbParams& params); void removeTimeoutEvent(); static AsyncCall::Pointer DequeueRequest(unsigned int requestId); - static void Abort(void* param); - void close(); -private: - ActionParams params; ///< action parameters to pass to the other side - HttpRequest* request; ///< HTTP client request for detailing errors - StoreEntry* entry; ///< Store entry expecting the response - int fd; ///< HTTP client connection descriptor - unsigned int requestId; ///< request id - AsyncCall::Pointer closer; ///< comm_close handler for the HTTP connection +protected: + Request::Pointer request; + const double timeout; ///< response wait timeout in seconds - /// maps requestId to Forwarder::handleRemoteAck callback + /// maps request->id to Forwarder::handleRemoteAck callback typedef std::map RequestsMap; static RequestsMap TheRequestsMap; ///< pending Coordinator requests static unsigned int LastRequestId; ///< last requestId used CBDATA_CLASS2(Forwarder); }; -} // namespace Mgr +} // namespace Ipc -#endif /* SQUID_MGR_FORWARDER_H */ +#endif /* SQUID_IPC_FORWARDER_H */ === renamed file 'src/mgr/Inquirer.cc' => 'src/ipc/Inquirer.cc' --- src/mgr/Inquirer.cc 2010-11-27 01:46:22 +0000 +++ src/ipc/Inquirer.cc 2011-01-31 11:32:55 +0000 @@ -1,253 +1,206 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ #include "config.h" #include "base/TextException.h" #include "comm/Write.h" -#include "CommCalls.h" -#include "HttpReply.h" -#include "ipc/Coordinator.h" -#include "mgr/ActionWriter.h" -#include "mgr/Command.h" -#include "mgr/Inquirer.h" -#include "mgr/Request.h" -#include "mgr/Response.h" -#include "SquidTime.h" -#include +#include "ipc/Inquirer.h" +#include "ipc/Port.h" +#include "ipc/TypedMsgHdr.h" +#include "MemBuf.h" #include -CBDATA_NAMESPACED_CLASS_INIT(Mgr, Inquirer); +CBDATA_NAMESPACED_CLASS_INIT(Ipc, Inquirer); -Mgr::Inquirer::RequestsMap Mgr::Inquirer::TheRequestsMap; -unsigned int Mgr::Inquirer::LastRequestId = 0; +Ipc::Inquirer::RequestsMap Ipc::Inquirer::TheRequestsMap; +unsigned int Ipc::Inquirer::LastRequestId = 0; /// compare Ipc::StrandCoord using kidId, for std::sort() below static bool LesserStrandByKidId(const Ipc::StrandCoord &c1, const Ipc::StrandCoord &c2) { return c1.kidId < c2.kidId; } -Mgr::Inquirer::Inquirer(Action::Pointer anAction, int aFd, - const Request &aCause, const Ipc::StrandCoords &coords): - AsyncJob("Mgr::Inquirer"), - aggrAction(anAction), - cause(aCause), - fd(aFd), - strands(coords), pos(strands.begin()), - requestId(0), closer(NULL), timeout(aggrAction->atomic() ? 10 : 100) +Ipc::Inquirer::Inquirer(Request::Pointer aRequest, const StrandCoords& coords, + double aTimeout): + AsyncJob("Ipc::Inquirer"), + request(aRequest), strands(coords), pos(strands.begin()), timeout(aTimeout) { - debugs(16, 5, HERE << "FD " << aFd << " action: " << aggrAction); + debugs(54, 5, HERE); // order by ascending kid IDs; useful for non-aggregatable stats std::sort(strands.begin(), strands.end(), LesserStrandByKidId); - - closer = asyncCall(16, 5, "Mgr::Inquirer::noteCommClosed", - CommCbMemFunT(this, &Inquirer::noteCommClosed)); - comm_add_close_handler(fd, closer); } -Mgr::Inquirer::~Inquirer() +Ipc::Inquirer::~Inquirer() { - debugs(16, 5, HERE); - close(); + debugs(54, 5, HERE); + cleanup(); } -/// closes our copy of the client HTTP connection socket void -Mgr::Inquirer::close() +Ipc::Inquirer::cleanup() { - if (fd >= 0) { - removeCloseHandler(); - comm_close(fd); - fd = -1; - } } void -Mgr::Inquirer::removeCloseHandler() +Ipc::Inquirer::start() { - if (closer != NULL) { - comm_remove_close_handler(fd, closer); - closer = NULL; - } + request->requestId = 0; } void -Mgr::Inquirer::start() -{ - debugs(16, 5, HERE); - Must(fd >= 0); - Must(aggrAction != NULL); - - std::auto_ptr reply(new HttpReply); - reply->setHeaders(HTTP_OK, NULL, "text/plain", -1, squid_curtime, squid_curtime); - reply->header.putStr(HDR_CONNECTION, "close"); // until we chunk response - std::auto_ptr replyBuf(reply->pack()); - writer = asyncCall(16, 5, "Mgr::Inquirer::noteWroteHeader", - CommCbMemFunT(this, &Inquirer::noteWroteHeader)); - Comm::Write(fd, replyBuf.get(), writer); -} - -/// called when we wrote the response header -void -Mgr::Inquirer::noteWroteHeader(const CommIoCbParams& params) -{ - debugs(16, 5, HERE); - writer = NULL; - Must(params.flag == COMM_OK); - Must(params.fd == fd); - Must(params.size != 0); - // start inquiries at the initial pos - inquire(); -} - -void -Mgr::Inquirer::inquire() +Ipc::Inquirer::inquire() { if (pos == strands.end()) { Must(done()); return; } - Must(requestId == 0); - AsyncCall::Pointer callback = asyncCall(16, 5, "Mgr::Inquirer::handleRemoteAck", - HandleAckDialer(this, &Inquirer::handleRemoteAck, Response())); - if (++LastRequestId == 0) // don't use zero value as requestId + Must(request->requestId == 0); + AsyncCall::Pointer callback = asyncCall(54, 5, "Mgr::Inquirer::handleRemoteAck", + HandleAckDialer(this, &Inquirer::handleRemoteAck, NULL)); + if (++LastRequestId == 0) // don't use zero value as request->requestId ++LastRequestId; - requestId = LastRequestId; + request->requestId = LastRequestId; const int kidId = pos->kidId; - debugs(16, 4, HERE << "inquire kid: " << kidId << status()); - TheRequestsMap[requestId] = callback; - Request mgrRequest(KidIdentifier, requestId, fd, - aggrAction->command().params); - Ipc::TypedMsgHdr message; - mgrRequest.pack(message); - Ipc::SendMessage(Ipc::Port::MakeAddr(Ipc::strandAddrPfx, kidId), message); - eventAdd("Mgr::Inquirer::requestTimedOut", &Inquirer::RequestTimedOut, + debugs(54, 4, HERE << "inquire kid: " << kidId << status()); + TheRequestsMap[request->requestId] = callback; + TypedMsgHdr message; + request->pack(message); + SendMessage(Port::MakeAddr(strandAddrPfx, kidId), message); + eventAdd("Ipc::Inquirer::requestTimedOut", &Inquirer::RequestTimedOut, this, timeout, 0, false); } /// called when a strand is done writing its output void -Mgr::Inquirer::handleRemoteAck(const Response& response) +Ipc::Inquirer::handleRemoteAck(Response::Pointer response) { - debugs(16, 4, HERE << status()); - requestId = 0; + debugs(54, 4, HERE << status()); + request->requestId = 0; removeTimeoutEvent(); - if (response.hasAction()) - aggrAction->add(response.getAction()); - Must(!done()); // or we should not be called - ++pos; // advance after a successful inquiry - inquire(); + if (aggregate(response)) { + Must(!done()); // or we should not be called + ++pos; // advance after a successful inquiry + inquire(); + } else { + mustStop("error"); + } } -/// called when the HTTP client or some external force closed our socket void -Mgr::Inquirer::noteCommClosed(const CommCloseCbParams& params) +Ipc::Inquirer::swanSong() { - debugs(16, 5, HERE); - Must(fd < 0 || fd == params.fd); - fd = -1; - mustStop("commClosed"); + debugs(54, 5, HERE); + removeTimeoutEvent(); + if (request->requestId > 0) { + DequeueRequest(request->requestId); + request->requestId = 0; + } + sendResponse(); + cleanup(); +} + +bool +Ipc::Inquirer::doneAll() const +{ + return pos == strands.end(); } void -Mgr::Inquirer::swanSong() +Ipc::Inquirer::handleException(const std::exception& e) { - debugs(16, 5, HERE); - removeTimeoutEvent(); - if (requestId > 0) { - DequeueRequest(requestId); - requestId = 0; - } - if (aggrAction->aggregatable()) { - removeCloseHandler(); - AsyncJob::Start(new ActionWriter(aggrAction, fd)); - fd = -1; // should not close fd because we passed it to ActionWriter - } - close(); + debugs(54, 3, HERE << e.what()); + mustStop("exception"); } -bool -Mgr::Inquirer::doneAll() const +void +Ipc::Inquirer::callException(const std::exception& e) { - return !writer && pos == strands.end(); + debugs(54, 3, HERE); + try { + handleException(e); + } catch (const std::exception& ex) { + debugs(54, DBG_CRITICAL, HERE << ex.what()); + } + AsyncJob::callException(e); } /// returns and forgets the right Inquirer callback for strand request AsyncCall::Pointer -Mgr::Inquirer::DequeueRequest(unsigned int requestId) +Ipc::Inquirer::DequeueRequest(unsigned int requestId) { - debugs(16, 3, HERE << " requestId " << requestId); + debugs(54, 3, HERE << " requestId " << requestId); Must(requestId != 0); AsyncCall::Pointer call; RequestsMap::iterator request = TheRequestsMap.find(requestId); if (request != TheRequestsMap.end()) { call = request->second; Must(call != NULL); TheRequestsMap.erase(request); } return call; } void -Mgr::Inquirer::HandleRemoteAck(const Mgr::Response& response) +Ipc::Inquirer::HandleRemoteAck(const Response& response) { Must(response.requestId != 0); AsyncCall::Pointer call = DequeueRequest(response.requestId); if (call != NULL) { HandleAckDialer* dialer = dynamic_cast(call->getDialer()); Must(dialer); - dialer->arg1 = response; + dialer->arg1 = response.clone(); ScheduleCallHere(call); } } /// called when we are no longer waiting for the strand to respond void -Mgr::Inquirer::removeTimeoutEvent() +Ipc::Inquirer::removeTimeoutEvent() { if (eventFind(&Inquirer::RequestTimedOut, this)) eventDelete(&Inquirer::RequestTimedOut, this); } -/// Mgr::Inquirer::requestTimedOut wrapper +/// Ipc::Inquirer::requestTimedOut wrapper void -Mgr::Inquirer::RequestTimedOut(void* param) +Ipc::Inquirer::RequestTimedOut(void* param) { - debugs(16, 3, HERE); + debugs(54, 3, HERE); Must(param != NULL); Inquirer* cmi = static_cast(param); // use async call to enable job call protection that time events lack - CallJobHere(16, 5, cmi, Mgr::Inquirer, requestTimedOut); + CallJobHere(54, 5, cmi, Inquirer, requestTimedOut); } /// called when the strand failed to respond (or finish responding) in time void -Mgr::Inquirer::requestTimedOut() +Ipc::Inquirer::requestTimedOut() { - debugs(16, 3, HERE); - if (requestId != 0) { - DequeueRequest(requestId); - requestId = 0; + debugs(54, 3, HERE); + if (request->requestId != 0) { + DequeueRequest(request->requestId); + request->requestId = 0; Must(!done()); // or we should not be called ++pos; // advance after a failed inquiry inquire(); } } const char* -Mgr::Inquirer::status() const +Ipc::Inquirer::status() const { static MemBuf buf; buf.reset(); - buf.Printf(" [FD %d, requestId %u]", fd, requestId); + buf.Printf(" [request->requestId %u]", request->requestId); buf.terminate(); return buf.content(); } === renamed file 'src/mgr/Inquirer.h' => 'src/ipc/Inquirer.h' --- src/mgr/Inquirer.h 2010-10-29 00:12:28 +0000 +++ src/ipc/Inquirer.h 2011-01-31 11:32:55 +0000 @@ -1,87 +1,85 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ -#ifndef SQUID_MGR_INQUIRER_H -#define SQUID_MGR_INQUIRER_H +#ifndef SQUID_IPC_INQUIRER_H +#define SQUID_IPC_INQUIRER_H #include "base/AsyncJobCalls.h" #include "base/AsyncJob.h" +#include "ipc/forward.h" +#include "ipc/Request.h" +#include "ipc/Response.h" #include "ipc/StrandCoords.h" -#include "MemBuf.h" -#include "mgr/Action.h" -#include "mgr/Request.h" #include -class CommIoCbParams; -class CommCloseCbParams; -namespace Mgr +namespace Ipc { /// Coordinator's job that sends a cache manage request to each strand, /// aggregating individual strand responses and dumping the result if needed class Inquirer: public AsyncJob { public: - Inquirer(Action::Pointer anAction, int aFd, const Request &aCause, - const Ipc::StrandCoords &coords); + Inquirer(Request::Pointer aRequest, const Ipc::StrandCoords& coords, double aTimeout); virtual ~Inquirer(); /// finds and calls the right Inquirer upon strand's response - static void HandleRemoteAck(const Mgr::Response& response); + static void HandleRemoteAck(const Response& response); + + /* has-to-be-public AsyncJob API */ + virtual void callException(const std::exception& e); protected: /* AsyncJob API */ virtual void start(); virtual void swanSong(); virtual bool doneAll() const; virtual const char *status() const; -private: - typedef UnaryMemFunT HandleAckDialer; + /// inquire the next strand + virtual void inquire(); + /// perform cleanup actions on completion of job + virtual void cleanup(); + /// do specific exception handling + virtual void handleException(const std::exception& e); + /// send response to client + virtual void sendResponse() = 0; + /// perform aggregating of responses and returns true if need to continue + virtual bool aggregate(Response::Pointer aResponse) = 0; - void inquire(); - void noteWroteHeader(const CommIoCbParams& params); - void noteCommClosed(const CommCloseCbParams& params); +private: + typedef UnaryMemFunT HandleAckDialer; - void handleRemoteAck(const Response& response); + void handleRemoteAck(Response::Pointer response); static AsyncCall::Pointer DequeueRequest(unsigned int requestId); static void RequestTimedOut(void* param); void requestTimedOut(); void removeTimeoutEvent(); - void close(); - void removeCloseHandler(); - -private: - Action::Pointer aggrAction; //< action to aggregate - - Request cause; ///< cache manager request received from HTTP client - int fd; ///< HTTP client socket descriptor +protected: + Request::Pointer request; ///< cache manager request received from client Ipc::StrandCoords strands; ///< all strands we want to query, in order Ipc::StrandCoords::const_iterator pos; ///< strand we should query now - unsigned int requestId; ///< ID of our outstanding request to strand - AsyncCall::Pointer writer; ///< comm_write callback - AsyncCall::Pointer closer; ///< comm_close handler const double timeout; ///< number of seconds to wait for strand response - /// maps requestId to Inquirer::handleRemoteAck callback + /// maps request->id to Inquirer::handleRemoteAck callback typedef std::map RequestsMap; static RequestsMap TheRequestsMap; ///< pending strand requests static unsigned int LastRequestId; ///< last requestId used CBDATA_CLASS2(Inquirer); }; -} // namespace Mgr +} // namespace Ipc -#endif /* SQUID_MGR_INQUIRER_H */ +#endif /* SQUID_IPC_INQUIRER_H */ === modified file 'src/ipc/Makefile.am' --- src/ipc/Makefile.am 2010-10-28 18:52:59 +0000 +++ src/ipc/Makefile.am 2011-01-28 13:59:51 +0000 @@ -11,24 +11,29 @@ Kids.cc \ Kids.h \ Messages.h \ StartListening.cc \ StartListening.h \ StrandCoord.cc \ StrandCoord.h \ StrandCoords.h \ SharedListen.cc \ SharedListen.h \ TypedMsgHdr.cc \ TypedMsgHdr.h \ Coordinator.cc \ Coordinator.h \ UdsOp.cc \ UdsOp.h \ Port.cc \ Port.h \ Strand.cc \ Strand.h \ - \ - forward.h + forward.h \ + Forwarder.cc \ + Forwarder.h \ + Inquirer.cc \ + Inquirer.h \ + Request.h \ + Response.h DEFS += -DDEFAULT_PREFIX=\"$(prefix)\" === modified file 'src/ipc/Messages.h' --- src/ipc/Messages.h 2010-10-28 18:52:59 +0000 +++ src/ipc/Messages.h 2011-01-31 11:32:55 +0000 @@ -1,28 +1,28 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_MESSAGES_H #define SQUID_IPC_MESSAGES_H -#include "ipc/forward.h" -#include - -/** Declarations used by varios IPC messages */ +/** Declarations used by various IPC messages */ namespace Ipc { /// message class identifier typedef enum { mtNone = 0, mtRegistration, mtSharedListenRequest, mtSharedListenResponse, - mtCacheMgrRequest, mtCacheMgrResponse + mtCacheMgrRequest, mtCacheMgrResponse, +#if SQUID_SNMP + mtSnmpRequest, mtSnmpResponse +#endif } MessageType; } // namespace Ipc; #endif /* SQUID_IPC_MESSAGES_H */ === renamed file 'src/mgr/Request.h' => 'src/ipc/Request.h' --- src/mgr/Request.h 2010-10-29 00:12:28 +0000 +++ src/ipc/Request.h 2011-01-31 11:32:55 +0000 @@ -1,39 +1,43 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ -#ifndef SQUID_MGR_REQUEST_H -#define SQUID_MGR_REQUEST_H +#ifndef SQUID_IPC_REQUEST_H +#define SQUID_IPC_REQUEST_H -#include "ipc/TypedMsgHdr.h" -#include "mgr/ActionParams.h" +#include "ipc/forward.h" +#include "RefCount.h" -namespace Mgr +namespace Ipc { -/// cache manager request -class Request +/// IPC request +class Request: public RefCountable { public: - Request(int aRequestorId, unsigned int aRequestId, int aFd, - const ActionParams &aParams); + typedef RefCount Pointer; - explicit Request(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() - void pack(Ipc::TypedMsgHdr& msg) const; ///< prepare for sendmsg() +public: + Request(int aRequestorId, unsigned int aRequestId): + requestorId(aRequestorId), requestId(aRequestId) {} + + virtual void pack(TypedMsgHdr& msg) const = 0; ///< prepare for sendmsg() + virtual Pointer clone() const = 0; ///< returns a copy of this + +private: + Request(const Request&); // not implemented + Request& operator= (const Request&); // not implemented public: int requestorId; ///< kidId of the requestor; used for response destination unsigned int requestId; ///< unique for sender; matches request w/ response - int fd; ///< HTTP client connection descriptor - - ActionParams params; ///< action name and parameters }; -} // namespace Mgr +} // namespace Ipc -#endif /* SQUID_MGR_REQUEST_H */ +#endif /* SQUID_IPC_REQUEST_H */ === renamed file 'src/mgr/Response.h' => 'src/ipc/Response.h' --- src/mgr/Response.h 2010-10-28 18:52:59 +0000 +++ src/ipc/Response.h 2011-01-31 11:32:55 +0000 @@ -1,38 +1,49 @@ /* * $Id$ * - * DEBUG: section 16 Cache Manager API + * DEBUG: section 54 Interprocess Communication * */ -#ifndef SQUID_MGR_RESPONSE_H -#define SQUID_MGR_RESPONSE_H +#ifndef SQUID_IPC_RESPONSE_H +#define SQUID_IPC_RESPONSE_H -#include "mgr/Action.h" +#include "ipc/forward.h" +#include "RefCount.h" -namespace Mgr +namespace Ipc { -/// A response to Mgr::Request. -/// May carry strand action data to be aggregated with data from other strands. -class Response +/// A response to Ipc::Request. +class Response: public RefCountable { public: - Response(unsigned int aRequestId = 0, Action::Pointer anAction = NULL); + typedef RefCount Pointer; - explicit Response(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() - void pack(Ipc::TypedMsgHdr& msg) const; ///< prepare for sendmsg() - bool hasAction() const; ///< whether response contain action object - const Action& getAction() const; ///< returns action object +public: + explicit Response(unsigned int aRequestId): + requestId(aRequestId) {} + + virtual void pack(TypedMsgHdr& msg) const = 0; ///< prepare for sendmsg() + virtual Pointer clone() const = 0; ///< returns a copy of this + +private: + Response(const Response&); // not implemented + Response& operator= (const Response&); // not implemented public: unsigned int requestId; ///< ID of request we are responding to - Action::Pointer action; ///< action relating to response }; -extern std::ostream& operator <<(std::ostream &os, const Response &response); +inline +std::ostream& operator << (std::ostream &os, const Response& response) +{ + os << "[response.requestId %u]" << response.requestId << '}'; + return os; +} + -} // namespace Mgr +} // namespace Ipc -#endif /* SQUID_MGR_RESPONSE_H */ +#endif /* SQUID_IPC_RESPONSE_H */ === modified file 'src/ipc/Strand.cc' --- src/ipc/Strand.cc 2010-10-28 18:52:59 +0000 +++ src/ipc/Strand.cc 2011-01-31 11:32:55 +0000 @@ -1,39 +1,43 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #include "config.h" #include "base/TextException.h" #include "ipc/Strand.h" #include "ipc/StrandCoord.h" #include "ipc/Messages.h" #include "ipc/SharedListen.h" #include "ipc/Kids.h" #include "mgr/Request.h" #include "mgr/Response.h" #include "mgr/Forwarder.h" #include "CacheManager.h" - +#if SQUID_SNMP +#include "snmp/Forwarder.h" +#include "snmp/Request.h" +#include "snmp/Response.h" +#endif CBDATA_NAMESPACED_CLASS_INIT(Ipc, Strand); Ipc::Strand::Strand(): Port(MakeAddr(strandAddrPfx, KidIdentifier)), isRegistered(false) { } void Ipc::Strand::start() { Port::start(); registerSelf(); } void Ipc::Strand::registerSelf() { debugs(54, 6, HERE); Must(!isRegistered); @@ -47,57 +51,81 @@ { debugs(54, 6, HERE << message.type()); switch (message.type()) { case mtRegistration: handleRegistrationResponse(StrandCoord(message)); break; case mtSharedListenResponse: SharedListenJoined(SharedListenResponse(message)); break; case mtCacheMgrRequest: handleCacheMgrRequest(Mgr::Request(message)); break; case mtCacheMgrResponse: handleCacheMgrResponse(Mgr::Response(message)); break; +#if SQUID_SNMP + case mtSnmpRequest: + handleSnmpRequest(Snmp::Request(message)); + break; + + case mtSnmpResponse: + handleSnmpResponse(Snmp::Response(message)); + break; +#endif + default: debugs(54, 1, HERE << "Unhandled message type: " << message.type()); break; } } void Ipc::Strand::handleRegistrationResponse(const StrandCoord &strand) { // handle registration response from the coordinator; it could be stale if (strand.kidId == KidIdentifier && strand.pid == getpid()) { debugs(54, 6, "kid" << KidIdentifier << " registered"); clearTimeout(); // we are done } else { // could be an ACK to the registration message of our dead predecessor debugs(54, 6, "kid" << KidIdentifier << " is not yet registered"); // keep listening, with a timeout } } void Ipc::Strand::handleCacheMgrRequest(const Mgr::Request& request) { Mgr::Action::Pointer action = CacheManager::GetInstance()->createRequestedAction(request.params); action->respond(request); } void Ipc::Strand::handleCacheMgrResponse(const Mgr::Response& response) { Mgr::Forwarder::HandleRemoteAck(response.requestId); } +#if SQUID_SNMP +void Ipc::Strand::handleSnmpRequest(const Snmp::Request& request) +{ + debugs(54, 6, HERE); + Snmp::SendResponse(request.requestId, request.pdu); +} + +void Ipc::Strand::handleSnmpResponse(const Snmp::Response& response) +{ + debugs(54, 6, HERE); + Snmp::Forwarder::HandleRemoteAck(response.requestId); +} +#endif + void Ipc::Strand::timedout() { debugs(54, 6, HERE << isRegistered); if (!isRegistered) fatalf("kid%d registration timed out", KidIdentifier); } === modified file 'src/ipc/Strand.h' --- src/ipc/Strand.h 2010-10-28 18:52:59 +0000 +++ src/ipc/Strand.h 2011-01-31 11:32:55 +0000 @@ -1,52 +1,58 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_STRAND_H #define SQUID_IPC_STRAND_H #include "ipc/Port.h" #include "mgr/forward.h" - +#if SQUID_SNMP +#include "snmp/forward.h" +#endif namespace Ipc { class StrandCoord; /// Receives coordination messages on behalf of its process or thread class Strand: public Port { public: Strand(); virtual void start(); // Port (AsyncJob) API protected: virtual void timedout(); // Port (UsdOp) API virtual void receive(const TypedMsgHdr &message); // Port API private: void registerSelf(); /// let Coordinator know this strand exists void handleRegistrationResponse(const StrandCoord &strand); void handleCacheMgrRequest(const Mgr::Request& request); void handleCacheMgrResponse(const Mgr::Response& response); +#if SQUID_SNMP + void handleSnmpRequest(const Snmp::Request& request); + void handleSnmpResponse(const Snmp::Response& response); +#endif private: bool isRegistered; ///< whether Coordinator ACKed registration (unused) private: Strand(const Strand&); // not implemented Strand& operator =(const Strand&); // not implemented CBDATA_CLASS2(Strand); }; } #endif /* SQUID_IPC_STRAND_H */ === modified file 'src/ipc/UdsOp.cc' --- src/ipc/UdsOp.cc 2010-11-27 01:46:22 +0000 +++ src/ipc/UdsOp.cc 2011-01-31 11:32:55 +0000 @@ -115,20 +115,40 @@ { debugs(54, 5, HERE << "FD " << params.fd << " flag " << params.flag << " [" << this << ']'); writing = false; if (params.flag != COMM_OK && retries-- > 0) { sleep(1); // do not spend all tries at once; XXX: use an async timed event instead of blocking here; store the time when we started writing so that we do not sleep if not needed? write(); // XXX: should we close on error so that fd() reopens? } } void Ipc::UdsSender::timedout() { debugs(54, 5, HERE); mustStop("timedout"); } void Ipc::SendMessage(const String& toAddress, const TypedMsgHdr &message) { AsyncJob::Start(new UdsSender(toAddress, message)); } + +int Ipc::ImportFdIntoComm(int fd, int socktype, int protocol, Ipc::FdNoteId noteId) +{ + struct sockaddr_in addr; + socklen_t len = sizeof(addr); + if (getsockname(fd, reinterpret_cast(&addr), &len) == 0) { + Ip::Address ipAddr(addr); + struct addrinfo* addr_info = NULL; + ipAddr.GetAddrInfo(addr_info); + addr_info->ai_socktype = socktype; + addr_info->ai_protocol = protocol; + comm_import_opened(fd, ipAddr, COMM_NONBLOCKING, Ipc::FdNote(noteId), addr_info); + ipAddr.FreeAddrInfo(addr_info); + } else { + debugs(54, DBG_CRITICAL, HERE << "ERROR: FD " << fd << ' ' << xstrerror()); + ::close(fd); + fd = -1; + } + return fd; +} === modified file 'src/ipc/UdsOp.h' --- src/ipc/UdsOp.h 2010-07-06 18:58:38 +0000 +++ src/ipc/UdsOp.h 2011-01-31 11:32:55 +0000 @@ -1,34 +1,35 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_ASYNCUDSOP_H #define SQUID_IPC_ASYNCUDSOP_H #include "SquidString.h" #include "base/AsyncJob.h" #include "ipc/TypedMsgHdr.h" +#include "ipc/FdNotes.h" class CommTimeoutCbParams; class CommIoCbParams; namespace Ipc { /// code shared by unix-domain socket senders (e.g., UdsSender or Coordinator) /// and receivers (e.g. Port or Coordinator) class UdsOp: public AsyncJob { public: UdsOp(const String &pathAddr); virtual ~UdsOp(); public: struct sockaddr_un address; ///< UDS address from path; treat as read-only protected: virtual void timedout() {} ///< called after setTimeout() if timed out @@ -73,25 +74,27 @@ private: void write(); ///< schedule writing void wrote(const CommIoCbParams& params); ///< done writing or error private: TypedMsgHdr message; ///< what to send int retries; ///< how many times to try after a write error int timeout; ///< total time to send the message bool writing; ///< whether Comm started and did not finish writing private: UdsSender(const UdsSender&); // not implemented UdsSender& operator= (const UdsSender&); // not implemented CBDATA_CLASS2(UdsSender); }; void SendMessage(const String& toAddress, const TypedMsgHdr& message); +/// import socket fd from another strand into our Comm state +int ImportFdIntoComm(int fd, int socktype, int protocol, FdNoteId noteId); } #endif /* SQUID_IPC_ASYNCUDSOP_H */ === modified file 'src/ipc/forward.h' --- src/ipc/forward.h 2010-11-21 04:40:05 +0000 +++ src/ipc/forward.h 2011-01-31 11:32:55 +0000 @@ -1,19 +1,23 @@ /* * $Id$ * * DEBUG: section 54 Interprocess Communication * */ #ifndef SQUID_IPC_FORWARD_H #define SQUID_IPC_FORWARD_H namespace Ipc { class TypedMsgHdr; class StrandCoord; +class Forwarder; +class Inquirer; +class Request; +class Response; } // namespace Ipc #endif /* SQUID_IPC_FORWARD_H */ === added file 'src/mgr/Forwarder.cc' --- src/mgr/Forwarder.cc 1970-01-01 00:00:00 +0000 +++ src/mgr/Forwarder.cc 2011-01-17 15:10:36 +0000 @@ -0,0 +1,130 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#include "config.h" +#include "base/AsyncJobCalls.h" +#include "base/TextException.h" +#include "CommCalls.h" +#include "errorpage.h" +#include "HttpReply.h" +#include "HttpRequest.h" +#include "ipc/Port.h" +#include "mgr/Forwarder.h" +#include "mgr/Request.h" +#include "SquidTime.h" +#include "Store.h" + + +CBDATA_NAMESPACED_CLASS_INIT(Mgr, Forwarder); + + +Mgr::Forwarder::Forwarder(int aFd, const ActionParams &aParams, + HttpRequest* aRequest, StoreEntry* anEntry): + Ipc::Forwarder(new Request(KidIdentifier, 0, aFd, aParams), 10), + httpRequest(aRequest), entry(anEntry), fd(aFd) +{ + debugs(16, 5, HERE << "FD " << fd); + Must(fd >= 0); + Must(httpRequest != NULL); + Must(entry != NULL); + + HTTPMSGLOCK(httpRequest); + entry->lock(); + EBIT_SET(entry->flags, ENTRY_FWD_HDR_WAIT); + + closer = asyncCall(16, 5, "Mgr::Forwarder::noteCommClosed", + CommCbMemFunT(this, &Forwarder::noteCommClosed)); + comm_add_close_handler(fd, closer); +} + +Mgr::Forwarder::~Forwarder() +{ + debugs(16, 5, HERE); + Must(httpRequest != NULL); + Must(entry != NULL); + + HTTPMSGUNLOCK(httpRequest); + entry->unregisterAbort(); + entry->unlock(); + cleanup(); +} + +/// closes our copy of the client HTTP connection socket +void +Mgr::Forwarder::cleanup() +{ + if (fd >= 0) { + if (closer != NULL) { + comm_remove_close_handler(fd, closer); + closer = NULL; + } + comm_close(fd); + fd = -1; + } +} + +void +Mgr::Forwarder::handleError() +{ + debugs(16, DBG_CRITICAL, "ERROR: uri " << entry->url() << " exceeds buffer size"); + sendError(errorCon(ERR_INVALID_URL, HTTP_REQUEST_URI_TOO_LARGE, httpRequest)); + mustStop("long URI"); +} + +void +Mgr::Forwarder::handleTimeout() +{ + sendError(errorCon(ERR_LIFETIME_EXP, HTTP_REQUEST_TIMEOUT, httpRequest)); + Ipc::Forwarder::handleTimeout(); +} + +void +Mgr::Forwarder::handleException(const std::exception& e) +{ + if (entry != NULL && httpRequest != NULL && fd >= 0) + sendError(errorCon(ERR_INVALID_RESP, HTTP_INTERNAL_SERVER_ERROR, httpRequest)); + Ipc::Forwarder::handleException(e); +} + +/// called when the client socket gets closed by some external force +void +Mgr::Forwarder::noteCommClosed(const CommCloseCbParams& params) +{ + debugs(16, 5, HERE); + Must(fd == params.fd); + fd = -1; + mustStop("commClosed"); +} + +/// called when Coordinator starts processing the request +void +Mgr::Forwarder::handleRemoteAck() +{ + Ipc::Forwarder::handleRemoteAck(); + + Must(entry != NULL); + EBIT_CLR(entry->flags, ENTRY_FWD_HDR_WAIT); + entry->complete(); +} + +/// send error page +void +Mgr::Forwarder::sendError(ErrorState *error) +{ + debugs(16, 3, HERE); + Must(error != NULL); + Must(entry != NULL); + Must(httpRequest != NULL); + + EBIT_CLR(entry->flags, ENTRY_FWD_HDR_WAIT); + entry->buffer(); + entry->replaceHttpReply(error->BuildHttpReply()); + entry->expires = squid_curtime; + errorStateFree(error); + entry->flush(); + entry->complete(); +} === added file 'src/mgr/Forwarder.h' --- src/mgr/Forwarder.h 1970-01-01 00:00:00 +0000 +++ src/mgr/Forwarder.h 2011-01-17 15:10:36 +0000 @@ -0,0 +1,57 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#ifndef SQUID_MGR_FORWARDER_H +#define SQUID_MGR_FORWARDER_H + +#include "ipc/Forwarder.h" +#include "mgr/ActionParams.h" + + +class CommCloseCbParams; +class HttpRequest; +class StoreEntry; +class ErrorState; + +namespace Mgr +{ + +/** Forwards a single client cache manager request to Coordinator. + * Waits for an ACK from Coordinator while holding the Store entry. + * Fills the store entry with an error response if forwarding fails. + */ +class Forwarder: public Ipc::Forwarder +{ +public: + Forwarder(int aFd, const ActionParams &aParams, HttpRequest* aRequest, + StoreEntry* anEntry); + virtual ~Forwarder(); + +protected: + /* Ipc::Forwarder API */ + virtual void cleanup(); ///< perform cleanup actions + virtual void handleError(); + virtual void handleTimeout(); + virtual void handleException(const std::exception& e); + virtual void handleRemoteAck(); + +private: + void noteCommClosed(const CommCloseCbParams& params); + void sendError(ErrorState* error); + +private: + HttpRequest* httpRequest; ///< HTTP client request for detailing errors + StoreEntry* entry; ///< Store entry expecting the response + int fd; ///< HTTP client connection descriptor + AsyncCall::Pointer closer; ///< comm_close handler for the HTTP connection + + CBDATA_CLASS2(Forwarder); +}; + +} // namespace Mgr + +#endif /* SQUID_MGR_FORWARDER_H */ === modified file 'src/mgr/FunAction.cc' --- src/mgr/FunAction.cc 2010-10-29 00:12:28 +0000 +++ src/mgr/FunAction.cc 2011-01-17 15:10:36 +0000 @@ -1,50 +1,51 @@ /* * $Id$ * * DEBUG: section 16 Cache Manager API * */ #include "config.h" #include "base/TextException.h" +#include "ipc/UdsOp.h" #include "mgr/Command.h" #include "mgr/Filler.h" #include "mgr/FunAction.h" #include "mgr/Request.h" #include "Store.h" Mgr::FunAction::Pointer Mgr::FunAction::Create(const Command::Pointer &aCmd, OBJH* aHandler) { return new FunAction(aCmd, aHandler); } Mgr::FunAction::FunAction(const Command::Pointer &aCmd, OBJH* aHandler): Action(aCmd), handler(aHandler) { Must(handler != NULL); debugs(16, 5, HERE); } void Mgr::FunAction::respond(const Request& request) { debugs(16, 5, HERE); - const int fd = ImportHttpFdIntoComm(request.fd); + const int fd = Ipc::ImportFdIntoComm(request.fd, SOCK_STREAM, IPPROTO_TCP, Ipc::fdnHttpSocket); Must(fd >= 0); Must(request.requestId != 0); AsyncJob::Start(new Mgr::Filler(this, fd, request.requestId)); } void Mgr::FunAction::dump(StoreEntry* entry) { debugs(16, 5, HERE); Must(entry != NULL); if (UsingSmp() && IamWorkerProcess()) storeAppendPrintf(entry, "by kid%d {\n", KidIdentifier); handler(entry); if (atomic() && UsingSmp() && IamWorkerProcess()) storeAppendPrintf(entry, "} by kid%d\n\n", KidIdentifier); } === modified file 'src/mgr/InfoAction.cc' --- src/mgr/InfoAction.cc 2010-10-29 00:12:28 +0000 +++ src/mgr/InfoAction.cc 2011-01-17 15:10:36 +0000 @@ -1,31 +1,32 @@ /* * $Id$ * * DEBUG: section 16 Cache Manager API * */ #include "config.h" #include "base/TextException.h" #include "HttpReply.h" #include "ipc/Messages.h" +#include "ipc/UdsOp.h" #include "ipc/TypedMsgHdr.h" #include "mgr/Filler.h" #include "mgr/InfoAction.h" #include "mgr/Request.h" #include "mgr/Response.h" #include "SquidTime.h" #include "Store.h" extern void GetInfo(Mgr::InfoActionData& stats); extern void DumpInfo(Mgr::InfoActionData& stats, StoreEntry* sentry); extern void DumpMallocStatistics(StoreEntry* sentry); Mgr::InfoActionData::InfoActionData() { xmemset(this, 0, sizeof(*this)); } Mgr::InfoActionData& Mgr::InfoActionData::operator += (const InfoActionData& stats) @@ -138,41 +139,41 @@ return new InfoAction(cmd); } Mgr::InfoAction::InfoAction(const CommandPointer &cmd): Action(cmd), data() { debugs(16, 5, HERE); } void Mgr::InfoAction::add(const Action& action) { debugs(16, 5, HERE); data += dynamic_cast(action).data; } void Mgr::InfoAction::respond(const Request& request) { debugs(16, 5, HERE); - int fd = ImportHttpFdIntoComm(request.fd); + int fd = Ipc::ImportFdIntoComm(request.fd, SOCK_STREAM, IPPROTO_TCP, Ipc::fdnHttpSocket); Must(fd >= 0); Must(request.requestId != 0); AsyncJob::Start(new Mgr::Filler(this, fd, request.requestId)); } void Mgr::InfoAction::collect() { GetInfo(data); } void Mgr::InfoAction::dump(StoreEntry* entry) { debugs(16, 5, HERE); Must(entry != NULL); #if XMALLOC_STATISTICS if (UsingSmp() && IamWorkerProcess()) storeAppendPrintf(entry, "by kid%d {\n", KidIdentifier); === added file 'src/mgr/Inquirer.cc' --- src/mgr/Inquirer.cc 1970-01-01 00:00:00 +0000 +++ src/mgr/Inquirer.cc 2011-01-17 16:57:44 +0000 @@ -0,0 +1,121 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "comm/Write.h" +#include "CommCalls.h" +#include "HttpReply.h" +#include "ipc/UdsOp.h" +#include "mgr/ActionWriter.h" +#include "mgr/Inquirer.h" +#include "mgr/Request.h" +#include "mgr/Response.h" +#include "SquidTime.h" +#include + + +CBDATA_NAMESPACED_CLASS_INIT(Mgr, Inquirer); + + +Mgr::Inquirer::Inquirer(Action::Pointer anAction, + const Request &aCause, const Ipc::StrandCoords &coords): + Ipc::Inquirer(aCause.clone(), coords, anAction->atomic() ? 10 : 100), + aggrAction(anAction), + fd(Ipc::ImportFdIntoComm(aCause.fd, SOCK_STREAM, IPPROTO_TCP, Ipc::fdnHttpSocket)) +{ + debugs(16, 5, HERE << "FD " << fd << " action: " << aggrAction); + + closer = asyncCall(16, 5, "Mgr::Inquirer::noteCommClosed", + CommCbMemFunT(this, &Inquirer::noteCommClosed)); + comm_add_close_handler(fd, closer); +} + +/// closes our copy of the client HTTP connection socket +void +Mgr::Inquirer::cleanup() +{ + if (fd >= 0) { + removeCloseHandler(); + comm_close(fd); + fd = -1; + } +} + +void +Mgr::Inquirer::removeCloseHandler() +{ + if (closer != NULL) { + comm_remove_close_handler(fd, closer); + closer = NULL; + } +} + +void +Mgr::Inquirer::start() +{ + debugs(16, 5, HERE); + Ipc::Inquirer::start(); + Must(fd >= 0); + Must(aggrAction != NULL); + + std::auto_ptr reply(new HttpReply); + reply->setHeaders(HTTP_OK, NULL, "text/plain", -1, squid_curtime, squid_curtime); + reply->header.putStr(HDR_CONNECTION, "close"); // until we chunk response + std::auto_ptr replyBuf(reply->pack()); + writer = asyncCall(16, 5, "Mgr::Inquirer::noteWroteHeader", + CommCbMemFunT(this, &Inquirer::noteWroteHeader)); + Comm::Write(fd, replyBuf.get(), writer); +} + +/// called when we wrote the response header +void +Mgr::Inquirer::noteWroteHeader(const CommIoCbParams& params) +{ + debugs(16, 5, HERE); + writer = NULL; + Must(params.flag == COMM_OK); + Must(params.fd == fd); + Must(params.size != 0); + // start inquiries at the initial pos + inquire(); +} + +/// called when the HTTP client or some external force closed our socket +void +Mgr::Inquirer::noteCommClosed(const CommCloseCbParams& params) +{ + debugs(16, 5, HERE); + Must(fd < 0 || fd == params.fd); + fd = -1; + mustStop("commClosed"); +} + +bool +Mgr::Inquirer::aggregate(Ipc::Response::Pointer aResponse) +{ + Mgr::Response& response = static_cast(*aResponse); + if (response.hasAction()) + aggrAction->add(response.getAction()); + return true; +} + +void +Mgr::Inquirer::sendResponse() +{ + if (aggrAction->aggregatable()) { + removeCloseHandler(); + AsyncJob::Start(new ActionWriter(aggrAction, fd)); + fd = -1; // should not close fd because we passed it to ActionWriter + } +} + +bool +Mgr::Inquirer::doneAll() const +{ + return !writer && Ipc::Inquirer::doneAll(); +} === added file 'src/mgr/Inquirer.h' --- src/mgr/Inquirer.h 1970-01-01 00:00:00 +0000 +++ src/mgr/Inquirer.h 2011-01-17 15:10:36 +0000 @@ -0,0 +1,56 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#ifndef SQUID_MGR_INQUIRER_H +#define SQUID_MGR_INQUIRER_H + +#include "ipc/Inquirer.h" +#include "mgr/Action.h" + +class CommIoCbParams; +class CommCloseCbParams; + +namespace Mgr +{ + +/// Coordinator's job that sends a cache manage request to each strand, +/// aggregating individual strand responses and dumping the result if needed +class Inquirer: public Ipc::Inquirer +{ +public: + Inquirer(Action::Pointer anAction, const Request &aCause, + const Ipc::StrandCoords &coords); + +protected: + /* AsyncJob API */ + virtual void start(); + virtual bool doneAll() const; + + /* Ipc::Inquirer API */ + virtual void cleanup(); + virtual void sendResponse(); + virtual bool aggregate(Ipc::Response::Pointer aResponse); + +private: + void noteWroteHeader(const CommIoCbParams& params); + void noteCommClosed(const CommCloseCbParams& params); + void removeCloseHandler(); + +private: + Action::Pointer aggrAction; //< action to aggregate + + int fd; ///< HTTP client socket descriptor + + AsyncCall::Pointer writer; ///< comm_write callback + AsyncCall::Pointer closer; ///< comm_close handler + + CBDATA_CLASS2(Inquirer); +}; + +} // namespace Mgr + +#endif /* SQUID_MGR_INQUIRER_H */ === added file 'src/mgr/Request.cc' --- src/mgr/Request.cc 1970-01-01 00:00:00 +0000 +++ src/mgr/Request.cc 2011-01-17 15:10:36 +0000 @@ -0,0 +1,56 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "ipc/Messages.h" +#include "ipc/TypedMsgHdr.h" +#include "mgr/ActionParams.h" +#include "mgr/Request.h" + + +Mgr::Request::Request(int aRequestorId, unsigned int aRequestId, int aFd, + const ActionParams &aParams): + Ipc::Request(aRequestorId, aRequestId), + fd(aFd), params(aParams) +{ + Must(requestorId > 0); +} + +Mgr::Request::Request(const Request& request): + Ipc::Request(request.requestorId, request.requestId), + fd(request.fd), params(request.params) +{ +} + +Mgr::Request::Request(const Ipc::TypedMsgHdr& msg): + Ipc::Request(0, 0) +{ + msg.checkType(Ipc::mtCacheMgrRequest); + msg.getPod(requestorId); + msg.getPod(requestId); + params = ActionParams(msg); + + fd = msg.getFd(); +} + +void +Mgr::Request::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.setType(Ipc::mtCacheMgrRequest); + msg.putPod(requestorId); + msg.putPod(requestId); + params.pack(msg); + + msg.putFd(fd); +} + +Ipc::Request::Pointer +Mgr::Request::clone() const +{ + return new Request(*this); +} === removed file 'src/mgr/Request.cc' --- src/mgr/Request.cc 2010-10-29 00:12:28 +0000 +++ src/mgr/Request.cc 1970-01-01 00:00:00 +0000 @@ -1,43 +0,0 @@ -/* - * $Id$ - * - * DEBUG: section 16 Cache Manager API - * - */ - -#include "config.h" -#include "base/TextException.h" -#include "ipc/Messages.h" -#include "mgr/ActionParams.h" -#include "mgr/Request.h" - - -Mgr::Request::Request(int aRequestorId, unsigned int aRequestId, int aFd, - const ActionParams &aParams): - requestorId(aRequestorId), requestId(aRequestId), - fd(aFd), params(aParams) -{ - Must(requestorId > 0); - Must(requestId != 0); -} - -Mgr::Request::Request(const Ipc::TypedMsgHdr& msg) -{ - msg.checkType(Ipc::mtCacheMgrRequest); - msg.getPod(requestorId); - msg.getPod(requestId); - params = ActionParams(msg); - - fd = msg.getFd(); -} - -void -Mgr::Request::pack(Ipc::TypedMsgHdr& msg) const -{ - msg.setType(Ipc::mtCacheMgrRequest); - msg.putPod(requestorId); - msg.putPod(requestId); - params.pack(msg); - - msg.putFd(fd); -} === added file 'src/mgr/Request.h' --- src/mgr/Request.h 1970-01-01 00:00:00 +0000 +++ src/mgr/Request.h 2011-01-17 15:10:36 +0000 @@ -0,0 +1,43 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#ifndef SQUID_MGR_REQUEST_H +#define SQUID_MGR_REQUEST_H + +#include "ipc/forward.h" +#include "ipc/Request.h" +#include "mgr/ActionParams.h" + + +namespace Mgr +{ + +/// cache manager request +class Request: public Ipc::Request +{ +public: + Request(int aRequestorId, unsigned int aRequestId, int aFd, + const ActionParams &aParams); + + explicit Request(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() + /* Ipc::Request API */ + virtual void pack(Ipc::TypedMsgHdr& msg) const; + virtual Pointer clone() const; + +private: + Request(const Request& request); + +public: + int fd; ///< HTTP client connection descriptor + + ActionParams params; ///< action name and parameters +}; + + +} // namespace Mgr + +#endif /* SQUID_MGR_REQUEST_H */ === added file 'src/mgr/Response.cc' --- src/mgr/Response.cc 1970-01-01 00:00:00 +0000 +++ src/mgr/Response.cc 2011-01-17 15:10:36 +0000 @@ -0,0 +1,74 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "CacheManager.h" +#include "ipc/Messages.h" +#include "ipc/TypedMsgHdr.h" +#include "mgr/ActionCreator.h" +#include "mgr/ActionProfile.h" +#include "mgr/Response.h" + + +Mgr::Response::Response(unsigned int aRequestId, Action::Pointer anAction): + Ipc::Response(aRequestId), action(anAction) +{ + Must(!action || action->name()); // if there is an action, it must be named +} + +Mgr::Response::Response(const Response& response): + Ipc::Response(response.requestId), action(response.action) +{ +} + +Mgr::Response::Response(const Ipc::TypedMsgHdr& msg): + Ipc::Response(0) +{ + msg.checkType(Ipc::mtCacheMgrResponse); + msg.getPod(requestId); + Must(requestId != 0); + + if (msg.hasMoreData()) { + String actionName; + msg.getString(actionName); + action = CacheManager::GetInstance()->createNamedAction(actionName.termedBuf()); + Must(hasAction()); + action->unpack(msg); + } +} + +void +Mgr::Response::pack(Ipc::TypedMsgHdr& msg) const +{ + Must(requestId != 0); + msg.setType(Ipc::mtCacheMgrResponse); + msg.putPod(requestId); + if (hasAction()) { + msg.putString(action->name()); + action->pack(msg); + } +} + +Ipc::Response::Pointer +Mgr::Response::clone() const +{ + return new Response(*this); +} + +bool +Mgr::Response::hasAction() const +{ + return action != NULL; +} + +const Mgr::Action& +Mgr::Response::getAction() const +{ + Must(hasAction()); + return *action; +} === removed file 'src/mgr/Response.cc' --- src/mgr/Response.cc 2010-10-29 00:12:28 +0000 +++ src/mgr/Response.cc 1970-01-01 00:00:00 +0000 @@ -1,68 +0,0 @@ -/* - * $Id$ - * - * DEBUG: section 16 Cache Manager API - * - */ - -#include "config.h" -#include "base/TextException.h" -#include "CacheManager.h" -#include "ipc/Messages.h" -#include "ipc/TypedMsgHdr.h" -#include "mgr/ActionCreator.h" -#include "mgr/ActionProfile.h" -#include "mgr/Response.h" - - -std::ostream& Mgr::operator << (std::ostream &os, const Response& response) -{ - os << "response: {requestId: " << response.requestId << '}'; - return os; -} - -Mgr::Response::Response(unsigned int aRequestId, Action::Pointer anAction): - requestId(aRequestId), action(anAction) -{ - Must(!action || action->name()); // if there is an action, it must be named -} - -Mgr::Response::Response(const Ipc::TypedMsgHdr& msg) -{ - msg.checkType(Ipc::mtCacheMgrResponse); - msg.getPod(requestId); - Must(requestId != 0); - - if (msg.hasMoreData()) { - String actionName; - msg.getString(actionName); - action = CacheManager::GetInstance()->createNamedAction(actionName.termedBuf()); - Must(hasAction()); - action->unpack(msg); - } -} - -void -Mgr::Response::pack(Ipc::TypedMsgHdr& msg) const -{ - Must(requestId != 0); - msg.setType(Ipc::mtCacheMgrResponse); - msg.putPod(requestId); - if (hasAction()) { - msg.putString(action->name()); - action->pack(msg); - } -} - -bool -Mgr::Response::hasAction() const -{ - return action != NULL; -} - -const Mgr::Action& -Mgr::Response::getAction() const -{ - Must(hasAction()); - return *action; -} === added file 'src/mgr/Response.h' --- src/mgr/Response.h 1970-01-01 00:00:00 +0000 +++ src/mgr/Response.h 2011-01-17 15:10:36 +0000 @@ -0,0 +1,44 @@ +/* + * $Id$ + * + * DEBUG: section 16 Cache Manager API + * + */ + +#ifndef SQUID_MGR_RESPONSE_H +#define SQUID_MGR_RESPONSE_H + +#include "ipc/forward.h" +#include "ipc/Response.h" +#include "mgr/Action.h" + + +namespace Mgr +{ + +/// A response to Mgr::Request. +/// May carry strand action data to be aggregated with data from other strands. +class Response: public Ipc::Response +{ +public: + Response(unsigned int aRequestId, Action::Pointer anAction = NULL); + + explicit Response(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() + + /* Ipc::Response API */ + virtual void pack(Ipc::TypedMsgHdr& msg) const; + virtual Ipc::Response::Pointer clone() const; + + bool hasAction() const; ///< whether response contain action object + const Action& getAction() const; ///< returns action object + +private: + Response(const Response& response); + +public: + Action::Pointer action; ///< action relating to response +}; + +} // namespace Mgr + +#endif /* SQUID_MGR_RESPONSE_H */ === modified file 'src/mgr/StoreToCommWriter.cc' --- src/mgr/StoreToCommWriter.cc 2010-11-27 01:46:22 +0000 +++ src/mgr/StoreToCommWriter.cc 2011-01-17 15:10:36 +0000 @@ -147,42 +147,20 @@ entry = NULL; } close(); } bool Mgr::StoreToCommWriter::doneAll() const { return entry && entry->store_status == STORE_OK && // the action is over writeOffset >= entry->objectLen(); // we wrote all the results } void Mgr::StoreToCommWriter::Abort(void* param) { StoreToCommWriter* mgrWriter = static_cast(param); if (mgrWriter->fd >= 0) comm_close(mgrWriter->fd); } - - -int -Mgr::ImportHttpFdIntoComm(int fd) -{ - struct sockaddr_in addr; - socklen_t len = sizeof(addr); - if (getsockname(fd, reinterpret_cast(&addr), &len) == 0) { - Ip::Address ipAddr(addr); - struct addrinfo* addr_info = NULL; - ipAddr.GetAddrInfo(addr_info); - addr_info->ai_socktype = SOCK_STREAM; - addr_info->ai_protocol = IPPROTO_TCP; - comm_import_opened(fd, ipAddr, COMM_NONBLOCKING, Ipc::FdNote(Ipc::fdnHttpSocket), addr_info); - ipAddr.FreeAddrInfo(addr_info); - } else { - debugs(16, DBG_CRITICAL, HERE << "ERROR: FD " << fd << ' ' << xstrerror()); - ::close(fd); - fd = -1; - } - return fd; -} === modified file 'src/mgr/StoreToCommWriter.h' --- src/mgr/StoreToCommWriter.h 2010-10-28 18:52:59 +0000 +++ src/mgr/StoreToCommWriter.h 2011-01-17 15:10:36 +0000 @@ -48,26 +48,23 @@ void noteCommWrote(const CommIoCbParams& params); /// called by Comm if the client socket got closed void noteCommClosed(const CommCloseCbParams& params); /// closes the local connection to the HTTP client, if any void close(); protected: int fd; ///< HTTP client descriptor StoreEntry* entry; ///< store entry with the cache manager response store_client* sc; ///< our registration with the store int64_t writeOffset; ///< number of bytes written to the client AsyncCall::Pointer closer; ///< comm_close handler char buffer[HTTP_REQBUF_SZ]; ///< action results; Store fills, Comm writes CBDATA_CLASS2(StoreToCommWriter); }; -/// import HTTP socket fd from another strand into our Comm state -extern int ImportHttpFdIntoComm(int fd); - } // namespace Mgr #endif /* SQUID_MGR_STORE_TO_COMM_WRITER_H */ === added directory 'src/snmp' === added file 'src/snmp/Forwarder.cc' --- src/snmp/Forwarder.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Forwarder.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,107 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "CommCalls.h" +#include "ipc/Port.h" +#include "snmp_core.h" +#include "snmp/Forwarder.h" +#include "snmp/Request.h" +#include "snmp/Response.h" + + +CBDATA_NAMESPACED_CLASS_INIT(Snmp, Forwarder); + + +Snmp::Forwarder::Forwarder(const Pdu& aPdu, const Session& aSession, int aFd, + const Ip::Address& anAddress): + Ipc::Forwarder(new Request(KidIdentifier, 0, aPdu, aSession, aFd, anAddress), 2), + fd(aFd) +{ + debugs(49, 5, HERE << "FD " << aFd); + Must(fd >= 0); + closer = asyncCall(49, 5, "Snmp::Forwarder::noteCommClosed", + CommCbMemFunT(this, &Forwarder::noteCommClosed)); + comm_add_close_handler(fd, closer); +} + +/// removes our cleanup handler of the client connection socket +void +Snmp::Forwarder::cleanup() +{ + if (fd >= 0) { + if (closer != NULL) { + comm_remove_close_handler(fd, closer); + closer = NULL; + } + fd = -1; + } +} + +/// called when the client socket gets closed by some external force +void +Snmp::Forwarder::noteCommClosed(const CommCloseCbParams& params) +{ + debugs(49, 5, HERE); + Must(fd == params.fd); + fd = -1; + mustStop("commClosed"); +} + +void +Snmp::Forwarder::handleTimeout() +{ + sendError(SNMP_ERR_RESOURCEUNAVAILABLE); + Ipc::Forwarder::handleTimeout(); +} + +void +Snmp::Forwarder::handleException(const std::exception& e) +{ + debugs(49, 3, HERE << e.what()); + if (fd >= 0) + sendError(SNMP_ERR_GENERR); + Ipc::Forwarder::handleException(e); +} + +/// send error SNMP response +void +Snmp::Forwarder::sendError(int error) +{ + debugs(49, 3, HERE); + Snmp::Request& req = static_cast(*request); + req.pdu.command = SNMP_PDU_RESPONSE; + req.pdu.errstat = error; + u_char buffer[SNMP_REQUEST_SIZE]; + int len = sizeof(buffer); + snmp_build(&req.session, &req.pdu, buffer, &len); + comm_udp_sendto(fd, req.address, buffer, len); +} + +void +Snmp::SendResponse(unsigned int requestId, const Pdu& pdu) +{ + debugs(49, 5, HERE); + // snmpAgentResponse() can modify arg + Pdu tmp = pdu; + Snmp::Response response(requestId); + snmp_pdu* response_pdu = NULL; + try { + response_pdu = snmpAgentResponse(&tmp); + Must(response_pdu != NULL); + response.pdu = static_cast(*response_pdu); + snmp_free_pdu(response_pdu); + } catch (const std::exception& e) { + debugs(49, DBG_CRITICAL, HERE << e.what()); + response.pdu.command = SNMP_PDU_RESPONSE; + response.pdu.errstat = SNMP_ERR_GENERR; + } + Ipc::TypedMsgHdr message; + response.pack(message); + Ipc::SendMessage(Ipc::coordinatorAddr, message); +} === added file 'src/snmp/Forwarder.h' --- src/snmp/Forwarder.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Forwarder.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,52 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_FORWARDER_H +#define SQUID_SNMPX_FORWARDER_H + +#include "ipc/Forwarder.h" +#include "snmp/Pdu.h" +#include "snmp/Session.h" + + +class CommCloseCbParams; + +namespace Snmp +{ + +/** Forwards a single client SNMP request to Coordinator. + * Waits for an ACK from Coordinator + * Send the data unit with an error response if forwarding fails. + */ +class Forwarder: public Ipc::Forwarder +{ +public: + Forwarder(const Pdu& aPdu, const Session& aSession, int aFd, + const Ip::Address& anAddress); + +protected: + /* Ipc::Forwarder API */ + virtual void cleanup(); ///< perform cleanup actions + virtual void handleTimeout(); + virtual void handleException(const std::exception& e); + +private: + void noteCommClosed(const CommCloseCbParams& params); + void sendError(int error); + +private: + int fd; ///< client connection descriptor + AsyncCall::Pointer closer; ///< comm_close handler for the connection + + CBDATA_CLASS2(Forwarder); +}; + +extern void SendResponse(unsigned int requestId, const Pdu& pdu); + +} // namespace Snmp + +#endif /* SQUID_SNMPX_FORWARDER_H */ === added file 'src/snmp/Inquirer.cc' --- src/snmp/Inquirer.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Inquirer.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,102 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "CommCalls.h" +#include "ipc/UdsOp.h" +#include "snmp_core.h" +#include "snmp/Inquirer.h" +#include "snmp/Response.h" +#include "snmp/Request.h" + + +CBDATA_NAMESPACED_CLASS_INIT(Snmp, Inquirer); + + +Snmp::Inquirer::Inquirer(const Request& aRequest, const Ipc::StrandCoords& coords): + Ipc::Inquirer(aRequest.clone(), coords, 2), + aggrPdu(aRequest.pdu), + fd(ImportFdIntoComm(aRequest.fd, SOCK_DGRAM, IPPROTO_UDP, Ipc::fdnInSnmpSocket)) +{ + debugs(49, 5, HERE); + closer = asyncCall(49, 5, "Snmp::Inquirer::noteCommClosed", + CommCbMemFunT(this, &Inquirer::noteCommClosed)); + comm_add_close_handler(fd, closer); +} + +/// closes our copy of the client connection socket +void +Snmp::Inquirer::cleanup() +{ + if (fd >= 0) { + if (closer != NULL) { + comm_remove_close_handler(fd, closer); + closer = NULL; + } + comm_close(fd); + fd = -1; + } +} + +void +Snmp::Inquirer::start() +{ + debugs(49, 5, HERE); + Ipc::Inquirer::start(); + Must(fd >= 0); + inquire(); +} + +void +Snmp::Inquirer::handleException(const std::exception& e) +{ + aggrPdu.errstat = SNMP_ERR_GENERR; + Ipc::Inquirer::handleException(e); +} + +bool +Snmp::Inquirer::aggregate(Response::Pointer aResponse) +{ + Snmp::Response& response = static_cast(*aResponse); + bool error = response.pdu.errstat != SNMP_ERR_NOERROR; + if (error) { + aggrPdu = response.pdu; + } else { + aggrPdu.aggregate(response.pdu); + } + return !error; +} + +/// called when the some external force closed our socket +void +Snmp::Inquirer::noteCommClosed(const CommCloseCbParams& params) +{ + debugs(49, 5, HERE); + Must(fd < 0 || fd == params.fd); + fd = -1; + mustStop("commClosed"); +} + +bool +Snmp::Inquirer::doneAll() const +{ + return !writer && Ipc::Inquirer::doneAll(); +} + +void +Snmp::Inquirer::sendResponse() +{ + debugs(49, 5, HERE); + aggrPdu.fixAggregate(); + aggrPdu.command = SNMP_PDU_RESPONSE; + u_char buffer[SNMP_REQUEST_SIZE]; + int len = sizeof(buffer); + Snmp::Request& req = static_cast(*request); + snmp_build(&req.session, &aggrPdu, buffer, &len); + comm_udp_sendto(fd, req.address, buffer, len); +} === added file 'src/snmp/Inquirer.h' --- src/snmp/Inquirer.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Inquirer.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,54 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_INQUIRER_H +#define SQUID_SNMPX_INQUIRER_H + +#include "ipc/Inquirer.h" +#include "snmp/forward.h" +#include "snmp/Pdu.h" + + +class CommCloseCbParams; + +namespace Snmp +{ + +/// Coordinator's job that sends a PDU request to each strand, +/// aggregates strand responses and send back the result to client +class Inquirer: public Ipc::Inquirer +{ +public: + Inquirer(const Request& aRequest, const Ipc::StrandCoords& coords); + +protected: + /* AsyncJob API */ + virtual void start(); + virtual bool doneAll() const; + + /* Ipc::Inquirer API */ + virtual void cleanup(); + virtual void handleException(const std::exception& e); + virtual void sendResponse(); + virtual bool aggregate(Ipc::Response::Pointer aResponse); + +private: + void noteCommClosed(const CommCloseCbParams& params); + +private: + Pdu aggrPdu; ///< aggregated pdu + int fd; ///< client connection descriptor + + AsyncCall::Pointer writer; ///< comm_write callback + AsyncCall::Pointer closer; ///< comm_close handler + + CBDATA_CLASS2(Inquirer); +}; + +} // namespace Snmp + +#endif /* SQUID_SNMPX_INQUIRER_H */ === added file 'src/snmp/Pdu.cc' --- src/snmp/Pdu.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Pdu.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,239 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "ipc/TypedMsgHdr.h" +#include "protos.h" +#include "snmp_core.h" +#include "snmp/Pdu.h" +#include "snmp/Var.h" + + +Snmp::Pdu::Pdu() +{ + init(); +} + +Snmp::Pdu::Pdu(const Pdu& pdu) +{ + init(); + assign(pdu); +} + +Snmp::Pdu::~Pdu() +{ + clear(); +} + +Snmp::Pdu& +Snmp::Pdu::operator = (const Pdu& pdu) +{ + clear(); + assign(pdu); + return *this; +} + +void +Snmp::Pdu::init() +{ + xmemset(this, 0, sizeof(*this)); + errstat = SNMP_DEFAULT_ERRSTAT; + errindex = SNMP_DEFAULT_ERRINDEX; +} + +void +Snmp::Pdu::aggregate(const Pdu& pdu) +{ + Must(varCount() == pdu.varCount()); + aggrCount++; + for (variable_list* p_aggr = variables, *p_var = pdu.variables; p_var != NULL; + p_aggr = p_aggr->next_variable, p_var = p_var->next_variable) + { + Must(p_aggr != NULL); + Var& aggr = static_cast(*p_aggr); + Var& var = static_cast(*p_var); + if (aggr.isNull()) { + aggr.setName(var.getName()); + aggr.copyValue(var); + } else { + switch(snmpAggrType(aggr.name, aggr.name_length)) + { + case atSum: + case atAverage: + // The mean-average division is done later + // when the Snmp::Pdu::fixAggregate() called + aggr += var; + break; + case atMax: + if (var > aggr) + aggr.copyValue(var); + break; + case atMin: + if (var < aggr) + aggr.copyValue(var); + break; + default: + break; + } + } + } +} + +void +Snmp::Pdu::clear() +{ + clearSystemOid(); + clearVars(); + init(); +} + +void +Snmp::Pdu::assign(const Pdu& pdu) +{ + command = pdu.command; + address.sin_addr.s_addr = pdu.address.sin_addr.s_addr; + reqid = pdu.reqid; + errstat = pdu.errstat; + errindex = pdu.errindex; + non_repeaters = pdu.non_repeaters; + max_repetitions = pdu.max_repetitions; + agent_addr.sin_addr.s_addr = pdu.agent_addr.sin_addr.s_addr; + trap_type = pdu.trap_type; + specific_type = pdu.specific_type; + time = pdu.time; + aggrCount = pdu.aggrCount; + setSystemOid(pdu.getSystemOid()); + setVars(pdu.variables); +} + +void +Snmp::Pdu::clearVars() +{ + variable_list* var = variables; + while (var != NULL) { + variable_list* tmp = var; + var = var->next_variable; + snmp_var_free(tmp); + } + variables = NULL; +} + +void +Snmp::Pdu::setVars(variable_list* vars) +{ + clearVars(); + for (variable_list** p_var = &variables; vars != NULL; + vars = vars->next_variable, p_var = &(*p_var)->next_variable) + { + *p_var = new Var(static_cast(*vars)); + } +} + +void +Snmp::Pdu::clearSystemOid() +{ + if (enterprise != NULL) { + xfree(enterprise); + enterprise = NULL; + } + enterprise_length = 0; +} + +Range +Snmp::Pdu::getSystemOid() const +{ + return Range(enterprise, enterprise + enterprise_length); +} + +void +Snmp::Pdu::setSystemOid(const Range& systemOid) +{ + clearSystemOid(); + if (systemOid.start != NULL && systemOid.size() != 0) { + enterprise_length = systemOid.size(); + enterprise = static_cast(xmalloc(enterprise_length * sizeof(oid))); + std::copy(systemOid.start, systemOid.end, enterprise); + } +} + +void +Snmp::Pdu::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.putPod(command); + msg.putPod(address); + msg.putPod(reqid); + msg.putPod(errstat); + msg.putPod(errindex); + msg.putPod(non_repeaters); + msg.putPod(max_repetitions); + msg.putInt(enterprise_length); + if (enterprise_length > 0) { + Must(enterprise != NULL); + msg.putFixed(enterprise, enterprise_length * sizeof(oid)); + } + msg.putPod(agent_addr); + msg.putPod(trap_type); + msg.putPod(specific_type); + msg.putPod(time); + msg.putInt(varCount()); + for (variable_list* var = variables; var != NULL; var = var->next_variable) + static_cast(var)->pack(msg); +} + +void +Snmp::Pdu::unpack(const Ipc::TypedMsgHdr& msg) +{ + clear(); + msg.getPod(command); + msg.getPod(address); + msg.getPod(reqid); + msg.getPod(errstat); + msg.getPod(errindex); + msg.getPod(non_repeaters); + msg.getPod(max_repetitions); + enterprise_length = msg.getInt(); + if (enterprise_length > 0) { + enterprise = static_cast(xmalloc(enterprise_length * sizeof(oid))); + msg.getFixed(enterprise, enterprise_length * sizeof(oid)); + } + msg.getPod(agent_addr); + msg.getPod(trap_type); + msg.getPod(specific_type); + msg.getPod(time); + int count = msg.getInt(); + for (variable_list** p_var = &variables; count > 0; + p_var = &(*p_var)->next_variable, --count) + { + Var* var = new Var(); + var->unpack(msg); + *p_var = var; + } +} + +int +Snmp::Pdu::varCount() const +{ + int count = 0; + for (variable_list* var = variables; var != NULL; var = var->next_variable) + ++count; + return count; +} + +void +Snmp::Pdu::fixAggregate() +{ + if (aggrCount < 2) + return; + for (variable_list* p_aggr = variables; p_aggr != NULL; p_aggr = p_aggr->next_variable) { + Var& aggr = static_cast(*p_aggr); + if(snmpAggrType(aggr.name, aggr.name_length) == atAverage) { + aggr /= aggrCount; + } + } + aggrCount = 0; +} === added file 'src/snmp/Pdu.h' --- src/snmp/Pdu.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Pdu.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,50 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_PDU_H +#define SQUID_SNMPX_PDU_H + +#include "config.h" +#include "ipc/forward.h" +#include "Range.h" +#include "snmp.h" + + +namespace Snmp +{ + +/// snmp_pdu wrapper introduce the feature +/// to aggregate variables and to pack/unpack message +class Pdu: public snmp_pdu +{ +public: + Pdu(); + Pdu(const Pdu& pdu); + Pdu& operator = (const Pdu& pdu); + ~Pdu(); + + void aggregate(const Pdu& pdu); + void fixAggregate(); + void pack(Ipc::TypedMsgHdr& msg) const; ///< prepare for sendmsg() + void unpack(const Ipc::TypedMsgHdr& msg); ///< restore struct from the message + int varCount() const; ///< size of variables list + void clear(); ///< clear all internal members + void setVars(variable_list* vars); ///< perform assignment of variables list + void clearVars(); ///< clear variables list + Range getSystemOid() const; + void setSystemOid(const Range& systemOid); + void clearSystemOid(); + +private: + void init(); ///< initialize members + void assign(const Pdu& pdu); ///< perform full assignment + unsigned int aggrCount; ///< The number of other Pdus merged into +}; + +} // namespace Snmp + +#endif /* SQUID_SNMPX_PDU_H */ === added file 'src/snmp/Request.cc' --- src/snmp/Request.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Request.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,59 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "ipc/Messages.h" +#include "ipc/TypedMsgHdr.h" +#include "snmp/Request.h" + + +Snmp::Request::Request(int aRequestorId, unsigned int aRequestId, + const Pdu& aPdu, const Session& aSession, + int aFd, const Ip::Address& anAddress): + Ipc::Request(aRequestorId, aRequestId), + pdu(aPdu), session(aSession), fd(aFd), address(anAddress) +{ +} + +Snmp::Request::Request(const Request& request): + Ipc::Request(request.requestorId, request.requestId), + pdu(request.pdu), session(request.session), + fd(request.fd), address(request.address) +{ +} + +Snmp::Request::Request(const Ipc::TypedMsgHdr& msg): + Ipc::Request(0, 0) +{ + msg.checkType(Ipc::mtSnmpRequest); + msg.getPod(requestorId); + msg.getPod(requestId); + pdu.unpack(msg); + session.unpack(msg); + msg.getPod(address); + + fd = msg.getFd(); +} + +void +Snmp::Request::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.setType(Ipc::mtSnmpRequest); + msg.putPod(requestorId); + msg.putPod(requestId); + pdu.pack(msg); + session.pack(msg); + msg.putPod(address); + + msg.putFd(fd); +} + +Ipc::Request::Pointer +Snmp::Request::clone() const +{ + return new Request(*this); +} === added file 'src/snmp/Request.h' --- src/snmp/Request.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Request.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,46 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_REQUEST_H +#define SQUID_SNMPX_REQUEST_H + +#include "ip/Address.h" +#include "ipc/forward.h" +#include "ipc/Request.h" +#include "snmp/Pdu.h" +#include "snmp/Session.h" + + +namespace Snmp +{ + +/// SNMP request +class Request: public Ipc::Request +{ +public: + Request(int aRequestorId, unsigned int aRequestId, const Pdu& aPdu, + const Session& aSession, int aFd, const Ip::Address& anAddress); + + explicit Request(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() + /* Ipc::Request API */ + virtual void pack(Ipc::TypedMsgHdr& msg) const; + virtual Pointer clone() const; + +private: + Request(const Request& request); + +public: + Pdu pdu; ///< SNMP protocol data unit + Session session; ///< SNMP session + int fd; ///< client connection descriptor + Ip::Address address; ///< client address +}; + + +} // namespace Snmp + +#endif /* SQUID_SNMPX_REQUEST_H */ === added file 'src/snmp/Response.cc' --- src/snmp/Response.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Response.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,51 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "ipc/Messages.h" +#include "ipc/TypedMsgHdr.h" +#include "snmp/Response.h" + + +std::ostream& Snmp::operator << (std::ostream& os, const Response& response) +{ + os << "response: {requestId: " << response.requestId << '}'; + return os; +} + +Snmp::Response::Response(unsigned int aRequestId): + Ipc::Response(aRequestId), pdu() +{ +} + +Snmp::Response::Response(const Response& response): + Ipc::Response(response.requestId), pdu(response.pdu) +{ +} + +Snmp::Response::Response(const Ipc::TypedMsgHdr& msg): + Ipc::Response(0) +{ + msg.checkType(Ipc::mtSnmpResponse); + msg.getPod(requestId); + pdu.unpack(msg); +} + +void +Snmp::Response::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.setType(Ipc::mtSnmpResponse); + msg.putPod(requestId); + pdu.pack(msg); +} + +Ipc::Response::Pointer +Snmp::Response::clone() const +{ + return new Response(*this); +} === added file 'src/snmp/Response.h' --- src/snmp/Response.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Response.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,40 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_RESPONSE_H +#define SQUID_SNMPX_RESPONSE_H + +#include "ipc/forward.h" +#include "ipc/Response.h" +#include "snmp/Pdu.h" +#include + +namespace Snmp +{ + +/// +class Response: public Ipc::Response +{ +public: + Response(unsigned int aRequestId); + explicit Response(const Ipc::TypedMsgHdr& msg); ///< from recvmsg() + /* Ipc::Response API */ + virtual void pack(Ipc::TypedMsgHdr& msg) const; + virtual Ipc::Response::Pointer clone() const; + +private: + Response(const Response& response); + +public: + Pdu pdu; ///< SNMP protocol data unit +}; + +extern std::ostream& operator << (std::ostream& os, const Response& response); + +} // namespace Snmp + +#endif /* SQUID_SNMPX_RESPONSE_H */ === added file 'src/snmp/Session.cc' --- src/snmp/Session.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Session.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,112 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "ipc/TypedMsgHdr.h" +#include "protos.h" +#include "snmp/Session.h" + + +Snmp::Session::Session() +{ + clear(); +} + +Snmp::Session::Session(const Session& session) +{ + assign(session); +} + +Snmp::Session::~Session() +{ + free(); +} + +Snmp::Session& +Snmp::Session::operator = (const Session& session) +{ + free(); + assign(session); + return *this; +} + +void +Snmp::Session::clear() +{ + xmemset(this, 0, sizeof(*this)); +} + +void +Snmp::Session::free() +{ + if (community_len > 0) { + Must(community != NULL); + xfree(community); + } + if (peername != NULL) + xfree(peername); + clear(); +} + +void +Snmp::Session::assign(const Session& session) +{ + memcpy(this, &session, sizeof(*this)); + if (session.community != NULL) { + community = (u_char*)xstrdup((char*)session.community); + Must(community != NULL); + } + if (session.peername != NULL) { + peername = xstrdup(session.peername); + Must(peername != NULL); + } +} + +void +Snmp::Session::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.putPod(Version); + msg.putInt(community_len); + if (community_len > 0) { + Must(community != NULL); + msg.putFixed(community, community_len); + } + msg.putPod(retries); + msg.putPod(timeout); + int len = peername != NULL ? strlen(peername) : 0; + msg.putInt(len); + if (len > 0) + msg.putFixed(peername, len); + msg.putPod(remote_port); + msg.putPod(local_port); +} + +void +Snmp::Session::unpack(const Ipc::TypedMsgHdr& msg) +{ + free(); + msg.getPod(Version); + community_len = msg.getInt(); + if (community_len > 0) { + community = static_cast(xmalloc(community_len + 1)); + Must(community != NULL); + msg.getFixed(community, community_len); + community[community_len] = 0; + } + msg.getPod(retries); + msg.getPod(timeout); + int len = msg.getInt(); + if (len > 0) { + peername = static_cast(xmalloc(len + 1)); + Must(peername != NULL); + msg.getFixed(peername, len); + peername[len] = 0; + } + msg.getPod(remote_port); + msg.getPod(local_port); +} === added file 'src/snmp/Session.h' --- src/snmp/Session.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Session.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,40 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_SESSION_H +#define SQUID_SNMPX_SESSION_H + +#include "config.h" +#include "ipc/forward.h" +#include "snmp.h" +#include "snmp_session.h" + + +namespace Snmp +{ + +/// snmp_session wrapper add pack/unpack feature +class Session: public snmp_session +{ +public: + Session(); + Session(const Session& session); + Session& operator = (const Session& session); + ~Session(); + + void pack(Ipc::TypedMsgHdr& msg) const; ///< prepare for sendmsg() + void unpack(const Ipc::TypedMsgHdr& msg); ///< restore struct from the message + void clear(); ///< clear internal members + +private: + void free(); ///< free internal members + void assign(const Session& session); ///< perform full assignment +}; + +} // namespace Snmp + +#endif /* SQUID_SNMPX_SESSION_H */ === added file 'src/snmp/Var.cc' --- src/snmp/Var.cc 1970-01-01 00:00:00 +0000 +++ src/snmp/Var.cc 2011-01-31 11:32:01 +0000 @@ -0,0 +1,360 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#include "config.h" +#include "base/TextException.h" +#include "ipc/TypedMsgHdr.h" +#include "protos.h" +#include "snmp/Var.h" + + +Snmp::Var::Var() +{ + init(); +} + +Snmp::Var::Var(const Var& var) +{ + init(); + assign(var); +} + +Snmp::Var::~Var() +{ + clear(); +} + +Snmp::Var& +Snmp::Var::operator = (const Var& var) +{ + clear(); + assign(var); + return *this; +} + +void +Snmp::Var::init() +{ + xmemset(this, 0, sizeof(*this)); +} + +Snmp::Var& +Snmp::Var::operator += (const Var& var) +{ + switch(type) + { + case SMI_INTEGER: + setInt(asInt() + var.asInt()); + break; + case SMI_GAUGE32: + setGauge(asGauge() + var.asGauge()); + break; + case SMI_COUNTER32: + setCounter(asCounter() + var.asCounter()); + break; + case SMI_COUNTER64: + setCounter64(asCounter64() + var.asCounter64()); + break; + case SMI_TIMETICKS: + setTimeTicks(asTimeTicks() + var.asTimeTicks()); + break; + default: + debugs(49, DBG_CRITICAL, HERE << "Unsupported type: " << type); + throw TexcHere("Unsupported type"); + break; + } + return *this; +} + +Snmp::Var& +Snmp::Var::operator /= (int num) +{ + Must(num != 0); + switch(type) + { + case SMI_INTEGER: + setInt(asInt() / num); + break; + case SMI_GAUGE32: + setGauge(asGauge() / num); + break; + case SMI_COUNTER32: + setCounter(asCounter() / num); + break; + case SMI_COUNTER64: + setCounter64(asCounter64() / num); + break; + case SMI_TIMETICKS: + setTimeTicks(asTimeTicks() / num); + break; + default: + debugs(49, DBG_CRITICAL, HERE << "Unsupported type: " << type); + throw TexcHere("Unsupported type"); + break; + } + return *this; +} + +bool +Snmp::Var::operator < (const Var& var) const +{ + switch(type) + { + case SMI_INTEGER: + return asInt() < var.asInt(); + case SMI_GAUGE32: + return asGauge() < var.asGauge(); + case SMI_COUNTER32: + return asCounter() < var.asCounter(); + case SMI_COUNTER64: + return asCounter64() < var.asCounter64(); + case SMI_TIMETICKS: + return asTimeTicks() < var.asTimeTicks(); + default: + debugs(49, DBG_CRITICAL, HERE << "Unsupported type: " << type); + throw TexcHere("Unsupported type"); + break; + } + return false; // unreachable +} + +bool +Snmp::Var::operator > (const Var& var) const +{ + switch(type) + { + case SMI_INTEGER: + return asInt() > var.asInt(); + case SMI_GAUGE32: + return asGauge() > var.asGauge(); + case SMI_COUNTER32: + return asCounter() > var.asCounter(); + case SMI_COUNTER64: + return asCounter64() > var.asCounter64(); + case SMI_TIMETICKS: + return asTimeTicks() > var.asTimeTicks(); + default: + debugs(49, DBG_CRITICAL, HERE << "Unsupported type: " << type); + throw TexcHere("Unsupported type"); + break; + } + return false; // unreachable +} + +void +Snmp::Var::assign(const Var& var) +{ + setName(var.getName()); + copyValue(var); +} + +void +Snmp::Var::clearName() +{ + if (name != NULL) { + xfree(name); + name = NULL; + } + name_length = 0; +} + +Range +Snmp::Var::getName() const +{ + return Range(name, name + name_length); +} + +void +Snmp::Var::setName(const Range& aName) +{ + clearName(); + if (aName.start != NULL && aName.size() != 0) { + name_length = aName.size(); + name = static_cast(xmalloc(name_length * sizeof(oid))); + std::copy(aName.start, aName.end, name); + } +} + +void +Snmp::Var::clearValue() +{ + if (val.string != NULL) { + xfree(val.string); + val.string = NULL; + } + val_len = 0; + type = 0; +} + +bool +Snmp::Var::isNull() const +{ + return type == SMI_NULLOBJ; +} + +int +Snmp::Var::asInt() const +{ + Must(type == SMI_INTEGER); + Must(val.integer != NULL && val_len == sizeof(int)); + return *val.integer; +} + +unsigned int +Snmp::Var::asGauge() const +{ + Must(type == SMI_GAUGE32); + Must(val.integer != NULL && val_len == 4); + return *reinterpret_cast(val.integer); +} + +int +Snmp::Var::asCounter() const +{ + Must(type == SMI_COUNTER32); + Must(val.integer != NULL && val_len == 4); + return *reinterpret_cast(val.integer); +} + +long long int +Snmp::Var::asCounter64() const +{ + Must(type == SMI_COUNTER64); + Must(val.integer != NULL && val_len == 8); + return *reinterpret_cast(val.integer); +} + +unsigned int +Snmp::Var::asTimeTicks() const +{ + Must(type == SMI_TIMETICKS); + Must(val.integer != NULL && val_len == sizeof(unsigned int)); + return *reinterpret_cast(val.integer); +} + +Range +Snmp::Var::asObject() const +{ + Must(type == SMI_OBJID); + Must(val_len % sizeof(oid) == 0); + int length = val_len / sizeof(oid); + Must(val.objid != NULL && length > 0); + return Range(val.objid, val.objid + length); +} + +Range +Snmp::Var::asString() const +{ + Must(type == SMI_STRING); + Must(val.string != NULL && val_len > 0); + return Range(val.string, val.string + val_len); +} + +void +Snmp::Var::setInt(int value) +{ + setValue(&value, sizeof(value), SMI_INTEGER); +} + +void +Snmp::Var::setCounter(int value) +{ + setValue(&value, sizeof(value), SMI_COUNTER32); +} + +void +Snmp::Var::setGauge(unsigned int value) +{ + setValue(&value, sizeof(value), SMI_GAUGE32); +} + +void +Snmp::Var::setString(const Range& string) +{ + setValue(string.start, string.size(), SMI_STRING); +} + +void +Snmp::Var::setObject(const Range& object) +{ + setValue(object.start, object.size() * sizeof(oid), SMI_OBJID); +} + +void +Snmp::Var::setCounter64(long long int counter) +{ + setValue(&counter, sizeof(counter), SMI_COUNTER64); +} + +void +Snmp::Var::setTimeTicks(unsigned int ticks) +{ + setValue(&ticks, sizeof(ticks), SMI_TIMETICKS); +} + +void +Snmp::Var::copyValue(const Var& var) +{ + setValue(var.val.string, var.val_len, var.type); +} + +void +Snmp::Var::setValue(const void* value, int length, int aType) +{ + clearValue(); + if (value != NULL) { + Must(length > 0 && aType > 0); + val.string = static_cast(xmalloc(length)); + memcpy(val.string, value, length); + } + val_len = length; + type = aType; +} + +void +Snmp::Var::clear() +{ + clearName(); + clearValue(); + init(); +} + +void +Snmp::Var::pack(Ipc::TypedMsgHdr& msg) const +{ + msg.putInt(name_length); + if (name_length > 0) { + Must(name != NULL); + msg.putFixed(name, name_length * sizeof(oid)); + } + msg.putPod(type); + msg.putPod(val_len); + if (val_len > 0) { + Must(val.string != NULL); + msg.putFixed(val.string, val_len); + } +} + +void +Snmp::Var::unpack(const Ipc::TypedMsgHdr& msg) +{ + clearName(); + clearValue(); + name_length = msg.getInt(); + Must(name_length >= 0); + if (name_length > 0) { + name = static_cast(xmalloc(name_length * sizeof(oid))); + msg.getFixed(name, name_length * sizeof(oid)); + } + msg.getPod(type); + val_len = msg.getInt(); + Must(val_len >= 0); + if (val_len > 0) { + val.string = static_cast(xmalloc(val_len)); + msg.getFixed(val.string, val_len); + } +} === added file 'src/snmp/Var.h' --- src/snmp/Var.h 1970-01-01 00:00:00 +0000 +++ src/snmp/Var.h 2011-01-31 11:32:01 +0000 @@ -0,0 +1,72 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMPX_VAR_H +#define SQUID_SNMPX_VAR_H + +#include "config.h" +#include "ipc/forward.h" +#include "Range.h" +#include "snmp_vars.h" + + +namespace Snmp +{ + +/// variable_list wrapper implement the feature to change +/// the name/value of variable and to pack/unpack message +class Var: public variable_list +{ +public: + Var(); + Var(const Var& var); + Var& operator = (const Var& var); + ~Var(); + + Var& operator += (const Var& var); + Var& operator /= (int num); + bool operator < (const Var& var) const; + bool operator > (const Var& var) const; + + void pack(Ipc::TypedMsgHdr& msg) const; ///< prepare for sendmsg() + void unpack(const Ipc::TypedMsgHdr& msg); ///< restore struct from the message + + Range getName() const; ///< returns variable name + void setName(const Range& aName); ///< set new variable name + void clearName(); ///< clear variable name + + bool isNull() const; + + int asInt() const; ///< returns variable value as integer + unsigned int asGauge() const; ///< returns variable value as unsigned int + int asCounter() const; ///< returns variable value as Counter32 + long long int asCounter64() const; ///< returns variable value as Counter64 + unsigned int asTimeTicks() const; ///< returns variable value as time ticks + Range asObject() const; ///< returns variable value as object oid + Range asString() const; ///< returns variable value as chars string + + void setInt(int value); ///< assign int value to variable + void setCounter(int value); ///< assign Counter32 value to variable + void setGauge(unsigned int value); ///< assign unsigned int value to variable + void setString(const Range& string); ///< assign string to variable + void setObject(const Range& object); ///< assign object oid to variable + void setTimeTicks(unsigned int ticks); ///from); snmp_free_pdu(PDU); } if (Community) xfree(Community); } /* * Packet OK, ACL Check OK, Create reponse. */ static void snmpConstructReponse(snmp_request_t * rq) { struct snmp_pdu *RespPDU; debugs(49, 5, "snmpConstructReponse: Called."); + + if (UsingSmp() && IamWorkerProcess()) { + AsyncJob::Start(new Snmp::Forwarder(static_cast(*rq->PDU), + static_cast(rq->session), rq->sock, rq->from)); + snmp_free_pdu(rq->PDU); + return; + } + RespPDU = snmpAgentResponse(rq->PDU); snmp_free_pdu(rq->PDU); if (RespPDU != NULL) { snmp_build(&rq->session, RespPDU, rq->outbuf, &rq->outlen); comm_udp_sendto(rq->sock, rq->from, rq->outbuf, rq->outlen); snmp_free_pdu(RespPDU); } } /* * Decide how to respond to the request, construct a response and * return the response to the requester. */ -static struct snmp_pdu * +struct snmp_pdu * snmpAgentResponse(struct snmp_pdu *PDU) { struct snmp_pdu *Answer = NULL; debugs(49, 5, "snmpAgentResponse: Called."); if ((Answer = snmp_pdu_create(SNMP_PDU_RESPONSE))) { Answer->reqid = PDU->reqid; Answer->errindex = 0; if (PDU->command == SNMP_PDU_GET || PDU->command == SNMP_PDU_GETNEXT) { /* Indirect way */ int get_next = (PDU->command == SNMP_PDU_GETNEXT); variable_list *VarPtr_; variable_list **RespVars = &(Answer->variables); oid_ParseFn *ParseFn; int index = 0; /* Loop through all variables */ for (VarPtr_ = PDU->variables; VarPtr_; VarPtr_ = VarPtr_->next_variable) { @@ -648,40 +648,63 @@ mibTreeEntry = mib_tree_head; if (Current[count] == mibTreeEntry->name[count]) { count++; while ((mibTreeEntry) && (count < CurrentLen) && (!mibTreeEntry->parsefunction)) { mibTreeEntry = snmpTreeEntry(Current[count], count, mibTreeEntry); count++; } } if (mibTreeEntry && mibTreeEntry->parsefunction) Fn = mibTreeEntry->parsefunction; debugs(49, 5, "snmpTreeGet: return"); return (Fn); } +AggrType +snmpAggrType(oid* Current, snint CurrentLen) +{ + debugs(49, 5, HERE); + + mib_tree_entry* mibTreeEntry = mib_tree_head; + AggrType type = atNone; + int count = 0; + + if (Current[count] == mibTreeEntry->name[count]) { + count++; + + while (mibTreeEntry != NULL && count < CurrentLen) { + mibTreeEntry = snmpTreeEntry(Current[count], count, mibTreeEntry); + if (mibTreeEntry != NULL) + type = mibTreeEntry->aggrType; + count++; + } + } + + return type; +} + static oid_ParseFn * snmpTreeNext(oid * Current, snint CurrentLen, oid ** Next, snint * NextLen) { oid_ParseFn *Fn = NULL; mib_tree_entry *mibTreeEntry = NULL, *nextoid = NULL; int count = 0; debugs(49, 5, "snmpTreeNext: Called"); MemBuf tmp; debugs(49, 6, "snmpTreeNext: Current : " << snmpDebugOid(Current, CurrentLen, tmp)); mibTreeEntry = mib_tree_head; if (Current[count] == mibTreeEntry->name[count]) { count++; while ((mibTreeEntry) && (count < CurrentLen) && (!mibTreeEntry->parsefunction)) { mib_tree_entry *nextmibTreeEntry = snmpTreeEntry(Current[count], count, mibTreeEntry); @@ -1015,91 +1038,92 @@ *nl = 0; char *s = xstrdup(str); char *s_ = s; /* Parse the OID string into oid bits */ while ( (p = strsep(&s_, delim)) != NULL) { *name = (oid*)xrealloc(*name, sizeof(oid) * ((*nl) + 1)); (*name)[*nl] = atoi(p); (*nl)++; } xfree(s); return 1; } /* * Create an entry. Return a pointer to the newly created node, or NULL * on failure. */ static mib_tree_entry * -snmpAddNodeStr(const char *base_str, int o, oid_ParseFn * parsefunction, instance_Fn * instancefunction) +snmpAddNodeStr(const char *base_str, int o, oid_ParseFn * parsefunction, instance_Fn * instancefunction, AggrType aggrType) { mib_tree_entry *m, *b; oid *n; int nl; char s[1024]; /* Find base node */ b = snmpLookupNodeStr(mib_tree_head, base_str); if (! b) return NULL; debugs(49, 5, "snmpAddNodeStr: " << base_str << ": -> " << b); /* Create OID string for new entry */ snprintf(s, 1024, "%s.%d", base_str, o); if (! snmpCreateOidFromStr(s, &n, &nl)) return NULL; /* Create a node */ - m = snmpAddNode(n, nl, parsefunction, instancefunction, 0); + m = snmpAddNode(n, nl, parsefunction, instancefunction, aggrType, 0); /* Link it into the existing tree */ snmpAddNodeChild(b, m); /* Return the node */ return m; } /* * Adds a node to the MIB tree structure and adds the appropriate children */ static mib_tree_entry * -snmpAddNode(oid * name, int len, oid_ParseFn * parsefunction, instance_Fn * instancefunction, int children,...) +snmpAddNode(oid * name, int len, oid_ParseFn * parsefunction, instance_Fn * instancefunction, AggrType aggrType, int children,...) { va_list args; int loop; mib_tree_entry *entry = NULL; va_start(args, children); MemBuf tmp; debugs(49, 6, "snmpAddNode: Children : " << children << ", Oid : " << snmpDebugOid(name, len, tmp)); va_start(args, children); entry = (mib_tree_entry *)xmalloc(sizeof(mib_tree_entry)); entry->name = name; entry->len = len; entry->parsefunction = parsefunction; entry->instancefunction = instancefunction; entry->children = children; entry->leaves = NULL; + entry->aggrType = aggrType; if (children > 0) { entry->leaves = (mib_tree_entry **)xmalloc(sizeof(mib_tree_entry *) * children); for (loop = 0; loop < children; loop++) { entry->leaves[loop] = va_arg(args, mib_tree_entry *); entry->leaves[loop]->parent = entry; } } return (entry); } /* End of tree utility functions */ /* * Returns the list of parameters in an oid */ static oid * snmpCreateOid(int length,...) { === added file 'src/snmp_core.h' --- src/snmp_core.h 1970-01-01 00:00:00 +0000 +++ src/snmp_core.h 2011-01-17 15:10:36 +0000 @@ -0,0 +1,38 @@ +/* + * $Id$ + * + * DEBUG: section 49 SNMP Interface + * + */ + +#ifndef SQUID_SNMP_CORE_H +#define SQUID_SNMP_CORE_H + +#include "config.h" +#include "cache_snmp.h" + +#define SNMP_REQUEST_SIZE 4096 +#define MAX_PROTOSTAT 5 + + +typedef struct _mib_tree_entry mib_tree_entry; +typedef oid *(instance_Fn) (oid * name, snint * len, mib_tree_entry * current, oid_ParseFn ** Fn); +typedef enum {atNone = 0, atSum, atAverage, atMax, atMin} AggrType; + +struct _mib_tree_entry { + oid *name; + int len; + oid_ParseFn *parsefunction; + instance_Fn *instancefunction; + int children; + + struct _mib_tree_entry **leaves; + + struct _mib_tree_entry *parent; + AggrType aggrType; +}; + +extern struct snmp_pdu* snmpAgentResponse(struct snmp_pdu* PDU); +extern AggrType snmpAggrType(oid* Current, snint CurrentLen); + +#endif /* SQUID_SNMP_CORE_H */