semi_synchronous_engine.hpp

/**  
 * Copyright (c) 2009 Carnegie Mellon University. 
 *     All rights reserved.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an "AS
 *  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 *  express or implied.  See the License for the specific language
 *  governing permissions and limitations under the License.
 *
 * For more about this software visit:
 *
 *      http://www.graphlab.ml.cmu.edu
 *
 */



#ifndef GRAPHLAB_SEMI_SYNCHRONOUS_ENGINE_HPP
#define GRAPHLAB_SEMI_SYNCHRONOUS_ENGINE_HPP

#include <deque>
#include <algorithm>
#include <boost/bind.hpp>

#include <graphlab/engine/iengine.hpp>

#include <graphlab/vertex_program/ivertex_program.hpp>
#include <graphlab/vertex_program/icontext.hpp>
#include <graphlab/vertex_program/context.hpp>

#include <graphlab/engine/execution_status.hpp>
#include <graphlab/options/graphlab_options.hpp>




#include <graphlab/parallel/pthread_tools.hpp>
#include <graphlab/parallel/atomic_add_vector.hpp>
#include <graphlab/parallel/lockfree_push_back.hpp>
#include <graphlab/util/tracepoint.hpp>
#include <graphlab/util/memory_info.hpp>

#include <graphlab/rpc/dc_dist_object.hpp>
#include <graphlab/rpc/distributed_event_log.hpp>
#include <graphlab/rpc/buffered_exchange.hpp>






#include <graphlab/macros_def.hpp>

namespace graphlab {
  
  
  /**
   * \ingroup engines
   * 
   * \brief The semi synchronous engine executes a fraction of all active 
   * vertex program synchronously in a sequence of super-step (iterations) 
   * in both the shared and distributed memory settings.
   * 
   * \tparam VertexProgram The user defined vertex program which
   * should implement the \ref graphlab::ivertex_program interface.   
   *
   *  
   * ### Execution Semantics
   *
   * The semi synchronous engine is functionally "in between" the synchronous 
   * and the asynchronous engine. It behaves like the synchronous engine and 
   * runs vertex progams in super-steps, but it only runs a subset of the set of
   * active vertices each round, thus achieving asynchronous-like operation 
   * using a synchronous execution.
   *
   * On start() the \ref graphlab::ivertex_program::init function is invoked
   * on the subset of active vertex programs to initialize the vertex 
   * program, vertex data, and possibly signal vertices.
   * The engine then proceeds to execute a sequence of
   * super-steps (iterations) each of which is further decomposed into a 
   * sequence of minor-steps which are also executed synchronously:
   * \li Receive all incoming messages (signals) by invoking the 
   * \ref graphlab::ivertex_program::init function on all
   * vertex-programs that have incoming messages.  If a
   * vertex-program does not have any incoming messages then it is
   * not active during this super-step.  
   * \li Execute all gathers for active vertex programs by invoking
   * the user defined \ref graphlab::ivertex_program::gather function
   * on the edge direction returned by the 
   * \ref graphlab::ivertex_program::gather_edges function.  The gather
   * functions can modify edge data but cannot modify the vertex
   * program or vertex data and therefore can be executed on multiple
   * edges in parallel.  The gather type is used to accumulate (sum)
   * the result of the gather function calls.
   * \li Execute all apply functions for active vertex-programs by
   * invoking the user defined \ref graphlab::ivertex_program::apply
   * function passing the sum of the gather functions.  If \ref
   * graphlab::ivertex_program::gather_edges returns no edges then
   * the default gather value is passed to apply.  The apply function
   * can modify the vertex program and vertex data.
   * \li Execute all scatters for active vertex programs by invoking
   * the user defined \ref graphlab::ivertex_program::scatter function
   * on the edge direction returned by the 
   * \ref graphlab::ivertex_program::scatter_edges function.  The scatter
   * functions can modify edge data but cannot modify the vertex
   * program or vertex data and therefore can be executed on multiple
   * edges in parallel.  
   * 
   * ### Construction
   *
   * The semi-synchronous engine is constructed by passing in a 
   * \ref graphlab::distributed_control object which manages coordination 
   * between engine threads and a \ref graphlab::distributed_graph object 
   * which is the graph on which the engine should be run.  The graph should
   * already be populated and cannot change after the engine is constructed. 
   * In the distributed setting all program instances (running on each machine)
   * should construct an instance of the engine at the same time.
   * 
   * Computation is initiated by signaling vertices using either 
   * \ref graphlab::semi_synchronous_engine::signal or 
   * \ref graphlab::semi_synchronous_engine::signal_all.  In either case all
   * machines should invoke signal or signal all at the same time.  Finally,
   * computation is initiated by calling the 
   * \ref graphlab::semi_synchronous_engine::start function. 
   * 
   * ### Example Usage
   *
   * The following is a simple example demonstrating how to use the engine:
   * \code
   * #include <graphlab.hpp>
   * 
   * struct vertex_data { 
   *   // code
   * };
   * struct edge_data { 
   *   // code
   * };
   * typedef graphlab::distributed_graph<vertex_data, edge_data> graph_type;
   * typedef float gather_type;
   * struct pagerank_vprog : 
   *   public graphlab::ivertex_program<graph_type, gather_type> {
   *   // code
   * };
   * 
   * int main(int argc, char** argv) {
   *   // Initialize control plain using mpi
   *   graphlab::mpi_tools::init(argc, argv);
   *   graphlab::distributed_control dc;
   *   // Parse command line options
   *   graphlab::command_line_options clopts("PageRank algorithm.");
   *   std::string graph_dir; 
   *   clopts.attach_option("graph", &graph_dir, graph_dir,
   *                        "The graph file.");
   *   if(!clopts.parse(argc, argv)) {
   *     std::cout << "Error in parsing arguments." << std::endl;
   *     return EXIT_FAILURE;
   *   }
   *   graph_type graph(dc, clopts);
   *   graph.load_structure(graph_dir, "tsv");
   *   graph.finalize();
   *   std::cout << "#vertices: " << graph.num_vertices() 
   *             << " #edges:" << graph.num_edges() << std::endl;
   *   graphlab::semi_synchronous_engine<pagerank_vprog> engine(dc, graph, clopts);
   *   engine.signal_all();
   *   engine.start();
   *   std::cout << "Runtime: " << engine.elapsed_time();
   *   graphlab::mpi_tools::finalize();
   * }
   * \endcode
   *
   *
   * 
   * <a name=engineopts>Engine Options</a>
   * =====================
   * The semi synchronous engine supports several engine options which can
   * be set as command line arguments using \c --engine_opts :
   * 
   * \li <b>max_iterations</b>: (default: infinity) The maximum number
   * of iterations (super-steps) to run.
   *
   * \li <b>timeout</b>: (default: infinity) The maximum time in
   * seconds that the engine may run. When the time runs out the
   * current iteration is completed and then the engine terminates.
   *
   * \li <b>use_cache</b>: (default: false) This is used to enable
   * caching.  When caching is enabled the gather phase is skipped for
   * vertices that already have a cached value.  To use caching the
   * vertex program must either clear (\ref icontext::clear_gather_cache) 
   * or update (\ref icontext::post_delta) the cache values of 
   * neighboring vertices during the scatter phase.
   *
   * \li \b snapshot_interval If set to a positive value, a snapshot
   * is taken every this number of iterations. If set to 0, a snapshot
   * is taken before the first iteration. If set to a negative value,
   * no snapshots are taken. Defaults to -1. A snapshot is a binary
   * dump of the graph.
   *
   * \li \b snapshot_path If snapshot_interval is set to a value >=0,
   * this option must be specified and should contain a target basename 
   * for the snapshot. The path including folder and file prefix in 
   * which the snapshots should be saved.
   *
   * \see graphlab::omni_engine
   * \see graphlab::async_consistent_engine
   * \see graphlab::synchronous_engine
   */
  template<typename VertexProgram>
  class semi_synchronous_engine : 
    public iengine<VertexProgram> {

  public:
    /**
     * \brief The user defined vertex program type. Equivalent to the
     * VertexProgram template argument.
     * 
     * The user defined vertex program type which should implement the
     * \ref graphlab::ivertex_program interface.
     */
    typedef VertexProgram vertex_program_type;

    /**
     * \brief The user defined type returned by the gather function.
     * 
     * The gather type is defined in the \ref graphlab::ivertex_program 
     * interface and is the value returned by the 
     * \ref graphlab::ivertex_program::gather function.  The
     * gather type must have an <code>operator+=(const gather_type&
     * other)</code> function and must be \ref sec_serializable.
     */
    typedef typename VertexProgram::gather_type gather_type;


    /** 
     * \brief The user defined message type used to signal neighboring 
     * vertex programs.
     * 
     * The message type is defined in the \ref graphlab::ivertex_program 
     * interface and used in the call to \ref graphlab::icontext::signal.  
     * The message type must have an 
     * <code>operator+=(const gather_type& other)</code> function and
     * must be \ref sec_serializable.
     */
    typedef typename VertexProgram::message_type message_type;

    /**
     * \brief The type of data associated with each vertex in the graph
     *
     * The vertex data type must be \ref sec_serializable. 
     */
    typedef typename VertexProgram::vertex_data_type vertex_data_type;

    /**
     * \brief The type of data associated with each edge in the graph
     *
     * The edge data type must be \ref sec_serializable. 
     */
    typedef typename VertexProgram::edge_data_type edge_data_type;

    /**
     * \brief The type of graph supported by this vertex program
     * 
     * See graphlab::distributed_graph 
     */
    typedef typename VertexProgram::graph_type  graph_type;

    /**
     * \brief The type used to represent a vertex in the graph.
     * See \ref graphlab::distributed_graph::vertex_type for details
     *
     * The vertex type contains the function 
     * \ref graphlab::distributed_graph::vertex_type::data which 
     * returns a reference to the vertex data as well as other functions 
     * like \ref graphlab::distributed_graph::vertex_type::num_in_edges
     * which returns the number of in edges.
     * 
     */
    typedef typename graph_type::vertex_type          vertex_type;
    
    /**
     * \brief The type used to represent an edge in the graph.  
     * See \ref graphlab::distributed_graph::edge_type for details.
     * 
     * The edge type contains the function
     * \ref graphlab::distributed_graph::edge_type::data which returns a 
     * reference to the edge data.  In addition the edge type contains 
     * the function \ref graphlab::distributed_graph::edge_type::source and
     * \ref graphlab::distributed_graph::edge_type::target.
     * 
     */
    typedef typename graph_type::edge_type            edge_type;

    /**
     * \brief The type of the callback interface passed by the engine to vertex 
     * programs.  See \ref graphlab::icontext for details.
     *
     * The context callback is passed to the vertex program functions and is
     * used to signal other vertices, get the current iteration, and access
     * information about the engine.
     */
    typedef icontext<graph_type, gather_type, message_type> icontext_type;
           
  private:

    /**
     * \brief Local vertex type used by the engine for fast indexing
     */
    typedef typename graph_type::local_vertex_type    local_vertex_type;
    
    /**
     * \brief Local edge type used by the engine for fast indexing
     */
    typedef typename graph_type::local_edge_type      local_edge_type;
    
    /**
     * \brief Local vertex id type used by the engine for fast indexing
     */
    typedef typename graph_type::lvid_type            lvid_type;

    std::vector<double> per_thread_compute_time; 
    /**
     * \brief The actual instance of the context type used by this engine.
     */
    typedef context<semi_synchronous_engine> context_type;
    friend class context<semi_synchronous_engine>;


    /**
     * \brief The type of the distributed aggregator inherited from iengine
     */
    typedef typename iengine<vertex_program_type>::aggregator_type aggregator_type;

    /**
     * \brief The object used to communicate with remote copies of the
     * synchronous engine.
     */
    dc_dist_object< semi_synchronous_engine<VertexProgram> > rmi;

    /**
     * \brief A reference to the distributed graph on which this
     * synchronous engine is running.
     */
    graph_type& graph;

    /**
     * \brief The local worker threads used by this engine
     */
    thread_pool threads;

    /**
     * \brief A thread barrier that is used to control the threads in the
     * thread pool.
     */
    graphlab::barrier thread_barrier;

    /**
     * \brief The maximum number of super-steps (iterations) to run
     * before terminating.  If the max iterations is reached the
     * engine will terminate if their are no messages remaining.
     */
    size_t max_iterations;

    /**
     * \brief A snapshot is taken every this number of iterations.
     * If snapshot_interval == 0, a snapshot is only taken before the first
     * iteration. If snapshot_interval < 0, no snapshots are taken.
     */
    int snapshot_interval;

    /// \brief The target base name the snapshot is saved in.
    std::string snapshot_path;

    /**
     * \brief A counter that tracks the current iteration number since
     * start was last invoked.
     */
    size_t iteration_counter;

    /**
     * \brief set to true if engine is running
     */
    bool started;

    /**
     * \brief The time in seconds at which the engine started.
     */
    float start_time;

    /**
     * \brief The timeout time in seconds
     */
    float timeout;

    /**
     * \brief Used to stop the engine prematurely
     */
    bool force_abort;


    /**
     * \brief the minimum number of active vertices per round
     */
    size_t max_active_vertices;

    /**
     * \brief The vertex locks protect access to vertex specific
     * data-structures including 
     * \ref graphlab::semi_synchronous_engine::gather_accum
     * and \ref graphlab::semi_synchronous_engine::messages.
     */
    std::vector<simple_spinlock> vlocks;


    /**
     * \brief The elocks protect individual edges during gather and
     * scatter.  Technically there is a potential race since gather
     * and scatter can modify edge values and can overlap.  The edge
     * lock ensures that only one gather or scatter occurs on an edge
     * at a time.
     */
    std::vector<simple_spinlock> elocks;
    


    /**
     * \brief The vertex programs associated with each vertex on this
     * machine.
     */
    std::vector<vertex_program_type> vertex_programs;


    /// A pointer to the scheduler object
    ischeduler<message_type>* scheduler_ptr;


    dense_bitset has_remote_message;

    /** \brief used by transfer_scheduler_to_active. The number of vertices 
     *  to use in the next superstep.
     */
    atomic<int> num_to_activate;

    /**
     * \brief Gather accumulator used for each master vertex to merge
     * the result of all the machine specific accumulators (or
     * caches).
     *
     * The gather accumulator can be accessed by multiple threads at
     * once and therefore must be guarded by a vertex locks in 
     * \ref graphlab::semi_synchronous_engine::vlocks
     */
    std::vector<gather_type>  gather_accum;
    
    /**
     * \brief Bit indicating if the gather has accumulator contains any
     * values.  
     *
     * While dense bitsets are thread safe the value of this bit must
     * change concurrently with the 
     * \ref graphlab::semi_synchronous_engine::gather_accum and therefore is
     * set while holding the lock in
     * \ref graphlab::semi_synchronous_engine::vlocks.
     */
    dense_bitset has_gather_accum;


    /**
     * \brief This optional vector contains caches of previous gather
     * contributions for each machine.  
     *
     * Caching is done locally and therefore a high-degree vertex may
     * have multiple caches (one per machine).
     */
    std::vector<gather_type>  gather_cache;

    /**
     * \brief A bit indicating if the local gather for that vertex is
     * available.
     */
    dense_bitset has_cache;   

    std::vector<lvid_type> active_superstep;
    lockfree_push_back<std::vector<lvid_type> > active_superstep_pushback;

    /**
     * \brief  The number of local vertices (masters) that are active on this
     * iteration.
     */
    atomic<size_t> num_active_vertices;


    /**
     * \brief A bit indicating (for all vertices) whether to
     * participate in the current minor-step (gather or scatter).
     */
    std::vector<lvid_type> active_minorstep;
    lockfree_push_back<std::vector<lvid_type> > active_minorstep_pushback;

    /**
     * \brief A counter measuring the number of applys that have been completed
     */
    atomic<size_t> completed_applys;


    /**
     * \brief The shared counter used coordinate operations between
     * threads.
     */
    atomic<size_t> shared_lvid_counter;


    /**
     * \brief The pair type used to synchronize vertex programs across machines.
     */
    typedef std::pair<vertex_id_type, vertex_program_type> vid_prog_pair_type;

    /**
     * \brief The type of the exchange used to synchronize vertex programs
     */
    typedef buffered_exchange<vid_prog_pair_type> vprog_exchange_type;
   
    /**
     * \brief The distributed exchange used to synchronize changes to
     * vertex programs.
     */
    vprog_exchange_type vprog_exchange;

    /**
     * \brief The pair type used to synchronize vertex across across machines.
     */
    typedef std::pair<vertex_id_type, vertex_data_type> vid_vdata_pair_type;

    /**
     * \brief The type of the exchange used to synchronize vertex data
     */
    typedef buffered_exchange<vid_vdata_pair_type> vdata_exchange_type;

    /**
     * \brief The distributed exchange used to synchronize changes to
     * vertex programs.
     */
    vdata_exchange_type vdata_exchange;

    /**
     * \brief The pair type used to synchronize the results of the gather phase
     */
    typedef std::pair<vertex_id_type, gather_type> vid_gather_pair_type;

    /**
     * \brief The type of the exchange used to synchronize gather
     * accumulators
     */
    typedef buffered_exchange<vid_gather_pair_type> gather_exchange_type;
   
    /**
     * \brief The distributed exchange used to synchronize gather
     * accumulators.
     */
    gather_exchange_type gather_exchange;

    /**
     * \brief The pair type used to synchronize messages
     */
    typedef std::pair<vertex_id_type, message_type> vid_message_pair_type;
   
    /**
     * \brief The type of the exchange used to synchronize messages
     */
    typedef buffered_exchange<vid_message_pair_type> message_exchange_type;

    /**
     * \brief The distributed exchange used to synchronize messages
     */
    message_exchange_type message_exchange;


    /**
     * \brief The distributed aggregator used to manage background
     * aggregation.
     */
    aggregator_type aggregator;

    DECLARE_EVENT(EVENT_APPLIES);
    DECLARE_EVENT(EVENT_GATHERS);
    DECLARE_EVENT(EVENT_SCATTERS);
    DECLARE_EVENT(EVENT_ACTIVE_CPUS);
  public:

    /**
     * \brief Construct a semi synchronous engine for a given graph and options.
     *
     * The semi synchronous engine should be constructed after the graph
     * has been loaded (e.g., \ref graphlab::distributed_graph::load)
     * and the graphlab options have been set 
     * (e.g., \ref graphlab::command_line_options).
     *
     * In the distributed engine the semi synchronous engine must be called
     * on all machines at the same time (in the same order) passing
     * the \ref graphlab::distributed_control object.  Upon
     * construction the semi synchronous engine allocates several
     * data-structures to store messages, gather accumulants, and
     * vertex programs and therefore may require considerable memory.
     *
     * The number of threads to create are read from 
     * \ref graphlab_options::get_ncpus "opts.get_ncpus()". 
     *
     * See the <a href="#engineopts">main class documentation</a>
     * for details on the available options.
     *
     * @param [in] dc Distributed controller to associate with
     * @param [in,out] graph A reference to the graph object that this
     * engine will modify. The graph must be fully constructed and 
     * finalized.
     * @param [in] opts A graphlab::graphlab_options object specifying engine
     *                  parameters.  This is typically constructed using
     *                  \ref graphlab::command_line_options.
     */
    semi_synchronous_engine(distributed_control& dc, graph_type& graph,
                       const graphlab_options& opts = graphlab_options());


    ~semi_synchronous_engine() {
      delete scheduler_ptr;
    }

    /**
     * \brief Start execution of the semi synchronous engine.
     * 
     * The start function begins computation and does not return until
     * there are no remaining messages or until max_iterations has
     * been reached. 
     * 
     * The start() function modifies the data graph through the vertex
     * programs and so upon return the data graph should contain the
     * result of the computation.
     *
     * @return The reason for termination
     */
    execution_status::status_enum start();

    // documentation inherited from iengine
    size_t num_updates() const;

    // documentation inherited from iengine
    void signal(vertex_id_type vid,
                const message_type& message = message_type());
    
    // documentation inherited from iengine
    void signal_all(const message_type& message = message_type(),
                    const std::string& order = "shuffle");

    void signal_vset(const vertex_set& vset,
                    const message_type& message = message_type(),
                    const std::string& order = "shuffle");


    // documentation inherited from iengine
    float elapsed_seconds() const;
    
    /**
     * \brief Get the current iteration number since start was last
     * invoked.
     *
     *  \return the current iteration
     */
    int iteration() const; 


    /**
     * \brief Compute the total memory used by the entire distributed
     * system.
     * 
     * @return The total memory used in bytes.
     */
    size_t total_memory_usage() const;

    /**
     * \brief Get a pointer to the distributed aggregator object. 
     *
     * This is currently used by the \ref graphlab::iengine interface to 
     * implement the calls to aggregation.
     *
     * @return a pointer to the local aggregator.
     */
    aggregator_type* get_aggregator();

  private:

    /**
     * \brief This internal stop function is called by the \ref graphlab::context to
     * terminate execution of the engine.
     */
    void internal_stop();    

    /**
     * \brief This function is called remote by the rpc to force the
     * engine to stop.
     */
    void rpc_stop();

    /**
     * \brief Signal a vertex.
     *
     * This function is called by the \ref graphlab::context.
     *
     * @param [in] vertex the vertex to signal
     * @param [in] message the message to send to that vertex.
     */
    void internal_signal(const vertex_type& vertex,
                         const message_type& message = message_type()); 

    /**
     * \brief Called by the context to signal an arbitrary vertex.
     * This must be done by finding the owner of that vertex. 
     *
     * @param [in] gvid the global vertex id of the vertex to signal
     * @param [in] message the message to send to that vertex.
     */
    void internal_signal_broadcast(vertex_id_type gvid,
                                   const message_type& message = message_type());
    
    /**
     * \brief This function tests if this machine is the master of
     * gvid and signals if successful.
     */
    void internal_signal_rpc(vertex_id_type gvid,
                              const message_type& message = message_type());


    /** 
     * \brief Post a to a previous gather for a give vertex.
     *
     * This function is called by the \ref graphlab::context.
     *
     * @param [in] vertex The vertex to which to post a change in the sum
     * @param [in] delta The change in that sum
     */
    void internal_post_delta(const vertex_type& vertex,
                             const gather_type& delta);

    /**
     * \brief Clear the cached gather for a vertex if one is
     * available.
     *
     * This function is called by the \ref graphlab::context.
     *
     * @param [in] vertex the vertex for which to clear the cache
     */
    void internal_clear_gather_cache(const vertex_type& vertex);


    // Program Steps ==========================================================
   

    void thread_launch_wrapped_event_counter(size_t thread_id,
                                             boost::function<void(void)> fn) {
      INCREMENT_EVENT(EVENT_ACTIVE_CPUS, 1);
      rmi.dc().stop_handler_threads(thread_id, threads.size());
      fn();
      rmi.dc().start_handler_threads(thread_id, threads.size());
      DECREMENT_EVENT(EVENT_ACTIVE_CPUS, 1);
    }

    /**
     * \brief Executes ncpus copies of a member function each with a
     * unique consecutive id (thread id). 
     *
     * This function is used by the main loop to execute each of the
     * stages in parallel.
     *
     * The member function must have the type:
     *
     * \code
     * void semi_synchronous_engine::member_fun(size_t threadid);
     * \endcode
     *
     * This function runs an rmi barrier after termination
     *
     * @tparam the type of the member function.  
     * @param [in] member_fun the function to call.
     */
    template<typename MemberFunction>       
    void run_synchronous(MemberFunction member_fun) {
      shared_lvid_counter = 0;
      // launch the initialization threads
      for(size_t i = 0; i < threads.size(); ++i) {
        boost::function<void(void)> invoke = boost::bind(member_fun, this, i);
        threads.launch(boost::bind(
                &semi_synchronous_engine::thread_launch_wrapped_event_counter, 
                this,
                i,
                invoke), i);
      }
      // Wait for all threads to finish
      threads.join();
      rmi.barrier();
    } // end of run_synchronous

    // /** 
    //  * \brief Initialize all vertex programs by invoking 
    //  * \ref graphlab::ivertex_program::init on all vertices.
    //  *
    //  * @param thread_id the thread to run this as which determines
    //  * which vertices to process.
    //  */
    // void initialize_vertex_programs(size_t thread_id);

    /** 
     * \brief Synchronize all message data.
     *
     * @param thread_id the thread to run this as which determines
     * which vertices to process.
     */
    void exchange_messages(size_t thread_id);

    /** 
     * \brief Synchronize some message data and prepares some vertices for 
     *        execution
     *
     * @param thread_id the thread to run this as which determines
     * which vertices to process.
     */
    void transfer_scheduler_to_active(size_t thread_id);


    /** 
     * \brief Execute the \ref graphlab::ivertex_program::gather function on all 
     * vertices that received messages for the edges specified by the 
     * \ref graphlab::ivertex_program::gather_edges.
     *
     * @param thread_id the thread to run this as which determines
     * which vertices to process.
     */
    void execute_gathers(size_t thread_id);
    



    /** 
     * \brief Execute the \ref graphlab::ivertex_program::apply function on all
     * all vertices that received messages in this super-step (active).
     *
     * @param thread_id the thread to run this as which determines
     * which vertices to process.
     */
    void execute_applys(size_t thread_id);

    /** 
     * \brief Execute the \ref graphlab::ivertex_program::scatter function on all 
     * vertices that received messages for the edges specified by the 
     * \ref graphlab::ivertex_program::scatter_edges.
     *
     * @param thread_id the thread to run this as which determines
     * which vertices to process.
     */
    void execute_scatters(size_t thread_id);

    // Data Synchronization ===================================================
    /**
     * \brief Send the vertex program for the local vertex id to all
     * of its mirrors. 
     *
     * @param [in] lvid the vertex to sync.  This muster must be the
     * master of that vertex.
     */
    void sync_vertex_program(lvid_type lvid, size_t thread_id);

    /**
     * \brief Receive all incoming vertex programs and update the
     * local mirrors.
     *
     * This function returns when there are no more incoming vertex
     * programs and should be called after a flush of the vertex
     * program exchange.
     */
    void recv_vertex_programs(size_t threadid, const bool try_to_recv = false);

    /**
     * \brief Send the vertex data for the local vertex id to all of
     * its mirrors.
     *
     * @param [in] lvid the vertex to sync.  This machine must be the master
     * of that vertex.
     */
    void sync_vertex_data(lvid_type lvid, size_t thread_id);
    
    /**
     * \brief Receive all incoming vertex data and update the local
     * mirrors.
     *
     * This function returns when there are no more incoming vertex
     * data and should be called after a flush of the vertex data
     * exchange.
     */
    void recv_vertex_data(size_t threadid, const bool try_to_recv = false);

    /**
     * \brief Send the gather value for the vertex id to its master.
     *
     * @param [in] lvid the vertex to send the gather value to
     * @param [in] accum the locally computed gather value. 
     */
    void sync_gather(lvid_type lvid, const gather_type& accum, 
                     size_t thread_id);


    /**
     * \brief Receive the gather values from the buffered exchange.
     *
     * This function returns when there is nothing left in the
     * buffered exchange and should be called after the buffered
     * exchange has been flushed
     */
    void recv_gathers(size_t threadid, const bool try_to_recv = false);

    /**
     * \brief Send the accumulated message for the local vertex to its
     * master.
     *
     * @param [in] lvid the vertex to send 
     */
    void sync_message(lvid_type lvid, 
                      const size_t thread_id, 
                      const message_type& msg);

    /**
     * \brief Receive the messages from the buffered exchange.
     *
     * This function returns when there is nothing left in the
     * buffered exchange and should be called after the buffered
     * exchange has been flushed
     */
    void recv_messages(size_t threadid, const bool try_to_recv = false);


  }; // end of class semi synchronous engine

























  /**
   * Constructs a semi synchronous distributed engine.
   * The number of threads to create are read from
   * opts::get_ncpus().
   *
   * Valid engine options (graphlab_options::get_engine_args()):
   * \arg \c max_iterations Sets the maximum number of iterations the
   * engine will run for. 
   * \arg \c use_cache If set to true, partial gathers are cached.
   * See \ref gather_caching to understand the behavior of the
   * gather caching model and how it may be used to accelerate program
   * performance.
   *
   * \param dc Distributed controller to associate with
   * \param graph The graph to schedule over. The graph must be fully
   *              constructed and finalized.
   * \param opts A graphlab_options object containing options and parameters
   *             for the engine.
   */
  template<typename VertexProgram>
  semi_synchronous_engine<VertexProgram>::
  semi_synchronous_engine(distributed_control &dc, 
                     graph_type& graph,
                     const graphlab_options& opts) :
    rmi(dc, this), graph(graph), 
    threads(opts.get_ncpus()), 
    thread_barrier(opts.get_ncpus()),
    max_iterations(-1), snapshot_interval(-1), iteration_counter(0),
    started(false), timeout(0),
    max_active_vertices(1000),
    scheduler_ptr(NULL),
    active_superstep(128),
    active_superstep_pushback(active_superstep, 0), 
    active_minorstep(128),
    active_minorstep_pushback(active_minorstep, 0), 
    vprog_exchange(dc, opts.get_ncpus(), 65536), 
    vdata_exchange(dc, opts.get_ncpus(), 65536), 
    gather_exchange(dc, opts.get_ncpus(), 65536), 
    message_exchange(dc, opts.get_ncpus(), 65536),
    aggregator(dc, graph, new context_type(*this, graph)) {
    // Process any additional options
    std::vector<std::string> keys = opts.get_engine_args().get_option_keys();
    per_thread_compute_time.resize(opts.get_ncpus());
    bool use_cache = false;

    graph.finalize();

    max_active_vertices = graph.num_local_vertices() * 0.1;
    max_active_vertices = std::max<size_t>(max_active_vertices, 1000);

    foreach(std::string opt, keys) {
      if (opt == "max_iterations") {
        opts.get_engine_args().get_option("max_iterations", max_iterations);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: max_iterations = " 
                              << max_iterations << std::endl;
      } else if (opt == "timeout") {
        opts.get_engine_args().get_option("timeout", timeout);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: timeout = " 
                              << timeout << std::endl;
      } else if (opt == "use_cache") {
        opts.get_engine_args().get_option("use_cache", use_cache);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: use_cache = " 
                              << use_cache << std::endl;
      } else if (opt == "snapshot_interval") {
        opts.get_engine_args().get_option("snapshot_interval", snapshot_interval);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: snapshot_interval = " 
                              << snapshot_interval << std::endl;
      } else if (opt == "snapshot_path") {
        opts.get_engine_args().get_option("snapshot_path", snapshot_path);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: snapshot_path = " 
                              << snapshot_path << std::endl;
      } else if (opt == "max_active_vertices") {
        opts.get_engine_args().get_option("max_active_vertices", max_active_vertices);
        if (rmi.procid() == 0)
          logstream(LOG_EMPH) << "Engine Option: max_active_vertices = " 
                              << max_active_vertices << std::endl;
      } else if (opt == "max_active_fraction") {
        float max_active_fraction = 0.1;
        opts.get_engine_args().get_option("max_active_fraction", max_active_fraction);
        if (max_active_fraction > 0) {
          max_active_vertices = graph.num_local_vertices() * max_active_fraction;
          max_active_vertices += (max_active_vertices == 0);
        }
        else {
          max_active_vertices = (size_t)(-1);
        }
        if (rmi.procid() == 0) {
          logstream(LOG_EMPH) << "Engine Option: max_active_fraction = " 
                              << max_active_fraction << std::endl;
        }
      } else {
        logstream(LOG_FATAL) << "Unexpected Engine Option: " << opt << std::endl;
      }
    }

    if (snapshot_interval >= 0 && snapshot_path.length() == 0) {
      logstream(LOG_FATAL) 
          << "Snapshot interval specified, but no snapshot path" << std::endl;
    }
    INITIALIZE_EVENT_LOG(dc);
    ADD_CUMULATIVE_EVENT(EVENT_APPLIES, "Applies", "Calls");
    ADD_CUMULATIVE_EVENT(EVENT_GATHERS , "Gathers", "Calls");
    ADD_CUMULATIVE_EVENT(EVENT_SCATTERS , "Scatters", "Calls");
    ADD_INSTANTANEOUS_EVENT(EVENT_ACTIVE_CPUS, "Active Threads", "Threads");

    // Finalize the graph
    memory_info::log_usage("Before Engine Initialization");
    // Allocate vertex locks and vertex programs
    active_superstep.resize(2 * max_active_vertices);
    active_minorstep.resize(2 * max_active_vertices);
    vlocks.resize(graph.num_local_vertices());
    vertex_programs.resize(graph.num_local_vertices());
    has_remote_message.resize(graph.num_local_vertices());
    has_remote_message.clear();
    // allocate the edge locks
    //elocks.resize(graph.num_local_edges());
    // Allocate gather accumulators and accumulator bitset
    gather_accum.resize(graph.num_local_vertices(), gather_type());
    has_gather_accum.resize(graph.num_local_vertices());
    has_gather_accum.clear();
    // If caching is used then allocate cache data-structures
    if (use_cache) {
      gather_cache.resize(graph.num_local_vertices(), gather_type());
      has_cache.resize(graph.num_local_vertices());
      has_cache.clear();
    }
    // Allocate bitset to track active vertices on each bitset.
    active_superstep.resize(opts.get_ncpus());
    // Print memory usage after initialization
    memory_info::log_usage("After Engine Initialization");

    graphlab_options opts_copy = opts;

    // set a default scheduler if none
    if (opts_copy.get_scheduler_type() == "") {
      opts_copy.set_scheduler_type("queued_fifo");
    }

    // construct scheduler
    scheduler_ptr = scheduler_factory<message_type>::
                    new_scheduler(graph.num_local_vertices(),
                                  opts_copy);

    rmi.barrier();
  } // end of synchronous engine
  






  template<typename VertexProgram>
  typename semi_synchronous_engine<VertexProgram>::aggregator_type*
  semi_synchronous_engine<VertexProgram>::get_aggregator() {
    return &aggregator;
  } // end of get_aggregator



  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::internal_stop() {
    for (size_t i = 0; i < rmi.numprocs(); ++i) 
      rmi.remote_call(i, &semi_synchronous_engine<VertexProgram>::rpc_stop);
  } // end of internal_stop

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::rpc_stop() {
    force_abort = true;
  } // end of rpc_stop


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  signal(vertex_id_type gvid, const message_type& message) {
    rmi.barrier();
    internal_signal_rpc(gvid, message);
    rmi.barrier();
  } // end of signal



  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  signal_all(const message_type& message, const std::string& order) {
    for(lvid_type lvid = 0; lvid < graph.num_local_vertices(); ++lvid) {
      if(graph.l_is_master(lvid)) {
        internal_signal(vertex_type(graph.l_vertex(lvid)), message);
      }
    }
  } // end of signal all
  

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  signal_vset(const vertex_set& vset,
             const message_type& message, const std::string& order) {
    for(lvid_type lvid = 0; lvid < graph.num_local_vertices(); ++lvid) {
      if(graph.l_is_master(lvid) && vset.l_contains(lvid)) {
        internal_signal(vertex_type(graph.l_vertex(lvid)), message);
      }
    }
  } // end of signal all
 

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  internal_signal(const vertex_type& vertex,
                  const message_type& message) {
    const lvid_type lvid = vertex.local_id();
    if (!graph.l_is_master(lvid)) {
      scheduler_ptr->place(lvid, message);
      has_remote_message.set_bit(lvid);
    }
    else {
      if (started) {
        scheduler_ptr->schedule_from_execution_thread(thread::thread_id(),
                                                      lvid, message);
      } else {
        scheduler_ptr->schedule(lvid, message);
      }
    }
  } // end of internal_signal


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  internal_signal_broadcast(vertex_id_type gvid, const message_type& message) {
    for (size_t i = 0; i < rmi.numprocs(); ++i) {
      if(i == rmi.procid()) internal_signal_rpc(gvid, message);
      else rmi.remote_call(i, &semi_synchronous_engine<VertexProgram>::internal_signal_rpc,
                          gvid, message);
    }
  } // end of internal_signal_broadcast

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  internal_signal_rpc(vertex_id_type gvid,
                      const message_type& message) {
    if (graph.is_master(gvid)) {
      internal_signal(graph.vertex(gvid), message);
    }
  } // end of internal_signal_rpc



  

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  internal_post_delta(const vertex_type& vertex, const gather_type& delta) {
    const bool caching_enabled = !gather_cache.empty();
    if(caching_enabled) {
      const lvid_type lvid = vertex.local_id();      
      vlocks[lvid].lock();
      if( has_cache.get(lvid) ) {
        gather_cache[lvid] += delta;
      } else {
        // You cannot add a delta to an empty cache.  A complete
        // gather must have been run.
        // gather_cache[lvid] = delta;
        // has_cache.set_bit(lvid);
      }
      vlocks[lvid].unlock();       
    }
  } // end of post_delta


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  internal_clear_gather_cache(const vertex_type& vertex) {
    const bool caching_enabled = !gather_cache.empty();
    const lvid_type lvid = vertex.local_id();
    if(caching_enabled && has_cache.get(lvid)) {
      vlocks[lvid].lock();
      gather_cache[lvid] = gather_type();
      has_cache.clear_bit(lvid);
      vlocks[lvid].unlock();
    }
  } // end of clear_gather_cache
  



  template<typename VertexProgram>
  size_t semi_synchronous_engine<VertexProgram>::
  num_updates() const { return completed_applys.value; }

  template<typename VertexProgram>
  float semi_synchronous_engine<VertexProgram>::
  elapsed_seconds() const { return timer::approx_time_seconds() - start_time; }

  template<typename VertexProgram>
  int semi_synchronous_engine<VertexProgram>::
  iteration() const { return iteration_counter; }



  template<typename VertexProgram>
  size_t semi_synchronous_engine<VertexProgram>::total_memory_usage() const {
    size_t allocated_memory = memory_info::allocated_bytes();
    rmi.all_reduce(allocated_memory);
    return allocated_memory;
  } // compute the total memory usage of the GraphLab system


  template<typename VertexProgram> execution_status::status_enum
  semi_synchronous_engine<VertexProgram>::start() {
    rmi.barrier();
    graph.finalize();
    // Initialization code ==================================================     
    // Reset event log counters? 
    // Start the timer
    graphlab::timer timer; timer.start();
    start_time = timer::approx_time_seconds();
    iteration_counter = 0;
    force_abort = false;
    execution_status::status_enum termination_reason = execution_status::UNSET; 
    aggregator.start();
    scheduler_ptr->start();
    started = true;

    rmi.barrier();
    if (snapshot_interval == 0) {
      graph.save_binary(snapshot_path);
    }

    // set to -5 to force it to print the first round
    float last_print = -5;
    if (rmi.procid() == 0) {
      logstream(LOG_EMPH) << "Iteration counter will only output every 5 seconds." 
                        << std::endl;
    }


    // Program Main loop ====================================================      
    while(iteration_counter < max_iterations && !force_abort ) {

      // Check first to see if we are out of time
      if(timeout != 0 && timeout < elapsed_seconds()) {
        termination_reason = execution_status::TIMEOUT;
        break;
      }
      
      bool print_this_round = (elapsed_seconds() - last_print) >= 5;

      if(rmi.procid() == 0 && print_this_round) {
        logstream(LOG_EMPH) 
          << rmi.procid() << ": Starting iteration: " << iteration_counter 
          << std::endl;
        last_print = elapsed_seconds();
      }

      run_synchronous( &semi_synchronous_engine::exchange_messages);
      has_remote_message.clear();

      // Reset Active vertices ---------------------------------------------- 
      // Clear the active super-step and minor-step bits which will
      // be set upon receiving messages
      active_superstep_pushback.set_size(0);
      active_minorstep_pushback.set_size(0);
      has_gather_accum.clear(); 


      // how many to activate?
      num_to_activate = max_active_vertices;
      num_active_vertices = 0;

      rmi.barrier();
      // Exchange Messages --------------------------------------------------
      // Exchange any messages in the local message vectors
      // if (rmi.procid() == 0) std::cout << "Exchange messages..." << std::endl;
      run_synchronous( &semi_synchronous_engine::transfer_scheduler_to_active);

      /**
       * Post conditions:
       *   1) only master vertices have messages
       */

      /**
       * Post conditions:
       *   1) there are no messages remaining
       *   2) All masters that received messages have their
       *      active_superstep bit set
       *   3) All masters and mirrors that are to participate in the
       *      next gather phases have their active_minorstep bit
       *      set.
       *   4) num_active_vertices is the number of vertices that
       *      received messages.
       */
      
      // Check termination condition  ---------------------------------------
      // TODO: optimize by joining the reduces 
      size_t total_active_vertices = num_active_vertices; 
      rmi.all_reduce(total_active_vertices);
      if (rmi.procid() == 0 && print_this_round) 
        logstream(LOG_EMPH)
          << "\tActive vertices: " << total_active_vertices << std::endl;
      if(total_active_vertices == 0) {
        termination_reason = execution_status::TASK_DEPLETION;
        break;
      }

      // Execute gather operations-------------------------------------------
      // Execute the gather operation for all vertices that are active
      // in this minor-step (active-minorstep bit set).
      // if (rmi.procid() == 0) std::cout << "Gathering..." << std::endl;
      run_synchronous( &semi_synchronous_engine::execute_gathers );
      // Clear the minor step bit since only super-step vertices
      // (only master vertices are required to participate in the
      // apply step)
      active_minorstep_pushback.set_size(0);
      rmi.barrier();

      /**
       * Post conditions:
       *   1) gather_accum for all master vertices contains the
       *      result of all the gathers (even if they are drawn from
       *      cache)
       *   2) No minor-step bits are set
       */

      // Execute Apply Operations -------------------------------------------
      // Run the apply function on all active vertices
      // if (rmi.procid() == 0) std::cout << "Applying..." << std::endl;
      run_synchronous( &semi_synchronous_engine::execute_applys );
      /**
       * Post conditions:
       *   1) any changes to the vertex data have been synchronized
       *      with all mirrors.
       *   2) all gather accumulators have been cleared
       *   3) If a vertex program is participating in the scatter
       *      phase its minor-step bit has been set to active (both
       *      masters and mirrors) and the vertex program has been
       *      synchronized with the mirrors.         
       */


      // Execute Scatter Operations -----------------------------------------
      // Execute each of the scatters on all minor-step active vertices.



      run_synchronous( &semi_synchronous_engine::execute_scatters );

      /**
       * Post conditions:
       *   1) NONE
       */
      if(rmi.procid() == 0 && print_this_round) 
        logstream(LOG_EMPH) << "\t Running Aggregators" << std::endl;
      // probe the aggregator
      aggregator.tick_synchronous();
      
      ++iteration_counter;
      
      if (snapshot_interval > 0 && iteration_counter % snapshot_interval == 0) {
        graph.save_binary(snapshot_path);
      }
    }

    if (rmi.procid() == 0) {
      logstream(LOG_EMPH) << iteration_counter 
                        << " iterations completed." << std::endl;
    }
    // Final barrier to ensure that all engines terminate at the same time
    double total_compute_time = 0;
    for (size_t i = 0;i < per_thread_compute_time.size(); ++i) {
      total_compute_time += per_thread_compute_time[i];
    }
    std::vector<double> all_compute_time_vec(rmi.numprocs());
    all_compute_time_vec[rmi.procid()] = total_compute_time;
    rmi.all_gather(all_compute_time_vec);

    size_t global_completed = completed_applys;
    rmi.all_reduce(global_completed);
    completed_applys = global_completed;
    rmi.cout() << "Updates: " << completed_applys.value << "\n";
    if (rmi.procid() == 0) { 
      logstream(LOG_INFO) << "Compute Balance: ";
      for (size_t i = 0;i < all_compute_time_vec.size(); ++i) {
        logstream(LOG_INFO) << all_compute_time_vec[i] << " ";
      }
      logstream(LOG_INFO) << std::endl;
    } 
    rmi.full_barrier();
    // Stop the aggregator
    aggregator.stop();
    started = false;
    // return the final reason for termination
    return termination_reason;
  } // end of start

  
  
  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  exchange_messages(const size_t thread_id) {
    context_type context(*this, graph);
    const bool TRY_TO_RECV = true;
    const size_t TRY_RECV_MOD = 100;
    size_t vcount = 0;
    fixed_dense_bitset<sizeof(size_t)> local_bitset;
    // for(lvid_type lvid = thread_id; lvid < graph.num_local_vertices(); 
    //     lvid += threads.size()) {
    while (1) {
      // increment by a word at a time 
      lvid_type lvid_block_start = 
                  shared_lvid_counter.inc_ret_last(8 * sizeof(size_t));
      if (lvid_block_start >= graph.num_local_vertices()) break;
      // get the bit field from has_message
      size_t lvid_bit_block = has_remote_message.containing_word(lvid_block_start);
      if (lvid_bit_block == 0) continue;
      // initialize a word sized bitfield 
      local_bitset.clear();
      local_bitset.initialize_from_mem(&lvid_bit_block, sizeof(size_t));
      foreach(size_t lvid_block_offset, local_bitset) {
        lvid_type lvid = lvid_block_start + lvid_block_offset; 
        if (lvid >= graph.num_local_vertices()) break;
        // if the vertex is not local and has a message send the
        // message and clear the bit
        message_type msg;
        ASSERT_FALSE(graph.l_is_master(lvid)); 
        if(scheduler_ptr->get_specific(lvid, msg) != sched_status::EMPTY) {
          sync_message(lvid, thread_id, msg); 
        }
        if(++vcount % TRY_RECV_MOD == 0) recv_messages(thread_id, TRY_TO_RECV); 
      }
    } // end of loop over vertices to send messages
    message_exchange.partial_flush(thread_id);
    // Finish sending and receiving all messages
    rmi.dc().start_handler_threads(thread_id, threads.size());
    thread_barrier.wait();
    if(thread_id == 0) message_exchange.flush(); 
    thread_barrier.wait();
    rmi.dc().stop_handler_threads(thread_id, threads.size());
    recv_messages(thread_id);
  } // end of exchange_messages




  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  transfer_scheduler_to_active(const size_t thread_id) {
    context_type context(*this, graph);
    const bool TRY_TO_RECV = true;
    const size_t TRY_RECV_MOD = 100;
    size_t vcount = 0;
    size_t curthread_num_to_activate = num_to_activate / threads.size();
    curthread_num_to_activate += (curthread_num_to_activate == 0);
    size_t nactive_inc = 0;
    while (nactive_inc < curthread_num_to_activate) {
      lvid_type lvid;
      message_type msg;
      sched_status::status_enum stat =
          scheduler_ptr->get_next(thread_id, lvid, msg);
      bool has_sched_msg = stat != sched_status::EMPTY;
      if (has_sched_msg) {
        // if the vertex is not local and has a message send the
        // message and clear the bit
        ASSERT_TRUE(graph.l_is_master(lvid));
        nactive_inc++;
        // The vertex becomes active for this superstep 
        // Pass the message to the vertex program
        active_superstep_pushback.push_back(lvid);
        vertex_type vertex = vertex_type(graph.l_vertex(lvid));
        vertex_programs[lvid].init(context, vertex, msg);
        // Determine if the gather should be run
        const vertex_program_type& const_vprog = vertex_programs[lvid];
        const vertex_type const_vertex = vertex;
        if(const_vprog.gather_edges(context, const_vertex) != 
           graphlab::NO_EDGES) {
          active_minorstep_pushback.push_back(lvid);
          sync_vertex_program(lvid, thread_id);
        }  
        if (++vcount % TRY_RECV_MOD == 0) {
          // to avoid popping the same task multiple times, it is
          // of critical importance that we do not recv_message here.
          // but only recv_messages at the end of the phase
          // recv_messages(TRY_TO_RECV); 
          recv_vertex_programs(thread_id, TRY_TO_RECV);
        }
      } else {
        break;
      }
    } // end of loop over vertices to send messages
    vprog_exchange.partial_flush(thread_id);
    num_active_vertices.inc(nactive_inc); 
    // Finish sending and receiving all messages
    rmi.dc().start_handler_threads(thread_id, threads.size());
    thread_barrier.wait();
    if(thread_id == 0) {
      vprog_exchange.flush();
    }
    thread_barrier.wait();
    rmi.dc().stop_handler_threads(thread_id, threads.size());
    recv_vertex_programs(thread_id);
  } // end of exchange_messages


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  execute_gathers(const size_t thread_id) {
    context_type context(*this, graph);
    const bool TRY_TO_RECV = true;
    const size_t TRY_RECV_MOD = 1000;
    size_t vcount = 0;
    const bool caching_enabled = !gather_cache.empty();
    timer ti;

    size_t numminorstep = active_minorstep_pushback.size();
    while(1) { 
      size_t i = shared_lvid_counter.inc_ret_last();
      if (i >= numminorstep) break;
      lvid_type lvid = active_minorstep[i];

      bool accum_is_set = false;
      gather_type accum = gather_type();         
      // if caching is enabled and we have a cache entry then use
      // that as the accum
      if( caching_enabled && has_cache.get(lvid) ) {
        accum = gather_cache[lvid];
        accum_is_set = true;
      } else {
        // recompute the local contribution to the gather
        const vertex_program_type& vprog = vertex_programs[lvid];
        local_vertex_type local_vertex = graph.l_vertex(lvid);
        const vertex_type vertex(local_vertex);
        const edge_dir_type gather_dir = vprog.gather_edges(context, vertex);
        // Loop over in edges
        size_t edges_touched = 0;
        vprog.pre_local_gather(accum); 
        if(gather_dir == IN_EDGES || gather_dir == ALL_EDGES) {
          foreach(local_edge_type local_edge, local_vertex.in_edges()) {
            edge_type edge(local_edge);
            // elocks[local_edge.id()].lock();
            if(accum_is_set) { // \todo hint likely                
              accum += vprog.gather(context, vertex, edge);
            } else {
              accum = vprog.gather(context, vertex, edge); 
              accum_is_set = true;
            }
            ++edges_touched;
            // elocks[local_edge.id()].unlock();
          }
        } // end of if in_edges/all_edges
        // Loop over out edges
        if(gather_dir == OUT_EDGES || gather_dir == ALL_EDGES) {
          foreach(local_edge_type local_edge, local_vertex.out_edges()) {
            edge_type edge(local_edge);
            // elocks[local_edge.id()].lock();
            if(accum_is_set) { // \todo hint likely
              accum += vprog.gather(context, vertex, edge);              
            } else {
              accum = vprog.gather(context, vertex, edge);
              accum_is_set = true;
            }
            // elocks[local_edge.id()].unlock();
            ++edges_touched;
          }
          INCREMENT_EVENT(EVENT_GATHERS, edges_touched);
        } // end of if out_edges/all_edges
        vprog.post_local_gather(accum); 
        // If caching is enabled then save the accumulator to the
        // cache for future iterations.  Note that it is possible
        // that the accumulator was never set in which case we are
        // effectively "zeroing out" the cache.
        if(caching_enabled && accum_is_set) {              
          gather_cache[lvid] = accum; has_cache.set_bit(lvid); 
        } // end of if caching enabled            
      }
      // If the accum contains a value for the local gather we put
      // that estimate in the gather exchange.
      if(accum_is_set) sync_gather(lvid, accum, thread_id);  
      if(!graph.l_is_master(lvid)) {
        // if this is not the master clear the vertex program
        vertex_programs[lvid] = vertex_program_type();
      }

      // try to recv gathers if there are any in the buffer
      if(++vcount % TRY_RECV_MOD == 0) recv_gathers(thread_id, TRY_TO_RECV);
    } // end of loop over vertices to compute gather accumulators
    per_thread_compute_time[thread_id] += ti.current_time();
    gather_exchange.partial_flush(thread_id);
      // Finish sending and receiving all gather operations
    rmi.dc().start_handler_threads(thread_id, threads.size());
    thread_barrier.wait();
    if(thread_id == 0) gather_exchange.flush();
    thread_barrier.wait();
    rmi.dc().stop_handler_threads(thread_id, threads.size());
    recv_gathers(thread_id);
  } // end of execute_gathers


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  execute_applys(const size_t thread_id) {
    context_type context(*this, graph);
    const bool TRY_TO_RECV = true;
    const size_t TRY_RECV_MOD = 1000;
    size_t vcount = 0;
    timer ti;

    size_t numactive = active_superstep_pushback.size();
    while(1) { 
      size_t i = shared_lvid_counter.inc_ret_last();
      if (i >= numactive) break;
      lvid_type lvid = active_superstep[i];
      // Only master vertices can be active in a super-step
      ASSERT_TRUE(graph.l_is_master(lvid));
      vertex_type vertex(graph.l_vertex(lvid));
      // Get the local accumulator.  Note that it is possible that
      // the gather_accum was not set during the gather.
      const gather_type& accum = gather_accum[lvid];
      INCREMENT_EVENT(EVENT_APPLIES, 1);
      vertex_programs[lvid].apply(context, vertex, accum);
      // record an apply as a completed task
      ++completed_applys;
      // Clear the accumulator to save some memory
      gather_accum[lvid] = gather_type();
      // synchronize the changed vertex data with all mirrors
      sync_vertex_data(lvid, thread_id);  
      // determine if a scatter operation is needed
      const vertex_program_type& const_vprog = vertex_programs[lvid];
      const vertex_type const_vertex = vertex;
      if(const_vprog.scatter_edges(context, const_vertex) != 
         graphlab::NO_EDGES) {
        active_minorstep_pushback.push_back(lvid);
        sync_vertex_program(lvid, thread_id);
      } else { // we are done so clear the vertex program
        vertex_programs[lvid] = vertex_program_type();
      }
      // try to receive vertex data
      if(++vcount % TRY_RECV_MOD == 0) {
        recv_vertex_programs(thread_id, TRY_TO_RECV);
        recv_vertex_data(thread_id, TRY_TO_RECV); 
      }
    } // end of loop over vertices to run apply

    per_thread_compute_time[thread_id] += ti.current_time();
    vprog_exchange.partial_flush(thread_id);
    vdata_exchange.partial_flush(thread_id);
    // Finish sending and receiving all changes due to apply operations
    rmi.dc().start_handler_threads(thread_id, threads.size());
    thread_barrier.wait();
    if(thread_id == 0) { 
      vprog_exchange.flush(); vdata_exchange.flush(); 
    }
    thread_barrier.wait();
    rmi.dc().stop_handler_threads(thread_id, threads.size());
    recv_vertex_programs(thread_id);
    recv_vertex_data(thread_id);

  } // end of execute_applys




  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  execute_scatters(const size_t thread_id) {
    context_type context(*this, graph);
    timer ti;
    
    size_t numminorstep = active_minorstep_pushback.size();
    while(1) { 
      size_t i = shared_lvid_counter.inc_ret_last();
      if (i >= numminorstep) break;
      lvid_type lvid = active_minorstep[i];

      const vertex_program_type& vprog = vertex_programs[lvid];
      local_vertex_type local_vertex = graph.l_vertex(lvid);
      const vertex_type vertex(local_vertex);
      const edge_dir_type scatter_dir = vprog.scatter_edges(context, vertex);
      size_t edges_touched = 0; 
      // Loop over in edges
      if(scatter_dir == IN_EDGES || scatter_dir == ALL_EDGES) {
        foreach(local_edge_type local_edge, local_vertex.in_edges()) {
          edge_type edge(local_edge);
          // elocks[local_edge.id()].lock();
          vprog.scatter(context, vertex, edge);
          // elocks[local_edge.id()].unlock();
        }
        ++edges_touched;
      } // end of if in_edges/all_edges
      // Loop over out edges
      if(scatter_dir == OUT_EDGES || scatter_dir == ALL_EDGES) {
        foreach(local_edge_type local_edge, local_vertex.out_edges()) {
          edge_type edge(local_edge);
          // elocks[local_edge.id()].lock();
          vprog.scatter(context, vertex, edge);
          // elocks[local_edge.id()].unlock();
        }
        ++edges_touched;
      } // end of if out_edges/all_edges
      INCREMENT_EVENT(EVENT_SCATTERS, edges_touched);
      // Clear the vertex program
      vertex_programs[lvid] = vertex_program_type();
    } // end of loop over vertices to complete scatter operation

    per_thread_compute_time[thread_id] += ti.current_time();
  } // end of execute_scatters



  // Data Synchronization ===================================================
  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  sync_vertex_program(lvid_type lvid, const size_t thread_id) {
    ASSERT_TRUE(graph.l_is_master(lvid));
    const vertex_id_type vid = graph.global_vid(lvid);
    local_vertex_type vertex = graph.l_vertex(lvid);
    foreach(const procid_t& mirror, vertex.mirrors()) {
      vprog_exchange.send(mirror, 
                          std::make_pair(vid, vertex_programs[lvid]),
                          thread_id);
    }
  } // end of sync_vertex_program



  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  recv_vertex_programs(size_t threadid, const bool try_to_recv) {
    rmi.dc().handle_incoming_calls(threadid, threads.size());
    procid_t procid(-1);
    typename vprog_exchange_type::buffer_type buffer;
    while(vprog_exchange.recv(procid, buffer, try_to_recv)) {
      foreach(const vid_prog_pair_type& pair, buffer) {
        const lvid_type lvid = graph.local_vid(pair.first);
  //      ASSERT_FALSE(graph.l_is_master(lvid));
        vertex_programs[lvid] = pair.second;
        active_minorstep_pushback.push_back(lvid);
      }
    }
  } // end of recv vertex programs


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  sync_vertex_data(lvid_type lvid, const size_t thread_id) {
    ASSERT_TRUE(graph.l_is_master(lvid));
    const vertex_id_type vid = graph.global_vid(lvid);
    local_vertex_type vertex = graph.l_vertex(lvid);
    foreach(const procid_t& mirror, vertex.mirrors()) {
      vdata_exchange.send(mirror, std::make_pair(vid, vertex.data()), thread_id);
    }
  } // end of sync_vertex_data





  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  recv_vertex_data(size_t threadid, bool try_to_recv) {
    rmi.dc().handle_incoming_calls(threadid, threads.size());
    procid_t procid(-1);
    typename vdata_exchange_type::buffer_type buffer;
    while(vdata_exchange.recv(procid, buffer, try_to_recv)) {
      foreach(const vid_vdata_pair_type& pair, buffer) {
        const lvid_type lvid = graph.local_vid(pair.first);
        ASSERT_FALSE(graph.l_is_master(lvid));
        graph.l_vertex(lvid).data() = pair.second;
      }
    }
  } // end of recv vertex data


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  sync_gather(lvid_type lvid, const gather_type& accum, const size_t thread_id) {
    if(graph.l_is_master(lvid)) {
      vlocks[lvid].lock();
      if(has_gather_accum.get(lvid)) {
        gather_accum[lvid] += accum;
      } else {
        gather_accum[lvid] = accum;
        has_gather_accum.set_bit(lvid);
      }
      vlocks[lvid].unlock();
    } else {
      const procid_t master = graph.l_master(lvid);
      const vertex_id_type vid = graph.global_vid(lvid);
      gather_exchange.send(master, std::make_pair(vid, accum), thread_id);
    }
  } // end of sync_gather

  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  recv_gathers(size_t threadid, const bool try_to_recv) {
    rmi.dc().handle_incoming_calls(threadid, threads.size());
    procid_t procid(-1);
    typename gather_exchange_type::buffer_type buffer;
    while(gather_exchange.recv(procid, buffer, try_to_recv)) {
      foreach(const vid_gather_pair_type& pair, buffer) {
        const lvid_type lvid = graph.local_vid(pair.first);
        const gather_type& accum = pair.second;
        ASSERT_TRUE(graph.l_is_master(lvid));
        vlocks[lvid].lock();
        if( has_gather_accum.get(lvid) ) {
          gather_accum[lvid] += accum;
        } else {
          gather_accum[lvid] = accum;
          has_gather_accum.set_bit(lvid);
        }
        vlocks[lvid].unlock();
      }
    }
  } // end of recv_gather


  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  sync_message(lvid_type lvid, const size_t thread_id, const message_type& msg) {
    ASSERT_FALSE(graph.l_is_master(lvid));
    const procid_t master = graph.l_master(lvid);
    const vertex_id_type vid = graph.global_vid(lvid);
    message_exchange.send(master, std::make_pair(vid, msg), thread_id);
  } // end of send_message




  template<typename VertexProgram>
  void semi_synchronous_engine<VertexProgram>::
  recv_messages(size_t threadid, const bool try_to_recv) {
    rmi.dc().handle_incoming_calls(threadid, threads.size());
    procid_t procid(-1);
    typename message_exchange_type::buffer_type buffer;
    while(message_exchange.recv(procid, buffer, try_to_recv)) {
      foreach(const vid_message_pair_type& pair, buffer) {
        internal_signal(graph.vertex(pair.first), pair.second);
      }
    }
  } // end of recv_messages











}; // namespace


#include <graphlab/macros_undef.hpp>

#endif