Make batch size configurable (#159)

* env var

* add

* grr

* store size in manager

* fix bug

* cleanup

* pr comments

* bug fix

Co-authored-by: Jayant Krishnamurthy <[email protected]>
Co-authored-by: Jayant Krishnamurthy <[email protected]>

Author: 3 people

pc/manager.cpp

@@ -9,6 +9,7 @@ using namespace pc;
 #define PC_BLOCKHASH_TIMEOUT  3
 #define PC_PUB_INTERVAL       (227L*PC_NSECS_IN_MSEC)
 #define PC_RPC_HOST           "localhost"
+#define PC_MAX_BATCH          8
 
 ///////////////////////////////////////////////////////////////////////////
 // manager_sub
@@ -70,6 +71,7 @@ manager::manager()
   do_tx_( true ),
   is_pub_( false ),
   cmt_( commitment::e_confirmed ),
+  max_batch_( PC_MAX_BATCH ),
   sreq_{ { commitment::e_processed } }
 {
   tconn_.set_sub( this );
@@ -178,6 +180,16 @@ int64_t manager::get_publish_interval() const
   return pub_int_ / PC_NSECS_IN_MSEC;
 }
 
+void manager::set_max_batch_size( unsigned batch_size )
+{
+  max_batch_ = batch_size;
+}
+
+unsigned manager::get_max_batch_size() const
+{
+  return max_batch_;
+}
+
 void manager::set_do_capture( bool do_cap )
 {
   do_cap_ = do_cap;
@@ -746,7 +758,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();

pc/manager.hpp

@@ -100,6 +100,10 @@ namespace pc
     void set_publish_interval( int64_t mill_secs );
     int64_t get_publish_interval() const;
 
+    // override the default maximum number of price updates to send in a batch
+    void set_max_batch_size( unsigned batch_size );
+    unsigned get_max_batch_size() const;
+
     // event subscription callback
     void set_manager_sub( manager_sub * );
     manager_sub *get_manager_sub() const;
@@ -253,6 +257,7 @@ namespace pc
     capture      cap_;      // aggregate price capture
     tx_parser    txp_;      // handle unexpected errors
     commitment   cmt_;      // account get/subscribe commitment
+    unsigned     max_batch_;// maximum number of price updates that can be sent in a single batch
 
     // requests
     rpc::get_slot              sreq_[1]; // slot subscription

pc/request.cpp

@@ -773,7 +773,7 @@ bool price::send( price *prices[], const unsigned n )
     upds_.emplace_back( p->preq_ );
 
     if (
-      upds_.size() >= rpc::upd_price::MAX_UPDATES
+      upds_.size() >= mgr->get_max_batch_size()
       || ( upds_.size() && ( i + 1 ) == n )
     ) {
       if ( mgr->get_do_tx() ) {

pc/rpc_client.hpp

@@ -403,8 +403,6 @@ namespace pc
     class upd_price : public tx_request, public rpc_request
     {
     public:
-      static constexpr unsigned MAX_UPDATES = 8;
-
       // parameters
       void set_symbol_status( symbol_status );
       void set_publish( key_pair * );

pcapps/pythd.cpp

@@ -101,8 +101,9 @@ int main(int argc, char **argv)
   std::string tx_host  = get_tx_host();
   int pyth_port = get_port();
   int opt = 0;
+  unsigned max_batch_size = 0;
   bool do_wait = true, do_tx = true, do_ws = true, do_debug = false;
-  while( (opt = ::getopt(argc,argv, "r:t:p:k:w:c:l:m:dnxhz" )) != -1 ) {
+  while( (opt = ::getopt(argc,argv, "r:t:p:k:w:c:l:m:b:dnxhz" )) != -1 ) {
     switch(opt) {
       case 'r': rpc_host = optarg; break;
       case 't': tx_host = optarg; break;
@@ -112,6 +113,7 @@ int main(int argc, char **argv)
       case 'w': cnt_dir = optarg; break;
       case 'l': log_file = optarg; break;
       case 'm': cmt = str_to_commitment(optarg); break;
+      case 'b': max_batch_size = strtoul(optarg, NULL, 0); break;
       case 'n': do_wait = false; break;
       case 'x': do_tx = false; break;
       case 'z': do_ws = false; break;
@@ -154,6 +156,12 @@ int main(int argc, char **argv)
       << mgr.get_mapping_pub_key_file() << "]" << std::endl;
     return 1;
   }
+
+  if (max_batch_size > 0) {
+    mgr.set_max_batch_size(max_batch_size);
+  }
+  std::cout << "pythd: max batch size " << mgr.get_max_batch_size() << std::endl;
+
   // set up signal handing
   signal( SIGINT, sig_handle );
   signal( SIGHUP, sig_handle );