Skip to content

Commit

Permalink
Provide better concurrency modes
Browse files Browse the repository at this point in the history
  • Loading branch information
reyang committed May 22, 2024
1 parent 3cff6db commit 72a2553
Show file tree
Hide file tree
Showing 5 changed files with 100 additions and 0 deletions.
1 change: 1 addition & 0 deletions docs/logs/extending-the-sdk/MyExporter.cs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
using OpenTelemetry;
using OpenTelemetry.Logs;

[ConcurrencyModes(ConcurrencyModes.Multithreaded | ConcurrencyModes.Reentrant)]
internal class MyExporter : BaseExporter<LogRecord>
{
private readonly string name;
Expand Down
7 changes: 7 additions & 0 deletions src/OpenTelemetry/.publicApi/Stable/PublicAPI.Unshipped.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,10 @@
OpenTelemetry.ConcurrencyModes
OpenTelemetry.ConcurrencyModes.Global = 128 -> OpenTelemetry.ConcurrencyModes
OpenTelemetry.ConcurrencyModes.Multithreaded = 2 -> OpenTelemetry.ConcurrencyModes
OpenTelemetry.ConcurrencyModes.Reentrant = 1 -> OpenTelemetry.ConcurrencyModes
OpenTelemetry.ConcurrencyModesAttribute
OpenTelemetry.ConcurrencyModesAttribute.ConcurrencyModesAttribute(OpenTelemetry.ConcurrencyModes supported) -> void
OpenTelemetry.ConcurrencyModesAttribute.Supported.get -> OpenTelemetry.ConcurrencyModes
OpenTelemetry.Metrics.Exemplar
OpenTelemetry.Metrics.Exemplar.DoubleValue.get -> double
OpenTelemetry.Metrics.Exemplar.Exemplar() -> void
Expand Down
43 changes: 43 additions & 0 deletions src/OpenTelemetry/ConcurrencyModes.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

namespace OpenTelemetry;

/// <summary>
/// Describes the concurrency mode of an OpenTelemetry extension.
/// </summary>
[Flags]
public enum ConcurrencyModes : byte
{
/*
0 0 0 0 0 0 0 0
| | | | | | | |
| | | | | | | +--- Reentrant
| | | | | | +----- Multithreaded
| | | | | +------- (reserved)
| | | | +--------- (reserved)
| | | +----------- (reserved)
| | +------------- (reserved)
| +--------------- (reserved)
+----------------- Global
*/

/// <summary>
/// Reentrant, the component can be invoked recursively without resulting
/// a deadlock or infinite loop.
/// </summary>
Reentrant = 0b1,

/// <summary>
/// Multithreaded, the component can be invoked concurrently across
/// multiple threads.
/// </summary>
Multithreaded = 0b10,

/// <summary>
/// Global, when combined with other flags, indicates that a per-instance
/// synchronization is insufficient, a global synchronization is required
/// across all instances of the component.
/// </summary>
Global = 0b10000000,
}
27 changes: 27 additions & 0 deletions src/OpenTelemetry/ConcurrencyModesAttribute.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

namespace OpenTelemetry;

/// <summary>
/// An attribute for declaring the supported <see cref="ConcurrencyModes"/> of an OpenTelemetry extension.
/// </summary>
[AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = true)]
public sealed class ConcurrencyModesAttribute : Attribute
{
private readonly ConcurrencyModes supportedConcurrencyModes;

/// <summary>
/// Initializes a new instance of the <see cref="ConcurrencyModesAttribute"/> class.
/// </summary>
/// <param name="supported"><see cref="ConcurrencyModes"/>.</param>
public ConcurrencyModesAttribute(ConcurrencyModes supported)
{
this.supportedConcurrencyModes = supported;
}

/// <summary>
/// Gets the supported <see cref="ConcurrencyModes"/>.
/// </summary>
public ConcurrencyModes Supported => this.supportedConcurrencyModes;
}
22 changes: 22 additions & 0 deletions src/OpenTelemetry/SimpleExportProcessor.cs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ public abstract class SimpleExportProcessor<T> : BaseExportProcessor<T>
where T : class
{
private readonly object syncObject = new();
private readonly ConcurrencyModes supportedConcurrencyModes;

/// <summary>
/// Initializes a new instance of the <see cref="SimpleExportProcessor{T}"/> class.
Expand All @@ -21,11 +22,32 @@ public abstract class SimpleExportProcessor<T> : BaseExportProcessor<T>
protected SimpleExportProcessor(BaseExporter<T> exporter)
: base(exporter)
{
var exporterType = exporter.GetType();
var attributes = exporterType.GetCustomAttributes(typeof(ConcurrencyModesAttribute), true);
if (attributes.Length > 0)
{
var attr = (ConcurrencyModesAttribute)attributes[attributes.Length - 1];
this.supportedConcurrencyModes = attr.Supported;
}
}

/// <inheritdoc />
protected override void OnExport(T data)
{
if (this.supportedConcurrencyModes.HasFlag(ConcurrencyModes.Multithreaded))
{
try
{
this.exporter.Export(new Batch<T>(data));
}
catch (Exception ex)
{
OpenTelemetrySdkEventSource.Log.SpanProcessorException(nameof(this.OnExport), ex);
}

return;
}

lock (this.syncObject)
{
try
Expand Down

0 comments on commit 72a2553

Please sign in to comment.