Only send one price update per price per slot

Author: tompntn

pc/manager.cpp

@@ -1021,6 +1021,11 @@ price *manager::get_price( const pub_key& acc )
 
 void manager::add_dirty_price(price* sptr)
 {
+  // Skip this update if we have already attempted an update in this slot.
+  if ( sptr->get_last_attempted_update_slot() == get_slot() ) {
+    return;
+  }
+
   if( std::find(pending_upds_.begin(), pending_upds_.end(), sptr) == pending_upds_.end() ) {
     pending_upds_.emplace_back( sptr );
   }

pc/request.cpp

@@ -458,7 +458,8 @@ price::price( const pub_key& acc, product *prod )
   prod_( prod ),
   sched_( this ),
   pinit_( this ),
-  pptr_(nullptr)
+  pptr_(nullptr),
+  last_attempted_update_slot_( 0UL )
 {
   areq_->set_account( &apub_ );
   preq_->set_account( &apub_ );
@@ -558,6 +559,16 @@ uint64_t price::get_twac() const
   return static_cast< uint64_t >( pptr_->twac_.val_ );
 }
 
+uint64_t price::get_last_attempted_update_slot() const
+{
+  return last_attempted_update_slot_;
+}
+
+void price::set_last_attempted_update_slot( uint64_t slot )
+{
+  last_attempted_update_slot_ = slot;
+}
+
 uint64_t price::get_prev_slot() const
 {
   return pptr_->prev_slot_;
@@ -738,6 +749,7 @@ bool price::send( price *prices[], const unsigned n )
   for ( unsigned i = 0, j = 0; i < n; ++i ) {
     price *const p = prices[ i ];
     manager *const mgr = p->get_manager();
+    p->set_last_attempted_update_slot( mgr->get_slot() );
     if ( PC_UNLIKELY( ! p->init_ && ! p->init_publish() ) ) {
       PC_LOG_ERR( "failed to initialize publisher" )
         .add( "secondary", mgr->get_is_secondary() )

pc/request.hpp

@@ -265,6 +265,10 @@ namespace pc
     // ready to publish (i.e. not waiting for confirmation)
     bool get_is_ready_publish() const;
 
+    // the slot when a price update was last attempted
+    uint64_t get_last_attempted_update_slot() const;
+    void set_last_attempted_update_slot( uint64_t );
+
     // submit new price update and update aggregate
     // will fail with false if in error (check get_is_err() )
     // or because symbol is not ready to publish (get_is_ready_publish())
@@ -359,6 +363,7 @@ namespace pc
     rpc::upd_price         preq_[1];
     pc_price_t            *pptr_;
     txid_vec_t             tvec_;
+    uint64_t               last_attempted_update_slot_;
   };
 
   template<class T>