Merge branch 'candidate-5.0.6' into candidate-5.2.0

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

Conflicts:
	thorlcr/thorutil/thmem.cpp

plugins/pyembed/pyembed.cpp

@@ -415,6 +415,45 @@ protected:
     CriticalSection lock;
 } globalState;
 
+MODULE_INIT(INIT_PRIORITY_STANDARD)
+{
+    // Make sure we are never unloaded (as Python may crash if we are)
+    // we do this by doing a dynamic load of the pyembed library
+#ifdef _WIN32
+    ::GetModuleFileName((HINSTANCE)&__ImageBase, helperLibraryName, _MAX_PATH);
+    if (strstr(path, "pyembed"))
+    {
+        HINSTANCE h = LoadSharedObject(helperLibraryName, false, false);
+        DBGLOG("LoadSharedObject returned %p", h);
+    }
+#else
+    FILE *diskfp = fopen("/proc/self/maps", "r");
+    if (diskfp)
+    {
+        char ln[_MAX_PATH];
+        while (fgets(ln, sizeof(ln), diskfp))
+        {
+            if (strstr(ln, "libpyembed"))
+            {
+                const char *fullName = strchr(ln, '/');
+                if (fullName)
+                {
+                    char *tail = (char *) strstr(fullName, SharedObjectExtension);
+                    if (tail)
+                    {
+                        tail[strlen(SharedObjectExtension)] = 0;
+                        HINSTANCE h = LoadSharedObject(fullName, false, false);
+                        break;
+                    }
+                }
+            }
+        }
+        fclose(diskfp);
+    }
+#endif
+    return true;
+}
+
 PyObject *PythonThreadContext::getNamedTupleType(const RtlTypeInfo *type)
 {
     if (!lru || (type!=lrutype))

system/jlib/jlzw.cpp

@@ -1806,6 +1806,7 @@ class CCompressedFile : public CInterface, implements ICompressedFileIO
     MemoryAttr prevrowbuf; 
     bool checkcrc;
     bool setcrc;
+    bool writeException;
     Owned<ICompressor> compressor;
     Owned<IExpander> expander;
 
@@ -1872,27 +1873,36 @@ class CCompressedFile : public CInterface, implements ICompressedFileIO
 
     void flush()
     {   
-        curblocknum++;
-        indexbuf.append((unsigned __int64) trailer.expandedSize-overflow.length());
-        offset_t p = ((offset_t)curblocknum)*((offset_t)trailer.blockSize);
-        if (trailer.recordSize==0) {
-            compressor->close();
-            compblklen = compressor->buflen();
-        }
-        if (compblklen) {
-            if (p>trailer.indexPos) { // fill gap
-                MemoryAttr fill;
-                size32_t fl = (size32_t)(p-trailer.indexPos);
-                memset(fill.allocate(fl),0xff,fl);
-                checkedwrite(trailer.indexPos,fl,fill.get());
-            }   
-            checkedwrite(p,compblklen,compblkptr);
-            p += compblklen;
-            compblklen = 0;
+        try
+        {
+            curblocknum++;
+            indexbuf.append((unsigned __int64) trailer.expandedSize-overflow.length());
+            offset_t p = ((offset_t)curblocknum)*((offset_t)trailer.blockSize);
+            if (trailer.recordSize==0) {
+                compressor->close();
+                compblklen = compressor->buflen();
+            }
+            if (compblklen) {
+                if (p>trailer.indexPos) { // fill gap
+                    MemoryAttr fill;
+                    size32_t fl = (size32_t)(p-trailer.indexPos);
+                    memset(fill.allocate(fl),0xff,fl);
+                    checkedwrite(trailer.indexPos,fl,fill.get());
+                }
+                checkedwrite(p,compblklen,compblkptr);
+                p += compblklen;
+                compblklen = 0;
+            }
+            trailer.indexPos = p;
+            if (trailer.recordSize==0) {
+                compressor->open(compblkptr, trailer.blockSize);
+            }
         }
-        trailer.indexPos = p; 
-        if (trailer.recordSize==0) {
-            compressor->open(compblkptr, trailer.blockSize);
+        catch (IException *e)
+        {
+            writeException = true;
+            EXCLOG(e, "CCompressedFile::flush");
+            throw;
         }
     }
 
@@ -1989,6 +1999,7 @@ public:
         compressor.set(_compressor);
         expander.set(_expander);
         setcrc = _setcrc;
+        writeException = false;
         memcpy(&trailer,&_trailer,sizeof(trailer));
         mode = _mode;
         curblockpos = 0;
@@ -2046,7 +2057,15 @@ public:
     }
     virtual ~CCompressedFile()
     {
-        close();
+        if (!writeException)
+        {
+            try { close(); }
+            catch (IException *e)
+            {
+                EXCLOG(e, "~CCompressedFile");
+                e->Release();
+            }
+        }
     }
 
     virtual offset_t size()                                             

thorlcr/thorutil/thmem.cpp

@@ -1250,31 +1250,41 @@ rowidx_t CThorSpillableRowArray::save(IFile &iFile, bool useCompression, const c
         nextCBI = nextCB->queryRecordNumber();
     }
     Owned<IExtRowWriter> writer = createRowWriter(&iFile, rowIf, rwFlags);
-    const void **rows = getBlock(n);
-    for (rowidx_t i=0; i < n; i++)
+    rowidx_t i=0;
+    try
     {
-        const void *row = rows[i];
-        assertex(row || allowNulls);
-        if (i == nextCBI)
+        const void **rows = getBlock(n);
+        while (i<n)
         {
-            writer->flush();
-            do
+            const void *row = rows[i];
+            assertex(row || allowNulls);
+            if (i == nextCBI)
             {
-                nextCB->filePosition(writer->getPosition());
-                if (cbCopy.ordinality())
+                writer->flush();
+                do
                 {
-                    nextCB = &cbCopy.popGet();
-                    nextCBI = nextCB->queryRecordNumber();
+                    nextCB->filePosition(writer->getPosition());
+                    if (cbCopy.ordinality())
+                    {
+                        nextCB = &cbCopy.popGet();
+                        nextCBI = nextCB->queryRecordNumber();
+                    }
+                    else
+                        nextCBI = RCIDXMAX; // indicating no more
                 }
-                else
-                    nextCBI = RCIDXMAX; // indicating no more
+                while (i == nextCBI); // loop as may be >1 IWritePosCallback at same pos
             }
-            while (i == nextCBI); // loop as may be >1 IWritePosCallback at same pos
+            rows[i++] = NULL;
+            writer->putRow(row); // NB: putRow takes ownership/should avoid leaking if fails
         }
-        writer->putRow(row);
-        rows[i] = NULL;
+        writer->flush();
+    }
+    catch (IException *e)
+    {
+        EXCLOG(e, "CThorSpillableRowArray::save");
+        firstRow += i; // ensure released rows are noted.
+        throw;
     }
-    writer->flush();
     firstRow += n;
     offset_t bytesWritten = writer->getPosition();
     writer.clear();
@@ -1437,10 +1447,9 @@ protected:
         tempPrefix.appendf("spill_%d", activity.queryActivityId());
         GetTempName(tempName, tempPrefix.str(), true);
         Owned<IFile> iFile = createIFile(tempName.str());
-        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
-
+        spillFiles.append(new CFileOwner(iFile.getLink()));
         ++overflowCount;
 
         return true;