Clean up obsolete configuration parameters of the Qserv worker

Author: iagaponenko

src/admin/templates/xrootd/etc/xrdssi.cf.jinja

@@ -66,13 +66,6 @@ maxsqlconn = 980
 # This value must be less than maxsqlconn. 
 reservedinteractivesqlconn = 930
 
-[transmits]
-# Maximum number of concurrent transmits to a czar.
-maxtransmits = 50
-# If more than this number of large transmits is happening at once, wait to
-# start more transmits until some are done.
-maxalreadytransmitting = 10
-
 [results]
 
 # The name of a folder where query results will be stored.

src/wconfig/WorkerConfig.h

@@ -150,14 +150,6 @@ class WorkerConfig {
         return _ReservedInteractiveSqlConnections->getVal();
     }
 
-    /// @return the maximum number of gigabytes that can be used by StreamBuffers
-    unsigned int getBufferMaxTotalGB() const { return _bufferMaxTotalGB->getVal(); }
-
-    /// @return the maximum number of concurrent transmits to a czar
-    unsigned int getMaxTransmits() const { return _maxTransmits->getVal(); }
-
-    int getMaxPerQid() const { return _maxPerQid->getVal(); }
-
     /// @return the name of a folder where query results will be stored
     std::string const resultsDirname() const { return _resultsDirname->getVal(); }
 
@@ -292,11 +284,6 @@ class WorkerConfig {
             util::ConfigValTUInt::create(_configValMap, "sqlconnections", "maxsqlconn", notReq, 800);
     CVTUIntPtr _ReservedInteractiveSqlConnections = util::ConfigValTUInt::create(
             _configValMap, "sqlconnections", "reservedinteractivesqlconn", notReq, 50);
-    CVTUIntPtr _bufferMaxTotalGB =
-            util::ConfigValTUInt::create(_configValMap, "transmit", "buffermaxtotalgb", notReq, 41);
-    CVTUIntPtr _maxTransmits =
-            util::ConfigValTUInt::create(_configValMap, "transmit", "maxtransmits", notReq, 40);
-    CVTIntPtr _maxPerQid = util::ConfigValTInt::create(_configValMap, "transmit", "maxperqid", notReq, 3);
     CVTStrPtr _resultsDirname =
             util::ConfigValTStr::create(_configValMap, "results", "dirname", notReq, "/qserv/data/results");
     CVTUIntPtr _resultsXrootdPort =

src/xrdsvc/SsiService.cc

@@ -156,8 +156,6 @@ SsiService::SsiService(XrdSsiLogger* log) {
         throw wconfig::WorkerConfigError("Unable to connect to MySQL");
     }
     auto const workerConfig = wconfig::WorkerConfig::instance();
-    int64_t bufferMaxTotalBytes = workerConfig->getBufferMaxTotalGB() * 1'000'000'000LL;
-    StreamBuffer::setMaxTotalBytes(bufferMaxTotalBytes);
 
     // Set thread pool size.
     unsigned int poolSize = ranges::max({wsched::BlendScheduler::getMinPoolSize(),

src/xrdsvc/StreamBuffer.cc

@@ -42,37 +42,8 @@ using namespace std;
 
 namespace lsst::qserv::xrdsvc {
 
-atomic<int64_t> StreamBuffer::_maxTotalBytes{40'000'000'000};
-atomic<int64_t> StreamBuffer::_totalBytes(0);
-mutex StreamBuffer::_createMtx;
-condition_variable StreamBuffer::_createCv;
-
-void StreamBuffer::setMaxTotalBytes(int64_t maxBytes) {
-    string const context = "StreamBuffer::" + string(__func__) + " ";
-    LOGS(_log, LOG_LVL_INFO, context << "maxBytes=" << maxBytes);
-    if (maxBytes < 0) {
-        throw invalid_argument(context + "negative " + to_string(maxBytes));
-    }
-    if (maxBytes < 1'000'000'000LL) {
-        LOGS(_log, LOG_LVL_ERROR, "Very small value for " << context << maxBytes);
-    }
-    _maxTotalBytes = maxBytes;
-}
-
-double StreamBuffer::percentOfMaxTotalBytesUsed() {
-    double percent = ((double)_totalBytes) / ((double)_maxTotalBytes);
-    if (percent < 0.0) percent = 0.0;
-    if (percent > 1.0) percent = 1.0;
-    return percent;
-}
-
 // Factory function, because this should be able to delete itself when Recycle() is called.
 StreamBuffer::Ptr StreamBuffer::createWithMove(std::string &input, std::shared_ptr<wbase::Task> const &task) {
-    unique_lock<mutex> uLock(_createMtx);
-    if (_totalBytes >= _maxTotalBytes) {
-        LOGS(_log, LOG_LVL_WARN, "StreamBuffer at memory limit " << _totalBytes);
-    }
-    _createCv.wait(uLock, []() { return _totalBytes < _maxTotalBytes; });
     Ptr ptr(new StreamBuffer(input, task));
     ptr->_selfKeepAlive = ptr;
     return ptr;
@@ -95,14 +66,6 @@ StreamBuffer::StreamBuffer(std::string &input, wbase::Task::Ptr const &task) : _
     if (_wStats != nullptr) {
         _wStats->startQueryRespConcurrentQueued(_createdTime);
     }
-
-    _totalBytes += _dataStr.size();
-    LOGS(_log, LOG_LVL_DEBUG, "StreamBuffer::_totalBytes=" << _totalBytes << " thisSize=" << _dataStr.size());
-}
-
-StreamBuffer::~StreamBuffer() {
-    _totalBytes -= _dataStr.size();
-    LOGS(_log, LOG_LVL_DEBUG, "~StreamBuffer::_totalBytes=" << _totalBytes);
 }
 
 void StreamBuffer::startTimer() {

src/xrdsvc/StreamBuffer.h

@@ -66,17 +66,8 @@ class StreamBuffer : public XrdSsiStream::Buffer {
     static StreamBuffer::Ptr createWithMove(std::string &input,
                                             std::shared_ptr<wbase::Task> const &task = nullptr);
 
-    /// Set the maximum number of bytes that can be used by all instances of this class.
-    static void setMaxTotalBytes(int64_t maxBytes);
-
-    /// @return the percent of totalBytes used out of _maxTotalByes.
-    static double percentOfMaxTotalBytesUsed();
-
     size_t getSize() const { return _dataStr.size(); }
 
-    /// @Return total number of bytes used by ALL StreamBuffer objects.
-    static size_t getTotalBytes() { return _totalBytes; }
-
     /// Call to recycle the buffer when finished (normally called by XrdSsi).
     void Recycle() override;
 
@@ -90,7 +81,7 @@ class StreamBuffer : public XrdSsiStream::Buffer {
     /// Unblock the condition variable on cancel.
     void cancel();
 
-    ~StreamBuffer() override;
+    ~StreamBuffer() override = default;
 
 private:
     /// This constructor will invalidate 'input'.
@@ -115,12 +106,6 @@ class StreamBuffer : public XrdSsiStream::Buffer {
     /// Pointer for worker statistics.
     /// NOTE: This will be nullptr for many things, so check before using.
     std::shared_ptr<wcontrol::WorkerStats> _wStats;
-
-    // Members associated with limiting memory use.
-    static std::atomic<int64_t> _totalBytes;  ///< Total bytes currently in use by all StreamBuffer instances.
-    static std::atomic<int64_t> _maxTotalBytes;
-    static std::mutex _createMtx;
-    static std::condition_variable _createCv;
 };
 
 }  // namespace lsst::qserv::xrdsvc