Small updates

This commit is contained in:
Michal Fousek 2023-05-10 08:47:03 +02:00
parent 92994b067e
commit 2501cea4b4
15 changed files with 134 additions and 57 deletions

View File

@ -39,6 +39,8 @@ enum CBPState: CaseIterable {
case stopped case stopped
} }
// this is replacement for CompactBlockProgress
enum ActionProgress { enum ActionProgress {
case scan case scan
} }
@ -56,4 +58,7 @@ protocol Action {
// different conditions. And action is the thing that knows these conditions. // different conditions. And action is the thing that knows these conditions.
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext
// Should be called on each existing action when processor wants to stop. Some actions may do it's own background work.
func stop()
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class ChecksBeforeSyncAction { class ChecksBeforeSyncAction {
init() { } init(container: DIContainer) { }
} }
extension ChecksBeforeSyncAction: Action { extension ChecksBeforeSyncAction: Action {
@ -24,7 +24,11 @@ extension ChecksBeforeSyncAction: Action {
// try await storage.create() // try await storage.create()
// } // }
await context.update(state: .scan) await context.update(state: .fetchUTXO)
return context return context
} }
func stop() {
}
} }

View File

@ -8,18 +8,18 @@
import Foundation import Foundation
class ClearAlreadyScannedBlocksAction { class ClearAlreadyScannedBlocksAction {
init() { } init(container: DIContainer) { }
} }
extension ClearAlreadyScannedBlocksAction: Action { extension ClearAlreadyScannedBlocksAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext { func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
// clear storage but delete only blocks that were already scanned, when doing parallel download all blocks can't be deleted // clear storage but delete only blocks that were already scanned, when doing parallel download all blocks can't be deleted
// if latestScannedHeight == context.scanRanges.downloadAndScanRange?.upperBound then set state `enhance`. Everything is scanned.
// If latestScannedHeight < context.scanRanges.downloadAndScanRange?.upperBound thne set state to `download` because there are blocks to
// download and scan.
await context.update(state: .enhance) await context.update(state: .enhance)
return context return context
} }
func stop() {
}
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class ClearCacheAction { class ClearCacheAction {
init() { } init(container: DIContainer) { }
} }
extension ClearCacheAction: Action { extension ClearCacheAction: Action {
@ -17,4 +17,8 @@ extension ClearCacheAction: Action {
await context.update(state: .finished) await context.update(state: .finished)
return context return context
} }
func stop() {
}
} }

View File

@ -8,13 +8,19 @@
import Foundation import Foundation
class ComputeSyncRangesAction { class ComputeSyncRangesAction {
init() { } init(container: DIContainer) { }
} }
extension ComputeSyncRangesAction: Action { extension ComputeSyncRangesAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext { func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
// call internalSyncProgress and compute sync ranges and store them in context // call internalSyncProgress and compute sync ranges and store them in context
// if there is nothing sync just switch to finished state
await context.update(state: .checksBeforeSync) await context.update(state: .checksBeforeSync)
return context return context
} }
func stop() {
}
} }

View File

@ -8,15 +8,30 @@
import Foundation import Foundation
class DownloadAction { class DownloadAction {
init() { } init(container: DIContainer) { }
} }
extension DownloadAction: Action { extension DownloadAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext { func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
// Use `BlockDownloader` to set download limit to latestScannedHeight + (2*batchSize) (after parallel is merged). // Use `BlockDownloader` to set download limit to latestScannedHeight + (2*batchSize) (after parallel is merged).
// And start download. // And start download.
// Compute batch sync range (range used by one loop in `downloadAndScanBlocks` method) and wait until blocks in this range are downloaded.
// do {
// await blockDownloader.setDownloadLimit(processingRange.upperBound + (2 * batchSize))
// await blockDownloader.startDownload(maxBlockBufferSize: config.downloadBufferSize)
//
// try await blockDownloader.waitUntilRequestedBlocksAreDownloaded(in: processingRange)
// } catch {
// await ifTaskIsNotCanceledClearCompactBlockCache(lastScannedHeight: lastScannedHeight)
// throw error
// }
await context.update(state: .validate) await context.update(state: .validate)
return context return context
} }
func stop() {
}
} }

View File

@ -8,14 +8,24 @@
import Foundation import Foundation
class EnhanceAction { class EnhanceAction {
init() { } init(container: DIContainer) { }
} }
extension EnhanceAction: Action { extension EnhanceAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext { func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
// Use `BlockEnhancer` to enhance blocks. // Use `BlockEnhancer` to enhance blocks.
// This action is executed on each downloaded and scanned batch (typically each 100 blocks). But we want to run enhancement each 1000 blocks.
// This action can use `InternalSyncProgress` and last scanned height to compute when it should do work.
await context.update(state: .fetchUTXO) // if latestScannedHeight == context.scanRanges.downloadAndScanRange?.upperBound then set state `enhance`. Everything is scanned.
// If latestScannedHeight < context.scanRanges.downloadAndScanRange?.upperBound thne set state to `download` because there are blocks to
// download and scan.
await context.update(state: .clearCache)
return context return context
} }
func stop() {
}
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class FetchUTXOsAction { class FetchUTXOsAction {
init() { } init(container: DIContainer) { }
} }
extension FetchUTXOsAction: Action { extension FetchUTXOsAction: Action {
@ -18,4 +18,8 @@ extension FetchUTXOsAction: Action {
await context.update(state: .handleSaplingParams) await context.update(state: .handleSaplingParams)
return context return context
} }
func stop() {
}
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class SaplingParamsAction { class SaplingParamsAction {
init() { } init(container: DIContainer) { }
} }
extension SaplingParamsAction: Action { extension SaplingParamsAction: Action {
@ -16,7 +16,11 @@ extension SaplingParamsAction: Action {
// Download files with sapling params. // Download files with sapling params.
await context.update(state: .clearCache) await context.update(state: .scanDownloaded)
return context return context
} }
func stop() {
}
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class ScanAction { class ScanAction {
init() { } init(container: DIContainer) { }
} }
extension ScanAction: Action { extension ScanAction: Action {
@ -18,4 +18,9 @@ extension ScanAction: Action {
await context.update(state: .clearAlreadyScannedBlocks) await context.update(state: .clearAlreadyScannedBlocks)
return context return context
} }
func stop() {
}
} }

View File

@ -0,0 +1,35 @@
//
// ScandownloadedButUnscannedAction.swift
//
//
// Created by Michal Fousek on 05.05.2023.
//
import Foundation
class ScanDownloadedButUnscannedAction {
init(container: DIContainer) { }
}
extension ScanDownloadedButUnscannedAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
// if let range = ranges.downloadedButUnscannedRange {
// logger.debug("Starting scan with downloaded but not scanned blocks with range: \(range.lowerBound)...\(range.upperBound)")
// try await blockScanner.scanBlocks(at: range, totalProgressRange: totalProgressRange) { [weak self] lastScannedHeight in
// let progress = BlockProgress(
// startHeight: totalProgressRange.lowerBound,
// targetHeight: totalProgressRange.upperBound,
// progressHeight: lastScannedHeight
// )
// await self?.notifyProgress(.syncing(progress))
// }
// }
await context.update(state: .download)
return context
}
func stop() {
}
}

View File

@ -1,23 +0,0 @@
//
// ScandownloadedButUnscannedAction.swift
//
//
// Created by Michal Fousek on 05.05.2023.
//
import Foundation
class ScandownloadedButUnscannedAction {
init() { }
}
extension ScandownloadedButUnscannedAction: Action {
func run(with context: ActionContext, didUpdate: @escaping (ActionProgress) async -> Void) async throws -> ActionContext {
if let downloadedButUnscannedRange = await context.syncRanges.downloadedButUnscannedRange {
// Use `BlockScanner` to do the scanning in this range.
}
await context.update(state: .download)
return context
}
}

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class ValidateAction { class ValidateAction {
init() { } init(container: DIContainer) { }
} }
extension ValidateAction: Action { extension ValidateAction: Action {
@ -19,4 +19,8 @@ extension ValidateAction: Action {
await context.update(state: .scan) await context.update(state: .scan)
return context return context
} }
func stop() {
}
} }

View File

@ -8,7 +8,7 @@
import Foundation import Foundation
class ValidateServerAction { class ValidateServerAction {
init() { } init(container: DIContainer) { }
} }
extension ValidateServerAction: Action { extension ValidateServerAction: Action {
@ -41,4 +41,8 @@ extension ValidateServerAction: Action {
await context.update(state: .computeSyncRanges) await context.update(state: .computeSyncRanges)
return context return context
} }
func stop() {
}
} }

View File

@ -20,41 +20,41 @@ class CompactBlockProcessorNG {
let logger: Logger let logger: Logger
init(logger: Logger) { init(container: DIContainer) {
context = ActionContext(state: .validateServer) context = ActionContext(state: .validateServer)
actions = Self.makeActions() actions = Self.makeActions(container: container)
self.logger = logger self.logger = container.resolve(Logger.self)
} }
// swiftlint:disable:next cyclomatic_complexity // swiftlint:disable:next cyclomatic_complexity
static func makeActions() -> [CBPState: Action] { static func makeActions(container: DIContainer) -> [CBPState: Action] {
let actionsDefinition = CBPState.allCases.compactMap { state -> (CBPState, Action)? in let actionsDefinition = CBPState.allCases.compactMap { state -> (CBPState, Action)? in
let action: Action let action: Action
switch state { switch state {
case .validateServer: case .validateServer:
action = ValidateServerAction() action = ValidateServerAction(container: container)
case .computeSyncRanges: case .computeSyncRanges:
action = ComputeSyncRangesAction() action = ComputeSyncRangesAction(container: container)
case .checksBeforeSync: case .checksBeforeSync:
action = ChecksBeforeSyncAction() action = ChecksBeforeSyncAction(container: container)
case .scanDownloaded: case .scanDownloaded:
action = ScandownloadedButUnscannedAction() action = ScanDownloadedButUnscannedAction(container: container)
case .download: case .download:
action = DownloadAction() action = DownloadAction(container: container)
case .validate: case .validate:
action = ValidateAction() action = ValidateAction(container: container)
case .scan: case .scan:
action = ScanAction() action = ScanAction(container: container)
case .clearAlreadyScannedBlocks: case .clearAlreadyScannedBlocks:
action = ClearAlreadyScannedBlocksAction() action = ClearAlreadyScannedBlocksAction(container: container)
case .enhance: case .enhance:
action = EnhanceAction() action = EnhanceAction(container: container)
case .fetchUTXO: case .fetchUTXO:
action = FetchUTXOsAction() action = FetchUTXOsAction(container: container)
case .handleSaplingParams: case .handleSaplingParams:
action = SaplingParamsAction() action = SaplingParamsAction(container: container)
case .clearCache: case .clearCache:
action = ClearCacheAction() action = ClearCacheAction(container: container)
case .finished, .failed, .stopped: case .finished, .failed, .stopped:
return nil return nil
} }