ciiOldbDataPointAsync.ipp

Go to the documentation of this file.

 
#include <chrono>
#include "hlcc/utils/ciiTypesToString/ciiTypesToString.hpp"
 
 
namespace hlcc::oldbmux {
 
//--------------------------------------------------------------+
//                      operator<<                              |
//                                                              |
// The buffer DiscardListener currently logs the dropped value. |
// For non-primitive types, we need a stream insertion operator |
// matching the template type T.                                |
//                                                              |
// Is it OK to "pollute" namespace hlcc::oldbmux with these     |
// operator definitions? Is there a better solution?            |
// Perhaps to somehow recognize a vector type in the discard    |
// listener lambda impl and to iterate there?                   |
//                                                              |
// For more data types, should their insertion operators be     |
// added centrally here, or should the user provide them in     |
// client code?                                                 |
//--------------------------------------------------------------+
 
template <typename T>
inline std::ostream& operator<<(std::ostream& os, const std::vector<T>& vec) {
    os << "{";
    for (std::size_t i = 0; i < vec.size(); ++i) {
        os << vec[i];
        if (i != vec.size() - 1) {
            os << ", ";
        }
    }
    os << "}";
    return os;
}
 
 
//--------------------------------------------------------------+
//                      CiiOldbDataPointAsync                   |
//                                                              |
//--------------------------------------------------------------+
 
template<typename T>
CiiOldbDataPointAsync<T>::CiiOldbDataPointAsync(
        std::string name,
        std::shared_ptr<elt::oldb::CiiOldbDataPoint<T>> delegate,
        boost::asio::thread_pool& async_exec,
        const log4cplus::Logger& logger,
        std::size_t buffer_capacity )
    : m_sync_dtor_cv {},
      m_sync_dtor_n_writing {0},
      m_sync_dtor_mutex {},
      logger {logger},
      name {name},
      delegate {delegate},
      buffer {name, buffer_capacity, logger},
      async_exec {async_exec},
      is_async_processing {false}
{
 
    buffer.SetDiscardListener(
        [this](OldbDataWithPromise& discarded_data) {
 
            // Check the oldest remaining elements in the buffer, if it is incomplete. If our discarded_data has the missing fields,
            // then they should be transferred to not get lost.
            auto lock = buffer.Lock(); // should already be locked by CircularBufferConcurrent::Push or ::Close calling us, but we do it also here for clarity.
 
            boost::circular_buffer<OldbDataWithPromise>& cb = buffer.GetCb(); 
 
            // If the buffer is full, which is expected if a buffer.Push call got us here 
            // (as opposed to a buffer.Close call where a full buffer would be coindicental), 
            // and the retained data has missing attributes that are present in the older to-be-discarded data,
            // then we transfer these attributes. For that we peek into the internal buffer to check the retained data.
            // We exceptionally use a raw pointer, which serves both as a flag for this situation and for data access.
            // std::optional does not work here because we have only a reference to the retained data, cannot move it out.
            OldbDataWithPromise* oldest_retained_data = ( cb.full() ? &cb.front() : nullptr );
 
            std::ostringstream msg {};
 
            if (discarded_data.GetValue()) {
                msg << "value=" << *discarded_data.GetValue();
                if (oldest_retained_data && !oldest_retained_data->GetValue()) {
                    oldest_retained_data->SetValue(*discarded_data.GetValue());
                    msg << " (transferred to retained newer data)";
                }
                msg << ", ";
            }
 
            if (discarded_data.GetTimestamp()) {
                msg << "timestamp=" << *discarded_data.GetTimestamp();
                if (oldest_retained_data && !oldest_retained_data->GetTimestamp()) {
                    oldest_retained_data->SetTimestamp(*discarded_data.GetTimestamp());
                    msg << " (transferred to retained newer data)";
                }
                msg << ", ";
            }
 
            if (discarded_data.GetQuality()) {
                std::string quality_str = hlcc::cpputil::CiiOldbDpQualityToString(*discarded_data.GetQuality());
                msg << "quality=" << quality_str;
                // no need to transfer, since data produced with WriteValue has quality default value, 
                // and data produced with SetQuality anyway has the quality.
            }
            LOG4CPLUS_DEBUG(this->logger, "OLDB async (" << this->name << ") discarded data: " << msg.str());
 
            // If a DP value was discarded from the buffer, we need to notify the client through its future object.
            // TODO: Is there a standardized (std or ELT etc) exception for cancelled operations? std::out_of_range does not really fit..
            // TODO use std::make_exception_ptr if we use std::promise instead of boost.
            discarded_data.GetPromise()->set_exception(boost::copy_exception(std::out_of_range{"Dropped data."}));
 
            std::lock_guard lck {this->m_sync_dtor_mutex};
            this->m_sync_dtor_n_writing--;
    });
}
 
 
template<typename T>
CiiOldbDataPointAsync<T>::~CiiOldbDataPointAsync() {
    RAD_TRACE(logger);
 
    // notify/release pool thread in Poll method, if any.
    buffer.Close(); 
 
 
    // We lock m_sync_dtor_mutex after calling Close(), because Close() calls the discard handler
    // that locks m_sync_dtor_mutex for a short time.
    // What matters is that we lock m_sync_dtor_mutex before calling m_sync_dtor_cv.wait.
    // It is the responsibility of the application to not call write methods when the dtor
    // is running, thus no need to lock m_sync_dtor_mutex at the beginning of the dtor.
    std::unique_lock lck {m_sync_dtor_mutex};
 
    // Block dtor (if needed) until pool thread has left WriteBufferToOldb.
    // We must not have this CiiOldbDataPointAsync object destroyed while another thread is using it.
    // Note that we cannot simply join the external pool thread, because it lives beyond this object.
    // There are also no callbacks offered by the boost thread pool that would tell us when a task has finished.
    m_sync_dtor_cv.wait(lck, [this]{ 
        LOG4CPLUS_TRACE(logger, "~CiiOldbDataPointAsync (" << name << ") m_sync_dtor_cv.wait predicate function: " 
                            << "m_sync_dtor_n_writing=" << m_sync_dtor_n_writing);
        return (m_sync_dtor_n_writing == 0); 
    });
 
    LOG4CPLUS_TRACE(logger, "~CiiOldbDataPointAsync (" << name << ") is done.");
}
 
 
template<typename T>
std::shared_ptr<elt::oldb::CiiOldbDpValue<T>> CiiOldbDataPointAsync<T>::ReadValue(bool check_bad_quality) {
    return delegate->ReadValue(check_bad_quality);
}
 

template<typename T>
boost::future<typename CiiOldbDataPointAsync<T>::OldbData> CiiOldbDataPointAsync<T>::WriteValue(
        const T& value, int64_t timestamp, elt::oldb::CiiOldbDpQuality quality, bool is_disable_publishing) {
 
    { // lock scope
        // If this write call is made before the dtor runs, then it will block the dtor.
        // The reverse situation is illegal: No new write calls must be made once the dtor runs. 
        std::lock_guard lck {m_sync_dtor_mutex};
        m_sync_dtor_n_writing++;
    }
    // Note that we store a copy of value in new_data. Thus no lifetime issues with our "T& value" parameter.
    typename CiiOldbDataPointAsync<T>::OldbData new_data {value, timestamp, quality};
    return CiiOldbDataPointAsync<T>::WriteAsync(std::move(new_data), is_disable_publishing);
}
 
 
template<typename T>
boost::future<typename CiiOldbDataPointAsync<T>::OldbData> CiiOldbDataPointAsync<T>::SetQuality(
        elt::oldb::CiiOldbDpQuality quality, bool is_disable_publishing) {
 
    { // lock scope
        std::lock_guard lck {m_sync_dtor_mutex};
        m_sync_dtor_n_writing++;
    }
    typename CiiOldbDataPointAsync<T>::OldbData new_data {quality};
    return CiiOldbDataPointAsync<T>::WriteAsync(std::move(new_data), is_disable_publishing);
}
 
 
template<typename T>
boost::future<typename CiiOldbDataPointAsync<T>::OldbData> CiiOldbDataPointAsync<T>::WriteAsync(
        OldbData&& new_data, bool is_disable_publishing) {
 
    // TODO: If specific write methods get added in the future, they all must call 
    // m_sync_dtor_n_writing++
    // the same way that WriteValue and SetQuality do. 
 
    std::chrono::steady_clock::time_point t_0 {std::chrono::steady_clock::now()};
 
    // We need a future/promise pair that is separate from any future we may obtain from running a thread.
    // The reasons are
    // (a) our lossy buffer, where we must "cancel" the future toward the client also
    //     when the data gets evicted from the buffer, without having attempted to write it to OLDB.
    //     See the SetDiscardListener call in our ctor.
    // (b) the optimization that we only push the data to the buffer without starting a thread
    //     when there is already another thread busy with moving data from the buffer to the OLDB.
    // The future will inform the client about any success or failure in writing the data to OLDB.
    boost::promise<OldbData> promise {};
    boost::future<OldbData> future = promise.get_future();
 
    // TODO: handle is_disable_publishing
 
    std::chrono::steady_clock::time_point t_1 {std::chrono::steady_clock::now()};
 
    typename CiiOldbDataPointAsync<T>::OldbDataWithPromise new_data_with_promise {
        // no std::move on the value, otherwise warning "moving a temporary object prevents copy elision"
        new_data.GetValue(), new_data.GetTimestamp(), new_data.GetQuality(), std::move(promise)};
 
    // Write data to the circular buffer. 
    // This discards older data if the buffer is full.
    // Once the data is in the buffer, it may be processed immediately by a background
    // thread that was started for the previously pushed data.
    buffer.Push(std::move(new_data_with_promise));
 
    std::chrono::steady_clock::time_point t_2 {std::chrono::steady_clock::now()};
 
    if (new_data.GetValue()) {
        LOG4CPLUS_TRACE(logger, "OLDB async (" << name << ") wrote to buffer: " 
            << *new_data_with_promise.GetValue() << ", threadId=" << std::this_thread::get_id());
    } else {
        LOG4CPLUS_TRACE(logger, "OLDB async (" << name << ") wrote to buffer: <no value>" 
            << ", threadId=" << std::this_thread::get_id());
    }
 
    // Trigger background processing of buffered data, unless we are already processing.
    bool expected = false; // TODO: how to pass bool& as a 'false' literal, without declaring the variable?
 
    std::chrono::steady_clock::time_point t_3 {};
    std::chrono::steady_clock::time_point t_4 {};
 
    // if is_async_processing == false then set it to true and execute the code block that triggers background processing.
    if (is_async_processing.compare_exchange_strong(expected, true)) {  
 
        LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name << LOG4CPLUS_TEXT(") will use a writer thread."));
 
        // Write the buffer content to the OLDB, using sync CII OLDB client API calls in a separate thread.
        // New data can come in concurrently and will be included. The thread finishes when the buffer is fully drained.
        //
        // We use a shared thread pool to cut thread creation overhead and to centrally synchronize
        // on thread completion during application shutdown (otherwise segfaults, see ETCS-629).
        //
        // Note that the externally managed thread pool resolves also an issue that we would otherwise have with using
        // std::async. There would be the "waiting destructor" issue of the special version of future returned,
        // see https://en.cppreference.com/w/cpp/thread/future/~future. The returned future
        // runs out of scope before we return from this method. Thus this method itself waits for
        // the background thread execution to finish, making OLDB access an unwanted sync call.
        // This problem almost caused deprecation of std::async, see http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2013/n3780.pdf
        // With std::thread we would have the same scope and blocking issue, unless we externally manage the single thread
        // or we detach the thread. But then the C++ standard does not define what happens with detached threads when
        // the application exits, and we cannot sync application shutdown on the detached thread. All messed up.
        //
        // Note on the number of threads and the task queue: In the Java prototype we used infinite maximumPoolSize
        // (well, Integer.MAX_VALUE). Here in C++ with the much larger thread stack size and possibly other thread
        // overhead, we use a quite limited number of threads.
        // OLDB's underlying Redis is single-threaded per instance. The ECM installation uses 6 Redis instances,
        // so that our 5 threads seem reasonable also from that perspective.
        // The finite thread pool has an unlimited task queue in front of it.
        // This should ensure that we return quickly from the 'post' call (where in reality it is sometimes very slow!)
        // If new data arrives while a thread is already writing to OLDB, it will also include the new data.
        // Thus it may happen that another thread will later find an empty data buffer. That should be almost no overhead
        // thanks to reusing threads from a pool, and on the other hand it ensures that even when all threads are busy,
        // no data will starve in the buffer.
 
        t_3 = std::chrono::steady_clock::now();
 
        // The lambda function helps to wrap our non-static member function WriteBufferToOldb as a std::function.
        boost::asio::post(async_exec, [this](){ WriteBufferToOldb(); });
 
        t_4 = std::chrono::steady_clock::now();
    } else {
        t_3 = t_4 = std::chrono::steady_clock::now();
        // There is already a thread draining the buffer (writing to OLDB).
        // Our data that we've pushed into the buffer will be considered by that other thread.
        LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name << LOG4CPLUS_TEXT(") will rely on running writer thread."));
    }
 
    LOG4CPLUS_TRACE(logger, "OLDB async (" << name << ") Returning from WriteAsync");
 
    int d1 = std::chrono::duration_cast<std::chrono::microseconds>(t_1 - t_0).count();
    int d2 = std::chrono::duration_cast<std::chrono::microseconds>(t_2 - t_1).count();
    int d3 = std::chrono::duration_cast<std::chrono::microseconds>(t_3 - t_2).count();
    int d4 = std::chrono::duration_cast<std::chrono::microseconds>(t_4 - t_3).count();
 
    if (d1 + d2 + d3 + d4 > 1000) {
        LOG4CPLUS_DEBUG(logger, "CiiOldbDataPointAsync " << name << " WriteAsync steps in micros: "
                << d1 << ", " << d2 << ", " << d3 << ", " << d4);;
    }
    return future; // compiler will do copy elision or implied std::move
}
 
 
template<typename T>
void CiiOldbDataPointAsync<T>::WriteBufferToOldb() {
    LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name << LOG4CPLUS_TEXT(") writer thread started. is_async_processing=") << is_async_processing <<
            ", threadId=" << std::this_thread::get_id()); // note that the remote logs contain threadId also in the "SourceID" field.
 
    int write_count = 0;
 
    // Fetch data, but without waiting. 
    // Normally there should be data available, as otherwise the worker thread would not be running here.
    std::optional<OldbDataWithPromise> data_to_write_opt {buffer.Poll()};
 
    if (!data_to_write_opt) {
        // There is the rare case that method WriteValue pushed new data to the buffer while a background thread 
        // from a previous write was still draining the buffer, but that background write finished before method WriteValue
        // checked variable is_async_processing. Then we get here with an empty buffer.
        // All we need to do is restore this flag.
        is_async_processing = false;
    } else {
        // Process new data as it gets added to the buffer (possibly also replacing older data, while we write).
        // Once the destructor runs, we just finish current data, but discard queued data.
        while (data_to_write_opt.has_value()) {
 
            LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name << ") thread got data and will call the sync writeValue.");
 
            OldbDataWithPromise data_to_write {std::move(data_to_write_opt.value())};
 
            try {
                auto start = std::chrono::steady_clock::now(); // TODO do we have a Stopwatch utility class?
 
                // make the synchronous OLDB call
                if (data_to_write.GetValue() && data_to_write.GetTimestamp() && data_to_write.GetQuality()) {
                    delegate->WriteValue(*data_to_write.GetValue(), *data_to_write.GetTimestamp(), *data_to_write.GetQuality(), false);
                    LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name <<
                            LOG4CPLUS_TEXT(") wrote DP to OLDB in ") << std::chrono::duration_cast<std::chrono::milliseconds>((std::chrono::steady_clock::now() - start)).count() <<
                            " ms, data=" << LOG4CPLUS_TEXT(*data_to_write.GetValue()) );  // TODO log also time and quality? 
                } 
                else if (data_to_write.GetQuality()) {
                    delegate->SetQuality(*data_to_write.GetQuality());
                    LOG4CPLUS_TRACE(logger, "OLDB async (" << name <<
                            ") wrote DP quality to OLDB in " << std::chrono::duration_cast<std::chrono::milliseconds>((std::chrono::steady_clock::now() - start)).count() <<
                            " ms, quality=" << hlcc::cpputil::CiiOldbDpQualityToString(*data_to_write.GetQuality()) );
                }
                else {
                    LOG4CPLUS_DEBUG(logger, LOG4CPLUS_TEXT("Programming error: OLDB async (") << name <<
                            ") could not write DP because of missing data in OldbDataWithPromise");
                }
 
                data_to_write.GetPromise()->set_value(std::move(data_to_write));
                write_count++;
            } catch (const elt::oldb::CiiOldbException& ex) {
                LOG4CPLUS_DEBUG(logger, LOG4CPLUS_TEXT("OLDB async (") << name << ") writing to OLDB failed: " << ex.what());
                try {
                    // store exception in the promise/future
                    data_to_write.GetPromise()->set_exception(boost::copy_exception(ex));
                } catch(...) {} // set_exception() may throw too
            } catch (...) {
                // TODO: should we let the application crash, or also set an ex on the promise?
                LOG4CPLUS_WARN(logger, LOG4CPLUS_TEXT("OLDB async (") << name << ") unknown exception.");
            }
 
            // Check if there is more data in the buffer that needs to be written to OLDB.
            // We lock the buffer so that toggling of flag is_async_processing is safe from
            // concurrent Push calls, which might otherwise lead to data starving in the buffer.
            { // scope for the lock
                auto lock = buffer.Lock();
                data_to_write_opt = buffer.Poll();
                if (!data_to_write_opt.has_value()) {
                    is_async_processing.store(false); // from here on, newly written data will call WriteBufferToOldb again.
                    // we'll exit from the while loop.
                }
            }
        }
        LOG4CPLUS_TRACE(logger, LOG4CPLUS_TEXT("OLDB async (") << name <<
                LOG4CPLUS_TEXT(") performed ") << write_count << LOG4CPLUS_TEXT(" DP writes in the background, now releasing writer thread."));
    }
 
    // Decrement the n_writing flag and notify the CV, in case the dtor is waiting on it.
    // TODO: Should we wrap the code above in a try, and the code below in a catch(...), or is all critical code already in try-catch?
    //       The risk with an excpetion is to not decrement m_sync_dtor_n_writing and thus starve the dtor (until spurious wakeup).
    std::lock_guard lck {m_sync_dtor_mutex};
    m_sync_dtor_n_writing -= write_count;
    LOG4CPLUS_TRACE(logger, "OLDB async (" << name << ") WriteBufferToOldb done. m_sync_dtor_n_writing=" << m_sync_dtor_n_writing);
    // We notify while under the lock, to hold off a loitering dtor until this worker thread has ended in peace.
    m_sync_dtor_cv.notify_all();
}
 
}  // namespace hlcc::oldbmux