Function load_behaviour_graph_cache

example/spread.cc:421–544

template <unsigned sum_items_num, template <typename, typename Enable = void> class OBJECT_BUFFER>
void load_behaviour_graph_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, plato::bitmap_t<>& behaviour_bitmap) {

  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);

  plato::vid_t max_vid = 0;
  size_t edges = 0;
  {
    plato::thread_local_counter max_vid_counter;
    std::mutex mutex;
    std::vector<std::string> files = plato::get_files(FLAGS_input_behaviour_edges);

    plato::thread_local_buffer input_buffer;

    auto __send = [&] (plato::bsp_send_callback_t<plato::vertex_unit_t<behaviour_state_content_t<sum_items_num>>> send) {
      while (true) {
        std::string filename;
        {
          std::lock_guard<std::mutex> lock(mutex);
          if (files.empty()) break;
          filename = std::move(files.back());
          files.pop_back();
        }

        plato::with_file(filename, [&] (boost::iostreams::filtering_istream& is) {
          plato::vertex_csv_parser<boost::iostreams::filtering_istream, behaviour_state_content_t<sum_items_num>>(
            is,
            [&] (plato::vertex_unit_t<behaviour_state_content_t<sum_items_num>>* input, size_t size) {
              __sync_fetch_and_add(&edges, size);
              for (size_t i = 0; i < size; ++i) {
                plato::vertex_unit_t<behaviour_state_content_t<sum_items_num>>& v = input[i];
                send(get_behaviour_partition_id(v.vdata_.behaviour_id_), v);
              }
              return true;
            },
            [] (behaviour_state_content_t<sum_items_num>* output, char* s_input) {
              char* pSave  = nullptr;
              char* pLog = s_input;
              char* pToken = strtok_r(pLog, ", \t", &pSave);
              if (nullptr == pToken) {
                LOG(WARNING) << boost::format("can not extract behaviour id from (%s)") % pLog;
                return false;
              }
              output->behaviour_id_ = strtoul(pToken, nullptr, 10);

              for (unsigned i = 0; i < sum_items_num; ++i) {