WorkGraphPolicy( const graph_type & arg_graph )
    : m_graph(arg_graph)
    , m_queue( view_alloc( "queue" , WithoutInitializing )
             , arg_graph.numRows() * 2 + 2 )
  {
    { // Initialize
      using policy_type = RangePolicy<std::int32_t, execution_space, TagInit>;
      using closure_type = Kokkos::Impl::ParallelFor<self_type, policy_type>;
      const closure_type closure(*this, policy_type(0, m_queue.size()));
      closure.execute();
      execution_space::fence();
    }

    { // execute-after counts
      using policy_type = RangePolicy<std::int32_t, execution_space, TagCount>;
      using closure_type = Kokkos::Impl::ParallelFor<self_type, policy_type>;
      const closure_type closure(*this,policy_type(0,m_graph.entries.size()));
      closure.execute();
      execution_space::fence();
    }

    { // Scheduling ready tasks
      using policy_type = RangePolicy<std::int32_t, execution_space, TagReady>;
      using closure_type = Kokkos::Impl::ParallelFor<self_type, policy_type>;
      const closure_type closure(*this,policy_type(0,m_graph.numRows()));
      closure.execute();
      execution_space::fence();
    }
  }

void merge_values(vid_vdata_vector_type& element, graph_type& graph) {
    for (auto vertex : element) {
        if (graph.contains_vertex(vertex.first)) {
            const graphlab::lvid_type lvid = graph.local_vid(vertex.first);
            graph.l_vertex(lvid).data() = vertex.second;
        }
    }
}

    SynchronousEngine<algorithm_t>::SynchronousEngine(graph_type& graph):
    iteration_counter_(0), max_iterations_(5), graph_(graph) {
        vertex_programs_.resize(graph.num_local_vertices());
		gather_accum_.resize(graph.num_local_vertices());
        has_msg_.resize(graph.num_local_vertices(), 0);
        has_msg_.resize(graph.num_local_vertices(), message_type());
        active_superstep_.resize(graph.num_local_vertices(), 0);
        active_minorstep_.resize(graph.num_local_vertices(), 0);
    }

 graph_gather_apply<Graph,GatherType>::graph_gather_apply(
     graph_type& graph,
     gather_fun_type gather_fun,
     apply_fun_type apply_fun,
     const graphlab_options& opts) :
   gather_fun(gather_fun), apply_fun(apply_fun), rmi(graph.dc(), this), graph(graph),
   threads(opts.get_ncpus()),
   thread_barrier(opts.get_ncpus()),
   gather_exchange(graph.dc(), opts.get_ncpus(), 64 * 1024) { } 

 SynchronousEngine<algorithm_t>::SynchronousEngine(graph_type& graph):
 iteration_counter_(0), max_iterations_(5), graph_(graph) {
     vertex_programs_.resize(graph.num_local_vertices());
     gather_accum_.resize(graph.num_local_vertices());
     has_msg_.resize(graph.num_local_vertices(), 0);
     messages_.resize(graph.num_local_vertices(), message_type());
     active_superstep_.resize(graph.num_local_vertices(), 0);
     active_minorstep_.resize(graph.num_local_vertices(), 0);
     context = new context_type(*this, graph);
     aggregator = new aggregator_type(graph, context);
     cout<<"sync engine init"<<endl;
 }

    void visit_next_component(const graph_type& g, const int starting_vertex_id, Sequence& visited)
    {
        pk::queue<int, graph_type::max_num_of_vertices> q;

        q.push(starting_vertex_id);
        visited[starting_vertex_id] = true;

        while(!q.empty())
        {
            const int v = q.front();
            q.pop();

            component_ids[v] = number_of_components;

            const typename graph_type::adjacency_list& adj_v = g.get_adjacency_list(v);
            for(int i = 0; i < adj_v.size(); ++i)
            {
                const int u = adj_v[i].to;

                if(visited[u])
                    continue;

                q.push(u);
                visited[u] = true;
            }
        }
    }

// Second loader that only a single machine calls and pre-loads cameras.
bool vertex_loader(graph_type& graph, string img_path, vector<CameraParams>& cameras)
{
    // force a "/" at the end of the path
    // make sure to check that the path is non-empty. (you do not
    // want to make the empty path "" the root path "/" )
    string path = img_path;
    if (path.length() > 0 && path[path.length() - 1] != '/') path = path + "/";
    
    vector<string> graph_files;
    string search_prefix;
    graphlab::fs_util::list_files_with_prefix(path, search_prefix, graph_files);
    
    if (graph_files.size() == 0)
        logstream(LOG_WARNING) << "No files found in " << path << std::endl;
    
    // vertex data & id
    graphlab::vertex_id_type vid(-1);
    
    ///////////////////////////////////////////////////////
    // Loop over files
    for(size_t i = 0; i < graph_files.size(); ++i)
    {
        vid = i;
        vertex_data vdata;
        vdata.empty = false;
        vdata.img_path = graph_files[i];
        vdata.features.img_idx = i;
        
        vdata.camera = cameras[i]; // addition to above function.
        
        graph.add_vertex(vid, vdata);
    }
    
    return true;
}

    static bool dfs(
            const graph_type& graph,
            const int vertex_id,
            pk::vector<int, graph_type::max_num_of_vertices>& colors,
            pk::stack<int, graph_type::max_num_of_vertices>& s)
    {
        if(colors[vertex_id] == 1)
            return false;

        if(colors[vertex_id] == 2)
            return true;

        colors[vertex_id] = 1;

        const typename graph_type::adjacency_list& adjacent_edges = graph.get_adjacency_list(vertex_id);
        for(int i = 0; i < adjacent_edges.size(); ++i)
        {
            if(!dfs(graph, adjacent_edges[i].to, colors, s))
                return false;
        }

        colors[vertex_id] = 2;
        s.push(vertex_id);

        return true;
    }

void write_results(char* ofile, graph_type& graph) {
    std::cout << "PageRank: Writing results to: " << ofile << std::endl;
    std::ofstream fout(ofile);
    assert(fout.is_open());

    for(graphlab::lvid_type lvid = 0; lvid < graph.get_local_graph().num_vertices(); ++lvid) {
        if (!graph.l_is_master(lvid)) {
            continue;
        }
        size_t id = graph.global_vid(lvid);
        double val = graph.get_local_graph().vertex(lvid).data();
        fout << id << "\t" << val << std::endl;
    }
    fout.close();
    std::cout << "PageRank: Results written." << std::endl;
}

/////////////////////////////////////////////////////////////////////////
// Load the distributed UAI file
bool line_parser(graph_type& graph, const std::string& filename, const std::string& textline) {
    std::stringstream strm(textline);
    graphlab::vertex_id_type vid;
    vertex_data vdata;
    vdata.dual_contrib = 0.0;
    string type;
    strm >> type;
 
     if(type == "v") { 
      vdata.factor_type = VAR;
      vdata.nvars = 1;
      vdata.cards.resize(1);
      strm>>vid;
      strm >> vdata.cards[0];
      vdata.potentials.resize(vdata.cards[0]);
      //vdata.beliefs.setOnes(vdata.cards[0]);
      //vdata.beliefs /= vdata.cards[0];
      vdata.beliefs.setConstant(vdata.cards[0], 0.5);
      vdata.unary_degree.resize(vdata.cards[0], 0);
      //for(int i=0; i< vdata.cards[0]; i++){
      // strm>>vdata.potentials[i];
      //   vdata.potentials[i] = log10(vdata.potentials[i]);
         
      //   }
     //    vdata.potentials.maxCoeff(&vdata.best_configuration);
      graph.add_vertex(vid,vdata);
    }

void test_contains_edge(graph_type g,
                        vector<typename graph_type::location_type> present_locs,
                        vector<typename graph_type::location_type> absent_locs)
{
    for (auto loc : present_locs) {
        if (!g.contains_edge(loc)) {
            cout << "Test contains_egde failed on " << loc.first << ',' << loc.second << '\n';
            exit(1);
        }
    }
    for (auto loc : absent_locs) {
        if (g.contains_edge(loc)) {
            cout << "Test does not contains_egde failed on " << loc.first << ',' << loc.second << '\n';
            exit(1);
        }
    }
    cout << "Contains edge passed\n";
}

bool line_parser(graph_type& graph, const std::string& filename,
                 const std::string& textline)
{

    std::istringstream ssin(textline);
    graphlab::vertex_id_type vid;
    ssin >> vid;
    int out_nb;
    ssin >> out_nb;
    if (out_nb == 0)
        graph.add_vertex(vid);
    while (out_nb--) {
        graphlab::vertex_id_type other_vid;
        ssin >> other_vid;
        graph.add_edge(vid, other_vid);
    }
    return true;
}

void test_delete_edges(graph_type g,
                       vector<typename graph_type::location_type> kill_edges,
                       vector<weight_type> expected_weights)
{
    for (auto k : kill_edges) {
        g.delete_edge(k);
    }
    test_graph_iterator(g, expected_weights);
    cout << "Delete edges passed\n";
}

/////////////////////////////////////////////////////////////////////////
// Edge Loader (used to read the adjacency list and add edges to the graph)
bool edge_loader(graph_type& graph, const std::string& fname,
                 const std::string& textline)
{
    if ( textline.length() == 0 || textline[0] == '#' )
        return true; // empty or comment line, return
    
    std::stringstream strm(textline);
    graphlab::vertex_id_type vid;
    
    // first entry in the line is a vertex ID
    strm >> vid;
    
    if (opts.verbose > 0)
        logstream(LOG_EMPH) << "Here's the input: "
        << textline << "\n"
        << vid << "\n";
    
    // Line should contain at least 1 more number (degree of node)
    if (!strm.good())
    {
        logstream(LOG_ERROR) << "The following ajacency list line is incomplete(check adj_list standard):\n"
        << textline << std::endl;
        return EXIT_FAILURE;
    }
    
    // second entry is the out-degree
    int outdeg;
    strm >> outdeg;
    
    graphlab::vertex_id_type other_vid;
    for (int i=0; i!=outdeg; ++i)
    {
        // Line should contain more numbers (id of neighbours)
        if (!strm.good())
        {
            logstream(LOG_ERROR) << "The following ajacency list line is incomplete(check adj_list standard):\n"
            << textline << std::endl;
            return EXIT_FAILURE;
        }
        
        strm >> other_vid;
        
        // only add edges in one direction
        if (other_vid < vid)
            continue;
        
        if (opts.verbose > 0)
            logstream(LOG_EMPH) << "Adding edge: (" << vid << "," << other_vid << ")\n";
        
        edge_data edata; edata.empty = false;
        graph.add_edge(vid,other_vid,edata);
    }
    
    return true;
}

void test_edges_at(graph_type g,
                   vector<typename graph_type::location_type> modify_edges,
                   weight_type new_value,
                   vector<weight_type> expected_weights)
{
    for (auto m : modify_edges) {
        g.edge_at(m).weight() = new_value;
    }
    test_graph_iterator(g, expected_weights);
    cout << "Edges at passed\n";
}

bool parse_vertex_line(graph_type &graph, const std::string &file, const std::string &line) {
    if (line.empty() || line[0] == '#') {
        return true;
    }

    char *dst;
    size_t id = strtoul(line.c_str(), &dst, 10);
    if (dst == line.c_str()) return false;

    graph.add_vertex(id);
    return true;
}

    static bool run(const graph_type& graph, output_iterator output)
    {
        pk::vector<int, graph_type::max_num_of_vertices> colors(0, graph.get_num_of_vertices());
        pk::stack<int, graph_type::max_num_of_vertices> s;

        for(int v = 0; v < graph.get_num_of_vertices(); ++v)
        {
            if(colors[v] != 0)
                continue;

            if(!dfs(graph, v, colors, s))
                return false;
        }

        while(!s.empty())
        {
            *output++ = s.top();
            s.pop();
        }

        return true;
    }

    void find_components(const graph_type& g)
    {
        number_of_components = 0;
        pk::vector<bool, graph_type::max_num_of_vertices> visited(false, g.get_num_of_vertices());

        for(int u = 0; u < visited.size(); ++u)
        {
            if(!visited[u])
            {
                visit_next_component(g, u, visited);
                ++number_of_components;
            }
        }
    }

bool parse_edge_line(graph_type &graph, const std::string &file, const std::string &line) {
    if (line.empty() || line[0] == '#') {
        return true;
    }

    char *dst;
    size_t source = strtoul(line.c_str(), &dst, 10);
    if (dst == line.c_str()) return false;

    char *end;
    size_t target = strtoul(dst, &end, 10);
    if (dst == end) return false;

    if (source != target) graph.add_edge(source, target);
    return true;
}

// Second Graph loader that only a single machine calls and pre-loads cameras.
// It adds only the selected edges to the subgraph of the fully connected graph.
// (Used in stitch_full_main)
bool graph_loader(graph_type& graph, string img_dir, vector<CameraParams>& cameras, 
                  vector<string>& img_path, vector<int> indices, vector<MatchesInfo>& pairwise_matches)
{
    // force a "/" at the end of the path
    // make sure to check that the path is non-empty. (you do not
    // want to make the empty path "" the root path "/" )
    string path = img_dir;
    if (path.length() > 0 && path[path.length() - 1] != '/') path = path + "/";
    
    // vertex data & id   
    graphlab::vertex_id_type vid(-1);
    graphlab::vertex_id_type other_vid;
    
    if (opts.verbose > 2)
        logstream(LOG_EMPH)
        << "Total number of vertices in the second graph: " << indices.size() << "\n";
    
    ///////////////////////////////////////////////////////
    // Loop over files
    for(size_t i = 0; i < indices.size(); ++i)
    {
        vid = i;
        vertex_data vdata;
        vdata.empty = false;
        vdata.img_path = img_path[i];
        vdata.features.img_idx = i;
        
        vdata.camera = cameras[i]; // addition to above function.
        
        graph.add_vertex(vid, vdata);
    }
    
    // Adding edges between selected pair of vertices to create a subgraph of the fully connected graph
    // no duplicate edges are added
    
    if (opts.verbose > 2)
        logstream(LOG_EMPH)
        << "Pairwise_matches size : " << pairwise_matches.size() << "\n";
    
    for(size_t i = 0; i < pairwise_matches.size(); ++i)
    {
        if (opts.verbose > 2)
            logstream(LOG_EMPH)
            << "Pairwise_matches : (" << pairwise_matches[i].src_img_idx << "," << pairwise_matches[i].dst_img_idx << ")\n";
        
        if (pairwise_matches[i].src_img_idx >= 0 && pairwise_matches[i].dst_img_idx >= 0)
        {
            if (pairwise_matches[i].src_img_idx < pairwise_matches[i].dst_img_idx)	// no duplicate edges are allowed
            {
                vid = pairwise_matches[i].src_img_idx;
                other_vid = pairwise_matches[i].dst_img_idx;
                if (opts.verbose > 0)
            	    logstream(LOG_EMPH) << "Adding edge: (" << vid << "," << other_vid << ")\n";
                
                edge_data edata; edata.empty = false;
                graph.add_edge(vid,other_vid,edata);
            }
        }
    }
    
    return true;
}