Address Collect/Shutdown thread safety for MetricReader #2506
Conversation
Codecov Report
@@ Coverage Diff @@
## main #2506 +/- ##
==========================================
+ Coverage 80.51% 80.52% +0.01%
==========================================
Files 253 253
Lines 8503 8529 +26
==========================================
+ Hits 6846 6868 +22
- Misses 1657 1661 +4
|
reyang
changed the title
| @@ -71,13 +76,48 @@ public bool Collect(int timeoutMilliseconds = Timeout.Infinite) | |||
| { | |||
| Guard.InvalidTimeout(timeoutMilliseconds, nameof(timeoutMilliseconds)); | |||
|
|
|||
| var kickoff = false; | |||
There was a problem hiding this comment.
I realized that kickoff might be a confusing name, maybe shouldRunCollect is better?
|
|
||
| if (tcs == null) | ||
| { | ||
| lock (this.newTaskLock) |
There was a problem hiding this comment.
I guess making this a spin lock might give better performance as the actual wait time should be short in normal cases.
There was a problem hiding this comment.
I think that usually threads will not wait here and the cost of a lock is very cheap in such a case. If the threads will wait here - it may be caused by some troubles and the wait times may occur not so short. I suggest postponing it.
| try | ||
| { | ||
| return this.OnCollect(timeoutMilliseconds); | ||
| if (kickoff) |
There was a problem hiding this comment.
Consider simplify the flow (I realized it when I rewrote the logic in #2505):
if (!shouldRunCollect)
{
// this shouldn't throw exception since tcs has it's value explicitly set
return Task.WaitAny(tcs.Task, this.shutdownTcs.Task, Task.Delay(timeoutMilliseconds)) == 0 ? tcs.Task.Result :
}
var result = false;
try
{
lock (this.onCollectLock)
{
this.collectionTcs = null;
result = this.OnCollect(timeoutMilliseconds);
}
}
catch (Exception)
{
// TODO: OpenTelemetrySdkEventSource.Log.SpanProcessorException(nameof(this.Shutdown), ex);
}
tcs.TrySetResult(result);
return result;
| private AggregationTemporality preferredAggregationTemporality = CumulativeAndDelta; | ||
| private AggregationTemporality supportedAggregationTemporality = CumulativeAndDelta; | ||
| private int shutdownCount; | ||
| private int shutdownTimeout = int.MinValue; |
There was a problem hiding this comment.
We don't need this anymore?
shutdownCount might be better?
if (Interlocked.CompareExchange(ref this.shutdownCount, 1, 0) != 0)
{
return false; // shutdown already called
}Up to you.
| { | ||
| return false; // shutdown already called | ||
| } | ||
|
|
||
| this.shutdownTimeout = timeoutMilliseconds; |
There was a problem hiding this comment.
Remove this line (it's already implied by CompareExchange).
(no idea why would I add it, most likely I was sleepy).
|
FYI I merged PR and then @reyang is going to open a new one to clean up some of the feedback. |
@reyang For comparison I took the code from #2499 (MetricReader task version) and switched it to use
TaskCompletionSource<T>instead ofTask<T>directly. What that does it take away the need to actually use a second thread to do the work while still allowing the same signaling/result storage mechanics. Makes the performance good like #2505 (wait handle version) but maybe a bit easier to read/maintain?