-
Notifications
You must be signed in to change notification settings - Fork 780
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Refactor exporter - step 4 #1085
Changes from 3 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,129 @@ | ||
// <copyright file="CircularBuffer.cs" company="OpenTelemetry Authors"> | ||
// Copyright The OpenTelemetry Authors | ||
// | ||
// Licensed under the Apache License, Version 2.0 (the "License"); | ||
// you may not use this file except in compliance with the License. | ||
// You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, software | ||
// distributed under the License is distributed on an "AS IS" BASIS, | ||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
// See the License for the specific language governing permissions and | ||
// limitations under the License. | ||
// </copyright> | ||
|
||
using System; | ||
using System.Collections.Generic; | ||
using System.Runtime.CompilerServices; | ||
using System.Threading; | ||
|
||
namespace OpenTelemetry.Internal | ||
{ | ||
/// <summary> | ||
/// Lock free implementation of single-reader multi-writer circular buffer. | ||
/// </summary> | ||
/// <typeparam name="T">The type of the underlying value.</typeparam> | ||
internal class CircularBuffer<T> | ||
where T : class | ||
{ | ||
private readonly int capacity; | ||
private readonly T[] trait; | ||
private long head = 0; | ||
private long tail = 0; | ||
|
||
public CircularBuffer(int capacity) | ||
{ | ||
if (capacity <= 0) | ||
{ | ||
throw new ArgumentOutOfRangeException(nameof(capacity)); | ||
} | ||
|
||
this.capacity = capacity; | ||
this.trait = new T[capacity]; | ||
} | ||
|
||
/// <summary> | ||
/// Gets the number of items contained in the <see cref="CircularBuffer{T}"/>. | ||
/// </summary> | ||
public int Count | ||
{ | ||
get | ||
{ | ||
var tailSnapshot = this.tail; | ||
return (int)(this.head - tailSnapshot); | ||
} | ||
} | ||
|
||
/// <summary> | ||
/// Attempts to add the specified item to the buffer. | ||
/// </summary> | ||
/// <param name="value">The value to add.</param> | ||
/// <returns>Returns true if the item was added to the buffer successfully; false if the buffer is full.</returns> | ||
public bool TryAdd(T value) | ||
{ | ||
if (value == null) | ||
{ | ||
throw new ArgumentNullException(nameof(value)); | ||
} | ||
|
||
while (true) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Retry here introduces uncertainty of running time of There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The intention of this function is to return false only when queue is full. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @ThomsonTan you've actually worried me! 😅 Here goes a PR #1088 to address it. |
||
{ | ||
var tailSnapshot = this.tail; | ||
var headSnapshot = this.head; | ||
|
||
if (headSnapshot - tailSnapshot >= this.capacity) | ||
{ | ||
return false; // buffer is full | ||
} | ||
|
||
var index = (int)(headSnapshot % this.capacity); | ||
|
||
if (this.SwapIfNull(index, value)) | ||
{ | ||
if (Interlocked.CompareExchange(ref this.head, headSnapshot + 1, headSnapshot) == headSnapshot) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It's is required, otherwise it'll break the 🧙♀️. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I see, yes, it is required because the item pointed by |
||
{ | ||
return true; | ||
} | ||
else | ||
{ | ||
this.trait[index] = null; | ||
} | ||
} | ||
} | ||
} | ||
|
||
public IEnumerable<T> Consume(int count) | ||
{ | ||
if (count <= 0) | ||
{ | ||
yield break; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. When the queue is empty, wait for notification of new items instead of polling? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. That would make it a normal lock based queue instead of lock-free. |
||
} | ||
|
||
count = Math.Min(count, this.Count); | ||
|
||
for (int i = 0; i < count; i++) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do you want to provide an additional check with a boolean and an Interlock to prevent 2 threads from accidently entering if the single reader pattern is accidently broken. |
||
{ | ||
var index = (int)(this.tail % this.capacity); | ||
var value = this.trait[index]; | ||
this.trait[index] = null; | ||
this.tail++; | ||
yield return value; | ||
} | ||
} | ||
|
||
[MethodImpl(MethodImplOptions.AggressiveInlining)] | ||
private bool CompareAndSwap(int index, T value, T comparand) | ||
{ | ||
var result = Interlocked.CompareExchange(ref this.trait[index], value, comparand); | ||
return object.ReferenceEquals(result, comparand); | ||
} | ||
|
||
[MethodImpl(MethodImplOptions.AggressiveInlining)] | ||
private bool SwapIfNull(int index, T value) | ||
{ | ||
return this.CompareAndSwap(index, value, null); | ||
} | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Restrict capacity to be power of 2, so the modulo operation
%
could replaced with a simple bitwise and?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How much (X%) perf improvement would you expect from modulo?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
div
could take tens times of CPU cycles than bitwise arithmetic instruction, but there is also trade off taking more memory (forceiling(2**size)
). This could be a low priority optimization and needs data to prove.