replicator.h

Go to the documentation of this file.

/*
 *  This file is a part of TiledArray.
 *  Copyright (C) 2013  Virginia Tech
 *
 *  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 TILEDARRAY_REPLICATOR_H__INCLUDED
#define TILEDARRAY_REPLICATOR_H__INCLUDED

#include <TiledArray/madness.h>

namespace TiledArray {
  namespace detail {


    template <typename A>
    class Replicator : public madness::WorldObject<Replicator<A> >, private madness::Spinlock {
    private:
      typedef Replicator<A> Replicator_; 
      typedef madness::WorldObject<Replicator_> wobj_type; 
      typedef std::stack<madness::CallbackInterface*, std::vector<madness::CallbackInterface*> > callback_type; 

      A destination_; 
      std::vector<typename A::size_type> indices_; 
      std::vector<Future<typename A::value_type> > data_; 
      madness::AtomicInt sent_; 
      World& world_;
      volatile callback_type callbacks_; 
      volatile mutable bool probe_; 

      void do_callbacks() {
        callback_type& callbacks = const_cast<callback_type&>(callbacks_);
        while(! callbacks.empty()) {
          callbacks.top()->notify();
          callbacks.pop();
        }
      }

      class DelaySend : public madness::TaskInterface {
      private:
        Replicator_& parent_; 

      public:

        DelaySend(Replicator_& parent) :
          madness::TaskInterface(madness::TaskAttributes::hipri()),
          parent_(parent)
        {
          typename std::vector<Future<typename A::value_type> >::iterator it =
              parent_.data_.begin();
          typename std::vector<Future<typename A::value_type> >::iterator end =
              parent_.data_.end();
          for(; it != end; ++it) {
            if(! it->probe()) {
              madness::DependencyInterface::inc();
              it->register_callback(this);
            }
          }
        }

        virtual ~DelaySend() { }

        virtual void run(const madness::TaskThreadEnv&) { parent_.send(); }

      }; // class DelaySend


      bool probe() const {
        madness::ScopedMutex<madness::Spinlock> locker(this);

        if(! probe_) {
          typename std::vector<Future<typename A::value_type> >::const_iterator it =
              data_.begin();
          typename std::vector<Future<typename A::value_type> >::const_iterator end =
              data_.end();
          for(; it != end; ++it)
            if(! it->probe())
              break;

          probe_ = (it == end);
        }

        return probe_;
      }

      void delay_send() {
        if(probe()) {
          // The data is ready so send it now.
          send(); // Replication is done
        } else {
          // The local data is not ready to be sent, so create a task that will
          // send it when it is ready.
          DelaySend* delay_send_task = new DelaySend(*this);
          world_.taskq.add(delay_send_task);
        }
      }

      void send() {
        const long sent = ++sent_;
        const ProcessID dest = (world_.rank() + sent) % world_.size();

        if(dest != world_.rank()) {
          wobj_type::task(dest, &  Replicator_::send_handler, indices_, data_,
              madness::TaskAttributes::hipri());
        } else
          do_callbacks(); // Replication is done
      }

      void send_handler(const std::vector<typename A::size_type>& indices,
          const std::vector<Future<typename A::value_type> >& data)
      {
        typename std::vector<typename A::size_type>::const_iterator index_it =
            indices.begin();
        typename std::vector<Future<typename A::value_type> >::const_iterator data_it =
            data.begin();
        typename std::vector<Future<typename A::value_type> >::const_iterator data_end =
            data.end();

        for(; data_it != data_end; ++data_it, ++index_it)
          destination_.set(*index_it, data_it->get());

        delay_send();
      }

    public:

      Replicator(const A& source, const A destination) :
        wobj_type(source.world()), madness::Spinlock(),
        destination_(destination), indices_(), data_(), sent_(),
        world_(source.world()), callbacks_(), probe_(false)
      {
        sent_ = 0;

        // Generate a list of local tiles from other.
        typename A::pmap_interface::const_iterator end = source.pmap()->end();
        typename A::pmap_interface::const_iterator it = source.pmap()->begin();
        indices_.reserve(source.pmap()->local_size());
        data_.reserve(source.pmap()->local_size());
        if(source.is_dense()) {
          // When dense, all tiles are present
          for(; it != end; ++it) {
            indices_.push_back(*it);
            data_.push_back(source.find(*it));
            destination_.set(*it, data_.back());
          }
        } else {
          // When sparse, we need to generate a list
          for(; it != end; ++it)
            if(! source.is_zero(*it)) {
              indices_.push_back(*it);
              data_.push_back(source.find(*it));
              destination_.set(*it, data_.back());
            }
        }

        delay_send();

        // Process any pending messages
        wobj_type::process_pending();
      }


      bool done() {
        madness::ScopedMutex<madness::Spinlock> locker(this);
        return sent_ == world_.size();
      }



      void register_callback(madness::CallbackInterface* callback) {
          madness::ScopedMutex<madness::Spinlock> locker(this);
          if(sent_ == world_.size())
            callback->notify();
          else
            const_cast<callback_type&>(callbacks_).push(callback);
      }

    }; // class Replicator

  }  // namespace detail
}  // namespace TiledArray


#endif // TILEDARRAY_REPLICATOR_H__INCLUDED