Only send one price update per price per slot

Author: tompntn

pc/manager.cpp

@@ -847,6 +847,14 @@ void manager::on_response( rpc::get_slot *res )
     cap_.flush();
   }
 
+  // mark all prices as unsent in the current slot
+  for (size_t i = 0; i < svec_.size(); i++) {
+    product* product = svec_[i];
+    for (size_t j = 0; j < product->get_num_price(); j++) {
+      product->get_price( j )->set_sent_in_current_slot( false );
+    }
+  }
+
   if (
     has_status( PC_PYTH_RPC_CONNECTED )
   ) {
@@ -1021,6 +1029,10 @@ price *manager::get_price( const pub_key& acc )
 
 void manager::add_dirty_price(price* sptr)
 {
+  if ( sptr->get_sent_in_current_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),
+  sent_in_curr_slot_( false )
 {
   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_ );
 }
 
+bool price::get_sent_in_current_slot() const
+{
+  return sent_in_curr_slot_;
+}
+
+void price::set_sent_in_current_slot( bool sent )
+{
+  sent_in_curr_slot_ = sent;
+}
+
 uint64_t price::get_prev_slot() const
 {
   return pptr_->prev_slot_;
@@ -738,6 +749,10 @@ 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();
+
+    // Mark that we have attempted to send the price update in the current slot
+    p->set_sent_in_current_slot( true );
+
     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;
 
+    // a price update has been attempted in the current slot
+    bool get_sent_in_current_slot() const;
+    void set_sent_in_current_slot( bool );
+
     // 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_;
+    bool                   sent_in_curr_slot_;
   };
 
   template<class T>