Merge branch 'candidate-4.2.8' into closedown-5.0.x

Conflicts:
	thorlcr/thorutil/thmem.cpp
	thorlcr/thorutil/thmem.hpp

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

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -1214,7 +1214,8 @@ public:
     virtual bool addLocalRHSRow(CThorSpillableRowArray &localRhsRows, const void *row)
     {
         LinkThorRow(row);
-        localRhsRows.append(row);
+        if (!localRhsRows.append(row))
+            throw MakeActivityException(this, 0, "Out of memory: Cannot append local rhs row");
         return true;
     }
 // ISmartBufferNotify
@@ -1337,6 +1338,7 @@ protected:
     inline void setBroadcastingSpilt(bool tf) { atomic_set(&spiltBroadcastingRHS, (int)tf); }
     rowidx_t clearNonLocalRows(CThorSpillableRowArray &rows, rowidx_t startPos)
     {
+        CThorArrayLockBlock block(rows);
         rowidx_t clearedRows = 0;
         rowidx_t numRows = rows.numCommitted();
         for (rowidx_t r=startPos; r<numRows; r++)

thorlcr/thorutil/thbuf.cpp

@@ -1693,9 +1693,7 @@ public:
                             eos = true;
                             return NULL;
                         }
-                        const void **toRead = rows.getBlock(rowsToRead);
-                        memcpy(readRows, toRead, rowsToRead * sizeof(void *));
-                        rows.noteSpilled(rowsToRead);
+                        rows.readBlock(readRows, rowsToRead);
                         rowPos = 0;
                         if (writersBlocked)
                         {

thorlcr/thorutil/thmem.cpp

@@ -197,7 +197,6 @@ protected:
 
         VStringBuffer spillPrefixStr("SpillableStream(%d)", SPILL_PRIORITY_SPILLABLE_STREAM); // const for now
         rows.save(*spillFile, useCompression, spillPrefixStr.str()); // saves committed rows
-        rows.noteSpilled(numRows);
         return true;
     }
 
@@ -369,9 +368,7 @@ public:
             if (fetch >= granularity)
                 fetch = granularity;
             // consume 'fetch' rows
-            const void **toRead = rows.getBlock(fetch);
-            memcpy(readRows, toRead, fetch * sizeof(void *));
-            rows.noteSpilled(fetch);
+            rows.readBlock(readRows, fetch);
             numReadRows = fetch;
             pos = 0;
         }
@@ -1173,6 +1170,7 @@ void CThorSpillableRowArray::clearRows()
 
 void CThorSpillableRowArray::compact()
 {
+    CThorArrayLockBlock block(*this);
     assertex(0 == firstRow && numRows == commitRows);
     CThorExpandingRowArray::compact();
     commitRows = numRows;
@@ -1259,6 +1257,7 @@ rowidx_t CThorSpillableRowArray::save(IFile &iFile, bool useCompression, const c
         rows[i] = NULL;
     }
     writer->flush();
+    firstRow += n;
     offset_t bytesWritten = writer->getPosition();
     writer.clear();
     ActPrintLog(&activity, "%s: CThorSpillableRowArray::save done, bytes = %"I64F"d", tracingPrefix, (__int64)bytesWritten);
@@ -1333,12 +1332,21 @@ void CThorSpillableRowArray::swap(CThorSpillableRowArray &other)
     commitRows = otherCommitRows;
 }
 
+void CThorSpillableRowArray::readBlock(const void **outRows, rowidx_t readRows)
+{
+    CThorArrayLockBlock block(*this);
+    dbgassertex(firstRow + readRows <= commitRows);
+    memcpy(outRows, rows + firstRow, readRows*sizeof(void *));
+    firstRow += readRows;
+}
+
 void CThorSpillableRowArray::transferRowsCopy(const void **outRows, bool takeOwnership)
 {
+    CThorArrayLockBlock block(*this);
     if (0 == numRows)
         return;
     assertex(numRows == commitRows);
-    memcpy(outRows, rows, numRows*sizeof(void **));
+    memcpy(outRows, rows, numRows*sizeof(void *));
     if (takeOwnership)
         firstRow = commitRows = numRows = 0;
     else
@@ -1385,6 +1393,7 @@ protected:
 
     bool spillRows()
     {
+        //This must only be called while a lock is held on spillableRows()
         rowidx_t numRows = spillableRows.numCommitted();
         if (numRows == 0)
             return false;
@@ -1405,7 +1414,6 @@ protected:
         spillFiles.append(new CFileOwner(iFile.getLink()));
         VStringBuffer spillPrefixStr("RowCollector(%d)", spillPriority);
         spillableRows.save(*iFile, activity.getOptBool(THOROPT_COMPRESS_SPILLS, true), spillPrefixStr.str()); // saves committed rows
-        spillableRows.noteSpilled(numRows);
 
         ++overflowCount;
 

thorlcr/thorutil/thmem.hpp

@@ -409,11 +409,11 @@ public:
     void unregisterWriteCallback(IWritePosCallback &cb);
     inline void setAllowNulls(bool b) { CThorExpandingRowArray::setAllowNulls(b); }
     void kill();
-    void clearRows();
     void compact();
     void flush();
-    inline bool append(const void *row)
+    inline bool append(const void *row) __attribute__((warn_unused_result))
     {
+        //GH->JCS Should this really be inline?
         assertex(row || allowNulls);
         if (numRows >= maxRows)
         {
@@ -431,36 +431,26 @@ public:
     }
     bool appendRows(CThorExpandingRowArray &inRows, bool takeOwnership);
 
-    //The following can be accessed from the reader without any need to lock
+    //The following must either be accessed within a lock, or when no rows can be appended,
+    //(otherwise flush() might move all the rows, invalidating the indexes - or for query() the row)
     inline const void *query(rowidx_t i) const
     {
-        CThorArrayLockBlock block(*this);
         return CThorExpandingRowArray::query(i);
     }
     inline const void *get(rowidx_t i) const
     {
-        CThorArrayLockBlock block(*this);
         return CThorExpandingRowArray::get(i);
     }
     inline const void *getClear(rowidx_t i)
     {
-        CThorArrayLockBlock block(*this);
         return CThorExpandingRowArray::getClear(i);
     }
 
     //A thread calling the following functions must own the lock, or guarantee no other thread will access
     void sort(ICompare & compare, unsigned maxcores);
     rowidx_t save(IFile &file, bool useCompression, const char *tracingPrefix);
-    const void **getBlock(rowidx_t readRows);
-    inline void noteSpilled(rowidx_t spilledRows)
-    {
-        firstRow += spilledRows;
-    }
-
-    //The block returned is only valid until the critical section is released
 
-    inline rowidx_t firstCommitted() const { return firstRow; }
-    inline rowidx_t numCommitted() const { return commitRows - firstRow; }
+    inline rowidx_t numCommitted() const { return commitRows - firstRow; } //MORE::Not convinced this is very safe!
 
 // access to
     void swap(CThorSpillableRowArray &src);
@@ -490,11 +480,16 @@ public:
     void deserializeRow(IRowDeserializerSource &in) { CThorExpandingRowArray::deserializeRow(in); }
     bool ensure(rowidx_t requiredRows) { return CThorExpandingRowArray::ensure(requiredRows); }
     void transferRowsCopy(const void **outRows, bool takeOwnership);
+    void readBlock(const void **outRows, rowidx_t readRows);
 
     virtual IThorArrayLock &queryLock() { return *this; }
 // IThorArrayLock
     virtual void lock() const { cs.enter(); }
     virtual void unlock() const { cs.leave(); }
+
+private:
+    void clearRows();
+    const void **getBlock(rowidx_t readRows);
 };