#include <hpx/hpx_init.hpp>
#include <hpx/hpx.hpp>
#include <hpx/include/iostreams.hpp>
#include <hpx/runtime/actions/plain_action.hpp>
#include <hpx/include/components.hpp>

#include <hpx/runtime/serialization/serialize.hpp>
#include <boost/shared_array.hpp>

#include <string>

struct cell {
    cell()  {p=0;}

    cell(double a) {p=a;}

    double p;
};

template<typename T>
class array_deleter {
    public:
        void operator()(T const* p)
        {
            delete [] p;
        }
};

struct partition_data
{
private:
    typedef hpx::serialization::serialize_buffer<cell> buffer_type;

public:
    double c;
    partition_data()
    : size_x_(0),
      size_y_(0),
      size_(0)
    {}

    partition_data(std::size_t size_x, std::size_t size_y)
    : data_(new cell [size_x * size_y], size_x * size_y, buffer_type::take, array_deleter<cell>()),
      size_x_(size_x),
      size_y_(size_y),
      size_(size_x * size_y)
    {
        HPX_ASSERT(size_x >= size_y);
    }

    partition_data(std::size_t size_x, std::size_t size_y, double initial_value)
    : data_(new cell [size_x * size_y], size_x * size_y, buffer_type::take, array_deleter<cell>()),
      size_x_(size_x),
      size_y_(size_y),
      size_(size_x * size_y)
    {
        HPX_ASSERT(size_x >= size_y);

        for(std::size_t i = 0; i < size_; ++i)
            data_[i] = cell(initial_value);
    }

    partition_data(partition_data const& base)
    {

        data_ = buffer_type(base.data_.data(), base.size(), buffer_type::reference);
        size_x_ = base.size_x();
        size_y_ = base.size_y();
        size_ = base.size();
    }

    std::size_t size_x() const { return size_x_;}
    std::size_t size_y() const { return size_y_;}
    std::size_t size() const { return size_;}

    cell get_cell(std::size_t idx, std::size_t idy) const { return data_[index(idx, idy)];}
    cell& get_cell_ref(std::size_t idx, std::size_t idy) { return data_[index(idx, idy)];}

    cell operator[](std::size_t idx) const { return data_[idx];}
    cell& operator[](std::size_t idx) { return data_[idx];}

private:
    // Serialization support: even if all of the code below runs on one
    // locality only, we need to provide an (empty) implementation for the
    // serialization as all arguments passed to actions have to support this.
    friend class hpx::serialization::access;

    template <typename Archive>
    void serialize(Archive& ar, const unsigned int version)
    {
        ar & data_ & size_x_ & size_y_ & size_;
    }

    std::size_t index(std::size_t idx, std::size_t idy) const
    {
        std::size_t id = idy*size_x_ + idx;
        HPX_ASSERT(id >= 0 && id < size_);
        return id;
    }

private:
    buffer_type data_;
    std::size_t size_x_;
    std::size_t size_y_;
    std::size_t size_;
};

struct HPX_COMPONENT_EXPORT partition_server
    : hpx::components::component_base<partition_server>
{
    partition_server() {}

    partition_server(partition_data const& data)
    : data_(data)
    {}

    partition_server(std::size_t size_x, std::size_t size_y, double initial_value)
    : data_(size_x, size_y, initial_value)
    {}

    partition_data get_data() const
    {
        hpx::cout << "invoked get_data  on " << hpx::get_locality_id()
        << " getting size " << data_.size() << hpx::endl << hpx::flush;
        return partition_data(data_);
    }

    HPX_DEFINE_COMPONENT_DIRECT_ACTION(partition_server, get_data, get_data_action);

private:
    partition_data data_;
};

typedef hpx::components::component<partition_server> partition_server_type;

HPX_REGISTER_COMPONENT(partition_server_type, partition_server);

HPX_REGISTER_ACTION(partition_server::get_data_action);

struct partition
    : hpx::components::client_base<partition, partition_server>
{
    typedef hpx::components::client_base<partition, partition_server> base_type;

    partition() {}

    partition(hpx::id_type where, std::size_t size_x, std::size_t size_y, double initial_value)
        : base_type(hpx::new_<partition_server>(where, size_x, size_y, initial_value))
    {}

    // Create a new component on the locality co-located to the id 'where'. The
    // new instance will be initialized from the given partition_data.
    partition(hpx::id_type where, partition_data const& data)
      : base_type(hpx::new_<partition_server>(hpx::colocated(where), data))
    {}

    // Attach a future representing a (possibly remote) partition.
    partition(hpx::future<hpx::id_type> && id)
      : base_type(std::move(id))
    {}

    // Unwrap a future<partition> (a partition already holds a future to the
    // id of the referenced object, thus unwrapping accesses this inner future).
    partition(hpx::future<partition> && c)
      : base_type(std::move(c))
    {}

    hpx::future<partition_data> get_data() const
    {
        partition_server::get_data_action act;
        return hpx::async(act, get_id());
    }
};

int test(partition part) {
    hpx::cout << "received on " << hpx::get_locality_id() << ": " << hpx::endl <<  hpx::flush;
    partition_data pdata = part.get_data().get();
    hpx::cout << " pdata " << pdata[0].p << hpx::endl << hpx::flush;
    return 1;
}

HPX_PLAIN_ACTION(test, test_act);

int hpx_main(int argc, char* argv[])
{
    std::vector<hpx::id_type> localities = hpx::find_all_localities();
    std::size_t num_localities = localities.size();

    partition part = partition(hpx::find_here(), 2, 2, hpx::get_locality_id());
    test_act act;
    hpx::cout << "num locs " << num_localities << hpx::endl << hpx::flush;

    std::vector<hpx::future<int> > futures;

    partition_data pdata = part.get_data().get();

     for (int i = 0; i<num_localities; i++){
            if(localities[i] != hpx::find_here())
            {
                hpx::cout << "applying from " << hpx::get_locality_id() <<" on " << localities[i] << hpx::endl << hpx::flush;
                futures.push_back(hpx::async(act, localities[i], part));
            }
    }

    hpx::wait_all(futures);
    hpx::cout << "done" << hpx::endl << hpx::flush;

    return hpx::finalize();
}

int main(int argc, char* argv[])
{
    // Initialize and run HPX, this executes hpx_main above.
    return hpx::init(argc, argv);
}