uniface.h

Go to the documentation of this file.

/*****************************************************************************
* Multiscale Universal Interface Code Coupling Library                       *
*                                                                            *
* Copyright (C) 2019 Y. H. Tang, S. Kudo, X. Bian, Z. Li, G. E. Karniadakis, *
*                    S. M. Longshaw, A. Skillen                              *
*                                                                            *
* This software is jointly licensed under the Apache License, Version 2.0    *
* and the GNU General Public License version 3, you may use it according     *
* to either.                                                                 *
*                                                                            *
* ** Apache License, version 2.0 **                                          *
*                                                                            *
* Licensed under the Apache License, Version 2.0 (the "License");            *
* you may not use this file except in compliance with the License.           *
* You may obtain a copy of the License at                                    *
*                                                                            *
* http://www.apache.org/licenses/LICENSE-2.0                                 *
*                                                                            *
* Unless required by applicable law or agreed to in writing, software        *
* distributed under the License is distributed on an "AS IS" BASIS,          *
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   *
* See the License for the specific language governing permissions and        *
* limitations under the License.                                             *
*                                                                            *
* ** GNU General Public License, version 3 **                                *
*                                                                            *
* This program is free software: you can redistribute it and/or modify       *
* it under the terms of the GNU General Public License as published by       *
* the Free Software Foundation, either version 3 of the License, or          *
* (at your option) any later version.                                        *
*                                                                            *
* This program is distributed in the hope that it will be useful,            *
* but WITHOUT ANY WARRANTY; without even the implied warranty of             *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
* GNU General Public License for more details.                               *
*                                                                            *
* You should have received a copy of the GNU General Public License          *
* along with this program.  If not, see <http://www.gnu.org/licenses/>.      *
*****************************************************************************/
 
#ifndef UNIFACE_H_
#define UNIFACE_H_
 
#include "general/util.h"
#include "communication/comm.h"
#include "communication/comm_factory.h"
#include "config.h"
#include "storage/dynstorage.h"
#include "storage/spatial_storage.h"
#include "communication/lib_dispatcher.h"
#include "communication/message/message.h"
#include "communication/message/reader_variable.h"
#include "storage/stream_vector.h"
#include "storage/stream_unordered.h"
#include "storage/stream_string.h"
#include "storage/bin.h"
#include "storage/stream.h"
 
#ifdef PYTHON_BINDINGS
#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
#include <pybind11/numpy.h>
namespace py = pybind11;
#endif
 
namespace mui {
 
template<typename CONFIG = default_config>
class uniface {
public:
    // public typedefs (see config.h for descriptions)
    static const int  D     = CONFIG::D;
    static const bool FIXEDPOINTS = CONFIG::FIXEDPOINTS;
    static const bool QUIET = CONFIG::QUIET;
 
    using REAL = typename CONFIG::REAL;
    using point_type = typename CONFIG::point_type;
    using time_type = typename CONFIG::time_type;
    using iterator_type = typename CONFIG::iterator_type;
    using data_types = typename CONFIG::data_types;
    using span_t = geometry::any_shape<CONFIG>;
private:
    // meta functions to split tuple and add vector<pair<point_type,_1> >
    template<typename T> struct add_vp_ { using type = std::vector<std::pair<point_type,T> >; };
        template<typename T> struct def_storage_;
    template<typename... TYPES> struct def_storage_<type_list<TYPES...> >{
        using type = storage<typename add_vp_<TYPES>::type...>;
    };
 
    template<typename T> struct add_vi_ { using type = std::vector<std::pair<size_t, T> >; };
    template<typename T> struct def_storage_raw_;
    template<typename... TYPES> struct def_storage_raw_<type_list<TYPES...> >{
        using type = storage<typename add_vi_<TYPES>::type...>;
    };
 
    template<typename T> struct def_storage_single_;
    template<typename... TYPES> struct def_storage_single_<type_list<TYPES...> >{
        using type = storage<TYPES...>;
    };
 
    // internal typedefinitions for full frame
    using storage_t = typename def_storage_<data_types>::type;
    using spatial_t = spatial_storage<bin_t<CONFIG>,storage_t,CONFIG>;
    using frame_type = std::unordered_map<std::string, storage_t>;
    using bin_frame_type = std::unordered_map<std::string, spatial_t>;
    // internal typdefinitions for data values only (static points)
    using storage_raw_t = typename def_storage_raw_<data_types>::type;
    using frame_raw_type = std::unordered_map<std::string, storage_raw_t>;
    // internal typedefinitions for single value
    using storage_single_t = typename def_storage_single_<data_types>::type;
 
    struct peer_state {
        peer_state() : disable_send(false), disable_recv(false), ss_stat_send(false), ss_stat_recv(false) {}
 
        using spans_type = std::map<std::pair<time_type,time_type>,span_t>;
 
        bool is_recving(time_type t, const span_t& s) const {
            return scan_spans_(t,s,recving_spans);
        }
        
        void set_recving( time_type start, time_type end, span_t s ) {
            recving_spans.emplace(std::make_pair(start,end),std::move(s));
        }
        
        bool is_sending(time_type t, const span_t& s) const {
            return scan_spans_(t,s,sending_spans);
        }
        
        void set_sending(time_type start, time_type end, span_t s) {
            sending_spans.emplace(std::make_pair(start,end), std::move(s));
        }
 
        void set_pts(std::vector<point_type>& pts) {
            pts_ = pts;
        }
 
        const std::vector<point_type>& pts() const {
            return pts_;
        }
        
        void set_send_disable() {
            disable_send = true;
        }
        
        void set_recv_disable() {
            disable_recv = true;
        }
        
        bool is_send_disabled() const {
            return disable_send;
        }
        
        bool is_recv_disabled() const {
            return disable_recv;
        }
 
        void set_ss_send_status(bool status) {
            ss_stat_send = status;
        }
 
        bool ss_send_status() const {
            return ss_stat_send;
        }
 
        void set_ss_recv_status(bool status) {
            ss_stat_recv = status;
        }
 
        bool ss_recv_status() const {
            return ss_stat_recv;
        }
 
        time_type current_t() const { return latest_timestamp; }
        iterator_type current_it() const { return latest_subiter; }
        time_type next_t() const { return next_timestamp; }
        iterator_type next_it() const { return next_subiter; }
        void set_current_t( time_type t ) { latest_timestamp = t; }
        void set_current_sub( iterator_type i ) { latest_subiter = i; }
        void set_next_t( time_type t ) { next_timestamp = t; }
        void set_next_sub( iterator_type i ) { next_subiter = i; }
    private:
        bool scan_spans_(time_type t, const span_t& s, const spans_type& spans ) const {
            bool prefetched = false;
            auto end = spans.lower_bound(std::make_pair(t,t));
            if( spans.size() == 1 ) end = spans.end();
 
            for( auto itr = spans.begin(); itr != end; ++itr ) {
                if( t < itr->first.second || almost_equal(t, itr->first.second) ) {
                    prefetched = true;
                    if( collide(s,itr->second) ) return true;
                }
            }
 
            // if prefetched at t, but no overlap region, then return false;
            // otherwise return true;
            return !prefetched;
        }
 
        time_type latest_timestamp = std::numeric_limits<time_type>::lowest();
        iterator_type latest_subiter = std::numeric_limits<iterator_type>::lowest();
        time_type next_timestamp = std::numeric_limits<time_type>::lowest();
        iterator_type next_subiter = std::numeric_limits<iterator_type>::lowest();
        spans_type recving_spans;
        spans_type sending_spans;
        std::vector<point_type> pts_;
        std::unordered_map<std::string, storage_single_t> assigned_vals_;
        bool disable_send;
        bool disable_recv;
        bool ss_stat_send;
        bool ss_stat_recv;
    };
 
private: // data members
    std::unique_ptr<communicator> comm;
    dispatcher<message::id_type, std::function<void(message)> > readers;
 
    std::map<std::pair<time_type, iterator_type>, bin_frame_type> log;
 
    frame_type push_buffer;
    frame_raw_type push_buffer_raw;
    std::vector<point_type> push_buffer_pts;
 
    std::unordered_map<std::string, storage_single_t > assigned_values;
 
    std::vector<peer_state> peers;
    std::vector<bool> peer_is_sending;
    bool smart_send_set_ = true;
    time_type span_start = std::numeric_limits<time_type>::lowest();
    time_type span_timeout = std::numeric_limits<time_type>::lowest();
    span_t current_span;
    time_type recv_start = std::numeric_limits<time_type>::lowest();
    time_type recv_timeout = std::numeric_limits<time_type>::lowest();
    span_t recv_span;
    time_type memory_length = std::numeric_limits<time_type>::max();
    std::mutex mutex;
    bool initialized_pts_;
    size_t fixedPointCount_;
    time_type fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
    iterator_type fetch_i_hist_ = std::numeric_limits<iterator_type>::lowest();
 
public:
    uniface( const char URI[] ) : uniface( comm_factory::create_comm(URI, QUIET) ) {}
    uniface( std::string const &URI ) : uniface( comm_factory::create_comm(URI.c_str(), QUIET) ) {}
    uniface( communicator* comm_ ) : comm(comm_), initialized_pts_(false), fixedPointCount_(0) {
        using namespace std::placeholders;
 
        peers.resize(comm->remote_size());
        peer_is_sending.resize(comm->remote_size(), true);
 
        readers.link("timestamp", reader_variables<int32_t, std::pair<time_type,iterator_type> >(
                     std::bind(&uniface::on_recv_confirm, this, _1, _2)));
        readers.link("forecast", reader_variables<int32_t, std::pair<time_type,iterator_type>>(
                     std::bind(&uniface::on_recv_forecast, this, _1, _2)));
        readers.link("data", reader_variables<std::pair<time_type,iterator_type>, frame_type>(
                     std::bind(&uniface::on_recv_data, this, _1, _2)));
        readers.link("rawdata", reader_variables<int32_t, std::pair<time_type,iterator_type>, frame_raw_type>(
                     std::bind(&uniface::on_recv_rawdata, this, _1, _2, _3)));
        readers.link("points", reader_variables<int32_t, std::vector<point_type>>(
                     std::bind(&uniface::on_recv_points, this, _1, _2)));
        readers.link("assignedVals", reader_variables<std::string, storage_single_t>(
                     std::bind(&uniface::on_recv_assignedVals, this, _1, _2)));
        readers.link("receivingSpan", reader_variables<int32_t, time_type,iterator_type, span_t>(
               std::bind(&uniface::on_recv_span, this, _1, _2, _3, _4)));
        readers.link("sendingSpan", reader_variables<int32_t, time_type,iterator_type, span_t>(
               std::bind(&uniface::on_send_span, this, _1, _2, _3, _4)));
        readers.link("receivingDisable", reader_variables<int32_t>(
                     std::bind(&uniface::on_send_disable, this, _1)));
        readers.link("sendingDisable", reader_variables<int32_t>(
                     std::bind(&uniface::on_recv_disable, this, _1)));
    }
 
    uniface( const uniface& ) = delete;
    uniface& operator=( const uniface& ) = delete;
 
    template<typename TYPE>
    void push( const std::string& attr, const TYPE& value ) {
        comm->send(message::make("assignedVals", attr, storage_single_t(TYPE(value))));
    }
 
    template<typename TYPE>
    void push( const std::string& attr, const point_type& loc, const TYPE& value ) {
        if( FIXEDPOINTS ) {
            // If this push is before first commit then build local points list
            if( !initialized_pts_ ) push_buffer_pts.emplace_back( loc );
 
            storage_raw_t& n = push_buffer_raw[attr];
            if( !n ) n = storage_raw_t(std::vector<std::pair<size_t,TYPE> >());
            storage_cast<std::vector<std::pair<size_t,TYPE> >&>(n).emplace_back( fixedPointCount_, value );
 
            // Increment counter for flat fixed point list
            fixedPointCount_++;
        } 
        else {
            storage_t& n = push_buffer[attr];
            if( !n ) n = storage_t(std::vector<std::pair<point_type,TYPE> >());
            storage_cast<std::vector<std::pair<point_type,TYPE> >&>(n).emplace_back( loc, value );
        }
    }
 
#ifdef PYTHON_BINDINGS
    template<typename TYPE>
    void push_many(const std::string& attr, const class py::array_t<REAL>& points,
                   const class py::array_t<TYPE>& values) {
        // Arrays must have ndim = d; can be non-writeable
        point_type p = 0;
        auto points_arr = points.template unchecked<2>();
        auto values_arr = values.template unchecked<1>();
        assert(points_arr.shape(0) == values_arr.shape(0));
        for (ssize_t i = 0; i < points_arr.shape(0); i++) {
            for (ssize_t j = 0; j < points_arr.shape(1); j++)
                p[j] = points_arr(i,j);
            push<TYPE>(attr, p, values_arr(i));
        }
    }
 
    template<class SAMPLER, class TIME_SAMPLER>
    py::array_t<typename SAMPLER::OTYPE,py::array::c_style>
    fetch_many(const std::string& attr,const py::array_t<REAL,py::array::c_style> points, const time_type t,
        const SAMPLER &sampler, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true) {
        // Arrays must have ndim = d; can be non-writeable
        point_type p = 0;
        auto points_arr = points.template unchecked<2>();
        py::array_t<typename SAMPLER::OTYPE,py::array::c_style> values(points_arr.shape(0));
        auto values_arr = values.template mutable_unchecked<1>();
        for (ssize_t i = 0; i < points_arr.shape(0); i++) {
            for (ssize_t j = 0; j < points_arr.shape(1); j++)
                p[j] = points_arr(i,j);
            values_arr(i)  = fetch(attr, p, t, sampler, t_sampler, barrier_enabled);
        }
        return values;
    }
 
    template<class SAMPLER, class TIME_SAMPLER>
    py::array_t<typename SAMPLER::OTYPE,py::array::c_style>
    fetch_many(const std::string& attr,const py::array_t<REAL,py::array::c_style> points, const time_type t,
        const iterator_type it, const SAMPLER &sampler, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true) {
        // Arrays must have ndim = d; can be non-writeable
        point_type p = 0;
        auto points_arr = points.template unchecked<2>();
        py::array_t<typename SAMPLER::OTYPE,py::array::c_style> values(points_arr.shape(0));
        auto values_arr = values.template mutable_unchecked<1>();
        for (ssize_t i = 0; i < points_arr.shape(0); i++) {
            for (ssize_t j = 0; j < points_arr.shape(1); j++)
                p[j] = points_arr(i,j);
            values_arr(i)  = fetch(attr, p, t, it, sampler, t_sampler, barrier_enabled);
        }
        return values;
    }
 
    template<class SAMPLER, class TIME_SAMPLER, class ALGORITHM>
    py::array_t<typename SAMPLER::OTYPE,py::array::c_style>
    fetch_many(const std::string& attr,const py::array_t<REAL,py::array::c_style> points, const time_type t,
        const SAMPLER &sampler, const TIME_SAMPLER &t_sampler, const ALGORITHM &algorithm, bool barrier_enabled = true) {
        // Arrays must have ndim = d; can be non-writeable
        point_type p = 0;
        auto points_arr = points.template unchecked<2>();
        py::array_t<typename SAMPLER::OTYPE,py::array::c_style> values(points_arr.shape(0));
        auto values_arr = values.template mutable_unchecked<1>();
        for (ssize_t i = 0; i < points_arr.shape(0); i++) {
            for (ssize_t j = 0; j < points_arr.shape(1); j++)
                p[j] = points_arr(i,j);
            values_arr(i)  = fetch(attr, p, t, sampler, t_sampler, algorithm, barrier_enabled);
        }
        return values;
    }
 
    template<class SAMPLER, class TIME_SAMPLER, class ALGORITHM>
    py::array_t<typename SAMPLER::OTYPE,py::array::c_style>
    fetch_many(const std::string& attr,const py::array_t<REAL,py::array::c_style> points, const time_type t,
        const iterator_type it, const SAMPLER &sampler, const TIME_SAMPLER &t_sampler, const ALGORITHM &algorithm, bool barrier_enabled = true) {
        // Arrays must have ndim = d; can be non-writeable
        point_type p = 0;
        auto points_arr = points.template unchecked<2>();
        py::array_t<typename SAMPLER::OTYPE,py::array::c_style> values(points_arr.shape(0));
        auto values_arr = values.template mutable_unchecked<1>();
        for (ssize_t i = 0; i < points_arr.shape(0); i++) {
            for (ssize_t j = 0; j < points_arr.shape(1); j++)
                p[j] = points_arr(i,j);
            values_arr(i)  = fetch(attr, p, t, it, sampler, t_sampler, algorithm, barrier_enabled);
        }
        return values;
    }
 
 
    template<typename TYPE, class TIME_SAMPLER>
    py::array_t<REAL, py::array::c_style>
    fetch_points_np(const std::string& attr, const time_type t, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, TYPE test_value = static_cast<TYPE>(0)) {
        std::vector<point_type> points = fetch_points<TYPE>(attr, t, t_sampler, barrier_enabled);
        size_t n = points.size();
        test_value += 1;
        py::array_t<REAL, py::array::c_style> points_np({n, static_cast<size_t>(D)});
        auto points_np_arr = points_np.template mutable_unchecked<2>();
        for (std::size_t i = 0; i < n; i++)
            for (std::size_t j = 0; j < D; j++)
                points_np_arr(i,j) = (points[i].data())[j];
        return points_np;
    }
 
    template<typename TYPE, class TIME_SAMPLER>
    py::array_t<REAL, py::array::c_style>
    fetch_points_np(const std::string& attr, const time_type t, const iterator_type it, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, TYPE test_value = static_cast<TYPE>(0)) {
        std::vector<point_type> points = fetch_points<TYPE>(attr, t, it, t_sampler, barrier_enabled);
        size_t n = points.size();
        test_value += 1;
        py::array_t<REAL, py::array::c_style> points_np({n, static_cast<size_t>(D)});
        auto points_np_arr = points_np.template mutable_unchecked<2>();
        for (std::size_t i = 0; i < n; i++)
            for (std::size_t j = 0; j < D; j++)
                points_np_arr(i,j) = (points[i].data())[j];
        return points_np;
    }
#endif
 
    template<typename TYPE>
    TYPE fetch( const std::string& attr ) {
        storage_single_t& n = assigned_values[attr];
        if( !n ) return TYPE();
        return storage_cast<TYPE&>(n);
    }
 
    template<class SAMPLER, class TIME_SAMPLER, typename ... ADDITIONAL>
    typename SAMPLER::OTYPE
    fetch( const std::string& attr,const point_type& focus, const time_type t,
           SAMPLER& sampler, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true,
           ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t
        if( fetch_t_hist_ != t && barrier_enabled )
            barrier(t_sampler.get_upper_bound(t));
 
        fetch_t_hist_ = t;
 
        std::vector<std::pair<std::pair<time_type,iterator_type>,typename SAMPLER::OTYPE> > v;
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ) {
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            v.emplace_back( start->first, iter->second.build_and_query_ts( focus, sampler, additional... ) );
        }
 
        return t_sampler.filter(t, v);
    }
 
    template<class SAMPLER, class TIME_SAMPLER, typename ... ADDITIONAL>
    typename SAMPLER::OTYPE
    fetch( const std::string& attr,const point_type& focus, const time_type t, const iterator_type it,
           SAMPLER& sampler, const TIME_SAMPLER &t_sampler, bool barrier_enabled = true,
           ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t,iteration=it
        if((fetch_t_hist_ != t || fetch_i_hist_ != it) && barrier_enabled)
            barrier(t_sampler.get_upper_bound(t),t_sampler.get_upper_bound(it));
 
        fetch_t_hist_ = t;
        fetch_i_hist_ = it;
 
        std::vector<std::pair<std::pair<time_type,iterator_type>,typename SAMPLER::OTYPE> > v;
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           t_sampler.get_lower_bound(it)-threshold(it));
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           t_sampler.get_upper_bound(it)+threshold(it));
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ) {
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            v.emplace_back( start->first, iter->second.build_and_query_ts( focus, sampler, additional... ) );
        }
 
        return t_sampler.filter(std::make_pair(t,it), v);
    }
 
    template<class SAMPLER, class TIME_SAMPLER, class COUPLING_ALGO, typename ... ADDITIONAL>
    typename SAMPLER::OTYPE
    fetch( const std::string& attr,const point_type& focus, const time_type t,
           SAMPLER& sampler, const TIME_SAMPLER &t_sampler, const COUPLING_ALGO &cpl_algo, 
           bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t
        if( fetch_t_hist_ != t && barrier_enabled )
            barrier(t_sampler.get_upper_bound(t));
 
        fetch_t_hist_ = t;
 
        std::vector<std::pair<std::pair<time_type,iterator_type>,typename SAMPLER::OTYPE> > v;
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                        std::numeric_limits<iterator_type>::lowest());
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                        std::numeric_limits<iterator_type>::lowest());
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ) {
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            v.emplace_back( start->first, iter->second.build_and_query_ts( focus, sampler, additional... ) );
        }
 
        return cpl_algo.relaxation(std::make_pair(std::numeric_limits<time_type>::lowest(), static_cast<iterator_type>(t)), focus, t_sampler.filter(t, v));
    }
 
    template<class SAMPLER, class TIME_SAMPLER, class COUPLING_ALGO, typename ... ADDITIONAL>
    typename SAMPLER::OTYPE
    fetch( const std::string& attr,const point_type& focus, const time_type t, const iterator_type it,
           SAMPLER& sampler, const TIME_SAMPLER &t_sampler, const COUPLING_ALGO &cpl_algo, 
           bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t,iteration=it
        if((fetch_t_hist_ != t || fetch_i_hist_ != it) && barrier_enabled)
            barrier(t_sampler.get_upper_bound(t),t_sampler.get_upper_bound(it));
 
        fetch_t_hist_ = t;
        fetch_i_hist_ = it;
 
        std::vector<std::pair<std::pair<time_type,iterator_type>,typename SAMPLER::OTYPE> > v;
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           t_sampler.get_lower_bound(it)-threshold(it));
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           t_sampler.get_upper_bound(it)+threshold(it));
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ) {
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            v.emplace_back( start->first, iter->second.build_and_query_ts( focus, sampler, additional... ) );
        }
 
        return cpl_algo.relaxation(std::make_pair(t,it), focus, t_sampler.filter(std::make_pair(t,it), v));
    }
 
    template<typename TYPE, class TIME_SAMPLER, typename ... ADDITIONAL>
    std::vector<point_type>
    fetch_points( const std::string& attr, const time_type t,
                  const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t
        if( fetch_t_hist_ != t && barrier_enabled )
            barrier(t_sampler.get_upper_bound(t));
 
        fetch_t_hist_ = t;
 
        using vec = std::vector<std::pair<point_type,TYPE> >;
        std::vector <point_type> return_points;
 
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ){
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            const vec& ds = iter->second.template return_data<TYPE>();
            return_points.reserve(ds.size());
            for( size_t i=0; i<ds.size(); i++ ) {
                return_points.emplace_back(ds[i].first);
            }
        }
 
        return return_points;
    }
 
    template<typename TYPE, class TIME_SAMPLER, typename ... ADDITIONAL>
    std::vector<point_type>
    fetch_points( const std::string& attr, const time_type t, const iterator_type it,
                  const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t,iteration=it
        if( fetch_t_hist_ != t && fetch_i_hist_ != it && barrier_enabled)
            barrier(t_sampler.get_upper_bound(t),t_sampler.get_upper_bound(it));
 
        fetch_t_hist_ = t;
        fetch_i_hist_ = it;
 
        using vec = std::vector<std::pair<point_type,TYPE> >;
        std::vector <point_type> return_points;
 
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           t_sampler.get_lower_bound(it)-threshold(it));
 
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           t_sampler.get_upper_bound(it)+threshold(it));
 
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ){
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            const vec& ds = iter->second.template return_data<TYPE>();
            return_points.reserve(ds.size());
            for( size_t i=0; i<ds.size(); i++ ) {
                return_points.emplace_back(ds[i].first);
            }
        }
 
        return return_points;
    }
 
    template<typename TYPE, class TIME_SAMPLER, typename ... ADDITIONAL>
    std::vector<TYPE>
    fetch_values( const std::string& attr, const time_type t,
                  const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t,iteration=it
        if( fetch_t_hist_ != t && barrier_enabled )
            barrier(t_sampler.get_upper_bound(t));
 
        fetch_t_hist_ = t;
 
        using vec = std::vector<std::pair<point_type,TYPE> >;
        std::vector<TYPE> return_values;
 
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           std::numeric_limits<iterator_type>::lowest());
 
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ){
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            const vec& ds = iter->second.template return_data<TYPE>();
            return_values.reserve(ds.size());
            for( size_t i=0; i<ds.size(); i++ ) {
                return_values.emplace_back(ds[i].second);
            }
        }
 
        return return_values;
    }
 
    template<typename TYPE, class TIME_SAMPLER, typename ... ADDITIONAL>
    std::vector<TYPE>
    fetch_values( const std::string& attr, const time_type t, const iterator_type it,
                  const TIME_SAMPLER &t_sampler, bool barrier_enabled = true, ADDITIONAL && ... additional ) {
        // Only enter barrier on first fetch for time=t,iteration=it
        if( fetch_t_hist_ != t && fetch_i_hist_ != it && barrier_enabled)
            barrier(t_sampler.get_upper_bound(t),t_sampler.get_upper_bound(it));
 
        fetch_t_hist_ = t;
        fetch_i_hist_ = it;
 
        using vec = std::vector<std::pair<point_type,TYPE> >;
        std::vector<TYPE> return_values;
 
        std::pair<time_type,iterator_type> curr_time_lower(t_sampler.get_lower_bound(t)-threshold(t),
                                                           t_sampler.get_lower_bound(it)-threshold(it));
        std::pair<time_type,iterator_type> curr_time_upper(t_sampler.get_upper_bound(t)+threshold(t),
                                                           t_sampler.get_upper_bound(it)+threshold(it));
 
        auto end = log.upper_bound(curr_time_upper);
 
        if( log.size() == 1 ) end = log.end();
 
        for( auto start = log.lower_bound(curr_time_lower); start != end; ++start ){
            const auto& iter = start->second.find(attr);
            if( iter == start->second.end() ) continue;
            const vec& ds = iter->second.template return_data<TYPE>();
            return_values.reserve(ds.size());
            for( size_t i=0; i<ds.size(); i++ ) {
                return_values.emplace_back(ds[i].second);
            }
        }
 
        return return_values;
    }
 
    int commit( time_type t, iterator_type it = std::numeric_limits<iterator_type>::lowest() ) {
        std::pair<time_type, iterator_type> time(t, it);
 
        // Check Smart Send if announcement made
        if ( !smart_send_set_ ) {
            // Reset all peers to default of enabled
            std::fill(peer_is_sending.begin(), peer_is_sending.end(), true);
            update_smart_send(t);
            smart_send_set_ = true;
        }
 
        if( FIXEDPOINTS ) {
            // This only happens during the first commit
            if( push_buffer_pts.size() > 0 ) {
                comm->send( message::make("points",comm->local_rank(),std::move(push_buffer_pts)),peer_is_sending );
                initialized_pts_ = true;
                push_buffer_pts.clear();
            }
 
            // Reset counter for flat point structure
            fixedPointCount_ = 0;
 
            if( push_buffer_raw.size() > 0 ) {
                comm->send( message::make("rawdata",comm->local_rank(),time,std::move(push_buffer_raw)),peer_is_sending );
                push_buffer_raw.clear();
            }
        }
        else {
            if( push_buffer.size() > 0 ) {
                comm->send( message::make("data",time,std::move(push_buffer)),peer_is_sending );
                push_buffer.clear();
            }
        }
 
        comm->send( message::make("timestamp",comm->local_rank(),time),peer_is_sending );
 
        return std::count( peer_is_sending.begin(),peer_is_sending.end(),true );
    }
 
    void update_smart_send( time_type t ) {
        if( (((span_start < t) || almost_equal(span_start, t)) &&
            ((t < span_timeout) || almost_equal(t, span_timeout))) ) {
            for( size_t i=0; i < peers.size(); i++ ) {
                // Check if peer is explicitly disabled
                if( peers[i].is_recv_disabled() ) {
                    peer_is_sending[i] = false;
                    continue;
                }
 
                // Perform geometric check against defined regions
                peer_is_sending[i] = peers[i].is_recving( t, current_span );
            }
      }
      else { // Ensure explicitly disabled peers are taken into account if outside Smart Send time bounds
          for( size_t i=0; i < peers.size(); i++ ) {
              if( peers[i].is_recv_disabled() ) peer_is_sending[i] = false;
          }
      }
    }
 
    void forecast( time_type t, iterator_type it = std::numeric_limits<iterator_type>::lowest()) {
        std::pair<time_type,iterator_type> time(t,it);
        comm->send(message::make("forecast", comm->local_rank(), time));
    }
 
    bool is_ready( const std::string& attr, time_type t ) const {
        using logitem_ref_t = typename decltype(log)::const_reference;
        return std::any_of(log.begin(), log.end(), [=](logitem_ref_t time_frame) {
            return time_frame.second.find(attr) != time_frame.second.end(); }) // return false for attributes that don't exist.
            && std::all_of(peers.begin(), peers.end(), [=](const peer_state& p) {
            return (p.is_send_disabled()) || (!p.is_sending(t, recv_span)) ||
                   ((((p.current_t() > t) || almost_equal(p.current_t(), t)) || (p.next_t() > t))); });
    }
 
    bool is_ready( const std::string& attr, time_type t, iterator_type it ) const {
        using logitem_ref_t = typename decltype(log)::const_reference;
        return std::any_of(log.begin(), log.end(), [=](logitem_ref_t time_frame) {
            return time_frame.second.find(attr) != time_frame.second.end(); }) // return false for attributes that don't exist.
            && std::all_of(peers.begin(), peers.end(), [=](const peer_state& p) {
            return (p.is_send_disabled()) || (!p.is_sending(t, recv_span)) ||
                   ((((p.current_t() > t) || almost_equal(p.current_t(), t)) || (p.next_t() > t)) &&
                    (((p.current_it() > it) || almost_equal(p.current_it(), it)) || (p.current_it() > it))); });
    }
 
    void barrier( time_type t ) {
        // barrier must be thread-safe because it is called in fetch()
        std::lock_guard<std::mutex> lock(mutex);
 
        auto start = std::chrono::system_clock::now();
 
        for(;;) {
            size_t peers_unblocked = 0;
            for( size_t p = 0; p < peers.size(); p++ ) {
                if( peers[p].is_send_disabled() ) { peers_unblocked++; continue; } // Rank disabled, immediate break
                if( !peers[p].is_sending(t, recv_span) ) { peers_unblocked++; continue; } // Rank disabled due to Smart Send geometry check
                if( (peers[p].current_t() > t || almost_equal(peers[p].current_t(), t)) || peers[p].next_t() > t ) { // Final time check
                    peers_unblocked++;
                    continue;
                }
            }
            // All peers unblocked, break loop
            if( peers_unblocked == peers.size() )
                break;
            else // Acquire messages
                acquire();
        }
 
        if( !QUIET ) {
            if( (std::chrono::system_clock::now() - start) > std::chrono::seconds(5) ) {
                std::cout << "MUI Warning [uniface.h]: Communication barrier spent over 5 seconds" << std::endl;
            }
        }
    }
 
    void barrier( time_type t, iterator_type it ) {
        // barrier must be thread-safe because it is called in fetch()
        std::lock_guard<std::mutex> lock(mutex);
 
        auto start = std::chrono::system_clock::now();
 
        for(;;) {
            size_t peers_unblocked = 0;
            for( size_t p = 0; p < peers.size(); p++ ) {
                if( peers[p].is_send_disabled() ) { peers_unblocked++; continue; } // Rank disabled, immediate break
                if( !peers[p].is_sending(t, recv_span) ) { peers_unblocked++; continue; } // Rank disabled due to Smart Send geometry check
                if( ((peers[p].current_t() > t || almost_equal(peers[p].current_t(), t)) || peers[p].next_t() > t) && // Final time check
                    ((peers[p].current_it() > it || almost_equal(peers[p].current_it(), it)) || peers[p].next_it() > it) ) {
                    peers_unblocked++;
                    continue;
                }
            }
            // All peers unblocked, break loop
            if( peers_unblocked == peers.size() )
                break;
            else // Acquire messages
                acquire();
        }
 
        if( !QUIET ) {
            if( (std::chrono::system_clock::now() - start) > std::chrono::seconds(5) ) {
                std::cout << "MUI Warning [uniface.h]: Communication barrier spent over 5 seconds" << std::endl;
            }
        }
    }
 
    void barrier_ss_send( ) {
        // barrier must be thread-safe because it is called in fetch()
        std::lock_guard<std::mutex> lock(mutex);
 
        auto start = std::chrono::system_clock::now();
 
        for(;;) {    // barrier must be thread-safe because it is called in fetch()
            if( std::all_of(peers.begin(), peers.end(), [=](const peer_state& p) {
                return (p.ss_send_status()); }) ) break;
            acquire(); // To avoid infinite-loop when synchronous communication
        }
 
        for(size_t i=0; i<peers.size(); i++) {
            peers[i].set_ss_send_status(false);
        }
 
        if( !QUIET ) {
            if( (std::chrono::system_clock::now() - start) > std::chrono::seconds(5) ) {
                std::cout << "MUI Warning [uniface.h]: Smart Send communication barrier spent over 5 seconds" << std::endl;
            }
        }
    }
 
    void barrier_ss_recv( ) {
        // barrier must be thread-safe because it is called in fetch()
        std::lock_guard<std::mutex> lock(mutex);
 
        auto start = std::chrono::system_clock::now();
 
        for(;;) {    // barrier must be thread-safe because it is called in fetch()
            if( std::all_of(peers.begin(), peers.end(), [=](const peer_state& p) {
                return (p.ss_recv_status()); }) ) break;
            acquire(); // To avoid infinite-loop when synchronous communication
        }
 
        for(size_t i=0; i<peers.size(); i++) {
            peers[i].set_ss_recv_status(false);
        }
 
        if( (std::chrono::system_clock::now() - start) > std::chrono::seconds(5) ) {
            if( !QUIET )
                std::cout << "MUI Warning [uniface.h]: Smart Send communication barrier spent over 5 seconds" << std::endl;
        }
    }
 
    void announce_send_span( time_type start, time_type timeout, span_t s, bool synchronised = false) {
        span_start = start;
        span_timeout = timeout;
        current_span.swap(s);
        comm->send(message::make("sendingSpan", comm->local_rank(), start, timeout, std::move(current_span)));
        if( synchronised ) barrier_ss_send();
        smart_send_set_ = false;
    }
 
    void announce_send_disable( bool synchronised = false ) {
        comm->send(message::make("sendingDisable", comm->local_rank()));
        if( synchronised ) barrier_ss_send();
    }
 
    void announce_recv_span( time_type start, time_type timeout, span_t s, bool synchronised = false ) {
        recv_start = start;
        recv_timeout = timeout;
        recv_span.swap(s);
        comm->send(message::make("receivingSpan", comm->local_rank(), start, timeout, std::move(recv_span)));
        if( synchronised ) barrier_ss_recv();
        smart_send_set_ = false;
    }
 
    void announce_recv_disable( bool synchronised = false ) {
        comm->send(message::make("receivingDisable", comm->local_rank()));
        if( synchronised ) barrier_ss_recv();
    }
 
    void forget( time_type last, bool reset_log = false ) {
        std::pair<time_type,iterator_type> upper_limit(last+threshold(last),
                                                       std::numeric_limits<iterator_type>::lowest());
 
        log.erase(log.begin(), log.upper_bound(upper_limit));
 
        if( reset_log ) {
            std::pair<time_type,iterator_type> curr_time(std::numeric_limits<time_type>::lowest(),
                                                         std::numeric_limits<iterator_type>::lowest());
 
            if( !log.empty() ) curr_time = log.rbegin()->first;
 
            for( size_t i=0; i < peers.size(); i++ ) {
                peers[i].set_current_t(curr_time.first);
                peers[i].set_current_sub(curr_time.second);
            }
        }
 
        fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
    }
 
    void forget( std::pair<time_type,iterator_type> last, bool reset_log = false ) {
        std::pair<time_type,iterator_type> upper_limit(last.first+threshold(last.first),
                                                   last.second+threshold(last.second));
 
        log.erase(log.begin(), log.upper_bound(upper_limit));
 
        if( reset_log ) {
            std::pair<time_type,iterator_type> curr_time(std::numeric_limits<time_type>::lowest(),
                                                         std::numeric_limits<iterator_type>::lowest());
 
            if( !log.empty() ) curr_time = log.rbegin()->first;
 
            for( size_t i=0; i < peers.size(); i++ ) {
                peers[i].set_current_t(curr_time.first);
                peers[i].set_current_sub(curr_time.second);
            }
        }
 
        fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
        fetch_i_hist_ = std::numeric_limits<iterator_type>::lowest();
    }
 
    void forget( time_type first, time_type last, bool reset_log = false ) {
        std::pair<time_type,iterator_type> lower_limit(first-threshold(first),
                                                       std::numeric_limits<iterator_type>::lowest());
        std::pair<time_type,iterator_type> upper_limit(last+threshold(last),
                                                       std::numeric_limits<iterator_type>::lowest());
 
        log.erase(log.lower_bound(lower_limit), log.upper_bound(upper_limit));
 
        if( reset_log ) {
            std::pair<time_type,iterator_type> curr_time(std::numeric_limits<time_type>::lowest(),
                                                         std::numeric_limits<iterator_type>::lowest());
 
            if( !log.empty() ) curr_time = log.rbegin()->first;
 
            for( size_t i=0; i < peers.size(); i++ ) {
                peers[i].set_current_t(curr_time.first);
                peers[i].set_current_sub(curr_time.second);
            }
        }
 
        fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
    }
 
    void forget( std::pair<time_type,iterator_type> first, std::pair<time_type,iterator_type> last, bool reset_log = false ) {
        std::pair<time_type,iterator_type> lower_limit(first.first-threshold(first.first),
                                                       first.second-threshold(first.second));
        std::pair<time_type,iterator_type> upper_limit(last.first+threshold(last.first),
                                                       last.second+threshold(last.second));
 
        log.erase(log.lower_bound(lower_limit), log.upper_bound(upper_limit));
 
        if( reset_log ) {
            std::pair<time_type,iterator_type> curr_time(std::numeric_limits<time_type>::lowest(),
                                                         std::numeric_limits<iterator_type>::lowest());
 
            if( !log.empty() ) curr_time = log.rbegin()->first;
 
            for( size_t i=0; i<peers.size(); i++ ) {
                peers[i].set_current_t(curr_time.first);
                peers[i].set_current_sub(curr_time.second);
            }
        }
 
        fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
        fetch_i_hist_ = std::numeric_limits<iterator_type>::lowest();
    }
 
    void set_memory( time_type length ) {
        memory_length = length;
 
        fetch_t_hist_ = std::numeric_limits<time_type>::lowest();
        fetch_i_hist_ = std::numeric_limits<iterator_type>::lowest();
    }
    
    std::string uri_host() {
        return comm->uri_host();
    }
 
    std::string uri_path() {
        return comm->uri_path();
    }
 
    std::string uri_protocol() {
        return comm->uri_protocol();
    }
 
private:
    void acquire() {
        message m = comm->recv();
        if( m.has_id() ) readers[m.id()](m);
    }
 
    void on_recv_confirm( int32_t sender, std::pair<time_type,iterator_type> timestamp ) {
        peers[sender].set_current_t(timestamp.first);
        peers[sender].set_current_sub(timestamp.second);
    }
 
    void on_recv_forecast( int32_t sender, std::pair<time_type,iterator_type> timestamp ) {
        peers[sender].set_next_t(timestamp.first);
        peers[sender].set_next_sub(timestamp.second);
    }
 
    void on_recv_data( std::pair<time_type,iterator_type> timestamp, frame_type frame ) {
        auto itr = log.find(timestamp);
 
        if( itr == log.end() )
            std::tie(itr,std::ignore) = log.insert(std::make_pair(timestamp,bin_frame_type()));
 
        auto& cur = itr->second;
 
        for( auto& p: frame ){
            auto pstr = cur.find(p.first);
            if( pstr == cur.end() ) cur.insert(std::make_pair(std::move(p.first),spatial_t(std::move(p.second))));
            else pstr->second.insert(p.second);
        }
 
        log.erase(log.begin(), log.upper_bound({timestamp.first-memory_length, timestamp.second}));
    }
 
    void on_recv_rawdata( int32_t sender, std::pair<time_type,iterator_type> timestamp, frame_raw_type frame ) {
        on_recv_data( timestamp, associate( sender, frame ) );
    }
 
    void on_recv_span( int32_t sender, time_type start, time_type timeout, span_t s ) {
        peers[sender].set_recving(start,timeout,std::move(s));
        peers[sender].set_ss_recv_status(true);
    }
 
    void on_send_span( int32_t sender, time_type start, time_type timeout, span_t s ) {
        peers[sender].set_sending(start,timeout,std::move(s));
        peers[sender].set_ss_send_status(true);
    }
 
    void on_recv_disable( int32_t sender ) {
        peers[sender].set_recv_disable();
        peers[sender].set_ss_recv_status(true);
        peer_is_sending[sender] = false;
    }
 
    void on_send_disable( int32_t sender ) {
        peers[sender].set_send_disable();
        peers[sender].set_ss_send_status(true);
    }
 
    void on_recv_points( int32_t sender, std::vector<point_type> points ) {
        peers[sender].set_pts(points);
    }
 
    void on_recv_assignedVals( std::string attr, storage_single_t data ) {
        typename std::unordered_map<std::string, storage_single_t >::iterator it = assigned_values.find(attr);
        if (it != assigned_values.end())
            it->second = data;
        else 
            assigned_values.insert( std::pair<std::string, storage_single_t>( attr, data ) );
    }
 
    inline frame_type associate( int32_t sender, frame_raw_type& frame ) {
        frame_type buf;
        const auto& pts = peers[sender].pts();
 
        for( auto& p: frame ) {
            const auto& data = storage_cast<const std::vector<std::pair<size_t,REAL> >&>(p.second);
 
            buf.insert(std::make_pair(p.first, storage_t(std::vector<std::pair<point_type,REAL> >())));
            std::vector<std::pair<point_type,REAL> >& data_store = storage_cast<std::vector<std::pair<point_type,REAL> >&>(buf[p.first]);
 
            data_store.resize(data.size());
 
            for( size_t i=0; i<data.size(); i++ ) {
                data_store[i].first = pts[data[i].first];
                data_store[i].second = data[i].second;
            }
        }
 
        return buf;
    }
};
 
}
 
#endif // _UNIFACE_H