Skip to content
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

Merged
merged 5 commits into from
Aug 16, 2020
Merged
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 135 additions & 0 deletions src/OpenTelemetry/Internal/CircularBuffer.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
// <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)
Copy link
Contributor

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?

Copy link
Member Author

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?

Copy link
Contributor

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 (for ceiling(2**size)). This could be a low priority optimization and needs data to prove.

{
if (capacity <= 0)
{
throw new ArgumentOutOfRangeException(nameof(capacity));
}

this.capacity = capacity;
this.trait = new T[capacity];
}

public int Capacity
{
get
{
return this.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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Retry here introduces uncertainty of running time of TryAdd, perhaps return false instead of retry in the while(true) loop, or limit retry count?

Copy link
Member Author

Choose a reason for hiding this comment

The 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.

Copy link
Member Author

Choose a reason for hiding this comment

The 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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is headSnapshot validation needed, or Interloaded.Add can work here?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's is required, otherwise it'll break the 🧙‍♀️.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, yes, it is required because the item pointed by headSnapshot maybe not in the range of [this.tail, this.head).

{
return true;
}

this.trait[index] = null;
}
}
}

public IEnumerable<T> Consume(int count)
{
if (count <= 0)
{
yield break;
Copy link
Contributor

Choose a reason for hiding this comment

The 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?

Copy link
Member Author

Choose a reason for hiding this comment

The 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++)

Choose a reason for hiding this comment

The 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);
}
}
}