HPCC-16426 Uninitialized fields reported by coverity

None of the issues fixed here would have resulted in incorrect program
behaviour, I think.

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

common/thorhelper/roxiedebug.ipp

@@ -125,7 +125,6 @@ interface IActivityDebugContext : extends IInterface
 
     virtual const char *queryEdgeId() const = 0;
     virtual const char *querySourceId() const = 0;
-    virtual unsigned queryChildGraphId() const = 0;
 
     virtual memsize_t queryProxyId() const = 0;
 };

ecl/eclagent/eclagent.cpp

@@ -3762,7 +3762,6 @@ class DebugProbe : public InputProbe, implements IActivityDebugContext
     unsigned historySize;
     unsigned historyCapacity;
     unsigned nextHistorySlot;
-    unsigned childGraphId;
     
     mutable memsize_t proxyId; // MORE - do we need a critsec to protect too?
 
@@ -3861,11 +3860,6 @@ public:
         return proxyId;
     }
 
-    virtual unsigned queryChildGraphId() const
-    {
-        return childGraphId;
-    }
-
     virtual void resetEOF() 
     { 
         forceEOF = false;

esp/services/ws_ecl/CMakeLists.txt

@@ -49,7 +49,6 @@ include_directories (
          ./../../../system/include 
          ./../../../common/workunit 
          ./../../../common/wuwebview
-         ./../../../roxie/roxieclient 
          ./../../../common/fileview2 
          ./../../../dali/base 
     )

roxie/CMakeLists.txt

@@ -15,8 +15,6 @@
 ################################################################################
 HPCC_ADD_SUBDIRECTORY (ccd)
 HPCC_ADD_SUBDIRECTORY (roxie "PLATFORM")
-HPCC_ADD_SUBDIRECTORY (roxieclient "PLATFORM")
-HPCC_ADD_SUBDIRECTORY (roxieclienttest "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (roxiemem)
 HPCC_ADD_SUBDIRECTORY (roxiepipe "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (udplib)

roxie/ccd/ccdcontext.cpp

@@ -2393,7 +2393,10 @@ public:
             }
         }
         else
+        {
             assertex(selfTestMode);
+            header = nullptr;
+        }
     }
     virtual void beforeDispose()
     {

roxie/ccd/ccddebug.cpp

@@ -508,7 +508,6 @@ class DebugProbe : public InputProbe, implements IActivityDebugContext
     unsigned historySize;
     unsigned historyCapacity;
     unsigned nextHistorySlot;
-    unsigned childGraphId;
 
     mutable memsize_t proxyId; // MORE - do we need a critsec to protect too?
 
@@ -615,11 +614,6 @@ public:
         return proxyId;
     }
 
-    virtual unsigned queryChildGraphId() const
-    {
-        return childGraphId;
-    }
-
     virtual void resetEOF()
     {
         forceEOF = false;

roxie/ccd/ccdmain.cpp

@@ -433,7 +433,6 @@ int STARTQUERY_API start_query(int argc, const char *argv[])
     init_signals();
     // We need to do the above BEFORE we call InitModuleObjects
     InitModuleObjects();
-    getDaliServixPort();
     init_signals();
 
     // stand alone usage only, not server

roxie/ccd/ccdprotocol.cpp

@@ -447,8 +447,7 @@ protected:
     bool isBlocked;
     bool isRaw;
     bool isHTTP;
-    bool trim;
-    bool failed;
+    bool trim;   // MORE - this is never set!
     bool adaptiveRoot = false;
     bool onlyUseFirstRow = false;
 
@@ -937,7 +936,6 @@ protected:
     CriticalSection contentsCrit;
     unsigned protocolFlags;
     bool isHTTP;
-    bool failed;
 
 public:
     IMPLEMENT_IINTERFACE;

roxie/ccd/ccdquery.cpp

@@ -1049,6 +1049,7 @@ public:
         : id(_id), package(_package), dll(_dll), channelNo(_channelNo), hashValue(_hashValue), sharedOnceContext(_sharedOnceContext), dynamic(_dynamic)
     {
         package.Link();
+        targetClusterType = RoxieCluster;
         isSuspended = false;
         isLoadFailed = false;
         libraryInterfaceHash = 0;

roxie/ccd/ccdserver.cpp

@@ -975,6 +975,7 @@ public:
     {
         activityId = 0;
         input = NULL;
+        sourceIdx = 0;
         inputStream = NULL;
         ctx = NULL;
         meta.set(basehelper.queryOutputMeta());
@@ -5871,11 +5872,11 @@ class CRoxieServerStrandedInlineTableActivity : public CRoxieServerStrandedActiv
     protected:
         IHThorInlineTableArg &helper;
         unsigned whichStrand;
-        unsigned numStrands;
+        unsigned numStrands = 0;
         __uint64 sectionSize;
-        __uint64 curRow;
-        __uint64 sectionMaxRows;
-        __uint64 maxRows;
+        __uint64 curRow = 0;
+        __uint64 sectionMaxRows = 0;
+        __uint64 maxRows = 0;
         bool isOrdered;
         bool eosPending;
 
@@ -14325,7 +14326,10 @@ class CRoxieServerPrefetchProjectActivity : public CRoxieServerActivity, impleme
                 recordCount = _recordCount;
             }
             else
+            {
                 ReleaseRoxieRow(_in);
+                recordCount = 0;
+            }
         }
         OwnedConstRoxieRow in;
         unsigned __int64 recordCount;
@@ -18293,8 +18297,8 @@ private:
     protected:
         __uint64 *table;
         ConstPointerArray rowtable;
-        mutable unsigned currentMatch;
-        mutable unsigned matchCount;
+        mutable unsigned currentMatch = 0;
+        mutable unsigned matchCount = 0;
     };
 
     IHThorHashJoinArg &helper;
@@ -22743,11 +22747,11 @@ class CRoxieServerIndexReadActivity : public CRoxieServerIndexReadBaseActivity,
 {
 protected:
     IHThorCompoundReadExtra & readHelper;
-    ISteppingMeta *rawMeta;
+    ISteppingMeta *rawMeta = nullptr;
     CSteppingMeta steppingMeta;
-    unsigned * seekSizes;
+    unsigned * seekSizes = nullptr;
     bool optimizeSteppedPostFilter;
-    ISteppingMeta * projectedMeta;
+    ISteppingMeta * projectedMeta = nullptr;
     unsigned maxSeekLookahead;
 
 public:

roxie/roxieclient/CMakeLists.txt

@@ -1,47 +0,0 @@
-################################################################################
-#    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.
-################################################################################
-
-
-# Component: roxieclient 
-
-#####################################################
-# Description:
-# ------------
-#    Cmake Input File for roxieclient
-#####################################################
-
-
-project( roxieclient ) 
-
-set (    SRCS 
-         roxieclient.cpp 
-    )
-
-include_directories ( 
-         ./../../system/include 
-         ./../../system/jlib 
-         ./../../roxie/roxie
-    )
-
-ADD_DEFINITIONS( -D_USRDLL -DROXIECLIENT_EXPORTS )
-
-HPCC_ADD_LIBRARY( roxieclient SHARED ${SRCS} )
-install ( TARGETS roxieclient RUNTIME DESTINATION ${EXEC_DIR} LIBRARY DESTINATION ${LIB_DIR} )
-target_link_libraries ( roxieclient
-         jlib
-    )
-
-

roxie/roxieclient/roxieclient.cpp

@@ -1,628 +0,0 @@
-/*##############################################################################
-
-    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 "roxieclient.ipp"
-#include "roxie.hpp"
-
-#define MAXBUFSIZE 0x8000
-
-#ifdef _DEBUG
-//#define DEBUG_ROXIECLIENT_
-#endif
-
-int CDataInputCache::readInput()
-{
-    inputLen = 0;
-    curInput = 0;
-    synchronized procedure(input->readMutex);
-
-    IByteInputStream* inputstream = input->stream.get();
-
-    if(inputstream == NULL)
-        return 0;
-
-    int bytes = 0;
-    if(input->record_width > 0)
-    {
-        char* inbuf = (char*)buffer.str();
-
-        while ((inputLen < bytesPerQuery) && (bytes = inputstream->readBytes(inbuf + inputLen, bytesPerQuery - inputLen)) > 0)
-        {
-            inputLen += bytes;
-        }
-    }
-    else
-    {
-        buffer.clear();
-        char inbuf[2048];
-        while ((bytes = inputstream->readBytes(inbuf, 2048)) > 0)
-        {
-            buffer.append(bytes, inbuf);
-            inputLen += bytes;
-        }
-    }
-
-    if (bytes < 0)
-    {
-        throw MakeStringException(-1, "Error in readInput(): errno = %d", errno);
-    }
-
-#ifdef DEBUG_ROXIECLIENT_
-    if(input->record_width > 0)
-        DBGLOG("Read %d bytes (%d records) from input stream %s", inputLen, inputLen/input->record_width, name.str());
-    else
-        DBGLOG("Read %d bytes from input stream %s", inputLen, name.str());
-#endif
-
-    return inputLen;
-}
-
-int CDataOutputCache::writeOutput()
-{
-
-    int bytes = 0;
-
-    synchronized procedure(output->writeMutex);
-
-    IByteOutputStream* outputstream = output->stream.get();
-    if(outputstream == NULL)
-        return 0;
-
-    while (outputQ.ordinality() > 0)
-    {
-        MemoryBuffer *mb = outputQ.dequeue();
-        size32_t len = mb->length();
-        const char *obuf = mb->toByteArray();
-        const char *finger = obuf;
-        
-        if(!isNonBlocked)
-        {   
-            finger = strchr(obuf+9, '\0') + 1;
-            len -= (finger-obuf);
-        }
-        try {
-            outputstream->writeBytes(finger, len);
-        }
-        catch (IException *) {
-            delete mb;
-            throw;
-        }
-
-#ifdef DEBUG_ROXIECLIENT_
-        DBGLOG("Written %d bytes to output %s", len, name.str());
-#endif
-        bytes += len;
-        delete mb;
-    }
-
-    return bytes;
-}
-
-void CDataOutputCache::clearOutput()
-{
-    while (outputQ.ordinality() > 0)
-        delete outputQ.dequeue();
-}
-
-int CDataOutputCache::immediateOutput(size32_t len, const char* result)
-{
-    synchronized procedure(output->writeMutex);
-
-    IByteOutputStream* outputstream = output->stream.get();
-    if(outputstream == NULL)
-        return 0;
-    
-    if(result != NULL)
-    {
-        outputstream->writeBytes(result, len);
-#ifdef DEBUG_ROXIECLIENT_
-        DBGLOG("Written %d bytes to output %s", len, name.str());
-#endif
-
-        return len;
-    }
-    return 0;
-}
-
-
-RoxieThread::RoxieThread(CRoxieClient* _roxieclient, const char *_query, unsigned recordsPerQuery, unsigned maxRetries, int readTimeout) 
-    : roxieclient(_roxieclient), query(_query), m_recordsPerQuery(recordsPerQuery)
-{
-    bytesRead = bytesWritten = 0;
-
-    CDataInput* dinput = roxieclient->m_default_input.get();
-    if(dinput != NULL)
-    {
-        m_default_incache.setown(new CDataInputCache(dinput, m_recordsPerQuery));
-    }
-    CDataOutput* doutput = roxieclient->m_default_output.get();
-    if(doutput != NULL)
-    {
-        m_default_outcache.setown(new CDataOutputCache(doutput));
-    }
-
-    ForEachItemIn(i, roxieclient->m_inputs)
-    {
-        CDataInput* oneinput = &(roxieclient->m_inputs.item(i));
-        if(oneinput != NULL)
-        {
-            m_incaches.append(*(new CDataInputCache(oneinput, m_recordsPerQuery)));
-        }
-    }
-
-    ForEachItemIn(j, roxieclient->m_outputs)
-    {
-        CDataOutput* oneoutput = &(roxieclient->m_outputs.item(j));
-        if(oneoutput != NULL)
-        {
-            m_outcaches.append(*(new CDataOutputCache(oneoutput)));
-        }
-    }
-    
-    m_maxRetries = maxRetries;
-    m_readTimeout = readTimeout;
-}
-
-void RoxieThread::sendQuery()
-{
-    MemoryBuffer sendBuffer;
-
-    unsigned queryLen = query.length();
-
-    unsigned revQueryLen = queryLen;
-    _WINREV(revQueryLen);
-    sendBuffer.append(revQueryLen);
-    sendBuffer.append(queryLen, query.str());
-
-    roxieSock->write(sendBuffer.toByteArray(), sendBuffer.length());
-#ifdef DEBUG_ROXIECLIENT_
-    DBGLOG("Sent query %s", query.str());
-#endif
-}
-
-bool RoxieThread::readInput()
-{
-    bool done = true;
-    if(m_default_incache != NULL)
-    {
-        int len = m_default_incache->readInput();
-        done = done && (len == 0);
-        bytesRead += len;
-    }
-    
-    ForEachItemIn(i, m_incaches)
-    {
-        CDataInputCache* oneincache = &(m_incaches.item(i));
-        if(oneincache != NULL)
-        {
-            int len = oneincache->readInput();
-            done = done && (len == 0);
-            bytesRead += len;
-        }
-    }
-
-    return done;
-}
-
-void RoxieThread::writeOutput()
-{
-    if(m_default_outcache != NULL)
-    {
-        int bytes = m_default_outcache->writeOutput();
-        bytesWritten += bytes;
-    }
-    
-    ForEachItemIn(i, m_outcaches)
-    {
-        CDataOutputCache* oneoutcache = &(m_outcaches.item(i));
-        if(oneoutcache != NULL)
-        {
-            int bytes = oneoutcache->writeOutput();
-            bytesWritten += bytes;
-        }
-    }
-
-}
-
-void RoxieThread::clearOutput()
-{
-    if(m_default_outcache != NULL)
-        m_default_outcache->clearOutput();
-    
-    ForEachItemIn(i, m_outcaches)
-    {
-        CDataOutputCache* oneoutcache = &(m_outcaches.item(i));
-        if(oneoutcache != NULL)
-            oneoutcache->clearOutput();
-    }
-
-}
-
-void RoxieThread::immediateOutput(size32_t len, const char* result)
-{
-    if(m_default_outcache != NULL)
-    {
-        m_default_outcache->immediateOutput(len, result);
-        bytesWritten += len;
-    }
-    else
-    {
-        throw MakeStringException(-1, "Output stream not set. Please set default outputstream, or use bxml as output format");
-    }
-}
-
-void RoxieThread::sendData(const char *reqbuf)
-{
-    static unsigned sequenceNumber = 0;
-
-    const char *inputName = reqbuf+9;
-    int inputNameLen = strlen(inputName);
-    
-    CDataInputCache* icache = m_default_incache.get();
-
-    ForEachItemIn(i, m_incaches)
-    {
-        CDataInputCache* oneincache = &(m_incaches.item(i));
-        if(oneincache != NULL && strcmp(inputName, oneincache->name.str()) == 0)
-        {
-            icache = oneincache;
-        }
-    }
-
-    if(icache == NULL)
-        return;
-
-    if (icache->curInput < icache->inputLen)
-    {
-        unsigned recordsLeft, numRecords, numBytes;
-        int in_width = icache->input->record_width;
-        if(in_width == 0)
-        {
-            numRecords = 0; // Wouldn't be able to know, so set to 0.
-            numBytes = icache->inputLen;
-        }
-        else
-        {
-            recordsLeft = (icache->inputLen - icache->curInput) / in_width;
-            numRecords = (recordsLeft < icache->recordsPerBlock)? recordsLeft: icache->recordsPerBlock;
-            numBytes = numRecords * in_width;
-        }
-
-        MemoryBuffer sendBuffer;
-        unsigned rev = numBytes + inputNameLen + 10;
-        rev |= 0x80000000;
-        _WINREV(rev);
-        sendBuffer.append(rev);
-        sendBuffer.append('R');
-        rev = sequenceNumber++;
-        _WINREV(rev);
-        sendBuffer.append(rev);
-        rev = numRecords;
-        _WINREV(rev);
-        sendBuffer.append(rev);
-        sendBuffer.append(inputNameLen+1, inputName);
-        sendBuffer.append(numBytes, icache->buffer.str()+icache->curInput);
-
-        roxieSock->write(sendBuffer.toByteArray(), sendBuffer.length());
-#ifdef DEBUG_ROXIECLIENT_
-        DBGLOG("Wrote %d bytes from inputQ to socket", numBytes);
-#endif
-        icache->curInput += numBytes;
-    }
-    else
-    {
-        unsigned zeroLen = 0;
-        roxieSock->write(&zeroLen, sizeof(zeroLen));
-#ifdef DEBUG_ROXIECLIENT_
-        DBGLOG("Wrote %d bytes to socket", sizeof(zeroLen));
-#endif
-    }
-}
-
-void RoxieThread::processQuery()
-{
-    char buf[MAXBUFSIZE];
-    char *obuf;
-
-    while (1)
-    {
-        bool blockedMsg = false;
-        size32_t r = 0;
-
-        roxieSock->read(buf,4,4,r, m_readTimeout);
-        size32_t len = *(unsigned *)buf;
-        _WINREV4(len);
-
-        if (len & 0x80000000)
-        {
-            len ^= 0x80000000;
-            blockedMsg = true;
-        }
-
-        if (len == 0)
-            break;
-
-        r=0;
-        MemoryBuffer x;
-        obuf = (char *) x.reserve(len);
-        roxieSock->read(obuf, len, len, r, m_readTimeout);
-
-        if (blockedMsg)
-        {
-            if (*obuf == 'D')
-            {
-                sendData(obuf);
-            }
-            else if (*obuf == 'R')
-            {
-                const char *finger = obuf+9;
-                const char *payload = finger + strlen(finger) + 1;
-
-                if (strcmp(finger, "Exception") == 0)
-                {
-                    StringBuffer body(x.length() - (payload-obuf), payload);
-                    StringBuffer xml;
-                    xml.append("<Exception>").append(body).append("</Exception>");
-                    Owned<IPropertyTree> ep = createPTreeFromXMLString(xml.str());
-                    int code = ep->getPropInt("./Code", 0);
-                    SocketEndpoint peerEp;
-                    StringBuffer peerStr;
-                    DBGLOG("Connected to %s", roxieSock->querySocket()->getPeerEndpoint(peerEp).getUrlStr(peerStr).str());
-                    DBGLOG("Roxie exception: %s", body.str());
-                    throw new ReceivedRoxieException(code, body.str());
-                }
-                else
-                {
-                    CDataOutputCache* ocache = m_default_outcache.get();
-
-                    ForEachItemIn(i, m_outcaches)
-                    {
-                        CDataOutputCache* oneoutcache = &(m_outcaches.item(i));
-                        if(oneoutcache != NULL && strcmp(finger, oneoutcache->name.str()) == 0)
-                        {
-                            ocache = oneoutcache;
-                        }
-                    }
-                    if(ocache != NULL)
-                    {
-                        int out_width = ocache->output->record_width;
-                        if(out_width > 0 && (len - (payload-obuf)) % out_width != 0)
-                        {
-                            StringBuffer errmsg;
-                            errmsg.appendf("received %d bytes of data, not a multiple of -ow %d", (unsigned) (len-(payload-obuf)), out_width);
-                            throw new RoxieClientException(-1, errmsg.str());
-                        }
-                        MemoryBuffer* y = new MemoryBuffer;
-                        y->setBuffer(len, x.detach(), true);
-                        ocache->outputQ.enqueue(y);
-                        continue;
-                    }
-                }
-            }
-        }
-        else
-        {
-            CDataOutputCache* ocache = m_default_outcache.get();
-            if(ocache != NULL)
-            {
-                ocache->isNonBlocked = true;
-                MemoryBuffer* y = new MemoryBuffer;
-                y->setBuffer(len, x.detach(), true);
-                ocache->outputQ.enqueue(y);
-            }
-            //immediateOutput(x.length(), x.toByteArray());
-        }
-    }
-}
-
-int RoxieThread::run()
-{
-    bool done = false;
-    unsigned attempts;
-    int retryInterval;
-    Owned<IRandomNumberGenerator> random = createRandomNumberGenerator();
-    random->seed((unsigned)get_cycles_now());
-    
-    while (!done)
-    {
-        try
-        {
-            done = readInput();
-        }
-        catch (IException *E)
-        {
-            EXCLOG(E, "readInput");
-            throw MakeStringException(-1, "Error readInput");
-        }
-        catch (...)
-        {
-            throw MakeStringException(-1, "Unknown exception in readInput");
-        }
-
-        attempts = 0;
-        retryInterval = 1;
-
-        while(1)
-        {
-            try
-            {
-                roxieSock.setown(roxieclient->querySocketFactory()->connect_timeout(9000));
-
-                SocketEndpoint peerEp;
-                StringBuffer peerStr;
-                DBGLOG("Connected to %s", roxieSock->querySocket()->getPeerEndpoint(peerEp).getUrlStr(peerStr).str());
-
-                sendQuery();
-                processQuery();
-
-                break;
-            }
-            catch (ISmartSocketException *e)
-            {
-                StringBuffer errmsg;
-                e->errorMessage(errmsg);
-                ERRLOG("Fatal exception: %s", errmsg.str());
-                StringBuffer buf;
-                buf.append("<Exception><Source>roxieclient</Source><Message>").append(errmsg.str()).append("</Message></Exception>");
-                immediateOutput(buf.length(), buf.str());
-                clearOutput();
-                e->Release();
-                return 0;
-            }
-            catch (IReceivedRoxieException *e)
-            {
-                if (e->errorCode() == ROXIE_TOO_MANY_QUERIES) // server busy exception
-                {
-                    int sleepTime = retryInterval + random->next() % retryInterval;
-                    DBGLOG("Thread sleeping for %d seconds", sleepTime);
-                    Sleep(sleepTime*1000);
-                    retryInterval = (retryInterval*2 >= 30)? 30: retryInterval*2;
-                }
-                else
-                {
-                    StringBuffer errmsg;
-                    e->errorMessage(errmsg);
-                    DBGLOG("Roxieclient received exception from Roxie: %s", errmsg.str());
-                    if (attempts < m_maxRetries)
-                    {
-                        DBGLOG("Retrying (%d times out of maxRetries %d)", attempts, m_maxRetries);
-                        e->Release();
-                        attempts++;
-                    }
-                    else
-                    {
-                        DBGLOG("Returning: maxRetries exceeded");
-                        StringBuffer buf("<Exception><Source>roxieclient</Source><Message>");
-                        buf.append(errmsg.str()).append(" and Roxie maxRetries exceeded</Message></Exception>");
-                        immediateOutput(buf.length(), buf.str());
-                        clearOutput();
-                        e->Release();
-                        return 0;
-                    }
-                }
-
-                clearOutput();
-                e->Release();
-            }
-            catch (IRoxieClientException* e)
-            {
-                StringBuffer errmsg;
-                e->errorMessage(errmsg);
-                DBGLOG("Roxieclient caught exception: %s", errmsg.str());
-                StringBuffer buf("<Exception><Source>roxieclient</Source><Message>");
-                buf.append(errmsg.str()).append("</Message></Exception>");
-                immediateOutput(buf.length(), buf.str());
-                clearOutput();
-                e->Release();
-                return 0;
-            }
-            catch (IException *e)
-            {
-                StringBuffer errmsg;
-                e->errorMessage(errmsg);
-                DBGLOG("Roxieclient caught exception: %s", errmsg.str());
-                if (attempts < m_maxRetries)
-                {
-                    DBGLOG("Retrying (%d times out of maxRetries %d)", attempts, m_maxRetries);
-                    e->Release();
-                    attempts++;
-                }
-                else
-                {
-                    DBGLOG("Returning: maxRetries exceeded");
-                    StringBuffer buf("<Exception><Source>roxieclient</Source><Message>");
-                    buf.append(errmsg.str()).append(" and Roxie maxRetries exceeded</Message></Exception>");
-                    immediateOutput(buf.length(), buf.str());
-                    clearOutput();
-                    e->Release();
-                    return 0;
-                }
-            }
-            catch(...)
-            {
-                ERRLOG("Caught unexpected Roxie exception");
-                StringBuffer buf("<Exception><Source>roxieclient</Source><Message>Unexpected exception</Message></Exception>");
-                immediateOutput(buf.length(), buf.str());
-                clearOutput();
-                return 0;
-            }
-        }
-
-        writeOutput();
-
-#ifdef DEBUG_ROXIECLIENT_
-        DBGLOG("Thread progress: %d bytes read, %d  bytes written", bytesRead, bytesWritten);
-#endif
-    }
-
-    return 0;
-}
-
-void CRoxieClient::runQuery(const char* query, bool trim)
-{
-    TIME_SECTION("CRoxieClient::runQuery");
-    IArrayOf<RoxieThread> threads;
-
-    if(!query || !query)
-        throw MakeStringException(-1, "query is empty");
-
-    StringBuffer qxml;
-    if(trim)
-    {
-        Owned<IPropertyTree> qtree = createPTreeFromXMLString(query);
-        if(qtree)
-        {
-            qtree->setPropBool("@trim", true);
-            toXML(qtree, qxml);
-        }
-        else
-            qxml.append(query);
-    }
-
-    unsigned i;
-
-    for (i=0; i< m_numThreads; i++)
-    {
-        RoxieThread* onethread = new RoxieThread(this, trim?qxml.str():query, m_recordsPerQuery, m_maxRetries, m_readTimeout);
-        threads.append(*onethread);
-        DBGLOG("Starting thread %d", i);
-        onethread->start();
-    }
-
-    unsigned totalRead = 0;
-    unsigned totalWritten = 0;
-
-    for (i=0; i< m_numThreads; i++)
-    {
-        DBGLOG("Waiting for thread %d to finish", i);
-        RoxieThread* onethread = &threads.item(i);
-        if(onethread == NULL)
-            continue;
-        onethread->join();
-        DBGLOG("Final stats for thread %d: %d bytes read, %d bytes written", i, onethread->bytesRead, onethread->bytesWritten);
-        totalRead += onethread->bytesRead;
-        totalWritten += onethread->bytesWritten;
-    }
-
-    DBGLOG("Final roxieclient stats: %d bytes read, %d bytes written", totalRead, totalWritten);
-}
-
-extern "C" ROXIECLIENT_API IRoxieClient* createRoxieClient(ISmartSocketFactory* socketfactory, int numThreads, int maxretries)
-{
-    return new CRoxieClient(socketfactory, numThreads, maxretries);
-}
-

roxie/roxieclient/roxieclient.hpp

@@ -1,54 +0,0 @@
-/*##############################################################################
-
-    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.
-############################################################################## */
-
-#ifndef _ROXIECLIENT_HPP__
-#define _ROXIECLIENT_HPP__
-
-#ifndef ROXIECLIENT_API
-
-#ifdef _WIN32
-    #ifndef ROXIECLIENT_EXPORTS
-        #define ROXIECLIENT_API __declspec(dllimport)
-    #else
-        #define ROXIECLIENT_API __declspec(dllexport)
-    #endif //ROXIECLIENT_EXPORTS
-#else
-    #define ROXIECLIENT_API
-#endif //_WIN32
-
-#endif 
-
-#include "jiface.hpp"
-#include "jsmartsock.hpp"
-
-const unsigned MAX_ROXIECLIENT_THREADS = 50;
-const unsigned DEFAULT_RECORDSPERQUERY = 10000;
-const unsigned BYTESPERBLOCK = 0x8000;
-
-interface IRoxieClient: implements IInterface
-{
-    virtual void setInput(const char* id, unsigned in_width, IByteInputStream* istream) = 0;
-    virtual void setOutput(const char* resultName, unsigned out_width, IByteOutputStream* ostream) = 0;
-    virtual void setMaxRetries(unsigned retries) = 0;
-    virtual void setReadTimeout(int readTimeout) = 0;
-    virtual void runQuery(const char* query, bool trim=true) = 0;
-    virtual void setRecordsPerQuery(unsigned recordsPerQuery) = 0;
-};
-
-extern "C" ROXIECLIENT_API IRoxieClient* createRoxieClient(ISmartSocketFactory* socketfactory, int numThreads, int maxretries=0);
-
-#endif

roxie/roxieclient/roxieclient.ipp

@@ -1,321 +0,0 @@
-/*##############################################################################
-
-    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.
-############################################################################## */
-
-#ifndef _ROXIECLIENT_IPP__
-#define _ROXIECLIENT_IPP__
-
-#include <stdio.h>
-#include <string.h>
-#include <fcntl.h>
-#include <stdlib.h>
-
-#include "jlib.hpp"
-#include "jsmartsock.hpp"
-#include "jmisc.hpp"
-#include "jqueue.tpp"
-#include "jstream.hpp"
-
-#include "roxieclient.hpp"
-
-
-#define MAXBUFSIZE 0x8000
-
-interface IReceivedRoxieException : extends IException
-{
-};
-
-class ReceivedRoxieException: public IReceivedRoxieException, public CInterface
-{
-public:
-    IMPLEMENT_IINTERFACE;
-
-    ReceivedRoxieException(int code, const char *_msg) : errcode(code), msg(_msg) { };
-    int errorCode() const { return (errcode); };
-    StringBuffer &  errorMessage(StringBuffer &str) const
-    {
-        return str.append("ReceivedRoxieException: (").append(msg).append(")");
-    };
-    MessageAudience errorAudience() const { return (MSGAUD_user); };
-
-private:
-    int errcode;
-    StringAttr msg;
-};
-
-interface IRoxieClientException : extends IException
-{
-};
-
-class RoxieClientException: public IRoxieClientException, public CInterface
-{
-public:
-    IMPLEMENT_IINTERFACE;
-
-    RoxieClientException(int code, const char *_msg) : errcode(code), msg(_msg) { };
-    int errorCode() const { return (errcode); };
-    StringBuffer &  errorMessage(StringBuffer &str) const
-    {
-        return str.append("RoxieClientException: (").append(msg).append(")");
-    };
-    MessageAudience errorAudience() const { return (MSGAUD_user); };
-
-private:
-    int errcode;
-    StringAttr msg;
-};
-
-
-class CDataInput : public CInterface
-{
-public:
-    CDataInput(const char* _name, int iw, IByteInputStream* _stream)
-    {
-        name.append(_name);
-        record_width = iw;
-        stream.set(_stream);
-    }
-
-    StringBuffer name;
-    int          record_width;
-    Linked<IByteInputStream> stream;
-    Mutex        readMutex; 
-};
-
-class CDataOutput : public CInterface
-{
-public:
-    CDataOutput(const char* _name, int ow, IByteOutputStream* _stream)
-    {
-        name.append(_name);
-        record_width = ow;
-        stream.set(_stream);
-    }
-
-    StringBuffer name;
-    int          record_width;
-    Linked<IByteOutputStream> stream;
-    Mutex        writeMutex;
-};
-
-class CDataInputCache : public CInterface, implements IInterface
-{
-public:
-    IMPLEMENT_IINTERFACE;
-    CDataInputCache(CDataInput* _input, unsigned _recordsPerQuery)
-    {
-        recordsPerQuery = _recordsPerQuery;
-        inputLen = 0;
-        curInput = 0;
-        if(_input != NULL)
-        {
-            name.append(_input->name);
-            input = _input;
-
-            if(input->record_width == 0)
-            {
-                bytesPerQuery = 0;
-                recordsPerBlock = 0;
-            }
-            else
-            {
-                bytesPerQuery = recordsPerQuery * _input->record_width;
-                recordsPerBlock = BYTESPERBLOCK / _input->record_width;
-                recordsPerBlock = (recordsPerBlock < 1)? 1: recordsPerBlock;
-                buffer.reserve(bytesPerQuery);
-            }
-        }
-    }
-
-    int readInput();
-
-    StringBuffer name;
-    int          inputLen;
-    int          curInput;
-    StringBuffer buffer;
-    unsigned     recordsPerQuery;
-    unsigned     bytesPerQuery;
-    unsigned     recordsPerBlock;
-    CDataInput   *input;
-};
-
-
-class CDataOutputCache : public CInterface, implements IInterface
-{
-public:
-    IMPLEMENT_IINTERFACE;
-
-    CDataOutputCache(CDataOutput* _output)
-    {
-        isNonBlocked = false;
-
-        if(_output != NULL)
-        {
-            name.append(_output->name);
-            output = _output;
-        }
-    }
-    virtual ~CDataOutputCache()
-    {
-        while (outputQ.ordinality() > 0)
-            delete outputQ.dequeue();
-    }
-
-    int writeOutput();
-    void clearOutput();
-    int immediateOutput(size32_t len, const char* result);
-    StringBuffer name;
-    QueueOf<MemoryBuffer, false> outputQ;
-    CDataOutput *output;
-    bool isNonBlocked;
-};
-
-class CRoxieClient;
-
-class RoxieThread : public Thread
-{
-private:
-    Owned<ISmartSocket> roxieSock;
-    StringBuffer query;
-
-    CRoxieClient* roxieclient;
-    
-    Owned<CDataInputCache>  m_default_incache;
-    Owned<CDataOutputCache> m_default_outcache;
-
-    IArrayOf<CDataInputCache>  m_incaches;
-    IArrayOf<CDataOutputCache> m_outcaches;
-    
-    unsigned m_recordsPerQuery;
-
-    unsigned m_maxRetries;
-    int      m_readTimeout;
-
-public:
-    RoxieThread(CRoxieClient* _roxieclient, const char *_query, unsigned recordsPerQuery, unsigned maxRetries, int readTimeout);
-
-    void sendQuery();
-    bool readInput();
-    void writeOutput();
-    void clearOutput();
-    void immediateOutput(size32_t len, const char* result);
-    void sendData(const char *reqbuf);
-    void processQuery();
-    int run();
-
-    unsigned     bytesRead;
-    unsigned     bytesWritten;
-};
-
-class CRoxieClient : public CInterface, implements IRoxieClient
-{
-public:
-    IMPLEMENT_IINTERFACE;
-
-    Mutex readMutex;    
-    Mutex writeMutex;
-
-    CRoxieClient(ISmartSocketFactory* socketfactory, int numThreads, int maxretries)
-    {
-        m_smartSocketFactory.set(socketfactory);
-        m_numThreads = (numThreads < 1)? 1: ((numThreads > MAX_ROXIECLIENT_THREADS)? MAX_ROXIECLIENT_THREADS: numThreads);
-        m_recordsPerQuery = DEFAULT_RECORDSPERQUERY;
-        m_maxRetries = maxretries;
-        m_readTimeout = 300;
-    }
-
-    virtual ~CRoxieClient()
-    {
-    }
-
-    virtual ISmartSocketFactory* querySocketFactory()
-    {
-        return m_smartSocketFactory.get();
-    }
-
-    virtual void setInput(const char* id, unsigned in_width, IByteInputStream* istream)
-    {
-        if(istream != NULL)
-        {
-            if(id == NULL || id[0] == '\0')
-            {
-                m_default_input.setown(new CDataInput("", in_width, istream));
-            }
-            else
-            {
-                m_inputs.append(*(new CDataInput(id, in_width, istream)));
-            }
-        }
-        else
-        {
-            ERRLOG("can't set input stream, it's NULL");
-        }
-    }
-    
-    virtual void setOutput(const char* resultName, unsigned out_width, IByteOutputStream* ostream)
-    {
-        if(ostream != NULL)
-        {
-            if(resultName == NULL || resultName[0] == '\0')
-            {
-                m_default_output.setown(new CDataOutput("", out_width, ostream));
-            }
-            else
-            {
-                m_outputs.append(*(new CDataOutput(resultName, out_width, ostream)));
-            }
-        }
-        else
-        {
-            ERRLOG("can't set output stream, it's NULL");
-        }
-    }
-    virtual void setRecordsPerQuery(unsigned recordsPerQuery)
-    {
-        m_recordsPerQuery = recordsPerQuery;
-    }
-
-    virtual void setMaxRetries(unsigned retries)
-    {
-        m_maxRetries = retries;
-    }
-
-    virtual void setReadTimeout(int readTimeout)
-    {
-        m_readTimeout = readTimeout;
-    }
-
-    virtual void runQuery(const char* query, bool trim);
-
-private:
-    Linked<ISmartSocketFactory> m_smartSocketFactory;
-    unsigned                    m_numThreads;
-    unsigned                    m_maxRetries;
-    int                         m_readTimeout;
-
-protected:
-    CIArrayOf<CDataInput>       m_inputs;
-    CIArrayOf<CDataOutput>      m_outputs;
-    Owned<CDataInput>           m_default_input;
-    Owned<CDataOutput>          m_default_output;
-    unsigned                    m_recordsPerQuery;
-
-friend class RoxieThread;
-
-};
-
-#endif
-

roxie/roxieclient/sourcedoc.xml

@@ -1,26 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<!--
-################################################################################
-#    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.
-################################################################################
--->
-<!DOCTYPE section PUBLIC "-//OASIS//DTD DocBook XML V4.3//EN" "http://www.oasis-open.org/docbook/xml/4.3/docbookx.dtd">
-<section>
-    <title>roxie/roxieclient</title>
-
-    <para>
-        The roxie/roxieclient directory contains the sources for the roxie/roxieclient library.
-    </para>
-</section>

roxie/roxieclienttest/CMakeLists.txt

@@ -1,44 +0,0 @@
-################################################################################
-#    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.
-################################################################################
-
-# Component: roxieclienttest 
-#####################################################
-# Description:
-# ------------
-#    Cmake Input File for roxieclienttest
-#####################################################
-
-project( roxieclienttest ) 
-
-set (    SRCS 
-         main.cpp 
-    )
-
-include_directories ( 
-         ./../../system/include 
-         ./../../system/jlib 
-         ./../roxieclient 
-    )
-
-ADD_DEFINITIONS( -D_CONSOLE )
-
-HPCC_ADD_EXECUTABLE ( roxieclienttest ${SRCS} )
-target_link_libraries ( roxieclienttest 
-         jlib
-         roxieclient 
-    )
-
-

roxie/roxieclienttest/main.cpp

@@ -1,190 +0,0 @@
-/*##############################################################################
-
-    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 "jliball.hpp"
-#include "roxieclient.hpp"
-
-int main(int argc, char *argv[])
-{
-    InitModuleObjects();
-
-    Owned<ISmartSocketFactory> smartSocketFactory;
-    int in_width = 0;
-    int out_width = 0;
-    int recordsPerQuery = 0;
-    StringBuffer query;
-    StringBuffer hosts;
-    StringBuffer resultName;
-    attachStandardFileLogMsgMonitor("C:\\roxieclienttest.log", NULL, MSGFIELD_STANDARD, MSGAUD_all, MSGCLS_all, TopDetail, false);
-    unsigned numThreads = 1;
-    EnableSEHtoExceptionMapping();
-    StringBuffer inputfile, outputfile;
-    unsigned maxRetries = 0;
-    int readTimeout = 300;
-
-    int i;
-
-    for (i=1; i<argc; i++)
-    {
-        if (stricmp(argv[i], "-iw") == 0)
-        {
-            i++;
-            if (i < argc)
-                in_width = atoi(argv[i]);
-        }
-        else if (stricmp(argv[i], "-ow") == 0)
-        {
-            i++;
-            if (i < argc)
-                out_width = atoi(argv[i]);
-        }
-        else if (stricmp(argv[i], "-q") == 0)
-        {
-            i++;
-            if (i < argc)
-                query.append(argv[i]);
-        }
-        else if (stricmp(argv[i], "-h") == 0)
-        {
-            i++;
-            if (i < argc)
-                hosts.append(argv[i]);
-        }
-        else if (stricmp(argv[i], "-r") == 0)
-        {
-            i++;
-            if (i < argc)
-                resultName.append(argv[i]);
-        }
-        else if (stricmp(argv[i], "-i") == 0)
-        {
-            i++;
-            if (i < argc)
-                inputfile.append(argv[i]);
-        }
-        else if (stricmp(argv[i], "-o") == 0)
-        {
-            i++;
-            if (i < argc)
-                outputfile.append(argv[i]);
-        }
-        else if (stricmp(argv[i], "-n") == 0)
-        {
-            i++;
-        }
-        else if (stricmp(argv[i], "-t") == 0)
-        {
-            i++;
-            if (i < argc)
-                numThreads = atoi(argv[i]);
-        }
-        else if (stricmp(argv[i], "-b") == 0)
-        {
-            i++;
-            if (i < argc)
-                recordsPerQuery = atoi(argv[i]);
-        }
-        else if (stricmp(argv[i], "-mr") == 0)
-        {
-            i++;
-            if (i <argc)
-                maxRetries = atoi(argv[i]);
-        }
-        else if (stricmp(argv[i], "-to") == 0)
-        {
-            i++;
-            if (i <argc)
-                readTimeout = atoi(argv[i]);
-        }
-#if 0
-        else if (stricmp(argv[i], "-d") == 0)
-        {
-            DebugBreak();
-        }
-#endif
-    }
-
-    if (query.length() == 0 || hosts.length() == 0)
-    {
-        DBGLOG("Missing required parameter (-q & -h are all required)");
-        exit(0);
-    }
-    
-    numThreads = (numThreads < 1)? 1: ((numThreads > MAX_ROXIECLIENT_THREADS)? MAX_ROXIECLIENT_THREADS: numThreads);
-    recordsPerQuery = (recordsPerQuery < 1)? DEFAULT_RECORDSPERQUERY : recordsPerQuery;
-
-
-    smartSocketFactory.setown(createSmartSocketFactory(hosts.str()));
-    Owned<IRoxieClient> roxieclient = createRoxieClient(smartSocketFactory, numThreads);
-    
-    // File input/output stream example
-    int fh1 = -1;
-    Owned<IByteInputStream> istream = NULL;
-    if(inputfile.length() > 0)
-    {
-        fh1 = open( inputfile.str(), _O_RDONLY );
-        if( fh1 == -1 )
-        {
-            perror( "Open failed on input file" );
-            exit(-1);
-        }
-        else
-        {
-            printf( "Open succeeded on input file\n" );
-        }
-#ifdef _WIN32
-        _setmode(fh1, _O_BINARY);
-#endif
-        Owned<IByteInputStream> istream = createInputStream(fh1);
-        roxieclient->setInput("", in_width, istream);
-
-    }
-
-    int fh2;
-    fh2 = open(outputfile.str(), _O_WRONLY | _O_CREAT | _O_TRUNC, _S_IREAD | _S_IWRITE );
-    if( fh2 == -1 )
-    {
-        perror( "Open failed on output file" );
-        exit(-1);
-    }
-    else
-    {
-        printf( "Open succeeded on output file\n" );
-    }
-
-#ifdef _WIN32
-    _setmode(fh2, _O_BINARY);
-#endif
-
-    Owned<IByteOutputStream> ostream = createOutputStream(fh2);
-    roxieclient->setRecordsPerQuery(recordsPerQuery);
-    roxieclient->setOutput("", out_width, ostream);
-    roxieclient->setMaxRetries(maxRetries);
-    roxieclient->setReadTimeout(readTimeout);
-    roxieclient->runQuery(query.str());
-
-    if(smartSocketFactory.get())
-        smartSocketFactory->stop();
-
-    if(fh1 != -1)   
-        _close( fh1 );
-    _close( fh2 );
-
-    releaseAtoms();
-
-    return 0;
-}

roxie/roxieclienttest/sourcedoc.xml

@@ -1,26 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<!--
-################################################################################
-#    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.
-################################################################################
--->
-<!DOCTYPE section PUBLIC "-//OASIS//DTD DocBook XML V4.3//EN" "http://www.oasis-open.org/docbook/xml/4.3/docbookx.dtd">
-<section>
-    <title>roxie/roxieclienttest</title>
-
-    <para>
-        The roxie/roxieclienttest directory contains the sources for the roxie/roxieclienttest library.
-    </para>
-</section>

roxie/roxiemem/roxiemem.cpp

@@ -6254,7 +6254,7 @@ protected:
         Semaphore & sem;
         const unsigned size;
         const unsigned numThreads;
-        volatile memsize_t final;
+        volatile memsize_t final = 0;
     };
     void testBitmapThreading(unsigned size, unsigned numThreads)
     {

roxie/roxiepipe/roxiepipe.cpp

@@ -79,16 +79,16 @@ private:
 
     QueueOf<MemoryBuffer, false> outputQ;
 
-    char *inputBuffer;
-    unsigned inputLen;
-    unsigned curInput;
+    char *inputBuffer = nullptr;;
+    unsigned inputLen = 0;
+    unsigned curInput = 0;
 
-    unsigned bytesPerQuery;
-    unsigned recordsPerBlock;
+    unsigned bytesPerQuery = 0;
+    unsigned recordsPerBlock = 0;
 
 public:
-    unsigned recordsRead;
-    unsigned recordsWritten;
+    unsigned recordsRead = 0;
+    unsigned recordsWritten = 0;
 
     RoxieThread(const char *_query, const char *_resultName) : query(_query), resultName(_resultName)
     {
@@ -101,9 +101,7 @@ public:
     void sendQuery()
     {
         MemoryBuffer sendBuffer;
-
         unsigned queryLen = query.length();
-
         unsigned revQueryLen = queryLen;
         _WINREV(revQueryLen);
         sendBuffer.append(revQueryLen);

roxie/udplib/udpsha.cpp

@@ -108,6 +108,12 @@ queue_t::queue_t(unsigned int queue_s)
 queue_t::queue_t() 
 {
     signal_free_sl = 0;
+    queue_size = 0;
+    element_count = 0;
+    elements = nullptr;
+    active_buffers = 0;
+    first = 0;
+    last = 0;
 }
 
 queue_t::~queue_t() 

roxie/udplib/udpsha.hpp

@@ -69,7 +69,6 @@ class queue_t
     Semaphore       free_space;
     Semaphore       free_sl;
     unsigned        signal_free_sl;
-    int             total_data;
 
     void removeElement(int ix);
     
@@ -81,7 +80,7 @@ public:
     bool dataQueued(void *key, PKT_CMP_FUN pkCmpFn);
     bool removeData(void *key, PKT_CMP_FUN pkCmpFn);
     int  free_slots(); //block if no free slots
-    void set_queue_size(unsigned int queue_size); //must be called after immitialy after contructor if default contructor is used
+    void set_queue_size(unsigned int queue_size); //must be called immediately after constructor if default constructor is used
     queue_t(unsigned int queue_size);
     queue_t();
     ~queue_t();

roxie/udplib/udptrr.cpp

@@ -605,6 +605,7 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             flow_socket->set_receive_buffer_size(udpFlowSocketsSize);
             size32_t actualSize = flow_socket->get_receive_buffer_size();
             DBGLOG("UdpReceiver: rcv_flow_socket created port=%d sockbuffsize=%d actual %d", flow_port, udpFlowSocketsSize, actualSize);
+            running = false;
         }
         
         ~receive_receive_flow() 
@@ -698,6 +699,7 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             receive_socket->set_receive_buffer_size(ip_buffer);
             size32_t actualSize = receive_socket->get_receive_buffer_size();
             DBGLOG("UdpReceiver: rcv_data_socket created port=%d requested sockbuffsize=%d actual sockbuffsize=%d", parent.data_port, ip_buffer, actualSize);
+            running = false;
         }
 
         virtual void start()
@@ -838,6 +840,8 @@ public:
         receive_flow = new receive_receive_flow(*this, server_flow_port);
         if (udpSnifferEnabled)
             sniffer = new receive_sniffer(*this, snif_port, multicast_ip, getNumNodes());
+        else
+            sniffer = nullptr;
 
         running = true;
         collatorThread.start();

roxie/udplib/udptrs.cpp

@@ -370,6 +370,7 @@ public:
         retryDataCount = 0;
         retryDataIdx = 0;
         retryData = NULL;
+        maxRetryData = 0;
     }
 
     void init(unsigned destNodeIndex, unsigned _numQueues, unsigned queueSize, unsigned _maxRetryData, unsigned sendFlowPort, unsigned dataPort, bool isLocal)
@@ -1028,7 +1029,6 @@ class CMessagePacker : implements IMessagePacker, public CInterface
     unsigned mem_buffer_size;
     unsigned totalSize;
     bool            packed_request;
-    unsigned        requested_size;
     MemoryBuffer    metaInfo;
     bool            last_message_done;
     int             queue_number;