Merge branch 'master' of bitbucket.org:bluetradesys/spark

doc/user/docs/release_notes.md

@@ -1,5 +1,9 @@
 # Release Notes
 
+## 1.4.10
+### Bug Fixes
+  - Removed duplicate entries in performance log
+  
 ## 1.4.9
 ### SDK
   - Added ticker events to the order API

src/client/spark_core/TestRunner/TestEnumerator.cs

@@ -23,6 +23,11 @@ namespace bts.test {
       }
 
       public IReadOnlyCollection<TestClass> TestGroups { get; }
+
+      public IEnumerable<TestClass> get_groups_of_type(TestType tt) {
+         return TestGroups.Where(wt => wt.test_type == tt);
+      }
+
       public IEnumerable<TestMethod> Tests => TestGroups.SelectMany(t => t.methods);
 
       public IEnumerable<TestMethod> get_of_type(TestType tt) {

src/client/sparkcmd/Cmd/TestCommand.cs

@@ -64,7 +64,7 @@ namespace sparkcmd {
 
       void list_tests() {
          Console.WriteLine("All tests: ");
-         foreach (var t in Enumerator.TestGroups) {
+         foreach (var t in Enumerator.get_groups_of_type(test_type)) {
             ConsoleEx.WriteLine(t.name, ConsoleColor.Green);
             foreach (var test in t.methods) {
                Console.Write($"  {test.name}");

src/data_files/files/perf_trace/perf_log.sql

@@ -62,7 +62,7 @@ select
    fix_reads.channel,
    fix_reads.sn
 from
-   (select session_start, channel, sn, hts,ts,rowid entry_id from trace where entry_kind='fix_msg_r') fix_reads,
+   (select session_start, channel, sn, hts,ts,rowid entry_id from trace where entry_kind='fix_msg_r' order by ts) fix_reads,
    (select session_start, channel, sn, hts,ts,rowid entry_id from trace where entry_kind='before_fix_socket_w') fix_writes
 where
    fix_reads.session_start = fix_writes.session_start and
@@ -73,14 +73,14 @@ select
    md_reads.session_start,
    "requote" event_type,
    md_reads.hts hts_start,
-   md_reads.entry_id event_id,
+   md_reads.rowid event_id,
    md_reads.ts start,
    fix_writes.ts stop,
    (fix_writes.ts - md_reads.ts) / 1000.0 duration,
    md_reads.channel,
    md_reads.sn
 from
-   (select session_start, channel, sn, hts,ts,rowid entry_id from trace where entry_kind='md') md_reads,
+   (select distinct session_start, channel, sn, hts,ts from trace where entry_kind='md' order by ts) md_reads,
    (select session_start, channel, sn, hts,ts,rowid entry_id from trace where entry_kind='before_fix_socket_w') fix_writes
 where
    md_reads.session_start = fix_writes.session_start and
@@ -302,22 +302,17 @@ insert into requote_trace_archive select * from requote_trace_view;
 insert into event_archive select * from event_view;
 insert into log_archive select * from log_view;
 insert into event_log_archive select * from event_log_aux;
-
 delete from trace;
-
 delete from event_log_archive where 
 rowid in (select rowid from event_log_archive order by rowid desc limit -1 offset 10000123);
-
 delete from hedge_trace_archive where 
 rowid in (select rowid from hedge_trace_archive order by rowid desc limit -1 offset 10000123);
-
 delete from requote_trace_archive where 
 rowid in (select rowid from requote_trace_archive order by rowid desc limit -1 offset 10000123);
-
 delete from log_archive where 
 rowid in (select rowid from log_archive order by rowid desc limit -1 offset 10000123);
-
 delete from event_log_archive where 
 rowid in (select rowid from event_log_archive order by rowid desc limit -1 offset 10000123);
 commit;
 
+

src/lib/orderman/actors/MDSource.h

@@ -14,6 +14,7 @@
 #include "lib/tracing/PerfLog.h"
 #include "lib/tracing/bts_trace.h"
 #include "lib/utils/TimeStamp.h"
+#include "lib/domain_model/Security.h"
 
 #include <date/date.h>
 
@@ -80,6 +81,7 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
                    switch (ev.md_type) {
                    case MDRequestType::BOOK:
                       on_book_sub(k, *synthetic_count_metric);
+                      security_count_metric->set(_handles.size());
                       break;
                    case MDRequestType::TICKER:
                       on_ticker_sub(k, *ticker_channel_cout_metric);
@@ -106,6 +108,8 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
             book._security_key = SecurityKey(handle.key());
             trace::CorID trace{h.external_event_channel, handle.sequence()};
             _perf_log.log(trace, h.last, "md", handle.key());
+            _perf_log.log(trace, "md_r", h.last, h.symbol->c_str(),
+                          handle.key());
             publish(_book_update, book, trace);
          }
       }
@@ -122,7 +126,7 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
                if (trade.key() == per_sec_ev.key().value()) {
                   trade_count++;
                   per_sec_ev.trades.emplace_back(TickerUpdateEvent::Trade{
-                        datetime::utc_ll_to_timepoint(trade.time()),
+                      datetime::utc_ll_to_timepoint(trade.time()),
                       Price(md::IntPrice(trade.mantissa(),
                                          (int8_t)trade.exponent())
                                 .price()),
@@ -161,7 +165,9 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
       auto handle = _md_consumer.get_handle(k);
 
       _handles.push_back({utils::TimeStamp(), handle,
-                          get_external_event_channel(handle.get_usersymbol())});
+                          get_external_event_channel(handle.get_usersymbol()),
+                          std::make_unique<std::string>(securities::descriptor(
+                              handle.get_normalized_security()))});
       if (handle.is_synthetic())
          synthetic_count_metric.increment();
    }
@@ -202,10 +208,11 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
           "key: {}",
           user_symbol.to_string(), sk.value());
 
-// We do not want to get old trades so we drain the trades buffer at
-// this point. This makes the first subscription to a channel
-// consistent with subsquent ones in terms of late join semantics.
-      ChannelRecord c {0, _md_consumer.get_channel(user_symbol), {}, user_symbol} ;
+      // We do not want to get old trades so we drain the trades buffer at
+      // this point. This makes the first subscription to a channel
+      // consistent with subsquent ones in terms of late join semantics.
+      ChannelRecord c{
+          0, _md_consumer.get_channel(user_symbol), {}, user_symbol};
       _trades_buffer.Clear();
       c.cursor = c.channel.read_trades(_trades_buffer, c.cursor);
       c.securities.push_back(sk);
@@ -215,6 +222,7 @@ class MDSource : public mercury::Actor, public mercury::IPollableActor {
       utils::TimeStamp last;
       md::MDConsumerHandle handle;
       const char *external_event_channel;
+      std::unique_ptr<std::string> symbol;
    };
    boost::container::small_vector<HandleRecord, 100> _handles;
 

src/lib/orderman/orders/spreader_orders/spreader/SpreaderChildOrderState.cpp

@@ -22,6 +22,7 @@ _filled_quantity(src._filled_quantity),
 _last_filled_price(src._last_filled_price),
 _last_reported_order_price(src._last_reported_order_price),
 _last_filled_quantity(src._last_filled_quantity),
+_last_update_trace(src._last_update_trace),
 _out_price(src._out_price),
 _out_quantity(src._out_quantity),
 _order_place_time(src._order_place_time)
@@ -54,6 +55,7 @@ void SpreaderChildOrderState::re_init (ChildOrder& child_order){
    _last_filled_price = Price(0);
    _last_reported_order_price=Price(0);
    _last_filled_quantity = Quantity(0);
+   _last_update_trace=trace::CorID();
    _out_price = Price(0);
    _out_quantity = Quantity(0);
    _order_place_time=bts::TimeStamp();

src/lib/orderman/orders/spreader_orders/spreader/SpreaderChildOrderState.h

@@ -44,6 +44,9 @@ private:
    Price _last_filled_price;
    Price _last_reported_order_price;
    Quantity _last_filled_quantity;
+   // this is the trace of the order event
+   // from the last on_order_update call
+   trace::CorID _last_update_trace;
    // this is latest price, qty sent out
    Price _out_price;
    Quantity _out_quantity;   
@@ -113,6 +116,9 @@ public:
    bts::TimeStamp order_place_time() const{
       return _order_place_time;
    }
+   const trace::CorID& last_update_trace() const{
+      return _last_update_trace;
+   }
    // init ///////////////////
    ///////////////////////////
    // this assignes a new order, and subsumes the earlier order's
@@ -134,6 +140,7 @@ public:
                                                 - _filled_quantity.value());
       _last_filled_price = Price(e.last_price);
       _last_reported_order_price=Price(e.price);
+      _last_update_trace=e.trace;
       SpreaderUtils::compute_average_price(_filled_price,
                                           _filled_quantity,
                                           _last_filled_price,

src/lib/orderman/orders/spreader_orders/spreader/quoter/quoter_states/QuoterWorkingStateWorker.cpp

@@ -17,7 +17,9 @@ void QuoterWorkingStateWorker::on_transition (){
    // waiting for the ack, things could have changed (namely
    // fills in other legs, books, order qty change), hence
    // we analyze and potentially modify/cxl immediatly
-   _longpath_analyze_and_act ();
+   trace::CorID ack_message_trace_id=
+      _quoter->_child_order_state.last_update_trace();
+   _longpath_analyze_and_act (ack_message_trace_id);
 }
 void QuoterWorkingStateWorker::one_time_activate (Price,
                                                   Quantity){
@@ -25,10 +27,11 @@ void QuoterWorkingStateWorker::one_time_activate (Price,
 }
 void QuoterWorkingStateWorker::on_order_update (const SpreaderChildOrderState& u){
    bts_assert (u.acknowledged());
+   trace::CorID ack_message_trace_id=u.last_update_trace();
    if (u.terminal_status())
       _quoter->_transform_to_END ();
    else
-      _longpath_analyze_and_act ();
+      _longpath_analyze_and_act (ack_message_trace_id);
 }
 void QuoterWorkingStateWorker::on_book (size_t,
                                         const bts::orderman::Book&,