Merge remote-tracking branch 'origin/candidate-3.10.x'

Conflicts:
	esp/files/scripts/ResultsControl.js

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

cmake_modules/dependencies/el5.cmake

@@ -1 +1 @@
-set ( CPACK_RPM_PACKAGE_REQUIRES "boost141-regex, openldap, libicu, m4, libtool, xalan-c, xerces-c, gcc-c++, openssh-server, openssh-clients, expect, libarchive")
+set ( CPACK_RPM_PACKAGE_REQUIRES "boost141-regex, openldap, libicu, m4, libtool, xalan-c, xerces-c, gcc-c++, openssh-server, openssh-clients, expect, libarchive, rsync")

cmake_modules/dependencies/el6.cmake

@@ -1 +1 @@
-set ( CPACK_RPM_PACKAGE_REQUIRES "boost-regex, openldap, libicu, m4, libtool, libxslt, libxml2, gcc-c++, openssh-server, openssh-clients, expect, libarchive")
+set ( CPACK_RPM_PACKAGE_REQUIRES "boost-regex, openldap, libicu, m4, libtool, libxslt, libxml2, gcc-c++, openssh-server, openssh-clients, expect, libarchive, rsync")

cmake_modules/dependencies/lenny.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.34.1, libicu38, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, sudo, openssh-client, openssh-server, expect, libarchive")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.34.1, libicu38, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, sudo, openssh-client, openssh-server, expect, libarchive, rsync")

cmake_modules/dependencies/lucid.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.40.0, libicu42, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.40.0, libicu42, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1, rsync")

cmake_modules/dependencies/natty.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.42.0, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.42.0, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1, rsync")

cmake_modules/dependencies/oneiric.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.46.1, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.46.1, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1, rsync")

cmake_modules/dependencies/precise.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.46.1, libicu48, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive12")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.46.1, libicu48, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive12, rsync")

cmake_modules/dependencies/squeeze.cmake

@@ -1 +1 @@
-set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.42.0, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1")
+set ( CPACK_DEBIAN_PACKAGE_DEPENDS "libboost-regex1.42.0, libicu44, libxalan110, libxerces-c28, binutils, libldap-2.4-2, openssl, zlib1g, g++, openssh-client, openssh-server, expect, libarchive1, rsync")

cmake_modules/dependencies/suse11.3.cmake

@@ -1 +1 @@
-set ( CPACK_RPM_PACKAGE_REQUIRES  "binutils, gcc-c++, openssh, libldap-2_4-2, libicu, libboost_regex1_42_0, libxerces-c-3_0, libxalan-c110, expect, libarchive2" )
+set ( CPACK_RPM_PACKAGE_REQUIRES  "binutils, gcc-c++, openssh, libldap-2_4-2, libicu, libboost_regex1_42_0, libxerces-c-3_0, libxalan-c110, expect, libarchive2, rsync" )

cmake_modules/dependencies/suse11.4.cmake

@@ -1 +1 @@
-set ( CPACK_RPM_PACKAGE_REQUIRES  "binutils, gcc-c++, openssh, libldap-2_4-2, libicu, libboost_regex1_44_0, libxerces-c-3_0, libxalan-c110, expect, libarchive2" )
+set ( CPACK_RPM_PACKAGE_REQUIRES  "binutils, gcc-c++, openssh, libldap-2_4-2, libicu, libboost_regex1_44_0, libxerces-c-3_0, libxalan-c110, expect, libarchive2, rsync" )

dali/base/dasds.cpp

@@ -34,8 +34,10 @@
 #define DEBUG_DIR "debug"
 #define DEFAULT_KEEP_LASTN_STORES 1
 #define MAXDELAYS 5
-static const char *deltaHeader = "<CRC>0000000000</CRC>"; // fill in later
+static const char *deltaHeader = "<CRC>0000000000</CRC><SIZE>0000000000000000</SIZE>"; // fill in later
 static unsigned deltaHeaderCrcOff = 5;
+static unsigned deltaHeaderSizeStart = 21;
+static unsigned deltaHeaderSizeOff = 27;
 
 static unsigned readWriteSlowTracing = 10000; // 10s default
 static bool readWriteStackTracing = false;
@@ -870,9 +872,11 @@ void writeDelta(StringBuffer &xml, IFile &iFile, const char *msg="", unsigned re
     Owned<IFileIOStream> stream;
     offset_t lastGood = 0;
     unsigned startCrc = ~0;
-    char strNum[11];
+    MemoryBuffer header;
+    char strNum[17];
     loop
     {
+        header.append(deltaHeader);
         try
         {
             iFileIO.setown(iFile.open(IFOreadwrite));
@@ -897,9 +901,14 @@ void writeDelta(StringBuffer &xml, IFile &iFile, const char *msg="", unsigned re
             stream->write(xml.length(), xml.toCharArray());
             stream->flush();
             stream.clear();
+            offset_t fLen = lastGood + xml.length();
             unsigned crc = crc32(xml.toCharArray(), xml.length(), startCrc);
+            char *headerPtr = (char *)header.bufferBase();
             sprintf(strNum, "%010u", ~crc);
-            iFileIO->write(deltaHeaderCrcOff, 10, strNum);
+            memcpy(headerPtr + deltaHeaderCrcOff, strNum, 10);
+            sprintf(strNum, "%016"I64F"X", fLen);
+            memcpy(headerPtr + deltaHeaderSizeOff, strNum, 16);
+            iFileIO->write(0, strlen(deltaHeader), headerPtr);
         }
         catch (IException *e)
         {
@@ -5137,23 +5146,48 @@ public:
         constructStoreName(DELTADETACHED, deltaInfo.edition, detachName);
         return detachName;
     }
-    virtual bool loadDelta(const char *filename, IFileIO *iFileIO, IPropertyTree *root)
+    virtual bool loadDelta(const char *filename, IFile *iFile, IPropertyTree *root)
     {
-        OwnedIFileIOStream iFileIOStream = createIOStream(iFileIO);
+        Owned<IFileIO> iFileIO = iFile->open(IFOread);
+        if (!iFileIO) // no delta to load
+            return true;
         MemoryBuffer tmp;
         char *ptr = (char *) tmp.reserveTruncate(strlen(deltaHeader));
         unsigned embeddedCrc = 0;
+        offset_t pos = 0;
         bool hasCrcHeader = false; // check really only needed for deltas proceeding CRC header
-        if (strlen(deltaHeader) == iFileIOStream->read(strlen(deltaHeader), ptr))
+        if (strlen(deltaHeader) == iFileIO->read(0, strlen(deltaHeader), ptr))
         {
             if (0 == memicmp(deltaHeader, ptr, 5))
             {
+                pos = deltaHeaderSizeStart;
                 hasCrcHeader = true;
                 embeddedCrc = (unsigned)atoi64_l(ptr+deltaHeaderCrcOff, 10);
+                if (0 == memicmp(deltaHeader+deltaHeaderSizeStart, ptr+deltaHeaderSizeStart, 6)) // has <SIZE> too
+                {
+                    pos = strlen(deltaHeader);
+                    offset_t lastGood;
+                    if (sscanf(ptr+deltaHeaderSizeOff, "%"I64F"X", &lastGood))
+                    {
+                        offset_t fSize = iFileIO->size();
+                        if (fSize > lastGood)
+                        {
+                            size32_t diff = fSize - lastGood;
+                            LOG(MCoperatorError, unknownJob, "Delta file '%s', has %d bytes of trailing data (possible power loss during save?), file size: %"I64F"d, last committed size: %"I64F"d", filename, diff, fSize, lastGood);
+                            LOG(MCoperatorError, unknownJob, "Resetting delta file '%s' to size: %"I64F"d", filename, lastGood);
+                            iFileIO->close();
+                            backup(filename);
+                            iFileIO.setown(iFile->open(IFOreadwrite));
+                            iFileIO->setSize(lastGood);
+                            iFileIO->close();
+                            iFileIO.setown(iFile->open(IFOread));
+                        }
+                    }
+                }
             }
         }
-        if (!hasCrcHeader)
-            iFileIOStream->seek(0, IFSbegin);
+        OwnedIFileIOStream iFileIOStream = createIOStream(iFileIO);
+        iFileIOStream->seek(pos, IFSbegin);
         Owned<ICrcIOStream> crcPipeStream = createCrcPipeStream(iFileIOStream); // crc *rest* of stream
         Owned<IIOStream> ios = createBufferedIOStream(crcPipeStream);
         bool noErrors;
@@ -5184,32 +5218,28 @@ public:
         OwnedIFile deltaIFile = createIFile(deltaFilename.str());
         loop
         {
-            OwnedIFileIO iFileIO;
             StringAttr filename;
             IFile *iFile;
             if (detached)
             {
                 iFile = detachedDeltaIFile;
                 filename.set(iFile->queryFilename());
-                iFileIO.setown(iFile->open(IFOread));
             }
             else
             {
                 iFile = deltaIFile;
                 filename.set(iFile->queryFilename());
-                iFileIO.setown(iFile->open(IFOread));
-                if (!iFileIO) // no delta to load
+                if (!iFile->exists())
                     break;
             }
             PROGLOG("Loading delta: %s", filename.get());
 
             bool noError;
             Owned<IException> deltaE;
-            try { noError = loadDelta(filename, iFileIO, root); }
+            try { noError = loadDelta(filename, iFile, root); }
             catch (IException *e) { deltaE.setown(e); noError = false; }
             if (!noError)
             {
-                iFileIO.clear();
                 backup(filename);
                 if (errors) *errors = true;
                 if (deltaE)

dali/base/dasds.hpp

@@ -233,7 +233,7 @@ extern da_decl unsigned querySDSLockTimeoutCount();
 interface IStoreHelper : extends IInterface
 {
     virtual StringBuffer &getDetachedDeltaName(StringBuffer &detachName) = 0;
-    virtual bool loadDelta(const char *filename, IFileIO *iFile, IPropertyTree *root) = 0;
+    virtual bool loadDelta(const char *filename, IFile *iFile, IPropertyTree *root) = 0;
     virtual bool loadDeltas(IPropertyTree *root, bool *errors=NULL) = 0;
     virtual bool detachCurrentDelta() = 0;
     virtual void saveStore(IPropertyTree *root, unsigned *newEdition=NULL, bool currentEdition=false) = 0;

dali/daliadmin/daliadmin.cpp

@@ -50,6 +50,8 @@
 #define _putch putchar
 #endif
 
+#define DEFAULT_DALICONNECT_TIMEOUT 5 // seconds
+static unsigned daliConnectTimeoutMs = 5000;
 
 static bool noninteractive=false;
 
@@ -96,6 +98,7 @@ void usage(const char *exe)
   printf("\n");
   printf("Workunit commands:\n");
   printf("  listworkunits <workunit-mask> [<prop>=<val> <lower> <upper>]\n");
+  printf("  listmatches <connection xpath> [<match xpath>=<val> [<property xpath>]]\n");
   printf("  workunittimings <WUID>\n");
   printf("\n");
   printf("Other dali server and misc commands:\n");
@@ -109,13 +112,14 @@ void usage(const char *exe)
   printf("  dalilocks [ <ip-pattern> ] [ files ] -- get all locked files/xpaths\n");
   printf("  unlock <xpath or logicalfile>   --  unlocks either matching xpath(s) or matching logical file(s), can contain wildcards\n");
   printf("\n");
-  printf("Common options (can be placed in dfuutil.ini)\n");
+  printf("Common options\n");
   printf("  server=<dali-server-ip>         -- server ip\n");
   printf("                                  -- can be 1st param if numeric ip (or '.')\n");
   printf("  user=<username>                 -- for file operations\n");
   printf("  password=<password>             -- for file operations\n");
   printf("  logfile=<filename>              -- filename blank for no log\n");
   printf("  rawlog=0|1                      -- if raw omits timestamps etc\n");
+  printf("  timeout=<seconds>               -- set dali connect timeout\n");
 }
 
 #define SDS_LOCK_TIMEOUT  60000
@@ -194,11 +198,11 @@ static IRemoteConnection *connectXPathOrFile(const char *path,bool safe,StringBu
     }
     else if (strchr(path+((*path=='/')?1:0),'/')==NULL)
         safe = true;    // all root trees safe
-    Owned<IRemoteConnection> conn = querySDS().connect(remLeading(path),myProcessSession(),safe?0:RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(remLeading(path),myProcessSession(),safe?0:RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn&&lfnpath.length()) {
         lfn.makeFullnameQuery(lfnpath.clear(),DXB_SuperFile);
         path = lfnpath.str();
-        conn.setown(querySDS().connect(remLeading(path),myProcessSession(),safe?0:RTM_LOCK_READ, INFINITE));
+        conn.setown(querySDS().connect(remLeading(path),myProcessSession(),safe?0:RTM_LOCK_READ, daliConnectTimeoutMs));
     }
     if (conn.get())
         xpath.append(path);
@@ -246,7 +250,7 @@ static void import(const char *path,const char *src,bool add)
     if (!tail)
         return;
     if (!add) {
-        Owned<IRemoteConnection> bconn = querySDS().connect(remLeading(path),myProcessSession(),RTM_LOCK_READ|RTM_SUB, INFINITE);
+        Owned<IRemoteConnection> bconn = querySDS().connect(remLeading(path),myProcessSession(),RTM_LOCK_READ|RTM_SUB, daliConnectTimeoutMs);
         if (bconn) {
             Owned<IPropertyTree> broot = bconn->getRoot();
             StringBuffer bakname;
@@ -256,7 +260,7 @@ static void import(const char *path,const char *src,bool add)
             toXML(broot, *fstream);         // formatted (default)
         }
     }
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),0, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),0, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -294,7 +298,7 @@ static void _delete_(const char *path,bool backup)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -328,7 +332,7 @@ static void set(const char *path,const char *val)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -353,7 +357,7 @@ static void get(const char *path)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -375,7 +379,7 @@ static void bget(const char *path,const char *outfn)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -395,7 +399,7 @@ static void xget(const char *path)
 {
     if (!path||!*path)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect("/",myProcessSession(),RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect("/",myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to /");
         return;
@@ -444,7 +448,7 @@ static void wget(const char *path)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -468,7 +472,7 @@ static void add(const char *path,const char *val)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -491,7 +495,7 @@ static void delv(const char *path)
     const char *tail=splitpath(path,head,tmp);
     if (!tail)
         return;
-    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(head.str(),myProcessSession(),RTM_LOCK_WRITE, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",path);
         return;
@@ -520,7 +524,7 @@ static void dfsfile(const char *lname,IUserDescriptor *userDesc, UnsignedArray *
     CDfsLogicalFileName lfn;
     lfn.set(lname);
     if (!lfn.isExternal()) {
-        Owned<IPropertyTree> tree = queryDistributedFileDirectory().getFileTree(lname,userDesc,NULL,1000*60*5,true); //,userDesc);
+        Owned<IPropertyTree> tree = queryDistributedFileDirectory().getFileTree(lname,userDesc,NULL,daliConnectTimeoutMs,true); //,userDesc);
         if (partslist)
             filterParts(tree,*partslist);
         if (!tree) {
@@ -1000,7 +1004,7 @@ static void checksuperfile(const char *lfn,bool fix=false)
     lname.set(lfn);
     StringBuffer query;
     lname.makeFullnameQuery(query, DXB_SuperFile, true);
-    Owned<IRemoteConnection> conn = querySDS().connect(query.str(),myProcessSession(),fix?RTM_LOCK_WRITE:0, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(query.str(),myProcessSession(),fix?RTM_LOCK_WRITE:0, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to %s",lfn);
         ERRLOG("Superfile %s FAILED",lname.get());
@@ -1044,10 +1048,10 @@ static void checksuperfile(const char *lfn,bool fix=false)
         if (!sublname.isExternal()&&!sublname.isForeign()) {
             StringBuffer subquery;
             sublname.makeFullnameQuery(subquery, DXB_File, true);
-            Owned<IRemoteConnection> subconn = querySDS().connect(subquery.str(),myProcessSession(),fix?RTM_LOCK_WRITE:0, INFINITE);
+            Owned<IRemoteConnection> subconn = querySDS().connect(subquery.str(),myProcessSession(),fix?RTM_LOCK_WRITE:0, daliConnectTimeoutMs);
             if (!subconn) {
                 sublname.makeFullnameQuery(subquery.clear(), DXB_SuperFile, true);
-                subconn.setown(querySDS().connect(subquery.str(),myProcessSession(),0, INFINITE));
+                subconn.setown(querySDS().connect(subquery.str(),myProcessSession(),0, daliConnectTimeoutMs));
             }
             if (!subconn) {
                 ERRLOG("SuperFile %s is missing sub-file file %s",lname.get(),subname.str());
@@ -1081,7 +1085,7 @@ static void checksuperfile(const char *lfn,bool fix=false)
                     sdlname.set(pname.str());
                     StringBuffer sdquery;
                     sdlname.makeFullnameQuery(sdquery, DXB_SuperFile, true);
-                    Owned<IRemoteConnection> sdconn = querySDS().connect(sdquery.str(),myProcessSession(),0, INFINITE);
+                    Owned<IRemoteConnection> sdconn = querySDS().connect(sdquery.str(),myProcessSession(),0, daliConnectTimeoutMs);
                     if (!conn) {
                         WARNLOG("SubFile %s has missing owner superfile %s",sublname.get(),sdlname.get());
                     }
@@ -1211,10 +1215,10 @@ static void checksubfile(const char *lfn)
     lname.set(lfn);
     StringBuffer query;
     lname.makeFullnameQuery(query, DXB_File, true);
-    Owned<IRemoteConnection> conn = querySDS().connect(query.str(),myProcessSession(),0, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect(query.str(),myProcessSession(),0, daliConnectTimeoutMs);
     if (!conn) {
         lname.makeFullnameQuery(query.clear(), DXB_SuperFile, true);
-        conn.setown(querySDS().connect(query.str(),myProcessSession(),0, INFINITE));
+        conn.setown(querySDS().connect(query.str(),myProcessSession(),0, daliConnectTimeoutMs));
     }
     if (!conn) {
         ERRLOG("Could not connect to %s",lfn);
@@ -1231,7 +1235,7 @@ static void checksubfile(const char *lfn)
         sdlname.set(pname.str());
         StringBuffer sdquery;
         sdlname.makeFullnameQuery(sdquery, DXB_SuperFile, true);
-        Owned<IRemoteConnection> sdconn = querySDS().connect(sdquery.str(),myProcessSession(),0, INFINITE);
+        Owned<IRemoteConnection> sdconn = querySDS().connect(sdquery.str(),myProcessSession(),0, daliConnectTimeoutMs);
         if (!conn) {
             ERRLOG("SubFile %s has missing owner superfile %s",lname.get(),sdlname.get());
             ok = false;
@@ -1449,7 +1453,7 @@ static void dfsscopes(const char *name, IUserDescriptor *user)
         StringBuffer s;
         dlfn.makeScopeQuery(s,true);
         ln.clear().append("SCOPE '").append(iter->query()).append('\'');
-        Owned<IRemoteConnection> conn = querySDS().connect(s.str(),myProcessSession(),RTM_LOCK_READ, INFINITE);
+        Owned<IRemoteConnection> conn = querySDS().connect(s.str(),myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
         if (!conn)
             ERRLOG("%s - Could not connect using %s",ln.str(),s.str());
         else {
@@ -1498,7 +1502,7 @@ static void cleanscopes(IUserDescriptor *user)
         scope.append(iter->query());
         dlfn.set(scope.str(),"x");
         dlfn.makeScopeQuery(s.clear(),true);
-        Owned<IRemoteConnection> conn = querySDS().connect(s.str(),myProcessSession(),RTM_LOCK_READ, INFINITE);
+        Owned<IRemoteConnection> conn = querySDS().connect(s.str(),myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
         if (!conn)  
             DBGLOG("Could not connect to '%s' using %s",iter->query(),s.str());
         else {
@@ -1525,7 +1529,7 @@ static void cleanscopes(IUserDescriptor *user)
 
 static void listworkunits(const char *test,const char *min, const char *max)
 {
-    Owned<IRemoteConnection> conn = querySDS().connect("/", myProcessSession(), 0, 5*60*1000);
+    Owned<IRemoteConnection> conn = querySDS().connect("/", myProcessSession(), 0, daliConnectTimeoutMs);
     Owned<IPropertyTreeIterator> iter = conn->queryRoot()->getElements("WorkUnits/*");
     ForEach(*iter) {
         IPropertyTree &e=iter->query();
@@ -1554,6 +1558,37 @@ static void listworkunits(const char *test,const char *min, const char *max)
 
 //=============================================================================
 
+static void listmatches(const char *path, const char *match, const char *pval)
+{
+    Owned<IRemoteConnection> conn = querySDS().connect(path, myProcessSession(), 0, daliConnectTimeoutMs);
+    if (!conn)
+    {
+        PROGLOG("Failed to connect to %s", path);
+        return;
+    }
+    StringBuffer output("Listing matches for path=");
+    output.append(path);
+    if (match)
+    {
+        output.append(", match=").append(match);
+        if (pval)
+            output.append(", property value = ").append(pval);
+    }
+    Owned<IPropertyTreeIterator> iter = conn->queryRoot()->getElements(match?match:"*", iptiter_remote);
+    ForEach(*iter)
+    {
+        IPropertyTree &e=iter->query();
+        output.clear().append(e.queryName());
+        const char *val = e.queryProp(pval?pval:NULL);
+        if (val)
+            output.append(" = ").append(val);
+        outln(output.str());
+    }
+}
+
+//=============================================================================
+
+
 static const char *getNum(const char *s,unsigned &num)
 {
     while (*s&&!isdigit(*s))
@@ -1570,7 +1605,7 @@ static void workunittimings(const char *wuid)
 {
     StringBuffer path;
     path.append("/WorkUnits/").append(wuid);
-    Owned<IRemoteConnection> conn = querySDS().connect(path, myProcessSession(), 0, 5*60*1000);
+    Owned<IRemoteConnection> conn = querySDS().connect(path, myProcessSession(), 0, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("WU %s not found",wuid);
         return;
@@ -1688,7 +1723,7 @@ static unsigned clustersToGroups(IPropertyTree *envroot,const StringArray &cmpls
 
 static void clusterlist()
 {
-    Owned<IRemoteConnection> conn = querySDS().connect("/Environment/Software", myProcessSession(), RTM_LOCK_READ, 1000*60);
+    Owned<IRemoteConnection> conn = querySDS().connect("/Environment/Software", myProcessSession(), RTM_LOCK_READ, daliConnectTimeoutMs);
     if (!conn) {
         ERRLOG("Could not connect to /Environment/Software");
         return;
@@ -1835,7 +1870,7 @@ static void convertBinBranch(IPropertyTree &cluster,const char *branch)
 
 static void getxref(const char *dst)
 {
-    Owned<IRemoteConnection> conn = querySDS().connect("DFU/XREF",myProcessSession(),RTM_LOCK_READ, INFINITE);
+    Owned<IRemoteConnection> conn = querySDS().connect("DFU/XREF",myProcessSession(),RTM_LOCK_READ, daliConnectTimeoutMs);
     Owned<IPropertyTree> root = createPTreeFromIPT(conn->getRoot());
     Owned<IPropertyTreeIterator> iter = root->getElements("Cluster");
     ForEach(*iter) {
@@ -2078,7 +2113,8 @@ int main(int argc, char* argv[])
             (memcmp(param,"rawlog=",8)==0)||
             (memcmp(param,"user=",5)==0)||
             (memcmp(param,"password=",9)==0) ||
-            (memcmp(param,"fix=",4)==0))
+            (memcmp(param,"fix=",4)==0) ||
+            (memcmp(param,"timeout=",4)==0))
             props->loadProp(param);
         else if ((i==1)&&(isdigit(*param)||(*param=='.'))&&ep.set(((*param=='.')&&param[1])?(param+1):param,DALI_SERVER_PORT))
             props->setProp("server",ep.getUrlStr(tmps.clear()).str());
@@ -2149,6 +2185,7 @@ int main(int argc, char* argv[])
                 userDesc->set(tmps.str(),ps.str());
                 queryDistributedFileDirectory().setDefaultUser(userDesc);
             }
+            daliConnectTimeoutMs = 1000 * props->getPropInt("timeout", DEFAULT_DALICONNECT_TIMEOUT);
             unsigned np = params.ordinality()-1;
             const char *cmd = params.item(0);
             if (stricmp(cmd,"export")==0) {
@@ -2285,7 +2322,11 @@ int main(int argc, char* argv[])
             }
             else if (stricmp(cmd,"listworkunits")==0) {
                 CHECKPARAMS(0,3);
-                listworkunits((np>1)?params.item(1):NULL,(np>2)?params.item(2):NULL,(np>3)?params.item(3):NULL);
+                listworkunits((np>0)?params.item(1):NULL,(np>1)?params.item(2):NULL,(np>2)?params.item(3):NULL);
+            }
+            else if (stricmp(cmd,"listmatches")==0) {
+                CHECKPARAMS(0,3);
+                listmatches((np>0)?params.item(1):NULL,(np>1)?params.item(2):NULL,(np>2)?params.item(3):NULL);
             }
             else if (stricmp(cmd,"workunittimings")==0) {
                 CHECKPARAMS(1,1);

dali/dfu/dfurun.cpp

@@ -462,7 +462,7 @@ public:
     void startListener(const char *queuename,CSDSServerStatus *serverstatus)
     {
         PROGLOG("DFU server waiting on queue %s",queuename);
-        cDFUlistener *lt = new cDFUlistener(this,queuename,false,serverstatus, 1000*60);
+        cDFUlistener *lt = new cDFUlistener(this,queuename,false,serverstatus);
         listeners.append(*lt);
         lt->start();
     }

dali/sasha/sacoalescer.cpp

@@ -125,13 +125,11 @@ void coalesceDatastore(bool force)
             OwnedIFile detachedIFile = createIFile(detachPath.str());
             if (detachedIFile->exists() || iStoreHelper->detachCurrentDelta())
             {
-                OwnedIFileIO iFileIO = detachedIFile->open(IFOread);
-                PROGLOG("COALESCER: Loading delta: %s, size=%"I64F"d", detachName.str(), iFileIO->size());
+                PROGLOG("COALESCER: Loading delta: %s, size=%"I64F"d", detachName.str(), detachedIFile->size());
                 bool noError;
                 Owned<IException> deltaE;
-                try { noError = iStoreHelper->loadDelta(detachName.str(), iFileIO, root); }
+                try { noError = iStoreHelper->loadDelta(detachName.str(), detachedIFile, root); }
                 catch (IException *e) { deltaE.setown(e); noError = false; }
-                iFileIO.clear();
                 if (!noError && 0 != (SH_BackupErrorFiles & configFlags))
                 {
                     iStoreHelper->backup(detachPath.str());

docs/HPCCClientTools/CT_Mods/CT_ECL_CLI.xml

@@ -1446,13 +1446,14 @@ ecl queries copy //192.168.1.10:8010/thor/findperson thor
       <sect2 id="eclpackageadd" role="brk">
         <title>ecl packagemap add</title>
 
-        <para><emphasis role="bold">ecl packagemap add [options]
+        <para><emphasis role="bold">ecl packagemap add [--daliip][options]
         &lt;target&gt; &lt;filename&gt;</emphasis></para>
 
         <para>Examples:</para>
 
         <programlisting>ecl packagemap add  -s=192.168.1.10 roxie mypackagemap.pkg
 ecl packagemap add roxie mypackagemap.pkg --overwrite
+ecl packagemap add roxie mypackagemap.pkg --daliip=192.168.11.11
 </programlisting>
 
         <para></para>
@@ -1498,6 +1499,13 @@ ecl packagemap add roxie mypackagemap.pkg --overwrite
                 </row>
 
                 <row>
+                  <entry>--daliip=</entry>
+
+                  <entry>IP address or hostname of the remote Dali to use for
+                  logical file lookups</entry>
+                </row>
+
+                <row>
                   <entry><emphasis role="bold">Options</emphasis></entry>
                 </row>
 
@@ -1548,6 +1556,7 @@ ecl packagemap add roxie mypackagemap.pkg --overwrite
             </tgroup>
           </informaltable></para>
       </sect2>
+
       <sect2 id="eclpackagedelete" role="brk">
         <title>ecl packagemap delete</title>
 

docs/UsingConfigManager/UsingConfigManager.xml

@@ -1485,7 +1485,8 @@ sudo -u hpcc cp /etc/HPCCSystems/source/NewEnvironment.xml /etc/HPCCSystems/envi
                     <row>
                       <entry><emphasis>maxEclccProcesses</emphasis></entry>
 
-                      <entry>The maximum number of eclcc processes that will be launched in parallel</entry>
+                      <entry>The maximum number of eclcc processes that will
+                      be launched in parallel</entry>
                     </row>
 
                     <row>
@@ -1809,7 +1810,7 @@ sudo -u hpcc cp /etc/HPCCSystems/source/NewEnvironment.xml /etc/HPCCSystems/envi
 
             <para><informaltable colsep="0" frame="none" rowsep="0">
                 <tgroup cols="2">
-                  <colspec align="left" colwidth="155.50pt" />
+                  <colspec align="left" colwidth="165pt" />
 
                   <colspec />
 
@@ -2641,6 +2642,64 @@ sudo -u hpcc cp /etc/HPCCSystems/source/NewEnvironment.xml /etc/HPCCSystems/envi
           <?hard-pagebreak ?>
 
           <sect4>
+            <title>Changing Thor topology</title>
+
+            <para>If you want to designate a different node as the Thor master
+            when setting up a multi-node system, follow these steps.</para>
+
+            <itemizedlist>
+              <listitem>
+                <para>Select <emphasis role="bold">Thor Cluster -
+                mythor</emphasis> in the Navigator panel on the left
+                side.</para>
+              </listitem>
+
+              <listitem>
+                <para>Select the <emphasis role="bold">Topology</emphasis>
+                tab.</para>
+              </listitem>
+
+              <listitem>
+                <para>RT-CLICK on the Master node</para>
+              </listitem>
+
+              <listitem>
+                <?dbfo keep-together="always"?>
+
+                <para>Select the <emphasis role="bold">Swap Master</emphasis>
+                option.</para>
+
+                <para><graphic fileref="images/CM-111-2.jpg"
+                vendor="configmgrSS" /></para>
+
+                <para><informaltable colsep="1" frame="all" rowsep="1">
+                    <?dbfo keep-together="always"?>
+
+                    <tgroup cols="2">
+                      <colspec colwidth="49.50pt" />
+
+                      <colspec />
+
+                      <tbody>
+                        <row>
+                          <entry><inlinegraphic
+                          fileref="images/caution.png" /></entry>
+
+                          <entry>You should only use this feature when
+                          initially setting up your system. If there is data
+                          on the nodes when attempting to Swap Master, you run
+                          the risk of losing or corrupting some data.</entry>
+                        </row>
+                      </tbody>
+                    </tgroup>
+                  </informaltable></para>
+              </listitem>
+            </itemizedlist>
+          </sect4>
+
+          <?hard-pagebreak ?>
+
+          <sect4>
             <title>ThorCluster Attributes</title>
 
             <para>This section describes the Thor Cluster Attributes
@@ -3059,6 +3118,16 @@ sudo -u hpcc cp /etc/HPCCSystems/source/NewEnvironment.xml /etc/HPCCSystems/envi
             component's configuration. This can be useful to keep a record of
             changes and to communicate this information to peers.</para>
           </sect4>
+
+          <sect4>
+            <title>Swap Master</title>
+
+            <para>The swap master option now allows you to easily swap out the
+            Thor Master.***</para>
+
+            <para><graphic fileref="images/CM-186.jpg"
+            vendor="configmgrSS" /></para>
+          </sect4>
         </sect3>
 
         <sect3 id="Roxie" role="brk">

ecl/agentexec/agentexec.cpp

@@ -147,7 +147,7 @@ int CEclAgentExecutionServer::run()
         while (started)
         {
             PROGLOG("AgentExec: Waiting on queue(s) '%s'", queueNames.str());
-            Owned<IJobQueueItem> item = queue->dequeue(WAIT_FOREVER);
+            Owned<IJobQueueItem> item = queue->dequeue();
             if (item.get())
             {
                 StringAttr wuid;

ecl/ecl-package/ecl-package.cpp

@@ -226,12 +226,6 @@ public:
     }
     virtual bool parseCommandLineOptions(ArgvIterator &iter)
     {
-        if (iter.done())
-        {
-            usage();
-            return false;
-        }
-
         for (; !iter.done(); iter.next())
         {
             const char *arg = iter.query();
@@ -253,12 +247,6 @@ public:
     {
         if (!EclCmdCommon::finalizeOptions(globals))
             return false;
-        if (optTarget.isEmpty())
-        {
-            fprintf(stderr, "\nTarget cluster must be specified\n");
-            usage();
-            return false;
-        }
         return true;
     }
     virtual int processCMD()
@@ -267,6 +255,7 @@ public:
 
         Owned<IClientListPackageRequest> request = packageProcessClient->createListPackageRequest();
         request->setTarget(optTarget);
+        request->setProcess("*");
 
         Owned<IClientListPackageResponse> resp = packageProcessClient->ListPackage(request);
         if (resp->getExceptions().ordinality())
@@ -350,12 +339,6 @@ public:
     {
         if (!EclCmdCommon::finalizeOptions(globals))
             return false;
-        if (optTarget.isEmpty())
-        {
-            fprintf(stderr, "\nTarget cluster must be specified\n");
-            usage();
-            return false;
-        }
         return true;
     }
     virtual int processCMD()
@@ -364,6 +347,7 @@ public:
 
         Owned<IClientGetPackageRequest> request = packageProcessClient->createGetPackageRequest();
         request->setTarget(optTarget);
+        request->setProcess("*");
 
         Owned<IClientGetPackageResponse> resp = packageProcessClient->GetPackage(request);
         if (resp->getExceptions().ordinality())

ecl/eclccserver/eclccserver.cpp

@@ -433,6 +433,7 @@ class EclccServer : public CInterface, implements IThreadFactory, implements IAb
     unsigned maxThreadsActive;
     bool running;
     CSDSServerStatus serverstatus;
+    Owned<IJobQueue> queue;
 
 public:
     IMPLEMENT_IINTERFACE;
@@ -455,7 +456,7 @@ public:
     void run()
     {
         DBGLOG("eclccServer (%d threads) waiting for requests on queue(s) %s", poolSize, queueName.get());
-        Owned<IJobQueue> queue = createJobQueue(queueName.get());
+        queue.setown(createJobQueue(queueName.get()));
         queue->connect();
         running = true;
         LocalIAbortHandler abortHandler(*this);
@@ -463,7 +464,7 @@ public:
         {
             try
             {
-                Owned<IJobQueueItem> item = queue->dequeue(1000);
+                Owned<IJobQueueItem> item = queue->dequeue();
                 if (item.get())
                 {
                     try
@@ -508,6 +509,8 @@ public:
     virtual bool onAbort() 
     {
         running = false;
+        if (queue)
+            queue->cancelAcceptConversation();
         return false;
     }
 };

ecl/hqlcpp/hqlcpp.cpp

@@ -10721,7 +10721,10 @@ void HqlCppTranslator::assignCastUnknownLength(BuildCtx & ctx, const CHqlBoundTa
                                 throwError(HQLERR_CastInfiniteString);
 
                             ITranslationInfo * translator = queryDefaultTranslation(tgtset, srcset);
-                            funcName = createIdentifierAtom(translator->queryVarRtlFunction());
+                            if (translator)
+                                funcName = createIdentifierAtom(translator->queryVarRtlFunction());
+                            else
+                                funcName = str2StrXAtom;
                         }
                     }
                     break;

ecl/hqlcpp/hqlhtcpp.cpp

@@ -12164,7 +12164,7 @@ void HqlCppTranslator::doBuildAggregateProcessTransform(BuildCtx & ctx, BoundRow
 
         IHqlExpression * src = cur->queryChild(1);
         IHqlExpression * arg = src->queryChild(0);
-        IHqlExpression * cond = src->queryChild(1);
+        IHqlExpression * cond = queryRealChild(src, 1);
 
         BuildCtx condctx(ctx);
         node_operator srcOp = src->getOperator();

ecl/regress/issue8229.ecl

@@ -0,0 +1,34 @@
+/*##############################################################################
+
+    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.
+############################################################################## */
+
+
+namesRecord :=
+            RECORD
+string        surname;
+string        forename;
+integer6        age := 25;
+            END;
+
+d := dataset([{'Hawthorn','Gavin',35},
+              {'Hawthorn','Abigail',2},
+              {'Smith','John',57},
+              {'Smith','Gavin',12}
+              ], namesRecord);
+
+
+t := TABLE(d, { HASHMD5((data)surname[length(surname)-6..length(surname)] + (unsigned)age) } );
+output(t);

ecl/regress/issue8268.ecl

@@ -0,0 +1,35 @@
+/*##############################################################################
+
+    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.
+############################################################################## */
+
+import std;
+Tbl1RecDef := RECORD string10 fname; string10 lname; unsigned1 prange; string10 street; unsigned1 zips; unsigned1 age; string2 birth_state; string3 birth_month; unsigned1 one; unsigned8 id; unsigned8 __filepos {virtual(fileposition)}; END;
+
+Tbl1DS := DATASET('~certification::full_test_distributed', Tbl1RecDef,FLAT);
+
+Idx := INDEX(Tbl1DS, {lname, fname, prange, street, zips, age, birth_state, birth_month},{ __filepos },'~certification::full_test_distributed_index');
+
+IdxDS := Idx(KEYED(  age > 10  ) and WILD(  lname  ) and WILD(  fname  ) and WILD(  prange  ) and WILD(  street  ) and WILD(  zips  ) and (age > 10 ));
+
+SelectStruct := RECORD
+maxOut := MAX( IdxDS( age > 10 ), IdxDS.age, KEYED );
+string10 lname := IdxDS.lname;
+string10 fname := IdxDS.fname;
+END;
+
+IdxDSTable := TABLE(IdxDS, SelectStruct );
+
+OUTPUT(CHOOSEN(IdxDSTable,100),NAMED('JDBCSelectQueryResult'));

esp/bindings/SOAP/Platform/soapbind.cpp

@@ -270,9 +270,14 @@ void CSoapComplexType::appendContent(IEspContext* ctx, MemoryBuffer& buffer, Str
     StringBuffer content;
     if (ctx && ctx->getResponseFormat()==ESPSerializationJSON)
     {
+        const char *jsonp = ctx->queryRequestParameters()->queryProp("jsonp");
+        if (jsonp && *jsonp)
+            content.append(jsonp).append('(');
         content.append('{');
         serializeStruct(ctx, content, (const char *)NULL);
         content.append('}');
+        if (jsonp && *jsonp)
+            content.append(");");
         mimetype.set("application/json; charset=UTF-8");
     }
     else

esp/bindings/http/platform/httptransport.cpp

@@ -2339,35 +2339,65 @@ int CHttpResponse::sendException(IEspHttpException* e)
     send();
     return 0;
 }
+
+StringBuffer &toJSON(StringBuffer &json, IMultiException *me, const char *callback)
+{
+    IArrayOf<IException> &exs = me->getArray();
+    if (callback && *callback)
+        json.append(callback).append('(');
+    appendJSONName(json.append("{"), "Exceptions").append("{");
+    appendJSONValue(json, "Source", me->source());
+    appendJSONName(json, "Exception").append("[");
+    ForEachItemIn(i, exs)
+    {
+        IException &e = exs.item(i);
+        if (i>0)
+            json.append(",");
+        StringBuffer msg;
+        appendJSONValue(json.append("{"), "Code", e.errorCode());
+        appendJSONValue(json, "Message", e.errorMessage(msg).str());
+        json.append("}");
+    }
+    json.append("]}}");
+    if (callback && *callback)
+        json.append(");");
+    return json;
+}
+
 bool CHttpResponse::handleExceptions(IXslProcessor *xslp, IMultiException *me, const char *serv, const char *meth, const char *errorXslt)
 {
     IEspContext *context=queryContext();
     if (me->ordinality()>0)
     {
-        StringBuffer text;
-        me->errorMessage(text);
-        text.append('\n');
-        WARNLOG("Exception(s) in %s::%s - %s", serv, meth, text.str());
+        StringBuffer msg;
+        WARNLOG("Exception(s) in %s::%s - %s", serv, meth, me->errorMessage(msg).append('\n').str());
 
-        bool returnXml = context->queryRequestParameters()->hasProp("rawxml_");
-        if (errorXslt || returnXml)
+        StringBuffer content;
+        switch (context->getResponseFormat())
         {
-            me->serialize(text.clear());
-            if (returnXml)
-            {
-                setContent(text.str());
-                setContentType(HTTP_TYPE_APPLICATION_XML);
-            }
-            else
+        case ESPSerializationJSON:
+        {
+            setContentType(HTTP_TYPE_APPLICATION_JSON_UTF8);
+            toJSON(content, me, context->queryRequestParameters()->queryProp("jsonp"));
+            break;
+        }
+        case ESPSerializationXML:
+            setContentType(HTTP_TYPE_APPLICATION_XML);
+            me->serialize(content);
+            break;
+        case ESPSerializationANY:
+        default:
             {
-                StringBuffer theOutput;
-                xslTransformHelper(xslp, text.str(), errorXslt, theOutput, context->queryXslParameters());
-                setContent(theOutput.str());
-                setContentType("text/html");
+            if (!errorXslt || !*errorXslt)
+                return false;
+            setContentType("text/html");
+            StringBuffer xml;
+            xslTransformHelper(xslp, me->serialize(xml), errorXslt, content, context->queryXslParameters());
             }
-            send();
-            return true;
         }
+        setContent(content);
+        send();
+        return true;
     }
     return false;
 }

esp/bindings/http/platform/httptransport.hpp

@@ -60,7 +60,7 @@
 #define HTTP_TYPE_TEXT_XML_UTF8                 "text/xml; charset=UTF-8"
 #define HTTP_TYPE_APPLICATION_XML_UTF8          "application/xml; charset=UTF-8"
 #define HTTP_TYPE_SOAP_UTF8                     "application/soap; charset=UTF-8"
-
+#define HTTP_TYPE_APPLICATION_JSON_UTF8         "application/json; charset=UTF-8"
 
 #define HTTP_STATUS_OK_CODE                 200
 #define HTTP_STATUS_NO_CONTENT_CODE         204

esp/eclwatch/ws_XSLT/clusterprocesses.xslt

@@ -98,6 +98,9 @@
 
                   function onLoad()
                   {
+                    if (countTCs > 0)
+                      setReloadFunction('reloadPage');
+
                     document.getElementsByName('TargetClusters.itemcount')[0].value = countTCs;
                     initSelection('resultsTable');
                     initPreflightControls();
@@ -121,6 +124,11 @@
                       }
                     }
                     document.forms[0].submitBtn.disabled = clusterChecked == 0;
+                    if (countTCs > 1)
+                    {
+                      document.getElementById( 'TargetClusters.All' ).checked = clusterChecked==countTCs;
+                      document.getElementById( 'TargetClusters.All2' ).checked = clusterChecked==countTCs;
+                    }
 
                     var table = document.getElementById('resultsTable');
                     if (table)
@@ -226,23 +234,38 @@
 
                   function clickTCCheckbox(type, name, o) 
                   {
-                    if (o.checked)
-                      clusterChecked++;
-                    else
-                      clusterChecked--;
+                    if (countTCs < 1)
+                      return;
 
-                    selectObj = document.getElementById( 'methodObj' );
-                    if (selectObj.selectedIndex == 0)
+                    if ((o.id == 'TargetClusters.All') || (o.id == 'TargetClusters.All2'))
                     {
-                      if (clusterChecked > 0)
-                      {
-                        document.forms[0].submitBtn.disabled = false;
-                      }
+                      for (i=0; i<countTCs; i++)
+                        document.getElementById( 'TargetClusters.' + i ).checked = o.checked;
+                      clusterChecked = o.checked? countTCs:0;
+
+                      if (o.id == 'TargetClusters.All')
+                        document.getElementById( 'TargetClusters.All2' ).checked = o.checked;
                       else
+                        document.getElementById( 'TargetClusters.All' ).checked = o.checked;
+                    }
+                    else
+                    {
+                      if (o.checked)
+                        clusterChecked++;
+                      else
+                        clusterChecked--;
+
+                      if (countTCs > 1)
                       {
-                        document.forms[0].submitBtn.disabled = true;
+                        document.getElementById( 'TargetClusters.All' ).checked = (clusterChecked == countTCs)? true: false;
+                        document.getElementById( 'TargetClusters.All2' ).checked = (clusterChecked == countTCs)? true: false;
                       }
                     }
+
+                    if (clusterChecked > 0)
+                      document.forms[0].submitBtn.disabled = false;
+                    else
+                      document.forms[0].submitBtn.disabled = true;
                   }
                 ]]></xsl:text>
               </script>
@@ -251,9 +274,6 @@
             </xsl:if>
           </head>
           <body class="yui-skin-sam"  onload="nof5();onLoad();">
-          <xsl:if test="TargetClusterInfoList">
-            <xsl:attribute name="onload">setReloadFunction('reloadPage');onLoad()</xsl:attribute>
-          </xsl:if>
           <form id="listitems" action="/ws_machine/GetTargetClusterInfo" method="post">
             <input type="hidden" name="Path" value="{$reqInfo/Path}"/>
             <input type="hidden" name="Cluster" value="{$clusterName}"/>
@@ -292,12 +312,34 @@
                            </tr>
                            <tr>
                               <td>
+                                <xsl:if test="TargetClusterInfoList/TargetClusterInfo[2]">
+                                    <table cellpadding="0" width="100%">
+                                        <tr>
+                                            <th id="selectAll" width="1%" style="padding-left:4px">
+                                                <input type="checkbox" id="TargetClusters.All" name="TargetClusters.ALL"
+                                                       title="Select all target clusters" onclick="return clickTCCheckbox('', '', this);"></input>
+                                            </th>
+                                            <th colspan="5" align="left">Select All / None</th>
+                                        </tr>
+                                    </table>
+                                </xsl:if>
                                 <xsl:for-each select="TargetClusterInfoList/TargetClusterInfo">
                                   <xsl:call-template name="show-cluster">
                                     <xsl:with-param name="type" select="Type"/>
                                     <xsl:with-param name="name" select="Name"/>
                                   </xsl:call-template>
                                 </xsl:for-each>
+                                <xsl:if test="TargetClusterInfoList/TargetClusterInfo[2]">
+                                      <table cellpadding="0" width="100%">
+                                          <tr>
+                                              <th id="selectAll2" width="1%" style="padding-left:4px">
+                                                  <input type="checkbox" id="TargetClusters.All2" name="TargetClusters.ALL2"
+                                                         title="Select all target clusters" onclick="return clickTCCheckbox('', '', this);"></input>
+                                              </th>
+                                              <th colspan="5" align="left">Select All / None</th>
+                                          </tr>
+                                      </table>
+                                </xsl:if>
                                 <b>Fetched: </b>
                                 <xsl:value-of select="TimeStamp"/>
                                 <br/>
@@ -333,7 +375,7 @@
     <table id="resultsTable" class="sort-table" width="100%">
       <tr class="grey">
         <td valign="top" width="20">
-          <input type="checkbox" name="TargetClusters.{count(preceding::TargetClusterInfo)}" checked="1"
+          <input type="checkbox" id="TargetClusters.{count(preceding::TargetClusterInfo)}" name="TargetClusters.{count(preceding::TargetClusterInfo)}" checked="1"
                                 value="{$type}:{$name}" title="Select this target cluster" onclick="return clickTCCheckbox('{$type}', '{$name}', this);"></input>
         </td>
         <td align="left" width="20">

esp/eclwatch/ws_XSLT/index.xslt

@@ -711,6 +711,21 @@
                 <xsl:otherwise>0</xsl:otherwise>
             </xsl:choose>
         </xsl:variable>
+        <xsl:variable name="showWarning">
+            <xsl:choose>
+                <xsl:when test="$clusterType = 'DFUserver' or $clusterType = 'ECLCCserver' or $clusterType = 'ECLagent'">
+                    <xsl:choose>
+                        <xsl:when test="$status='paused'"> Queue paused </xsl:when>
+                        <xsl:when test="$status='stopped'"> Queue stopped </xsl:when>
+                        <xsl:otherwise></xsl:otherwise>
+                    </xsl:choose>
+                </xsl:when>
+                <xsl:when test="($status2='1') or ($status2='2')"> Queue paused </xsl:when>
+                <xsl:when test="$status2='3'"> Queue paused - Cluster stopped </xsl:when>
+                <xsl:when test="$status2='5'"> Cluster stopped </xsl:when>
+                <xsl:otherwise></xsl:otherwise>
+            </xsl:choose>
+        </xsl:variable>
         <table class="clusters" border="2" frame="box" rules="groups" style="margin-bottom:5px">
             <tr>
                 <xsl:variable name="pid" select="position()"/>
@@ -794,6 +809,11 @@
                                 <xsl:value-of select="$clusterType"/> - <xsl:value-of select="$queue"/>
                             </xsl:otherwise>
                         </xsl:choose>
+                        <xsl:if test="string-length($showWarning)">
+                            <span style="background: #C00">
+                                <xsl:copy-of select="$showWarning"/>
+                            </span>
+                        </xsl:if>
                     </a>
                 </td>
             </tr>

esp/eclwatch/ws_XSLT/targetclusters.xslt

@@ -106,16 +106,10 @@
             document.getElementsByName('TargetClusters.itemcount')[0].value = countTCs;
             initSelection('resultsTable');
 
-            if (countTCs > 0)
-            {
-              for (i=0; i<countTCs; i++)
-              { 
-                var ch = document.getElementById('TargetClusters.'+i);
-                if (ch && ch.checked)
-                {
-                  clusterChecked++;
-                }
-              }
+            if (countTCs > 1)
+            { //no target cluster is selected when the page is loaded.
+              document.getElementById( 'TargetClusters.All' ).checked = false;
+              document.getElementById( 'TargetClusters.All2' ).checked = false;
             }
 
             initPreflightControls();
@@ -201,24 +195,38 @@
 
           function clickTCCheckbox(type, name, o) 
           {
-            selectObj = document.getElementById( 'methodObj' );
+            if (countTCs < 1)
+              return;
 
-            if (o.checked)
-              clusterChecked++;
-            else
-              clusterChecked--;
+            if ((o.id == 'TargetClusters.All') || (o.id == 'TargetClusters.All2'))
+            {
+              for (i=0; i<countTCs; i++)
+                document.getElementById( 'TargetClusters.' + i ).checked = o.checked;
+              clusterChecked = o.checked? countTCs:0;
 
-            if (selectObj.selectedIndex == 0)
+              if (o.id == 'TargetClusters.All')
+                document.getElementById( 'TargetClusters.All2' ).checked = o.checked;
+              else
+                document.getElementById( 'TargetClusters.All' ).checked = o.checked;
+            }
+            else
             {
-              if (clusterChecked > 0)
-              {
-                document.forms[0].submitBtn.disabled = false;
-              }
+              if (o.checked)
+                clusterChecked++;
               else
+                clusterChecked--;
+
+              if (countTCs > 1)
               {
-                document.forms[0].submitBtn.disabled = true;
+                document.getElementById( 'TargetClusters.All' ).checked = (clusterChecked == countTCs)? true: false;
+                document.getElementById( 'TargetClusters.All2' ).checked = (clusterChecked == countTCs)? true: false;
               }
             }
+
+            if (clusterChecked > 0)
+              document.forms[0].submitBtn.disabled = false;
+            else
+              document.forms[0].submitBtn.disabled = true;
           }
 
           ]]></xsl:text>
@@ -229,12 +237,37 @@
                 <h3>Target Clusters:</h3>
                 <form id="listitems" action="/ws_machine/GetTargetClusterInfo" method="post">
                     <input type="hidden" name="TargetClusters.itemcount" value=""/>
+                    <xsl:if test="TpTargetClusters/TpTargetCluster[2]">
+                        <table cellpadding="0" width="100%">
+                            <tr>
+                                <th id="selectAll" width="1%" style="padding-left:4px">
+                                    <input type="checkbox" id="TargetClusters.All" name="TargetClusters.ALL"
+                                           title="Select all target clusters" onclick="return clickTCCheckbox('', '', this);"></input>
+                                </th>
+                                <th colspan="5" align="left">Select All / None</th>
+                            </tr>
+                        </table>
+                    </xsl:if>
                     <xsl:for-each select="TpTargetClusters/TpTargetCluster">
                         <xsl:call-template name="show-cluster">
                             <xsl:with-param name="type" select="Type"/>
                             <xsl:with-param name="name" select="Name"/>
                         </xsl:call-template>
                     </xsl:for-each>
+                    <xsl:if test="TpTargetClusters/TpTargetCluster[2]">
+                        <table cellpadding="0" width="100%">
+                            <tr>
+                                <th id="selectAll2" width="1%" style="padding-left:4px">
+                                    <input type="checkbox" id="TargetClusters.All2" name="TargetClusters.ALL2"
+                                           title="Select all target clusters" onclick="return clickTCCheckbox('', '', this);"></input>
+                                </th>
+                                <th colspan="5" align="left">Select All / None</th>
+                            </tr>
+                            <tr>
+                                <td height="20"/>
+                            </tr>
+                        </table>
+                    </xsl:if>
           <xsl:call-template name="ShowPreflightControls">
             <xsl:with-param name="method" select="'GetMachineInfo'"/>
             <xsl:with-param name="getProcessorInfo" select="1"/>

esp/scm/ws_packageprocess.ecm

@@ -72,6 +72,7 @@ ESPresponse [exceptions_inline] DeActivatePackageResponse
 ESPrequest GetPackageRequest
 {
     string Target;
+    string Process;
 };
 
 ESPresponse [exceptions_inline] GetPackageResponse
@@ -83,6 +84,7 @@ ESPresponse [exceptions_inline] GetPackageResponse
 ESPrequest ListPackageRequest
 {
     string Target;
+    string Process;
 };
 
 ESPstruct PackageListData
@@ -94,6 +96,7 @@ ESPstruct PackageListData
 ESPstruct PackageListMapData
 {
     string Id;
+    string Target;
     ESParray<ESPstruct PackageListData> PkgListData;
 };
 

esp/services/ws_ecl/ws_ecl_service.cpp

@@ -2702,6 +2702,14 @@ int CWsEclBinding::onGet(CHttpRequest* request, CHttpResponse* response)
             WsEclWuInfo wsinfo(wuid.str(), qs.str(), qid.str(), context->queryUserId(), context->queryPassword());
             return getWsEclLinks(*context, request, response, wsinfo);
         }
+        else if (strieq(methodName.str(), "soap"))
+        {
+            StringBuffer url;
+            url.append("/WsEcl/forms/soap/").append(thepath);
+            response->redirect(*request, url);
+            return 0;
+        }
+
     }
     catch (IMultiException* mex)
     {

esp/services/ws_packageprocess/ws_packageprocessService.cpp

@@ -268,6 +268,7 @@ void addPackageMapInfo(IPropertyTree *pkgSetRegistry, const char *target, const
 void getPackageListInfo(IPropertyTree *mapTree, IEspPackageListMapData *pkgList)
 {
     pkgList->setId(mapTree->queryProp("@id"));
+    pkgList->setTarget(mapTree->queryProp("@querySet"));
 
     Owned<IPropertyTreeIterator> iter = mapTree->getElements("Package");
     IArrayOf<IConstPackageListData> results;
@@ -298,34 +299,40 @@ void getAllPackageListInfo(IPropertyTree *mapTree, StringBuffer &info)
     }
     info.append("</PackageMap>");
 }
-void listPkgInfo(const char *target, IArrayOf<IConstPackageListMapData>* results)
+void listPkgInfo(const char *target, const char *process, IArrayOf<IConstPackageListMapData>* results)
 {
-    StringBuffer info;
     Owned<IRemoteConnection> globalLock = querySDS().connect("/PackageMaps/", myProcessSession(), RTM_LOCK_WRITE|RTM_CREATE_QUERY, SDS_LOCK_TIMEOUT);
     if (!globalLock)
         throw MakeStringException(PKG_DALI_LOOKUP_ERROR, "Unable to retrieve package information from dali /PackageMaps");
     IPropertyTree *root = globalLock->queryRoot();
+    Owned<IPropertyTree> pkgSetRegistry = getPkgSetRegistry((process && *process) ? process : "*", NULL, true);
+    if (!pkgSetRegistry)
+        throw MakeStringException(PKG_DALI_LOOKUP_ERROR, "Unable to retrieve package information from dali for process %s", (process && *process) ? process : "*");
+
+    StringBuffer xpath("PackageMap");
     if (!target || !*target)
     {
-        info.append("<PackageMaps>");
-        Owned<IPropertyTreeIterator> iter = root->getElements("PackageMap");
+        Owned<IPropertyTreeIterator> iter = pkgSetRegistry->getElements("PackageMap");
         ForEach(*iter)
         {
             Owned<IEspPackageListMapData> res = createPackageListMapData("", "");
             IPropertyTree &item = iter->query();
-            getPackageListInfo(&item, res);
-            results->append(*res.getClear());
+            StringBuffer xpath;
+            const char *id = item.queryProp("@id");
+            if (id)
+            {
+                xpath.append("PackageMap[@id='").append(id).append("']");
+                IPropertyTree *mapTree = root->queryPropTree(xpath);
+                Owned<IEspPackageListMapData> res = createPackageListMapData("", "");
+                getPackageListInfo(mapTree, res);
+                results->append(*res.getClear());
+            }
         }
-        info.append("</PackageMaps>");
     }
     else
     {
-        Owned<IPropertyTree> pkgSetRegistry = getPkgSetRegistry(target, NULL, true);
-        if (!pkgSetRegistry)
-            throw MakeStringException(PKG_DALI_LOOKUP_ERROR, "Unable to retrieve package information from dali for target %s", target);
-
-        Owned<IPropertyTreeIterator> iter = pkgSetRegistry->getElements("PackageMap");
-        info.append("<PackageMaps>");
+        xpath.appendf("[@querySet='%s']", target);
+        Owned<IPropertyTreeIterator> iter = pkgSetRegistry->getElements(xpath.str());
         ForEach(*iter)
         {
             IPropertyTree &item = iter->query();
@@ -340,36 +347,33 @@ void listPkgInfo(const char *target, IArrayOf<IConstPackageListMapData>* results
                 results->append(*res.getClear());
             }
         }
-        info.append("</PackageMaps>");
     }
 }
-void getPkgInfo(const char *target, StringBuffer &info)
+void getPkgInfo(const char *target, const char *process, StringBuffer &info)
 {
     Owned<IRemoteConnection> globalLock = querySDS().connect("/PackageMaps/", myProcessSession(), RTM_LOCK_WRITE|RTM_CREATE_QUERY, SDS_LOCK_TIMEOUT);
     if (!globalLock)
         throw MakeStringException(PKG_DALI_LOOKUP_ERROR, "Unable to retrieve package information from dali /PackageMaps");
     IPropertyTree *root = globalLock->queryRoot();
     Owned<IPropertyTree> tree = createPTree("PackageMaps");
-    if (target)
+    Owned<IPropertyTree> pkgSetRegistry = getPkgSetRegistry((process && *process) ? process : "*", NULL, true);
+    StringBuffer xpath("PackageMap[@active='1']");
+    if (target && *target)
+        xpath.appendf("[@querySet='%s']", target);
+    Owned<IPropertyTreeIterator> iter = pkgSetRegistry->getElements(xpath.str());
+    ForEach(*iter)
     {
-        Owned<IPropertyTree> pkgSetRegistry = getPkgSetRegistry(target, NULL, true);
-        Owned<IPropertyTreeIterator> iter = pkgSetRegistry->getElements("PackageMap[@active='1']");
-        ForEach(*iter)
+        IPropertyTree &item = iter->query();
+        const char *id = item.queryProp("@id");
+        if (id)
         {
-            IPropertyTree &item = iter->query();
-            const char *id = item.queryProp("@id");
-            if (id)
-            {
-                StringBuffer xpath;
-                xpath.append("PackageMap[@id='").append(id).append("']");
-                IPropertyTree *mapTree = root->queryPropTree(xpath);
-                if (mapTree)
-                    mergePTree(tree, mapTree);
-            }
+            StringBuffer xpath;
+            xpath.append("PackageMap[@id='").append(id).append("']");
+            IPropertyTree *mapTree = root->queryPropTree(xpath);
+            if (mapTree)
+                mergePTree(tree, mapTree);
         }
     }
-    else
-        throw MakeStringException(PKG_TARGET_NOT_DEFINED, "No target defined");
 
     toXML(tree, info);
 }
@@ -517,7 +521,8 @@ bool CWsPackageProcessEx::onListPackage(IEspContext &context, IEspListPackageReq
 {
     resp.updateStatus().setCode(0);
     IArrayOf<IConstPackageListMapData> results;
-    listPkgInfo(req.getTarget(), &results);
+    StringAttr process(req.getProcess());
+    listPkgInfo(req.getTarget(), process.length() ? process.get() : "*", &results);
     resp.setPkgListMapData(results);
     return true;
 }
@@ -525,9 +530,9 @@ bool CWsPackageProcessEx::onListPackage(IEspContext &context, IEspListPackageReq
 bool CWsPackageProcessEx::onGetPackage(IEspContext &context, IEspGetPackageRequest &req, IEspGetPackageResponse &resp)
 {
     resp.updateStatus().setCode(0);
-    StringAttr target(req.getTarget());
+    StringAttr process(req.getProcess());
     StringBuffer info;
-    getPkgInfo(target.length() ? target.get() : "*", info);
+    getPkgInfo(req.getTarget(), process.length() ? process.get() : "*", info);
     resp.setInfo(info);
     return true;
 }

esp/services/ws_workunits/ws_workunitsService.cpp

@@ -840,7 +840,7 @@ bool CWsWorkunitsEx::onWUUpdate(IEspContext &context, IEspWUUpdateRequest &req,
         if (origValueChanged(req.getJobname(), req.getJobnameOrig(), s))
             wu->setJobName(s.trim().str());
         if (origValueChanged(req.getDescription(), req.getDescriptionOrig(), s.clear()))
-            wu->setDebugValue("description", (req.getDescription()) ? s.trim().str() : NULL, true);
+            wu->setDebugValue("description", (req.getDescription() && *req.getDescription()) ? s.trim().str() : NULL, true);
 
         double version = context.getClientVersion();
         if (version > 1.04)

esp/xslt/wsecl3_links.xslt

@@ -36,6 +36,9 @@
                     Parameter XML:&nbsp;&nbsp;<a href="/WsEcl/definitions/query/{$pathval}/{$queryname}/resource/soap/{$queryname}.xml?display">display</a>
                 </li>
                 <li>
+                    SOAP (Post SOAP messages to this URL):&nbsp;&nbsp;<a href="/WsEcl/soap/query/{$pathval}/{$queryname}">/WsEcl/soap/query/<xsl:value-of select="$pathval"/>/<xsl:value-of select="$queryname"/></a>
+                </li>
+                <li>
                     WSDL:&nbsp;&nbsp;<a href="/WsEcl/definitions/query/{$pathval}/{$queryname}/main/{$queryname}.wsdl?display">display</a>&nbsp;&nbsp;<a  target="_blank" href="/WsEcl/definitions/query/{$pathval}/{$queryname}/main/{$queryname}.wsdl">link</a>
                 </li>
                 <li>

initfiles/bash/etc/init.d/hpcc_common.in

@@ -720,9 +720,9 @@ create_dropzone() {
     # Creating DropZone directory
     if [ ! -d ${D} ]; then
          mkdir -p $D > /dev/null 2>&1
+         chown -cR $user:$user $D > /dev/null 2>&1
+         chmod 777 $D > /dev/null 2>&1
     fi
-      chown -cR $user:$user $D > /dev/null 2>&1
-      chmod 777 $D > /dev/null 2>&1
     done
 }
 

initfiles/bin/init_eclagent.in

@@ -48,12 +48,12 @@ agentexec 1>/dev/null 2>/dev/null &
 echo $! > $PID_NAME
 wait
 
-if [ -e ${SENTINEL} ]; then
+while [ -e ${SENTINEL} ]; do
     sleep 1
     if [ -e ${SENTINEL} ]; then
         agentexec 1>/dev/null 2>/dev/null &
         echo $! > $PID_NAME
         wait
     fi
-fi
+done
 

initfiles/componentfiles/thor/CMakeLists.txt

@@ -12,7 +12,6 @@ FOREACH( iFILES
     ${CMAKE_CURRENT_SOURCE_DIR}/setup_nfs
     ${CMAKE_CURRENT_SOURCE_DIR}/setup_one_nfs
     ${CMAKE_CURRENT_SOURCE_DIR}/sshslaves
-    ${CMAKE_CURRENT_SOURCE_DIR}/start_slave
     ${CMAKE_CURRENT_SOURCE_DIR}/start_slaves
     ${CMAKE_CURRENT_SOURCE_DIR}/start_thor
     ${CMAKE_CURRENT_SOURCE_DIR}/run_thor

initfiles/componentfiles/thor/run_thor

@@ -40,12 +40,19 @@ while [ 1 ]; do
     echo 'failed to lookup dali group for $groupName'
         exit 1
     fi
-    ## multislaves for bkwd compat. with old environments
-    if [ ${slavespernode} -gt 1 ] || [ "$localthor" = "true" ] || [ "$multislaves" = "true" ]; then
-        $deploydir/makethorgroup
-    fi
+    $deploydir/makethorgroup
+    sort $instancedir/slaves | uniq > $instancedir/uslaves.start
+
+    echo --------------------------
+    echo starting thorslaves ...
 
-    $deploydir/start_slaves
+    logredirect="$logdir/start_slaves_$logpthtail.log"
+    # Would be simpler, if there was simple way to test if ip is local and get rid of 'localthor' setting
+    if [ "$localthor" = "true" ]; then
+        $deploydir/start_slaves $THORMASTER thorslave${LCR} $THORMASTER $THORMASTERPORT $logdir $instancedir $THORNAME $PATH_PRE $logredirect
+    else
+        $deploydir/frunssh $instancedir/uslaves.start "/bin/sh -c '$deploydir/start_slaves %a thorslave${LCR} $THORMASTER $THORMASTERPORT $logdir $instancedir $THORNAME $PATH_PRE $logredirect'" -i:$SSHidentityfile -u:$SSHusername -pe:$SSHpassword -t:$SSHtimeout -a:$SSHretries 2>&1
+    fi
 
     echo thormaster cmd : $instancedir/thormaster_$THORNAME MASTER=$THORMASTER:$THORMASTERPORT
     nohup $instancedir/thormaster_$THORNAME MASTER=$THORMASTER:$THORMASTERPORT 2> /dev/null 1>/dev/null &

initfiles/componentfiles/thor/start_slave

@@ -1,80 +0,0 @@
-#!/bin/bash
-################################################################################
-#    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.
-################################################################################
-
-slavenum=$1
-prog=$2
-master=$3
-logpth=$4
-instancedir=$5
-slaveport=$6
-hpcc_compname=$7
-hpcc_setenv=$8
-logredirect=$9
-
-source ${hpcc_setenv}
-PID_NAME="$PID/${hpcc_compname}_slave_${slavenum}.pid"
-
-if [ $# -lt 9 ]; then
-  echo usage: slavenum prog master logdir workingdir slaveport hpcc_compname hpcc_setenv logredirect
-  exit 1
-fi
-
-# this must match jsocket hard limit
-export handlelimit=32768
-
-sudo /etc/init.d/hpcc-init -c dafilesrv setup
-mkdir -p $instancedir
-mkdir -p `dirname $logredirect`
-exec >>$logredirect 2>&1
-
-cd $instancedir
-
-echo "slave init `date`"
-
-lckfile="$PID/start_slave_${hpcc_compname}_${slavenum}.pid"
-
-# prevent two slaves starting together
-while [ -e $lckfile ]; do
-  echo waiting on lckfile: $lckfile
-  oldpid=`cat $lckfile`
-  if ps h $oldpid; then
-     echo killing pid $oldpid start_slave
-     kill -9 $oldpid
-     rm $lckfile                   # just in case
-     sleep 1
-  else
-     rm -f $lckfile
-  fi
-done
-trap "rm -f $lckfile" exit
-echo $$ > $lckfile
-
-ulimit -c unlimited
-ulimit -n $handlelimit
-
-echo "slave starting `date`"
-
-echo $prog master=$master slave=.:$slaveport slavenum=$slavenum logDir=$logpth
-$prog master=$master slave=.:$slaveport slavenum=$slavenum logDir=$logpth 2>/dev/null 1>/dev/null &
-slavepid=$!
-echo $slavepid > $PID_NAME
-if [ "$slavepid" -eq "0" ]; then
-  echo "failed to start at `date`"
-else
-  echo "slave pid $slavepid started `date`"
-fi
-

initfiles/componentfiles/thor/start_slaves

@@ -15,43 +15,80 @@
 #    limitations under the License.
 ################################################################################
 
-echo --------------------------
-echo starting thorslaves ...
-
-if [ "$localthor" = "true" ]; then
-        ulimit -c unlimited
-        ulimit -n $handlelimit
-        let "n = 1";
-        for slave in $(cat $instancedir/thorgroup); do
-            slaveport=${slave/*:/}
-            if [ "$slaveport" = "" ]; then
-                slaveport=$THORSLAVEPORT
-            fi
-            $deploydir/start_slave $n thorslave${LCR} $THORMASTER:$THORMASTERPORT $logdir $instancedir $slaveport $THORNAME $PATH_PRE $logdir/start_slave_$logpthtail.$n.log
-            let "n += 1";
-        done
-else
-        ## multislaves for bkwd compat. with old environments
-        if [ ${slavespernode} -gt 1 ] || [ "$multislaves" = "true" ]; then
-            let "n = 1";
-            for slave in $(cat $instancedir/thorgroup); do
-                ip=${slave/:*/}
-                slaveport=${slave/*:/}
-                if [ "$slaveport" = "" ]; then
-                    slaveport=$THORSLAVEPORT
-                fi
-                logredirect="$logdir/start_slave_$logpthtail.$n.log"
-                frunssh $ip "/bin/sh -c '$deploydir/start_slave $n thorslave${LCR} $THORMASTER:$THORMASTERPORT $logdir $instancedir $slaveport $THORNAME $PATH_PRE $logredirect'" -i:$SSHidentityfile -u:$SSHusername -pe:$SSHpassword -t:$SSHtimeout -a:$SSHretries 2>&1 &
-                let "n += 1";
-            done
-            wait
-        else
-            if [ -e $instancedir/thorgroup ]; then
-                mv $instancedir/thorgroup $instancedir/thorgroup.local
-            fi
-            logredirect="$logdir/start_slave_$logpthtail.log"
-            frunssh $instancedir/slaves "/bin/sh -c '$deploydir/start_slave %n thorslave${LCR} $THORMASTER:$THORMASTERPORT $logdir $instancedir $THORSLAVEPORT $THORNAME $PATH_PRE $logredirect'" -i:$SSHidentityfile -u:$SSHusername -pe:$SSHpassword -t:$SSHtimeout -a:$SSHretries 2>&1
-        fi
+ip=$1
+prog=$2
+master=$3
+masterport=$4
+logpth=$5
+instancedir=$6
+hpcc_compname=$7
+hpcc_setenv=$8
+logredirect=$9
+
+source ${hpcc_setenv}
+
+if [ $# -lt 9 ]; then
+  echo usage: ip prog master masterport logdir workingdir hpcc_compname hpcc_setenv logredirect
+  exit 1
 fi
-echo thorslaves started
 
+# this must match jsocket hard limit
+export handlelimit=32768
+
+sudo /etc/init.d/hpcc-init -c dafilesrv setup
+mkdir -p $instancedir
+mkdir -p `dirname $logredirect`
+exec >>$logredirect 2>&1
+
+cd $instancedir
+
+echo "slave(${ip}) init `date`"
+
+lckfile="$PID/start_slaves_${hpcc_compname}_${ip}.pid"
+
+# prevent two of these scripts starting together
+while [ -e $lckfile ]; do
+  echo waiting on lckfile: $lckfile
+  oldpid=`cat $lckfile`
+  if ps h $oldpid; then
+     echo killing pid $oldpid start_slaves
+     kill -9 $oldpid
+     rm $lckfile                   # just in case
+     sleep 1
+  else
+     rm -f $lckfile
+  fi
+done
+trap "rm -f $lckfile" exit
+echo $$ > $lckfile
+
+ulimit -c unlimited
+ulimit -n $handlelimit
+
+echo "slave(s) starting `date`"
+
+# sync to current master thorgroup
+echo rsync $master:$instancedir/thorgroup $instancedir/thorgroup.slave
+rsync $master:$instancedir/thorgroup $instancedir/thorgroup.slave
+
+let "slavenum = 1";
+for slave in $(cat $instancedir/thorgroup.slave); do
+  slaveip=${slave/:*/}
+  if [ ${slaveip} = ${ip} ]; then
+    slaveport=${slave/*:/}
+    if [ "$slaveport" = "" ]; then
+      slaveport=$THORSLAVEPORT
+    fi
+    echo $prog master=$master:$masterport slave=.:$slaveport slavenum=$slavenum logDir=$logpth
+    $prog master=$master:$masterport slave=.:$slaveport slavenum=$slavenum logDir=$logpth 2>/dev/null 1>/dev/null &
+    slavepid=$!
+    PID_NAME="$PID/${hpcc_compname}_slave_${slavenum}.pid"
+    echo $slavepid > $PID_NAME
+    if [ "$slavepid" -eq "0" ]; then
+      echo "failed to start at `date`"
+    else
+      echo "slave pid $slavepid started `date`"
+    fi
+  fi
+  let "slavenum += 1";
+done

initfiles/etc/bash_completion/ecl

@@ -77,7 +77,7 @@ _ecl_nextcommand()
 
 _ecl_opts_common()
 {
-    echo "--help -v --verbose -s --server=<ip> --port=N -u --username=<name> -pw --password=<pw>"
+    echo "--help -v --verbose -s= --server= --port= -u= --username= -pw= --password="
 }
 
 _ecl_opts_queries()
@@ -89,16 +89,16 @@ _ecl_opts_queries()
             echo "--help list copy config"
             ;;
         copy)
-            echo -n "--no-reload --wait=MS --timeLimit=SEC --warnTimeLimit=SEC --memoryLimit=MEM "
+            echo -n "--no-reload --wait= --timeLimit= --warnTimeLimit= --memoryLimit= "
             _ecl_opts_common
             ;;
         config)
-            echo -n "-t --target=<target> --no-files --daliip=<ip> -A --activate --no-reload "
-            echo -n "-O --overwrite --wait=MS --timeLimit=SEC --warnTimeLimit=SEC --memoryLimit=MEM "
+            echo -n "-t= --target= --no-files --daliip= -A --activate --no-reload "
+            echo -n "-O --overwrite --wait= --timeLimit= --warnTimeLimit= --memoryLimit= "
             _ecl_opts_common
             ;;
         list)
-            echo -n "-t --target=<target> --show=<ASU> "
+            echo -n "-t= --target= --show= "
             _ecl_opts_common
             ;;
          *)
@@ -115,7 +115,7 @@ _ecl_opts_roxie()
             echo "--help attach detach check reload"
             ;;
         attach | detach | check | reload)
-            echo -n "--wait=MS "
+            echo -n "--wait= "
             _ecl_opts_common
             ;;
          *)
@@ -150,8 +150,8 @@ _ecl_opts_packagemap()
 
 _ecl_opts_deploy()
 {
-    echo -n "-t --target=<name> -n --name=<name> --main=<definition> --ecl-only --limit=N "
-    echo -n "--wait=<ms> -I<path> -L<path> -f<option>=value --manifest=<file> "
+    echo -n "-t= --target= -n= --name= --main= --ecl-only --limit= "
+    echo -n "--wait= --manifest= "
 }
 
 _ecl_opts_core_file()
@@ -166,12 +166,12 @@ _ecl_opts_core_file()
                 _ecl_opts_common
                 ;;
             publish)
-                echo -n "-A --activate --no-reload --timeLimit=ms --warnTimeLimit=ms --memoryLimit=value "
+                echo -n "-A --activate --no-reload --timeLimit= --warnTimeLimit= --memoryLimit= "
                 _ecl_opts_deploy
                 _ecl_opts_common
                 ;;
             run)
-                echo -n "-in --input=<xml|file> -X<name>=<value> "
+                echo -n "-in= --input= "
                 _ecl_opts_deploy
                 _ecl_opts_common
                 ;;

roxie/ccd/ccdserver.cpp

@@ -19654,13 +19654,13 @@ public:
                 if (map)
                     numParts = map->getNumParts();
                 else
-                {
                     numParts = 0;
-                    eof = true;
-                    return;
-                }
             }
-            if (useRemote())
+            if (!numParts)
+            {
+                eof = true;
+            }
+            else if (useRemote())
             {
                 remote->onStart(parentExtractSize, parentExtract);
                 remote->setLimits(rowLimit, (unsigned __int64) -1, stopAfter);
@@ -19999,7 +19999,10 @@ public:
 
     virtual const void *nextInGroup()
     {
-        // Note - in remote case this never gets called as input chain is routed to remoteResultAdaptor
+        if (eof)
+            return NULL;
+        else if (useRemote())
+            return remote->nextInGroup();
         assertex(xmlParser != NULL);
         try
         {
@@ -20104,7 +20107,10 @@ public:
 
     virtual const void *nextInGroup()
     {
-        // Note - in remote case this never gets called as input chain is routed to remoteResultAdaptor
+        if (eof)
+            return NULL;
+        else if (useRemote())
+            return remote->nextInGroup();
         try
         {
             while (!eof)
@@ -20211,6 +20217,8 @@ public:
     {
         if (eof)
             return NULL;
+        else if (useRemote())
+            return remote->nextInGroup();
         RtlDynamicRowBuilder rowBuilder(rowAllocator);
         unsigned transformedSize = 0;
         if (isKeyed)
@@ -20357,7 +20365,7 @@ public:
         done = true;
 
         unsigned __int64 totalCount = 0;
-        if (helper.canMatchAny())
+        if (helper.canMatchAny() && !eof)
         {
             if (useRemote())
             {
@@ -20486,28 +20494,31 @@ public:
         else
         {
             aggregateHelper.clearAggregate(rowBuilder);
-            if (isKeyed)
+            if (helper.canMatchAny() && !eof)
             {
-                loop
+                if (isKeyed)
                 {
-                    const void *next = cursor->nextMatch();
-                    if (!next)
-                        break;
-                    aggregateHelper.processRow(rowBuilder, next);
+                    loop
+                    {
+                        const void *next = cursor->nextMatch();
+                        if (!next)
+                            break;
+                        aggregateHelper.processRow(rowBuilder, next);
+                    }
                 }
-            }
-            else
-            {
-                assertex(reader != NULL);
-                while (!deserializeSource.eos())
+                else
                 {
-                    prefetcher->readAhead(deserializeSource);
-                    const byte *nextRec = deserializeSource.queryRow();
-                    if (!cursor || !cursor->isFiltered(nextRec))
+                    assertex(reader != NULL);
+                    while (!deserializeSource.eos())
                     {
-                        aggregateHelper.processRow(rowBuilder, nextRec);
+                        prefetcher->readAhead(deserializeSource);
+                        const byte *nextRec = deserializeSource.queryRow();
+                        if (!cursor || !cursor->isFiltered(nextRec))
+                        {
+                            aggregateHelper.processRow(rowBuilder, nextRec);
+                        }
+                        deserializeSource.finishedRow();
                     }
-                    deserializeSource.finishedRow();
                 }
             }
             finalSize = meta.getRecordSize(rowBuilder.getSelf());
@@ -20570,11 +20581,14 @@ public:
         }
         else
         {
-            Owned<IInMemoryFileProcessor> processor = isKeyed ? 
-                createKeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper) :
-                createUnkeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper, manager->createReader(0, 0, 1),
-                                                           ctx->queryCodeContext(), activityId);
-            processor->doQuery(NULL, 0, 0, 0);
+            if (helper.canMatchAny() && !eof)
+            {
+                Owned<IInMemoryFileProcessor> processor = isKeyed ?
+                    createKeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper) :
+                    createUnkeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper, manager->createReader(0, 0, 1),
+                                                               ctx->queryCodeContext(), activityId);
+                processor->doQuery(NULL, 0, 0, 0);
+            }
         }
         gathered = true;
     }
@@ -31097,6 +31111,7 @@ private:
 
 class RoxieWorkUnitListener : public RoxieListener
 {
+    Owned<IJobQueue> queue;
 public:
     RoxieWorkUnitListener(unsigned _poolSize, bool _suspended)
       : RoxieListener(_poolSize, _suspended)
@@ -31118,9 +31133,23 @@ public:
         UNIMPLEMENTED;
     }
 
+    virtual bool stop(unsigned timeout)
+    {
+        if (queue)
+        {
+            DBGLOG("RoxieWorkUnitListener::stop");
+            queue->cancelAcceptConversation();
+        }
+        return RoxieListener::stop(timeout);
+    }
+
     virtual void stopListening()
     {
-        // Nothing to do
+        if (queue)
+        {
+            DBGLOG("RoxieWorkUnitListener::stopListening");
+            queue->cancelAcceptConversation();
+        }
     }
 
     virtual int run()
@@ -31140,12 +31169,12 @@ public:
                         DBGLOG("roxie: Waiting on queue(s) '%s'", queueNames.str());
                     try
                     {
-                        Owned<IJobQueue> queue = createJobQueue(queueNames.str());
+                        queue.setown(createJobQueue(queueNames.str()));
                         queue->connect();
                         daliHelper->noteQueuesRunning(queueNames.str());
                         while (running)
                         {
-                            Owned<IJobQueueItem> item = queue->dequeue(5000);
+                            Owned<IJobQueueItem> item = queue->dequeue();
                             if (item.get())
                             {
                                 if (traceLevel)
@@ -31153,6 +31182,7 @@ public:
                                 pool->start((void *) item->queryWUID());
                             }
                         }
+                        queue.clear();
                     }
                     catch (IDaliClient_Exception *E)
                     {
@@ -31160,6 +31190,7 @@ public:
                             EXCLOG(E, "roxie: Dali connection lost");
                         E->Release();
                         daliHelper->disconnect();
+                        queue.clear();
                     }
                 }
             }

system/mp/mpcomm.cpp

@@ -169,7 +169,18 @@ struct MultiPacketHeader
     size32_t size;
     unsigned idx;
     unsigned numparts;
-    size32_t total;     //
+    size32_t total;
+    StringBuffer &getDetails(StringBuffer &out) const
+    {
+        out.append("MultiPacketHeader: ");
+        out.append("tag=").append((unsigned)tag);
+        out.append(",ofs=").append(ofs);
+        out.append(",size=").append(size);
+        out.append(",idx=").append(idx);
+        out.append(",numparts=").append(numparts);
+        out.append(",total=").append(total);
+        return out;
+    }
 };
 
 
@@ -1008,7 +1019,25 @@ class MultiPacketHandler // TAG_SYS_MULTI
 {
     CIArrayOf<CMultiPacketReceiver> inprogress; // should be ok as not many in progress hopefully (TBD orphans)
     CriticalSection sect;
+    unsigned lastErrMs;
+
+    void logError(unsigned code, MultiPacketHeader &mhdr, CMessageBuffer &msg)
+    {
+        unsigned ms = msTick();
+        if (lastErrMs-ms > 1000) // avoid logging too much
+        {
+            StringBuffer errorMsg("sender=");
+            msg.getSender().getUrlStr(errorMsg).newline();
+            mhdr.getDetails(errorMsg).newline();
+            msg.getDetails(errorMsg);
+            LOG(MCerror, unknownJob, "MultiPacketHandler: protocol error (%d) %s", code, errorMsg.str());
+        }
+        lastErrMs = ms;
+    }
 public:
+    MultiPacketHandler() : lastErrMs(0)
+    {
+    }
     CMessageBuffer *handle(CMessageBuffer * msg)
     {
         if (!msg) 
@@ -1026,7 +1055,8 @@ public:
         }
         if (mhdr.idx==0) {
             if ((mhdr.ofs!=0)||(recv!=NULL)) {
-                LOG(MCerror, unknownJob, "MultiPacketHandler: protocol error (1)");
+                logError(1, mhdr, *msg);
+                delete msg;
                 return NULL;
             }
             recv = new CMultiPacketReceiver;
@@ -1043,7 +1073,7 @@ public:
                  (recv->info.idx+1!=mhdr.idx)||
                  (recv->info.total!=mhdr.total)||
                  (mhdr.ofs+mhdr.size>mhdr.total)) {
-                LOG(MCerror, unknownJob, "MultiPacketHandler: protocol error (2)");
+                logError(2, mhdr, *msg);
                 delete msg;
                 return NULL;
             }
@@ -1054,7 +1084,7 @@ public:
         recv->info = mhdr;
         if (mhdr.idx+1==mhdr.numparts) {
             if (mhdr.ofs+mhdr.size!=mhdr.total) {
-                LOG(MCerror, unknownJob, "MultiPacketHandler: protocol error (3)");
+                logError(3, mhdr, *msg);
                 return NULL;
             }
             msg = recv->msg;

testing/ecl/csv-escaped.ecl

@@ -15,6 +15,10 @@
     limitations under the License.
 ############################################################################## */
 
+// Roxie needs this to resolve files at run time
+#option ('allowVariableRoxieFilenames', 1);
+VarString EmptyString := '' : STORED('dummy');
+
 rec := RECORD
   string foo;
   integer id;
@@ -23,54 +27,54 @@ END;
 
 // Default is no escape
 orig := DATASET([{'this is an \\\'escaped\\\' string', 10, 'while this is not'}], rec);
-OUTPUT(orig, , 'regress::csv-orig', OVERWRITE, CSV);
-escaped := DATASET('regress::csv-orig', rec, CSV);
+OUTPUT(orig, ,'regress::csv-orig'+EmptyString, OVERWRITE, CSV);
+escaped := DATASET('regress::csv-orig'+EmptyString, rec, CSV);
 OUTPUT(escaped);
 
 // Standard escape
 orig2 := DATASET([{'this is an \\\'escaped\\\' string', 10, 'while this is not'}], rec);
-OUTPUT(orig2, , 'regress::csv-escaped', OVERWRITE, CSV);
-escaped2 := DATASET('regress::csv-escaped', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig2, ,'regress::csv-escaped'+EmptyString, OVERWRITE, CSV);
+escaped2 := DATASET('regress::csv-escaped'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped2);
 
 // Multi-char escape
 orig3 := DATASET([{'this is an -=-\'escaped-=-\' string', 10, 'while this is not'}], rec);
-OUTPUT(orig3, , 'regress::csv-escaped-multi', OVERWRITE, CSV);
-escaped3 := DATASET('regress::csv-escaped-multi', rec, CSV(ESCAPE('-=-')));
+OUTPUT(orig3, ,'regress::csv-escaped-multi'+EmptyString, OVERWRITE, CSV);
+escaped3 := DATASET('regress::csv-escaped-multi'+EmptyString, rec, CSV(ESCAPE('-=-')));
 OUTPUT(escaped3);
 
 // Escape the escape
 orig4 := DATASET([{'escape the \\\\ escape', 10, 'escape at the end \\\\'}], rec);
-OUTPUT(orig4, , 'regress::csv-escaped-escape', OVERWRITE, CSV);
-escaped4 := DATASET('regress::csv-escaped-escape', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig4, ,'regress::csv-escaped-escape'+EmptyString, OVERWRITE, CSV);
+escaped4 := DATASET('regress::csv-escaped-escape'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped4);
 
 // Multi-escapes in a row
 orig5 := DATASET([{'multiple escapes \\\\\\\\ in a row', 10, 'multiple at end \\\\\\\\'}], rec);
-OUTPUT(orig5, , 'regress::csv-escaped-many', OVERWRITE, CSV);
-escaped5 := DATASET('regress::csv-escaped-many', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig5, ,'regress::csv-escaped-many'+EmptyString, OVERWRITE, CSV);
+escaped5 := DATASET('regress::csv-escaped-many'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped5);
 
 // Many escapes
 orig6 := DATASET([{'many escapes like \\\'\\\' \\\'  \\\' and \\\\\\\\ \\\\ \\\\  \\\\  \\\\ escape', 10, 'escape at the end \\\''}], rec);
-OUTPUT(orig6, , 'regress::csv-escaped-many-more', OVERWRITE, CSV);
-escaped6 := DATASET('regress::csv-escaped-many-more', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig6, ,'regress::csv-escaped-many-more'+EmptyString, OVERWRITE, CSV);
+escaped6 := DATASET('regress::csv-escaped-many-more'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped6);
 
 // Escape separator
 orig7 := DATASET([{'escaping \\, the \\,\\, \\, \\, separator', 10, 'escape at the end \\,'}], rec);
-OUTPUT(orig7, , 'regress::csv-escaped-separator', OVERWRITE, CSV);
-escaped7 := DATASET('regress::csv-escaped-separator', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig7, ,'regress::csv-escaped-separator'+EmptyString, OVERWRITE, CSV);
+escaped7 := DATASET('regress::csv-escaped-separator'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped7);
 
 // Escape with quotes
 orig8 := DATASET([{'\'escaping\'\'the quote\'', 10, 'au naturel'}], rec);
-OUTPUT(orig8, , 'regress::csv-escaped-escaped', OVERWRITE, CSV);
-escaped8 := DATASET('regress::csv-escaped-escaped', rec, CSV);
+OUTPUT(orig8, ,'regress::csv-escaped-escaped'+EmptyString, OVERWRITE, CSV);
+escaped8 := DATASET('regress::csv-escaped-escaped'+EmptyString, rec, CSV);
 OUTPUT(escaped8);
 
 // Escape with quotes with ESCAPE()
 orig9 := DATASET([{'\'escaping\'\'the quote\'', 10, 'with user defined escape'}], rec);
-OUTPUT(orig9, , 'regress::csv-escaped-escaped2', OVERWRITE, CSV);
-escaped9 := DATASET('regress::csv-escaped-escaped2', rec, CSV(ESCAPE('\\')));
+OUTPUT(orig9, ,'regress::csv-escaped-escaped2'+EmptyString, OVERWRITE, CSV);
+escaped9 := DATASET('regress::csv-escaped-escaped2'+EmptyString, rec, CSV(ESCAPE('\\')));
 OUTPUT(escaped9);

testing/ecl/roxie/key/csv-escaped.xml

@@ -0,0 +1,27 @@
+<Dataset name='Result 2'>
+ <Row><foo>this is an \&apos;escaped\&apos; string</foo><id>10</id><bar>while this is not</bar></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><foo>this is an &apos;escaped&apos; string</foo><id>10</id><bar>while this is not</bar></Row>
+</Dataset>
+<Dataset name='Result 6'>
+ <Row><foo>this is an &apos;escaped&apos; string</foo><id>10</id><bar>while this is not</bar></Row>
+</Dataset>
+<Dataset name='Result 8'>
+ <Row><foo>escape the \ escape</foo><id>10</id><bar>escape at the end \</bar></Row>
+</Dataset>
+<Dataset name='Result 10'>
+ <Row><foo>multiple escapes \\ in a row</foo><id>10</id><bar>multiple at end \\</bar></Row>
+</Dataset>
+<Dataset name='Result 12'>
+ <Row><foo>many escapes like &apos;&apos; &apos;  &apos; and \\ \ \  \  \ escape</foo><id>10</id><bar>escape at the end &apos;</bar></Row>
+</Dataset>
+<Dataset name='Result 14'>
+ <Row><foo>escaping , the ,, , , separator</foo><id>10</id><bar>escape at the end ,</bar></Row>
+</Dataset>
+<Dataset name='Result 16'>
+ <Row><foo>escaping&apos;the quote</foo><id>10</id><bar>au naturel</bar></Row>
+</Dataset>
+<Dataset name='Result 18'>
+ <Row><foo>escaping&apos;the quote</foo><id>10</id><bar>with user defined escape</bar></Row>
+</Dataset>

thorlcr/activities/join/thjoinslave.cpp

@@ -118,7 +118,7 @@ class JoinSlaveActivity : public CSlaveActivity, public CThorDataLink, implement
         {
             if (firstrow)
                 return firstrow.getClear();
-            return base->nextRow();
+            return base->ungroupedNextRow();
         }
 
         void stop()

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -256,6 +256,7 @@ public:
             return;
         inputsConfigured = true;
         unsigned noutputs = container.connectedOutputs.getCount();
+        ActPrintLog("Number of connected outputs: %d", noutputs);
         if (1 == noutputs)
         {
             CIOConnection *io = NULL;
@@ -337,7 +338,7 @@ public:
             try {
                 startInput(input);
                 grouped = input->isGrouped();
-                nstopped = outputs.ordinality();
+                nstopped = container.connectedOutputs.getCount();
                 if (smartBuf)
                     smartBuf->reset();
                 else
@@ -354,6 +355,13 @@ public:
                         ActPrintLog("Spill is 'balanced'");
                         smartBuf.setown(createSharedSmartMemBuffer(this, outputs.ordinality(), queryRowInterfaces(input), NSPLITTER_SPILL_BUFFER_SIZE));
                     }
+                    // mark any unconnected outputs of smartBuf as already stopped.
+                    ForEachItemIn(o, outputs)
+                    {
+                        CDelayedInput *delayedInput = (CDelayedInput *)outputs.item(o);
+                        if (NULL == container.connectedOutputs.queryItem(o))
+                            smartBuf->queryOutput(o)->stop();
+                    }
                 }
                 writer.start();
             }

thorlcr/graph/thgraph.cpp

@@ -836,6 +836,12 @@ bool isGlobalActivity(CGraphElementBase &container)
         }
         case TAKspill:
             return false;
+        case TAKcsvread:
+        {
+            Owned<IHThorCsvReadArg> helper = (IHThorCsvReadArg *)container.helperFactory();
+            // if header lines, then [may] need to co-ordinate across slaves
+            return helper->queryCsvParameters()->queryHeaderLen() > 0;
+        }
 // dependent on child acts?
         case TAKlooprow:
         case TAKloopcount:
@@ -938,7 +944,6 @@ bool isGlobalActivity(CGraphElementBase &container)
 
         case TAKindexread:
         case TAKindexnormalize:
-        case TAKcsvread:
         case TAKxmlread:
         case TAKdiskexists:
         case TAKindexexists:

thorlcr/thorutil/thmem.cpp

@@ -1247,6 +1247,10 @@ protected:
         }
 
         {
+            // Ensure existing callback is cleared, before locked section below, which in turn may add a new callback
+            // Otherwise there is potential for deadlock with the callback mechanism, if it is in the midst of calling this callback.
+            clearSpillingCallback();
+
             CThorSpillableRowArray::CThorSpillableRowArrayLock block(spillableRows);
             if (spillableRowSet)
                 instrms.append(*spillableRowSet->createRowStream());
@@ -1294,8 +1298,15 @@ protected:
         totalRows = 0;
         overflowCount = outStreams = 0;
     }
-
     inline bool spillingEnabled() const { return SPILL_PRIORITY_DISABLE != spillPriority; }
+    void clearSpillingCallback()
+    {
+        if (mmRegistered)
+        {
+            activity.queryJob().queryRowManager()->removeRowBuffer(this);
+            mmRegistered = false;
+        }
+    }
 public:
     CThorRowCollectorBase(CActivityBase &_activity, IRowInterfaces *_rowIf, ICompare *_iCompare, bool _isStable, RowCollectorFlags _diskMemMix, unsigned _spillPriority)
         : activity(_activity),
@@ -1320,8 +1331,7 @@ public:
     ~CThorRowCollectorBase()
     {
         reset();
-        if (mmRegistered)
-            activity.queryJob().queryRowManager()->removeRowBuffer(this);
+        clearSpillingCallback();
     }
     void transferRowsOut(CThorExpandingRowArray &out, bool sort)
     {