Flush batches early (#167)

* add code to flush batches early

* fix build

* fix build

* fix build

* comments

* fix test?

* better approach

* fix

* fix build

* fix

* pr comments

Co-authored-by: Jayant Krishnamurthy <jkrishnamurthy@jumptrading.com>

Author: jayantk

pc/manager.cpp

@@ -500,11 +500,16 @@ void manager::poll( bool do_wait )
     tconn_.reconnect();
   }
 
-  // submit new quotes while connected
   if ( has_status( PC_PYTH_RPC_CONNECTED ) &&
        !hconn_.get_is_err() &&
        ( !wconn_ || !wconn_->get_is_err() ) ) {
+    // request product quotes from pythd's clients while connected
     poll_schedule();
+
+    // Flush any pending complete batches of price updates by submitting solana TXs.
+    for ( user *uptr = olist_.first(); uptr; uptr = uptr->get_next() ) {
+      uptr->send_pending_upds();
+    }
   } else {
     reconnect_rpc();
   }
@@ -758,11 +763,6 @@ void manager::on_response( rpc::get_slot *res )
   if (
     has_status( PC_PYTH_RPC_CONNECTED )
   ) {
-      // New slot received, so flush all pending updates for all active users
-      for( user *uptr = olist_.first(); uptr; uptr = uptr->get_next() ) {
-        uptr->send_pending_upds();
-      }
-
     if ( sub_ ) {
       sub_->on_slot_publish( this );
     }

pc/user.cpp

@@ -13,6 +13,8 @@
 #define PC_JSON_MISSING_PERMS   -32001
 #define PC_JSON_NOT_READY       -32002
 #define PC_BATCH_SEND_FAILED    -32010
+// Flush partial batches if not completed within 400 ms.
+#define PC_FLUSH_INTERVAL       (400L*PC_NSECS_IN_MSEC)
 
 using namespace pc;
 
@@ -34,6 +36,7 @@ user::user()
   hsvr_.set_net_connect( this );
   hsvr_.set_ws_parser( this );
   set_net_parser( &hsvr_ );
+  last_upd_ts_ = get_now();
 }
 
 void user::set_rpc_client( rpc_client *rptr )
@@ -332,15 +335,24 @@ void user::parse_get_product( uint32_t tok, uint32_t itok )
 
 void user::send_pending_upds()
 {
-  if ( pending_vec_.empty() ) {
+  uint32_t n_to_send = 0;
+  int64_t curr_ts = get_now();
+  if (curr_ts - last_upd_ts_ > PC_FLUSH_INTERVAL) {
+    n_to_send = pending_vec_.size();
+  } else if (pending_vec_.size() >= sptr_->get_max_batch_size()) {
+    n_to_send = sptr_->get_max_batch_size();
+  }
+
+  if (n_to_send == 0) {
     return;
   }
 
-  if ( !price::send( pending_vec_.data(), pending_vec_.size()) ) {
+  if ( !price::send( pending_vec_.data(), n_to_send) ) {
     add_error( 0, PC_BATCH_SEND_FAILED, "batch send failed - please check the pyth logs" );
   }
 
-  pending_vec_.clear();
+  pending_vec_.erase(pending_vec_.begin(), pending_vec_.begin() + n_to_send);
+  last_upd_ts_ = curr_ts;
 }
 
 void user::parse_get_all_products( uint32_t itok )

pc/user.hpp

@@ -43,7 +43,10 @@ namespace pc
     // symbol price schedule callback
     void on_response( price_sched *, uint64_t ) override;
 
-    // send all pending updates
+    // send a batch of pending price updates. This function eagerly sends any complete batches.
+    // It also sends partial batches that have not been completed within a short interval of time.
+    // At most one complete batch will be sent. Additional price updates remain queued until the next
+    // time this function is invoked.
     void send_pending_upds();
 
   private:
@@ -85,6 +88,7 @@ namespace pc
     def_vec_t       dvec_;        // deferred subscriptions
     request_sub_set psub_;        // price subscriptions
     pending_vec_t   pending_vec_; // prices with pending updates
+    int64_t         last_upd_ts_; // timestamp of last price update transaction
   };
 
 }