Cocurrency otlp http session

api/include/opentelemetry/common/spin_lock_mutex.h

@@ -59,6 +59,26 @@ class SpinLockMutex
  SpinLockMutex &operator=(const SpinLockMutex &) = delete;
  SpinLockMutex &operator=(const SpinLockMutex &) volatile = delete;
 
+ static inline void fast_yield() noexcept
+ {
+// Issue a Pause/Yield instruction while spinning.
+#if defined(_MSC_VER)
+ YieldProcessor();
+#elif defined(__i386__) || defined(__x86_64__)
+# if defined(__clang__)
+ _mm_pause();
+# else
+ __builtin_ia32_pause();
+# endif
+#elif defined(__arm__)
+ // This intrinsic should fail to be found if YIELD is not supported on the current
+ // processor.
+ __yield();
+#else
+ // TODO: Issue PAGE/YIELD on other architectures.
+#endif
+ }
+
  /**
  * Attempts to lock the mutex. Return immediately with `true` (success) or `false` (failure).
  */
@@ -91,22 +111,7 @@ class SpinLockMutex
  {
  return;
  }
-// Issue a Pause/Yield instruction while spinning.
-#if defined(_MSC_VER)
- YieldProcessor();
-#elif defined(__i386__) || defined(__x86_64__)
-# if defined(__clang__)
- _mm_pause();
-# else
- __builtin_ia32_pause();
-# endif
-#elif defined(__arm__)
- // This intrinsic should fail to be found if YIELD is not supported on the current
- // processor.
- __yield();
-#else
- // TODO: Issue PAGE/YIELD on other architectures.
-#endif
+ fast_yield();
  }
  // Yield then try again (goal ~100ns)
  std::this_thread::yield();

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_client.h

@@ -11,6 +11,7 @@
 
 #include "opentelemetry/common/spin_lock_mutex.h"
 #include "opentelemetry/ext/http/client/http_client.h"
+#include "opentelemetry/nostd/variant.h"
 #include "opentelemetry/sdk/common/exporter_utils.h"
 
 #include "opentelemetry/exporters/otlp/otlp_environment.h"
@@ -19,6 +20,7 @@
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
+#include <functional>
 #include <list>
 #include <memory>
 #include <mutex>
@@ -76,8 +78,8 @@ struct OtlpHttpClientOptions
  // Additional HTTP headers
  OtlpHeaders http_headers = GetOtlpDefaultHeaders();
 
- // Concurrent sessions
- std::size_t concurrent_sessions = 8;
+ // Concurrent requests
+ std::size_t max_concurrent_requests = 8;
 
  inline OtlpHttpClientOptions(nostd::string_view input_url,
  HttpRequestContentType input_content_type,
@@ -94,7 +96,7 @@ struct OtlpHttpClientOptions
  console_debug(input_console_debug),
  timeout(input_timeout),
  http_headers(input_http_headers),
- concurrent_sessions(input_concurrent_sessions)
+ max_concurrent_requests(input_concurrent_sessions)
  {}
 };
 
@@ -112,12 +114,22 @@ class OtlpHttpClient
  ~OtlpHttpClient();
 
  /**
- * Export
+ * Sync export
  * @param message message to export, it should be ExportTraceServiceRequest,
  * ExportMetricsServiceRequest or ExportLogsServiceRequest
  */
  sdk::common::ExportResult Export(const google::protobuf::Message &message) noexcept;
 
+ /**
+ * Async export
+ * @param message message to export, it should be ExportTraceServiceRequest,
+ * ExportMetricsServiceRequest or ExportLogsServiceRequest
+ * @param result_callback callback to call when the exporting is done
+ */
+ void Export(
+ const google::protobuf::Message &message,
+ std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept;
+
  /**
  * Shut down the HTTP client.
  * @param timeout an optional timeout, the default timeout of 0 means that no
@@ -134,14 +146,49 @@ class OtlpHttpClient
  void ReleaseSession(const opentelemetry::ext::http::client::Session &session) noexcept;
 
 private:
+ struct HttpSessionData
+ {
+ std::shared_ptr<opentelemetry::ext::http::client::Session> session;
+ std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle;
+
+ inline HttpSessionData() = default;
+
+ inline explicit HttpSessionData(
+ std::shared_ptr<opentelemetry::ext::http::client::Session> &&input_session,
+ std::shared_ptr<opentelemetry::ext::http::client::EventHandler> &&input_handle)
+ {
+ session.swap(input_session);
+ event_handle.swap(input_handle);
+ }
+
+ inline explicit HttpSessionData(HttpSessionData &&other)
+ {
+ session.swap(other.session);
+ event_handle.swap(other.event_handle);
+ }
+
+ inline HttpSessionData &operator=(HttpSessionData &&other) noexcept
+ {
+ session.swap(other.session);
+ event_handle.swap(other.event_handle);
+ return *this;
+ }
+ };
+
+ /**
+ * @brief Create a Session object or return a error result
+ *
+ * @param message The message to send
+ */
+ nostd::variant<sdk::common::ExportResult, HttpSessionData> createSession(
+ const google::protobuf::Message &message,
+ std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept;
+
  /**
  * Add http session and hold it's lifetime.
- * @param session the session to add
- * @param event_handle the event handle of this session
+ * @param session_data the session to add
  */
- void addSession(
- std::shared_ptr<opentelemetry::ext::http::client::Session> session,
- std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle) noexcept;
+ void addSession(HttpSessionData &&session_data) noexcept;
 
  /**
  * @brief Real delete all sessions and event handles.
@@ -165,28 +212,6 @@ class OtlpHttpClient
  OtlpHttpClient(OtlpHttpClientOptions &&options,
  std::shared_ptr<ext::http::client::HttpClient> http_client);
 
- struct HttpSessionData
- {
- std::shared_ptr<opentelemetry::ext::http::client::Session> session;
- std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle;
-
- inline HttpSessionData() = default;
-
- inline explicit HttpSessionData(
- std::shared_ptr<opentelemetry::ext::http::client::Session> &&input_session,
- std::shared_ptr<opentelemetry::ext::http::client::EventHandler> &&input_handle)
- {
- session.swap(input_session);
- event_handle.swap(input_handle);
- }
-
- inline explicit HttpSessionData(HttpSessionData &&other)
- {
- session.swap(other.session);
- event_handle.swap(other.event_handle);
- }
- };
-
  // Stores if this HTTP client had its Shutdown() method called
  bool is_shutdown_;
 

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h

@@ -52,8 +52,9 @@ struct OtlpHttpExporterOptions
  // Additional HTTP headers
  OtlpHeaders http_headers = GetOtlpDefaultHeaders();
 
- // Concurrent sessions
- std::size_t concurrent_sessions = 8;
+ // Concurrent requests
+ // https:/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/otlp.md#otlpgrpc-concurrent-requests
+ std::size_t max_concurrent_requests = 8;
 };
 
 /**

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h

@@ -52,8 +52,9 @@ struct OtlpHttpLogExporterOptions
  // Additional HTTP headers
  OtlpHeaders http_headers = GetOtlpDefaultLogHeaders();
 
- // Concurrent sessions
- std::size_t concurrent_sessions = 8;
+ // Concurrent requests
+ // https:/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/otlp.md#otlpgrpc-concurrent-requests
+ std::size_t max_concurrent_requests = 8;
 };
 
 /**