Skip to content

Complete NIOCore strict concurrency #2959

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.

Sign up for GitHub

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

Merged
merged 2 commits into from
Oct 31, 2024
Merged

Complete NIOCore strict concurrency #2959

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cb7d29b
Complete NIOCore strict concurrency
Lukasa Oct 31, 2024
2c8d72c
Formatting
Lukasa Oct 31, 2024
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
7 changes: 5 additions & 2 deletions Package.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,8 @@ let package = Package(
"_NIODataStructures",
swiftCollections,
swiftAtomics,
]
],
swiftSettings: strictConcurrencySettings
),
.target(
name: "_NIODataStructures",
Expand Down Expand Up @@ -414,11 +415,13 @@ let package = Package(
.testTarget(
name: "NIOCoreTests",
dependencies: [
"NIOConcurrencyHelpers",
"NIOCore",
"NIOEmbedded",
"NIOFoundationCompat",
swiftAtomics,
]
],
swiftSettings: strictConcurrencySettings
),
.testTarget(
name: "NIOEmbeddedTests",
Expand Down
2 changes: 1 addition & 1 deletion Sources/NIOCore/BSDSocketAPI.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ let SO_TIMESTAMP = CNIOLinux_SO_TIMESTAMP
let SO_RCVTIMEO = CNIOLinux_SO_RCVTIMEO
#endif

public enum NIOBSDSocket {
public enum NIOBSDSocket: Sendable {
#if os(Windows)
public typealias Handle = SOCKET
#else
Expand Down
6 changes: 3 additions & 3 deletions Tests/NIOCoreTests/AsyncChannel/AsyncChannelTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,8 @@ final class AsyncChannelTests: XCTestCase {
let strongSentinel: Sentinel? = Sentinel()
sentinel = strongSentinel!
try await XCTAsyncAssertNotNil(
await channel.pipeline.handler(type: NIOAsyncChannelHandler<Sentinel, Sentinel, Never>.self).map { _ in
await channel.pipeline.handler(type: NIOAsyncChannelHandler<Sentinel, Sentinel, Never>.self).map {
_ -> Bool in
true
}.get()
)
Expand Down Expand Up @@ -428,9 +429,8 @@ private final class CloseRecorder: ChannelOutboundHandler, @unchecked Sendable {
}
}

private final class CloseSuppressor: ChannelOutboundHandler, RemovableChannelHandler {
private final class CloseSuppressor: ChannelOutboundHandler, RemovableChannelHandler, Sendable {
typealias OutboundIn = Any
typealias outbound = Any

func close(context: ChannelHandlerContext, mode: CloseMode, promise: EventLoopPromise<Void>?) {
// We drop the close here.
Expand Down
4 changes: 2 additions & 2 deletions Tests/NIOCoreTests/AsyncSequences/NIOAsyncSequenceTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -519,7 +519,7 @@ final class NIOAsyncSequenceProducerTests: XCTestCase {
let value = await iterator.next()
resumed.fulfill()

await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
return value
}

Expand Down Expand Up @@ -562,7 +562,7 @@ final class NIOAsyncSequenceProducerTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task: Task<Int?, Never> = Task {
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
let iterator = sequence.makeAsyncIterator()
return await iterator.next()
}
Expand Down
13 changes: 7 additions & 6 deletions Tests/NIOCoreTests/AsyncSequences/NIOAsyncWriterTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,8 @@ final class NIOAsyncWriterTests: XCTestCase {
override func setUp() {
super.setUp()

self.delegate = .init()
let delegate = MockAsyncWriterDelegate()
self.delegate = delegate
let newWriter = NIOAsyncWriter.makeWriter(
elementType: String.self,
isWritable: true,
Expand All @@ -78,7 +79,7 @@ final class NIOAsyncWriterTests: XCTestCase {
)
self.writer = newWriter.writer
self.sink = newWriter.sink
self.sink._storage._setDidSuspend { self.delegate.didSuspend() }
self.sink._storage._setDidSuspend { delegate.didSuspend() }
}

override func tearDown() {
Expand Down Expand Up @@ -411,7 +412,7 @@ final class NIOAsyncWriterTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task = Task { [writer] in
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
try await writer!.yield("message2")
}

Expand Down Expand Up @@ -470,7 +471,7 @@ final class NIOAsyncWriterTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task = Task { [writer] in
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
try await writer!.yield("message1")
}

Expand All @@ -491,7 +492,7 @@ final class NIOAsyncWriterTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task = Task { [writer] in
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
try await writer!.yield("message2")
}

Expand Down Expand Up @@ -545,7 +546,7 @@ final class NIOAsyncWriterTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task = Task { [writer] in
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
try await writer!.yield("message1")
}

Expand Down
13 changes: 7 additions & 6 deletions Tests/NIOCoreTests/AsyncSequences/NIOThrowingAsyncSequenceTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -611,7 +611,7 @@ final class NIOThrowingAsyncSequenceProducerTests: XCTestCase {
let iterator = sequence.makeAsyncIterator()
let element = try await iterator.next()
resumed.fulfill()
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
return element
}

Expand Down Expand Up @@ -655,7 +655,7 @@ final class NIOThrowingAsyncSequenceProducerTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task: Task<Int?, Error> = Task {
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
let iterator = sequence.makeAsyncIterator()
return try await iterator.next()
}
Expand Down Expand Up @@ -686,7 +686,7 @@ final class NIOThrowingAsyncSequenceProducerTests: XCTestCase {
let cancelled = expectation(description: "task cancelled")

let task: Task<Int?, Error> = Task {
await fulfillment(of: [cancelled], timeout: 1)
await XCTWaiter().fulfillment(of: [cancelled], timeout: 1)
let iterator = sequence.makeAsyncIterator()
return try await iterator.next()
}
Expand Down Expand Up @@ -879,12 +879,13 @@ final class NIOThrowingAsyncSequenceProducerTests: XCTestCase {

func testIteratorThrows_whenCancelled() async {
_ = self.source.yield(contentsOf: Array(1...100))
guard let sequence = self.sequence else {
return XCTFail("Expected to have an AsyncSequence")
}

await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
var itemsYieldedCounter = 0
guard let sequence = self.sequence else {
return XCTFail("Expected to have an AsyncSequence")
}

do {
for try await next in sequence {
Expand Down
43 changes: 26 additions & 17 deletions Tests/NIOCoreTests/ByteBufferTest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
//
//===----------------------------------------------------------------------===//

import Atomics
import NIOFoundationCompat
import XCTest
import _NIOBase64
Expand Down Expand Up @@ -2440,8 +2441,8 @@ class ByteBufferTest: XCTestCase {
}

func testReserveCapacityLargerUniquelyReferencedCallsRealloc() throws {
testReserveCapacityLarger_reallocCount = 0
testReserveCapacityLarger_mallocCount = 0
testReserveCapacityLarger_reallocCount.store(0, ordering: .sequentiallyConsistent)
testReserveCapacityLarger_mallocCount.store(0, ordering: .sequentiallyConsistent)

let alloc = ByteBufferAllocator(
hookedMalloc: testReserveCapacityLarger_mallocHook,
Expand All @@ -2453,17 +2454,17 @@ class ByteBufferTest: XCTestCase {

let oldCapacity = buf.capacity

XCTAssertEqual(testReserveCapacityLarger_mallocCount, 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount, 0)
XCTAssertEqual(testReserveCapacityLarger_mallocCount.load(ordering: .sequentiallyConsistent), 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount.load(ordering: .sequentiallyConsistent), 0)
buf.reserveCapacity(32)
XCTAssertEqual(testReserveCapacityLarger_mallocCount, 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount, 1)
XCTAssertEqual(testReserveCapacityLarger_mallocCount.load(ordering: .sequentiallyConsistent), 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount.load(ordering: .sequentiallyConsistent), 1)
XCTAssertNotEqual(buf.capacity, oldCapacity)
}

func testReserveCapacityLargerMultipleReferenceCallsMalloc() throws {
testReserveCapacityLarger_reallocCount = 0
testReserveCapacityLarger_mallocCount = 0
testReserveCapacityLarger_reallocCount.store(0, ordering: .sequentiallyConsistent)
testReserveCapacityLarger_mallocCount.store(0, ordering: .sequentiallyConsistent)

let alloc = ByteBufferAllocator(
hookedMalloc: testReserveCapacityLarger_mallocHook,
Expand All @@ -2480,11 +2481,11 @@ class ByteBufferTest: XCTestCase {
UInt(bitPattern: $0.baseAddress!)
}

XCTAssertEqual(testReserveCapacityLarger_mallocCount, 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount, 0)
XCTAssertEqual(testReserveCapacityLarger_mallocCount.load(ordering: .sequentiallyConsistent), 1)
XCTAssertEqual(testReserveCapacityLarger_reallocCount.load(ordering: .sequentiallyConsistent), 0)
buf.reserveCapacity(32)
XCTAssertEqual(testReserveCapacityLarger_mallocCount, 2)
XCTAssertEqual(testReserveCapacityLarger_reallocCount, 0)
XCTAssertEqual(testReserveCapacityLarger_mallocCount.load(ordering: .sequentiallyConsistent), 2)
XCTAssertEqual(testReserveCapacityLarger_reallocCount.load(ordering: .sequentiallyConsistent), 0)

let newPtrVal = buf.withVeryUnsafeBytes {
UInt(bitPattern: $0.baseAddress!)
Expand Down Expand Up @@ -3354,7 +3355,15 @@ private enum AllocationExpectationState: Int {
case freeDone
}

private var testAllocationOfReallyBigByteBuffer_state = AllocationExpectationState.begin
private let _testAllocationOfReallyBigByteBuffer_state = ManagedAtomic<Int>(AllocationExpectationState.begin.rawValue)
private var testAllocationOfReallyBigByteBuffer_state: AllocationExpectationState {
get {
.init(rawValue: _testAllocationOfReallyBigByteBuffer_state.load(ordering: .acquiring))!
}
set {
_testAllocationOfReallyBigByteBuffer_state.store(newValue.rawValue, ordering: .releasing)
}
}
private func testAllocationOfReallyBigByteBuffer_freeHook(_ ptr: UnsafeMutableRawPointer?) {
precondition(AllocationExpectationState.reallocDone == testAllocationOfReallyBigByteBuffer_state)
testAllocationOfReallyBigByteBuffer_state = .freeDone
Expand Down Expand Up @@ -3387,22 +3396,22 @@ private func testAllocationOfReallyBigByteBuffer_memcpyHook(
// not actually doing any copies
}

private var testReserveCapacityLarger_reallocCount = 0
private var testReserveCapacityLarger_mallocCount = 0
private let testReserveCapacityLarger_reallocCount = ManagedAtomic(0)
private let testReserveCapacityLarger_mallocCount = ManagedAtomic(0)
private func testReserveCapacityLarger_freeHook(_ ptr: UnsafeMutableRawPointer) {
free(ptr)
}

private func testReserveCapacityLarger_mallocHook(_ size: Int) -> UnsafeMutableRawPointer? {
testReserveCapacityLarger_mallocCount += 1
testReserveCapacityLarger_mallocCount.wrappingIncrement(ordering: .sequentiallyConsistent)
return malloc(size)
}

private func testReserveCapacityLarger_reallocHook(
_ ptr: UnsafeMutableRawPointer?,
_ count: Int
) -> UnsafeMutableRawPointer? {
testReserveCapacityLarger_reallocCount += 1
testReserveCapacityLarger_reallocCount.wrappingIncrement(ordering: .sequentiallyConsistent)
return realloc(ptr, count)
}

Expand Down
13 changes: 11 additions & 2 deletions Tests/NIOCoreTests/ChannelOptionStorageTest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
//
//===----------------------------------------------------------------------===//

import NIOConcurrencyHelpers
import NIOCore
import NIOEmbedded
import XCTest
Expand Down Expand Up @@ -110,8 +111,16 @@ class ChannelOptionStorageTest: XCTestCase {
}
}

class OptionsCollectingChannel: Channel {
var allOptions: [(Any, Any)] = []
final class OptionsCollectingChannel: Channel {
private let _allOptions = NIOLockedValueBox<[(any Sendable, any Sendable)]>([])
var allOptions: [(any Sendable, any Sendable)] {
get {
self._allOptions.withLockedValue { $0 }
}
set {
self._allOptions.withLockedValue { $0 = newValue }
}
}

var allocator: ByteBufferAllocator { fatalError() }

Expand Down
13 changes: 7 additions & 6 deletions Tests/NIOCoreTests/DispatchQueue+WithFutureTest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
//
//===----------------------------------------------------------------------===//

import Atomics
import Dispatch
import NIOCore
import NIOEmbedded
Expand All @@ -30,20 +31,20 @@ class DispatchQueueWithFutureTest: XCTestCase {
}
let eventLoop = group.next()
let sem = DispatchSemaphore(value: 0)
var nonBlockingRan = false
let nonBlockingRan = ManagedAtomic(false)
let futureResult: EventLoopFuture<String> = DispatchQueue.global().asyncWithFuture(eventLoop: eventLoop) {
() -> String in
sem.wait() // Block in callback
return "hello"
}
futureResult.whenSuccess { value in
XCTAssertEqual(value, "hello")
XCTAssertTrue(nonBlockingRan)
XCTAssertTrue(nonBlockingRan.load(ordering: .sequentiallyConsistent))
}

let p2 = eventLoop.makePromise(of: Bool.self)
p2.futureResult.whenSuccess { _ in
nonBlockingRan = true
nonBlockingRan.store(true, ordering: .sequentiallyConsistent)
}
p2.succeed(true)

Expand All @@ -57,20 +58,20 @@ class DispatchQueueWithFutureTest: XCTestCase {
}
let eventLoop = group.next()
let sem = DispatchSemaphore(value: 0)
var nonBlockingRan = false
let nonBlockingRan = ManagedAtomic(false)
let futureResult: EventLoopFuture<String> = DispatchQueue.global().asyncWithFuture(eventLoop: eventLoop) {
() -> String in
sem.wait() // Block in callback
throw DispatchQueueTestError.example
}
futureResult.whenFailure { err in
XCTAssertEqual(err as! DispatchQueueTestError, DispatchQueueTestError.example)
XCTAssertTrue(nonBlockingRan)
XCTAssertTrue(nonBlockingRan.load(ordering: .sequentiallyConsistent))
}

let p2 = eventLoop.makePromise(of: Bool.self)
p2.futureResult.whenSuccess { _ in
nonBlockingRan = true
nonBlockingRan.store(true, ordering: .sequentiallyConsistent)
}
p2.succeed(true)

Expand Down
2 changes: 1 addition & 1 deletion Tests/NIOCoreTests/NIOCloseOnErrorHandlerTest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import NIOCore
import NIOEmbedded
import XCTest

final class DummyFailingHandler1: ChannelInboundHandler {
final class DummyFailingHandler1: ChannelInboundHandler, Sendable {
typealias InboundIn = NIOAny

struct DummyError1: Error {}
Expand Down
2 changes: 1 addition & 1 deletion Tests/NIOCoreTests/XCTest+Extensions.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ func assertNoThrowWithValue<T>(

func withTemporaryFile<T>(content: String? = nil, _ body: (NIOCore.NIOFileHandle, String) throws -> T) throws -> T {
let temporaryFilePath = "\(temporaryDirectory)/nio_\(UUID())"
FileManager.default.createFile(atPath: temporaryFilePath, contents: content?.data(using: .utf8))
_ = FileManager.default.createFile(atPath: temporaryFilePath, contents: content?.data(using: .utf8))
defer {
XCTAssertNoThrow(try FileManager.default.removeItem(atPath: temporaryFilePath))
}
Expand Down
Loading