Merge pull request #519 from crypto-chassis/fix-threading

Fix threading

Author: cryptochassis

README.md

@@ -1,5 +1,6 @@
 # Some breaking changes introduced
-* Rename "PRICE_TIMES_QUANTITY_MIN" to "QUOTE_QUANTITY_MIN". Rename "CUMULATIVE_FILLED_PRICE_TIMES_QUANTITY" to "CUMULATIVE_FILLED_QUOTE_QUANTITY".
+* The return type of `EventHandler::processEvent` has been changed from `bool` to `void`.
+* Please read [Handle events in "immediate" vs. "batching" mode](#handle-events-in-immediate-vs-batching-mode).
 
 <!-- START doctoc generated TOC please keep comment here to allow auto update -->
 
@@ -57,6 +58,8 @@
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 
 
+
+
 # ccapi
 * A header-only C++ library for streaming market data and executing trades directly from cryptocurrency exchanges (i.e. the connections are between your server and the exchange server without anything in-between).
 * Bindings for other languages such as Python, Java, C#, Go, and Javascript are provided.
@@ -225,9 +228,8 @@ Logger* Logger::logger = nullptr;  // This line is needed.
 
 class MyEventHandler : public EventHandler {
  public:
-  bool processEvent(const Event& event, Session* sessionPtr) override {
+  void processEvent(const Event& event, Session* sessionPtr) override {
     std::cout << "Received an event:\n" + event.toStringPretty(2, 2) << std::endl;
-    return true;
   }
 };
 } /* namespace ccapi */
@@ -302,7 +304,7 @@ Logger* Logger::logger = nullptr;  // This line is needed.
 
 class MyEventHandler : public EventHandler {
  public:
-  bool processEvent(const Event& event, Session* sessionPtr) override {
+  void processEvent(const Event& event, Session* sessionPtr) override {
     if (event.getType() == Event::Type::SUBSCRIPTION_STATUS) {
       std::cout << "Received an event of type SUBSCRIPTION_STATUS:\n" + event.toStringPretty(2, 2) << std::endl;
     } else if (event.getType() == Event::Type::SUBSCRIPTION_DATA) {
@@ -314,7 +316,6 @@ class MyEventHandler : public EventHandler {
         }
       }
     }
-    return true;
   }
 };
 } /* namespace ccapi */
@@ -494,9 +495,8 @@ Logger* Logger::logger = nullptr;  // This line is needed.
 
 class MyEventHandler : public EventHandler {
  public:
-  bool processEvent(const Event& event, Session* sessionPtr) override {
+  void processEvent(const Event& event, Session* sessionPtr) override {
     std::cout << "Received an event:\n" + event.toStringPretty(2, 2) << std::endl;
-    return true;
   }
 };
 } /* namespace ccapi */
@@ -592,7 +592,7 @@ Logger* Logger::logger = nullptr;  // This line is needed.
 
 class MyEventHandler : public EventHandler {
  public:
-  bool processEvent(const Event& event, Session* sessionPtr) override {
+  void processEvent(const Event& event, Session* sessionPtr) override {
     if (event.getType() == Event::Type::SUBSCRIPTION_STATUS) {
       std::cout << "Received an event of type SUBSCRIPTION_STATUS:\n" + event.toStringPretty(2, 2) << std::endl;
       auto message = event.getMessageList().at(0);
@@ -609,7 +609,6 @@ class MyEventHandler : public EventHandler {
     } else if (event.getType() == Event::Type::SUBSCRIPTION_DATA) {
       std::cout << "Received an event of type SUBSCRIPTION_DATA:\n" + event.toStringPretty(2, 2) << std::endl;
     }
-    return true;
   }
 };
 } /* namespace ccapi */
@@ -836,7 +835,7 @@ namespace ccapi {
 Logger* Logger::logger = nullptr;  // This line is needed.
 class MyEventHandler : public EventHandler {
  public:
-  bool processEvent(const Event& event, Session* sessionPtr) override {
+  void processEvent(const Event& event, Session* sessionPtr) override {
     if (event.getType() == Event::Type::AUTHORIZATION_STATUS) {
       std::cout << "Received an event of type AUTHORIZATION_STATUS:\n" + event.toStringPretty(2, 2) << std::endl;
       auto message = event.getMessageList().at(0);
@@ -857,7 +856,6 @@ class MyEventHandler : public EventHandler {
     } else if (event.getType() == Event::Type::FIX) {
       std::cout << "Received an event of type FIX:\n" + event.toStringPretty(2, 2) << std::endl;
     }
-    return true;
   }
 };
 } /* namespace ccapi */
@@ -955,21 +953,16 @@ Bye
 #### Handle events in "immediate" vs. "batching" mode
 
 In general there are 2 ways to handle events.
-* When a `Session` is instantiated with an `eventHandler` argument, it will handle events in immediate mode. The `processEvent` method in the `eventHandler` will be invoked immediately when an `Event` is available.
-* When a `Session` is instantiated without an `eventHandler` argument, it will handle events in batching mode. The evetns will be batched into an internal `Queue<Event>` and can be retrieved by
+* When a `Session` is instantiated with an `eventHandler` argument, it will handle events in immediate mode. The `processEvent` method in the `eventHandler` will be invoked immediately when an `Event` is available, and the invocation will run on the thread where `boost::asio::io_context` runs. When a `Session` is instantiated with an `eventHandler` and an `eventDispatcher` argument, it will also handle events in immediate mode. The `processEvent` method in the `eventHandler` will also be invoked immediately when an `Event` is available, but the invocation will run in the thread(s) provided by the `eventDispatcher` therefore not blocking the thread where `boost::asio::io_context` runs. `EventHandler`s and/or `EventDispatcher`s can be shared among different sessions. Otherwise, different sessions are independent from each other.
+An example can be found [here](example/src/market_data_advanced_subscription/main.cpp).
+* When a `Session` is instantiated without an `eventHandler` argument, it will handle events in batching mode. The events will be batched into an internal `Queue<Event>` and can be retrieved by
 ```
 std::vector<Event> eventList = session.getEventQueue().purge();
 ```
 An example can be found [here](example/src/market_data_advanced_subscription/main.cpp).
 
 #### Thread safety
 * The following methods are implemented to be thread-safe: `Session::sendRequest`, `Session::subscribe`, `Session::sendRequestByFix`, `Session::subscribeByFix`, `Session::setTimer`, all public methods in `Queue`.
-* The `processEvent` method in the `eventHandler` is invoked on one of the internal threads in the `eventDispatcher`. A default `EventDispatcher` with 1 internal thread will be created if no `eventDispatcher` argument is provided in `Session` instantiation. To dispatch events to multiple threads, instantiate `EventDispatcher` with `numDispatcherThreads` set to be the desired number. `EventHandler`s and/or `EventDispatcher`s can be shared among different sessions. Otherwise, different sessions are independent from each other.
-```
-EventDispatcher eventDispatcher(2);
-Session session(sessionOptions, sessionConfigs, &eventHandler, &eventDispatcher);
-```
-An example can be found [here](example/src/market_data_advanced_subscription/main.cpp).
 
 #### Enable library logging
 
@@ -1015,7 +1008,6 @@ sessionPtr->setTimer(
 * Shorten constant strings used as key names in the returned `Element` (e.g. in CmakeLists.txt `add_compile_definitions(CCAPI_BEST_BID_N_PRICE="b")`).
 * Only enable the services and exchanges that you need.
 * Handle events in ["batching" mode](#handle-events-in-immediate-vs-batching-mode) if your application (e.g. market data archiver) isn't latency sensitive.
-* Define macro `CCAPI_USE_SINGLE_THREAD`. It reduces locking overhead for single threaded applications.
 
 ## Known Issues and Workarounds
 * Kraken invalid nonce errors. Give the API key a nonce window (https://support.kraken.com/hc/en-us/articles/360001148023-What-is-a-nonce-window-). We use unix timestamp with microsecond resolution as nonce and therefore a nonce window of 500000 translates to a tolerance of 0.5 second.

binding/csharp/example/execution_management_simple_request/MainProgram.cs

@@ -1,9 +1,8 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       System.Console.WriteLine(string.Format("Received an event:\n{0}", event_.ToStringPretty(2, 2)));
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     if (System.Environment.GetEnvironmentVariable("OKX_API_KEY") is null) {

binding/csharp/example/execution_management_simple_subscription/MainProgram.cs

@@ -1,6 +1,6 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       if (event_.GetType_() == ccapi.Event.Type.SUBSCRIPTION_STATUS) {
         System.Console.WriteLine(string.Format("Received an event of type SUBSCRIPTION_STATUS:\n{0}", event_.ToStringPretty(2, 2)));
         var message = event_.GetMessageList()[0];
@@ -16,8 +16,7 @@ public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       } else if (event_.GetType_() == ccapi.Event.Type.SUBSCRIPTION_DATA) {
         System.Console.WriteLine(string.Format("Received an event of type SUBSCRIPTION_DATA:\n{0}", event_.ToStringPretty(2, 2)));
       }
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     if (System.Environment.GetEnvironmentVariable("OKX_API_KEY") is null) {

binding/csharp/example/fix_simple/MainProgram.cs

@@ -1,6 +1,6 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       if (event_.GetType_() == ccapi.Event.Type.AUTHORIZATION_STATUS) {
         System.Console.WriteLine(string.Format("Received an event of type AUTHORIZATION_STATUS:\n{0}", event_.ToStringPretty(2, 2)));
         var message = event_.GetMessageList()[0];
@@ -21,8 +21,7 @@ public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       } else if (event_.GetType_() == ccapi.Event.Type.FIX) {
         System.Console.WriteLine(string.Format("Received an event of type FIX:\n{0}", event_.ToStringPretty(2, 2)));
       }
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     if (System.Environment.GetEnvironmentVariable("COINBASE_API_KEY") is null) {

binding/csharp/example/handle_exception/MainProgram.cs

@@ -1,14 +1,13 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       try {
         throw new System.Exception("oops");
       } catch (System.Exception e) {
         System.Console.WriteLine(e.ToString());
         System.Environment.Exit(1);
       }
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     var eventHandler = new MyEventHandler();

binding/csharp/example/market_data_multiple_subscription/MainProgram.cs

@@ -1,6 +1,6 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       if (event_.GetType_() == ccapi.Event.Type.SUBSCRIPTION_DATA) {
         foreach (var message in event_.GetMessageList()) {
           var correlationId = message.GetCorrelationIdList()[0];
@@ -15,8 +15,7 @@ public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
           }
         }
       }
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     var eventHandler = new MyEventHandler();

binding/csharp/example/market_data_simple_request/MainProgram.cs

@@ -1,9 +1,8 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       System.Console.WriteLine(string.Format("Received an event:\n{0}", event_.ToStringPretty(2, 2)));
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     var eventHandler = new MyEventHandler();

binding/csharp/example/market_data_simple_subscription/MainProgram.cs

@@ -1,6 +1,6 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
       if (event_.GetType_() == ccapi.Event.Type.SUBSCRIPTION_STATUS) {
         System.Console.WriteLine(string.Format("Received an event of type SUBSCRIPTION_STATUS:\n{0}", event_.ToStringPretty(2, 2)));
       } else if (event_.GetType_() == ccapi.Event.Type.SUBSCRIPTION_DATA) {
@@ -16,8 +16,7 @@ public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
           }
         }
       }
-      return true;
-    }
+          }
   }
   static void Main(string[] args) {
     var eventHandler = new MyEventHandler();

binding/csharp/test/MainProgram.cs

@@ -1,8 +1,7 @@
 class MainProgram {
   class MyEventHandler : ccapi.EventHandler {
-    public override bool ProcessEvent(ccapi.Event event_, ccapi.Session session) {
-      return true;
-    }
+    public override void ProcessEvent(ccapi.Event event_, ccapi.Session session) {
+          }
   }
   static void Main(string[] args) {
     var eventHandler = new MyEventHandler();

binding/go/example/execution_management_simple_request/main.go

@@ -11,10 +11,9 @@ type MyEventHandler struct {
 	ccapi.EventHandler
 }
 
-func (*MyEventHandler) ProcessEvent(event ccapi.Event, session ccapi.Session) bool {
+func (*MyEventHandler) ProcessEvent(event ccapi.Event, session ccapi.Session)  {
 	fmt.Printf("Received an event:\n%s\n", event.ToStringPretty(2, 2))
-	return true
-}
+	}
 
 func main() {
 	if len(os.Getenv("OKX_API_KEY")) == 0 {