Added max async export support using separate AsyncBatchSpan/LogProcessor

[DebajitDas]

Fixes #1239

Changes:

1. Added separate asyncbatchspan/log processor as discussed.
2. Added max async export support, where the number of running exports are tracked with unique id.
3. If the running exports are busy and no callback is receieved, any new recordables are discarded.
4. Added test case for AsyncShutdown if no callback is received. Shutdown should not block for ever.

Please provide a brief description of the changes here.

For significant contributions please make sure you have completed the following items:

• CHANGELOG.md updated for non-trivial changes
• Unit tests have been added
• Changes in public API reviewed

[ThomsonTan]

Could we add a CI build for ENABLE_ASYNC_EXPORT?

[DebajitDas]

Yes, I will add on this PR

[owent]

There are same problems between log processor and span processor.
Could you please add timeout mechanism for these two proccesors in case of the exporter miss the callback and add some tests about it to ensure the processors will not wait for ever?

[owent]

I think we shouldn't save span of recordables here, it will be invalid after this function return.Also the std::unique_ptr<Recordable> in spans_arr may be moved into exporter after calling exporter_->Export and left all elements nullptr in it.

[DebajitDas]

ok, should we save any state for async export? If not, will atomic counter suffice to keep track of all running exports?

[owent]

I think we can't check which callback is missing or timeout, so we can't recover it if we only has the atomic counter.
According to open-telemetry/opentelemetry-specification#2434 (comment)_ and open-telemetry/opentelemetry-specification#2434, it's the responsibility of exporters to do the retry logic. I think we can just save a sequence or a id related to the callback , but not every recordable.

[DebajitDas]

I have one more query:
Lets assume max_async_export is set to 8. Now, no callbacks are received on all 8 async export calls. Now, what should we do before we make the 9th async export - Should we discard the recordables and do not make any export call ? What is your suggestion?

[owent]

I think we can just drop it, just like the sync mode.

[lalitb]

Just to be clear of the question, if the max number of async requests is ongoing, and the circular queue is full, we should drop the incoming recordable ( as in sync mode).

[owent]

Should we still need a indefinite loop to wait all running exports to finish in destructor? Or we may start to destroy a exporter when it's still running.

[DebajitDas]

Waiting logic to wait on all running exports to finish is added in Shutdown function.

[owent]

We need to call async_export_waker.notify_all() here, or all threads call Export and Shutdown will be blocked util timeout.

[DebajitDas]

Yeah will do this.

[owent]

Should we use async_export_waker to wait a event instead of busy wait here?

[DebajitDas]

Are you talking about async_export_waker.notify_all() to be called in other areas?

[owent]

I mean while (CleanUpGarbageAsyncData() == false); is a busy wait. CleanUpGarbageAsyncData only lock async_export_data_m, and this lock has a very small critial section, it's almost busy wait when worker thread is exited when exporter is still exporting data in other thread.

[owent]

Why do we need sleep here?I think it's better to use a condition_variable to get notify if we want to know when jobs are done and it's ready to send next round datas.
Sometimes the CI runner is slow and can not finish this job in 1 millisecond.

[DebajitDas]

ok will take care this.

[owent]

@lalitb Could we add async-changes into .github/workflows/ci.yml ?

pull_request:
    branches: [ main, async-changes]

[lalitb]

@lalitb Could we add async-changes into .github/workflows/ci.yml ?

pull_request:
    branches: [ main, async-changes]

Yes, we should add it, and can be removed once the feature branch is merged to main.

[owent]

@lalitb Could we add async-changes into .github/workflows/ci.yml ?

pull_request:
    branches: [ main, async-changes]

Yes, we should add it, and can be removed once the feature branch is merged to main.

Thanks and I create #1309

[lalitb]

Why do we need a span of recordable here? It is just a reference and may get invalidated.

[owent]

Could you enable -DENABLE_ASYNC_EXPORT_PREVIEW for all jobs but left one without it?I think we also need benchmark,address sanitizer, thread sanitizer and so on for this.

[DebajitDas]

Could you please point me to some examples to refer to for benchmark, address sanitizer and thread sanitizer for this?

[owent]

Sorry, I missed something, bazel.asan, bazel.tsan and bazel.valgrind already enable async APIs now, but benchmark still use $BAZEL_OPTIONS, I think we should also use $BAZEL_OPTIONS_ASYNC.
And we should also add --copt=-DENABLE_ASYNC_EXPORT_PREVIEW or -DWITH_ASYNC_EXPORT_PREVIEW=ON into all actions in ci/do_ci.ps1, it's used on Windows.

[owent]

@owent I have not added the test case of async forceflush when no callback is received. Because currently forceflush block indefinitely if no callback is recieved. It blocks at wait_result, because wait_result never becomes true. Need your help in Force Flush.

On my suggestion, we can use a LRU map to replace export_ids and export_ids_flag to resolve this problem. After calling Export(); in DoBackgroundWork(), we can peek all running but timeout exporting and force to call callback.

We can use list to store exporting id and the state to check if the callback already called(in case of we call callback in DoBackgroundWork() and then it be called again by exporters), and we can use id as key of a unordered_map and use the iterator of the list before as value to implement this LRU map,
here is a sample of lru_map in my project.

[DebajitDas]

On my suggestion, we can use a LRU map to replace export_ids and export_ids_flag to resolve this problem. After calling Export(); in DoBackgroundWork(), we can peek all running but timeout exporting and force to call callback.

If we force call callback for all timed-out exports, we would free up the ids to be re-used by new exports. This way we would be allowing number of running exports > max_export_async. Do you have any thoughts on this?

[owent]

On my suggestion, we can use a LRU map to replace export_ids and export_ids_flag to resolve this problem. After calling Export(); in DoBackgroundWork(), we can peek all running but timeout exporting and force to call callback.

If we force call callback for all timed-out exports, we would free up the ids to be re-used by new exports. This way we would be allowing number of running exports > max_export_async. Do you have any thoughts on this?

Maybe we can add a interface to notify exporter to abort timeout exporting? What your thought about it ?
@lalitb

[lalitb]

Maybe we can add a interface to notify exporter to abort timeout exporting? What your thought about it ?
@lalitb

If there are max_export_async exports running forever, the processor should start ignoring the new exports. This is also the current behavior in the case of sequential exports. There may be underlying issues with the network, export endpoint, or the exporter logic, I don't think the processor should try to recover the situation. These are my view, but feel free to comment if they don't seem correct :)

[DebajitDas]

Maybe we can add a interface to notify exporter to abort timeout exporting? What your thought about it ?
@lalitb

If there are max_export_async exports running forever, the processor should start ignoring the new exports. This is also the current behavior in the case of sequential exports. There may be underlying issues with the network, export endpoint, or the exporter logic, I don't think the processor should try to recover the situation. These are my view, but feel free to comment if they don't seem correct :)

Just to confirm, I am re-iterating what the behaviour is now in this PR

1. In case of normal export calls, if all async exports are busy and no callbacks are recieved, any new exports would be discarded.
2. In case of force-flush is being called and no callback is received, force-flush blocks indefinitely.
3. In case shutdown or destructor is called, if any running exports do not return within timeout, shutdown will block till timeout and continue.

[lalitb]

Just to confirm, I am re-iterating what the behaviour is now in this PR

1. In case of normal export calls, if all async exports are busy and no callbacks are recieved, any new exports would be discarded.
2. In case of force-flush is being called and no callback is received, force-flush blocks indefinitely.
3. In case shutdown or destructor is called, if any running exports do not return within timeout, shutdown will block till timeout and continue.

Thanks for summarising - 1 and 3 look good to me. I need to check the force-flush code once again, but can't we make it return/fail (without changing the exporter interface) if exports don't return within the timeout. It should definitely not block indefinitely.

[DebajitDas]

Just to confirm, I am re-iterating what the behaviour is now in this PR

1. In case of normal export calls, if all async exports are busy and no callbacks are recieved, any new exports would be discarded.
2. In case of force-flush is being called and no callback is received, force-flush blocks indefinitely.
3. In case shutdown or destructor is called, if any running exports do not return within timeout, shutdown will block till timeout and continue.

Thanks for summarising - 1 and 3 look good to me. I need to check the force-flush code once again, but can't we make it return/fail (without changing the exporter interface) if exports don't return within the timeout. It should definitely not block indefinitely.

@owent I am not clear if there is any specific reason to keep ForceFlush blocking (even though there is timeout)? wait_result never becomes true even if timeout expires, without callback being called.

[lalitb]

Just to confirm, I am re-iterating what the behaviour is now in this PR

1. In case of normal export calls, if all async exports are busy and no callbacks are recieved, any new exports would be discarded.
2. In case of force-flush is being called and no callback is received, force-flush blocks indefinitely.
3. In case shutdown or destructor is called, if any running exports do not return within timeout, shutdown will block till timeout and continue.

Thanks for summarising - 1 and 3 look good to me. I need to check the force-flush code once again, but can't we make it return/fail (without changing the exporter interface) if exports don't return within the timeout. It should definitely not block indefinitely.

@owent I am not clear if there is any specific reason to keep ForceFlush blocking (even though there is timeout)? wait_result never becomes true even if timeout expires, without callback being called.

Even in the initial sync implementation for BatchSpanProcessor, force-flush block indefinitely, though specs clearly states it should abort/complete within the timeout. So seems we were never specs compliant :)

[DebajitDas]

compliant

In that case, let me not touch ForceFlush for now in this PR. We can take up in separate PR to fix that. Do you agree?

[owent]

compliant

In that case, let me not touch ForceFlush for now in this PR. We can take up in separate PR to fix that. Do you agree?

Agree. some exporters implement the timeout/abort while some do not, Ithink we can finish it in the future.

[lalitb]

compliant

In that case, let me not touch ForceFlush for now in this PR. We can take up in separate PR to fix that. Do you agree?

Agree. some exporters implement the timeout/abort while some do not, Ithink we can finish it in the future.

Agree.

[codecov]

Codecov Report

❗ No coverage uploaded for pull request base (async-changes@465158c). Click here to learn what that means.
The diff coverage is n/a.

@@               Coverage Diff                @@
##             async-changes    #1306   +/-   ##
================================================
  Coverage                 ?   91.46%           
================================================
  Files                    ?      220           
  Lines                    ?     8164           
  Branches                 ?        0           
================================================
  Hits                     ?     7466           
  Misses                   ?      698           
  Partials                 ?        0           

[owent]

LGTM

[lalitb]

LGTM

@owent There was a discussing with @DebajitDas on this PR in today's community meeting. We felt it would be better from maintainability prospective to introduce a separate AsyncBatchSpanProcessor / AsyncBatchLogProcessor and have the async changes for processor confined there. We can still have feature flag to enable/disable these two components. Let us know if that would be fine? We can afterwards see if similar thing can be done for Exporter.

[owent]

LGTM

@owent There was a discussing with @DebajitDas on this PR in today's community meeting. We felt it would be better from maintainability prospective to introduce a separate AsyncBatchSpanProcessor / AsyncBatchLogProcessor and have the async changes for processor confined there. We can still have feature flag to enable/disable these two components. Let us know if that would be fine? We can afterwards see if similar thing can be done for Exporter.

That's fine. I think we need think about how to share codes with aync and async exporter by this way.

[owent]

There a a lot similar code between async and sync processors and tests. Could we add some classes to maintain these codes instead of jusy copy them?

[owent]

Should we create one test for AddProcessor and another for AsyncBatchLogProcessor, just like exporters/otlp/test/otlp_http_exporter_test.cc ?

[DebajitDas]

Thanks, done in the next PR

[owent]

I think we can only remove this and only keep the constructor using const AsyncBatchLogProcessorOptions &options, just like AsyncBatchSpanProcessor. The similar constructor in BatchLogProcessor is just for compatibility.

[DebajitDas]

Thanks, added in the PR

[DebajitDas]

There a a lot similar code between async and sync processors and tests. Could we add some classes to maintain these codes instead of jusy copy them?

Can we make AsyncBatchSpanProcessor derived from BatchSpanProcessor ? This way we can reduce lots of duplicate code. What do you think?

[owent]

There a a lot similar code between async and sync processors and tests. Could we add some classes to maintain these codes instead of jusy copy them?

Can we make AsyncBatchSpanProcessor derived from BatchSpanProcessor ? This way we can reduce lots of duplicate code. What do you think?

I think it's fine right now. AsyncBatchSpanProcessor has all members of BatchSpanProcessor in current implement.
What's your opinion about this way? @lalitb

[lalitb]

I think it's fine right now. AsyncBatchSpanProcessor has all members of BatchSpanProcessor in current implement.
What's your opinion about this way? @lalitb

It looks me good idea to derive Async if that allows us to eliminate lots of duplicate code. Unless @owent you see any issue in this, will leave it to you to decide :)
I will be spending time over this weekend for in-depth review of exporter and processor changes.

[owent]

We should not add virtual keyword for overried methods. Clang will warning this.

[DebajitDas]

ok, will make the change.

[owent]

Just want to discuss, is it simpiler to declare struct AsyncBatchSpanProcessorOptions : public BatchSpanProcessorOptions ?

[DebajitDas]

Thanks, this is much better.

[owent]

We should shutdown here.If it's shutdown in destructor of base class. the AsyncBatchLogProcessor::Shutdown will not be called.

[DebajitDas]

Agreed

[owent]

I think we should loop wair_for scheduled_delay_millis_ in a loop until timeout is reached.When async_export_waker.notify_all() is called between predicate callback and the real lock operation.This may wait for ever.

[DebajitDas]

I am not clear on this. Could you please elaborate on how this might wait forever or share some sample code which fixes this?

[owent]

Just like BatchLogProcessor::ForceFlush at https:/DebajitDas/opentelemetry-cpp/blob/export-changes/sdk/src/logs/batch_log_processor.cc#L112.

The condition_variable::wait_for(...) implementation calls predicate callback and then lock API(pthread_mutex_lock for example). If condition_variable::notify_one() or condition_variable::notify_all() is call between callback and then lock API. Then this condition_variable will wait for timeout without notify any more, and timeout may be very large.

[DebajitDas]

Thanks, for the clarification. I understood your concern here.
I re-looked into the BatchLogProcessor::ForceFlush at https:/DebajitDas/opentelemetry-cpp/blob/export-changes/sdk/src/logs/batch_log_processor.cc#L112. So it handles this case as follows:

1. when the timeout is set as std::chrono::duration<Rep, Period>::max() or too large, we are waiting on conditiion variable for scheduled_delay_millis_ in loop untill the predicate is true.
2. When the timeout is set as some finite value, we wait till timeout.

Now, if we introduce the above concept in Shutdown, then the predicate in case 1 above would never be true if no callback is called and hence it would block indefinitely. Is this valid then?

[owent]

The problem is the predicate counld be true, but the condition_variable may not be notified any more when shutdown.
Let's looks into the implement of wait_for in libstdc++ (GCC 11.2.0).

template<typename _Clock, typename _Duration, typename _Predicate>
bool wait_until(unique_lock<mutex>& __lock, const chrono::time_point<_Clock, _Duration>& __atime, _Predicate __p) {
  while (!__p())
    if (wait_until(__lock, __atime) == cv_status::timeout) // The real lock operation
      return __p();
  return true;
}

template<typename _Rep, typename _Period>
cv_status  wait_for(unique_lock<mutex>& __lock, const chrono::duration<_Rep, _Period>& __rtime) {
  using __dur = typename steady_clock::duration;
  return wait_until(__lock, steady_clock::now() + chrono::__detail::ceil<__dur>(__rtime));
}

And the threads run as below:

The thread call Shutdown()	Background woker thread
while (!__p())	-	__p() returns false
-	export_data->export_ids.push(id);
-	export_data->export_ids_flag[id - 1] = true;	__p() should return true now
-	export_data_storage->async_export_waker.notify_all();	notify_all() is just ignored because it not locked yet.
if (wait_until(__lock, __atime) == cv_status::timeout)	-	async_export_waker is locked now, and wait until __atime

What I mean is just let async_export_waker have a change to run __p() again and break this loop in this situation.
And this happened several times in CI jobswhen I pushed my first async version of OtlpHttpClient before.

[owent]

BTW:

if(WITH_METRICS_PREVIEW)
  add_definitions(-DENABLE_METRICS_PREVIEW)
endif()

These scripts in CMakeLists.txt should be moved into api/CMakeLists.txt and be replaced by

if(WITH_METRICS_PREVIEW)
  target_compile_definitions(opentelemetry_api INTERFACE ENABLE_METRICS_PREVIEW)
endif()

Other options are already be moved in main branch. And we could merge from main some times later, could you please move ENABLE_METRICS_PREVIEW in this PR?

[DebajitDas]

BTW:

if(WITH_METRICS_PREVIEW)
  add_definitions(-DENABLE_METRICS_PREVIEW)
endif()

These scripts in CMakeLists.txt should be moved into api/CMakeLists.txt and be replaced by

if(WITH_METRICS_PREVIEW)
  target_compile_definitions(opentelemetry_api INTERFACE ENABLE_METRICS_PREVIEW)
endif()

Other options are already be moved in main branch. And we could merge from main some times later, could you please move ENABLE_METRICS_PREVIEW in this PR?

You must be talking about moving WITH_ASYNC_EXPORT_PREVIEW instead.

[owent]

BTW:

if(WITH_METRICS_PREVIEW)
  add_definitions(-DENABLE_METRICS_PREVIEW)
endif()

These scripts in CMakeLists.txt should be moved into api/CMakeLists.txt and be replaced by

if(WITH_METRICS_PREVIEW)
  target_compile_definitions(opentelemetry_api INTERFACE ENABLE_METRICS_PREVIEW)
endif()

Other options are already be moved in main branch. And we could merge from main some times later, could you please move ENABLE_METRICS_PREVIEW in this PR?

You must be talking about moving WITH_ASYNC_EXPORT_PREVIEW instead.

Yes, sorry for my mistake.

[owent]

LGTM now.