Function load_behaviour_graph_from_cache

example/spread.cc:547–645

template <unsigned sum_items_num, template <typename, typename Enable = void> class OBJECT_BUFFER>
plato::sparse_state_t<behaviour_state_t<sum_items_num>, plato::empty_t> load_behaviour_graph_from_cache(
  OBJECT_BUFFER<plato::vertex_unit_t<behaviour_state_content_t<sum_items_num>>>&& behaviour_cache,
  plato::graph_info_t& behaviour_graph_info, OBJECT_BUFFER<degree_unit>&& behaviour_degree_cache) {
  auto& cluster_info = plato::cluster_info_t::get_instance();

  plato::stop_watch_t watch;
  watch.mark("t0");
  watch.mark("t1");

  constexpr size_t mem_size = sizeof(plato::vid_t) * (size_t(std::numeric_limits<plato::vid_t>::max()) + 1);
  std::unique_ptr<plato::vid_t, plato::mmap_deleter> out_degree(
    (plato::vid_t*)mmap(nullptr, mem_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_NORESERVE, -1, 0),
    plato::mmap_deleter{mem_size});
  CHECK(MAP_FAILED != out_degree.get())
  << boost::format("WARNING: mmap failed, err code: %d, err msg: %s") % errno % strerror(errno);

  size_t local_degree_sum = 0;
  plato::vid_t local_vertices = 0;
  plato::vid_t max_vid = 0;

  behaviour_degree_cache.reset_traversal();
  #pragma omp parallel reduction(+:local_degree_sum) reduction(+:local_vertices) reduction(max:max_vid)
  {
    size_t chunk_size = 1;
    while (behaviour_degree_cache.next_chunk([&] (size_t, degree_unit* v) {
      out_degree.get()[v->vid_] = v->degree_;
      local_degree_sum += v->degree_;
      local_vertices++;
      max_vid = std::max(max_vid, v->vid_);
    }, &chunk_size)) {}
  }

  size_t degree_sum = 0;
  MPI_Allreduce(MPI_IN_PLACE, &max_vid, 1, plato::get_mpi_data_type<plato::vid_t>(), MPI_MAX, MPI_COMM_WORLD);
  MPI_Allreduce(&local_degree_sum, &degree_sum, 1, plato::get_mpi_data_type<size_t>(), MPI_SUM, MPI_COMM_WORLD);
  CHECK(behaviour_graph_info.edges_ == degree_sum);
  CHECK(behaviour_graph_info.max_v_i_ == max_vid);

  plato::sparse_state_t<behaviour_state_t<sum_items_num>, plato::empty_t> behaviour(local_vertices * 1.2, std::make_shared<plato::empty_t>());

  {
    watch.mark("t1");

    behaviour_mmap_p.reset(
      mmap(nullptr, sizeof(behaviour_state_t<sum_items_num>) * local_degree_sum, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_NORESERVE, -1, 0),
      plato::mmap_deleter{sizeof(behaviour_state_t<sum_items_num>) * local_degree_sum});
    CHECK(MAP_FAILED != behaviour_mmap_p.get())
    << boost::format("WARNING: mmap failed, err code: %d, err msg: %s") % errno % strerror(errno) << " local_degree_sum: " << local_degree_sum;

    behaviour.unlock();
    auto lock_defer = plato::defer([&]{ behaviour.lock(); });

    size_t adjs_num = 0;
    #pragma omp parallel for
    for (plato::vid_t v_i = 0; v_i <= max_vid; ++v_i) {
      plato::vid_t degree = *(out_degree.get() + v_i);
      if (degree) {
        behaviour.upsert(