support batch price updates

Author: jhunsaker

pc/request.cpp

@@ -655,9 +655,10 @@ bool price::update(
   if ( PC_UNLIKELY( !get_is_ready_publish() ) ) {
     return false;
   }
+  preq_->set_price( price, conf, st, is_agg );
   manager *mgr = get_manager();
   const uint64_t slot = mgr->get_slot();
-  preq_->set_price( price, conf, st, slot, is_agg );
+  preq_->set_slot( slot );
   preq_->set_block_hash( mgr->get_recent_block_hash() );
   if ( mgr->get_do_tx() )
     mgr->submit( preq_ );
@@ -688,6 +689,101 @@ bool price::update(
   return true;
 }
 
+void price::update_no_send(
+  const int64_t price, const uint64_t conf
+  , const symbol_status st, const bool is_agg
+)
+{
+  preq_->set_price( price, conf, st, is_agg );
+}
+
+bool price::send( price *prices[], const unsigned n )
+{
+  static std::vector< rpc::upd_price * > upds_;
+
+  upds_.clear();
+
+  manager *mgr1 = nullptr;
+
+  for ( unsigned i = 0, j = 0; i < n; ++i ) {
+    price *const p = prices[ i ];
+    if ( PC_UNLIKELY( ! p->init_ && ! p->init_publish() ) ) {
+      continue;
+    }
+    if ( PC_UNLIKELY( ! p->has_publisher() ) ) {
+      continue;
+    }
+    if ( PC_UNLIKELY( ! p->get_is_ready_publish() ) ) {
+      continue;
+    }
+    manager *const mgr = p->get_manager();
+    if ( ! mgr1 ) {
+      mgr1 = mgr;
+    }
+    else if ( mgr != mgr1 ) {
+      PC_LOG_ERR( "unexpected manager" ).end();
+      continue;
+    }
+    const uint64_t slot = mgr->get_slot();
+    p->preq_->set_slot( slot );
+    p->preq_->set_block_hash( mgr->get_recent_block_hash() );
+    upds_.emplace_back( p->preq_ );
+
+    if (
+      upds_.size() >= rpc::upd_price::MAX_UPDATES
+      || ( upds_.size() && ( i + 1 ) == n )
+    ) {
+      if ( mgr->get_do_tx() ) {
+        net_wtr msg;
+        if ( rpc::upd_price::build( msg, &upds_[ 0 ], upds_.size() ) ) {
+          mgr->submit( msg );
+        }
+        else {
+          PC_LOG_ERR( "failed to build msg" );
+        }
+      }
+      else {
+        p->get_rpc_client()->send( &upds_[ 0 ], upds_.size() );
+        for ( unsigned k = j; k <= i; ++k ) {
+          price *const p1 = prices[ k ];
+          p1->tvec_.emplace_back(
+            std::string( 100, '\0' ), p1->preq_->get_sent_time()
+          );
+          p1->preq_->get_signature()->enc_base58( p1->tvec_.back().first );
+          PC_LOG_DBG( "sent price update transaction" )
+            .add( "price_account", *p1->get_account() )
+            .add( "product_account", *p1->prod_->get_account() )
+            .add( "symbol", p1->get_symbol() )
+            .add( "price_type", price_type_to_str( p1->get_price_type() ) )
+            .add( "sig", p1->tvec_.back().first )
+            .add( "pub_slot", slot )
+            .end();
+          if ( PC_UNLIKELY( p1->tvec_.size() >= 100 ) ) {
+            PC_LOG_WRN( "too many unacked price update transactions" )
+              .add( "price_account", *p1->get_account() )
+              .add( "product_account", *p1->prod_->get_account() )
+              .add( "symbol", p1->get_symbol() )
+              .add( "price_type", price_type_to_str( p1->get_price_type() ) )
+              .add( "num_txid", p1->tvec_.size() )
+              .end();
+            p1->tvec_.erase( p1->tvec_.begin(), p1->tvec_.begin() + 50 );
+          }
+        }
+      }
+
+      for ( unsigned k = j; k <= i; ++k ) {
+        price *const p1 = prices[ k ];
+        p1->inc_sent();
+      }
+
+      j = i;
+      upds_.clear();
+    }
+  }
+
+  return true;
+}
+
 void price::submit()
 {
   if ( st_ == e_subscribe ) {

pc/request.hpp

@@ -261,6 +261,11 @@ namespace pc
     // or because publisher does not have permission (has_publisher())
     bool update( int64_t price, uint64_t conf, symbol_status );
 
+    void update_no_send(
+      int64_t price, uint64_t conf, symbol_status, bool aggr
+    );
+    static bool send( price * [], unsigned );
+
     // update aggregate price only
     bool update();
 
@@ -307,14 +312,6 @@ namespace pc
     void dump_json( json_wtr& wtr ) const;
 
   public:
-
-    void set_price_type( price_type );
-    void set_version( uint32_t );
-    void set_price( int64_t );
-    void set_conf( int64_t );
-    void set_symbol_status( symbol_status );
-    void set_product( product * );
-
     void reset();
     void unsubscribe();
     void submit() override;

pc/rpc_client.cpp

@@ -151,12 +151,11 @@ void rpc_client::send( rpc_request *rptr )
     reuse_.pop_back();
   } else {
     id = ++id_;
-    rv_.resize( 1 + id, nullptr );
   }
   rptr->set_id( id );
   rptr->set_rpc_client( this );
   rptr->set_sent_time( get_now() );
-  rv_[id] = rptr;
+  rv_.emplace( id, rptr );
 
   // construct json message
   json_wtr jw;
@@ -185,6 +184,56 @@ void rpc_client::send( rpc_request *rptr )
   }
 }
 
+void rpc_client::send( rpc::upd_price *upds[], const unsigned n )
+{
+  if ( ! n ) {
+    return;
+  }
+
+  // get request id
+  uint64_t id;
+  if ( !reuse_.empty() ) {
+    id = reuse_.back();
+    reuse_.pop_back();
+  } else {
+    id = ++id_;
+  }
+  const auto now = get_now();
+  for ( unsigned i = 0; i < n; ++i ) {
+    rpc_request *const rptr = upds[ i ];
+    rptr->set_id( id );
+    rptr->set_rpc_client( this );
+    rptr->set_sent_time( now );
+    rv_.emplace( id, rptr );
+  }
+
+  // construct json message
+  json_wtr jw;
+  jw.add_val( json_wtr::e_obj );
+  jw.add_key( "jsonrpc", "2.0" );
+  jw.add_key( "id", id );
+  rpc::upd_price::request( jw, upds, n );
+  jw.pop();
+//  jw.print();
+  if ( upds[ 0 ]->get_is_http() ) {
+    // submit http POST request
+    http_request msg;
+    msg.init( "POST", "/" );
+    msg.add_hdr( "Host", hptr_->get_host() );
+    msg.add_hdr( "Content-Type", "application/json" );
+    msg.commit( jw );
+    hptr_->add_send( msg );
+  } else if ( wptr_ ) {
+    // submit websocket message
+    ws_wtr msg;
+    msg.commit( ws_wtr::text_id, jw, true );
+    wptr_->add_send( msg );
+  }
+  else {
+    PC_LOG_WRN( "no ws connection to send msg" ).end();
+  }
+}
+
 void rpc_client::rpc_http::parse_content( const char *txt, size_t len )
 {
   cp_->parse_response( txt, len );
@@ -202,15 +251,14 @@ void rpc_client::parse_response( const char *txt, size_t len )
   uint32_t idtok = jp_.find_val( 1, "id" );
   if ( idtok ) {
     // response to http request
-    uint64_t id = jp_.get_uint( idtok );
-    if ( id < rv_.size() ) {
-      rpc_request *rptr = rv_[id];
-      if ( rptr ) {
-        rv_[id] = nullptr;
-        reuse_.push_back( id );
-        rptr->response( jp_ );
-      }
+    const uint64_t id = jp_.get_uint( idtok );
+    const auto range = rv_.equal_range( id );
+    for ( auto it = range.first; it != range.second; ++it ) {
+      rpc_request *const rptr = it->second;
+      rptr->response( jp_ );
     }
+    reuse_.push_back( id );
+    rv_.erase( range.first, range.second );
   } else {
     // websocket notification
     uint32_t ptok = jp_.find_val( 1, "params" );
@@ -798,16 +846,19 @@ void rpc::upd_price::set_block_hash( hash *bhash )
 void rpc::upd_price::set_price( int64_t px,
                                 uint64_t conf,
                                 symbol_status st,
-                                uint64_t pub_slot,
                                 bool is_agg )
 {
   price_ = px;
   conf_  = conf;
   st_    = st;
-  pub_slot_ = pub_slot;
   cmd_   = is_agg?e_cmd_agg_price:e_cmd_upd_price;
 }
 
+void rpc::upd_price::set_slot( const uint64_t pub_slot )
+{
+  pub_slot_ = pub_slot;
+}
+
 signature *rpc::upd_price::get_signature()
 {
   return &sig_;
@@ -974,7 +1025,7 @@ bool rpc::upd_price::request(
   // construct binary transaction
   net_buf *bptr = net_buf::alloc();
   bincode tx( bptr->buf_ );
-  if ( !  build_tx( tx, upds, n ) ) {
+  if ( ! build_tx( tx, upds, n ) ) {
     return false;
   }
 

pc/rpc_client.hpp

@@ -7,6 +7,8 @@
 #include <oracle/oracle.h>
 #include <pc/hash_map.hpp>
 
+#include <unordered_map>
+
 #define PC_RPC_ERROR_BLOCK_CLEANED_UP          -32001
 #define PC_RPC_ERROR_SEND_TX_PREFLIGHT_FAIL    -32002
 #define PC_RPC_ERROR_TX_SIG_VERIFY_FAILURE     -32003
@@ -62,6 +64,11 @@ namespace pc
   str commitment_to_str( commitment );
   commitment str_to_commitment( str );
 
+  namespace rpc
+  {
+    class upd_price;
+  }
+
   // solana rpc REST API client
   class rpc_client : public error
   {
@@ -80,6 +87,7 @@ namespace pc
 
     // submit rpc request (and bundled callback)
     void send( rpc_request * );
+    void send( rpc::upd_price *[], unsigned n );
 
   public:
 
@@ -123,7 +131,7 @@ namespace pc
       };
     };
 
-    typedef std::vector<rpc_request*> request_t;
+    typedef std::unordered_multimap< uint64_t, rpc_request* > request_t;
     typedef std::vector<uint64_t>     id_vec_t;
     typedef std::vector<char>         acc_buf_t;
     typedef hash_map<trait>           sub_map_t;
@@ -395,6 +403,8 @@ namespace pc
     class upd_price : public tx_request, public rpc_request
     {
     public:
+      static constexpr unsigned MAX_UPDATES = 10;
+
       // parameters
       void set_symbol_status( symbol_status );
       void set_publish( key_pair * );
@@ -403,7 +413,8 @@ namespace pc
       void set_program( pub_key * );
       void set_block_hash( hash * );
       void set_price( int64_t px, uint64_t conf, symbol_status,
-                      uint64_t pub_slot, bool is_aggregate );
+                      bool is_aggregate );
+      void set_slot( uint64_t );
 
       // results
       signature *get_signature();