/*############################################################################## HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. ############################################################################## */ #include #include #include #include #include #include #include "jisem.hpp" #include "udplib.hpp" #include "udptopo.hpp" #include "ccd.hpp" #include "ccddebug.hpp" #include "ccdquery.hpp" #include "ccdstate.hpp" #include "ccdqueue.ipp" #include "ccdsnmp.hpp" #ifdef _USE_CPPUNIT #include #endif using roxiemem::OwnedRoxieRow; using roxiemem::OwnedConstRoxieRow; using roxiemem::IRowManager; using roxiemem::DataBuffer; //============================================================================================ RoxiePacketHeader::RoxiePacketHeader(const RemoteActivityId &_remoteId, ruid_t _uid, unsigned _channel, unsigned _overflowSequence) { packetlength = sizeof(RoxiePacketHeader); #ifdef TIME_PACKETS tick = 0; #endif init(_remoteId, _uid, _channel, _overflowSequence); } RoxiePacketHeader::RoxiePacketHeader(const RoxiePacketHeader &source, unsigned _activityId, unsigned subChannel) : serverId(source.serverId) { // Used to create the header to send a callback to originating server or an IBYTI to a buddy activityId = _activityId; uid = source.uid; queryHash = source.queryHash; channel = source.channel; overflowSequence = source.overflowSequence; continueSequence = source.continueSequence; if (_activityId >= ROXIE_ACTIVITY_SPECIAL_FIRST && _activityId <= ROXIE_ACTIVITY_SPECIAL_LAST) overflowSequence |= OUTOFBAND_SEQUENCE; // Need to make sure it is not treated as dup of actual reply in the udp layer retries = getSubChannelMask(subChannel) | (source.retries & ~ROXIE_RETRIES_MASK); #ifdef TIME_PACKETS tick = source.tick; #endif packetlength = sizeof(RoxiePacketHeader); } unsigned RoxiePacketHeader::getSubChannelMask(unsigned subChannel) { return SUBCHANNEL_MASK << (SUBCHANNEL_BITS * subChannel); } unsigned RoxiePacketHeader::priorityHash() const { // Used to determine which agent to act as primary and which as secondary for a given packet (thus spreading the load) // It's important that we do NOT include channel (since that would result in different values for the different agents responding to a broadcast) // We also don't include continueSequence since we'd prefer continuations to go the same way as original unsigned hash = serverId.hash(); hash = hashc((const unsigned char *) &uid, sizeof(uid), hash); hash += overflowSequence; // MORE - is this better than hashing? if (traceLevel > 9) { StringBuffer s; DBGLOG("Calculating hash: %s hash was %d", toString(s).str(), hash); } return hash; } bool RoxiePacketHeader::matchPacket(const RoxiePacketHeader &oh) const { // used when matching up a kill packet against a pending one... // DO NOT compare activityId - they are not supposed to match, since 0 in activityid identifies ibyti! return oh.uid==uid && (oh.overflowSequence & ~OUTOFBAND_SEQUENCE) == (overflowSequence & ~OUTOFBAND_SEQUENCE) && oh.continueSequence == continueSequence && oh.serverId==serverId && oh.channel==channel; } void RoxiePacketHeader::init(const RemoteActivityId &_remoteId, ruid_t _uid, unsigned _channel, unsigned _overflowSequence) { retries = 0; activityId = _remoteId.activityId; queryHash = _remoteId.queryHash; uid = _uid; serverId = myNode; channel = _channel; overflowSequence = _overflowSequence; continueSequence = 0; } StringBuffer &RoxiePacketHeader::toString(StringBuffer &ret) const { const IpAddress &serverIP = serverId.getNodeAddress(); ret.appendf("uid=" RUIDF " activityId=", uid); switch(activityId & ~ROXIE_PRIORITY_MASK) { case ROXIE_UNLOAD: ret.append("ROXIE_UNLOAD"); break; case ROXIE_PING: ret.append("ROXIE_PING"); break; case ROXIE_TRACEINFO: ret.append("ROXIE_TRACEINFO"); break; case ROXIE_DEBUGREQUEST: ret.append("ROXIE_DEBUGREQUEST"); break; case ROXIE_DEBUGCALLBACK: ret.append("ROXIE_DEBUGCALLBACK"); break; case ROXIE_FILECALLBACK: ret.append("ROXIE_FILECALLBACK"); break; case ROXIE_ALIVE: ret.append("ROXIE_ALIVE"); break; case ROXIE_KEYEDLIMIT_EXCEEDED: ret.append("ROXIE_KEYEDLIMIT_EXCEEDED"); break; case ROXIE_LIMIT_EXCEEDED: ret.append("ROXIE_LIMIT_EXCEEDED"); break; case ROXIE_EXCEPTION: ret.append("ROXIE_EXCEPTION"); break; default: ret.appendf("%u", (activityId & ~(ROXIE_ACTIVITY_FETCH | ROXIE_PRIORITY_MASK))); if (activityId & ROXIE_ACTIVITY_FETCH) ret.appendf(" (fetch part)"); break; } ret.append(" pri="); switch(activityId & ROXIE_PRIORITY_MASK) { case ROXIE_SLA_PRIORITY: ret.append("SLA"); break; case ROXIE_HIGH_PRIORITY: ret.append("HIGH"); break; case ROXIE_LOW_PRIORITY: ret.append("LOW"); break; default: ret.append("???"); break; } ret.appendf(" queryHash=%" I64F "x ch=%u seq=%d cont=%d server=", queryHash, channel, overflowSequence, continueSequence); serverIP.getIpText(ret); if (retries) { if (retries==QUERY_ABORTED) ret.append(" retries=QUERY_ABORTED"); else { if (retries & ROXIE_RETRIES_MASK) ret.appendf(" retries=%04x", retries); if (retries & ROXIE_FASTLANE) ret.appendf(" FASTLANE"); if (retries & ROXIE_BROADCAST) ret.appendf(" BROADCAST"); } } return ret; } bool RoxiePacketHeader::allChannelsFailed() { unsigned mask = (1 << (getNumAgents(channel) * SUBCHANNEL_BITS)) - 1; return (retries & mask) == mask; } bool RoxiePacketHeader::retry() { bool worthRetrying = false; unsigned mask = SUBCHANNEL_MASK; unsigned numAgents = getNumAgents(channel); for (unsigned subChannel = 0; subChannel < numAgents; subChannel++) { unsigned subRetries = (retries & mask) >> (subChannel * SUBCHANNEL_BITS); if (subRetries != SUBCHANNEL_MASK) subRetries++; if (subRetries != SUBCHANNEL_MASK) worthRetrying = true; retries = (retries & ~mask) | (subRetries << (subChannel * SUBCHANNEL_BITS)); mask <<= SUBCHANNEL_BITS; } return worthRetrying; } void RoxiePacketHeader::setException(unsigned subChannel) { retries |= SUBCHANNEL_MASK << (SUBCHANNEL_BITS * subChannel); } unsigned RoxiePacketHeader::thisChannelRetries(unsigned subChannel) { unsigned shift = SUBCHANNEL_BITS * subChannel; unsigned mask = SUBCHANNEL_MASK << shift; return (retries & mask) >> shift; } //============================================================================================ unsigned getReplicationLevel(unsigned channel) { if (!channel) return 0; Owned topology = getTopology(); return topology->queryChannelInfo(channel).replicationLevel(); } //============================================================================================ // This function maps a agent number to the multicast ip used to talk to it. IpAddress multicastBase("239.1.1.1"); // TBD IPv6 (need IPv6 multicast addresses? IpAddress multicastLast("239.1.5.254"); const IpAddress &getChannelIp(IpAddress &ip, unsigned _channel) { // need to be careful to avoid the .0's and the .255's (not sure why...) ip = multicastBase; if (!ip.ipincrement(_channel,1,254,1,0xffff) ||(ip.ipcompare(multicastLast)>0)) throw MakeStringException(ROXIE_MULTICAST_ERROR, "Out-of-range multicast channel %d", _channel); return ip; } static Owned multicastSocket; void joinMulticastChannel(unsigned channel) { IpAddress multicastIp; getChannelIp(multicastIp, channel); SocketEndpoint ep(ccdMulticastPort, multicastIp); StringBuffer epStr; ep.getUrlStr(epStr); if (!multicastSocket->join_multicast_group(ep)) throw MakeStringException(ROXIE_MULTICAST_ERROR, "Failed to join multicast channel %d (%s)", channel, epStr.str()); if (traceLevel) DBGLOG("Joined multicast channel %d (%s)", channel, epStr.str()); } static SocketEndpointArray multicastEndpoints; // indexed by channel void setMulticastEndpoints(unsigned numChannels) { for (unsigned channel = 0; channel <= numChannels; channel++) // NOTE - channel 0 is special, and numChannels does not include it { IpAddress multicastIp; getChannelIp(multicastIp, channel); multicastEndpoints.append(SocketEndpoint(ccdMulticastPort, multicastIp)); } } void openMulticastSocket() { if (!multicastSocket) { multicastSocket.setown(ISocket::udp_create(ccdMulticastPort)); if (multicastTTL) { multicastSocket->set_ttl(multicastTTL); DBGLOG("Roxie: multicastTTL: %u", multicastTTL); } else DBGLOG("Roxie: multicastTTL not set"); multicastSocket->set_receive_buffer_size(udpMulticastBufferSize); size32_t actualSize = multicastSocket->get_receive_buffer_size(); if (actualSize < udpMulticastBufferSize) { DBGLOG("Roxie: multicast socket buffer size could not be set (requested=%d actual %d", udpMulticastBufferSize, actualSize); throwUnexpected(); } if (traceLevel) DBGLOG("Roxie: multicast socket created port=%d sockbuffsize=%d actual %d", ccdMulticastPort, udpMulticastBufferSize, actualSize); if (roxieMulticastEnabled && !localAgent) { Owned topology = getTopology(); for (unsigned channel : topology->queryChannels()) { assertex(channel); joinMulticastChannel(channel); } joinMulticastChannel(0); // all agents also listen on channel 0 } } } void closeMulticastSockets() { multicastSocket.clear(); } size32_t channelWrite(unsigned channel, void const* buf, size32_t size) { size32_t minwrote = 0; if (roxieMulticastEnabled) { return multicastSocket->udp_write_to(multicastEndpoints.item(channel), buf, size); } else { Owned topo = getTopology(); const SocketEndpointArray &eps = topo->queryAgents(channel); if (!eps.ordinality()) throw makeStringExceptionV(0, "No agents available for channel %d", channel); ForEachItemIn(idx, eps) { size32_t wrote = multicastSocket->udp_write_to(eps.item(idx), buf, size); if (!idx || wrote < minwrote) minwrote = wrote; } } return minwrote; } //============================================================================================ class CRoxieQueryPacket : implements IRoxieQueryPacket, public CInterface { protected: RoxiePacketHeader *data; const byte *continuationData; unsigned continuationLength; const byte *smartStepInfoData; unsigned smartStepInfoLength; const byte *contextData; unsigned contextLength; const byte *traceInfo; unsigned traceLength; public: IMPLEMENT_IINTERFACE; CRoxieQueryPacket(const void *_data, int lengthRemaining) : data((RoxiePacketHeader *) _data) { assertex(lengthRemaining >= (int) sizeof(RoxiePacketHeader)); data->packetlength = lengthRemaining; const byte *finger = (const byte *) (data + 1); lengthRemaining -= sizeof(RoxiePacketHeader); if (data->activityId == ROXIE_FILECALLBACK || data->activityId == ROXIE_DEBUGCALLBACK || data->retries == QUERY_ABORTED) { continuationData = NULL; continuationLength = 0; smartStepInfoData = NULL; smartStepInfoLength = 0; traceInfo = NULL; traceLength = 0; } else { if (data->continueSequence & ~CONTINUE_SEQUENCE_SKIPTO) { assertex(lengthRemaining >= (int) sizeof(unsigned short)); continuationLength = *(unsigned short *) finger; continuationData = finger + sizeof(unsigned short); finger = continuationData + continuationLength; lengthRemaining -= continuationLength + sizeof(unsigned short); } else { continuationData = NULL; continuationLength = 0; } if (data->continueSequence & CONTINUE_SEQUENCE_SKIPTO) { assertex(lengthRemaining >= (int) sizeof(unsigned short)); smartStepInfoLength = *(unsigned short *) finger; smartStepInfoData = finger + sizeof(unsigned short); finger = smartStepInfoData + smartStepInfoLength; lengthRemaining -= smartStepInfoLength + sizeof(unsigned short); } else { smartStepInfoData = NULL; smartStepInfoLength = 0; } assertex(lengthRemaining > 1); traceInfo = finger; lengthRemaining--; if (*finger++ & LOGGING_DEBUGGERACTIVE) { assertex(lengthRemaining >= (int) sizeof(unsigned short)); unsigned short debugLen = *(unsigned short *) finger; finger += debugLen + sizeof(unsigned short); lengthRemaining -= debugLen + sizeof(unsigned short); } for (;;) { assertex(lengthRemaining>0); if (!*finger) { lengthRemaining--; finger++; break; } lengthRemaining--; finger++; } traceLength = finger - traceInfo; } assertex(lengthRemaining >= 0); contextData = finger; contextLength = lengthRemaining; } ~CRoxieQueryPacket() { free(data); } virtual RoxiePacketHeader &queryHeader() const { return *data; } virtual const void *queryContinuationData() const { return continuationData; } virtual unsigned getContinuationLength() const { return continuationLength; } virtual const byte *querySmartStepInfoData() const { return smartStepInfoData; } virtual unsigned getSmartStepInfoLength() const { return smartStepInfoLength; } virtual const byte *queryTraceInfo() const { return traceInfo; } virtual unsigned getTraceLength() const { return traceLength; } virtual const void *queryContextData() const { return contextData; } virtual unsigned getContextLength() const { return contextLength; } virtual IRoxieQueryPacket *clonePacket(unsigned channel) const { unsigned length = data->packetlength; RoxiePacketHeader *newdata = (RoxiePacketHeader *) malloc(length); memcpy(newdata, data, length); newdata->channel = channel; newdata->retries |= ROXIE_BROADCAST; return createRoxiePacket(newdata, length); } virtual IRoxieQueryPacket *insertSkipData(size32_t skipDataLen, const void *skipData) const { assertex((data->continueSequence & CONTINUE_SEQUENCE_SKIPTO) == 0); // Should not already be any skipto info in the source packet unsigned newDataSize = data->packetlength + sizeof(unsigned short) + skipDataLen; char *newdata = (char *) malloc(newDataSize); unsigned headSize = sizeof(RoxiePacketHeader); if (data->continueSequence & ~CONTINUE_SEQUENCE_SKIPTO) headSize += sizeof(unsigned short) + continuationLength; memcpy(newdata, data, headSize); // copy in leading part of old data ((RoxiePacketHeader *) newdata)->continueSequence |= CONTINUE_SEQUENCE_SKIPTO; // set flag indicating new data is present *(unsigned short *) (newdata + headSize) = skipDataLen; // add length field for new data memcpy(newdata + headSize + sizeof(unsigned short), skipData, skipDataLen); // copy in new data memcpy(newdata + headSize + sizeof(unsigned short) + skipDataLen, ((char *) data) + headSize, data->packetlength - headSize); // copy in remaining old data return createRoxiePacket(newdata, newDataSize); } virtual unsigned hash() const { // This is used for Roxie server-side caching. The hash includes some of the header and all of the payload. unsigned hash = 0; if (continuationLength) hash = hashc((const unsigned char *) continuationData, continuationLength, hash); if (smartStepInfoLength) hash = hashc((const unsigned char *) smartStepInfoData, smartStepInfoLength, hash); // NOTE - don't hash the trace info! hash = hashc((const unsigned char *) contextData, contextLength, hash); hash = hashc((const unsigned char *) &data->channel, sizeof(data->channel), hash); hash = hashc((const unsigned char *) &data->overflowSequence, sizeof(data->overflowSequence), hash); hash = hashc((const unsigned char *) &data->continueSequence, sizeof(data->continueSequence), hash); // MORE - sequence fields should always be zero for anything we are caching I think... (?) // Note - no point hashing activityId (as cache is local to one activity) or serverIP (likewise) return hash; } virtual bool cacheMatch(const IRoxieQueryPacket *c) const { // note - this checks whether it's a repeat from Roxie server's point-of-view // So fields that are compared are the same as the ones that are hashed.... RoxiePacketHeader &h = c->queryHeader(); if (data->channel == h.channel && data->overflowSequence == h.overflowSequence && data->continueSequence == h.continueSequence) { if (continuationLength) // note - we already checked that sequences match { if (continuationLength != c->getContinuationLength()) return false; if (memcmp(continuationData,c->queryContinuationData(),continuationLength)!=0) return false; } if (smartStepInfoLength) { if (smartStepInfoLength != c->getSmartStepInfoLength()) return false; if (memcmp(smartStepInfoData,c->querySmartStepInfoData(),smartStepInfoLength)!=0) return false; } // NOTE - trace info NOT compared if (contextLength == c->getContextLength() && memcmp(contextData, c->queryContextData(), contextLength)==0) return true; } return false; } }; extern IRoxieQueryPacket *createRoxiePacket(void *_data, unsigned _len) { if ((unsigned short)_len != _len && !localAgent) { StringBuffer s; RoxiePacketHeader *header = (RoxiePacketHeader *) _data; header->toString(s); free(_data); throw MakeStringException(ROXIE_PACKET_ERROR, "Packet length %d exceeded maximum sending packet %s", _len, s.str()); } return new CRoxieQueryPacket(_data, _len); } extern IRoxieQueryPacket *createRoxiePacket(MemoryBuffer &m) { unsigned length = m.length(); // don't make assumptions about evaluation order of parameters... return createRoxiePacket(m.detachOwn(), length); } //================================================================================= AgentContextLogger::AgentContextLogger() { GetHostIp(ip); set(NULL); } AgentContextLogger::AgentContextLogger(IRoxieQueryPacket *packet) { GetHostIp(ip); set(packet); } void AgentContextLogger::set(IRoxieQueryPacket *packet) { anyOutput = false; intercept = false; debuggerActive = false; checkingHeap = false; aborted = false; stats.reset(); start = msTick(); if (packet) { CriticalBlock b(crit); RoxiePacketHeader &header = packet->queryHeader(); const byte *traceInfo = packet->queryTraceInfo(); unsigned traceLength = packet->getTraceLength(); unsigned char loggingFlags = *traceInfo; if (loggingFlags & LOGGING_FLAGSPRESENT) // should always be true.... but this flag is handy to avoid flags byte ever being NULL { traceInfo++; traceLength--; if (loggingFlags & LOGGING_INTERCEPTED) intercept = true; if (loggingFlags & LOGGING_TRACELEVELSET) { ctxTraceLevel = (*traceInfo++ - 1); // avoid null byte here in case anyone still thinks there's just a null-terminated string traceLength--; } if (loggingFlags & LOGGING_BLIND) blind = true; if (loggingFlags & LOGGING_CHECKINGHEAP) checkingHeap = true; if (loggingFlags & LOGGING_DEBUGGERACTIVE) { assertex(traceLength > sizeof(unsigned short)); debuggerActive = true; unsigned short debugLen = *(unsigned short *) traceInfo; traceInfo += debugLen + sizeof(unsigned short); traceLength -= debugLen + sizeof(unsigned short); } // Passing the wuid via the logging context prefix is a lot of a hack... if (loggingFlags & LOGGING_WUID) { unsigned wuidLen = 0; while (wuidLen < traceLength) { if (traceInfo[wuidLen]=='@'||traceInfo[wuidLen]==':') break; wuidLen++; } wuid.set((const char *) traceInfo, wuidLen); } } channel = header.channel; StringBuffer s(traceLength, (const char *) traceInfo); s.append("|"); ip.getIpText(s); s.append(':').append(channel); StringContextLogger::set(s.str()); if (intercept || mergeAgentStatistics) { RoxiePacketHeader newHeader(header, ROXIE_TRACEINFO, 0); // subchannel not relevant output.setown(ROQ->createOutputStream(newHeader, true, *this)); } } else { StringContextLogger::set(""); channel = 0; } } void AgentContextLogger::putStatProcessed(unsigned subGraphId, unsigned actId, unsigned idx, unsigned processed, unsigned strands) const { if (output && mergeAgentStatistics) { MemoryBuffer buf; buf.append((char) LOG_CHILDCOUNT); // A special log entry for the stats buf.append(subGraphId); buf.append(actId); buf.append(idx); buf.append(processed); buf.append(strands); } } void AgentContextLogger::putStats(unsigned subGraphId, unsigned actId, const CRuntimeStatisticCollection &stats) const { if (output && mergeAgentStatistics) { MemoryBuffer buf; buf.append((char) LOG_CHILDSTATS); // A special log entry for the stats buf.append(subGraphId); buf.append(actId); if (stats.serialize(buf)) { unsigned len = buf.length(); void *ret = output->getBuffer(len, true); memcpy(ret, buf.toByteArray(), len); output->putBuffer(ret, len, true); anyOutput = true; } } } void AgentContextLogger::flush() { if (output) { CriticalBlock b(crit); if (mergeAgentStatistics) { MemoryBuffer buf; buf.append((char) LOG_STATVALUES); // A special log entry for the stats if (stats.serialize(buf)) { unsigned len = buf.length(); void *ret = output->getBuffer(len, true); memcpy(ret, buf.toByteArray(), len); output->putBuffer(ret, len, true); anyOutput = true; } } ForEachItemIn(idx, log) { MemoryBuffer buf; LogItem &logItem = log.item(idx); logItem.serialize(buf); unsigned len = buf.length(); void *ret = output->getBuffer(len, true); memcpy(ret, buf.toByteArray(), len); output->putBuffer(ret, len, true); anyOutput = true; } log.kill(); if (anyOutput) output->flush(); output.clear(); } } //================================================================================= static SpinLock onDemandQueriesCrit; static MapXToMyClass onDemandQueryCache; void sendUnloadMessage(hash64_t hash, const char *id, const IRoxieContextLogger &logctx) { unsigned packetSize = sizeof(RoxiePacketHeader) + sizeof(char) + strlen(id) + 1; void *packetData = malloc(packetSize); RoxiePacketHeader *header = (RoxiePacketHeader *) packetData; RemoteActivityId unloadId(ROXIE_UNLOAD, hash); header->init(unloadId, 0, 0, 0); char *finger = (char *) (header + 1); *finger++ = (char) LOGGING_FLAGSPRESENT; strcpy(finger, id); finger += strlen(id)+1; if (traceLevel > 1) DBGLOG("UNLOAD sent for query %s", id); Owned packet = createRoxiePacket(packetData, packetSize); ROQ->sendPacket(packet, logctx); } void doUnload(IRoxieQueryPacket *packet, const IRoxieContextLogger &logctx) { const RoxiePacketHeader &header = packet->queryHeader(); unsigned channelNo = header.channel; logctx.CTXLOG("Unload received for channel %d", channelNo); hash64_t hashValue = header.queryHash; hashValue = rtlHash64Data(sizeof(channelNo), &channelNo, hashValue); SpinBlock b(onDemandQueriesCrit); onDemandQueryCache.remove(hashValue); } void cacheOnDemandQuery(hash64_t hashValue, unsigned channelNo, IQueryFactory *query) { hashValue = rtlHash64Data(sizeof(channelNo), &channelNo, hashValue); SpinBlock b(onDemandQueriesCrit); onDemandQueryCache.setValue(hashValue, query); } //================================================================================= struct PingRecord { unsigned tick; IpAddress senderIP; }; void doPing(IRoxieQueryPacket *packet, const IRoxieContextLogger &logctx) { const RoxiePacketHeader &header = packet->queryHeader(); const IpAddress &serverIP = header.serverId.getNodeAddress(); unsigned contextLength = packet->getContextLength(); if (contextLength != sizeof(PingRecord)) { StringBuffer s; throw MakeStringException(ROXIE_UNKNOWN_SERVER, "Unexpected data size %d (expected %d) in PING: %s", contextLength, (unsigned) sizeof(PingRecord), header.toString(s).str()); } const PingRecord *data = (const PingRecord *) packet->queryContextData(); if (!serverIP.ipequals(data->senderIP)) { StringBuffer s; throw MakeStringException(ROXIE_UNKNOWN_SERVER, "Message received from unknown Roxie server %s", header.toString(s).str()); } RoxiePacketHeader newHeader(header, ROXIE_PING, 0); // subchannel not relevant Owned output = ROQ->createOutputStream(newHeader, true, logctx); void *ret = output->getBuffer(contextLength, false); memcpy(ret, data, contextLength); output->putBuffer(ret, contextLength, false); output->flush(); } //================================================================================= // // RoxieQueue - holds pending transactions on a roxie agent class RoxieQueue : public CInterface, implements IThreadFactory { Owned workers; QueueOf waiting; Semaphore available; CriticalSection qcrit; unsigned headRegionSize; unsigned numWorkers; RelaxedAtomic started; std::atomic idle; void noteQueued() { maxQueueLength.store_max(++queueLength); // NOTE - there is a small race condition here - if idle is 1 but two enqueue's happen // close enough together that the signal has not yet caused idle to come back down to zero, then the // desired new thread may not be created. It's unlikely, and it's benign in that the query is still // processed and the thread will be created next time the HWM is reached. if (started < numWorkers && idle==0) { workers->start(this); started++; } } public: IMPLEMENT_IINTERFACE; RoxieQueue(unsigned _headRegionSize, unsigned _numWorkers) { headRegionSize = _headRegionSize; numWorkers = _numWorkers; workers.setown(createThreadPool("RoxieWorkers", this, NULL, numWorkers)); started = 0; idle = 0; } virtual IPooledThread *createNew(); void abortChannel(unsigned channel); void start() { if (prestartAgentThreads) { while (started < numWorkers) { workers->start(this); started++; } } } IPooledThreadIterator *running() { return workers->running(); } void stopAll() { workers->stopAll(true); signal(workers->runningCount()); } void join() { workers->joinAll(true); workers.clear(); // Breaks a cyclic reference count that would stop us from releasing RoxieReceiverThread otherwise } void enqueue(IRoxieQueryPacket *x) { { #ifdef TIME_PACKETS x->queryHeader().tick = msTick(); #endif CriticalBlock qc(qcrit); waiting.enqueue(x); noteQueued(); } available.signal(); } void enqueueUnique(IRoxieQueryPacket *x, unsigned subChannel) { RoxiePacketHeader &header = x->queryHeader(); #ifdef TIME_PACKETS header.tick = msTick(); #endif bool found = false; { CriticalBlock qc(qcrit); unsigned len = waiting.ordinality(); unsigned i; for (i = 0; i < len; i++) { IRoxieQueryPacket *queued = waiting.item(i); if (queued && queued->queryHeader().matchPacket(header)) { found = true; break; } } if (!found) waiting.enqueue(x); } if (found) { if (traceLevel > 0) { StringBuffer xx; AgentContextLogger l(x); l.CTXLOG("Ignored retry on subchannel %u for queued activity %s", subChannel, header.toString(xx).str()); } if (!subChannel) retriesIgnoredPrm++; else retriesIgnoredSec++; x->Release(); } else { available.signal(); noteQueued(); if (traceLevel > 10) { AgentContextLogger l(x); StringBuffer xx; l.CTXLOG("enqueued %s", header.toString(xx).str()); } } } bool remove(RoxiePacketHeader &x) { unsigned scanLength = 0; IRoxieQueryPacket *found = nullptr; { CriticalBlock qc(qcrit); unsigned len = waiting.ordinality(); unsigned i; for (i = 0; i < len; i++) { IRoxieQueryPacket *queued = waiting.item(i); if (queued) { scanLength++; if (queued->queryHeader().matchPacket(x)) { waiting.set(i, NULL); found = queued; break; } } } } if (found) { #ifdef _DEBUG RoxiePacketHeader &header = found->queryHeader(); AgentContextLogger l(found); StringBuffer xx; l.CTXLOG("discarded %s", header.toString(xx).str()); #endif found->Release(); queueLength--; if (scanLength > maxScanLength) maxScanLength = scanLength; totScanLength += scanLength; totScans++; return true; } else return false; } void wait() { idle++; available.wait(); idle--; } void signal(unsigned num) { available.signal(num); } IRoxieQueryPacket *dequeue() { CriticalBlock qc(qcrit); unsigned lim = waiting.ordinality(); if (lim) { if (headRegionSize) { if (lim > headRegionSize) lim = headRegionSize; return waiting.dequeue(fastRand() % lim); } return waiting.dequeue(); } else return NULL; } unsigned getHeadRegionSize() const { return headRegionSize; } unsigned setHeadRegionSize(unsigned newsize) { unsigned ret = headRegionSize; headRegionSize = newsize; return ret; } }; class CRoxieWorker : public CInterface, implements IPooledThread { RoxieQueue *queue; CriticalSection actCrit; Semaphore ibytiSem; bool stopped; bool abortJob; bool busy; Owned activity; Owned packet; Owned topology; AgentContextLogger logctx; public: IMPLEMENT_IINTERFACE; CRoxieWorker() { queue = NULL; stopped = false; busy = false; abortJob = false; } virtual void init(void *_r) override { queue = (RoxieQueue *) _r; stopped = false; busy = false; abortJob = false; } virtual bool canReuse() const override { return true; } virtual bool stop() override { stopped = true; return true; } inline void setActivity(IRoxieAgentActivity *act) { CriticalBlock b(actCrit); activity.setown(act); } inline bool match(RoxiePacketHeader &h) { // There is a window between getting packet from queue and being able to match it. // This could cause some deduping to fail, but it does not matter if it does (so long as it is rare!) CriticalBlock b(actCrit); return packet && packet->queryHeader().matchPacket(h); } void abortChannel(unsigned channel) { CriticalBlock b(actCrit); if (packet && packet->queryHeader().channel==channel) { abortJob = true; if (doIbytiDelay) ibytiSem.signal(); if (activity) activity->abort(); } } bool checkAbort(RoxiePacketHeader &h, bool checkRank, bool &queryFound, bool &preActivity) { CriticalBlock b(actCrit); if (packet && packet->queryHeader().matchPacket(h)) { queryFound = true; abortJob = true; if (doIbytiDelay) ibytiSem.signal(); if (activity) { // Try to stop/abort a job after it starts only if IBYTI comes from a higher priority agent // (more primary in the rank). The agents with higher rank will hold the lower bits of the retries field in IBYTI packet). if (!checkRank || topology->queryChannelInfo(h.channel).otherAgentHasPriority(h.priorityHash(), h.getRespondingSubChannel())) { activity->abort(); return true; } else { return false; } } if (busy) { preActivity = true; return true; } } return false; } void throwRemoteException(IException *E, IRoxieAgentActivity *activity, IRoxieQueryPacket *packet, bool isUser) { try { if (activity && (logctx.queryTraceLevel() > 1)) { StringBuffer act; activity->toString(act); logctx.CTXLOG("throwRemoteException, activity %s, isUser=%d", act.str(), (int) isUser); if (!isUser) EXCLOG(E, "throwRemoteException"); } RoxiePacketHeader &header = packet->queryHeader(); unsigned mySubChannel = topology->queryChannelInfo(header.channel).subChannel(); // I failed to do the query, but already sent out IBYTI - resend it so someone else can try if (!isUser) { StringBuffer s; s.append("Exception in agent for packet "); header.toString(s); logctx.logOperatorException(E, NULL, 0, "%s", s.str()); header.setException(mySubChannel); if (!header.allChannelsFailed() && !localAgent) { if (logctx.queryTraceLevel() > 1) logctx.CTXLOG("resending packet from agent in case others want to try it"); ROQ->sendPacket(packet, logctx); } } RoxiePacketHeader newHeader(header, ROXIE_EXCEPTION, mySubChannel); if (isUser) newHeader.retries = (unsigned short) -1; Owned output = ROQ->createOutputStream(newHeader, true, logctx); StringBuffer message(""); message.appendf("%d", E->errorCode()); StringBuffer err; E->errorMessage(err); encodeXML(err.str(), message); message.append(""); unsigned len = message.length(); void *ret = output->getBuffer(len+1, true); memcpy(ret, message.str(), len+1); output->putBuffer(ret, len+1, true); output->flush(); E->Release(); } catch (IException *EInE) { EXCLOG(EInE, "Exception during throwRemoteException"); E->Release(); EInE->Release(); } catch (...) { logctx.CTXLOG("Unknown Exception during throwRemoteException"); E->Release(); } } void doActivity() { RoxiePacketHeader &header = packet->queryHeader(); unsigned channel = header.channel; hash64_t queryHash = packet->queryHeader().queryHash; unsigned activityId = packet->queryHeader().activityId & ~ROXIE_PRIORITY_MASK; Owned queryFactory = getQueryFactory(queryHash, channel); unsigned numAgents = topology->queryAgents(channel).ordinality(); unsigned mySubChannel = topology->queryChannelInfo(channel).subChannel(); if (!queryFactory && logctx.queryWuid()) { Owned daliHelper = connectToDali(); Owned wu = daliHelper->attachWorkunit(logctx.queryWuid(), NULL); queryFactory.setown(createAgentQueryFactoryFromWu(wu, channel)); if (queryFactory) cacheOnDemandQuery(queryHash, channel, queryFactory); } if (!queryFactory) { StringBuffer hdr; IException *E = MakeStringException(MSGAUD_operator, ROXIE_UNKNOWN_QUERY, "Roxie agent received request for unregistered query: %s", packet->queryHeader().toString(hdr).str()); EXCLOG(E, "doActivity"); throwRemoteException(E, activity, packet, false); return; } try { if (logctx.queryTraceLevel() > 8) { StringBuffer x; logctx.CTXLOG("IBYTI delay controls : doIbytiDelay=%s numagents=%u subchnl=%u : %s", doIbytiDelay?"YES":"NO", numAgents, topology->queryChannelInfo(channel).subChannel(), header.toString(x).str()); } bool debugging = logctx.queryDebuggerActive(); if (debugging) { if (mySubChannel) abortJob = true; // when debugging, we always run on primary only... } else if (doIbytiDelay && (numAgents > 1)) { unsigned hdrHashVal = header.priorityHash(); unsigned primarySubChannel = (hdrHashVal % numAgents); if (primarySubChannel != mySubChannel) { unsigned delay = topology->queryChannelInfo(channel).getIbytiDelay(primarySubChannel); if (logctx.queryTraceLevel() > 6) { StringBuffer x; logctx.CTXLOG("YES myTurnToDelayIBYTI subchannel=%u delay=%u hash=%u %s", mySubChannel, delay, hdrHashVal, header.toString(x).str()); } // MORE: if we are dealing with a query that was on channel 0, we may want a longer delay // (since the theory about duplicated work not mattering when cluster is idle does not hold up) if (delay) { ibytiSem.wait(delay); if (!abortJob) topology->queryChannelInfo(channel).noteChannelsSick(primarySubChannel); if (logctx.queryTraceLevel() > 8) { StringBuffer x; logctx.CTXLOG("Buddy did%s send IBYTI, updated delay : %s", abortJob ? "" : " NOT", header.toString(x).str()); } } } else { #ifndef NO_IBYTI_DELAYS_COUNT if (!mySubChannel) ibytiNoDelaysPrm++; else ibytiNoDelaysSec++; #endif if (logctx.queryTraceLevel() > 6) { StringBuffer x; logctx.CTXLOG("NOT myTurnToDelayIBYTI subchannel=%u hash=%u %s", mySubChannel, hdrHashVal, header.toString(x).str()); } } } if (abortJob) { CriticalBlock b(actCrit); busy = false; // Keep order - before setActivity below if (logctx.queryTraceLevel() > 5) { StringBuffer x; logctx.CTXLOG("Stop before processing - activity aborted %s", header.toString(x).str()); } return; } if (!debugging) ROQ->sendIbyti(header, logctx, mySubChannel); activitiesStarted++; Owned factory = queryFactory->getAgentActivityFactory(activityId); assertex(factory); setActivity(factory->createActivity(logctx, packet)); Owned output = activity->process(); if (logctx.queryTraceLevel() > 5) { StringBuffer x; logctx.CTXLOG("done processing %s", header.toString(x).str()); } if (output) { activitiesCompleted++; busy = false; // Keep order - before setActivity below setActivity(NULL); // Ensures all stats are merged from child queries etc logctx.flush(); output->flush(); } } catch (IUserException *E) { throwRemoteException(E, activity, packet, true); } catch (IException *E) { if (E->errorCode()!=ROXIE_ABORT_ERROR) throwRemoteException(E, activity, packet, false); else E->Release(); } catch (...) { throwRemoteException(MakeStringException(ROXIE_MULTICAST_ERROR, "Unknown exception"), activity, packet, false); } busy = false; // Keep order - before setActivity below setActivity(NULL); } virtual void threadmain() override { while (!stopped) { try { for (;;) { queue->wait(); if (stopped) break; agentsActive++; maxAgentsActive.store_max(agentsActive); abortJob = false; busy = true; if (doIbytiDelay) ibytiSem.reinit(0U); // Make sure sem is is in no-signaled state packet.setown(queue->dequeue()); if (packet) { queueLength--; RoxiePacketHeader &header = packet->queryHeader(); logctx.set(packet); #ifdef TIME_PACKETS { unsigned now = msTick(); unsigned packetWait = now-header.tick; header.tick = now; packetWaitMax.store_max(packetWait); packetWaitElapsed += packetWait; packetWaitCount++; } #endif topology.setown(getTopology()); if (logctx.queryTraceLevel() > 10) { StringBuffer x; logctx.CTXLOG("dequeued %s", header.toString(x).str()); } if ((header.activityId & ~ROXIE_PRIORITY_MASK) == ROXIE_UNLOAD) { doUnload(packet, logctx); } else if ((header.activityId & ~ROXIE_PRIORITY_MASK) == ROXIE_PING) { doPing(packet, logctx); } else if ((header.activityId & ~ROXIE_PRIORITY_MASK) == ROXIE_DEBUGREQUEST) { // MORE - we need to make sure only executed on primary, and that the proxyId (== pointer to DebugGraphManager) is still valid. // It may be that there is not a lot of point using the pointer - may as well use an non-reused ID and look it up in a global hash table of active ones doDebugRequest(packet, logctx); } else if (header.channel) doActivity(); else throwUnexpected(); // channel 0 requests translated earlier now #ifdef TIME_PACKETS { unsigned now = msTick(); unsigned packetRun = now-header.tick; packetRunMax.store_max(packetRun); packetRunElapsed += packetRun; packetRunCount++; } #endif } busy = false; { CriticalBlock b(actCrit); packet.clear(); topology.clear(); logctx.set(NULL); } agentsActive--; } } catch(IException *E) { CriticalBlock b(actCrit); EXCLOG(E); if (packet) { throwRemoteException(E, NULL, packet, false); packet.clear(); } else E->Release(); topology.clear(); } catch(...) { CriticalBlock b(actCrit); Owned E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unexpected exception in Roxie worker thread"); EXCLOG(E); if (packet) { throwRemoteException(E.getClear(), NULL, packet, false); packet.clear(); } topology.clear(); } } } }; IPooledThread *RoxieQueue::createNew() { return new CRoxieWorker; } void RoxieQueue::abortChannel(unsigned channel) { Owned wi = workers->running(); ForEach(*wi) { CRoxieWorker &w = (CRoxieWorker &) wi->query(); w.abortChannel(channel); } } //================================================================================= class CallbackEntry : implements IPendingCallback, public CInterface { const RoxiePacketHeader &header; StringAttr lfn; InterruptableSemaphore ready; MemoryBuffer data; bool gotData; public: IMPLEMENT_IINTERFACE; CallbackEntry(const RoxiePacketHeader &_header, const char *_lfn) : header(_header), lfn(_lfn) { gotData = false; } virtual bool wait(unsigned msecs) { return ready.wait(msecs); } virtual MemoryBuffer &queryData() { return data; } bool matches(RoxiePacketHeader &cand, const char *_lfn) { return (cand.matchPacket(header) && (!_lfn|| stricmp(_lfn, lfn)==0)); } void doFileCallback(unsigned _len, const void *_data, bool aborted) { // MORE - make sure we call this for whole query abort as well as for callback abort if (aborted) ready.interrupt(MakeStringException(0, "Interrupted")); else if (!gotData) { gotData = true; data.append(_len, _data); ready.signal(); } } }; class RoxieReceiverBase : implements IRoxieOutputQueueManager, public CInterface { protected: #ifdef ROXIE_SLA_LOGIC RoxieQueue slaQueue; #endif RoxieQueue hiQueue; RoxieQueue loQueue; unsigned numWorkers; public: IMPLEMENT_IINTERFACE; #ifdef ROXIE_SLA_LOGIC RoxieReceiverBase(unsigned _numWorkers) : slaQueue(headRegionSize, _numWorkers), hiQueue(headRegionSize, _numWorkers), loQueue(headRegionSize, _numWorkers), numWorkers(_numWorkers) #else RoxieReceiverBase(unsigned _numWorkers) : hiQueue(headRegionSize, _numWorkers), loQueue(headRegionSize, _numWorkers), numWorkers(_numWorkers) #endif { } virtual unsigned getHeadRegionSize() const { return loQueue.getHeadRegionSize(); } virtual void setHeadRegionSize(unsigned newSize) { #ifdef ROXIE_SLA_LOGIC slaQueue.setHeadRegionSize(newSize); #endif hiQueue.setHeadRegionSize(newSize); loQueue.setHeadRegionSize(newSize); } virtual void start() { loQueue.start(); hiQueue.start(); #ifdef ROXIE_SLA_LOGIC slaQueue.start(); #endif } virtual void stop() { loQueue.stopAll(); hiQueue.stopAll(); #ifdef ROXIE_SLA_LOGIC slaQueue.stopAll(); #endif } virtual void join() { loQueue.join(); hiQueue.join(); #ifdef ROXIE_SLA_LOGIC slaQueue.join(); #endif } IArrayOf callbacks; CriticalSection callbacksCrit; virtual IPendingCallback *notePendingCallback(const RoxiePacketHeader &header, const char *lfn) { CriticalBlock b(callbacksCrit); CallbackEntry *callback = new CallbackEntry(header, lfn); callbacks.append(*callback); return callback; } virtual void removePendingCallback(IPendingCallback *goer) { if (goer) { CriticalBlock b(callbacksCrit); callbacks.zap(static_cast(*goer)); } } protected: void doFileCallback(IRoxieQueryPacket *packet) { // This is called on the main agent reader thread so needs to be as fast as possible to avoid lost packets const char *lfn; const char *data; unsigned len; RoxiePacketHeader &header = packet->queryHeader(); if (header.activityId == ROXIE_FILECALLBACK || header.activityId == ROXIE_DEBUGCALLBACK) { lfn = (const char *) packet->queryContextData(); unsigned namelen = strlen(lfn) + 1; data = lfn + namelen; len = packet->getContextLength() - namelen; } else { lfn = data = NULL; // used when query aborted len = 0; } CriticalBlock b(callbacksCrit); ForEachItemIn(idx, callbacks) { CallbackEntry &c = callbacks.item(idx); if (c.matches(header, lfn)) { if (traceLevel > 10) DBGLOG("callback return matched a waiting query"); c.doFileCallback(len, data, header.retries==QUERY_ABORTED); } } } }; #ifdef _MSC_VER #pragma warning ( push ) #pragma warning ( disable: 4355 ) #endif class RoxieThrottledPacketSender : public Thread { TokenBucket &bucket; InterruptableSemaphore queued; Semaphore started; unsigned maxPacketSize; SafeQueueOf queue; class DECL_EXCEPTION StoppedException: public IException, public CInterface { public: IMPLEMENT_IINTERFACE; int errorCode() const { return 0; } StringBuffer & errorMessage(StringBuffer &str) const { return str.append("Stopped"); } MessageAudience errorAudience() const { return MSGAUD_user; } }; void enqueue(IRoxieQueryPacket *packet) { packet->Link(); queue.enqueue(packet); queued.signal(); } IRoxieQueryPacket *dequeue() { queued.wait(); return queue.dequeue(); } public: RoxieThrottledPacketSender(TokenBucket &_bucket, unsigned _maxPacketSize) : Thread("RoxieThrottledPacketSender"), bucket(_bucket), maxPacketSize(_maxPacketSize) { start(); started.wait(); } ~RoxieThrottledPacketSender() { stop(); join(); } virtual int run() { started.signal(); for (;;) { try { Owned packet = dequeue(); RoxiePacketHeader &header = packet->queryHeader(); unsigned length = packet->queryHeader().packetlength; { MTIME_SECTION(queryActiveTimer(), "bucket_wait"); bucket.wait((length / 1024) + 1); } if (channelWrite(header.channel, &header, length) != length) DBGLOG("multicast write wrote too little"); packetsSent++; } catch (StoppedException *E) { E->Release(); break; } catch (IException *E) { EXCLOG(E); E->Release(); } catch (...) { } } return 0; } virtual void sendPacket(IRoxieQueryPacket *x, const IRoxieContextLogger &logctx) { RoxiePacketHeader &header = x->queryHeader(); unsigned length = x->queryHeader().packetlength; assertex (header.activityId & ~ROXIE_PRIORITY_MASK); switch (header.retries & ROXIE_RETRIES_MASK) { case (QUERY_ABORTED & ROXIE_RETRIES_MASK): { StringBuffer s; logctx.CTXLOG("Aborting packet size=%d: %s", length, header.toString(s).str()); } break; default: { StringBuffer s; logctx.CTXLOG("Resending packet size=%d: %s", length, header.toString(s).str()); } break; case 0: if (logctx.queryTraceLevel() > 8) { StringBuffer s; logctx.CTXLOG("Sending packet size=%d: %s", length, header.toString(s).str()); } break; } if (length > maxPacketSize) { StringBuffer s; throw MakeStringException(ROXIE_PACKET_ERROR, "Maximum packet length %d exceeded sending packet %s", maxPacketSize, header.toString(s).str()); } enqueue(x); } void stop() { // bucket.stop(); queued.interrupt(new StoppedException); } }; class RoxieSocketQueueManager : public RoxieReceiverBase { protected: Linked sendManager; Linked receiveManager; Owned throttledPacketSendManager; Owned bucket; unsigned maxPacketSize = 0; std::atomic running = { false }; class ReceiverThread : public Thread { RoxieSocketQueueManager &parent; public: ReceiverThread(RoxieSocketQueueManager &_parent) : Thread("RoxieSocketQueueManager"), parent(_parent) {} int run() { // Raise the priority so ibyti's get through in a timely fashion #if defined( __linux__) || defined(__APPLE__) setLinuxThreadPriority(3); #else adjustPriority(1); #endif return parent.run(); } } readThread; public: RoxieSocketQueueManager(unsigned _numWorkers) : RoxieReceiverBase(_numWorkers), readThread(*this) { maxPacketSize = multicastSocket->get_max_send_size(); if ((maxPacketSize==0)||(maxPacketSize>65535)) maxPacketSize = 65535; } virtual void sendPacket(IRoxieQueryPacket *x, const IRoxieContextLogger &logctx) { if (throttledPacketSendManager) throttledPacketSendManager->sendPacket(x, logctx); else { MTIME_SECTION(queryActiveTimer(), "RoxieSocketQueueManager::sendPacket"); RoxiePacketHeader &header = x->queryHeader(); unsigned length = x->queryHeader().packetlength; assertex (header.activityId & ~ROXIE_PRIORITY_MASK); StringBuffer s; switch (header.retries & ROXIE_RETRIES_MASK) { case (QUERY_ABORTED & ROXIE_RETRIES_MASK): logctx.CTXLOG("Aborting packet size=%d: %s", length, header.toString(s).str()); break; default: logctx.CTXLOG("Resending packet size=%d: %s", length, header.toString(s).str()); break; case 0: if (logctx.queryTraceLevel() > 8) logctx.CTXLOG("Sending packet size=%d: %s", length, header.toString(s).str()); break; } if (length > maxPacketSize) { StringBuffer s; throw MakeStringException(ROXIE_PACKET_ERROR, "Maximum packet length %d exceeded sending packet %s", maxPacketSize, header.toString(s).str()); } if (channelWrite(header.channel, &header, length) != length) logctx.CTXLOG("multicast write wrote too little"); packetsSent++; } } virtual void sendIbyti(RoxiePacketHeader &header, const IRoxieContextLogger &logctx, unsigned subChannel) override { MTIME_SECTION(queryActiveTimer(), "RoxieSocketQueueManager::sendIbyti"); RoxiePacketHeader ibytiHeader(header, header.activityId & ROXIE_PRIORITY_MASK, subChannel); if (logctx.queryTraceLevel() > 8) { StringBuffer s; logctx.CTXLOG("Sending IBYTI packet %s", ibytiHeader.toString(s).str()); } if (channelWrite(header.channel, &ibytiHeader, sizeof(RoxiePacketHeader)) != sizeof(RoxiePacketHeader)) logctx.CTXLOG("sendIbyti wrote too little"); ibytiPacketsSent++; } virtual void sendAbort(RoxiePacketHeader &header, const IRoxieContextLogger &logctx) override { MTIME_SECTION(queryActiveTimer(), "RoxieSocketQueueManager::sendAbort"); RoxiePacketHeader abortHeader(header, header.activityId & ROXIE_PRIORITY_MASK, 0); // subChannel irrelevant - we are about to overwrite retries anyway abortHeader.retries = QUERY_ABORTED; if (logctx.queryTraceLevel() > 8) { StringBuffer s; logctx.CTXLOG("Sending ABORT packet %s", abortHeader.toString(s).str()); } if (channelWrite(header.channel, &abortHeader, sizeof(RoxiePacketHeader)) != sizeof(RoxiePacketHeader)) logctx.CTXLOG("sendAbort wrote too little"); abortsSent++; } virtual void sendAbortCallback(const RoxiePacketHeader &header, const char *lfn, const IRoxieContextLogger &logctx) override { MTIME_SECTION(queryActiveTimer(), "RoxieSocketQueueManager::sendAbortCallback"); RoxiePacketHeader abortHeader(header, ROXIE_FILECALLBACK, 0); // subChannel irrelevant - we are about to overwrite retries anyway abortHeader.retries = QUERY_ABORTED; MemoryBuffer data; data.append(sizeof(abortHeader), &abortHeader).append(lfn); if (logctx.queryTraceLevel() > 5) { StringBuffer s; logctx.CTXLOG("Sending ABORT FILECALLBACK packet %s for file %s", abortHeader.toString(s).str(), lfn); } if (channelWrite(header.channel, data.toByteArray(), data.length()) != data.length()) logctx.CTXLOG("tr->write wrote too little"); abortsSent++; } virtual IMessagePacker *createOutputStream(RoxiePacketHeader &header, bool outOfBand, const IRoxieContextLogger &logctx) { unsigned qnum = outOfBand ? 0 : ((header.retries & ROXIE_FASTLANE) || !fastLaneQueue) ? 1 : 2; if (logctx.queryTraceLevel() > 8) { StringBuffer s; logctx.CTXLOG("Creating Output Stream for reply packet on Q=%d - %s", qnum, header.toString(s).str()); } return sendManager->createMessagePacker(header.uid, header.getSequenceId(), &header, sizeof(RoxiePacketHeader), header.serverId, qnum); } virtual bool replyPending(RoxiePacketHeader &header) { return sendManager->dataQueued(header.uid, header.getSequenceId(), header.serverId); } virtual bool abortCompleted(RoxiePacketHeader &header) { return sendManager->abortData(header.uid, header.getSequenceId(), header.serverId); } bool abortRunning(RoxiePacketHeader &header, RoxieQueue &queue, bool checkRank, bool &preActivity) { bool queryFound = false; bool ret = false; Owned wi = queue.running(); ForEach(*wi) { CRoxieWorker &w = (CRoxieWorker &) wi->query(); if (w.checkAbort(header, checkRank, queryFound, preActivity)) { ret = true; break; } else if (queryFound) { ret = false; break; } } if (!checkRank) { if (traceLevel > 8) DBGLOG("discarding data for aborted query"); ROQ->abortCompleted(header); } return ret; } void doIbyti(RoxiePacketHeader &header, RoxieQueue &queue, const ITopologyServer* topology) { assertex(!localAgent); bool preActivity = false; const ChannelInfo &channelInfo = topology->queryChannelInfo(header.channel); unsigned mySubChannel = channelInfo.subChannel(); if (traceLevel > 10) { IpAddress peer; StringBuffer s, s1; multicastSocket->getPeerAddress(peer).getIpText(s); header.toString(s1); DBGLOG("doIBYTI %s from %s", s1.str(), s.str()); DBGLOG("header.retries=%x header.getSubChannelMask(header.channel)=%x", header.retries, header.getSubChannelMask(mySubChannel)); } if (header.retries == QUERY_ABORTED) { abortRunning(header, queue, false, preActivity); queue.remove(header); if (traceLevel > 10) { StringBuffer s; DBGLOG("Abort activity %s", header.toString(s).str()); } } else { ibytiPacketsReceived++; unsigned subChannel = header.getRespondingSubChannel(); if (subChannel == mySubChannel) { if (traceLevel > 10) DBGLOG("doIBYTI packet was from self"); ibytiPacketsFromSelf++; } else { channelInfo.noteChannelHealthy(subChannel); bool foundInQ = queue.remove(header); if (foundInQ) { if (traceLevel > 10) { StringBuffer s; DBGLOG("Removed activity from Q : %s", header.toString(s).str()); } ibytiPacketsWorked++; return; } if (abortRunning(header, queue, true, preActivity)) { if (preActivity) ibytiPacketsWorked++; else ibytiPacketsHalfWorked++; return; } if (traceLevel > 10) DBGLOG("doIBYTI packet was too late"); ibytiPacketsTooLate++; // meaning either I started and reserve the right to finish, or I finished already } } } void processMessage(MemoryBuffer &mb, RoxiePacketHeader &header, RoxieQueue &queue) { // NOTE - this thread needs to do as little as possible - just read packets and queue them up - otherwise we can get packet loss due to buffer overflow // DO NOT put tracing on this thread except at very high tracelevels! Owned topology = getTopology(); if (!header.channel) { // Turn broadcast packet (channel 0), as early as possible, into non-0 channel packets. // So retries and other communication with Roxie server (which uses non-0 channel numbers) will not cause double work or confusion. // Unfortunately this is bad news for dropping packets const std::vector channels = topology->queryChannels(); Owned packet = createRoxiePacket(mb); for (unsigned i = 1; i < channels.size(); i++) queue.enqueue(packet->clonePacket(channels[i])); header.channel = channels[0]; queue.enqueue(packet.getClear()); return; } unsigned mySubchannel = topology->queryChannelInfo(header.channel).subChannel(); if (header.activityId == ROXIE_FILECALLBACK || header.activityId == ROXIE_DEBUGCALLBACK ) { Owned packet = createRoxiePacket(mb); if (traceLevel > 10) { StringBuffer s; DBGLOG("ROXIE_CALLBACK %s", header.toString(s).str()); } doFileCallback(packet); } else if ((header.activityId & ~ROXIE_PRIORITY_MASK) == 0) doIbyti(header, queue, topology); // MORE - check how fast this is! else { Owned packet = createRoxiePacket(mb); AgentContextLogger logctx(packet); unsigned retries = header.thisChannelRetries(mySubchannel); if (retries) { // MORE - is this fast enough? By the time I am seeing retries I may already be under load. Could move onto a separate thread assertex(header.channel); // should never see a retry on channel 0 if (retries >= SUBCHANNEL_MASK) return; // someone sent a failure or something - ignore it // Send back an out-of-band immediately, to let Roxie server know that channel is still active if (!(testAgentFailure & 0x800)) { RoxiePacketHeader newHeader(header, ROXIE_ALIVE, mySubchannel); Owned output = ROQ->createOutputStream(newHeader, true, logctx); output->flush(); } // If it's a retry, look it up against already running, or output stream, or input queue // if found, send an IBYTI and discard retry request if (!mySubchannel) retriesReceivedPrm++; else retriesReceivedSec++; bool alreadyRunning = false; Owned wi = queue.running(); ForEach(*wi) { CRoxieWorker &w = (CRoxieWorker &) wi->query(); if (w.match(header)) { alreadyRunning = true; if (!mySubchannel) retriesIgnoredPrm++; else retriesIgnoredSec++; ROQ->sendIbyti(header, logctx, mySubchannel); if (logctx.queryTraceLevel() > 10) { StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for running activity %s", mySubchannel, header.toString(xx).str()); } break; } } if (!alreadyRunning && checkCompleted && ROQ->replyPending(header)) { alreadyRunning = true; if (!mySubchannel) retriesIgnoredPrm++; else retriesIgnoredSec++; ROQ->sendIbyti(header, logctx, mySubchannel); if (logctx.queryTraceLevel() > 10) { StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for completed activity %s", mySubchannel, header.toString(xx).str()); } } if (!alreadyRunning) { if (logctx.queryTraceLevel() > 10) { StringBuffer xx; logctx.CTXLOG("Retry %d received on subchannel %u for %s", retries+1, mySubchannel, header.toString(xx).str()); } queue.enqueueUnique(packet.getClear(), mySubchannel); } } else // first time (not a retry). queue.enqueue(packet.getClear()); } } int run() { if (traceLevel) DBGLOG("RoxieSocketQueueManager::run() starting: doIbytiDelay=%s minIbytiDelay=%u initIbytiDelay=%u", doIbytiDelay?"YES":"NO", minIbytiDelay, initIbytiDelay); for (;;) { MemoryBuffer mb; try { // NOTE - this thread needs to do as little as possible - just read packets and queue them up - otherwise we can get packet loss due to buffer overflow // DO NOT put tracing on this thread except at very high tracelevels! unsigned l; multicastSocket->read(mb.reserve(maxPacketSize), sizeof(RoxiePacketHeader), maxPacketSize, l, 5); mb.setLength(l); packetsReceived++; RoxiePacketHeader &header = *(RoxiePacketHeader *) mb.toByteArray(); if (l != header.packetlength) DBGLOG("sock->read returned %d but packetlength was %d", l, header.packetlength); if (traceLevel > 10) { StringBuffer s; DBGLOG("Read from multicast: %s", header.toString(s).str()); } #ifdef ROXIE_SLA_LOGIC if (header.activityId & ROXIE_SLA_PRIORITY) processMessage(mb, header, slaQueue); else #endif if (header.activityId & ROXIE_HIGH_PRIORITY) processMessage(mb, header, hiQueue); else processMessage(mb, header, loQueue); } catch (IException *E) { if (running) { // MORE: Maybe we should utilize IException::errorCode - not just text ?? if (E->errorCode()==JSOCKERR_timeout_expired) E->Release(); else if (roxiemem::memPoolExhausted()) { //MORE: I think this should probably be based on the error code instead. EXCLOG(E, "Exception reading or processing multicast msg"); E->Release(); MilliSleep(1000); // Give a chance for mem free } else { EXCLOG(E, "Exception reading or processing multicast msg"); E->Release(); // MORE: Protect with try logic, in case udp_create throws exception ? // What to do if create fails (ie exception is caught) ? if (multicastSocket) { multicastSocket->close(); multicastSocket.clear(); openMulticastSocket(); } } } else { E->Release(); break; } } } return 0; } void start() { RoxieReceiverBase::start(); running = true; readThread.start(); } void stop() { if (running) { running = false; multicastSocket->close(); } RoxieReceiverBase::stop(); } void join() { readThread.join(); RoxieReceiverBase::join(); } virtual IReceiveManager *queryReceiveManager() { return receiveManager; } }; class RoxieUdpSocketQueueManager : public RoxieSocketQueueManager { public: RoxieUdpSocketQueueManager(unsigned snifferChannel, unsigned _numWorkers) : RoxieSocketQueueManager(_numWorkers) { int udpQueueSize = topology->getPropInt("@udpQueueSize", UDP_QUEUE_SIZE); int udpSendQueueSize = topology->getPropInt("@udpSendQueueSize", UDP_SEND_QUEUE_SIZE); int udpMaxSlotsPerClient = topology->getPropInt("@udpMaxSlotsPerClient", 0x7fffffff); if (topology->getPropInt("@sendMaxRate", 0)) { unsigned sendMaxRate = topology->getPropInt("@sendMaxRate"); unsigned sendMaxRatePeriod = topology->getPropInt("@sendMaxRatePeriod", 1); bucket.setown(new TokenBucket(sendMaxRate, sendMaxRatePeriod, sendMaxRate)); throttledPacketSendManager.setown(new RoxieThrottledPacketSender(*bucket, maxPacketSize)); } IpAddress snifferIp; getChannelIp(snifferIp, snifferChannel); if (udpMaxSlotsPerClient > udpQueueSize) udpMaxSlotsPerClient = udpQueueSize; unsigned serverFlowPort = topology->getPropInt("@serverFlowPort", CCD_SERVER_FLOW_PORT); unsigned dataPort = topology->getPropInt("@dataPort", CCD_DATA_PORT); unsigned clientFlowPort = topology->getPropInt("@clientFlowPort", CCD_CLIENT_FLOW_PORT); unsigned snifferPort = topology->getPropInt("@snifferPort", CCD_SNIFFER_PORT); receiveManager.setown(createReceiveManager(serverFlowPort, dataPort, clientFlowPort, snifferPort, snifferIp, udpQueueSize, udpMaxSlotsPerClient)); sendManager.setown(createSendManager(serverFlowPort, dataPort, clientFlowPort, snifferPort, snifferIp, udpSendQueueSize, fastLaneQueue ? 3 : 2, bucket)); } }; class RoxieAeronSocketQueueManager : public RoxieSocketQueueManager { public: RoxieAeronSocketQueueManager(unsigned _numWorkers) : RoxieSocketQueueManager(_numWorkers) { unsigned dataPort = topology->getPropInt("@dataPort", CCD_DATA_PORT); SocketEndpoint ep(dataPort, myNode.getNodeAddress()); receiveManager.setown(createAeronReceiveManager(ep)); assertex(!myNode.getNodeAddress().isNull()); sendManager.setown(createAeronSendManager(dataPort, fastLaneQueue ? 3 : 2, myNode.getNodeAddress())); } }; #ifdef _MSC_VER #pragma warning( pop ) #endif //================================================================================================== interface ILocalMessageCollator : extends IMessageCollator { virtual void enqueueMessage(bool outOfBand, void *data, unsigned datalen, void *meta, unsigned metalen, void *header, unsigned headerlen) = 0; }; interface ILocalReceiveManager : extends IReceiveManager { virtual ILocalMessageCollator *lookupCollator(ruid_t id) = 0; }; class LocalMessagePacker : public CDummyMessagePacker { MemoryBuffer meta; MemoryBuffer header; Linked rm; ruid_t id; bool outOfBand; public: IMPLEMENT_IINTERFACE; LocalMessagePacker(RoxiePacketHeader &_header, bool _outOfBand, ILocalReceiveManager *_rm) : rm(_rm), outOfBand(_outOfBand) { id = _header.uid; header.append(sizeof(RoxiePacketHeader), &_header); } virtual void flush() override; virtual void sendMetaInfo(const void *buf, unsigned len) override { meta.append(len, buf); } }; class CLocalMessageUnpackCursor : implements IMessageUnpackCursor, public CInterface { void *data; unsigned datalen; unsigned pos; Linked rowManager; public: IMPLEMENT_IINTERFACE; CLocalMessageUnpackCursor(IRowManager *_rowManager, void *_data, unsigned _datalen) : rowManager(_rowManager) { datalen = _datalen; data = _data; pos = 0; } ~CLocalMessageUnpackCursor() { } virtual bool atEOF() const { return datalen==pos; } virtual bool isSerialized() const { // NOTE: tempting to think that we could avoid serializing in localAgent case, but have to be careful about the lifespan of the rowManager... return true; } virtual const void * getNext(int length) { if (pos==datalen) return NULL; assertex(pos + length <= datalen); void * cur = ((char *) data) + pos; pos += length; void * ret = rowManager->allocate(length, 0); memcpy(ret, cur, length); //No need for finalize since only contains plain data. return ret; } }; class CLocalMessageResult : implements IMessageResult, public CInterface { void *data; void *meta; void *header; unsigned datalen, metalen, headerlen; unsigned pos; public: IMPLEMENT_IINTERFACE; CLocalMessageResult(void *_data, unsigned _datalen, void *_meta, unsigned _metalen, void *_header, unsigned _headerlen) { datalen = _datalen; metalen = _metalen; headerlen = _headerlen; data = _data; meta = _meta; header = _header; pos = 0; } ~CLocalMessageResult() { free(data); free(meta); free(header); } virtual IMessageUnpackCursor *getCursor(IRowManager *rowMgr) const { return new CLocalMessageUnpackCursor(rowMgr, data, datalen); } virtual const void *getMessageHeader(unsigned &length) const { length = headerlen; return header; } virtual const void *getMessageMetadata(unsigned &length) const { length = metalen; return meta; } virtual void discard() const { } }; class CLocalMessageCollator : implements ILocalMessageCollator, public CInterface { InterruptableSemaphore sem; QueueOf pending; CriticalSection crit; Linked rowManager; // Linked to ensure it lives longer than me Linked receiveManager; ruid_t id; unsigned totalBytesReceived; public: IMPLEMENT_IINTERFACE; CLocalMessageCollator(IRowManager *_rowManager, ruid_t _ruid); ~CLocalMessageCollator(); virtual ruid_t queryRUID() const { return id; } virtual IMessageResult* getNextResult(unsigned time_out, bool &anyActivity) { anyActivity = false; if (!sem.wait(time_out)) return NULL; anyActivity = true; CriticalBlock c(crit); return pending.dequeue(); } virtual void interrupt(IException *E) { sem.interrupt(E); } virtual void enqueueMessage(bool outOfBand, void *data, unsigned datalen, void *meta, unsigned metalen, void *header, unsigned headerlen) { CriticalBlock c(crit); if (outOfBand) pending.enqueueHead(new CLocalMessageResult(data, datalen, meta, metalen, header, headerlen)); else pending.enqueue(new CLocalMessageResult(data, datalen, meta, metalen, header, headerlen)); sem.signal(); totalBytesReceived += datalen + metalen + headerlen; } virtual unsigned queryBytesReceived() const { return totalBytesReceived; } }; class RoxieLocalReceiveManager : implements ILocalReceiveManager, public CInterface { MapXToMyClass collators; CriticalSection crit; Owned logctx; public: IMPLEMENT_IINTERFACE; RoxieLocalReceiveManager() : logctx(new StringContextLogger("RoxieLocalReceiveManager")) { } virtual IMessageCollator *createMessageCollator(IRowManager *manager, ruid_t ruid) { ILocalMessageCollator *collator = new CLocalMessageCollator(manager, ruid); CriticalBlock b(crit); collators.setValue(ruid, collator); return collator; } virtual void detachCollator(const IMessageCollator *collator) { ruid_t id = collator->queryRUID(); CriticalBlock b(crit); collators.setValue(id, NULL); } virtual ILocalMessageCollator *lookupCollator(ruid_t id) { CriticalBlock b(crit); ILocalMessageCollator *ret = collators.getValue(id); if (!ret) ret = collators.getValue(RUID_DISCARD); return LINK(ret); } }; void LocalMessagePacker::flush() { // MORE - I think this means we don't send anything until whole message available in localAgent mode, which // may not be optimal. data.setLength(lastput); Owned collator = rm->lookupCollator(id); if (collator) { unsigned datalen = data.length(); unsigned metalen = meta.length(); unsigned headerlen = header.length(); collator->enqueueMessage(outOfBand, data.detach(), datalen, meta.detach(), metalen, header.detach(), headerlen); } // otherwise Roxie server is no longer interested and we can simply discard } CLocalMessageCollator::CLocalMessageCollator(IRowManager *_rowManager, ruid_t _ruid) : rowManager(_rowManager), id(_ruid) { totalBytesReceived = 0; } CLocalMessageCollator::~CLocalMessageCollator() { IMessageResult *goer; for (;;) { goer = pending.dequeue(); if (!goer) break; goer->Release(); } } class RoxieLocalQueueManager : public RoxieReceiverBase { Linked receiveManager; public: RoxieLocalQueueManager(unsigned _numWorkers) : RoxieReceiverBase(_numWorkers) { receiveManager.setown(new RoxieLocalReceiveManager); } virtual void sendPacket(IRoxieQueryPacket *packet, const IRoxieContextLogger &logctx) override { RoxiePacketHeader &header = packet->queryHeader(); unsigned retries = header.thisChannelRetries(0); if (header.activityId == ROXIE_FILECALLBACK || header.activityId == ROXIE_DEBUGCALLBACK ) { if (traceLevel > 5) { StringBuffer s; DBGLOG("ROXIE_CALLBACK %s", header.toString(s).str()); } doFileCallback(packet); } else if (retries < SUBCHANNEL_MASK) { if (retries) { // Send back an out-of-band immediately, to let Roxie server know that channel is still active RoxiePacketHeader newHeader(header, ROXIE_ALIVE, 0); Owned output = createOutputStream(newHeader, true, logctx); output->flush(); return; // No point sending the retry in localAgent mode } RoxieQueue *targetQueue; #ifdef ROXIE_SLA_LOGIC if (header.activityId & ROXIE_SLA_PRIORITY) targetQueue = &slaQueue; else #endif if (header.activityId & ROXIE_HIGH_PRIORITY) targetQueue = &hiQueue; else targetQueue = &loQueue; if (header.channel) { targetQueue->enqueue(LINK(packet)); } else { // Turn broadcast packet (channel 0), as early as possible, into non-0 channel packets. // So retries and other communication with Roxie server (which uses non-0 channel numbers) will not cause double work or confusion. for (unsigned i = 0; i < numChannels; i++) { targetQueue->enqueue(packet->clonePacket(i+1)); } } } } virtual void sendIbyti(RoxiePacketHeader &header, const IRoxieContextLogger &logctx, unsigned subChannel) override { // Don't do IBYTI's when local agent - no buddy to talk to anyway } virtual void sendAbort(RoxiePacketHeader &header, const IRoxieContextLogger &logctx) override { MTIME_SECTION(queryActiveTimer(), "RoxieLocalQueueManager::sendAbort"); RoxiePacketHeader abortHeader(header, header.activityId & ROXIE_PRIORITY_MASK, 0); abortHeader.retries = QUERY_ABORTED; if (logctx.queryTraceLevel() > 8) { StringBuffer s; logctx.CTXLOG("Sending ABORT packet %s", abortHeader.toString(s).str()); } MemoryBuffer data; data.append(sizeof(abortHeader), &abortHeader); Owned packet = createRoxiePacket(data); sendPacket(packet, logctx); abortsSent++; } virtual void sendAbortCallback(const RoxiePacketHeader &header, const char *lfn, const IRoxieContextLogger &logctx) override { MTIME_SECTION(queryActiveTimer(), "RoxieLocalQueueManager::sendAbortCallback"); RoxiePacketHeader abortHeader(header, ROXIE_FILECALLBACK, 0); abortHeader.retries = QUERY_ABORTED; MemoryBuffer data; data.append(sizeof(abortHeader), &abortHeader).append(lfn); if (logctx.queryTraceLevel() > 5) { StringBuffer s; logctx.CTXLOG("Sending ABORT FILECALLBACK packet %s for file %s", abortHeader.toString(s).str(), lfn); } Owned packet = createRoxiePacket(data); sendPacket(packet, logctx); abortsSent++; } virtual IMessagePacker *createOutputStream(RoxiePacketHeader &header, bool outOfBand, const IRoxieContextLogger &logctx) override { return new LocalMessagePacker(header, outOfBand, receiveManager); } virtual IReceiveManager *queryReceiveManager() override { return receiveManager; } virtual bool replyPending(RoxiePacketHeader &header) override { // MORE - should really have some code here! But returning true is a reasonable approximation. return true; } virtual bool abortCompleted(RoxiePacketHeader &header) override { // MORE - should really have some code here! return false; } }; IRoxieOutputQueueManager *ROQ; extern IRoxieOutputQueueManager *createOutputQueueManager(unsigned snifferChannel, unsigned numWorkers) { if (localAgent) return new RoxieLocalQueueManager(numWorkers); else if (useAeron) return new RoxieAeronSocketQueueManager(numWorkers); else return new RoxieUdpSocketQueueManager(snifferChannel, numWorkers); } //================================================================================================================================ class PacketDiscarder : public Thread, implements IPacketDiscarder { bool aborted; Owned rowManager; // not completely sure I need one... maybe I do Owned mc; public: IMPLEMENT_IINTERFACE; PacketDiscarder() { aborted = false; }; ~PacketDiscarder() { if (mc) ROQ->queryReceiveManager()->detachCollator(mc); mc.clear(); } virtual int run() { Owned logctx = new StringContextLogger("PacketDiscarder"); rowManager.setown(roxiemem::createRowManager(0, NULL, *logctx, NULL, false)); mc.setown(ROQ->queryReceiveManager()->createMessageCollator(rowManager, RUID_DISCARD)); while (!aborted) { bool anyActivity = false; Owned mr = mc->getNextResult(5000, anyActivity); if (mr) { if (traceLevel > 4) DBGLOG("Discarding unwanted message"); unsigned headerLen; const RoxiePacketHeader &header = *(const RoxiePacketHeader *) mr->getMessageHeader(headerLen); if (headerLen) { switch (header.activityId) { case ROXIE_FILECALLBACK: { Owned callbackData = mr->getCursor(rowManager); OwnedConstRoxieRow len = callbackData->getNext(sizeof(RecordLengthType)); if (len) { RecordLengthType *rowlen = (RecordLengthType *) len.get(); OwnedConstRoxieRow row = callbackData->getNext(*rowlen); const char *rowdata = (const char *) row.get(); // bool isOpt = * (bool *) rowdata; // bool isLocal = * (bool *) (rowdata+1); ROQ->sendAbortCallback(header, rowdata+2, *logctx); } else DBGLOG("Unrecognized format in discarded file callback"); break; } // MORE - ROXIE_ALIVE perhaps should go here too? debug callbacks? Actually any standard query results should too (though by the time I see them here it's too late (that may change once start streaming) } } else DBGLOG("Unwanted message had no header?!"); } else if (!anyActivity) { // to avoid leaking partial unwanted packets, we clear out mc periodically... ROQ->queryReceiveManager()->detachCollator(mc); mc.setown(ROQ->queryReceiveManager()->createMessageCollator(rowManager, RUID_DISCARD)); } } return 0; } virtual void start() { Thread::start(); } virtual void stop() { if (mc) mc->interrupt(); aborted = true; join(); } }; IPacketDiscarder *createPacketDiscarder() { IPacketDiscarder *packetDiscarder = new PacketDiscarder; packetDiscarder->start(); return packetDiscarder; } //================================================================================================================================ // There are various possibly interesting ways to reply to a ping: // Reply as soon as receive, or put it on the queue like other messages? // Reply for every channel, or just once for every agent? // Should I send on channel 0 or round-robin the channels? // My gut feeling is that knowing what channels are responding is useful so should reply on every unsuspended channel, // and that the delay caused by queuing system is an interesting part of what we want to measure (though nice to know minimum possible too) unsigned pingInterval = 60; class PingTimer : public Thread { bool aborted; Owned rowManager; Owned mc; StringContextLogger logctx; void sendPing(unsigned priorityMask) { try { unsigned packetSize = sizeof(RoxiePacketHeader) + sizeof(char) + strlen("PING") + 1 + sizeof(PingRecord); void *packetData = malloc(packetSize); RoxiePacketHeader *header = (RoxiePacketHeader *) packetData; RemoteActivityId pingId(ROXIE_PING | priorityMask, 0); header->init(pingId, 0, 0, 0); char *finger = (char *) (header + 1); *finger++ = (char) LOGGING_FLAGSPRESENT; strcpy(finger, "PING"); finger += strlen("PING")+1; if (traceLevel > 1) DBGLOG("PING sent"); PingRecord data; data.senderIP.ipset(myNode.getNodeAddress()); data.tick = usTick(); memcpy(finger, &data, sizeof(PingRecord)); Owned packet = createRoxiePacket(packetData, packetSize); ROQ->sendPacket(packet, logctx); } catch (IException *E) { EXCLOG(E); E->Release(); } } public: PingTimer() : logctx("PingTimer") { aborted = false; }; ~PingTimer() { if (mc) ROQ->queryReceiveManager()->detachCollator(mc); mc.clear(); } virtual int run() { rowManager.setown(roxiemem::createRowManager(1, NULL, queryDummyContextLogger(), NULL, false)); mc.setown(ROQ->queryReceiveManager()->createMessageCollator(rowManager, RUID_PING)); unsigned pingsReceived = 0; unsigned pingsElapsed = 0; sendPing(ROXIE_HIGH_PRIORITY); while (!aborted) { bool anyActivity = false; Owned mr = mc->getNextResult(pingInterval*1000, anyActivity); if (mr) { unsigned headerLen; const RoxiePacketHeader *header = (const RoxiePacketHeader *) mr->getMessageHeader(headerLen); Owned mu = mr->getCursor(rowManager); PingRecord *answer = (PingRecord *) mu->getNext(sizeof(PingRecord)); if (answer && mu->atEOF() && headerLen==sizeof(RoxiePacketHeader)) { unsigned elapsed = usTick() - answer->tick; pingsReceived++; pingsElapsed += elapsed; if (traceLevel > 10) DBGLOG("PING reply channel=%d, time %d", header->channel, elapsed); // DBGLOG is slower than the pings so be careful! } else DBGLOG("PING reply, garbled result"); ReleaseRoxieRow(answer); } else if (!anyActivity) { if (!pingsReceived && roxieMulticastEnabled) DBGLOG("PING: NO replies received! Please check multicast settings, and that your network supports multicast."); else if (traceLevel) DBGLOG("PING: %d replies received, average delay %uus", pingsReceived, pingsReceived ? pingsElapsed / pingsReceived : 0); pingsReceived = 0; pingsElapsed = 0; sendPing(ROXIE_HIGH_PRIORITY); // MORE - we could think about alternating the priority or sending pings on high and low at the same time... } } return 0; } void stop() { if (mc) mc->interrupt(); aborted = true; } static CriticalSection crit; } *pingTimer; CriticalSection PingTimer::crit; extern void startPingTimer() { CriticalBlock b(PingTimer::crit); if (!pingTimer) { pingTimer = new PingTimer(); pingTimer->start(); } } extern void stopPingTimer() { CriticalBlock b(PingTimer::crit); if (pingTimer) { pingTimer->stop(); pingTimer->join(); delete pingTimer; pingTimer = NULL; } }