+ Fix condition variables and notifications when ForceFlush.

+ Rename `max_sessions` to `max_concurrent_requests` according to https:/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/otlp.md#otlpgrpc-concurrent-requests

Signed-off-by: owent <[email protected]>

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

@@ -78,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,
@@ -96,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)
  {}
 };
 

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;
 };
 
 /**

exporters/otlp/src/otlp_http_client.cc

@@ -751,7 +751,7 @@ void OtlpHttpClient::Export(
  std::unique_lock<std::mutex> lock(session_waker_lock_);
  session_waker_.wait_for(lock, options_.timeout, [this] {
  std::lock_guard<std::recursive_mutex> guard{session_manager_lock_};
- return running_sessions_.size() <= options_.concurrent_sessions;
+ return running_sessions_.size() <= options_.max_concurrent_requests;
  });
 
  cleanupGCSessions();

exporters/otlp/src/otlp_http_exporter.cc

@@ -32,7 +32,7 @@ OtlpHttpExporter::OtlpHttpExporter(const OtlpHttpExporterOptions &options)
  options.console_debug,
  options.timeout,
  options.http_headers,
- options.concurrent_sessions)))
+ options.max_concurrent_requests)))
 {}
 
 OtlpHttpExporter::OtlpHttpExporter(std::unique_ptr<OtlpHttpClient> http_client)

exporters/otlp/src/otlp_http_log_exporter.cc

@@ -34,7 +34,7 @@ OtlpHttpLogExporter::OtlpHttpLogExporter(const OtlpHttpLogExporterOptions &optio
  options.console_debug,
  options.timeout,
  options.http_headers,
- options.concurrent_sessions)))
+ options.max_concurrent_requests)))
 {}
 
 OtlpHttpLogExporter::OtlpHttpLogExporter(std::unique_ptr<OtlpHttpClient> http_client)

sdk/include/opentelemetry/sdk/logs/batch_log_processor.h

@@ -10,6 +10,7 @@
 
 # include <atomic>
 # include <condition_variable>
+# include <cstdint>
 # include <memory>
 # include <thread>
 
@@ -124,12 +125,8 @@ class BatchLogProcessor : public LogProcessor
  /**
  * Exports all logs to the configured exporter.
  *
- * @param was_force_flush_called - A flag to check if the current export is the result
- * of a call to ForceFlush method. If true, then we have to
- * notify the main thread to wake it up in the ForceFlush
- * method.
  */
- void Export(const bool was_for_flush_called);
+ void Export();
 
  /**
  * Called when Shutdown() is invoked. Completely drains the queue of all log records and
@@ -143,14 +140,15 @@ class BatchLogProcessor : public LogProcessor
  struct SynchronizationData
  {
  /* Synchronization primitives */
- std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_;
- std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_;
+ std::condition_variable cv, force_flush_cv, async_shutdown_cv;
+ std::mutex cv_m, force_flush_cv_m, shutdown_m, async_shutdown_m;
 
  /* Important boolean flags to handle the workflow of the processor */
- std::atomic<bool> is_shutdown_;
- std::atomic<bool> is_force_flush_;
- std::atomic<bool> is_force_flush_notified_;
- std::atomic<bool> is_async_shutdown_notified_;
+ std::atomic<bool> is_force_wakeup_background_worker;
+ std::atomic<bool> is_force_flush_pending;
+ std::atomic<bool> is_force_flush_notified;
+ std::atomic<bool> is_shutdown;
+ std::atomic<bool> is_async_shutdown_notified;
  };
 
  /**

sdk/include/opentelemetry/sdk/trace/batch_span_processor.h

@@ -122,12 +122,8 @@ class BatchSpanProcessor : public SpanProcessor
  /**
  * Exports all ended spans to the configured exporter.
  *
- * @param was_force_flush_called - A flag to check if the current export is the result
- * of a call to ForceFlush method. If true, then we have to
- * notify the main thread to wake it up in the ForceFlush
- * method.
  */
- void Export(const bool was_for_flush_called);
+ void Export();
 
  /**
  * Called when Shutdown() is invoked. Completely drains the queue of all its ended spans and
@@ -141,14 +137,15 @@ class BatchSpanProcessor : public SpanProcessor
  struct SynchronizationData
  {
  /* Synchronization primitives */
- std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_;
- std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_;
+ std::condition_variable cv, force_flush_cv, async_shutdown_cv;
+ std::mutex cv_m, force_flush_cv_m, shutdown_m, async_shutdown_m;
 
  /* Important boolean flags to handle the workflow of the processor */
- std::atomic<bool> is_shutdown_;
- std::atomic<bool> is_force_flush_;
- std::atomic<bool> is_force_flush_notified_;
- std::atomic<bool> is_async_shutdown_notified_;
+ std::atomic<bool> is_force_wakeup_background_worker;
+ std::atomic<bool> is_force_flush_pending;
+ std::atomic<bool> is_force_flush_notified;
+ std::atomic<bool> is_shutdown;
+ std::atomic<bool> is_async_shutdown_notified;
  };
 
  /**