HPCC-32584 Ensure that dfuserver closes files cleanly so errors are r…

…eported

Signed-off-by: Gavin Halliday <[email protected]>

common/dllserver/thorplugin.cpp

@@ -559,6 +559,8 @@ const StringArray &HelperDll::queryManifestFiles(const char *type, const char *w
                 OwnedIFileIO o = f->open(IFOcreate);
                 assertex(o.get() != nullptr);
                 o->write(0, len, data);
+                o->close();
+
                 list->append(extractName);
                 if (doTrace(traceJava) && streq(type, "jar"))
                     DBGLOG("Extracted jar resource %u size %u to %s in %u ms", id, len, extractName.str(), msTick() - start);

dali/base/dasds.cpp

@@ -5289,6 +5289,7 @@ class CStoreHelper : implements IStoreHelper, public CInterface
             iFileIO->write(0, sizeof(unsigned), crc);
         if (storeInfo)
             storeInfo->cache.set(filename.str());
+        iFileIO->close();
     }
 
     void updateStoreInfo(const char *base, const char *location, unsigned edition, unsigned *crc, CStoreInfo *storeInfo=NULL)
@@ -5400,6 +5401,7 @@ class CStoreHelper : implements IStoreHelper, public CInterface
 
             OwnedIFileIO detachIPIO = detachIPIFile->open(IFOcreate);
             detachIPIO->write(0, sizeof(storeHelper.mySessId), &storeHelper.mySessId);
+            detachIPIO->close();
             detachIPIO.clear();
             detachIPIFile->rename(activeDetachIPStr.str());
             // check often do not wait any longer than necessary

dali/ft/filecopy.cpp

@@ -1023,6 +1023,7 @@ void FileSprayer::beforeTransfer()
                         io->write(lastOutputOffset-sizeof(null), sizeof(null), &null);
                     }
                 }
+                io->close();
             }
         }
     }

dali/ft/ftbase.ipp

@@ -134,6 +134,10 @@ public:
     virtual offset_t tell() override;
     virtual size32_t write(size32_t len, const void * data) override;
     virtual unsigned __int64 getStatistic(StatisticKind kind) override { return stream->getStatistic(kind); }
+    virtual void close() override
+    {
+        stream->close();
+    }
 
     unsigned getCRC()               { return crc; }
     void setCRC(unsigned long _crc)     { crc = _crc; }

dali/ft/fttransform.cpp

@@ -790,6 +790,8 @@ bool TransferServer::pull()
             assertex(curProgress.status != OutputProgress::StatusRenamed);
             if (curProgress.status != OutputProgress::StatusCopied)
             {
+                if (out)
+                    out->close();
                 out.setown(createIOStream(outio));
                 out->seek(progressOffset, IFSbegin);
                 wrapOutInCRC(curProgress.outputCRC);
@@ -885,6 +887,8 @@ bool TransferServer::pull()
                 }
             }
 
+            if (out)
+                out->close();
             out.setown(createIOStream(outio));
             out->seek(0, IFSbegin);
             wrapOutInCRC(0);
@@ -903,7 +907,10 @@ bool TransferServer::pull()
     }
 
     crcOut.clear();
+    if (out)
+        out->close();
     out.clear();
+
     //Once the transfers have completed, rename the files, and sync file times
     //if replicating...
     if (!isSafeMode)
@@ -997,6 +1004,7 @@ bool TransferServer::push()
                 }
                 outio.setown(createCompressedFileWriter(outio, false, 0, true, compressor, COMPRESS_METHOD_LZ4));
             }
+
             out.setown(createIOStream(outio));
             if (!compressOutput)
                 out->seek(curPartition.outputOffset + curProgress.outputLength, IFSbegin);
@@ -1011,6 +1019,7 @@ bool TransferServer::push()
                 sendProgress(curProgress);
             }
             crcOut.clear();
+            out->close();
             out.clear();
         }
     }

system/jlib/jfcmp.hpp

@@ -440,6 +440,12 @@ class CFcmpStream : public CSimpleInterfaceOf<IFileIOStream>
     {
         return baseio->getStatistic(kind);
     }
+
+    virtual void close() override
+    {
+        flush();
+        baseio->close();
+    }
 };
 
 #endif

system/jlib/jfile.cpp

@@ -2953,7 +2953,11 @@ class CBufferedFileIOStream : public CBufferedFileIOStreamBase
     {
         return io->getStatistic(kind);
     }
-
+    virtual void close() override
+    {
+        flush();
+        io->close();
+    }
 protected:
     IFileIOAttr             io;
 };
@@ -3064,6 +3068,13 @@ class CBufferedAsyncIOStream: public CBufferedFileIOStreamBase
     virtual size32_t directWrite(size32_t len, const void * data) { assertex(false); return 0; }    // shouldn't get called
     virtual offset_t directSize() { waitAsyncWrite(); return io->size(); }
     virtual unsigned __int64 getStatistic(StatisticKind kind) { return io->getStatistic(kind); }
+    virtual void close() override
+    {
+        flush();
+        waitAsyncWrite();
+        waitAsyncRead();
+        io->close();
+    }
 };
 
 
@@ -4571,6 +4582,10 @@ IFileIOStream *createProgressIFileIOStream(IFileIOStream *iFileIOStream, offset_
         {
             return iFileIOStream->getStatistic(kind);
         }
+        virtual void close() override
+        {
+            iFileIOStream->close();
+        }
     };
     return new CProgressIFileIOStream(iFileIOStream, totalSize, msg, periodSecs);
 }

system/jlib/jfile.hpp

@@ -222,6 +222,7 @@ interface IFileIOStream : extends IIOStream
     virtual offset_t size() = 0;
     virtual offset_t tell() = 0;
     virtual unsigned __int64 getStatistic(StatisticKind kind) = 0;
+    virtual void close() = 0;
 };
 
 interface IDiscretionaryLock: extends IInterface

system/jlib/jfile.ipp

@@ -219,6 +219,7 @@ public:
     virtual offset_t tell();
     virtual size32_t write(size32_t len, const void * data);
     virtual unsigned __int64 getStatistic(StatisticKind kind) { return io->getStatistic(kind); }
+    virtual void close() override { io->close(); }
 protected:
     Linked<IFileIO>     io;
     offset_t            curOffset;
@@ -238,6 +239,7 @@ public:
     virtual offset_t tell();
     virtual size32_t write(size32_t len, const void * data);
     virtual unsigned __int64 getStatistic(StatisticKind kind) { return stream->getStatistic(kind); }
+    virtual void close() override { stream->close(); }
 protected:
     Linked<IFileIOStream>     stream;
 };

tools/dumpkey/dumpkey.cpp

@@ -183,6 +183,7 @@ class DummyFileIOStream : public CInterfaceOf<IFileIOStream>
     virtual offset_t size() override { return hwm; }
     virtual offset_t tell() override { return offset; }
     virtual unsigned __int64 getStatistic(StatisticKind kind) { return stats.getStatistic(kind); }
+    virtual void close() override { }
 private:
     offset_t offset = 0;
     offset_t hwm = 0;