Batch price updates across all websocket users (#175)

* Batch price updates across all websocket users

* Fully move batching out of user, remove error notification

* Restore old flushing behavior, refactor to another function

* Apply naming and documentation suggestions

Co-authored-by: Karol Skrzypczynski <146karol@gmail.com>

Author: wijagels

pc/manager.cpp

@@ -1,6 +1,8 @@
 #include "manager.hpp"
 #include "log.hpp"
 
+#include <algorithm>
+
 using namespace pc;
 
 #define PC_TPU_PROXY_PORT     8898
@@ -10,6 +12,8 @@ using namespace pc;
 #define PC_PUB_INTERVAL       (227L*PC_NSECS_IN_MSEC)
 #define PC_RPC_HOST           "localhost"
 #define PC_MAX_BATCH          8
+// Flush partial batches if not completed within 400 ms.
+#define PC_FLUSH_INTERVAL       (400L*PC_NSECS_IN_MSEC)
 
 ///////////////////////////////////////////////////////////////////////////
 // manager_sub
@@ -437,6 +441,34 @@ void manager::reset_status( int status )
   status_ &= ~status;
 }
 
+void manager::send_pending_ups()
+{
+  uint32_t n_to_send = 0;
+
+  // the batch will be sent if its size is greater than max batch size
+  // or time since the previously sent batch is greater than PC_FLUSH_INTERVAL
+  // the buffer is being updated by user class un user::parse_upd_price
+  int64_t curr_ts = get_now();
+  if (curr_ts - last_upd_ts_> PC_FLUSH_INTERVAL) {
+    n_to_send = pending_upds_.size();
+  } else if (pending_upds_.size() >= get_max_batch_size()) {
+    n_to_send = get_max_batch_size();
+  }
+
+  if (n_to_send == 0) {
+    return;
+  }
+
+  // send batch of price updates to solana
+  price::send( pending_upds_.data(), n_to_send);
+
+  // remove the sent elements from the vector
+  pending_upds_.erase(pending_upds_.begin(), pending_upds_.begin() + n_to_send);
+
+  // record the current time
+  last_upd_ts_= curr_ts;
+}
+
 void manager::poll( bool do_wait )
 {
   // poll for any socket events
@@ -506,10 +538,7 @@ void manager::poll( bool do_wait )
     // 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();
-    }
+    send_pending_ups();
   } else {
     reconnect_rpc();
   }
@@ -927,6 +956,13 @@ price *manager::get_price( const pub_key& acc )
   return it ? dynamic_cast<price*>( amap_.obj( it ) ) : nullptr;
 }
 
+void manager::add_dirty_price(price* sptr)
+{
+  if( std::find(pending_upds_.begin(), pending_upds_.end(), sptr) == pending_upds_.end() ) {
+    pending_upds_.emplace_back( sptr );
+  }
+}
+
 unsigned manager::get_num_product() const
 {
   return svec_.size();

pc/manager.hpp

@@ -135,6 +135,9 @@ namespace pc
     product *get_product( const pub_key& );
     price   *get_price( const pub_key& );
 
+    // adds dirty price to pending updates buffer
+    void add_dirty_price(price* sptr);
+
     // submit pyth client api request
     void submit( request * );
     void submit( net_wtr& );
@@ -221,6 +224,12 @@ namespace pc
     void poll_schedule();
     void reset_status( int );
 
+    // 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_ups();
+
     net_loop     nl_;       // epoll loop
     tcp_connect  hconn_;    // rpc http connection
     ws_connect  *wconn_;    // rpc websocket sonnection
@@ -264,6 +273,12 @@ namespace pc
     rpc::get_recent_block_hash breq_[1]; // block hash request
     rpc::program_subscribe     preq_[1]; // program account subscription
     rpc::get_program_accounts  areq_[1]; // alternative to program_subscribe
+
+    // price updates that have not been sent yet
+    std::vector<price*> pending_upds_;
+
+    // Timestamp of the last batch
+    int64_t last_upd_ts_= 0;
   };
 
   inline bool manager::get_is_tx_connect() const

pc/user.cpp

@@ -13,8 +13,6 @@
 #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;
 
@@ -36,7 +34,6 @@ 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 )
@@ -211,9 +208,9 @@ void user::parse_upd_price( uint32_t tok, uint32_t itok )
 
     // Add the updated price to the pending updates
     sptr->update_no_send( price, conf, stype, false );
-    if( std::find(pending_vec_.begin(), pending_vec_.end(), sptr) == pending_vec_.end() ) {
-      pending_vec_.emplace_back( sptr );
-    }
+
+    // pass the updated price to manager
+    sptr_->add_dirty_price(sptr);
 
     // Send the result back
     add_header();
@@ -333,28 +330,6 @@ void user::parse_get_product( uint32_t tok, uint32_t itok )
   add_tail( itok );
 }
 
-void user::send_pending_upds()
-{
-  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(), n_to_send) ) {
-    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_to_send);
-  last_upd_ts_ = curr_ts;
-}
-
 void user::parse_get_all_products( uint32_t itok )
 {
   add_header();

pc/user.hpp

@@ -43,12 +43,6 @@ namespace pc
     // symbol price schedule callback
     void on_response( price_sched *, uint64_t ) override;
 
-    // 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:
 
     // http-only request parsing
@@ -63,7 +57,6 @@ namespace pc
     };
 
     typedef std::vector<deferred_sub> def_vec_t;
-    typedef std::vector<price*> pending_vec_t;
 
     void parse_request( uint32_t );
     void parse_get_product_list( uint32_t );
@@ -87,8 +80,6 @@ namespace pc
     json_wtr        jw_;          // json writer
     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
   };
 
 }