Merge pull request #14026 from mayx/HPCC-17585-SoapcallPersist

HPCC-17585 SOAPCALL support for persistent outgoing connections

Reviewed-By: Anthony Fishbeck <anthony.fishbeck@lexisnexis.com>
Reviewed-By: Jake Smith <jake.smith@lexisnexis.com>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

common/thorhelper/CMakeLists.txt

@@ -68,6 +68,8 @@ set (    SRCS
          thorrparse.ipp
          thorsort.cpp
          thorsort.hpp
+         persistent.cpp
+         persistent.hpp
          
          roxiedebug.hpp
          roxiedebug.ipp
@@ -80,24 +82,24 @@ set (    SRCS
 include_directories ( 
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
-         ./../../system/security/securesocket 
-         ./../../common/remote 
-         ./../../system/jhtree 
-         ./../../rtl/eclrtl 
-         ./../../rtl/nbcd 
-         ./../../system/include 
-         ./../../system/mp
-         ./../../system/jlib
-         ./../../system/security/zcrypt
-         ./../../dali/base
-         ./../deftype 
-         ./../workunit
-         ./../../rtl/include 
-         ./../../ecl/hql
-         ./../../roxie/roxiemem
-         ./../../testing/unittests
-         ./../../system/tbb_sm/tbb/include
-         ./../../system/security/shared
+         ${HPCC_SOURCE_DIR}/system/security/securesocket
+         ${HPCC_SOURCE_DIR}/common/remote
+         ${HPCC_SOURCE_DIR}/system/jhtree
+         ${HPCC_SOURCE_DIR}/rtl/eclrtl
+         ${HPCC_SOURCE_DIR}/rtl/nbcd
+         ${HPCC_SOURCE_DIR}/system/include
+         ${HPCC_SOURCE_DIR}/system/mp
+         ${HPCC_SOURCE_DIR}/system/jlib
+         ${HPCC_SOURCE_DIR}/system/security/zcrypt
+         ${HPCC_SOURCE_DIR}/dali/base
+         ${HPCC_SOURCE_DIR}/common/deftype
+         ${HPCC_SOURCE_DIR}/common/workunit
+         ${HPCC_SOURCE_DIR}/rtl/include
+         ${HPCC_SOURCE_DIR}/ecl/hql
+         ${HPCC_SOURCE_DIR}/roxie/roxiemem
+         ${HPCC_SOURCE_DIR}/testing/unittests
+         ${HPCC_SOURCE_DIR}/system/tbb_sm/tbb/include
+         ${HPCC_SOURCE_DIR}/system/security/shared
          ${HPCC_SOURCE_DIR}/fs/dafsclient
     )
 

esp/platform/persistent.cpp

@@ -20,6 +20,7 @@
 #include "jdebug.hpp"
 #include "jlog.hpp"
 #include <memory>
+#include <unordered_set>
 #ifdef _WIN32
 #include <windows.h>
 #include <winsock2.h>
@@ -27,30 +28,151 @@
 #endif
 
 #define PERSILOG(loglevel, ...) if(static_cast<int>(loglevel) <= static_cast<int>(m_loglevel)) DBGLOG(__VA_ARGS__)
+#define MAX_NO_CLEANUP_SOCKETSETS 1000
+
+static inline StringBuffer& addKeySuffix(PersistentProtocol proto, StringBuffer& keystr)
+{
+    switch (proto)
+    {
+        case PersistentProtocol::ProtoTCP:
+            break;
+        case PersistentProtocol::ProtoTLS:
+            keystr.append('~');
+            break;
+        default:
+            throw makeStringException(-1, "New suffix should be defined");
+    }
+    return keystr;
+}
 
 class CPersistentInfo : implements IInterface, public CInterface
 {
     friend class CPersistentHandler;
+    friend class CAvailKeeper;
 public:
     IMPLEMENT_IINTERFACE;
-    CPersistentInfo(bool _inUse, unsigned _timeUsed, unsigned _useCount, SocketEndpoint* _ep)
-        : inUse(_inUse), timeUsed(_timeUsed), useCount(_useCount), ep(_ep?(new SocketEndpoint(*_ep)):nullptr)
+    CPersistentInfo(bool _inUse, unsigned _timeUsed, unsigned _useCount, SocketEndpoint* _ep, PersistentProtocol _proto, ISocket* _sock)
+        : inUse(_inUse), timeUsed(_timeUsed), useCount(_useCount), ep(_ep?(new SocketEndpoint(*_ep)):nullptr), proto(_proto), sock(_sock)
     {
         if(_ep)
+        {
             _ep->getUrlStr(epstr);
+            keystr.set(epstr);
+            addKeySuffix(proto, keystr);
+        }
     }
-    virtual ~CPersistentInfo() { } //TODO remove trace
 protected:
     bool inUse;
     unsigned timeUsed;
     unsigned useCount;
     std::unique_ptr<SocketEndpoint> ep;
     StringBuffer epstr;
+    StringBuffer keystr;
+    PersistentProtocol proto;
+    Linked<ISocket> sock;
+};
+
+struct LinkedPersistentInfoHash
+{
+    size_t operator()(const Linked<CPersistentInfo>& linkedinfo) const
+    {
+        return std::hash<CPersistentInfo*>()(linkedinfo);
+    }
+};
+
+using SocketSet = std::unordered_set<Linked<CPersistentInfo>, LinkedPersistentInfoHash>;
+using EpSocketSetMap = MapStringTo<OwnedPtr<SocketSet>, SocketSet*>;
+using EpSocketSetMapping = MappingStringTo<OwnedPtr<SocketSet>, SocketSet*>;
+
+class CAvailKeeper
+{
+private:
+    SocketSet m_avail;
+    EpSocketSetMap m_avail4ep;
+public:
+    void add(CPersistentInfo* sockinfo)
+    {
+        findSet(sockinfo, true)->insert(sockinfo);
+        if (m_avail4ep.count() > MAX_NO_CLEANUP_SOCKETSETS)
+            cleanup();
+    }
+    void remove(CPersistentInfo* sockinfo)
+    {
+        SocketSet* sset = findSet(sockinfo);
+        if (sset)
+            sset->erase(sockinfo);
+    }
+    CPersistentInfo* get(SocketEndpoint* ep, PersistentProtocol proto)
+    {
+        SocketSet* sset = findSet(ep, proto);
+        if (sset)
+        {
+            //The first available socket will suffice
+            auto iter = sset->begin();
+            if (iter != sset->end())
+            {
+                Linked<CPersistentInfo> info = *iter;
+                sset->erase(iter);
+                return info.getClear();
+            }
+        }
+        return nullptr;
+    }
+private:
+    inline StringBuffer& calcKey(SocketEndpoint& ep, PersistentProtocol proto, StringBuffer& keystr)
+    {
+        ep.getUrlStr(keystr);
+        return addKeySuffix(proto, keystr);
+    }
+    SocketSet* findSet(CPersistentInfo* info, bool create = false)
+    {
+        if (!info->ep.get())
+            return &m_avail;
+        return findSet(info->keystr.str(), create);
+    }
+    SocketSet* findSet(SocketEndpoint* ep, PersistentProtocol proto, bool create = false)
+    {
+        if (!ep)
+            return &m_avail;
+        StringBuffer keystr;
+        calcKey(*ep, proto, keystr);
+        return findSet(keystr.str(), create);
+    }
+    SocketSet* findSet(const char* key, bool create = false)
+    {
+        auto ptrptr = m_avail4ep.getValue(key);
+        if (ptrptr)
+            return *ptrptr;
+        else if (create)
+        {
+            SocketSet* sset = new SocketSet();
+            m_avail4ep.setValue(key, sset);
+            return sset;
+        }
+        return nullptr;
+    }
+    void cleanup()
+    {
+        std::vector<EpSocketSetMapping*> elems;
+        for (auto& e : m_avail4ep)
+        {
+            if (e.getValue()->empty())
+                elems.push_back(&e);
+        }
+        for (auto& e : elems)
+            m_avail4ep.removeExact(e);
+    }
 };
 
 using SockInfoMap = MapBetween<Linked<ISocket>, ISocket*, Owned<CPersistentInfo>, CPersistentInfo*>;
 using StringIntMap = MapStringTo<int, int>;
 
+// Important data structures for the implementation:
+// m_selecHandler: used to detect incoming data on a socket or socket closure from the other end
+// m_infomap: keep track of the status of a socket, there's one entry for each reusable socket, no matter if the socket is being used or idle. The main purpose is to
+//   keep track of the status and life span of the socket so that it can be recycled properly.
+// m_availkeeper: keep track of available sockets that can be assigned for reusing. It's a map between <endpoint, protocal> and the set of available sockets. The
+//   main purpose is to speed up finding an available socket
 class CPersistentHandler : implements IPersistentHandler, implements ISocketSelectNotify, public Thread
 {
 private:
@@ -62,7 +184,8 @@ private:
     Semaphore m_waitsem;
     bool m_stop = false;
     SockInfoMap m_infomap;
-    Mutex m_mutex;
+    CAvailKeeper m_availkeeper;
+    CriticalSection m_critsect;
     PersistentLogLevel m_loglevel = PersistentLogLevel::PLogNormal;
     static int CurID;
     int m_id = 0;
@@ -82,12 +205,12 @@ public:
     {
     }
 
-    virtual void add(ISocket* sock, SocketEndpoint* ep = nullptr) override
+    virtual void add(ISocket* sock, SocketEndpoint* ep = nullptr, PersistentProtocol proto = PersistentProtocol::ProtoTCP) override
     {
         if (!sock || sock->OShandle() == INVALID_SOCKET)
             return;
-        synchronized block(m_mutex);
-        PERSILOG(PersistentLogLevel::PLogNormal, "PERSISTENT: adding socket %d to handler %d", sock->OShandle(), m_id);
+        PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: adding socket %d to handler %d", sock->OShandle(), m_id);
+        CriticalBlock block(m_critsect);
         if (m_enableDoNotReuseList && ep != nullptr)
         {
             StringBuffer epstr;
@@ -101,29 +224,36 @@ public:
             }
         }
         m_selectHandler->add(sock, SELECTMODE_READ, this);
-        m_infomap.setValue(sock, new CPersistentInfo(false, usTick()/1000, 0, ep));
+        Owned<CPersistentInfo> info = new CPersistentInfo(false, usTick()/1000, 0, ep, proto, sock);
+        m_infomap.setValue(sock, info.getLink());
+        m_availkeeper.add(info);
     }
 
     virtual void remove(ISocket* sock) override
     {
         if (!sock)
             return;
-        PERSILOG(PersistentLogLevel::PLogNormal, "PERSISTENT: Removing socket %d from handler %d", sock->OShandle(), m_id);
-        synchronized block(m_mutex);
+        PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: Removing socket %d from handler %d", sock->OShandle(), m_id);
+        CriticalBlock block(m_critsect);
         Owned<CPersistentInfo>* val = m_infomap.getValue(sock);
         CPersistentInfo* info = nullptr;
         if (val)
             info = *val;
         bool removedFromSelectHandler = info && info->inUse; //If inUse sock was already removed from select handler
         if (info)
+        {
+            if (!info->inUse)
+                m_availkeeper.remove(info);
             m_infomap.remove(sock);
+        }
         if (!removedFromSelectHandler)
             m_selectHandler->remove(sock);
     }
 
     virtual void doneUsing(ISocket* sock, bool keep, unsigned usesOverOne) override
     {
-        synchronized block(m_mutex);
+        PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: Done using socket %d, keep=%s", sock->OShandle(), boolToStr(keep));
+        CriticalBlock block(m_critsect);
         Owned<CPersistentInfo>* val = m_infomap.getValue(sock);
         CPersistentInfo* info = nullptr;
         if (val)
@@ -139,11 +269,12 @@ public:
                 info->inUse = false;
                 info->timeUsed = usTick()/1000;
                 m_selectHandler->add(sock, SELECTMODE_READ, this);
+                m_availkeeper.add(info);
             }
             else
             {
                 if (reachedQuota)
-                    PERSILOG(PersistentLogLevel::PLogMin, "PERSISTENT: Socket %d reached quota", sock->OShandle());
+                    PERSILOG(PersistentLogLevel::PLogNormal, "PERSISTENT: Socket %d reached quota", sock->OShandle());
                 if(!keep)
                     PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: Indicated not to keep socket %d", sock->OShandle());
                 remove(sock);
@@ -151,27 +282,21 @@ public:
         }
     }
 
-    virtual Linked<ISocket> getAvailable(SocketEndpoint* ep = nullptr, bool* pShouldClose = nullptr) override
+    virtual ISocket* getAvailable(SocketEndpoint* ep = nullptr, bool* pShouldClose = nullptr, PersistentProtocol proto =  PersistentProtocol::ProtoTCP) override
     {
-        synchronized block(m_mutex);
-        for (auto& si:m_infomap)
+        CriticalBlock block(m_critsect);
+        Owned<CPersistentInfo> info = m_availkeeper.get(ep, proto);
+        if (info)
         {
-            CPersistentInfo* info = si.getValue();
-            if (info && !info->inUse && (ep == nullptr || (info->ep != nullptr && *(info->ep) == *ep)))
-            {
-                ISocket* sock = *(ISocket**)(si.getKey());
-                if (sock)
-                {
-                    info->inUse = true;
-                    info->timeUsed = usTick()/1000;
-                    info->useCount++;
-                    if (pShouldClose != nullptr)
-                        *pShouldClose = m_maxReqs > 0 && m_maxReqs <= info->useCount;
-                    m_selectHandler->remove(sock);
-                    PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: Obtained persistent socket %d from handler %d", sock->OShandle(), m_id);
-                    return sock;
-                }
-            }
+            Linked<ISocket> sock = info->sock;
+            info->inUse = true;
+            info->timeUsed = usTick()/1000;
+            info->useCount++;
+            if (pShouldClose != nullptr)
+                *pShouldClose = m_maxReqs > 0 && m_maxReqs <= info->useCount;
+            m_selectHandler->remove(sock);
+            PERSILOG(PersistentLogLevel::PLogMax, "PERSISTENT: Obtained persistent socket %d from handler %d", info->sock->OShandle(), m_id);
+            return sock.getClear();
         }
         return nullptr;
     }
@@ -182,10 +307,10 @@ public:
         size32_t x = sock->avail_read();
         if (x == 0)
         {
-            PERSILOG(PersistentLogLevel::PLogMin, "PERSISTENT: Detected closing of connection %d from the other end", sock->OShandle());
+            PERSILOG(PersistentLogLevel::PLogNormal, "PERSISTENT: Detected closing of connection %d from the other end", sock->OShandle());
             if (m_enableDoNotReuseList)
             {
-                synchronized block(m_mutex);
+                CriticalBlock block(m_critsect);
                 Owned<CPersistentInfo>* val = m_infomap.getValue(sock);
                 CPersistentInfo* info = nullptr;
                 if (val)
@@ -220,13 +345,14 @@ public:
             Owned<ISocket> mysock(LINK(sock));
             PERSILOG(PersistentLogLevel::PLogMax, "Data arrived on persistent connection %d", sock->OShandle());
             {
-                synchronized block(m_mutex);
+                CriticalBlock block(m_critsect);
                 Owned<CPersistentInfo>* val = m_infomap.getValue(sock);
                 CPersistentInfo* info = nullptr;
                 if (val)
                     info = *val;
                 if (info)
                 {
+                    m_availkeeper.remove(info);
                     info->inUse = true;
                     info->timeUsed = usTick()/1000;
                     info->useCount++;
@@ -261,7 +387,7 @@ public:
             if (m_stop)
                 break;
             unsigned now = usTick()/1000;
-            synchronized block(m_mutex);
+            CriticalBlock block(m_critsect);
             std::vector<ISocket*> socks1;
             std::vector<ISocket*> socks2;
             for (auto& si:m_infomap)
@@ -276,7 +402,7 @@ public:
             }
             for (auto& s:socks1)
             {
-                PERSILOG(PersistentLogLevel::PLogMin, "PERSISTENT: Socket %d has been idle for %d seconds so remove it", s->OShandle(), m_maxIdleTime);
+                PERSILOG(PersistentLogLevel::PLogNormal, "PERSISTENT: Socket %d has been idle for %d seconds so remove it", s->OShandle(), m_maxIdleTime);
                 remove(s);
             }
             for (auto& s:socks2)
@@ -311,6 +437,20 @@ public:
 
 };
 
+bool isHttpPersistable(const char* httpVer, const char* conHeader)
+{
+    if (isEmptyString(httpVer))
+        return false;
+    if (!isEmptyString(conHeader))
+    {
+        if (strieq(conHeader, "close"))
+            return false;
+        else if (strieq(conHeader, "Keep-Alive"))
+            return true;
+    }
+    return !streq(httpVer, "1.0");
+}
+
 int CPersistentHandler::CurID = 0;
 
 IPersistentHandler* createPersistentHandler(IPersistentSelectNotify* notify, int maxIdleTime, int maxReqs, PersistentLogLevel loglevel, bool enableDoNotReuseList)

esp/platform/persistent.hpp

@@ -18,20 +18,30 @@
 #ifndef __PERSISTENT_HPP__
 #define __PERSISTENT_HPP__
 
+#ifdef THORHELPER_EXPORTS
+ #define THORHELPER_API DECL_EXPORT
+#else
+ #define THORHELPER_API DECL_IMPORT
+#endif
+
 #include "jlib.hpp"
 #include "jsocket.hpp"
 
 #define DEFAULT_MAX_PERSISTENT_IDLE_TIME 60
 #define DEFAULT_MAX_PERSISTENT_REQUESTS  100
 
-enum class PersistentLogLevel { PLogNone=0, PLogMin=1, PLogNormal=5, PLogMax=10};
-
+enum class PersistentLogLevel { PLogNone=0, PLogMin=1, PLogNormal=5, PLogMax=10 };
+enum class PersistentProtocol { ProtoTCP=0, ProtoTLS=1 };
 interface IPersistentHandler : implements IInterface
 {
-    virtual void add(ISocket* sock, SocketEndpoint* ep = nullptr) = 0;
+    // Add a new socket to the pool for reuse
+    virtual void add(ISocket* sock, SocketEndpoint* ep = nullptr, PersistentProtocol proto = PersistentProtocol::ProtoTCP) = 0;
+    // Remove a socket from the pool
     virtual void remove(ISocket* sock) = 0;
+    // Put a socket back to the pool for further reuse, or remove its record from the pool when "keep" is false
     virtual void doneUsing(ISocket* sock, bool keep, unsigned usesOverOne = 0) = 0;
-    virtual Linked<ISocket> getAvailable(SocketEndpoint* ep = nullptr, bool* pShouldClose = nullptr) = 0;
+    // Get one available socket from the pool
+    virtual ISocket* getAvailable(SocketEndpoint* ep = nullptr, bool* pShouldClose = nullptr, PersistentProtocol proto = PersistentProtocol::ProtoTCP) = 0;
     virtual void stop(bool wait) = 0;
     virtual bool inDoNotReuseList(SocketEndpoint* ep) = 0;
 };
@@ -41,6 +51,8 @@ interface IPersistentSelectNotify
     virtual bool notifySelected(ISocket *sock,unsigned selected, IPersistentHandler* handler, bool shouldClose) = 0;
 };
 
-IPersistentHandler* createPersistentHandler(IPersistentSelectNotify* notify, int maxIdleTime = DEFAULT_MAX_PERSISTENT_IDLE_TIME, int maxReqs = DEFAULT_MAX_PERSISTENT_REQUESTS, PersistentLogLevel loglevel=PersistentLogLevel::PLogMin, bool enableDoNotReuseList=false);
+inline size32_t httpVerOffset() { return 5; /*strlen("HTTP ")*/ }
+THORHELPER_API bool isHttpPersistable(const char* httpVer, const char* conHeader);
+THORHELPER_API IPersistentHandler* createPersistentHandler(IPersistentSelectNotify* notify, int maxIdleTime = DEFAULT_MAX_PERSISTENT_IDLE_TIME, int maxReqs = DEFAULT_MAX_PERSISTENT_REQUESTS, PersistentLogLevel loglevel=PersistentLogLevel::PLogMin, bool enableDoNotReuseList=false);
 
 #endif //__PERSISTENT_HPP__

common/thorhelper/thorsoapcall.cpp

@@ -31,6 +31,7 @@
 #include "eclrtl.hpp"
 #include "roxiemem.hpp"
 #include "zcrypt.hpp"
+#include "persistent.hpp"
 
 using roxiemem::OwnedRoxieString;
 
@@ -43,6 +44,7 @@ using roxiemem::OwnedRoxieString;
 #define CONTENT_LENGTH "Content-Length: "
 #define CONTENT_ENCODING "Content-Encoding"
 #define ACCEPT_ENCODING "Accept-Encoding"
+#define CONNECTION "Connection"
 
 unsigned soapTraceLevel = 1;
 
@@ -511,6 +513,28 @@ public:
     }
 } *blacklist;
 
+static IPersistentHandler* persistentHandler = nullptr;
+static CriticalSection persistentCrit;
+static std::atomic<bool> persistentInitDone{false};
+
+void initPersistentHandler()
+{
+    CriticalBlock block(persistentCrit);
+    if (!persistentInitDone)
+    {
+#ifndef _CONTAINERIZED
+        const IProperties &conf = queryEnvironmentConf();
+        int maxPersistentRequests = conf.getPropInt("maxPersistentRequests", DEFAULT_MAX_PERSISTENT_REQUESTS);
+#else
+        const IPropertyTree& conf = queryComponentConfig();
+        int maxPersistentRequests = conf.getPropInt("@maxPersistentRequests", DEFAULT_MAX_PERSISTENT_REQUESTS);
+#endif
+        if (maxPersistentRequests != 0)
+            persistentHandler = createPersistentHandler(nullptr, DEFAULT_MAX_PERSISTENT_IDLE_TIME, maxPersistentRequests, PersistentLogLevel::PLogMin, true);
+        persistentInitDone = true;
+    }
+}
+
 MODULE_INIT(INIT_PRIORITY_STANDARD)
 {
     blacklist = new BlackLister;
@@ -522,6 +546,12 @@ MODULE_EXIT()
 {
     blacklist->stop();
     delete blacklist;
+
+    if (persistentHandler)
+    {
+        persistentHandler->stop(true);
+        ::Release(persistentHandler);
+    }
 }
 
 //=================================================================================================
@@ -648,7 +678,7 @@ interface IWSCAsyncFor: public IInterface
     virtual void checkTimeLimitExceeded(unsigned * _remainingMS) = 0;
 
     virtual void createHttpRequest(Url &url, StringBuffer &request) = 0;
-    virtual int readHttpResponse(StringBuffer &response, ISocket *socket) = 0;
+    virtual int readHttpResponse(StringBuffer &response, ISocket *socket, bool &keepAlive) = 0;
     virtual void processResponse(Url &url, StringBuffer &response, ColumnProvider * meta) = 0;
 
     virtual const char *getResponsePath() = 0;
@@ -1561,7 +1591,7 @@ private:
     Owned<CSocketDataProvider> dataProvider;
     PTreeReaderOptions options;
     unsigned remainingMS;
-    cycle_t startSoapCallCycles;
+    CCycleTimer mTimer;
 
     inline void checkRoxieAbortMonitor(IRoxieAbortMonitor * roxieAbortMonitor)
     {
@@ -1771,7 +1801,7 @@ private:
             master->logctx.CTXLOG("%s: request(%s:%u)", master->wscCallTypeText(), url.host.str(), url.port);
     }
 
-    int readHttpResponse(StringBuffer &response, ISocket *socket)
+    int readHttpResponse(StringBuffer &response, ISocket *socket, bool &keepAlive)
     {
         // Read the POST reply
         // not doesn't *assume* is valid HTTP post format but if it is takes advantage of
@@ -1780,6 +1810,7 @@ private:
         MemoryAttr buf;
         char *buffer=(char *)buf.allocate(WSCBUFFERSIZE+1);
         int rval = 200;
+        keepAlive = false;
 
         // first read header
         size32_t payloadofs = 0;
@@ -1925,6 +1956,8 @@ private:
                 }
             }
         }
+        if (rval == 200 && response.length() > 0)
+            keepAlive = checkKeepAlive(dbgheader);
         if (checkContentDecoding(dbgheader, response, contentEncoding))
             decodeContent(contentEncoding.str(), response);
         if (soapTraceLevel > 6 || master->logXML)
@@ -2071,6 +2104,20 @@ private:
             throw MakeStringException(TIMELIMIT_EXCEEDED, "%sCALL TIMELIMIT(%ums) exceeded", master->wscType == STsoap ? "SOAP" : "HTTP", master->timeLimitMS);
     }
 
+    inline bool checkKeepAlive(StringBuffer& headers)
+    {
+        size32_t verOffset = httpVerOffset();
+        if (headers.length() <= verOffset)
+            return false;
+        StringBuffer httpVer;
+        const char* ptr = headers.str() + verOffset;
+        while (*ptr && !isspace(*ptr))
+            httpVer.append(*ptr++);
+        StringBuffer conHeader;
+        getHTTPHeader(ptr, CONNECTION, conHeader);
+        return isHttpPersistable(httpVer.str(), conHeader.str());
+    }
+
 public:
     CWSCAsyncFor(CWSCHelper * _master, IXmlWriterExt &_xmlWriter, ConstPointerArray &_inputRows, PTreeReaderOptions _options): xmlWriter(_xmlWriter), inputRows(_inputRows), options(_options)
     {
@@ -2086,14 +2133,11 @@ public:
         }
         responsePath.append(master->service).append("Response");
         remainingMS = 0;
-        startSoapCallCycles = get_cycles_now();
     }
 
     ~CWSCAsyncFor()
     {
-        cycle_t endCycles = get_cycles_now();
-        __int64 elapsedNs = cycle_to_nanosec(endCycles-startSoapCallCycles);
-        master->logctx.noteStatistic(StTimeSoapcall, elapsedNs);
+        master->logctx.noteStatistic(StTimeSoapcall, mTimer.elapsedNs());
     }
 
     IMPLEMENT_IINTERFACE;
@@ -2115,41 +2159,61 @@ public:
         unsigned startidx = idx;
         while (!master->aborted)
         {
+            bool keepAlive = false;
+            bool isReused = false;
+            PersistentProtocol proto = PersistentProtocol::ProtoTCP;
+            SocketEndpoint ep;
             Owned<ISocket> socket;
-            cycle_t startCycles;
-            startCycles = get_cycles_now();
+            CCycleTimer timer;
             for (;;)
             {
                 try
                 {
                     checkTimeLimitExceeded(&remainingMS);
                     Url &connUrl = master->proxyUrlArray.empty() ? url : master->proxyUrlArray.item(0);
-                    socket.setown(blacklist->connect(connUrl.port, connUrl.host, master->logctx, (unsigned)master->maxRetries, master->timeoutMS, master->roxieAbortMonitor));
-                    if (stricmp(url.method, "https") == 0)
+                    ep.set(connUrl.host.get(), connUrl.port);
+                    if (strieq(url.method, "https"))
+                        proto = PersistentProtocol::ProtoTLS;
+                    bool shouldClose = false;
+                    Owned<ISocket> psock = persistentHandler?persistentHandler->getAvailable(&ep, &shouldClose, proto):nullptr;
+                    if (psock)
                     {
-#ifdef _USE_OPENSSL
-                        Owned<ISecureSocket> ssock = master->createSecureSocket(socket.getClear());
-                        if (ssock) 
+                        isReused = true;
+                        keepAlive = !shouldClose;
+                        socket.setown(psock.getClear());
+                    }
+                    else
+                    {
+                        isReused = false;
+                        keepAlive = true;
+                        socket.setown(blacklist->connect(connUrl.port, connUrl.host, master->logctx, (unsigned)master->maxRetries, master->timeoutMS, master->roxieAbortMonitor));
+                        if (proto == PersistentProtocol::ProtoTLS)
                         {
-                            checkTimeLimitExceeded(&remainingMS);
-                            int status = ssock->secure_connect();
-                            if (status < 0)
+#ifdef _USE_OPENSSL
+                            Owned<ISecureSocket> ssock = master->createSecureSocket(socket.getClear());
+                            if (ssock)
                             {
-                                StringBuffer err;
-                                err.append("Failure to establish secure connection to ");
-                                connUrl.getUrlString(err);
-                                err.append(": returned ").append(status);
-                                throw MakeStringExceptionDirect(0, err.str());
+                                checkTimeLimitExceeded(&remainingMS);
+                                int status = ssock->secure_connect();
+                                if (status < 0)
+                                {
+                                    StringBuffer err;
+                                    err.append("Failure to establish secure connection to ");
+                                    connUrl.getUrlString(err);
+                                    err.append(": returned ").append(status);
+                                    throw makeStringException(0, err.str());
+                                }
+                                socket.setown(ssock.getClear());
                             }
-                            socket.setown(ssock.getLink());
-                        }
 #else
-                        StringBuffer err;
-                        err.append("Failure to establish secure connection to ");
-                        connUrl.getUrlString(err);
-                        err.append(": OpenSSL disabled in build");
-                        throw MakeStringExceptionDirect(0, err.str());
+                            StringBuffer err;
+                            err.append("Failure to establish secure connection to ");
+                            connUrl.getUrlString(err);
+                            err.append(": OpenSSL disabled in build");
+                            throw makeStringExceptionDirect(0, err.str());
 #endif
+
+                        }
                     }
                     break;
                 }
@@ -2194,7 +2258,9 @@ public:
                 checkTimeLimitExceeded(&remainingMS);
                 checkRoxieAbortMonitor(master->roxieAbortMonitor);
 
-                int rval = readHttpResponse(response, socket);
+                bool keepAlive2;
+                int rval = readHttpResponse(response, socket, keepAlive2);
+                keepAlive = keepAlive && keepAlive2;
 
                 if (soapTraceLevel > 4)
                     multiLog(master->logctx, "%sCALL: received response (%s) from %s:%d", master->wscType == STsoap ? "SOAP" : "HTTP",master->service.str(), url.host.str(), url.port);
@@ -2213,16 +2279,24 @@ public:
                 {
                     throw MakeStringException(-1, "Zero length response in processQuery");
                 }
-                cycle_t endCycles = get_cycles_now();
-                __int64 elapsedNs = cycle_to_nanosec(endCycles-startCycles);
                 checkTimeLimitExceeded(&remainingMS);
-                ColumnProvider * meta = (ColumnProvider*)CreateColumnProvider((unsigned)nanoToMilli(elapsedNs), master->flags&SOAPFencoding?true:false);
+                ColumnProvider * meta = (ColumnProvider*)CreateColumnProvider((unsigned)nanoToMilli(timer.elapsedNs()), master->flags&SOAPFencoding?true:false);
                 processResponse(url, response, meta);
                 delete meta;
+
+                if (persistentHandler)
+                {
+                    if (isReused)
+                        persistentHandler->doneUsing(socket, keepAlive);
+                    else if (keepAlive)
+                        persistentHandler->add(socket, &ep, proto);
+                }
                 break;
             }
             catch (IReceivedRoxieException *e)
             {
+                if (persistentHandler && isReused)
+                    persistentHandler->doneUsing(socket, false);
                 // server busy ... Sleep and retry
                 if (e->errorCode() == 1001)
                 {
@@ -2253,6 +2327,8 @@ public:
             }
             catch (IException *e)
             {
+                if (persistentHandler && isReused)
+                    persistentHandler->doneUsing(socket, false);
                 if (master->timeLimitExceeded)
                 {
                     processException(url, inputRows, e);
@@ -2284,12 +2360,16 @@ public:
             }
             catch (std::exception & es)
             {
+                if (persistentHandler && isReused)
+                    persistentHandler->doneUsing(socket, false);
                 if(dynamic_cast<std::bad_alloc *>(&es))
                     throw MakeStringException(-1, "std::exception: out of memory (std::bad_alloc) in CWSCAsyncFor processQuery");
                 throw MakeStringException(-1, "std::exception: standard library exception (%s) in CWSCAsyncFor processQuery",es.what());
             }
             catch (...)
             {
+                if (persistentHandler && isReused)
+                    persistentHandler->doneUsing(socket, false);
                 throw MakeStringException(-1, "Unknown exception in processQuery");
             }
         }
@@ -2302,5 +2382,7 @@ public:
 
 IWSCAsyncFor * createWSCAsyncFor(CWSCHelper * _master, IXmlWriterExt &_xmlWriter, ConstPointerArray &_inputRows, PTreeReaderOptions _options)
 {
+    if (!persistentInitDone)
+        initPersistentHandler();
     return new CWSCAsyncFor(_master, _xmlWriter, _inputRows, _options);
 }

dali/dfuplus/CMakeLists.txt

@@ -52,6 +52,7 @@ include_directories (
          ./../../fs/dafsserver
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 set_source_files_properties (dfuplus.cpp PROPERTIES COMPILE_FLAGS -DDAFILESRV_LOCAL)

ecl/ecl-bundle/CMakeLists.txt

@@ -53,6 +53,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/system/xmllib
          ${HPCC_SOURCE_DIR}/ecl/eclcmd
          ${HPCC_SOURCE_DIR}/ecl/eclcc
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

ecl/ecl-package/CMakeLists.txt

@@ -52,6 +52,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/system/xmllib
          ${HPCC_SOURCE_DIR}/ecl/eclcmd
          ${HPCC_SOURCE_DIR}/ecl/eclcc
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

ecl/eclcmd/CMakeLists.txt

@@ -69,6 +69,7 @@ include_directories (
          ./../../esp/smc/SMCLib
          ./../../system/jlib
          ./../../system/xmllib
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

ecl/eclcmd/queries/CMakeLists.txt

@@ -43,6 +43,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/system/xmllib
          ${HPCC_SOURCE_DIR}/ecl/eclcmd
          ${HPCC_SOURCE_DIR}/ecl/eclcc
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

ecl/eclcmd/roxie/CMakeLists.txt

@@ -43,6 +43,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/system/xmllib
          ${HPCC_SOURCE_DIR}/ecl/eclcmd
          ${HPCC_SOURCE_DIR}/ecl/eclcc
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

ecl/eclplus/CMakeLists.txt

@@ -59,6 +59,7 @@ include_directories (
          ./../../esp/platform 
          ./../../system/jlib 
          ./../../system/xmllib 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -DECLPLUS_EXPORTS -D_CONSOLE )

esp/bindings/http/client/httpclient.cpp

@@ -189,7 +189,7 @@ CHttpClient::~CHttpClient()
         }
         else if (m_persistentHandler && !m_disableKeepAlive && m_persistable)
         {
-            m_persistentHandler->add(m_socket, &m_ep);
+            m_persistentHandler->add(m_socket, &m_ep, strieq(m_protocol.get(), "HTTPS")?PersistentProtocol::ProtoTLS:PersistentProtocol::ProtoTCP);
         }
         else
         {
@@ -288,7 +288,8 @@ int CHttpClient::connect(StringBuffer& errmsg, bool forceNewConnection)
         m_disableKeepAlive = true;
 
     bool shouldClose = false;
-    Linked<ISocket> pSock = (m_disableKeepAlive || forceNewConnection)?nullptr:m_persistentHandler->getAvailable(&ep, &shouldClose);
+    Owned<ISocket> pSock = (m_disableKeepAlive || forceNewConnection)?nullptr:m_persistentHandler->getAvailable(&ep, &shouldClose,
+            strieq(m_protocol.get(), "HTTPS")?PersistentProtocol::ProtoTLS:PersistentProtocol::ProtoTCP);
     if(pSock)
     {
         m_isPersistentSocket = true;

esp/bindings/http/platform/httptransport.cpp

@@ -16,8 +16,8 @@
 ############################################################################## */
 #pragma warning(disable : 4786)
 #include "platform.h"
-
 #include "esphttp.hpp"
+#include "persistent.hpp"
 
 #ifdef _WIN32
   #include <algorithm>
@@ -1190,6 +1190,16 @@ void CHttpMessage::enableCompression()
     m_compressionEnabled = true;
 }
 
+bool CHttpMessage::checkPersistentEligible()
+{
+    size32_t verOffset = httpVerOffset();
+    if (m_version.length() <= verOffset)
+        return false;
+    StringBuffer conHeader;
+    getHeader("Connection", conHeader);
+    return isHttpPersistable(m_version.str() + verOffset, conHeader.str());
+}
+
 /******************************************************************************
               CHttpRequest Implementation
 *******************************************************************************/
@@ -1945,25 +1955,7 @@ bool CHttpRequest::checkPersistentEligible()
     if(m_content_length == -1 && !(m_httpMethod.length() > 0 && stricmp(m_httpMethod.get(), "GET") == 0))
         return false;
 
-    StringBuffer conheader;
-    getHeader("Connection", conheader);
-    conheader.trim().toLowerCase();
-    if(conheader.length() != 0)
-    {
-        if(strcmp(conheader.str(), "keep-alive") == 0)
-            return true;
-        else if(strcmp(conheader.str(), "close") == 0)
-            return false;
-    }
-
-    //HTTP 1.0 close by default
-    const char* httpver = nullptr;
-    if(m_version.length() > 5)
-        httpver = m_version.str() + 5;
-    if(httpver && strcmp(httpver, "1.0") == 0)
-        return false;
-
-    return true;
+    return CHttpMessage::checkPersistentEligible();
 }
 
 int CHttpRequest::processHeaders(IMultiException *me)
@@ -2432,13 +2424,7 @@ int CHttpResponse::processHeaders(IMultiException *me)
         lenread = m_bufferedsocket->readline(oneline, MAX_HTTP_HEADER_LEN, me);
     }
 
-    setPersistentEligible(true);
-    StringBuffer conheader;
-    getHeader("Connection", conheader);
-    if(conheader.length() != 0 && stricmp(conheader.str(), "Close") == 0)
-        setPersistentEligible(false);
-    if(m_content_length == 0)
-        setPersistentEligible(false);
+    setPersistentEligible(checkPersistentEligible());
 
     return 0;
 }
@@ -2860,3 +2846,11 @@ bool CHttpResponse::decompressContent(StringBuffer* originalContent, int compres
     return true;
 #endif
 }
+
+bool CHttpResponse::checkPersistentEligible()
+{
+    if(m_content_length <= 0)
+        return false;
+
+    return CHttpMessage::checkPersistentEligible();
+}

esp/bindings/http/platform/httptransport.ipp

@@ -99,6 +99,9 @@ protected:
     virtual StringBuffer& constructHeaderBuffer(StringBuffer& headerbuf, bool inclLength);
     virtual int processHeaders(IMultiException *me);
 
+protected:
+    virtual bool checkPersistentEligible();
+
 public:
     IMPLEMENT_IINTERFACE;
     
@@ -327,11 +330,11 @@ private:
     virtual StringBuffer& constructHeaderBuffer(StringBuffer& headerbuf, bool inclLen);
     virtual int processHeaders(IMultiException *me);
     virtual void parseCookieHeader(char* cookiestr);
-    inline bool checkPersistentEligible();
 
-public:
-    
+protected:
+    virtual bool checkPersistentEligible() override;
 
+public:
     CHttpRequest(ISocket& socket);
     virtual ~CHttpRequest();
     
@@ -379,6 +382,9 @@ private:
     virtual void parseCookieHeader(char* cookiestr);
     virtual void parseOneCookie(char* cookiestr);
 
+protected:
+    virtual bool checkPersistentEligible() override;
+
 public:
     CHttpResponse(ISocket& socket);
     virtual ~CHttpResponse();

esp/clients/WUManager/CMakeLists.txt

@@ -44,6 +44,7 @@ include_directories (
          ./../../../system/jlib 
          ./../../platform 
          ./../../clients 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

esp/clients/wsecl/CMakeLists.txt

@@ -44,6 +44,7 @@ include_directories (
          ./../../platform 
          ./../../bindings/SOAP/Platform 
          ./../../bindings/SOAP/ws_ecl_client 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL -DWS_ECL_CLIENT_EXPORTS )

esp/logging/loggingagent/cassandraloggingagent/CMakeLists.txt

@@ -38,6 +38,7 @@ if(USE_CASSANDRA)
      ${HPCC_SOURCE_DIR}/esp/bindings
      ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp
      ${HPCC_SOURCE_DIR}/esp/logging/logginglib
+     ${HPCC_SOURCE_DIR}/common/thorhelper
   )
 
   ADD_DEFINITIONS( -D_USRDLL -DCASSANDRALOGAGENT_EXPORTS )

esp/logging/loggingagent/espserverloggingagent/CMakeLists.txt

@@ -45,6 +45,7 @@ include_directories (
     ${HPCC_SOURCE_DIR}/esp/esdl
     ${HPCC_SOURCE_DIR}/esp/esplib
     ${HPCC_SOURCE_DIR}/esp/logging/logginglib
+    ${HPCC_SOURCE_DIR}/common/thorhelper
     ./../..
     ./..
 )

esp/logging/logginglib/CMakeLists.txt

@@ -29,6 +29,7 @@ include_directories (
     ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp        #XmlPullParser.h in generated/ws_loggingservice_esp.ipp
     ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/Platform   #CSoapRequestBinding
     ${HPCC_SOURCE_DIR}/esp/clients                  #edwin.h in generated/ws_loggingservice.esp
+    ${HPCC_SOURCE_DIR}/common/thorhelper
 )
 
 ADD_DEFINITIONS ( -D_USRDLL -DLOGGINGCOMMON_EXPORTS )
@@ -53,4 +54,4 @@ target_link_libraries ( logginglib
     jlib
     xmllib
     esphttp
-)
+)

esp/logging/loggingmanager/CMakeLists.txt

@@ -48,6 +48,7 @@ include_directories (
     ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp
 #    ./../../protocols/sidex
     ${HPCC_SOURCE_DIR}/esp/logging/logginglib
+    ${HPCC_SOURCE_DIR}/common/thorhelper
     ./..
 )
 

esp/logging/test/CMakeLists.txt

@@ -50,6 +50,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/esp/clients/LoggingClient
          ${HPCC_SOURCE_DIR}/esp/logging/logginglib
          ${HPCC_SOURCE_DIR}/esp/logging/loggingmanager
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

esp/protocols/http/CMakeLists.txt

@@ -47,7 +47,6 @@ set(SRCS
     ../../platform/espcache.cpp
     ../../platform/sechandler.cpp
     ../../platform/txsummary.cpp
-    ../../platform/persistent.cpp
     mapinfo.cpp
     plugin.cpp
     )
@@ -68,6 +67,7 @@ include_directories(
     ./../../../system/mp 
     ./../../../dali/base
     ./../../../common/workunit
+    ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 add_definitions(-DESPHTTP_EXPORTS -DESP_TIMING -D_USRDLL -DESP_PLUGIN)
@@ -82,6 +82,7 @@ target_link_libraries(
     jlib
     xmllib
     dalibase
+    thorhelper
     )
 
 if(LIBMEMCACHED_FOUND)

esp/services/WsDeploy/CMakeLists.txt

@@ -59,6 +59,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/deployment/configutils
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/espcontrol/CMakeLists.txt

@@ -48,6 +48,7 @@ include_directories (
          ./../../bindings/SOAP/xpp
          ./../../smc/SMCLib
          ./../../../common/environment
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_access/CMakeLists.txt

@@ -48,6 +48,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/esp/smc/SMCLib
          ${HPCC_SOURCE_DIR}/dali/base
          ${HPCC_SOURCE_DIR}/system/mp
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 # NOTE - this should not be needed, it's the result of poor encapsulation and using CLdapSecManager directly 

esp/services/ws_account/CMakeLists.txt

@@ -46,6 +46,7 @@ include_directories (
          ./../../bindings 
          ./../../bindings/SOAP/xpp
          ./../../smc/SMCLib  
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 if (USE_OPENLDAP)

esp/services/ws_codesign/CMakeLists.txt

@@ -57,6 +57,7 @@ include_directories (
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
          ${CMAKE_BINARY_DIR}/esp/services/ws_codesign
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_config/CMakeLists.txt

@@ -48,6 +48,7 @@ include_directories (
          ./../../../dali/base 
          ./../../bindings 
          ./../../bindings/SOAP/xpp 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_configmgr/CMakeLists.txt

@@ -51,6 +51,7 @@ include_directories(
     ${CMAKE_BINARY_DIR}/oss
     ${CMAKE_BINARY_DIR}/esp/services/ws_configmgr
     ${BOOST_REGEX_INCLUDE_DIR}
+    ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 add_definitions(-D_USRDLL)

esp/services/ws_decoupledlogging/CMakeLists.txt

@@ -48,6 +48,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp
          ${HPCC_SOURCE_DIR}/esp/logging/logginglib
          ${HPCC_SOURCE_DIR}/esp/smc/SMCLib
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL -DLOGGINGCOMMON_EXPORTS)

esp/services/ws_ecl/CMakeLists.txt

@@ -51,6 +51,7 @@ include_directories (
          ./../../../common/wuwebview
          ./../../../common/fileview2 
          ./../../../dali/base 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_elk/CMakeLists.txt

@@ -57,6 +57,7 @@ include_directories (
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
          ${CMAKE_BINARY_DIR}/esp/services/ws_elk
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_esdlconfig/CMakeLists.txt

@@ -56,6 +56,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/esp/esdllib
          ${HPCC_SOURCE_DIR}/esp/logging/logginglib
          ${HPCC_SOURCE_DIR}/esp/logging/loggingmanager
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_fileio/CMakeLists.txt

@@ -57,6 +57,7 @@ include_directories (
          ./../../bindings 
          ./../../smc/SMCLib 
          ./../../bindings/SOAP/xpp 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_fs/CMakeLists.txt

@@ -59,6 +59,7 @@ include_directories (
          ./../../bindings/SOAP/xpp 
          ./../../../fs/dafsclient 
          ./../../../common/workunit
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_loggingservice/CMakeLists.txt

@@ -38,6 +38,7 @@ include_directories (
      ${HPCC_SOURCE_DIR}/esp/clients
      ${HPCC_SOURCE_DIR}/esp/clients/LoggingClient
      ${HPCC_SOURCE_DIR}/esp/logging/logginglib
+     ${HPCC_SOURCE_DIR}/common/thorhelper
      ./..
 )
 

esp/services/ws_machine/CMakeLists.txt

@@ -56,6 +56,7 @@ include_directories (
          ./../../bindings/SOAP/xpp 
          ./../../../esp/platform 
          ./../../smc/SMCLib 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL )

esp/services/ws_smc/CMakeLists.txt

@@ -61,6 +61,7 @@ include_directories (
          ./../../../system/security/LdapSecurity
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 if (USE_OPENLDAP)

esp/services/ws_sql/CMakeLists.txt

@@ -94,6 +94,7 @@ if(WSSQL_SERVICE)
         ${libantlr3c_includes}
         ${ESPSCM_GENERATED_DIR}
         ${CMAKE_CURRENT_SOURCE_DIR}/SQL2ECL
+        ${HPCC_SOURCE_DIR}/common/thorhelper
         )
 
     if (CMAKE_COMPILER_IS_CLANG)

esp/services/ws_store/CMakeLists.txt

@@ -60,6 +60,7 @@ HPCC_ADD_SUBDIRECTORY (espstorelib)
         ${HPCC_SOURCE_DIR}/common/environment
         ${HPCC_SOURCE_DIR}/common/deftype
         ${ESPSCM_GENERATED_DIR}
+        ${HPCC_SOURCE_DIR}/common/thorhelper
         )
 
     add_definitions(-D_USRDLL)

esp/services/ws_topology/CMakeLists.txt

@@ -56,6 +56,7 @@ include_directories (
          ./../../smc/SMCLib 
          ./../../bindings/SOAP/xpp 
          ./../..//esplib
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL -DWS_TOPOLOGY_EXPORTS )

esp/smc/SMCLib/CMakeLists.txt

@@ -57,6 +57,7 @@ include_directories (
          ./../../bindings 
          ./../../bindings/SOAP/xpp 
          ./.. 
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_USRDLL -DSMCLIB_EXPORTS )

esp/tools/tankfile/CMakeLists.txt

@@ -46,6 +46,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/esp/bindings
          ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp
          ${HPCC_SOURCE_DIR}/esp/clients
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )

tools/esdlcmd/CMakeLists.txt

@@ -61,6 +61,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/tools/esdlcomp
          ${HPCC_SOURCE_DIR}/tools/esdlcmd-xml
          ${HPCC_SOURCE_DIR}/esp/esdllib
+         ${HPCC_SOURCE_DIR}/common/thorhelper
     )
 
 ADD_DEFINITIONS( -D_CONSOLE )