From 21450e3d026d782d2baee6f3946535dd722a42ca Mon Sep 17 00:00:00 2001 From: Qiwei Yang Date: Fri, 2 Oct 2026 09:35:28 +0800 Subject: [PATCH] refactor accumulation organization --- .../Accumulation/AccountChanges.swift | 129 +++++++++ .../Accumulation/Accumulation.swift | 19 ++ .../AccumulationExecution.swift} | 263 ++---------------- .../AccumulationInvocationTypes.swift} | 6 + .../Accumulation/AccumulationTypes.swift | 75 +++++ 5 files changed, 254 insertions(+), 238 deletions(-) create mode 100644 Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccountChanges.swift create mode 100644 Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/Accumulation.swift rename Blockchain/Sources/Blockchain/RuntimeProtocols/{Accumulation.swift => Accumulation/AccumulationExecution.swift} (72%) rename Blockchain/Sources/Blockchain/RuntimeProtocols/{AccumulateFunction.swift => Accumulation/AccumulationInvocationTypes.swift} (99%) create mode 100644 Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationTypes.swift diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccountChanges.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccountChanges.swift new file mode 100644 index 00000000..e2945379 --- /dev/null +++ b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccountChanges.swift @@ -0,0 +1,129 @@ +import Codec +import TracingUtils +import Utils + +private let logger = Logger(label: "Accumulation") + +public struct AccountChanges: Sendable { + public enum UpdateKind: Sendable { + case newAccount(ServiceIndex, ServiceAccount) + case removeAccount(ServiceIndex) + case updateAccount(ServiceIndex, ServiceAccountDetails) + case updateStorage(ServiceIndex, Data, Data?) + case updatePreimage(ServiceIndex, Data32, Data?) + case updateLookup(ServiceIndex, Data32, UInt32, StateKeys.ServiceAccountLookupKey.Value?) + } + + // records for checking conflicts + public var newAccounts: [ServiceIndex: ServiceAccount] + public var altered: Set + public var removed: Set + + /// array for apply sequential updates + public var updates: [UpdateKind] + + public init() { + newAccounts = [:] + altered = [] + removed = [] + updates = [] + } + + public mutating func addNewAccount(index: ServiceIndex, account: ServiceAccount) { + newAccounts[index] = account + updates.append(.newAccount(index, account)) + } + + public mutating func addRemovedAccount(index: ServiceIndex) { + removed.insert(index) + updates.append(.removeAccount(index)) + } + + public mutating func addAccountUpdate(index: ServiceIndex, account: ServiceAccountDetails) { + altered.insert(index) + updates.append(.updateAccount(index, account)) + } + + public mutating func addStorageUpdate(index: ServiceIndex, key: Data, value: Data?) { + altered.insert(index) + updates.append(.updateStorage(index, key, value)) + } + + public mutating func addPreimageUpdate(index: ServiceIndex, hash: Data32, value: Data?) { + altered.insert(index) + updates.append(.updatePreimage(index, hash, value)) + } + + public mutating func addLookupUpdate( + index: ServiceIndex, + hash: Data32, + length: UInt32, + value: StateKeys.ServiceAccountLookupKey.Value?, + ) { + altered.insert(index) + updates.append(.updateLookup(index, hash, length, value)) + } + + public func apply(to accounts: ServiceAccountsMutRef) async throws { + let removedIndices = Set(updates.compactMap { update -> ServiceIndex? in + if case let .removeAccount(index) = update { + return index + } + return nil + }) + + var pendingRemovals: [ServiceIndex] = [] + var seenRemovals: Set = [] + + for update in updates { + switch update { + case let .newAccount(index, account): + guard !removedIndices.contains(index) else { continue } + try await accounts.addNew(serviceAccount: index, account: account) + case let .removeAccount(index): + if seenRemovals.insert(index).inserted { + pendingRemovals.append(index) + } + case let .updateAccount(index, account): + guard !removedIndices.contains(index) else { continue } + accounts.set(serviceAccount: index, account: account) + case let .updateStorage(index, key, value): + guard !removedIndices.contains(index) else { continue } + try await accounts.set(serviceAccount: index, storageKey: key, value: value) + case let .updatePreimage(index, hash, value): + guard !removedIndices.contains(index) else { continue } + accounts.set(serviceAccount: index, preimageHash: hash, value: value) + case let .updateLookup(index, hash, length, value): + guard !removedIndices.contains(index) else { continue } + try await accounts.set(serviceAccount: index, preimageHash: hash, length: length, value: value) + } + } + + // Remove accounts last so removal dominates mixed update/remove batches. + for index in pendingRemovals { + try await accounts.remove(serviceAccount: index) + } + } + + public mutating func checkAndMerge(with other: AccountChanges) throws(AccumulationError) { + guard Set(newAccounts.keys).isDisjoint(with: other.newAccounts.keys) else { + logger.debug("new accounts have duplicates, self: \(newAccounts.keys), other: \(other.newAccounts.keys)") + throw .duplicatedNewService + } + guard altered.isDisjoint(with: other.altered) else { + logger.debug("same service being altered in parallel, self: \(altered), other: \(other.altered)") + throw .duplicatedContributionToService + } + guard removed.isDisjoint(with: other.removed) else { + logger.debug("removed accounts have duplicates, self: \(removed), other: \(other.removed)") + throw .duplicatedRemovedService + } + + for (index, account) in other.newAccounts { + newAccounts[index] = account + } + altered.formUnion(other.altered) + removed.formUnion(other.removed) + updates.append(contentsOf: other.updates) + } +} diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/Accumulation.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/Accumulation.swift new file mode 100644 index 00000000..6ae75d41 --- /dev/null +++ b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/Accumulation.swift @@ -0,0 +1,19 @@ +import Codec +import Utils + +public protocol Accumulation: ServiceAccounts { + var timeslot: TimeslotIndex { get } + var privilegedServices: PrivilegedServices { get set } + var validatorQueue: ConfigFixedSizeArray< + ValidatorKey, ProtocolConfig.TotalNumberOfValidators, + > { get set } + var authorizationQueue: ConfigFixedSizeArray< + ConfigFixedSizeArray< + Data32, + ProtocolConfig.MaxAuthorizationsQueueItems, + >, + ProtocolConfig.TotalNumberOfCores, + > { get set } + var accumulationQueue: StateKeys.AccumulationQueueKey.Value { get set } + var accumulationHistory: StateKeys.AccumulationHistoryKey.Value { get set } +} diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationExecution.swift similarity index 72% rename from Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation.swift rename to Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationExecution.swift index 281759e4..a446f1fb 100644 --- a/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation.swift +++ b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationExecution.swift @@ -4,220 +4,6 @@ import Utils private let logger = Logger(label: "Accumulation") -public enum AccumulationError: Error { - case invalidServiceIndex - case duplicatedNewService - case duplicatedContributionToService - case duplicatedRemovedService -} - -public struct AccumulationQueueItem: Sendable, Equatable, Codable { - public var workReport: WorkReport - @CodingAs> public var dependencies: Set - - public init(workReport: WorkReport, dependencies: Set) { - self.workReport = workReport - self.dependencies = dependencies - } -} - -/// accumulation output pairing -public struct Commitment: Hashable, Sendable, Equatable, Codable { - public var serviceIndex: ServiceIndex - public var hash: Data32 - - public init(service: ServiceIndex, hash: Data32) { - serviceIndex = service - self.hash = hash - } -} - -/// outer accumulation function ∆+ output -public struct AccumulationOutput { - // number of work results accumulated - public var numAccumulated: Int - public var state: AccumulateState - public var commitments: Set - public var gasUsed: [(serviceIndex: ServiceIndex, gas: Gas)] -} - -/// parallelized accumulation function ∆* output -public struct ParallelAccumulationOutput { - public var state: AccumulateState - public var transfers: [DeferredTransfers] - public var commitments: Set - public var gasUsed: [(serviceIndex: ServiceIndex, gas: Gas)] -} - -/// single-service accumulation function ∆1 output -public typealias SingleAccumulationOutput = AccumulationResult - -public struct ServicePreimagePair: Hashable, Sendable { - public var serviceIndex: ServiceIndex - public var preimage: Data - - public init(service: ServiceIndex, preimage: Data) { - serviceIndex = service - self.preimage = preimage - } -} - -public struct AccumulationResult: Sendable { - /// e - public var state: AccumulateState - /// t - public var transfers: [DeferredTransfers] - /// y - public var commitment: Data32? - /// u - public var gasUsed: Gas - /// p - public var provide: Set -} - -public struct AccountChanges: Sendable { - public enum UpdateKind: Sendable { - case newAccount(ServiceIndex, ServiceAccount) - case removeAccount(ServiceIndex) - case updateAccount(ServiceIndex, ServiceAccountDetails) - case updateStorage(ServiceIndex, Data, Data?) - case updatePreimage(ServiceIndex, Data32, Data?) - case updateLookup(ServiceIndex, Data32, UInt32, StateKeys.ServiceAccountLookupKey.Value?) - } - - // records for checking conflicts - public var newAccounts: [ServiceIndex: ServiceAccount] - public var altered: Set - public var removed: Set - - /// array for apply sequential updates - public var updates: [UpdateKind] - - public init() { - newAccounts = [:] - altered = [] - removed = [] - updates = [] - } - - public mutating func addNewAccount(index: ServiceIndex, account: ServiceAccount) { - newAccounts[index] = account - updates.append(.newAccount(index, account)) - } - - public mutating func addRemovedAccount(index: ServiceIndex) { - removed.insert(index) - updates.append(.removeAccount(index)) - } - - public mutating func addAccountUpdate(index: ServiceIndex, account: ServiceAccountDetails) { - altered.insert(index) - updates.append(.updateAccount(index, account)) - } - - public mutating func addStorageUpdate(index: ServiceIndex, key: Data, value: Data?) { - altered.insert(index) - updates.append(.updateStorage(index, key, value)) - } - - public mutating func addPreimageUpdate(index: ServiceIndex, hash: Data32, value: Data?) { - altered.insert(index) - updates.append(.updatePreimage(index, hash, value)) - } - - public mutating func addLookupUpdate( - index: ServiceIndex, - hash: Data32, - length: UInt32, - value: StateKeys.ServiceAccountLookupKey.Value?, - ) { - altered.insert(index) - updates.append(.updateLookup(index, hash, length, value)) - } - - public func apply(to accounts: ServiceAccountsMutRef) async throws { - let removedIndices = Set(updates.compactMap { update -> ServiceIndex? in - if case let .removeAccount(index) = update { - return index - } - return nil - }) - - var pendingRemovals: [ServiceIndex] = [] - var seenRemovals: Set = [] - - for update in updates { - switch update { - case let .newAccount(index, account): - guard !removedIndices.contains(index) else { continue } - try await accounts.addNew(serviceAccount: index, account: account) - case let .removeAccount(index): - if seenRemovals.insert(index).inserted { - pendingRemovals.append(index) - } - case let .updateAccount(index, account): - guard !removedIndices.contains(index) else { continue } - accounts.set(serviceAccount: index, account: account) - case let .updateStorage(index, key, value): - guard !removedIndices.contains(index) else { continue } - try await accounts.set(serviceAccount: index, storageKey: key, value: value) - case let .updatePreimage(index, hash, value): - guard !removedIndices.contains(index) else { continue } - accounts.set(serviceAccount: index, preimageHash: hash, value: value) - case let .updateLookup(index, hash, length, value): - guard !removedIndices.contains(index) else { continue } - try await accounts.set(serviceAccount: index, preimageHash: hash, length: length, value: value) - } - } - - // Remove accounts last so removal dominates mixed update/remove batches. - for index in pendingRemovals { - try await accounts.remove(serviceAccount: index) - } - } - - public mutating func checkAndMerge(with other: AccountChanges) throws(AccumulationError) { - guard Set(newAccounts.keys).isDisjoint(with: other.newAccounts.keys) else { - logger.debug("new accounts have duplicates, self: \(newAccounts.keys), other: \(other.newAccounts.keys)") - throw .duplicatedNewService - } - guard altered.isDisjoint(with: other.altered) else { - logger.debug("same service being altered in parallel, self: \(altered), other: \(other.altered)") - throw .duplicatedContributionToService - } - guard removed.isDisjoint(with: other.removed) else { - logger.debug("removed accounts have duplicates, self: \(removed), other: \(other.removed)") - throw .duplicatedRemovedService - } - - for (index, account) in other.newAccounts { - newAccounts[index] = account - } - altered.formUnion(other.altered) - removed.formUnion(other.removed) - updates.append(contentsOf: other.updates) - } -} - -public protocol Accumulation: ServiceAccounts { - var timeslot: TimeslotIndex { get } - var privilegedServices: PrivilegedServices { get set } - var validatorQueue: ConfigFixedSizeArray< - ValidatorKey, ProtocolConfig.TotalNumberOfValidators, - > { get set } - var authorizationQueue: ConfigFixedSizeArray< - ConfigFixedSizeArray< - Data32, - ProtocolConfig.MaxAuthorizationsQueueItems, - >, - ProtocolConfig.TotalNumberOfCores, - > { get set } - var accumulationQueue: StateKeys.AccumulationQueueKey.Value { get set } - var accumulationHistory: StateKeys.AccumulationHistoryKey.Value { get set } -} - -public typealias AccumulationStats = [ServiceIndex: (Gas, UInt32)] - extension Accumulation { /// single-service accumulate function ∆1 private static func singleAccumulate( @@ -530,7 +316,28 @@ extension Accumulation { } } - // E: edit the accumulation queue items when some work reports are accumulated + /// accumulate execution + private func execution( + config: ProtocolConfigRef, + workReports: [WorkReport], + state: AccumulateState, + timeslot: TimeslotIndex, + ) async throws -> AccumulationOutput { + let sumPrivilegedGas = privilegedServices.alwaysAcc.values.reduce(Gas(0)) { $0 + $1.value } + let minTotalGas = config.value.workReportAccumulationGas * Gas(config.value.totalNumberOfCores) + sumPrivilegedGas + let gasLimit = max(config.value.totalAccumulationGas, minTotalGas) + + return try await outerAccumulate( + config: config, + state: state, + transfers: [], + workReports: workReports, + alwaysAcc: privilegedServices.alwaysAcc, + gasLimit: gasLimit, + timeslot: timeslot, + ) + } + private func editQueue(items: inout [AccumulationQueueItem], accumulatedPackages: Set) { items = items.filter { !accumulatedPackages.contains($0.workReport.packageSpecification.workPackageHash) } @@ -592,28 +399,6 @@ extension Accumulation { return (zeroPrereqReports + getAccumulatables(items: &allQueueItems), newQueueItems) } - /// accumulate execution - private func execution( - config: ProtocolConfigRef, - workReports: [WorkReport], - state: AccumulateState, - timeslot: TimeslotIndex, - ) async throws -> AccumulationOutput { - let sumPrivilegedGas = privilegedServices.alwaysAcc.values.reduce(Gas(0)) { $0 + $1.value } - let minTotalGas = config.value.workReportAccumulationGas * Gas(config.value.totalNumberOfCores) + sumPrivilegedGas - let gasLimit = max(config.value.totalAccumulationGas, minTotalGas) - - return try await outerAccumulate( - config: config, - state: state, - transfers: [], - workReports: workReports, - alwaysAcc: privilegedServices.alwaysAcc, - gasLimit: gasLimit, - timeslot: timeslot, - ) - } - /// Accumulate execution, state integration and deferred transfers public mutating func update( config: ProtocolConfigRef, @@ -707,7 +492,9 @@ extension Accumulation { for (service, gasUsed) in gasUsedMap { let num = digestCounts[service] ?? 0 - if Int(gasUsed.value) + num == 0 { continue } + if Int(gasUsed.value) + num == 0 { + continue + } accumulateStats[service] = (gasUsed, UInt32(num)) } diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/AccumulateFunction.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationInvocationTypes.swift similarity index 99% rename from Blockchain/Sources/Blockchain/RuntimeProtocols/AccumulateFunction.swift rename to Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationInvocationTypes.swift index f24847b4..675a8594 100644 --- a/Blockchain/Sources/Blockchain/RuntimeProtocols/AccumulateFunction.swift +++ b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationInvocationTypes.swift @@ -2,6 +2,8 @@ import Codec import Foundation import Utils +// MARK: Inputs + public struct OperandTuple: Codable, Sendable { /// p public var packageHash: Data32 @@ -95,6 +97,8 @@ public struct AccumulationInput: Sendable, Codable { } } +// MARK: State + /// Characterization (i.e. values capable of representing) of state components /// which are both needed and mutable by the accumulation process. public struct AccumulateState: Sendable { @@ -140,6 +144,8 @@ public struct AccumulateState: Sendable { } } +// MARK: Result + public class AccumulateResultContext { /// s: the accumulating service account index public var serviceIndex: ServiceIndex diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationTypes.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationTypes.swift new file mode 100644 index 00000000..9ad4391d --- /dev/null +++ b/Blockchain/Sources/Blockchain/RuntimeProtocols/Accumulation/AccumulationTypes.swift @@ -0,0 +1,75 @@ +import Codec +import Utils + +public enum AccumulationError: Error { + case invalidServiceIndex + case duplicatedNewService + case duplicatedContributionToService + case duplicatedRemovedService +} + +public struct AccumulationQueueItem: Sendable, Equatable, Codable { + public var workReport: WorkReport + @CodingAs> public var dependencies: Set + + public init(workReport: WorkReport, dependencies: Set) { + self.workReport = workReport + self.dependencies = dependencies + } +} + +/// accumulation output pairing +public struct Commitment: Hashable, Sendable, Equatable, Codable { + public var serviceIndex: ServiceIndex + public var hash: Data32 + + public init(service: ServiceIndex, hash: Data32) { + serviceIndex = service + self.hash = hash + } +} + +/// outer accumulation function ∆+ output +public struct AccumulationOutput { + // number of work results accumulated + public var numAccumulated: Int + public var state: AccumulateState + public var commitments: Set + public var gasUsed: [(serviceIndex: ServiceIndex, gas: Gas)] +} + +/// parallelized accumulation function ∆* output +public struct ParallelAccumulationOutput { + public var state: AccumulateState + public var transfers: [DeferredTransfers] + public var commitments: Set + public var gasUsed: [(serviceIndex: ServiceIndex, gas: Gas)] +} + +/// single-service accumulation function ∆1 output +public typealias SingleAccumulationOutput = AccumulationResult + +public struct ServicePreimagePair: Hashable, Sendable { + public var serviceIndex: ServiceIndex + public var preimage: Data + + public init(service: ServiceIndex, preimage: Data) { + serviceIndex = service + self.preimage = preimage + } +} + +public struct AccumulationResult: Sendable { + /// e + public var state: AccumulateState + /// t + public var transfers: [DeferredTransfers] + /// y + public var commitment: Data32? + /// u + public var gasUsed: Gas + /// p + public var provide: Set +} + +public typealias AccumulationStats = [ServiceIndex: (Gas, UInt32)]