Added memory/disk hybrid for transfering csv files.

Author: jgates108

deploy/compose/docker-compose.yml

@@ -36,6 +36,7 @@ volumes:
     volume_czar_xrootd:
     volume_czar_home:
     volume_czar_cfg:
+    volume_czar_transfer:
 
     volume_czar_mariadb_data:
     volume_czar_mariadb_cfg:
@@ -272,6 +273,10 @@ services:
             - type: volume
               source: volume_czar_mariadb_run
               target: /qserv/mariadb/run
+            - type: volume
+              source: volume_czar_transfer
+              target: /tmp
+              
             - << : *log-volume
         expose:
             - "3306" # for czar-mariadb
@@ -308,6 +313,9 @@ services:
             - type: volume
               source: volume_czar_cfg
               target: /config-etc
+            - type: volume
+              source: volume_czar_transfer
+              target: /tmp
             - type: volume
               source: volume_czar_home
               target: /home/qserv

src/admin/templates/proxy/etc/qserv-czar.cnf.jinja

@@ -25,12 +25,29 @@ port = {{ czar_db_port }}
 # Any table in resultdb that hasn't been updated in this many days is deleted.
 oldestResultKeptDays = 7
 
+# Either this should be changed to a high performance docker volume directory
+# or /tmp should be mounted as a high performance docker volume directory
+# to avoid using limited docker memory to store the contents.
+transferDir = /tmp
+
 # maximum number of connection retries to SQL databse (per connection attempt)
 maxsqlconnectionattempts = 10
 
 # maximum user query result size in MB
 maxtablesize_mb = 5100
 
+# maximum number of MB of concurrent csv transfer files allowed to be kept in 
+# memory, after this point the will be temporarily written to disk.  
+# 0 is used for testing. 10000 is usually reasonable.
+maxTransferMemMB = 0
+
+# minimum number of MB for each csv transfer file to be kept in memory
+# before possibly going to disk.
+# 0 for testing, up to 10 should be reasonable.
+transferMinMBInMem = 0
+
+
+
 
 # database connection for QMeta database
 [qmeta]

src/cconfig/CzarConfig.h

@@ -130,8 +130,16 @@ class CzarConfig {
     /// Getters for result aggregation options.
     int getMaxTableSizeMB() const { return _maxTableSizeMB->getVal(); }
     int getMaxSqlConnectionAttempts() const { return _maxSqlConnectionAttempts->getVal(); }
+    unsigned int getMaxTransferMemMB() const { return _resultMaxTransferMemMB->getVal(); }
+    /// Return the transfer directory. This is customizable to allow for a
+    /// high performance volume.
+    std::string getTransferDir() const { return _resultTransferDir->getVal(); }
 
-    /// The size of the TCP connection pool witin the client API that is used
+    /// Return the minimum amount of memory per UberJob to keep in memory. This much transfer
+    /// data will be stored in memory regardless of other conditions.
+    unsigned int getTransferMinMBInMem() const { return _resultTransferMinMBInMem->getVal(); }
+
+    /// The size of the TCP connection pool within the client API that is used
     /// by the merger to pool result files from workers via the HTTP protocol.
     int getResultMaxHttpConnections() const { return _resultMaxHttpConnections->getVal(); }
 
@@ -169,13 +177,6 @@ class CzarConfig {
     /// the method then the monitoring will be disabled.
     unsigned int czarStatsUpdateIvalSec() const { return _czarStatsUpdateIvalSec->getVal(); }
 
-    /// @return The maximum retain period for keeping in memory the relevant metrics
-    /// captured by the Czar monitoring system. If 0 is returned by the method then
-    /// query history archiving will be disabled.
-    /// @note Setting the limit too high may be potentially result in runing onto
-    /// the OOM situation.
-    unsigned int czarStatsRetainPeriodSec() const { return _czarStatsRetainPeriodSec->getVal(); }
-
     /// A worker is considered fully ALIVE if the last update from the worker has been
     /// heard in less than _activeWorkerTimeoutAliveSecs seconds.
     int getActiveWorkerTimeoutAliveSecs() const { return _activeWorkerTimeoutAliveSecs->getVal(); }
@@ -306,6 +307,14 @@ class CzarConfig {
     CVTIntPtr _oldestAsyncResultKeptSeconds = util::ConfigValTInt::create(
             _configValMap, "resultdb", "oldestAsyncResultKeptSeconds", notReq, 3600);
 
+    // This must be larger than _maxTableSizeMB when using the "memory" TransferMethod
+    CVTUIntPtr _resultMaxTransferMemMB =
+            util::ConfigValTUInt::create(_configValMap, "resultdb", "maxTransferMemMB", notReq, 10000);
+    CVTStrPtr _resultTransferDir =
+            util::ConfigValTStr::create(_configValMap, "resultdb", "transferDir", notReq, "/tmp");
+    CVTUIntPtr _resultTransferMinMBInMem =
+            util::ConfigValTUInt::create(_configValMap, "resultdb", "transferMinMBInMem", notReq, 10);
+
     /// Get all the elements in the css section.
     CVTStrPtr _cssTechnology =
             util::ConfigValTStr::create(_configValMap, "css", "technology", notReq, "mysql");

src/ccontrol/MergingHandler.cc

@@ -39,17 +39,17 @@
 #include "lsst/log/Log.h"
 
 // Qserv headers
+#include "cconfig/CzarConfig.h"
 #include "ccontrol/msgCode.h"
 #include "global/clock_defs.h"
 #include "global/debugUtil.h"
 #include "http/Client.h"
 #include "http/ClientConnPool.h"
 #include "http/Method.h"
-#include "mysql/CsvBuffer.h"
+#include "mysql/CsvMemDisk.h"
 #include "qdisp/CzarStats.h"
 #include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
-#include "qdisp/QueryRequest.h"
 #include "qdisp/UberJob.h"
 #include "rproc/InfileMerger.h"
 #include "util/Bug.h"
@@ -84,7 +84,6 @@ lsst::qserv::TimeCountTracker<double>::CALLBACKFUNC const reportFileRecvRate =
             }
         };
 
-
 string readHttpFileAndMerge(lsst::qserv::qdisp::UberJob::Ptr const& uberJob, string const& httpUrl,
                             size_t fileSize, function<void(char const*, uint32_t)> const& messageIsReady,
                             shared_ptr<http::ClientConnPool> const& httpConnPool) {
@@ -180,52 +179,6 @@ MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> const& merge
 
 MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_TRACE, __func__); }
 
-
-bool MergingHandler::flush(proto::ResponseSummary const& resp) {
-    _wName = resp.wname();
-
-    // This is needed to ensure the job query would be staying alive for the duration
-    // of the operation to prevent inconsistency within the application.
-    auto const jobQuery = getJobQuery().lock();
-    if (jobQuery == nullptr) {
-        LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, jobQuery was NULL");
-        return false;
-    }
-    auto const jobQuery = std::dynamic_pointer_cast<qdisp::JobQuery>(jobBase);
-
-    LOGS(_log, LOG_LVL_TRACE,
-         "MergingHandler::" << __func__ << " jobid=" << resp.jobid() << " transmitsize="
-                            << resp.transmitsize() << " rowcount=" << resp.rowcount() << " rowSize="
-                            << " attemptcount=" << resp.attemptcount() << " errorcode=" << resp.errorcode()
-                            << " errormsg=" << resp.errormsg());
-
-    if (resp.errorcode() != 0 || !resp.errormsg().empty()) {
-        _error = util::Error(resp.errorcode(), resp.errormsg(), util::ErrorCode::MYSQLEXEC);
-        _setError(ccontrol::MSG_RESULT_ERROR, _error.getMsg());
-        LOGS(_log, LOG_LVL_ERROR,
-             "MergingHandler::" << __func__ << " error from worker:" << resp.wname() << " error: " << _error);
-        // This way we can track if the worker has reported this error. The current implementation
-        // requires the large result size to be reported as an error via the InfileMerger regardless
-        // of an origin of the error (Czar or the worker). Note that large results can be produced
-        // by the Czar itself, e.g., when the aggregate result of multiple worker queries is too large
-        // or by the worker when the result set of a single query is too large.
-        // The error will be reported to the Czar as a part of the response summary.
-        if (resp.errorcode() == util::ErrorCode::WORKER_RESULT_TOO_LARGE) {
-            _infileMerger->setResultSizeLimitExceeded();
-        }
-        return false;
-    }
-
-    bool const success = _merge(resp, jobQuery);
-
-    if (success) {
-        _infileMerger->mergeCompleteFor(resp.jobid());
-        qdisp::CzarStats::get()->addTotalRowsRecv(resp.rowcount());
-        qdisp::CzarStats::get()->addTotalBytesRecv(resp.transmitsize());
-    }
-    return success;
-}
-
 void MergingHandler::errorFlush(std::string const& msg, int code) {
     _setError(code, msg, util::ErrorCode::RESULT_IMPORT);
     // Might want more info from result service.
@@ -244,13 +197,8 @@ qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uber
     }
 
     if (fileSize == 0) return qdisp::MergeEndStatus(true);
-
-    // Read from the http stream and push records into the CSV stream in a separate thread.
-    // Note the fixed capacity of the stream which allows up to 2 records to be buffered
-    // in the stream. This is enough to hide the latency of the HTTP connection and
-    // the time needed to read the file.
-    auto csvStream = mysql::CsvStream::create(2);
-    _csvStream = csvStream;
+    auto csvMemDisk = mysql::CsvMemDisk::create(fileSize, uberJob->getQueryId(), uberJob->getUjId());
+    _csvMemDisk = csvMemDisk;
 
     // This must be after setting _csvStream to avoid cancelFileMerge()
     // race issues, and it needs to be before the thread starts.
@@ -260,46 +208,46 @@ qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uber
     }
 
     string fileReadErrorMsg;
-    thread csvThread([uberJob, csvStream, fileUrl, fileSize, &fileReadErrorMsg]() {
+    auto transferFunc = [&]() {
         size_t bytesRead = 0;
         fileReadErrorMsg = ::readHttpFileAndMerge(
                 uberJob, fileUrl, fileSize,
-                [uberJob, csvStream, fileSize, &bytesRead](char const* buf, uint32_t size) {
+                [&](char const* buf, uint32_t size) {
                     bool last = false;
                     if (buf == nullptr || size == 0) {
                         last = true;
                     } else {
-                        csvStream->push(buf, size);
+                        csvMemDisk->push(buf, size);
                         bytesRead += size;
                         last = bytesRead >= fileSize;
                     }
                     if (last) {
-                        csvStream->push(nullptr, 0);
+                        csvMemDisk->push(nullptr, 0);
                     }
                 },
                 MergingHandler::_getHttpConnPool());
         // Push the stream terminator to indicate the end of the stream.
         // It may be needed to unblock the table merger which may be still attempting to read
         // from the CSV stream.
         if (!fileReadErrorMsg.empty()) {
-            csvStream->push(nullptr, 0);
+            csvMemDisk->push(nullptr, 0);
         }
-    });
+    };
+    csvMemDisk->transferDataFromWorker(transferFunc);
 
     // Attempt the actual merge.
-    bool fileMergeSuccess = _infileMerger->mergeHttp(uberJob, fileSize, csvStream);
+    bool fileMergeSuccess = _infileMerger->mergeHttp(uberJob, fileSize, csvMemDisk);
     if (!fileMergeSuccess) {
         LOGS(_log, LOG_LVL_WARN, __func__ << " merge failed");
         util::Error const& err = _infileMerger->getError();
         _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg(), util::ErrorCode::RESULT_IMPORT);
     }
-    if (csvStream->getContaminated()) {
+    if (csvMemDisk->getContaminated()) {
         LOGS(_log, LOG_LVL_ERROR, __func__ << " merge stream contaminated");
         fileMergeSuccess = false;
         _setError(ccontrol::MSG_RESULT_ERROR, "merge stream contaminated", util::ErrorCode::RESULT_IMPORT);
     }
 
-    csvThread.join();
     if (!fileReadErrorMsg.empty()) {
         LOGS(_log, LOG_LVL_WARN, __func__ << " result file read failed");
         _setError(ccontrol::MSG_HTTP_RESULT, fileReadErrorMsg, util::ErrorCode::RESULT_IMPORT);
@@ -310,14 +258,14 @@ qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uber
     if (!mergeEStatus.success) {
         // This error check needs to come after the csvThread.join() to ensure writing
         // is finished. If any bytes were written, the result table is ruined.
-        mergeEStatus.contaminated = csvStream->getBytesWritten() > 0;
+        mergeEStatus.contaminated = csvMemDisk->getBytesFetched() > 0;
     }
 
     return mergeEStatus;
 }
 
 void MergingHandler::cancelFileMerge() {
-    auto csvStrm = _csvStream.lock();
+    auto csvStrm = _csvMemDisk.lock();
     if (csvStrm != nullptr) {
         csvStrm->cancel();
     }
@@ -343,9 +291,6 @@ qdisp::MergeEndStatus MergingHandler::flushHttp(string const& fileUrl, uint64_t
          "MergingHandler::" << __func__ << " uberJob=" << uberJob->getIdStr() << " fileUrl=" << fileUrl);
 
     qdisp::MergeEndStatus mergeStatus = _mergeHttp(uberJob, fileUrl, fileSize);
-    if (mergeStatus.success) {
-        _infileMerger->mergeCompleteFor(uberJob->getUjId());
-    }
     return mergeStatus;
 }
 

src/ccontrol/MergingHandler.h

@@ -37,7 +37,7 @@ class ClientConnPool;
 }  // namespace lsst::qserv::http
 
 namespace lsst::qserv::mysql {
-class CsvStream;
+class CsvMemDisk;
 }  // namespace lsst::qserv::mysql
 
 namespace lsst::qserv::qdisp {
@@ -91,20 +91,21 @@ class MergingHandler : public qdisp::ResponseHandler {
     /// Set error code and string.
     void _setError(int code, std::string const& msg, int errorState);
 
-    /// Check if the query is no longer active.
-    /// This is used to prevent the query from being processed after it has been cancelled
-    /// or finished for any reason.
-    /// @param jobQuery the query to check
-    /// @return true if the query is no longer active
-    bool _queryIsNoLongerActive(std::shared_ptr<qdisp::JobQuery> const& jobQuery) const;
+    // All instances of the HTTP client class are members of the same pool. This allows
+    // connection reuse and a significant reduction of the kernel memory pressure.
+    // Note that the pool gets instantiated at the very first call to method _getHttpConnPool()
+    // because the instantiation depends on the availability of the Czar configuration.
+    static std::shared_ptr<http::ClientConnPool> const& _getHttpConnPool();
+    static std::shared_ptr<http::ClientConnPool> _httpConnPool;
+    static std::mutex _httpConnPoolMutex;
 
     std::shared_ptr<rproc::InfileMerger> _infileMerger;  ///< Merging delegate
     std::atomic<bool> _errorSet{false};                  ///< Set to true when an error is set.
     bool _flushed{false};                                ///< flushed to InfileMerger?
     std::string _wName{"~"};                             ///< worker name
 
-    std::weak_ptr<qdisp::Executive> _executive;  ///< Weak pointer to the executive for errors.
-    std::weak_ptr<mysql::CsvStream> _csvStream;  ///< Weak pointer to cancel infile merge.
+    std::weak_ptr<qdisp::Executive> _executive;    ///< Weak pointer to the executive for errors.
+    std::weak_ptr<mysql::CsvMemDisk> _csvMemDisk;  ///< Weak pointer to cancel infile merge.
 };
 
 }  // namespace lsst::qserv::ccontrol

src/ccontrol/UserQueryFactory.cc

@@ -135,7 +135,7 @@ std::shared_ptr<UserQuery> _makeUserQueryProcessList(query::SelectStmt::Ptr& stm
     LOGS(_log, LOG_LVL_DEBUG, "SELECT query is a PROCESSLIST");
     try {
         return std::make_shared<UserQueryProcessList>(stmt, sharedResources->qMetaSelect,
-                                                      sharedResources->qMetaCzarId, userQueryId, resultDb);
+                                                      sharedResources->czarId, userQueryId, resultDb);
     } catch (std::exception const& exc) {
         return std::make_shared<UserQueryInvalid>(exc.what());
     }
@@ -297,6 +297,7 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         }
         auto stmt = parser->getSelectStmt();
 
+        std::lock_guard focatoryLock(_factoryMtx);
         // handle special database/table names
         if (_stmtRefersToProcessListTable(stmt, defaultDb)) {
             return _makeUserQueryProcessList(stmt, _userQuerySharedResources, userQueryId, resultDb, aQuery,
@@ -331,7 +332,6 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         }
 
         // This is a regular SELECT for qserv
-
         // Currently using the database for results to get schema information.
         auto qs = std::make_shared<qproc::QuerySession>(_userQuerySharedResources->css,
                                                         _userQuerySharedResources->databaseModels, defaultDb,
@@ -372,7 +372,8 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
                 qs, messageStore, executive, _userQuerySharedResources->databaseModels, infileMergerConfig,
                 _userQuerySharedResources->secondaryIndex, _userQuerySharedResources->queryMetadata,
                 _userQuerySharedResources->queryProgress, _userQuerySharedResources->czarId, errorExtra,
-                async, resultDb);
+                async, resultDb, uberJobMaxChunks);
+
         if (sessionValid) {
             uq->qMetaRegister(resultLocation, msgTableName);
             uq->setupMerger();
@@ -381,11 +382,13 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         }
         return uq;
     } else if (UserQueryType::isSelectResult(query, userJobId)) {
+        std::lock_guard factoryLock(_factoryMtx);
         auto uq = std::make_shared<UserQueryAsyncResult>(userJobId, _userQuerySharedResources->czarId,
                                                          _userQuerySharedResources->queryMetadata);
         LOGS(_log, LOG_LVL_DEBUG, "make UserQueryAsyncResult: userJobId=" << userJobId);
         return uq;
     } else if (UserQueryType::isShowProcessList(query, full)) {
+        std::lock_guard factoryLock(_factoryMtx);
         LOGS(_log, LOG_LVL_DEBUG, "make UserQueryProcessList: full=" << (full ? 'y' : 'n'));
         try {
             return std::make_shared<UserQueryProcessList>(full, _userQuerySharedResources->qMetaSelect,
@@ -395,6 +398,7 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
             return std::make_shared<UserQueryInvalid>(exc.what());
         }
     } else if (UserQueryType::isCall(query)) {
+        std::lock_guard factoryLock(_factoryMtx);
         auto parser = std::make_shared<ParseRunner>(
                 query, _userQuerySharedResources->makeUserQueryResources(userQueryId, resultDb));
         return parser->getUserQuery();
@@ -406,6 +410,7 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
             return std::make_shared<UserQueryInvalid>(std::string("ParseException:") + e.what());
         }
         auto uq = parser->getUserQuery();
+        std::lock_guard factoryLock(_factoryMtx);
         auto setQuery = std::static_pointer_cast<UserQuerySet>(uq);
         if (setQuery->varName() == "QSERV_ROW_COUNTER_OPTIMIZATION") {
             _useQservRowCounterOptimization = setQuery->varValue() != "0";
@@ -417,6 +422,7 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         }
         return uq;
     } else {
+        std::lock_guard factoryLock(_factoryMtx);
         // something that we don't recognize
         auto uq = std::make_shared<UserQueryInvalid>("Invalid or unsupported query: " + query);
         return uq;