1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
//===----------------------------------------------------------------------===//
//
// This source file is part of the Swift open source project
//
// Copyright (c) 2023-2024 Apple Inc. and the Swift project authors
// Licensed under Apache License v2.0 with Runtime Library Exception
//
// See https://swift.org/LICENSE.txt for license information
// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
//
//===----------------------------------------------------------------------===//
import _AsyncFileSystem
import Basics
import Crypto
package func withQueryEngine(
_ fileSystem: some AsyncFileSystem,
_ observabilityScope: ObservabilityScope,
cacheLocation: SQLite.Location,
_ body: @Sendable (QueryEngine) async throws -> Void
) async throws {
let engine = QueryEngine(
fileSystem,
observabilityScope,
cacheLocation: cacheLocation
)
try await withAsyncThrowing {
try await body(engine)
} defer: {
try await engine.shutDown()
}
}
/// Cacheable computations engine. Currently the engine makes an assumption that computations produce same results for
/// the same query values and write results to a single file path.
package actor QueryEngine {
private(set) var cacheHits = 0
private(set) var cacheMisses = 0
package let fileSystem: any AsyncFileSystem
package let httpClient = HTTPClient()
package let observabilityScope: ObservabilityScope
private let resultsCache: SQLiteBackedCache<FileCacheRecord>
private var isShutDown = false
/// Creates a new instance of the ``QueryEngine`` actor. Requires an explicit call
/// to ``QueryEngine//shutdown`` before the instance is deinitialized. The recommended approach to resource
/// management is to place `engine.shutDown()` when the engine is no longer used, but is not deinitialized yet.
/// - Parameter fileSystem: Implementation of a file system this engine should use.
/// - Parameter cacheLocation: Location of cache storage used by the engine.
/// - Parameter logger: Logger to use during queries execution.
init(
_ fileSystem: any AsyncFileSystem,
_ observabilityScope: ObservabilityScope,
cacheLocation: SQLite.Location
) {
self.fileSystem = fileSystem
self.observabilityScope = observabilityScope
self.resultsCache = SQLiteBackedCache(tableName: "cache_table", location: cacheLocation)
}
package func shutDown() async throws {
precondition(!self.isShutDown, "`QueryEngine/shutDown` should be called only once")
try self.resultsCache.close()
self.isShutDown = true
}
deinit {
let isShutDown = self.isShutDown
precondition(
isShutDown,
"`QueryEngine/shutDown` should be called explicitly on instances of `Engine` before deinitialization"
)
}
/// Executes a given query if no cached result of it is available. Otherwise fetches the result from engine's cache.
/// - Parameter query: A query value to execute.
/// - Returns: A file path to query's result recorded in a file.
package subscript(_ query: some Query) -> FileCacheRecord {
get async throws {
let hashEncoder = HashEncoder<SHA512>()
try hashEncoder.encode(query.cacheKey)
let key = hashEncoder.finalize()
if let fileRecord = try resultsCache.get(blobKey: key) {
let fileHash = try await self.fileSystem.withOpenReadableFile(fileRecord.path) {
var hashFunction = SHA512()
try await $0.hash(with: &hashFunction)
return hashFunction.finalize().description
}
if fileHash == fileRecord.hash {
self.cacheHits += 1
return fileRecord
}
}
self.cacheMisses += 1
let resultPath = try await query.run(engine: self)
let resultHash = try await self.fileSystem.withOpenReadableFile(resultPath) {
var hashFunction = SHA512()
try await $0.hash(with: &hashFunction)
return hashFunction.finalize().description
}
let result = FileCacheRecord(path: resultPath, hash: resultHash)
// FIXME: update `SQLiteBackedCache` to store `resultHash` directly instead of relying on string conversions
try self.resultsCache.put(blobKey: key, value: result)
return result
}
}
}
|
