10
10
#include < ostream>
11
11
#include < ratio>
12
12
#include < thread>
13
- #include < type_traits>
14
13
#include < utility>
15
14
16
15
#include " opentelemetry/common/timestamp.h"
24
23
#include " opentelemetry/sdk/metrics/push_metric_exporter.h"
25
24
#include " opentelemetry/version.h"
26
25
27
- #if defined(_MSC_VER)
28
- # pragma warning(suppress : 5204)
29
- # include < future>
30
- #else
31
- # include < future>
32
- #endif
33
-
34
26
#if OPENTELEMETRY_HAVE_EXCEPTIONS
35
27
# include < exception>
36
28
#endif
@@ -98,11 +90,9 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
98
90
worker_thread_instrumentation_->OnStart ();
99
91
}
100
92
#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
101
-
102
93
do
103
94
{
104
95
auto start = std::chrono::steady_clock::now ();
105
-
106
96
#ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
107
97
if (worker_thread_instrumentation_ != nullptr )
108
98
{
@@ -134,7 +124,6 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
134
124
worker_thread_instrumentation_->BeforeWait ();
135
125
}
136
126
#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
137
-
138
127
std::unique_lock<std::mutex> lk (cv_m_);
139
128
cv_.wait_for (lk, remaining_wait_interval_ms, [this ]() {
140
129
if (is_force_wakeup_background_worker_.load (std::memory_order_acquire))
@@ -151,7 +140,6 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
151
140
worker_thread_instrumentation_->AfterWait ();
152
141
}
153
142
#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
154
-
155
143
} while (IsShutdown () != true );
156
144
157
145
#ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
@@ -164,61 +152,39 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
164
152
165
153
bool PeriodicExportingMetricReader::CollectAndExportOnce ()
166
154
{
167
- std::atomic<bool > cancel_export_for_timeout{false };
168
-
169
155
std::uint64_t notify_force_flush = force_flush_pending_sequence_.load (std::memory_order_acquire);
170
- std::unique_ptr<std::thread> task_thread;
171
-
172
156
#if OPENTELEMETRY_HAVE_EXCEPTIONS
173
157
try
174
158
{
175
159
#endif
176
- std::promise<void > sender;
177
- auto receiver = sender.get_future ();
178
-
179
- task_thread.reset (
180
- new std::thread ([this , &cancel_export_for_timeout, sender = std::move (sender)] {
181
160
#ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
182
- if (collect_thread_instrumentation_ != nullptr )
183
- {
184
- collect_thread_instrumentation_->OnStart ();
185
- collect_thread_instrumentation_->BeforeLoad ();
186
- }
161
+ if (collect_thread_instrumentation_ != nullptr )
162
+ {
163
+ collect_thread_instrumentation_->OnStart ();
164
+ collect_thread_instrumentation_->BeforeLoad ();
165
+ }
187
166
#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
188
-
189
- this ->Collect ([this , &cancel_export_for_timeout](ResourceMetrics &metric_data) {
190
- if (cancel_export_for_timeout.load (std::memory_order_acquire))
191
- {
192
- OTEL_INTERNAL_LOG_ERROR (
193
- " [Periodic Exporting Metric Reader] Collect took longer configured time: "
194
- << this ->export_timeout_millis_ .count () << " ms, and timed out" );
195
- return false ;
196
- }
197
- this ->exporter_ ->Export (metric_data);
198
- return true ;
199
- });
200
-
201
- const_cast <std::promise<void > &>(sender).set_value ();
167
+ auto start = std::chrono::steady_clock::now ();
168
+ this ->Collect ([this , &start](ResourceMetrics &metric_data) {
169
+ auto end = std::chrono::steady_clock::now ();
170
+ if ((end - start) > this ->export_timeout_millis_ )
171
+ {
172
+ OTEL_INTERNAL_LOG_ERROR (
173
+ " [Periodic Exporting Metric Reader] Collect took longer configured time: "
174
+ << this ->export_timeout_millis_ .count () << " ms, and timed out" );
175
+ return false ;
176
+ }
177
+ this ->exporter_ ->Export (metric_data);
178
+ return true ;
179
+ });
202
180
203
181
#ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
204
- if (collect_thread_instrumentation_ != nullptr )
205
- {
206
- collect_thread_instrumentation_->AfterLoad ();
207
- collect_thread_instrumentation_->OnEnd ();
208
- }
209
- #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
210
- }));
211
-
212
- std::future_status status;
213
- do
182
+ if (collect_thread_instrumentation_ != nullptr )
214
183
{
215
- status = receiver.wait_for (std::chrono::milliseconds (export_timeout_millis_));
216
- if (status == std::future_status::timeout)
217
- {
218
- cancel_export_for_timeout.store (true , std::memory_order_release);
219
- break ;
220
- }
221
- } while (status != std::future_status::ready);
184
+ collect_thread_instrumentation_->AfterLoad ();
185
+ collect_thread_instrumentation_->OnEnd ();
186
+ }
187
+ #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
222
188
#if OPENTELEMETRY_HAVE_EXCEPTIONS
223
189
}
224
190
catch (std::exception &e)
@@ -235,11 +201,6 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
235
201
}
236
202
#endif
237
203
238
- if (task_thread && task_thread->joinable ())
239
- {
240
- task_thread->join ();
241
- }
242
-
243
204
std::uint64_t notified_sequence = force_flush_notified_sequence_.load (std::memory_order_acquire);
244
205
while (notify_force_flush > notified_sequence)
245
206
{
0 commit comments