Merge branch 'candidate-5.2.8' into candidate-5.4.0

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

initfiles/componentfiles/configxml/roxie.xsd.in

@@ -448,7 +448,7 @@
         </xs:appinfo>
       </xs:annotation>
     </xs:attribute>
-    <xs:attribute name="callbackTimeout" type="xs:nonNegativeInteger" use="optional" default="500">
+    <xs:attribute name="callbackTimeout" type="xs:nonNegativeInteger" use="optional" default="5000">
       <xs:annotation>
         <xs:appinfo>
           <tooltip>Timeout (in ms) before callbacks from agents to server are resent</tooltip>

initfiles/etc/DIR_NAME/environment.xml.in

@@ -799,7 +799,7 @@
                 build="_"
                 buildSet="roxie"
                 callbackRetries="3"
-                callbackTimeout="500"
+                callbackTimeout="5000"
                 channelsPerNode="1"
                 checkCompleted="true"
                 checkFileDate="true"

roxie/ccd/ccdmain.cpp

@@ -49,7 +49,7 @@
 bool shuttingDown = false;
 unsigned numChannels;
 unsigned callbackRetries = 3;
-unsigned callbackTimeout = 500;
+unsigned callbackTimeout = 5000;
 unsigned lowTimeout = 10000;
 unsigned highTimeout = 2000;
 unsigned slaTimeout = 2000;

roxie/ccd/ccdserver.cpp

@@ -24636,9 +24636,9 @@ public:
 
     void processCompletedGroups()
     {
-        loop
+        CriticalBlock c(groupsCrit);
+        while (groups.ordinality())
         {
-            CriticalBlock c(groupsCrit); 
             if (!groups.head()->complete())
                 break;
             Owned<CJoinGroup> head = groups.dequeue();
@@ -24657,8 +24657,6 @@ public:
             }
             else
                 doJoinGroup(head);
-            if (!groups.ordinality())
-                break;
         }
     }
 

testing/regress/ecl/groupchild.ecl

@@ -43,9 +43,10 @@ ds := dataset(
 ], namesRecord);
 
 namesRecord t(namesRecord l) := transform
-    deduped := dedup(group(l.children, name), name);
+    deduped := dedup(group(SORT(l.children, name), name), name);
     cnt(string search) := count(table(deduped(name != search), {count(group)}));
-    self.children := group(project(deduped, transform(childRecord, self.dups := cnt(left.name); self.name := left.name)));
+    notdoe := HAVING(deduped, LEFT.name != 'Doe');
+    self.children := group(project(notdoe, transform(childRecord, self.dups := cnt(left.name); self.name := left.name)));
     self := l;
 end;
 

testing/regress/ecl/key/groupchild.xml

@@ -1,4 +1,4 @@
 <Dataset name='Result 1'>
- <Row><surname>Gavin               </surname><children><Row><name>Smith</name><dups>2</dups></Row><Row><name>Jones</name><dups>3</dups></Row><Row><name>Doe</name><dups>3</dups></Row><Row><name>Smith</name><dups>2</dups></Row></children></Row>
+ <Row><surname>Gavin               </surname><children><Row><name>Jones</name><dups>2</dups></Row><Row><name>Smith</name><dups>2</dups></Row></children></Row>
  <Row><surname>John                </surname><children><Row><name>Bib</name><dups>1</dups></Row><Row><name>Bob</name><dups>1</dups></Row></children></Row>
 </Dataset>

testing/regress/environment.xml.in

@@ -799,7 +799,7 @@
                 build="_"
                 buildSet="roxie"
                 callbackRetries="3"
-                callbackTimeout="500"
+                callbackTimeout="5000"
                 channelsPerNode="1"
                 checkCompleted="true"
                 checkFileDate="true"

thorlcr/graph/thgraph.cpp

@@ -992,6 +992,7 @@ bool isGlobalActivity(CGraphElementBase &container)
 // always local
         case TAKfilter:
         case TAKfilterproject:
+        case TAKfiltergroup:
         case TAKsplit:
         case TAKpipewrite:
         case TAKdegroup:

thorlcr/thorutil/thmem.cpp

@@ -270,12 +270,12 @@ class CSharedSpillableRowSet : public CSpillableStreamBase, implements IInterfac
         {
             pos = 0;
             outputOffset = (offset_t)-1;
-            owner->rows.registerWriteCallback(*this);
+            owner->rows.registerWriteCallback(*this); // NB: CStream constructor called within rows lock
         }
         ~CStream()
         {
             spillStream.clear(); // NB: clear stream 1st
-            owner->rows.unregisterWriteCallback(*this);
+            owner->rows.safeUnregisterWriteCallback(*this);
             owner.clear();
         }
     // IRowStream
@@ -325,17 +325,14 @@ public:
     }
     IRowStream *createRowStream()
     {
+        CRowsLockBlock block(*this);
+        if (spillFile) // already spilled?
         {
-            // already spilled?
-            CRowsLockBlock block(*this);
-            if (spillFile)
-            {
-                block.clearCB = true;
-                unsigned rwFlags = DEFAULT_RWFLAGS;
-                if (preserveNulls)
-                    rwFlags |= rw_grouped;
-                return ::createRowStream(spillFile, rowIf, rwFlags);
-            }
+            block.clearCB = true;
+            unsigned rwFlags = DEFAULT_RWFLAGS;
+            if (preserveNulls)
+                rwFlags |= rw_grouped;
+            return ::createRowStream(spillFile, rowIf, rwFlags);
         }
         return new CStream(*this);
     }
@@ -1174,12 +1171,22 @@ void CThorExpandingRowArray::deserializeExpand(size32_t sz, const void *data)
 
 void CThorSpillableRowArray::registerWriteCallback(IWritePosCallback &cb)
 {
-    CThorArrayLockBlock block(*this);
     writeCallbacks.append(cb); // NB not linked to avoid circular dependency
 }
 
 void CThorSpillableRowArray::unregisterWriteCallback(IWritePosCallback &cb)
 {
+    writeCallbacks.zap(cb);
+}
+
+void CThorSpillableRowArray::safeRegisterWriteCallback(IWritePosCallback &cb)
+{
+    CThorArrayLockBlock block(*this);
+    writeCallbacks.append(cb); // NB not linked to avoid circular dependency
+}
+
+void CThorSpillableRowArray::safeUnregisterWriteCallback(IWritePosCallback &cb)
+{
     CThorArrayLockBlock block(*this);
     writeCallbacks.zap(cb);
 }

thorlcr/thorutil/thmem.hpp

@@ -414,6 +414,8 @@ public:
     }
     void registerWriteCallback(IWritePosCallback &cb);
     void unregisterWriteCallback(IWritePosCallback &cb);
+    void safeRegisterWriteCallback(IWritePosCallback &cb);
+    void safeUnregisterWriteCallback(IWritePosCallback &cb);
     inline void setAllowNulls(bool b) { CThorExpandingRowArray::setAllowNulls(b); }
     inline void setDefaultMaxSpillCost(unsigned defaultMaxSpillCost) { CThorExpandingRowArray::setDefaultMaxSpillCost(defaultMaxSpillCost); }
     inline unsigned queryDefaultMaxSpillCost() const { return CThorExpandingRowArray::queryDefaultMaxSpillCost(); }