[FEATURE][ML] Shallow copy memory mapped vector and matrix types (#340)

Author: tveasey

include/core/CDataFrame.h

@@ -320,6 +320,9 @@ class CORE_EXPORT CDataFrame final {
     //! Get the memory used by the data frame.
     std::size_t memoryUsage() const;
 
+    //! Get a checksum of all the data stored in the data frame.
+    std::uint64_t checksum() const;
+
     // TODO Better error case diagnostics.
 
     // TODO We may want an architecture agnostic check pointing mechanism for long

include/core/CDataFrameRowSlice.h

@@ -75,6 +75,7 @@ class CORE_EXPORT CDataFrameRowSlice {
     virtual void write(const TFloatVec& values) = 0;
     virtual std::size_t staticSize() const = 0;
     virtual std::size_t memoryUsage() const = 0;
+    virtual std::uint64_t checksum() const = 0;
 };
 
 //! \brief In main memory CDataFrame slice storage.
@@ -95,6 +96,7 @@ class CORE_EXPORT CMainMemoryDataFrameRowSlice final : public CDataFrameRowSlice
     virtual void write(const TFloatVec& values);
     virtual std::size_t staticSize() const;
     virtual std::size_t memoryUsage() const;
+    virtual std::uint64_t checksum() const;
 
 private:
     std::size_t m_FirstRow;
@@ -155,6 +157,7 @@ class CORE_EXPORT COnDiskDataFrameRowSlice final : public CDataFrameRowSlice {
     virtual void write(const TFloatVec& values);
     virtual std::size_t staticSize() const;
     virtual std::size_t memoryUsage() const;
+    virtual std::uint64_t checksum() const;
 
 private:
     void writeToDisk(const TFloatVec& state);

include/maths/CLinearAlgebraEigen.h

@@ -281,7 +281,16 @@ struct SConstant<CDenseVector<SCALAR>> {
     }
 };
 
-//! \brief Decorates an Eigen::Map of a dense matrix with some useful methods.
+//! \brief Decorates an Eigen::Map of a dense matrix with some useful methods
+//! and changes default copy semantics to shallow copy.
+//!
+//! IMPLEMENTATION:\n
+//! This effectively acts like a std::reference_wrapper of an Eigen::Map for
+//! an Eigen matrix. In particular, all copying is shallow unlike Eigen::Map
+//! that acts directly on the referenced memory. This is to match the behaviour
+//! of CMemoryMappedDenseVector.
+//!
+//! \sa CMemoryMappedDenseVector for more information.
 template<typename SCALAR>
 class CMemoryMappedDenseMatrix
     : public Eigen::Map<typename CDenseMatrix<SCALAR>::TBase> {
@@ -302,11 +311,32 @@ class CMemoryMappedDenseMatrix
     CMemoryMappedDenseMatrix(CMemoryMappedDenseMatrix& other)
         : CMemoryMappedDenseMatrix{static_cast<const CMemoryMappedDenseMatrix&>(other)} {}
     CMemoryMappedDenseMatrix(const CMemoryMappedDenseMatrix& other)
-        : TBase{static_cast<const TBase&>(other)} {}
-    CMemoryMappedDenseMatrix(CMemoryMappedDenseMatrix&& other) = default;
-    CMemoryMappedDenseMatrix& operator=(const CMemoryMappedDenseMatrix& other) = default;
-    CMemoryMappedDenseMatrix& operator=(CMemoryMappedDenseMatrix&& other) = default;
+        : TBase{nullptr, 1, 1} {
+        this->reseat(other);
+    }
+    CMemoryMappedDenseMatrix(CMemoryMappedDenseMatrix&& other)
+        : TBase{nullptr, 1, 1} {
+        this->reseat(other);
+    }
+    CMemoryMappedDenseMatrix& operator=(const CMemoryMappedDenseMatrix& other) {
+        if (this != &other) {
+            this->reseat(other);
+        }
+        return *this;
+    }
+    CMemoryMappedDenseMatrix& operator=(CMemoryMappedDenseMatrix&& other) {
+        if (this != &other) {
+            this->reseat(other);
+        }
+        return *this;
+    }
     //@}
+
+private:
+    void reseat(const CMemoryMappedDenseMatrix& other) {
+        TBase* base{static_cast<TBase*>(this)};
+        new (base) TBase{const_cast<SCALAR*>(other.data()), other.rows(), other.cols()};
+    }
 };
 
 //! \brief Gets a constant square dense matrix with specified dimension or with
@@ -323,7 +353,37 @@ struct SConstant<CMemoryMappedDenseMatrix<SCALAR>> {
     }
 };
 
-//! \brief Decorates an Eigen::Map of a dense vector with some useful methods.
+//! \brief Decorates an Eigen::Map of a dense vector with some useful methods
+//! and changes default copy semantics to shallow.
+//!
+//! IMPLEMENTATION:\n
+//! This effectively acts like a std::reference_wrapper of an Eigen::Map for
+//! an Eigen vector. In particular, all copying is shallow unlike Eigen::Map
+//! that acts directly on the referenced memory, i.e.
+//! \code{.cpp}
+//! double values1[]{1.0, 1.0};
+//! double values2[]{2.0, 2.0};
+//!
+//! CMemoryMappedDenseVector<double> mm1{values1, 2};
+//! CMemoryMappedDenseVector<double> mm2{values2, 2};
+//!
+//! mm1 = mm2;
+//! std::cout << mm1(0) << "," << mm1(1) << "," << values1[0] << "," << values1[1] << std::endl;
+//!
+//! Eigen::Map<Eigen::VectorXd> map1{values1, 2};
+//! Eigen::Map<Eigen::VectorXd> map2{values2, 2};
+//!
+//! map1 = map2;
+//! std::cout << map1(0) << "," << map1(1) << "," << values1[0] << "," << values1[1] << std::endl;
+//! \endcode
+//!
+//! Outputs:\n
+//! 2,2,1,1\n
+//! 2,2,2,2
+//!
+//! This better fits our needs with data frames where we want to reference the
+//! memory stored in the data frame rows, but never modify it directly through
+//! this vector type.
 template<typename SCALAR>
 class CMemoryMappedDenseVector
     : public Eigen::Map<typename CDenseVector<SCALAR>::TBase> {
@@ -337,17 +397,32 @@ class CMemoryMappedDenseVector
     //! Forwarding constructor.
     template<typename... ARGS>
     CMemoryMappedDenseVector(ARGS&&... args)
-        : TBase(std::forward<ARGS>(args)...) {}
+        : TBase{std::forward<ARGS>(args)...} {}
 
     //! \name Copy and Move Semantics
     //@{
     CMemoryMappedDenseVector(CMemoryMappedDenseVector& other)
         : CMemoryMappedDenseVector{static_cast<const CMemoryMappedDenseVector&>(other)} {}
     CMemoryMappedDenseVector(const CMemoryMappedDenseVector& other)
-        : TBase{static_cast<const TBase&>(other)} {}
-    CMemoryMappedDenseVector(CMemoryMappedDenseVector&& other) = default;
-    CMemoryMappedDenseVector& operator=(const CMemoryMappedDenseVector& other) = default;
-    CMemoryMappedDenseVector& operator=(CMemoryMappedDenseVector&& other) = default;
+        : TBase{nullptr, 1} {
+        this->reseat(other);
+    }
+    CMemoryMappedDenseVector(CMemoryMappedDenseVector&& other)
+        : TBase{nullptr, 1} {
+        this->reseat(other);
+    }
+    CMemoryMappedDenseVector& operator=(const CMemoryMappedDenseVector& other) {
+        if (this != &other) {
+            this->reseat(other);
+        }
+        return *this;
+    }
+    CMemoryMappedDenseVector& operator=(CMemoryMappedDenseVector&& other) {
+        if (this != &other) {
+            this->reseat(other);
+        }
+        return *this;
+    }
     //@}
 
     //! Get a checksum of this object.
@@ -357,6 +432,12 @@ class CMemoryMappedDenseVector
         }
         return seed;
     }
+
+private:
+    void reseat(const CMemoryMappedDenseVector& other) {
+        TBase* base{static_cast<TBase*>(this)};
+        new (base) TBase{const_cast<SCALAR*>(other.data()), other.size()};
+    }
 };
 
 //! \brief Gets a constant dense vector with specified dimension.

lib/api/CDataFrameAnalyzer.cc

@@ -158,7 +158,8 @@ bool CDataFrameAnalyzer::handleControlMessage(const TStrVec& fieldValues) {
     case ' ':
         // Spaces are just used to fill the buffers and force prior messages
         // through the system - we don't need to do anything else.
-        LOG_TRACE(<< "Received pad of length " << controlMessage.length());
+        LOG_TRACE(<< "Received pad of length "
+                  << fieldValues[m_ControlFieldIndex].length());
         return true;
     case FINISHED_DATA_CONTROL_MESSAGE_FIELD_VALUE:
         this->receivedAllRows();

lib/core/CDataFrame.cc

@@ -8,6 +8,7 @@
 
 #include <core/CConcurrentWrapper.h>
 #include <core/CDataFrameRowSlice.h>
+#include <core/CHashing.h>
 #include <core/CLogger.h>
 #include <core/CMemory.h>
 #include <core/Concurrency.h>
@@ -250,6 +251,19 @@ std::size_t CDataFrame::memoryUsage() const {
     return CMemory::dynamicSize(m_Slices) + CMemory::dynamicSize(m_Writer);
 }
 
+std::uint64_t CDataFrame::checksum() const {
+    std::vector<std::uint64_t> checksums(m_Slices.size(), 0);
+    parallel_for_each(0, m_Slices.size(), [&](std::size_t index) {
+        checksums[index] = m_Slices[index]->checksum();
+    });
+
+    std::uint64_t result{0};
+    for (auto checksum : checksums) {
+        result = CHashing::hashCombine(result, checksum);
+    }
+    return result;
+}
+
 CDataFrame::CDataFrameRowSliceWriter::CDataFrameRowSliceWriter(
     std::size_t numberRows,
     std::size_t rowCapacity,

lib/core/CDataFrameRowSlice.cc

@@ -86,6 +86,12 @@ class CBadDataFrameRowSliceHandle : public CDataFrameRowSliceHandleImpl {
     //! Stub for the values.
     mutable TFloatVec m_Empty;
 };
+
+//! Checksum \p slice.
+uint64_t computeChecksum(const TFloatVec& slice) {
+    return CHashing::murmurHash64(
+        slice.data(), static_cast<int>(sizeof(CFloatStorage) * slice.size()), 0);
+}
 }
 
 //////// CDataFrameRowSliceHandle ////////
@@ -181,6 +187,10 @@ std::size_t CMainMemoryDataFrameRowSlice::memoryUsage() const {
     return CMemory::dynamicSize(m_State);
 }
 
+std::uint64_t CMainMemoryDataFrameRowSlice::checksum() const {
+    return computeChecksum(m_State);
+}
+
 //////// COnDiskDataFrameRowSlice ////////
 
 namespace {
@@ -201,12 +211,6 @@ bool sufficientDiskSpaceAvailable(const boost::filesystem::path& path, std::size
     }
     return true;
 }
-
-//! Checksum \p slice.
-uint64_t checksum(const TFloatVec& slice) {
-    return CHashing::murmurHash64(
-        slice.data(), static_cast<int>(sizeof(CFloatStorage) * slice.size()), 0);
-}
 }
 
 COnDiskDataFrameRowSlice::COnDiskDataFrameRowSlice(const TTemporaryDirectoryPtr& directory,
@@ -282,7 +286,7 @@ COnDiskDataFrameRowSlice::TSizeHandlePr COnDiskDataFrameRowSlice::read() {
             return {0, {boost::make_unique<CBadDataFrameRowSliceHandle>()}};
         }
 
-        if (checksum(result) != m_Checksum) {
+        if (computeChecksum(result) != m_Checksum) {
             LOG_ERROR(<< "Corrupt from row " << m_FirstRow);
             m_StateIsBad = true;
             return {0, {boost::make_unique<CBadDataFrameRowSliceHandle>()}};
@@ -313,7 +317,7 @@ std::size_t COnDiskDataFrameRowSlice::memoryUsage() const {
 
 void COnDiskDataFrameRowSlice::writeToDisk(const TFloatVec& state) {
     m_Capacity = state.size();
-    m_Checksum = checksum(state);
+    m_Checksum = computeChecksum(state);
     LOG_TRACE(<< "Checksum = " << m_Checksum);
 
     std::size_t bytes{sizeof(CFloatStorage) * state.size()};
@@ -323,6 +327,10 @@ void COnDiskDataFrameRowSlice::writeToDisk(const TFloatVec& state) {
     file.write(reinterpret_cast<const char*>(state.data()), bytes);
 }
 
+std::uint64_t COnDiskDataFrameRowSlice::checksum() const {
+    return m_Checksum;
+}
+
 bool COnDiskDataFrameRowSlice::readFromDisk(TFloatVec& result) const {
     result.resize(m_Capacity);
 

lib/maths/CLocalOutlierFactors.cc

@@ -79,10 +79,13 @@ bool computeOutliersNoPartitions(std::size_t numberThreads, core::CDataFrame& fr
 
     auto rowsToPoints = [&points](TRowItr beginRows, TRowItr endRows) {
         for (auto row = beginRows; row != endRows; ++row) {
-            new (&points[row->index()]) TVector{row->data(), row->numberColumns()};
+            points[row->index()] =
+                TVector{row->data(), static_cast<long>(row->numberColumns())};
         }
     };
 
+    std::uint64_t checksum{frame.checksum()};
+
     bool successful;
     std::tie(std::ignore, successful) = frame.readRows(numberThreads, rowsToPoints);
 
@@ -94,6 +97,15 @@ bool computeOutliersNoPartitions(std::size_t numberThreads, core::CDataFrame& fr
     TDoubleVec scores;
     CLocalOutlierFactors::ensemble(std::move(points), scores);
 
+    // This never happens now, but it is a sanity check against someone
+    // changing CLocalOutlierFactors to accidentally write to the data
+    // frame via one of the memory mapped vectors. All bets are off as to
+    // if we generate anything meaningful if this happens.
+    if (checksum != frame.checksum()) {
+        LOG_ERROR(<< "Accidentally modified the data frame");
+        return false;
+    }
+
     auto writeScores = [&scores](TRowItr beginRows, TRowItr endRows) {
         for (auto row = beginRows; row != endRows; ++row) {
             row->writeColumn(row->numberColumns() - 1, scores[row->index()]);