Skip to content

Commit

Permalink
Use Lock instead of DispatchQueue
Browse files Browse the repository at this point in the history
  • Loading branch information
justinhporter committed Oct 26, 2023
1 parent 409f18b commit 33e1771
Show file tree
Hide file tree
Showing 4 changed files with 216 additions and 21 deletions.
198 changes: 198 additions & 0 deletions Sources/Exporters/OpenTelemetryProtocolHttp/Internal/Locks.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,198 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

//===----------------------------------------------------------------------===//
//
// This source file is part of the Swift Metrics API open source project
//
// Copyright (c) 2018-2019 Apple Inc. and the Swift Metrics API project authors
// Licensed under Apache License v2.0
//
// See LICENSE.txt for license information
// See CONTRIBUTORS.txt for the list of Swift Metrics API project authors
//
// SPDX-License-Identifier: Apache-2.0
//
//===----------------------------------------------------------------------===//

//===----------------------------------------------------------------------===//
//
// This source file is part of the SwiftNIO open source project
//
// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
// Licensed under Apache License v2.0
//
// See LICENSE.txt for license information
// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
//
// SPDX-License-Identifier: Apache-2.0
//
//===----------------------------------------------------------------------===//

#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS)
import Darwin
#else
import Glibc
#endif

/// A threading lock based on `libpthread` instead of `libdispatch`.
///
/// This object provides a lock on top of a single `pthread_mutex_t`. This kind
/// of lock is safe to use with `libpthread`-based threading models, such as the
/// one used by NIO.
internal final class Lock {
fileprivate let mutex: UnsafeMutablePointer<pthread_mutex_t> = UnsafeMutablePointer.allocate(capacity: 1)

/// Create a new lock.
public init() {
let err = pthread_mutex_init(self.mutex, nil)
precondition(err == 0, "pthread_mutex_init failed with error \(err)")
}

deinit {
let err = pthread_mutex_destroy(self.mutex)
precondition(err == 0, "pthread_mutex_destroy failed with error \(err)")
self.mutex.deallocate()
}

/// Acquire the lock.
///
/// Whenever possible, consider using `withLock` instead of this method and
/// `unlock`, to simplify lock handling.
public func lock() {
let err = pthread_mutex_lock(self.mutex)
precondition(err == 0, "pthread_mutex_lock failed with error \(err)")
}

/// Release the lock.
///
/// Whenever possible, consider using `withLock` instead of this method and
/// `lock`, to simplify lock handling.
public func unlock() {
let err = pthread_mutex_unlock(self.mutex)
precondition(err == 0, "pthread_mutex_unlock failed with error \(err)")
}
}

extension Lock {
/// Acquire the lock for the duration of the given block.
///
/// This convenience method should be preferred to `lock` and `unlock` in
/// most situations, as it ensures that the lock will be released regardless
/// of how `body` exits.
///
/// - Parameter body: The block to execute while holding the lock.
/// - Returns: The value returned by the block.
@inlinable
internal func withLock<T>(_ body: () throws -> T) rethrows -> T {
self.lock()
defer {
self.unlock()
}
return try body()
}

// specialise Void return (for performance)
@inlinable
internal func withLockVoid(_ body: () throws -> Void) rethrows {
try self.withLock(body)
}
}

/// A threading lock based on `libpthread` instead of `libdispatch`.
///
/// This object provides a lock on top of a single `pthread_mutex_t`. This kind
/// of lock is safe to use with `libpthread`-based threading models, such as the
/// one used by NIO.
internal final class ReadWriteLock {
fileprivate let rwlock: UnsafeMutablePointer<pthread_rwlock_t> = UnsafeMutablePointer.allocate(capacity: 1)

/// Create a new lock.
public init() {
let err = pthread_rwlock_init(self.rwlock, nil)
precondition(err == 0, "pthread_rwlock_init failed with error \(err)")
}

deinit {
let err = pthread_rwlock_destroy(self.rwlock)
precondition(err == 0, "pthread_rwlock_destroy failed with error \(err)")
self.rwlock.deallocate()
}

/// Acquire a reader lock.
///
/// Whenever possible, consider using `withLock` instead of this method and
/// `unlock`, to simplify lock handling.
public func lockRead() {
let err = pthread_rwlock_rdlock(self.rwlock)
precondition(err == 0, "pthread_rwlock_rdlock failed with error \(err)")
}

/// Acquire a writer lock.
///
/// Whenever possible, consider using `withLock` instead of this method and
/// `unlock`, to simplify lock handling.
public func lockWrite() {
let err = pthread_rwlock_wrlock(self.rwlock)
precondition(err == 0, "pthread_rwlock_wrlock failed with error \(err)")
}

/// Release the lock.
///
/// Whenever possible, consider using `withLock` instead of this method and
/// `lock`, to simplify lock handling.
public func unlock() {
let err = pthread_rwlock_unlock(self.rwlock)
precondition(err == 0, "pthread_rwlock_unlock failed with error \(err)")
}
}

extension ReadWriteLock {
/// Acquire the reader lock for the duration of the given block.
///
/// This convenience method should be preferred to `lock` and `unlock` in
/// most situations, as it ensures that the lock will be released regardless
/// of how `body` exits.
///
/// - Parameter body: The block to execute while holding the lock.
/// - Returns: The value returned by the block.
@inlinable
internal func withReaderLock<T>(_ body: () throws -> T) rethrows -> T {
self.lockRead()
defer {
self.unlock()
}
return try body()
}

/// Acquire the writer lock for the duration of the given block.
///
/// This convenience method should be preferred to `lock` and `unlock` in
/// most situations, as it ensures that the lock will be released regardless
/// of how `body` exits.
///
/// - Parameter body: The block to execute while holding the lock.
/// - Returns: The value returned by the block.
@inlinable
internal func withWriterLock<T>(_ body: () throws -> T) rethrows -> T {
self.lockWrite()
defer {
self.unlock()
}
return try body()
}

// specialise Void return (for performance)
@inlinable
internal func withReaderLockVoid(_ body: () throws -> Void) rethrows {
try self.withReaderLock(body)
}

// specialise Void return (for performance)
@inlinable
internal func withWriterLockVoid(_ body: () throws -> Void) rethrows {
try self.withWriterLock(body)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ public func defaultOltpHttpLoggingEndpoint() -> URL {
public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {

var pendingLogRecords: [ReadableLogRecord] = []
let dispatchQueue = DispatchQueue(label: "OtlpHttpLogExporter Queue")
private let exporterLock = Lock()

override public init(endpoint: URL = defaultOltpHttpLoggingEndpoint(),
config: OtlpConfiguration = OtlpConfiguration(),
Expand All @@ -24,8 +24,8 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
}

public func export(logRecords: [OpenTelemetrySdk.ReadableLogRecord], explicitTimeout: TimeInterval? = nil) -> OpenTelemetrySdk.ExportResult {
var sendingLogRecords: [ReadableLogRecord]!
dispatchQueue.sync {
var sendingLogRecords: [ReadableLogRecord] = []
exporterLock.withLockVoid {
pendingLogRecords.append(contentsOf: logRecords)
sendingLogRecords = pendingLogRecords
pendingLogRecords = []
Expand All @@ -37,13 +37,12 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {

var request = createRequest(body: body, endpoint: endpoint)
request.timeoutInterval = min(explicitTimeout ?? TimeInterval.greatestFiniteMagnitude , config.timeout)
httpClient.send(request: request) { [weak self] result in
guard let self = self else { return }
httpClient.send(request: request) { [weak self, exporterLock] result in
switch result {
case .success(_):
break
case .failure(let error):
self.dispatchQueue.sync { [weak self] in
exporterLock.withLockVoid {
self?.pendingLogRecords.append(contentsOf: sendingLogRecords)
}
print(error)
Expand All @@ -60,7 +59,7 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
public func flush(explicitTimeout: TimeInterval? = nil) -> ExportResult {
var exporterResult: ExportResult = .success
var pendingLogRecords: [ReadableLogRecord]!
dispatchQueue.sync {
exporterLock.withLockVoid {
pendingLogRecords = self.pendingLogRecords
}
if !pendingLogRecords.isEmpty {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,16 @@ public func defaultOltpHTTPMetricsEndpoint() -> URL {

public class OtlpHttpMetricExporter: OtlpHttpExporterBase, MetricExporter {
var pendingMetrics: [Metric] = []
let dispatchQueue = DispatchQueue(label: "OtlpHttpMetricExporter Queue")
private let exporterLock = Lock()

override
public init(endpoint: URL = defaultOltpHTTPMetricsEndpoint(), config : OtlpConfiguration = OtlpConfiguration(), useSession: URLSession? = nil, envVarHeaders: [(String,String)]? = EnvVarHeaders.attributes) {
super.init(endpoint: endpoint, config: config, useSession: useSession, envVarHeaders: envVarHeaders)
}

public func export(metrics: [Metric], shouldCancel: (() -> Bool)?) -> MetricExporterResultCode {
var sendingMetrics: [Metric]!
dispatchQueue.sync {
var sendingMetrics: [Metric] = []
exporterLock.withLockVoid {
pendingMetrics.append(contentsOf: metrics)
sendingMetrics = pendingMetrics
pendingMetrics = []
Expand All @@ -33,13 +33,12 @@ public class OtlpHttpMetricExporter: OtlpHttpExporterBase, MetricExporter {
}

let request = createRequest(body: body, endpoint: endpoint)
httpClient.send(request: request) { [weak self] result in
guard let self = self else { return }
httpClient.send(request: request) { [weak self, exporterLock] result in
switch result {
case .success(_):
break
case .failure(let error):
self.dispatchQueue.sync { [weak self] in
exporterLock.withLockVoid {
self?.pendingMetrics.append(contentsOf: sendingMetrics)
}
print(error)
Expand All @@ -52,7 +51,7 @@ public class OtlpHttpMetricExporter: OtlpHttpExporterBase, MetricExporter {
public func flush() -> MetricExporterResultCode {
var exporterResult: MetricExporterResultCode = .success
var pendingMetrics: [Metric]!
dispatchQueue.sync {
exporterLock.withLockVoid {
pendingMetrics = self.pendingMetrics
}
if !pendingMetrics.isEmpty {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {


var pendingSpans: [SpanData] = []
let dispatchQueue = DispatchQueue(label: "OtlpHttpTraceExporter Queue")
private let exporterLock = Lock()

override
public init(endpoint: URL = defaultOltpHttpTracesEndpoint(), config: OtlpConfiguration = OtlpConfiguration(),
Expand All @@ -24,8 +24,8 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
}

public func export(spans: [SpanData], explicitTimeout: TimeInterval? = nil) -> SpanExporterResultCode {
var sendingSpans: [SpanData]!
dispatchQueue.sync {
var sendingSpans: [SpanData] = []
exporterLock.withLockVoid {
pendingSpans.append(contentsOf: spans)
sendingSpans = pendingSpans
pendingSpans = []
Expand All @@ -45,13 +45,12 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
request.addValue(value, forHTTPHeaderField: key)
}
}
httpClient.send(request: request) { [weak self] result in
guard let self = self else { return }
httpClient.send(request: request) { [weak self, exporterLock] result in
switch result {
case .success:
break
case .failure(let error):
self.dispatchQueue.sync { [weak self] in
exporterLock.withLockVoid {
self?.pendingSpans.append(contentsOf: sendingSpans)
}
print(error)
Expand All @@ -63,7 +62,7 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
public func flush(explicitTimeout: TimeInterval? = nil) -> SpanExporterResultCode {
var resultValue: SpanExporterResultCode = .success
var pendingSpans: [SpanData]!
dispatchQueue.sync {
exporterLock.withLockVoid {
pendingSpans = self.pendingSpans
}
if !pendingSpans.isEmpty {
Expand Down

0 comments on commit 33e1771

Please sign in to comment.