submit on leader schedule

more on leader rotation

simpler publishing

end-to-end latency tracking and backwards in time checking

print price account name after add_price

support (re)initialization of price accounts

rust example to print all pyth price data on devnet

track publish latency statistics and hit rates

sort dashboard products by asset_type, symbol

Author: richardbrooks2000

.gitignore

@@ -2,3 +2,4 @@
 build
 debug
 target
+Cargo.lock

CMakeLists.txt

@@ -32,6 +32,7 @@ set( PC_SRC
   pc/mem_map.cpp;
   pc/misc.cpp;
   pc/net_socket.cpp;
+  pc/pub_stats.cpp;
   pc/replay.cpp;
   pc/request.cpp;
   pc/rpc_client.cpp;

dashboard/dashboard.js

@@ -72,10 +72,17 @@ class Prices
   }
   get_product_list( res ) {
     let tab = document.getElementById( "prices" );
+    let k = 1;
+    res.result.sort( function(a,b) {
+      return a['attr_dict']['symbol'] > b['attr_dict']['symbol'];
+    } );
+    res.result.sort( function(a,b) {
+      return a['attr_dict']['asset_type'] > b['attr_dict']['asset_type'];
+    } );
     for( let i = 0; i != res.result.length; ++i ) {
       let sym = res.result[i];
       let att = sym['attr_dict']
-      for( let j=0; j != sym.price.length; ++j ) {
+      for( let j=0; j != sym.price.length; ++j, ++k ) {
         let pxa = sym.price[j];
         let row = tab.insertRow(-1);
         row.appendChild( this.get_title( att, 'asset_type') );
@@ -96,7 +103,7 @@ class Prices
             'account' : pxa['account']
            }
         };
-        pxa.idx = i+1;
+        pxa.idx = k;
         this.send( msg, this.get_price.bind( this, pxa ) );
       }
     }

etc/products.json

@@ -125,17 +125,5 @@
       "nasdaq_symbol": "QQQ"
     }
   }
-  {
-    "account": "96NiWypEbNs56cBEiYACEzZtFTLqhV2TQ6wH2A2CqxbM",
-    "attr_dict":{
-      "symbol": "GBP/USD",
-      "asset_type": "Currency",
-      "country": "US",
-      "description": "BRITISH POUND STERLING",
-      "quote_currency" : "USD",
-      "tenor": "Spot",
-      "jlqd_symbol": "GBPUSD"
-    }
-  }
 ]
 

pc/manager.cpp

@@ -7,6 +7,7 @@ using namespace pc;
 #define PC_RPC_WEBSOCKET_PORT 8900
 #define PC_RECONNECT_TIMEOUT  (120L*1000000000L)
 #define PC_BLOCKHASH_TIMEOUT  3
+#define PC_PUB_INTERVAL       (293L*PC_NSECS_IN_MSEC)
 
 ///////////////////////////////////////////////////////////////////////////
 // manager_sub
@@ -39,17 +40,17 @@ manager::manager()
   status_( 0 ),
   num_sub_( 0 ),
   version_( PC_VERSION ),
-  kidx_( 0 ),
+  kidx_( (unsigned)-1 ),
   cts_( 0L ),
   ctimeout_( PC_NSECS_IN_SEC ),
-  slot_ts_( 0L ),
-  slot_int_( 0L ),
-  slot_min_( 0L ),
   slot_( 0UL ),
   slot_cnt_( 0UL ),
+  curr_ts_( 0L ),
+  pub_ts_( 0L ),
+  pub_int_( PC_PUB_INTERVAL ),
   wait_conn_( false ),
   do_cap_( false ),
-  first_ack_( true )
+  is_pub_( false )
 {
   breq_->set_sub( this );
   sreq_->set_sub( this );
@@ -105,6 +106,16 @@ std::string manager::get_capture_file() const
   return cap_.get_file();
 }
 
+void manager::set_publish_interval( int64_t pub_int )
+{
+  pub_int_ = pub_int * PC_NSECS_IN_MSEC;
+}
+
+int64_t manager::get_publish_interval() const
+{
+  return pub_int_ / PC_NSECS_IN_MSEC;
+}
+
 void manager::set_do_capture( bool do_cap )
 {
   do_cap_ = do_cap;
@@ -155,6 +166,11 @@ manager_sub *manager::get_manager_sub() const
   return sub_;
 }
 
+int64_t manager::get_curr_time() const
+{
+  return curr_ts_;
+}
+
 rpc_client *manager::get_rpc_client()
 {
   return &clnt_;
@@ -170,16 +186,6 @@ uint64_t manager::get_slot() const
   return slot_;
 }
 
-int64_t manager::get_slot_time() const
-{
-  return slot_ts_;
-}
-
-int64_t manager::get_slot_interval() const
-{
-  return slot_int_;
-}
-
 void manager::teardown()
 {
   PC_LOG_INF( "pythd_teardown" ).end();
@@ -258,6 +264,7 @@ bool manager::init()
   PC_LOG_INF( "initialized" )
     .add( "version", version_ )
     .add( "capture_file", get_capture_file() )
+    .add( "publish_interval(ms)", get_publish_interval() )
     .end();
 
   return true;
@@ -347,24 +354,30 @@ void manager::poll( bool do_wait )
   // destroy any users scheduled for deletion
   teardown_users();
 
-  // check if we need to reconnect rpc services
-  if ( PC_UNLIKELY( !has_status( PC_PYTH_RPC_CONNECTED ) ||
-                    hconn_.get_is_err() || wconn_.get_is_err() ))  {
+  // get current time
+  curr_ts_ = get_now();
+
+  // submit new quotes while connected
+  if ( has_status( PC_PYTH_RPC_CONNECTED ) &&
+       !hconn_.get_is_err() &&
+       !wconn_.get_is_err() ) {
+    poll_schedule();
+  } else {
     reconnect_rpc();
-    return;
   }
+}
 
-  // schedule next price update but only if we're connected to rpc port
-  int64_t ts = get_now();
-  while ( kidx_ < kvec_.size() ) {
+void manager::poll_schedule()
+{
+  while ( is_pub_ && kidx_ < kvec_.size() ) {
     price_sched *kptr = kvec_[kidx_];
-    int64_t tot_ts = slot_min_ - ack_ts_ - ack_ts_;
-    tot_ts = std::max( slot_min_/10, tot_ts );
-    int64_t pub_ts = slot_ts_ + ( tot_ts * kptr->get_hash() ) /
+    int64_t pub_ts = pub_ts_ + ( pub_int_ * kptr->get_hash() ) /
       price_sched::fraction;
-    if ( ts > pub_ts ) {
+    if ( curr_ts_ > pub_ts ) {
       kptr->schedule();
-      ++kidx_;
+      if ( ++kidx_ >= kvec_.size() ) {
+        is_pub_ = false;
+      }
     } else {
       break;
     }
@@ -391,15 +404,17 @@ void manager::reconnect_rpc()
 
     // reset state
     wait_conn_ = false;
-    first_ack_ = true;
+    is_pub_ = false;
+    kidx_ = 0;
     ctimeout_ = PC_NSECS_IN_SEC;
-    slot_ts_ = slot_int_ = 0L;
+    pub_ts_ = 0L;
     slot_cnt_ = 0UL;
+    slot_ = 0L;
     num_sub_ = 0;
     clnt_.reset();
     plist_.clear();
 
-    // resubmit slot subscription
+    // subscribe to slots and get first block hash
     clnt_.send( sreq_ );
 
     // resubscribe to mapping and symbol accounts
@@ -419,6 +434,13 @@ void manager::reconnect_rpc()
         add_map_sub();
       }
     }
+
+    // subscribe to mapping account if not done before
+    pub_key *mpub = get_mapping_pub_key();
+    if ( mvec_.empty() && mpub ) {
+      add_mapping( *mpub );
+    }
+
     // callback user with connection status
     if ( sub_ ) {
       sub_->on_connect( this );
@@ -518,36 +540,31 @@ void manager::schedule( price_sched *kptr )
 
 void manager::on_response( rpc::slot_subscribe *res )
 {
-  if ( !slot_ts_ ) {
-    slot_ts_ = res->get_recv_time();
+  // ignore slots that go back in time
+  uint64_t slot = res->get_slot();
+  int64_t ts = res->get_recv_time();
+  if ( slot <= slot_ ) {
     return;
   }
-  slot_ = res->get_slot();
-  slot_int_ = res->get_recv_time() - slot_ts_;
-  slot_ts_ = res->get_recv_time();
+  slot_ = slot;
+  PC_LOG_DBG( "receive slot" ).add( "slot", slot_ ).end();
+
+  // submit block hash every N slots
   if ( slot_cnt_++ % PC_BLOCKHASH_TIMEOUT == 0 ) {
-    // submit block hash every N slots
     clnt_.send( breq_ );
   }
-  // reset scheduler
-  kidx_ = 0;
 
-  // derive minimum slot interval
-  if ( slot_min_ ) {
-    slot_min_ = std::min( slot_int_, slot_min_ );
-  } else {
-    slot_min_ = slot_int_;
+  // reset submit
+  if ( !is_pub_ ) {
+    kidx_ = 0;
+    pub_ts_ = ts;
+    is_pub_ = true;
   }
 
   // flush capture
   if ( do_cap_ ) {
     cap_.flush();
   }
-  PC_LOG_DBG( "receive slot" )
-   .add( "slot_num", res->get_slot() )
-   .add( "slot_int", 1e-6*slot_int_ )
-   .add( "slot_min", 1e-6*slot_min_ )
-   .end();
 }
 
 void manager::on_response( rpc::get_recent_block_hash *m )
@@ -557,30 +574,18 @@ void manager::on_response( rpc::get_recent_block_hash *m )
         + m->get_err_msg()  + "]" );
     return;
   }
-  static const double afactor = 2./(1+8.);
   int64_t ack_ts = m->get_recv_time() - m->get_sent_time();
-  if ( !first_ack_ ) {
-    ack_ts_ = (1.-afactor)*ack_ts_ + afactor * ack_ts;
-  } else {
-    ack_ts_ = ack_ts;
-    first_ack_ = false;
-  }
   if ( has_status( PC_PYTH_HAS_BLOCK_HASH ) ) {
     return;
   }
   // set initialized status for block hash
   set_status( PC_PYTH_HAS_BLOCK_HASH );
   PC_LOG_INF( "received_recent_block_hash" )
-    .add( "slot", m->get_slot() )
-    .add( "slot_interval(ms)", 1e-6*slot_int_ )
-    .add( "rount_trip_time(ms)", 1e-6*ack_ts_ )
+    .add( "curr_slot", slot_ )
+    .add( "hash_slot", m->get_slot() )
+    .add( "rount_trip_time(ms)", 1e-6*ack_ts )
     .end();
 
-  // subscribe to mapping account if not done before
-  pub_key *mpub = get_mapping_pub_key();
-  if ( mvec_.empty() && mpub ) {
-    add_mapping( *mpub );
-  }
 }
 
 void manager::submit( request *req )

pc/manager.hpp

@@ -73,10 +73,17 @@ namespace pc
     void set_capture_file( const std::string& cap_file );
     std::string get_capture_file() const;
 
+    // override default publish interval (in milliseconds)
+    void set_publish_interval( int64_t mill_secs );
+    int64_t get_publish_interval() const;
+
     // event subscription callback
     void set_manager_sub( manager_sub * );
     manager_sub *get_manager_sub() const;
 
+    // current time in manager in nanoseconds from epoch
+    int64_t get_curr_time() const;
+
     // rpc client interface
     rpc_client *get_rpc_client();
 
@@ -86,12 +93,6 @@ namespace pc
     // get most recently processed slot
     uint64_t get_slot() const;
 
-    // slot start time esitimate
-    int64_t get_slot_time() const;
-
-    // slot interval time estimate
-    int64_t get_slot_interval() const;
-
     // add and subscribe to new mapping account
     void add_mapping( const pub_key& );
 
@@ -171,6 +172,7 @@ namespace pc
     void reconnect_rpc();
     void log_disconnect();
     void teardown_users();
+    void poll_schedule();
     void reset_status( int );
 
     net_loop     nl_;       // epoll loop
@@ -193,16 +195,15 @@ namespace pc
     uint32_t     kidx_;     // schedule index
     int64_t      cts_;      // (re)connect timestamp
     int64_t      ctimeout_; // connection timeout
-    int64_t      slot_ts_;  // slot start time estimate
-    int64_t      slot_int_; // slot interval
-    int64_t      slot_min_; // slot minimum interval
     uint64_t     slot_;     // current slot
     uint64_t     slot_cnt_; // slot count
-    int64_t      ack_ts_;   // ack time ema
+    int64_t      curr_ts_;  // current time
+    int64_t      pub_ts_;   // start publish time
+    int64_t      pub_int_;  // publish interval
     kpx_vec_t    kvec_;     // symbol price scheduling
     bool         wait_conn_;// waiting on connection
     bool         do_cap_;   // do capture flag
-    bool         first_ack_;// first ack flag
+    bool         is_pub_;   // is publishing mode
     capture      cap_;      // aggregate price capture
 
     // requests

pc/misc.hpp

@@ -6,6 +6,7 @@
 #define PC_PACKED        __attribute__((__packed__))
 #define PC_UNLIKELY(ARG) __builtin_expect((ARG),1)
 #define PC_NSECS_IN_SEC  1000000000L
+#define PC_NSECS_IN_MSEC 1000000L
 
 namespace pc
 {
@@ -44,6 +45,7 @@ namespace pc
     str( const std::string& );
     std::string as_string() const;
     bool operator==( const str& ) const;
+    bool operator!=( const str& ) const;
     const char *str_;
     size_t      len_;
   };
@@ -77,6 +79,12 @@ namespace pc
            0 == __builtin_strncmp( str_, obj.str_, len_);
   }
 
+  inline bool str::operator!=( const str& obj ) const
+  {
+    return len_ != obj.len_ ||
+           0 != __builtin_strncmp( str_, obj.str_, len_);
+  }
+
   inline std::string str::as_string() const
   {
     return std::string( str_, len_ );