subscribe to signatures to get accurate tx rate; scale conf. interval before applying sqrt(slot_diff)

Author: richardbrooks2000

pc/dbl_list.hpp

@@ -14,6 +14,7 @@ namespace pc
     void add( T * );
     void del( T * );
     T *first() const;
+    T *last() const;
   private:
     T *hd_;
     T *tl_;
@@ -103,6 +104,12 @@ namespace pc
     return hd_;
   }
 
+  template<class T>
+  T *dbl_list<T>::last() const
+  {
+    return tl_;
+  }
+
   template<class T>
   next<T>::next()
   : next_( nullptr )

pc/manager.cpp

@@ -79,6 +79,11 @@ manager::~manager()
     delete ptr;
   }
   svec_.clear();
+  while( !slist_.empty() ) {
+    price_sig *ptr = slist_.first();
+    slist_.del( ptr );
+    delete ptr;
+  }
 }
 
 void manager::add_map_sub()
@@ -727,3 +732,19 @@ product *manager::get_product( unsigned i ) const
 {
   return i < svec_.size() ? svec_[i] : nullptr;
 }
+
+void manager::alloc( price_sig*& ptr )
+{
+  ptr = slist_.last();
+  if ( ptr ) {
+    ptr->reset_notify();
+    slist_.del( ptr );
+  } else {
+    ptr = new price_sig;
+  }
+}
+
+void manager::dealloc( price_sig *ptr )
+{
+  slist_.add( ptr );
+}

pc/manager.hpp

@@ -151,6 +151,8 @@ namespace pc
     void del_map_sub();
     void schedule( price_sched* );
     void write( pc_pub_key_t *, pc_acc_t *ptr );
+    void alloc( price_sig*& );
+    void dealloc( price_sig* );
 
     // tx_sub callbacks
     void on_connect() override;
@@ -181,6 +183,7 @@ namespace pc
 
     typedef dbl_list<user>            user_list_t;
     typedef dbl_list<request>         req_list_t;
+    typedef dbl_list<price_sig>       sig_list_t;
     typedef std::vector<get_mapping*> map_vec_t;
     typedef std::vector<product*>     spx_vec_t;
     typedef std::vector<price_sched*> kpx_vec_t;
@@ -200,6 +203,7 @@ namespace pc
     tx_connect   tconn_;    // tx proxy connection
     user_list_t  olist_;    // open users list
     user_list_t  dlist_;    // to-be-deleted users list
+    sig_list_t   slist_;    // signature reuse list
     req_list_t   plist_;    // pending requests
     map_vec_t    mvec_;     // mapping account updates
     acc_map_t    amap_;     // account to symbol pricing info

pc/pub_stats.cpp

@@ -11,7 +11,7 @@ pub_stats::pub_stats()
 void pub_stats::clear_stats()
 {
   slots_.clear();
-  num_sent_ = num_recv_ = 0;
+  num_sent_ = num_recv_ = num_tx_ = num_coal_ = 0;
   __builtin_memset( shist_, 0, sizeof( shist_ ) );
   __builtin_memset( thist_, 0, sizeof( shist_ ) );
 }
@@ -26,16 +26,31 @@ uint64_t pub_stats::get_num_recv() const
   return num_recv_;
 }
 
+uint64_t pub_stats::get_num_tx() const
+{
+  return num_tx_;
+}
+
 uint64_t pub_stats::get_num_drop() const
 {
-  return num_sent_ - num_recv_;
+  return num_sent_ - num_tx_;
+}
+
+uint64_t pub_stats::get_num_coalesced() const
+{
+  return num_coal_;
 }
 
 double pub_stats::get_hit_rate() const
 {
   return (100.*num_recv_)/num_sent_;
 }
 
+double pub_stats::get_tx_rate() const
+{
+  return (100.*num_tx_)/num_sent_;
+}
+
 void pub_stats::add_send( uint64_t slot, int64_t ts )
 {
   ++num_sent_;

pc/pub_stats.hpp

@@ -22,11 +22,22 @@ namespace pc
     // get publish stats details
     uint64_t get_num_sent() const;
     uint64_t get_num_recv() const;
+    uint64_t get_num_tx() const;
     uint64_t get_num_drop() const;
 
+    // number of detected coalesce events (i.e. where valid slot does
+    // not match publish slot of previous update). Does not indicate
+    // how many submissions (transactions) were actually coalesced
+    uint64_t get_num_coalesced() const;
+
     // hit rate is percentage of quotes sent that appeared in an aggregate
+    // this may be undercounted because some submissions may be coalesced
+    // into a single slot
     double get_hit_rate() const;
 
+    // tx rate is the percentage of quotes send that are confirmed
+    double get_tx_rate() const;
+
     // get (rough) quartiles of publish end-to-end latency in seconds
     // to a max of 16 seconds
     void get_time_quartiles( float q[4] ) const;
@@ -38,6 +49,12 @@ namespace pc
     // clear-down statistics
     void clear_stats();
 
+    // increment signature count
+    void inc_tx();
+
+    // increment coalesce event count
+    void inc_coalesce();
+
   private:
 
     struct slot_time {
@@ -53,9 +70,21 @@ namespace pc
 
     slots_t  slots_;
     uint64_t num_sent_;
+    uint64_t num_tx_;
     uint64_t num_recv_;
+    uint64_t num_coal_;
     uint32_t thist_[num_buckets];
     uint32_t shist_[num_buckets];
   };
 
+  inline void pub_stats::inc_tx()
+  {
+    ++num_tx_;
+  }
+
+  inline void pub_stats::inc_coalesce()
+  {
+    ++num_coal_;
+  }
+
 }

pc/request.cpp

@@ -1981,9 +1981,62 @@ price *product::get_price( price_type pt ) const
   return nullptr;
 }
 
+///////////////////////////////////////////////////////////////////////////
+// price_sig
+
+price_sig::price_sig()
+: is_notify_( false )
+{
+  set_commitment( commitment::e_confirmed );
+  set_signature( this );
+}
+
+void price_sig::response( const jtree& jt )
+{
+  is_notify_ = true;
+  add_notify( jt );
+}
+
+bool price_sig::notify( const jtree& )
+{
+  ((price*)get_sub())->on_response( this );
+  return false;
+}
+
+void price_sig::reset_notify()
+{
+  is_notify_ = false;
+}
+
+bool price_sig::get_is_notify() const
+{
+  return is_notify_;
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // price
 
+void price::on_response( price_sig *res )
+{
+  // update transaction receive count
+  inc_tx();
+
+  // remove everything up to and including this result
+  for( price_sig *ptr = glist_.first(); ptr; ) {
+    price_sig *nxt = ptr->get_next();
+    if ( ptr->get_is_notify() ) {
+      ptr->remove_notify();
+      glist_.del( ptr );
+      get_manager()->dealloc( ptr );
+    }
+    if ( ptr != res ) {
+      ptr = nxt;
+    } else {
+      break;
+    }
+  }
+}
+
 price::price( const pub_key& acc, product *prod )
 : init_( false ),
   isched_( false ),
@@ -2013,6 +2066,15 @@ price::price( const pub_key& acc, product *prod )
   sreq_->set_sub( this );
 }
 
+price::~price()
+{
+  while( !glist_.empty() ) {
+    price_sig *ptr = glist_.first();
+    glist_.del( ptr );
+    delete ptr;
+  }
+}
+
 bool price::init_publish()
 {
   manager *cptr = get_manager();
@@ -2178,11 +2240,18 @@ bool price::update(
   if ( PC_UNLIKELY( !get_is_ready_publish() ) ) {
     return false;
   }
+  price_sig *sig;
   manager *mgr = get_manager();
+  mgr->alloc( sig );
   add_send( mgr->get_slot(), mgr->get_curr_time() );
   preq_->set_price( price, conf, st, mgr->get_slot(), is_agg );
   preq_->set_block_hash( mgr->get_recent_block_hash() );
+  preq_->set_signature( sig );
   mgr->submit( preq_ );
+  // subscribe to signature
+  sig->set_sub( this );
+  get_rpc_client()->send( sig );
+  glist_.add( sig );
   return true;
 }
 
@@ -2345,6 +2414,9 @@ void price::update( T *res )
     twap_  = pupd->twap_;
     avol_   = pupd->avol_;
     sym_st_ = (symbol_status)pupd->agg_.status_;
+    if ( pub_slot_ < pupd->valid_slot_ ) {
+      inc_coalesce();
+    }
     pub_slot_ = pupd->agg_.pub_slot_;
     valid_slot_ = pupd->valid_slot_;
 

pc/request.hpp

@@ -645,15 +645,31 @@ namespace pc
     uint64_t shash_;
   };
 
+  class price_sig : public rpc::signature_subscribe,
+                    public signature,
+                    public prev_next<price_sig>
+  {
+  public:
+    price_sig();
+    void response( const jtree& jt ) override;
+    bool notify( const jtree& jt ) override;
+    void reset_notify();
+    bool get_is_notify() const;
+  private:
+    bool is_notify_;
+  };
+
   // price subscriber and publisher
   class price : public request,
                 public pub_stats,
                 public rpc_sub_i<rpc::get_account_info>,
-                public rpc_sub_i<rpc::account_subscribe>
+                public rpc_sub_i<rpc::account_subscribe>,
+                public rpc_sub_i<price_sig>
   {
   public:
 
     price( const pub_key&, product *prod );
+    virtual ~price();
 
     // corresponding product definition
     product *get_product() const;
@@ -726,13 +742,16 @@ namespace pc
     void submit() override;
     void on_response( rpc::get_account_info * ) override;
     void on_response( rpc::account_subscribe * ) override;
+    void on_response( price_sig * ) override;
     bool get_is_done() const override;
 
   private:
 
     typedef enum {
       e_subscribe, e_sent_subscribe, e_publish, e_error } state_t;
 
+    typedef dbl_list<price_sig> sig_list_t;
+
     template<class T> void update( T *res );
 
     bool init_publish();
@@ -760,6 +779,7 @@ namespace pc
     int32_t                aexpo_;
     uint32_t               cnum_;
     product               *prod_;
+    sig_list_t             glist_;
     price_sched            sched_;
     pc_pub_key_t           cpub_[PC_COMP_SIZE];
     pc_price_info_t        cprice_[PC_COMP_SIZE];

pc/rpc_client.cpp

@@ -1,7 +1,6 @@
 #include "rpc_client.hpp"
 #include "bincode.hpp"
 #include <unistd.h>
-
 #include "log.hpp"
 
 using namespace pc;
@@ -2241,6 +2240,11 @@ void rpc::upd_price::set_block_hash( hash *bhash )
   bhash_ = bhash;
 }
 
+void rpc::upd_price::set_signature( signature *sig )
+{
+  sig_ = sig;
+}
+
 void rpc::upd_price::set_price( int64_t px,
                                 uint64_t conf,
                                 symbol_status st,
@@ -2318,5 +2322,6 @@ void rpc::upd_price::build( net_wtr& wtr )
 
   // all accounts need to sign transaction
   tx.sign( pub_idx, tx_idx, *ckey_ );
+  sig_->init_from_buf( (const uint8_t*)(tx.get_buf() + pub_idx) );
   ((tx_wtr&)wtr).commit( tx );
 }

pc/rpc_client.hpp

@@ -835,6 +835,7 @@ namespace pc
       void set_params( pub_key * );
       void set_program( pub_key * );
       void set_block_hash( hash * );
+      void set_signature( signature * );
       void set_price( int64_t px, uint64_t conf, symbol_status,
                       uint64_t pub_slot, bool is_aggregate );
       void build( net_wtr& ) override;
@@ -846,6 +847,7 @@ namespace pc
       pub_key      *gkey_;
       pub_key      *akey_;
       pub_key      *rkey_;
+      signature    *sig_;
       int64_t       price_;
       uint64_t      conf_;
       uint64_t      pub_slot_;;