/**
  * Receive all messages and compute the new belief. 
  */
 inline void update_belief(const vertex_type& vertex,
                           StateManager& state) {
   
   // Get the belief from the state manager
   belief_type* blf = state.checkout_belief(vertex);
   // Wipe out the old value for the belief
   if(vertex.is_variable()) {
     *blf = 1;
   } else if(vertex.is_factor()) {
     *blf = vertex.factor();
   } else {
     assert(false);
   }
   // For each of the neighbor variables 
   foreach(const vertex_type& vertex_source, state.neighbors(vertex)) {
     // get the in message
     message_type* in_msg = 
       state.try_checkout(vertex_source, vertex, Reading);
     if(in_msg != NULL) {
       // Combine the in_msg with the destination factor
       blf->combine_in(*in_msg, csr_.dot_op);
       // return the message to the state manager
       state.checkin(vertex_source, vertex, in_msg);
       // normalize the belief
       blf->normalize();
     }
   }  
   // Do an extra normalization (just in case no messages were
   // available)
   blf->normalize();
   // ASSERT WE BLF IS A VALID DISTRIBUTION (we should check this)
   // Save the belief
   state.checkin_belief(vertex, blf);
 }// End of update belief    

 /** The gather function computes XtX and Xy */
 gather_type gather(icontext_type& context, const vertex_type& vertex, 
                    edge_type& edge) const {
   if(edge.data().role == edge_data::TRAIN) {
     const vertex_type other_vertex = get_other_vertex(edge, vertex);
     return gather_type(other_vertex.data().factor, edge.data().obs);
   } else return gather_type();
 } // end of gather function

 /**
  * \brief If the distance is smaller then update
  */
 void apply(icontext_type& context, vertex_type& vertex,
            const graphlab::empty& empty) {
     changed = false;
     if(vertex.data().dist > min_dist) {
         changed = true;
         vertex.data().dist = min_dist;
     }
 }

    void scatter(icontext_type& context, const vertex_type& vertex,
            edge_type& edge) const {
        const vertex_type other = edge.target();
        pagerank_type value = vertex.data().pagerank / vertex.num_out_edges();
        assert(other.id() != vertex.id());
        const sum_pagerank_type msg(value);
        context.signal(other, msg);

    }

 void apply(icontext_type& context, vertex_type& vertex,
            const gather_type& total)
 {
     changed = false;
     if (vertex.data().color > total.color) {
         changed = true;
         vertex.data().color = total.color;
     }
 }

 // Change the vertex data if any of its neighbors have a lower data value.
 void apply(icontext_type& context, vertex_type& vertex, const gather_type& total) {
   // mark if values differ to determine which edges to scatter on.
   if (message_value < vertex.data()) {
       changed = true;
       vertex.data() = message_value;
   } else {
       changed = false;
   }
 }

 /**
  * \brief The scatter function just signal adjacent pages
  */
 void scatter(icontext_type& context, const vertex_type& vertex,
              edge_type& edge) const {
     const vertex_type other = get_other_vertex(edge, vertex);
     distance_type newd = vertex.data().dist + edge.data().dist;
     if (other.data().dist > newd) {
         const min_distance_type msg(newd);
         context.signal(other, msg);
     }
 } // end of scatter

        void apply(icontext_type& context, vertex_type& vertex, const gather_type &total) {
            vertex_data_type new_label = most_common(total);

            if (new_label != vertex.data()) {
                vertex.data() = new_label;
                changed = true;
            } else {
                changed = false;
            }
        }

  // Scatter to scatter_edges edges with the new message value.
  void scatter(icontext_type& context, const vertex_type& vertex, edge_type& edge) const {
    bool isEdgeSource = (vertex.id() == edge.source().id());
    bool hasSameData = isEdgeSource ? (vertex.data() == edge.target().data()) : (vertex.data() == edge.source().data()) ;
    if (!hasSameData) {
      min_combiner combiner;
      combiner.value = message_value;

      context.signal(isEdgeSource ? edge.target() : edge.source(), combiner);
    }
  }

 void apply(icontext_type& context, vertex_type& vertex,
            const gather_type& total)
 {
     converged = true;
     double new_pagerank = 0.15 + 0.85 * total.pagerank;
     double delta = fabs(vertex.data().pagerank - new_pagerank);
     vertex.data().pagerank = new_pagerank;
     if (delta > EPS) {
         converged = false;
     }
 }

 void apply(icontext_type& context, vertex_type& vertex,
            const factor_type& sum) {
   const size_t num_neighbors = vertex.num_in_edges() + vertex.num_out_edges();
   ASSERT_GT(num_neighbors, 0);
   // There should be no new edge data since the vertex program has been cleared
   vertex_data& vdata = vertex.data();
   ASSERT_EQ(sum.size(), NTOPICS);
   ASSERT_EQ(vdata.factor.size(), NTOPICS);
   vdata.nupdates++; vdata.nchanges = 0; 
   vdata.factor = sum;
 } // end of apply

        void apply(icontext_type& context, vertex_type& vertex, const gather_type &total) {
            if (context.iteration() == 0) {
                vertex.data().neighbors = total.get();
            } else {
                size_t d = vertex.data().neighbors.size();
                size_t t = last_msg;

                // Due to rounding errors, the results is sometimes not exactly
                // 0.0 even when it should be. Explicitly set LCC to 0 if that
                // is the case, other calculate it as tri / (degree * (degree - 1))
                double lcc = (d < 2 || t == 0) ? 0.0 : double(t) / (d * (d - 1));

                vertex.data().clustering_coef = lcc;
            }
        }

    /* Use the total rank of adjacent pages to update this page */
    void apply(icontext_type& context, vertex_type& vertex,
        const double& total) {

      //printf("Entered apply on node %d value %lg\n", vertex.id(), total);
      vertex_data & user = vertex.data();
      assert(mi.x_offset >=0 || mi.y_offset >= 0);
      assert(mi.r_offset >=0);

      /* perform orthogonalization of current vector */
      if (mi.orthogonalization){
         for (int i=mi.mat_offset; i< mi.vec_offset; i++){
            vertex.data().pvec[mi.vec_offset] -= alphas.pvec[i-mi.mat_offset] * vertex.data().pvec[i]; 
         }
         return;
      }

      double val = total;
      //assert(total != 0 || mi.y_offset >= 0);

      //store previous value for convergence detection
      if (mi.prev_offset >= 0)
        user.pvec[mi.prev_offset ] = user.pvec[mi.r_offset];

      assert(mi.x_offset >=0 || mi.y_offset>=0);
      if (mi.A_offset  && mi.x_offset >= 0){
        if  (info.is_square() && mi.use_diag)// add the diagonal term
          val += (/*mi.c**/ (user.A_ii+ regularization) * user.pvec[mi.x_offset]);
        //printf("node %d added diag term: %lg\n", vertex.id(), user.A_ii);
        val *= mi.c;
      }
      /***** COMPUTE r = c*I*x  *****/
      else if (!mi.A_offset && mi.x_offset >= 0){
        val = mi.c*user.pvec[mi.x_offset];
      }

      /**** COMPUTE r+= d*y (optional) ***/
      if (mi.y_offset>= 0){
        val += mi.d*user.pvec[mi.y_offset]; 
      }

      /***** compute r = (... ) / div */
      if (mi.div_offset >= 0){
        val /= user.pvec[mi.div_offset];
      }

      user.pvec[mi.r_offset] = val;
      //printf("Exit apply on node %d value %lg\n", vertex.id(), val);
    }

	gather_type gather(icontext_type& context, const vertex_type& vertex, edge_type& edge) const
	{
		cout << "gather(), edge=" << edge.source().id() << "->" << edge.target().id() << ", called from vid=" << vertex.id() << endl;
		
		gather_type gathered;
		
		// add id of other vertex of edge and add id->beta to map if message source
		if (edge.target().id()==vertex.id())
		{
			// incoming edge, outgoing message, only if target is non-observed
			if (!edge.source().data().is_observed)
			{
				gathered.message_targets.insert(edge.source().id());
				cout << "added " << edge.source().id() << " as message target" << endl;
			}
		}
		else
		{
			// outgoing edge, incoming message with beta
			gathered.message_source_betas[edge.target().id()]=edge.data().beta;
			cout << "added " << edge.target().id() << " as message source" << endl;	
		}
					
		cout << "gathered=" << gathered << endl;
		
		return gathered;
	}

	edge_dir_type scatter_edges(icontext_type& context, const vertex_type& vertex) const {
		if( vertex.data().changed == 1 ) {
			return OUT_EDGES;
		}
		else {
			return NO_EDGES;
		}
	}

 void aggregate(icontext_type& context, const vertex_type& vertex){
     if (!marked) {
         marked = true;
         saedb::IAggregator* active_node = context.getAggregator("active_node");
         float t = vertex.data();
         active_node->reduce(&t);
     }
 }

    void scatter(icontext_type& context, const vertex_type& vertex,
                 edge_type& edge) const
    {
        const vertex_type other = edge.target();
        distance_type newd = vertex.data().dist + edge.data().dist;

        const min_distance_type msg(newd);
        context.signal(other, msg);
    }

  // Receive inbound message (minimum data of adjacent vertices)
  void init(icontext_type& context, const vertex_type& vertex, const message_type& message) {
    // message.value == 4294967295 on first run, so init message_value to vertex data.
    if (message.value == 4294967295) {
      message_value = vertex.id();
    } else {
      // else, set the local copy to the message parameter.
      message_value = message.value;
    }

  }

  /**
   * \brief If the distance is smaller then update
   */
  void apply(icontext_type& context, vertex_type& vertex,
             const gather_type& total) {
    changed = false;
    if(context.iteration() == 0) {
    	changed = true;
    	vertex.data().dist = 0;
    	context.setUpdateFlag(changed);
    	return;
    	//lastchange = vertex.data().dist;
    }
    if(vertex.data().dist > total.dist) {
      changed = true;
      vertex.data().dist = total.dist;
      //lastchange = vertex.data().dist;
    }

    context.setUpdateFlag(changed);
    //std::cout << "vid=" << vertex.id() << ", val=" << vertex.data().dist << "\n";
  }

 gather_type gather(icontext_type& context, const vertex_type& vertex, edge_type& edge) const {
     if (context.iteration() == 0) {
         if (vertex.id() == edge.source().id()) {
             return gather_type(edge.target().id());
         } else {
             return gather_type(edge.source().id());
         }
     } else {
         return gather_type();
     }
 }