From 40ef9fa5f07ef0fefbf5a03fe1e35fd5becedc44 Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Fri, 4 Sep 2026 04:13:07 -0700 Subject: [PATCH 1/2] Decreased number of concurrent open file handles by force-reading entire files as quickly as possible --- .../Helpers/AsyncFileReader.swift | 195 ++++++++++++++++++ .../Helpers/Swift6.1+Compatibility.swift | 17 ++ .../Datastore/DatastoreIndexManifest.swift | 12 +- .../Datastore/DatastorePage.swift | 18 +- 4 files changed, 218 insertions(+), 24 deletions(-) create mode 100644 Sources/CodableDatastore/Helpers/AsyncFileReader.swift create mode 100644 Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift diff --git a/Sources/CodableDatastore/Helpers/AsyncFileReader.swift b/Sources/CodableDatastore/Helpers/AsyncFileReader.swift new file mode 100644 index 0000000..16d8bd1 --- /dev/null +++ b/Sources/CodableDatastore/Helpers/AsyncFileReader.swift @@ -0,0 +1,195 @@ +// +// AsyncFileReader.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2026-09-04. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +import Bytes +import Foundation +import QuestionableConcurrency + +/// An asynchronous, low-latency file reader that consumes the file as fast as possible, but allows multiple readers to read it at their own pace. +class AsyncFileReader: @unchecked Sendable { + static let chunkSize = 1024 + + let url: URL + var readerTask: Task! = nil + + var resultsLock = UnfairLock() + + var writeHead: Int = 0 + var byteChunks: [Bytes] = [] + var finalResult: Result? + var continuations: [CheckedContinuation] = [] + + init(contentsOf url: URL) { + self.url = url + self.readerTask = Task(name: "AsyncFileReader") { await self.startReading() } + } + + func startReading() async { + do { + #if canImport(Darwin) + if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) { + for try await byte in url.resourceBytes { + resultsLock.withLock { + let chunkIndex = writeHead/Self.chunkSize + let indexInChunk = writeHead % Self.chunkSize + + /// If there isn't enough space, allocate a new chunk. + if chunkIndex >= byteChunks.count { + byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize)) + } + + /// Write the byte and advance the head. + byteChunks[chunkIndex][indexInChunk] = byte + + writeHead += 1 + + /// If anyone is waiting on us, let them know. + if !continuations.isEmpty { + for continuation in continuations { + continuation.resume() + } + continuations.removeAll(keepingCapacity: true) + } + } + } + } else { + let data = try Data(contentsOf: url) + for byte in data { + resultsLock.withLock { + let chunkIndex = writeHead/Self.chunkSize + let indexInChunk = writeHead % Self.chunkSize + + /// If there isn't enough space, allocate a new chunk. + if chunkIndex >= byteChunks.count { + byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize)) + } + + /// Write the byte and advance the head. + byteChunks[chunkIndex][indexInChunk] = byte + + writeHead += 1 + + /// If anyone is waiting on us, let them know. + if !continuations.isEmpty { + for continuation in continuations { + continuation.resume() + } + continuations.removeAll(keepingCapacity: true) + } + } + } + } + #else + let data = try Data(contentsOf: url) + for byte in data { + resultsLock.withLock { + let chunkIndex = writeHead/Self.chunkSize + let indexInChunk = writeHead % Self.chunkSize + + /// If there isn't enough space, allocate a new chunk. + if chunkIndex >= byteChunks.count { + byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize)) + } + + /// Write the byte and advance the head. + byteChunks[chunkIndex][indexInChunk] = byte + + writeHead += 1 + + /// If anyone is waiting on us, let them know. + if !continuations.isEmpty { + for continuation in continuations { + continuation.resume() + } + continuations.removeAll(keepingCapacity: true) + } + } + } + #endif + + /// Let the reader know the sequence is done. + resultsLock.withLock { + finalResult = .success(()) + /// If anyone is waiting on us, let them know. + if !continuations.isEmpty { + for continuation in continuations { + continuation.resume() + } + continuations.removeAll(keepingCapacity: true) + } + } + } catch { + /// Let the reader know the sequence failed. + resultsLock.withLock { + finalResult = .failure(error) + /// If anyone is waiting on us, let them know. + if !continuations.isEmpty { + for continuation in continuations { + continuation.resume() + } + continuations.removeAll(keepingCapacity: true) + } + } + } + } + + func byte(for readHead: Int) async throws -> Byte? { + /// If we overtake the write head, check if we are done, otherwise suspend until there are more bytes + resultsLock.unsafeLock() + if readHead < writeHead { + resultsLock.unsafeUnlock() + } else { + if let finalResult { + resultsLock.unsafeUnlock() + /// Return nil or throw the error we encountered while reading. + try finalResult.get() + return nil + } + + await withCheckedContinuation { continuation in + continuations.append(continuation) + resultsLock.unsafeUnlock() + } + + /// Check one more time if we are actually at the end and an error was thrown or if we finished. + let (writeHead, finalResult) = resultsLock.withLock { (self.writeHead, self.finalResult) } + if readHead >= writeHead, let finalResult { + try finalResult.get() + return nil + } + } + + let chunkIndex = readHead/Self.chunkSize + let indexInChunk = readHead % Self.chunkSize + return resultsLock.withLock { byteChunks[chunkIndex][indexInChunk] } + } +} + +extension AsyncFileReader: AsyncSequence { + nonisolated func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(fileReader: self) + } + + struct AsyncIterator: AsyncIteratorProtocol { + let fileReader: AsyncFileReader + var readHead: Int = 0 + + mutating func next() async throws -> Byte? { + let readHead = readHead + self.readHead += 1 + return try await fileReader.byte(for: readHead) + } + + mutating func next(isolation actor: isolated (any Actor)?) async throws(any Error) -> Byte? { + let readHead = readHead + self.readHead += 1 + return try await fileReader.byte(for: readHead) + } + } +} diff --git a/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift b/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift new file mode 100644 index 0000000..0d62306 --- /dev/null +++ b/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift @@ -0,0 +1,17 @@ +// +// Swift6.1+Compatibility.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2026-08-31. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +#if compiler(<6.2) +extension Task where Failure == Never { + @discardableResult + init(name: String?, priority: TaskPriority? = nil, operation: sending @escaping @isolated(any) () async -> Success) { + self.init(priority: priority, operation: operation) + } +} +#endif diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift index ef4ba69..779df1c 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift @@ -95,18 +95,10 @@ extension DatastoreIndexManifest { extension DatastoreIndexManifest { init(contentsOf url: URL, id: ID) async throws { -#if canImport(Darwin) - if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) { - try await self.init(sequence: AnyReadableSequence(url.resourceBytes), id: id) - } else { - try await self.init(sequence: AnyReadableSequence(try Data(contentsOf: url)), id: id) - } -#else - try await self.init(sequence: AnyReadableSequence(try Data(contentsOf: url)), id: id) -#endif + try await self.init(sequence: AsyncFileReader(contentsOf: url), id: id) } - init(sequence: AnyReadableSequence, id: ID) async throws { + init(sequence: AsyncFileReader, id: ID) async throws { self.id = id var iterator = sequence.makeAsyncIterator() diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift index 94a4e3f..f35e6a0 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift @@ -92,21 +92,11 @@ extension DiskPersistence.Datastore.Page { // MARK: - Persistence extension DiskPersistence.Datastore.Page { - private var readableSequence: AnyReadableSequence { - get throws { -#if canImport(Darwin) - if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) { - return AnyReadableSequence(pageURL.resourceBytes) - } else { - return AnyReadableSequence(try Data(contentsOf: pageURL)) - } -#else - return AnyReadableSequence(try Data(contentsOf: pageURL)) -#endif - } + private var readableSequence: AsyncFileReader { + AsyncFileReader(contentsOf: pageURL) } - private nonisolated func performRead(sequence: AnyReadableSequence) async throws -> MultiplexedAsyncSequence> { + private nonisolated func performRead(sequence: AsyncFileReader) async throws -> MultiplexedAsyncSequence> { var iterator = sequence.makeBufferedIterator() try await iterator.check(Self.header) @@ -146,7 +136,7 @@ extension DiskPersistence.Datastore.Page { } let readerTask = Task { - try await performRead(sequence: try readableSequence) + try await performRead(sequence: readableSequence) } isPersisted = true blocksReaderTask = readerTask From f1bba8070c81d39cbf33dac67f63d689f9d05f86 Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Mon, 7 Sep 2026 14:09:31 -0700 Subject: [PATCH 2/2] Improved datastore writes by up to 24% by writing new files non-atomically --- .../Persistence/Disk Persistence/Datastore/DatastoreIndex.swift | 2 +- .../Persistence/Disk Persistence/Datastore/DatastorePage.swift | 2 +- .../Persistence/Disk Persistence/Datastore/DatastoreRoot.swift | 2 +- .../Persistence/Disk Persistence/Snapshot/Snapshot.swift | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndex.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndex.swift index 7ff575e..127622d 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndex.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndex.swift @@ -153,7 +153,7 @@ extension DiskPersistence.Datastore.Index { /// Encode the provided manifest, and write it to disk. let data = Data(manifest.bytes) - try data.write(to: manifestURL, options: .atomic) + try data.write(to: manifestURL, options: []) isPersisted = true await datastore.mark(identifier: id, asLoaded: true) } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift index f35e6a0..8fcc0a7 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift @@ -156,7 +156,7 @@ extension DiskPersistence.Datastore.Page { try FileManager.default.createDirectory(at: pageURL.deletingLastPathComponent(), withIntermediateDirectories: true) /// Write the bytes for the page to disk. - try Data(bytes).write(to: pageURL, options: .atomic) + try Data(bytes).write(to: pageURL, options: []) isPersisted = true await datastore.mark(identifier: id, asLoaded: true) } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRoot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRoot.swift index 2fc8d51..f5d3b19 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRoot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRoot.swift @@ -131,7 +131,7 @@ extension DiskPersistence.Datastore.RootObject { /// Encode the provided manifest, and write it to disk. let data = try JSONEncoder.shared.encode(rootObject) - try data.write(to: rootObjectURL, options: .atomic) + try data.write(to: rootObjectURL, options: []) isPersisted = true await datastore.mark(identifier: id, asLoaded: true) } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift index c904ce9..49fa4c5 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift @@ -169,7 +169,7 @@ extension Snapshot { /// Encode the provided iteration, and write it to disk. let data = try JSONEncoder.shared.encode(iteration) - try data.write(to: iterationURL, options: .atomic) + try data.write(to: iterationURL, options: []) /// Update the cache since we know what it should be. cachedIteration = iteration