view release on metacpan or  search on metacpan

src/boost/property_map/parallel/impl/distributed_property_map.ipp  view on Meta::CPAN

  ghost_cells_key_index_type const& key_index 
    = data->ghost_cells->template get<1>();

  std::size_t n = msg.size();
  for (std::size_t i = 0; i < n; ++i) {
    // Search for the ghost cell by key, and project back to the sequence
    iterator position
      = data->ghost_cells->template project<0>(key_index.find(msg[i].first));

    if (position != data->ghost_cells->end())
      const_cast<value_type&>(position->second) = msg[i].second;
  }
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
template<typename Reduce>
void
PBGL_DISTRIB_PMAP::handle_message<Reduce>::
handle_multiput
  (int source, int tag, 
   const std::vector<unsafe_pair<local_key_type, value_type> >& values,
   trigger_receive_context)
{
  using boost::get;

  shared_ptr<data_t> data(data_ptr);
  BOOST_ASSERT(data);

  std::size_t n = values.size();
  for (std::size_t i = 0; i < n; ++i) {
    local_key_type local_key = values[i].first;
    value_type local_value = get(data->storage, local_key);
    detail::maybe_put(data->storage, values[i].first,
                      reduce(values[i].first,
                             local_value,
                             values[i].second));
  }
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
template<typename Reduce>
void
PBGL_DISTRIB_PMAP::handle_message<Reduce>::
setup_triggers(process_group_type& pg)
{
  using boost::graph::parallel::simple_trigger;

  simple_trigger(pg, property_map_put, this, &handle_message::handle_put);
  simple_trigger(pg, property_map_get, this, &handle_message::handle_get);
  simple_trigger(pg, property_map_multiget, this, 
                 &handle_message::handle_multiget);
  simple_trigger(pg, property_map_multiget_reply, this, 
                 &handle_message::handle_multiget_reply);
  simple_trigger(pg, property_map_multiput, this, 
                 &handle_message::handle_multiput);
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void
PBGL_DISTRIB_PMAP
::on_synchronize::operator()()
{
  int stage=0; // we only get called at the start now
  shared_ptr<data_t> data(data_ptr);
  BOOST_ASSERT(data);

  // Determine in which stage backward consistency messages should be sent.
  int backward_stage = -1;
  if (data->model & cm_backward) {
    if (data->model & cm_flush) backward_stage = 1;
    else backward_stage = 0;
  }

  // Flush results in first stage
  if (stage == 0 && data->model & cm_flush)
    data->flush();

  // Backward consistency
  if (stage == backward_stage && !(data->model & (cm_clear | cm_reset)))
    data->refresh_ghost_cells();

  // Optionally clear results
  if (data->model & cm_clear)
    data->clear();

  // Optionally reset results
  if (data->model & cm_reset) {
    if (data->reset) ((*data).*data->reset)();
  }
}


template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void
PBGL_DISTRIB_PMAP::set_consistency_model(int model)
{
  data->model = model;

  bool need_on_synchronize = (model != cm_forward);

  // Backward consistency is a two-stage process.
  if (model & cm_backward) {
    // For backward consistency to work, we absolutely cannot throw
    // away any ghost cells.
    data->max_ghost_cells = 0;
  }

  // attach the on_synchronize handler.
  if (need_on_synchronize)
    data->process_group.replace_on_synchronize_handler(on_synchronize(data));
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void
PBGL_DISTRIB_PMAP::set_max_ghost_cells(std::size_t max_ghost_cells)
{
  if ((data->model & cm_backward) && max_ghost_cells > 0)
      boost::throw_exception(std::runtime_error("distributed_property_map::set_max_ghost_cells: "
                                                "cannot limit ghost-cell usage with a backward "
                                                "consistency model"));

  if (max_ghost_cells == 1)
    // It is not safe to have only 1 ghost cell; the cell() method
    // will fail.
    max_ghost_cells = 2;

  data->max_ghost_cells = max_ghost_cells;
  prune_ghost_cells();
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::clear()
{
  data->clear();
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::data_t::clear()
{
  ghost_cells->clear();
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::reset()
{
  if (data->reset) ((*data).*data->reset)();
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::flush()
{
  data->flush();
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::data_t::refresh_ghost_cells()
{
  using boost::get;

  std::vector<std::vector<key_type> > keys;
  keys.resize(num_processes(process_group));

  // Collect the set of keys for which we will request values
  for (iterator i = ghost_cells->begin(); i != ghost_cells->end(); ++i)
    keys[get(global, i->first).first].push_back(i->first);

  // Send multiget requests to each of the other processors
  typedef typename ProcessGroup::process_size_type process_size_type;
  process_size_type n = num_processes(process_group);
  process_id_type id = process_id(process_group);
  for (process_size_type p = (id + 1) % n ; p != id ; p = (p + 1) % n) {
    if (!keys[p].empty())
      send(process_group, p, property_map_multiget, keys[p]);
  }  
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::data_t::flush()
{
  using boost::get;

  int n = num_processes(process_group);
  std::vector<std::vector<unsafe_pair<local_key_type, value_type> > > values;
  values.resize(n);

  // Collect all of the flushed values
  for (iterator i = ghost_cells->begin(); i != ghost_cells->end(); ++i) {
    std::pair<int, local_key_type> g = get(global, i->first);
    values[g.first].push_back(std::make_pair(g.second, i->second));
  }

  // Transmit flushed values
  for (int p = 0; p < n; ++p) {
    if (!values[p].empty())
      send(process_group, p, property_map_multiput, values[p]);
  }
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
void PBGL_DISTRIB_PMAP::do_synchronize()
{
  if (data->model & cm_backward) {
    synchronize(data->process_group);
    return;
  }

  // Request refreshes of the values of our ghost cells
  data->refresh_ghost_cells();

  // Allows all of the multigets to get to their destinations
  synchronize(data->process_group);

  // Allows all of the multiget responses to get to their destinations
  synchronize(data->process_group);
}

template<typename ProcessGroup, typename GlobalMap, typename StorageMap>
template<typename Resolver>
void PBGL_DISTRIB_PMAP::data_t::do_reset()
{
  Resolver* resolver = get_default_value.template target<Resolver>();
  BOOST_ASSERT(resolver);

  for (iterator i = ghost_cells->begin(); i != ghost_cells->end(); ++i)
    const_cast<value_type&>(i->second) = (*resolver)(i->first);
}

} } // end namespace boost::parallel