[ML] Clearing JSON memory allocators (#30)

Use scoped JSON memory allocators to avoid excessive memory accumulation
where-ever the JSON output writer is persisted.

Fixes #26

Author: edsavage

include/core/CRapidJsonWriterBase.h

@@ -71,28 +71,40 @@ template<typename OUTPUT_STREAM,
          template<typename, typename, typename, typename, unsigned> class JSON_WRITER = rapidjson::Writer>
 class CRapidJsonWriterBase : public JSON_WRITER<OUTPUT_STREAM, SOURCE_ENCODING, TARGET_ENCODING, STACK_ALLOCATOR, WRITE_FLAGS> {
 public:
-    using TTimeVec = std::vector<core_t::TTime>;
-    using TStrVec = std::vector<std::string>;
-    using TDoubleVec = std::vector<double>;
-    using TDoubleDoublePr = std::pair<double, double>;
-    using TDoubleDoublePrVec = std::vector<TDoubleDoublePr>;
-    using TDoubleDoubleDoublePrPr = std::pair<double, TDoubleDoublePr>;
-    using TDoubleDoubleDoublePrPrVec = std::vector<TDoubleDoubleDoublePrPr>;
-    using TStrUSet = boost::unordered_set<std::string>;
-    using TDocument = rapidjson::Document;
-    using TValue = rapidjson::Value;
-    using TDocumentWeakPtr = boost::weak_ptr<TDocument>;
-    using TValuePtr = boost::shared_ptr<TValue>;
-
-    using TPoolAllocatorPtr = boost::shared_ptr<CRapidJsonPoolAllocator>;
-    using TPoolAllocatorPtrStack = std::stack<TPoolAllocatorPtr>;
-    using TStrPoolAllocatorPtrMap = boost::unordered_map<std::string, TPoolAllocatorPtr>;
-    using TStrPoolAllocatorPtrMapItr = TStrPoolAllocatorPtrMap::iterator;
-    using TStrPoolAllocatorPtrMapItrBoolPr = std::pair<TStrPoolAllocatorPtrMapItr, bool>;
+    typedef std::vector<core_t::TTime> TTimeVec;
+    typedef std::vector<std::string> TStrVec;
+    typedef std::vector<double> TDoubleVec;
+    typedef std::pair<double, double> TDoubleDoublePr;
+    typedef std::vector<TDoubleDoublePr> TDoubleDoublePrVec;
+    typedef std::pair<double, TDoubleDoublePr> TDoubleDoubleDoublePrPr;
+    typedef std::vector<TDoubleDoubleDoublePrPr> TDoubleDoubleDoublePrPrVec;
+    typedef boost::unordered_set<std::string> TStrUSet;
+    typedef rapidjson::Document TDocument;
+    typedef rapidjson::Value TValue;
+    typedef boost::weak_ptr<TDocument> TDocumentWeakPtr;
+    typedef boost::shared_ptr<TValue> TValuePtr;
+
+    typedef boost::shared_ptr<CRapidJsonPoolAllocator> TPoolAllocatorPtr;
+    typedef std::stack<TPoolAllocatorPtr> TPoolAllocatorPtrStack;
+    typedef boost::unordered_map<std::string, TPoolAllocatorPtr> TStrPoolAllocatorPtrMap;
+    typedef TStrPoolAllocatorPtrMap::iterator TStrPoolAllocatorPtrMapItr;
+    typedef std::pair<TStrPoolAllocatorPtrMapItr, bool> TStrPoolAllocatorPtrMapItrBoolPr;
 
 public:
     using TRapidJsonWriterBase = JSON_WRITER<OUTPUT_STREAM, SOURCE_ENCODING, TARGET_ENCODING, STACK_ALLOCATOR, WRITE_FLAGS>;
 
+    //! Instances of this class may very well be long lived, potentially for the lifetime of the application.
+    //! Over the course of that lifetime resources will accumulate in the underlying rapidjson memory
+    //! allocator. To prevent excessive memory expansion these resources will need to be cleaned regularly.
+    //!
+    //! In preference to clients of this class explicitly clearing the allocator a helper/wrapper class -
+    //! \p CScopedRapidJsonPoolAllocator - is provided. This helper has an RAII style interface that clears the
+    //! allocator when it goes out of scope which requires that the writer provides the push/popAllocator
+    //! functions.  The intent of this approach is to make it possible to use one or two separate allocators
+    //! for the writer at nested scope.
+    //!
+    //! Note that allocators are not destroyed by the pop operation, they persist for the lifetime of the
+    //! writer in a cache for swift retrieval.
     CRapidJsonWriterBase(OUTPUT_STREAM& os) : TRapidJsonWriterBase(os) {
         // push a default rapidjson allocator onto our stack
         m_JsonPoolAllocators.push(boost::make_shared<CRapidJsonPoolAllocator>());
@@ -105,10 +117,9 @@ class CRapidJsonWriterBase : public JSON_WRITER<OUTPUT_STREAM, SOURCE_ENCODING,
 
     CRapidJsonWriterBase(CRapidJsonWriterBase&& rhs) : TRapidJsonWriterBase(std::move(rhs)) {}
 
-    virtual ~CRapidJsonWriterBase() {
-        // clean up resources
-        m_JsonPoolAllocators.pop();
-    }
+    // No need for an explicit destructor here as the allocators clear themselves
+    // on destruction.
+    virtual ~CRapidJsonWriterBase() = default;
 
     //! Push a named allocator on to the stack
     //! Look in the cache for the allocator - creating it if not present

lib/api/CJsonOutputWriter.cc

@@ -15,6 +15,7 @@
 
 #include <api/CJsonOutputWriter.h>
 
+#include <core/CScopedRapidJsonPoolAllocator.h>
 #include <core/CStringUtils.h>
 #include <core/CTimeUtils.h>
 
@@ -345,6 +346,10 @@ const CJsonOutputWriter::TStrVec& CJsonOutputWriter::fieldNames() const {
 }
 
 bool CJsonOutputWriter::writeRow(const TStrStrUMap& dataRowFields, const TStrStrUMap& overrideDataRowFields) {
+    using TScopedAllocator = core::CScopedRapidJsonPoolAllocator<core::CRapidJsonConcurrentLineWriter>;
+
+    TScopedAllocator scopedAllocator("CJsonOutputWriter::writeRow", m_Writer);
+
     rapidjson::Document doc = m_Writer.makeDoc();
 
     // Write all the fields to the document as strings

lib/api/CLineifiedJsonOutputWriter.cc

@@ -14,6 +14,7 @@
  */
 #include <api/CLineifiedJsonOutputWriter.h>
 
+#include <core/CScopedRapidJsonPoolAllocator.h>
 #include <core/CSleep.h>
 #include <core/CStringUtils.h>
 
@@ -58,6 +59,9 @@ const CLineifiedJsonOutputWriter::TStrVec& CLineifiedJsonOutputWriter::fieldName
 }
 
 bool CLineifiedJsonOutputWriter::writeRow(const TStrStrUMap& dataRowFields, const TStrStrUMap& overrideDataRowFields) {
+    using TScopedAllocator = core::CScopedRapidJsonPoolAllocator<TGenericLineWriter>;
+    TScopedAllocator scopedAllocator("CLineifiedJsonOutputWriter::writeRow", m_Writer);
+
     rapidjson::Document doc = m_Writer.makeDoc();
 
     // Write all the fields to the document as strings

lib/model/CForecastDataSink.cc

@@ -16,6 +16,7 @@
 #include <model/CForecastDataSink.h>
 
 #include <core/CLogger.h>
+#include <core/CScopedRapidJsonPoolAllocator.h>
 
 #include <vector>
 
@@ -60,6 +61,8 @@ const std::string CForecastDataSink::PROCESSING_TIME_MS("processing_time_ms");
 const std::string CForecastDataSink::PROGRESS("forecast_progress");
 const std::string CForecastDataSink::STATUS("forecast_status");
 
+using TScopedAllocator = core::CScopedRapidJsonPoolAllocator<core::CRapidJsonConcurrentLineWriter>;
+
 CForecastDataSink::SForecastModelWrapper::SForecastModelWrapper(model_t::EFeature feature,
                                                                 TMathsModelPtr&& forecastModel,
                                                                 const std::string& byFieldValue)
@@ -102,6 +105,8 @@ CForecastDataSink::CForecastDataSink(const std::string& jobId,
 }
 
 void CForecastDataSink::writeStats(const double progress, uint64_t runtime, const TStrUMap& messages, bool successful) {
+    TScopedAllocator scopedAllocator("CForecastDataSink", m_Writer);
+
     rapidjson::Document doc = m_Writer.makeDoc();
 
     this->writeCommonStatsFields(doc);
@@ -134,6 +139,8 @@ void CForecastDataSink::writeScheduledMessage() {
 }
 
 void CForecastDataSink::writeErrorMessage(const std::string& message) {
+    TScopedAllocator scopedAllocator("CForecastDataSink", m_Writer);
+
     rapidjson::Document doc = m_Writer.makeDoc();
     this->writeCommonStatsFields(doc);
     TStrVec messages{message};
@@ -143,6 +150,8 @@ void CForecastDataSink::writeErrorMessage(const std::string& message) {
 }
 
 void CForecastDataSink::writeFinalMessage(const std::string& message) {
+    TScopedAllocator scopedAllocator("CForecastDataSink", m_Writer);
+
     rapidjson::Document doc = m_Writer.makeDoc();
     this->writeCommonStatsFields(doc);
     TStrVec messages{message};
@@ -165,6 +174,8 @@ void CForecastDataSink::writeCommonStatsFields(rapidjson::Value& doc) {
 }
 
 void CForecastDataSink::push(bool flush, rapidjson::Value& doc) {
+    TScopedAllocator scopedAllocator("CForecastDataSink", m_Writer);
+
     rapidjson::Document wrapper = m_Writer.makeDoc();
 
     m_Writer.addMember(MODEL_FORECAST_STATS, doc, wrapper);
@@ -186,6 +197,8 @@ void CForecastDataSink::push(const maths::SErrorBar errorBar,
                              const std::string& byFieldName,
                              const std::string& byFieldValue,
                              int detectorIndex) {
+    TScopedAllocator scopedAllocator("CForecastDataSink", m_Writer);
+
     ++m_NumRecordsWritten;
     rapidjson::Document doc = m_Writer.makeDoc();