Merge pull request #299 from crypto-chassis/develop

Release
crypto-chassis · Aug 24, 2022 · 67f5c30 · 67f5c30
2 parents 50931bc + b4bcbae
commit 67f5c30
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 1 deletion.
diff --git a/README.md b/README.md
@@ -914,6 +914,7 @@ session.serviceByServiceNameExchangeMap[CCAPI_EXECUTION_MANAGEMENT][CCAPI_EXCHAN
 * Only enable the services and exchanges that you need.
 * Use FIX API instead of REST API.
 * 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.
 
 ## Applications
 

diff --git a/include/ccapi_cpp/ccapi_queue.h b/include/ccapi_cpp/ccapi_queue.h
@@ -16,7 +16,9 @@ class Queue {
   std::string EXCEPTION_QUEUE_EMPTY = "queue is empty";
   explicit Queue(const size_t maxSize = 0) : maxSize(maxSize) {}
   void pushBack(T&& t) {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     if (this->maxSize <= 0 || this->queue.size() < this->maxSize) {
       CCAPI_LOGGER_TRACE("this->queue.size() = " + size_tToString(this->queue.size()));
       this->queue.push_back(t);
@@ -25,7 +27,9 @@ class Queue {
     }
   }
   T popBack() {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     if (this->queue.empty()) {
       throw std::runtime_error(EXCEPTION_QUEUE_EMPTY);
     } else {
@@ -35,13 +39,17 @@ class Queue {
     }
   }
   std::vector<T> purge() {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     std::vector<T> p;
     std::swap(p, this->queue);
     return p;
   }
   void removeAll(std::vector<T>& c) {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     if (c.empty()) {
       c = std::move(this->queue);
     } else {
@@ -51,19 +59,25 @@ class Queue {
     this->queue.clear();
   }
   size_t size() const {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     return this->queue.size();
   }
   bool empty() const {
+#ifndef CCAPI_USE_SINGLE_THREAD
     std::lock_guard<std::mutex> lock(this->m);
+#endif
     return this->queue.empty();
   }
 #ifndef CCAPI_EXPOSE_INTERNAL
 
  private:
 #endif
   std::vector<T> queue;
+#ifndef CCAPI_USE_SINGLE_THREAD
   mutable std::mutex m;
+#endif
   size_t maxSize{};
 };
 } /* namespace ccapi */

diff --git a/include/ccapi_cpp/ccapi_session.h b/include/ccapi_cpp/ccapi_session.h
@@ -620,6 +620,18 @@ class Session {
     } else {
       if (this->eventHandler) {
         CCAPI_LOGGER_TRACE("handle event in immediate mode");
+#ifdef CCAPI_USE_SINGLE_THREAD
+        bool shouldContinue = true;
+        try {
+          shouldContinue = this->eventHandler->processEvent(event, this);
+        } catch (const std::runtime_error& e) {
+          CCAPI_LOGGER_ERROR(e.what());
+        }
+        if (!shouldContinue) {
+          CCAPI_LOGGER_DEBUG("about to pause the event dispatcher");
+          this->eventDispatcher->pause();
+        }
+#else
         this->eventDispatcher->dispatch([that = this, event = std::move(event)] {
           bool shouldContinue = true;
           try {
@@ -632,6 +644,7 @@ class Session {
             that->eventDispatcher->pause();
           }
         });
+#endif
       } else {
         CCAPI_LOGGER_TRACE("handle event in batching mode");
         this->eventQueue.pushBack(std::move(event));

diff --git a/include/ccapi_cpp/service/ccapi_service_context.h b/include/ccapi_cpp/service/ccapi_service_context.h
@@ -15,7 +15,7 @@ class ServiceContext CCAPI_FINAL {
   typedef wspp::lib::asio::io_service IoContext;
   typedef wspp::lib::shared_ptr<wspp::lib::asio::io_service> IoContextPtr;
   struct CustomClientConfig : public wspp::config::asio_tls_client {
-#ifdef WEBSOCKETPP_ENABLE_SINGLE_THREADING
+#ifdef CCAPI_USE_SINGLE_THREAD
     typedef wspp::config::asio_tls_client base;
     static bool const enable_multithreading = false;
     struct transport_config : public base::transport_config {