better approach

Author: Jayant Krishnamurthy

pc/manager.cpp

@@ -507,10 +507,8 @@ void manager::poll( bool do_wait )
     poll_schedule();
 
     // Flush any pending complete batches of price updates by submitting solana TXs.
-    for( user *uptr = olist_.first(); uptr; uptr = uptr->get_next() ) {
-      if (uptr->num_pending_upds() >= get_max_batch_size()) {
-        uptr->send_pending_upds(get_max_batch_size());
-      }
+    for ( user *uptr = olist_.first(); uptr; uptr = uptr->get_next() ) {
+      uptr->send_pending_upds();
     }
   } else {
     reconnect_rpc();

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;
 
@@ -27,7 +29,8 @@ void user::user_http::parse_content( const char *txt, size_t len )
 user::user()
 : rptr_( nullptr ),
   sptr_( nullptr ),
-  psub_( this )
+  psub_( this ),
+  last_update_ts_(0)
 {
   // setup the plumbing
   hsvr_.ptr_ = this;
@@ -330,27 +333,26 @@ void user::parse_get_product( uint32_t tok, uint32_t itok )
   add_tail( itok );
 }
 
-uint32_t user::num_pending_upds()
+void user::send_pending_upds()
 {
-  return pending_vec_.size();
-}
-
-void user::send_pending_upds(uint32_t n)
-{
-  if ( pending_vec_.empty() ) {
-    return;
+  uint32_t n_sent = 0;
+  int64_t curr_ts = get_now();
+  if (curr_ts_ - price_upd_ts_ > PC_FLUSH_INTERVAL) {
+    n_sent = pending_vec_.size();
+  } else if (pending_vec_.size() > max_batch_size_) {
+    n_sent = max_batch_size_;
   }
 
-  uint32_t n_sent = n;
-  if (pending_vec_.size() < n) {
-    n_sent = pending_vec_.size();
+  if (n_sent == 0) {
+    return;
   }
 
   if ( !price::send( pending_vec_.data(), n_sent) ) {
     add_error( 0, PC_BATCH_SEND_FAILED, "batch send failed - please check the pyth logs" );
   }
 
   pending_vec_.erase(pending_vec_.begin(), pending_vec_.begin() + n_sent);
+  price_upd_ts_ = curr_ts_;
 }
 
 void user::parse_get_all_products( uint32_t itok )

pc/user.hpp

@@ -43,12 +43,9 @@ namespace pc
     // symbol price schedule callback
     void on_response( price_sched *, uint64_t ) override;
 
-    // Get the number of pending price updates that are queued to be sent.
-    uint32_t num_pending_upds();
-
-    // send up to n pending price updates. If n > the max batch size,
-    // this will split the price updates into multiple transactions.
-    void send_pending_upds(uint32_t n);
+    // 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.
+    void send_pending_upds();
 
   private:
 
@@ -89,6 +86,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
   };
 
 }