1//===----------------------------------------------------------------------===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8///
9/// \file
10/// Implementation of the LLVM CAS plugin API, for testing purposes.
11///
12/// It is backed by \c UnifiedOnDiskCache and can optionally be given a second
13/// on-disk path via the \c upstream-path option, which it uses to simulate
14/// "uploading"/"downloading" objects to/from a distributed CAS.
15///
16//===----------------------------------------------------------------------===//
17
18#include "llvm-c/CAS/PluginAPI_functions.h"
19#include "llvm/CAS/BuiltinObjectHasher.h"
20#include "llvm/CAS/CASID.h"
21#include "llvm/CAS/OnDiskKeyValueDB.h"
22#include "llvm/CAS/UnifiedOnDiskCache.h"
23#include "llvm/Support/CBindingWrapping.h"
24#include "llvm/Support/Errc.h"
25#include "llvm/Support/Error.h"
26#include "llvm/Support/MemoryBuffer.h"
27#include "llvm/Support/SHA1.h"
28#include "llvm/Support/ThreadPool.h"
29#include <mutex>
30
31using namespace llvm;
32using namespace llvm::cas;
33using namespace llvm::cas::ondisk;
34
35namespace llvm::cas::ondisk {
36/// Declared in the private "OnDiskCommon.h"; see \c setSmallMaxMappingSize.
37void setMaxMappingSize(uint64_t Size);
38} // namespace llvm::cas::ondisk
39
40/// This plugin exists only for testing, and a test process can create many
41/// instances of it. Keep the on-disk mappings small so that they stay cheap;
42/// the default sizes are measured in gigabytes per instance.
43///
44/// This has to happen inside the plugin: it links its own copy of LLVMCAS, so
45/// the setting the test binary applies to itself does not reach us.
46static void setSmallMaxMappingSize() {
47 static std::once_flag Flag;
48 std::call_once(once&: Flag, f: [] { setMaxMappingSize(100 * 1024 * 1024); });
49}
50
51static char *copyNewMallocString(StringRef Str) {
52 char *c_str = (char *)malloc(size: Str.size() + 1);
53 std::uninitialized_copy(first: Str.begin(), last: Str.end(), result: c_str);
54 c_str[Str.size()] = '\0';
55 return c_str;
56}
57
58template <typename ResT>
59static ResT reportError(Error &&E, char **error, ResT Result = ResT()) {
60 if (error)
61 *error = copyNewMallocString(Str: toString(E: std::move(E)));
62 return Result;
63}
64
65void llcas_get_plugin_version(unsigned *major, unsigned *minor) {
66 *major = LLCAS_VERSION_MAJOR;
67 *minor = LLCAS_VERSION_MINOR;
68}
69
70void llcas_string_dispose(char *str) { free(ptr: str); }
71
72namespace {
73
74struct CancellableState {
75 std::atomic<bool> Cancelled{false};
76};
77
78struct CancellableWrap {
79 std::shared_ptr<CancellableState> State;
80};
81
82DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CancellableWrap, llcas_cancellable_t)
83
84} // namespace
85
86void llcas_cancellable_cancel(llcas_cancellable_t c_cancellable) {
87 unwrap(P: c_cancellable)->State->Cancelled = true;
88}
89
90void llcas_cancellable_dispose(llcas_cancellable_t c_cancellable) {
91 delete unwrap(P: c_cancellable);
92}
93
94namespace {
95
96struct CASPluginOptions {
97 std::string OnDiskPath;
98 std::string UpstreamPath;
99 std::string FirstPrefix;
100 std::string SecondPrefix;
101 bool SimulateMissingObjects = false;
102 bool Logging = true;
103
104 Error setOption(StringRef Name, StringRef Value);
105};
106
107DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASPluginOptions, llcas_cas_options_t)
108
109} // namespace
110
111Error CASPluginOptions::setOption(StringRef Name, StringRef Value) {
112 if (Name == "first-prefix")
113 FirstPrefix = Value;
114 else if (Name == "second-prefix")
115 SecondPrefix = Value;
116 else if (Name == "upstream-path")
117 UpstreamPath = Value;
118 else if (Name == "simulate-missing-objects")
119 SimulateMissingObjects = true;
120 else if (Name == "no-logging")
121 Logging = false;
122 else
123 return createStringError(EC: errc::invalid_argument,
124 S: Twine("unknown option: ") + Name);
125 return Error::success();
126}
127
128llcas_cas_options_t llcas_cas_options_create(void) {
129 return wrap(P: new CASPluginOptions());
130}
131
132void llcas_cas_options_dispose(llcas_cas_options_t c_opts) {
133 delete unwrap(P: c_opts);
134}
135
136void llcas_cas_options_set_ondisk_path(llcas_cas_options_t c_opts,
137 const char *path) {
138 auto &Opts = *unwrap(P: c_opts);
139 Opts.OnDiskPath = path;
140}
141
142bool llcas_cas_options_set_option(llcas_cas_options_t c_opts, const char *name,
143 const char *value, char **error) {
144 auto &Opts = *unwrap(P: c_opts);
145 if (Error E = Opts.setOption(Name: name, Value: value))
146 return reportError(E: std::move(E), error, Result: true);
147 return false;
148}
149
150namespace {
151
152using HasherT = SHA1;
153using HashType = decltype(HasherT::hash(Data: std::declval<ArrayRef<uint8_t> &>()));
154
155class PluginCASContext : public CASContext {
156 void printIDImpl(raw_ostream &OS, const CASID &ID) const final {
157 PluginCASContext::printID(Digest: ID.getHash(), OS);
158 }
159
160public:
161 static StringRef getHashName() { return "SHA1"; }
162 StringRef getHashSchemaIdentifier() const final {
163 static const std::string ID =
164 ("llvm.cas.builtin.v2[" + getHashName() + "]").str();
165 return ID;
166 }
167
168 PluginCASContext() = default;
169
170 static Expected<HashType> parseID(StringRef Reference) {
171 if (!Reference.consume_front(Prefix: "llvmcas://"))
172 return createStringError(
173 EC: std::make_error_code(e: std::errc::invalid_argument),
174 S: "invalid cas-id '" + Reference + "'");
175
176 if (Reference.size() != 2 * sizeof(HashType))
177 return createStringError(
178 EC: std::make_error_code(e: std::errc::invalid_argument),
179 S: "wrong size for cas-id hash '" + Reference + "'");
180
181 std::string Binary;
182 if (!tryGetFromHex(Input: Reference, Output&: Binary))
183 return createStringError(
184 EC: std::make_error_code(e: std::errc::invalid_argument),
185 S: "invalid hash in cas-id '" + Reference + "'");
186
187 assert(Binary.size() == sizeof(HashType));
188 HashType Digest;
189 llvm::copy(Range&: Binary, Out: Digest.data());
190 return Digest;
191 }
192
193 static void printID(ArrayRef<uint8_t> Digest, raw_ostream &OS) {
194 SmallString<64> Hash;
195 toHex(Input: Digest, /*LowerCase=*/true, Output&: Hash);
196 OS << "llvmcas://" << Hash;
197 }
198};
199
200struct CASWrapper {
201 std::string FirstPrefix;
202 std::string SecondPrefix;
203 /// If true, asynchronous "download" of an object will treat it as missing.
204 bool SimulateMissingObjects = false;
205 bool Logging = true;
206 std::unique_ptr<UnifiedOnDiskCache> DB;
207 /// Used for testing the \c globally parameter of action cache APIs. Simulates
208 /// "uploading"/"downloading" objects from/to the primary on-disk path.
209 std::unique_ptr<UnifiedOnDiskCache> UpstreamDB;
210 DefaultThreadPool Pool{llvm::hardware_concurrency()};
211
212 std::mutex Lock{};
213
214 /// Check if the object is contained, in the "local" CAS only or "globally".
215 bool containsObject(ObjectID ID, bool Globally);
216
217 /// Load the object, potentially "downloading" it from upstream.
218 Expected<std::optional<ondisk::ObjectHandle>> loadObject(ObjectID ID);
219
220 /// "Uploads" a key and the associated full node graph.
221 Error upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value);
222
223 /// "Downloads" the ID associated with the key but not the node data. The node
224 /// itself and the rest of the nodes in the graph will be "downloaded" lazily
225 /// as they are visited.
226 Expected<std::optional<ObjectID>> downstreamKey(ArrayRef<uint8_t> Key);
227
228 /// Synchronized access to \c llvm::errs().
229 void syncErrs(llvm::function_ref<void(raw_ostream &OS)> Fn) {
230 if (!Logging) {
231 // Ignore log output.
232 SmallString<32> Buf;
233 raw_svector_ostream OS(Buf);
234 Fn(OS);
235 return;
236 }
237 std::unique_lock<std::mutex> LockGuard(Lock);
238 Fn(errs());
239 errs().flush();
240 }
241
242private:
243 /// "Uploads" the full object node graph.
244 Expected<ObjectID> upstreamNode(ObjectID Node);
245 /// "Downloads" only a single object node. The rest of the nodes in the graph
246 /// will be "downloaded" lazily as they are visited.
247 Expected<ObjectID> downstreamNode(ObjectID Node);
248};
249
250DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASWrapper, llcas_cas_t)
251
252} // namespace
253
254bool CASWrapper::containsObject(ObjectID ID, bool Globally) {
255 if (DB->getGraphDB().containsObject(Ref: ID))
256 return true;
257 if (!Globally || !UpstreamDB)
258 return false;
259
260 auto UpstreamID = expectedToOptional(
261 E: UpstreamDB->getGraphDB().getReference(Hash: DB->getGraphDB().getDigest(Ref: ID)));
262
263 if (!UpstreamID)
264 return false;
265
266 return UpstreamDB->getGraphDB().containsObject(Ref: *UpstreamID);
267}
268
269Expected<std::optional<ondisk::ObjectHandle>>
270CASWrapper::loadObject(ObjectID ID) {
271 std::optional<ondisk::ObjectHandle> Obj;
272 if (Error E = DB->getGraphDB().load(Ref: ID).moveInto(Value&: Obj))
273 return std::move(E);
274 if (Obj)
275 return Obj;
276 if (!UpstreamDB)
277 return std::nullopt;
278
279 // Try "downloading" the node from upstream.
280 auto UpstreamID =
281 UpstreamDB->getGraphDB().getReference(Hash: DB->getGraphDB().getDigest(Ref: ID));
282 if (!UpstreamID)
283 return UpstreamID.takeError();
284 std::optional<ObjectID> Ret;
285 if (Error E = downstreamNode(Node: *UpstreamID).moveInto(Value&: Ret))
286 return std::move(E);
287 return DB->getGraphDB().load(Ref: ID);
288}
289
290/// Imports a single object node.
291static Expected<ObjectID> importNode(ObjectID FromID, OnDiskGraphDB &FromDB,
292 OnDiskGraphDB &ToDB) {
293 auto ToID = ToDB.getReference(Hash: FromDB.getDigest(Ref: FromID));
294 if (!ToID)
295 return ToID.takeError();
296 if (ToDB.containsObject(Ref: *ToID))
297 return ToID;
298
299 std::optional<ondisk::ObjectHandle> FromH;
300 if (Error E = FromDB.load(Ref: FromID).moveInto(Value&: FromH))
301 return std::move(E);
302 if (!FromH)
303 return ToID;
304
305 auto Data = FromDB.getObjectData(Node: *FromH);
306 auto FromRefs = FromDB.getObjectRefs(Node: *FromH);
307 SmallVector<ObjectID> Refs;
308 for (ObjectID FromRef : FromRefs) {
309 auto Ref = ToDB.getReference(Hash: FromDB.getDigest(Ref: FromRef));
310 if (!Ref)
311 return Ref.takeError();
312 Refs.push_back(Elt: *Ref);
313 }
314
315 if (Error E = ToDB.store(ID: *ToID, Refs, Data))
316 return std::move(E);
317 return ToID;
318}
319
320Expected<ObjectID> CASWrapper::upstreamNode(ObjectID Node) {
321 OnDiskGraphDB &FromDB = DB->getGraphDB();
322 OnDiskGraphDB &ToDB = UpstreamDB->getGraphDB();
323
324 std::optional<ondisk::ObjectHandle> FromH;
325 if (Error E = FromDB.load(Ref: Node).moveInto(Value&: FromH))
326 return std::move(E);
327 if (!FromH)
328 return createStringError(EC: errc::invalid_argument, S: "node doesn't exist");
329
330 for (ObjectID Ref : FromDB.getObjectRefs(Node: *FromH)) {
331 std::optional<ObjectID> ID;
332 if (Error E = upstreamNode(Node: Ref).moveInto(Value&: ID))
333 return std::move(E);
334 }
335
336 return importNode(FromID: Node, FromDB, ToDB);
337}
338
339Expected<ObjectID> CASWrapper::downstreamNode(ObjectID Node) {
340 OnDiskGraphDB &FromDB = UpstreamDB->getGraphDB();
341 OnDiskGraphDB &ToDB = DB->getGraphDB();
342 return importNode(FromID: Node, FromDB, ToDB);
343}
344
345static Expected<ObjectID> cachePut(OnDiskKeyValueDB &DB, ArrayRef<uint8_t> Key,
346 ObjectID ID) {
347 auto Value = UnifiedOnDiskCache::getValueFromObjectID(ID);
348 auto Result = DB.put(Key, Value);
349 if (!Result)
350 return Result.takeError();
351 return UnifiedOnDiskCache::getObjectIDFromValue(Value: *Result);
352}
353
354static Expected<std::optional<ObjectID>> cacheGet(OnDiskKeyValueDB &DB,
355 ArrayRef<uint8_t> Key) {
356 auto Result = DB.get(Key);
357 if (!Result)
358 return Result.takeError();
359 if (!*Result)
360 return std::nullopt;
361 return UnifiedOnDiskCache::getObjectIDFromValue(Value: **Result);
362}
363
364Error CASWrapper::upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value) {
365 if (!UpstreamDB)
366 return Error::success();
367 Expected<ObjectID> UpstreamVal = upstreamNode(Node: Value);
368 if (!UpstreamVal)
369 return UpstreamVal.takeError();
370 Expected<ObjectID> PutValue =
371 cachePut(DB&: UpstreamDB->getKeyValueDB(), Key, ID: *UpstreamVal);
372 if (!PutValue)
373 return PutValue.takeError();
374 assert(*PutValue == *UpstreamVal);
375 return Error::success();
376}
377
378Expected<std::optional<ObjectID>>
379CASWrapper::downstreamKey(ArrayRef<uint8_t> Key) {
380 if (!UpstreamDB)
381 return std::nullopt;
382 std::optional<ObjectID> UpstreamValue;
383 if (Error E =
384 cacheGet(DB&: UpstreamDB->getKeyValueDB(), Key).moveInto(Value&: UpstreamValue))
385 return std::move(E);
386 if (!UpstreamValue)
387 return std::nullopt;
388
389 auto Value = DB->getGraphDB().getReference(
390 Hash: UpstreamDB->getGraphDB().getDigest(Ref: *UpstreamValue));
391 if (!Value)
392 return Value.takeError();
393 Expected<ObjectID> PutValue = cachePut(DB&: DB->getKeyValueDB(), Key, ID: *Value);
394 if (!PutValue)
395 return PutValue.takeError();
396 assert(*PutValue == *Value);
397 return PutValue;
398}
399
400llcas_cas_t llcas_cas_create(llcas_cas_options_t c_opts, char **error) {
401 auto &Opts = *unwrap(P: c_opts);
402 setSmallMaxMappingSize();
403 Expected<std::unique_ptr<UnifiedOnDiskCache>> DB = UnifiedOnDiskCache::open(
404 Path: Opts.OnDiskPath, /*SizeLimit=*/std::nullopt,
405 HashName: PluginCASContext::getHashName(), HashByteSize: sizeof(HashType));
406 if (!DB)
407 return reportError<llcas_cas_t>(E: DB.takeError(), error);
408
409 std::unique_ptr<UnifiedOnDiskCache> UpstreamDB;
410 if (!Opts.UpstreamPath.empty()) {
411 if (Error E = UnifiedOnDiskCache::open(
412 Path: Opts.UpstreamPath, /*SizeLimit=*/std::nullopt,
413 HashName: PluginCASContext::getHashName(), HashByteSize: sizeof(HashType))
414 .moveInto(Value&: UpstreamDB))
415 return reportError<llcas_cas_t>(E: std::move(E), error);
416 }
417
418 return wrap(P: new CASWrapper{.FirstPrefix: Opts.FirstPrefix, .SecondPrefix: Opts.SecondPrefix,
419 .SimulateMissingObjects: Opts.SimulateMissingObjects, .Logging: Opts.Logging,
420 .DB: std::move(*DB), .UpstreamDB: std::move(UpstreamDB)});
421}
422
423void llcas_cas_dispose(llcas_cas_t c_cas) { delete unwrap(P: c_cas); }
424
425int64_t llcas_cas_get_ondisk_size(llcas_cas_t c_cas, char **error) {
426 return unwrap(P: c_cas)->DB->getStorageSize();
427}
428
429bool llcas_cas_set_ondisk_size_limit(llcas_cas_t c_cas, int64_t size_limit,
430 char **error) {
431 std::optional<uint64_t> SizeLimit;
432 if (size_limit < 0) {
433 return reportError(
434 E: llvm::createStringError(
435 EC: llvm::inconvertibleErrorCode(),
436 S: "invalid size limit passed to llcas_cas_set_ondisk_size_limit"),
437 error, Result: true);
438 }
439 if (size_limit > 0) {
440 SizeLimit = size_limit;
441 }
442 unwrap(P: c_cas)->DB->setSizeLimit(SizeLimit);
443 return false;
444}
445
446bool llcas_cas_prune_ondisk_data(llcas_cas_t c_cas, char **error) {
447 if (Error E = unwrap(P: c_cas)->DB->collectGarbage())
448 return reportError(E: std::move(E), error, Result: true);
449 return false;
450}
451
452void llcas_cas_options_set_client_version(llcas_cas_options_t, unsigned major,
453 unsigned minor) {
454 // Ignore for now.
455}
456
457char *llcas_cas_get_hash_schema_name(llcas_cas_t) {
458 // Using same name as builtin CAS so that it's interchangeable for testing
459 // purposes.
460 return copyNewMallocString(Str: "llvm.cas.builtin.v2[BLAKE3]");
461}
462
463unsigned llcas_digest_parse(llcas_cas_t c_cas, const char *printed_digest,
464 uint8_t *bytes, size_t bytes_size, char **error) {
465 auto &Wrapper = *unwrap(P: c_cas);
466 if (bytes_size < sizeof(HashType))
467 return sizeof(HashType);
468
469 StringRef PrintedDigest = printed_digest;
470 bool Consumed = PrintedDigest.consume_front(Prefix: Wrapper.FirstPrefix);
471 assert(Consumed);
472 (void)Consumed;
473 Consumed = PrintedDigest.consume_front(Prefix: Wrapper.SecondPrefix);
474 assert(Consumed);
475 (void)Consumed;
476
477 Expected<HashType> Digest = PluginCASContext::parseID(Reference: PrintedDigest);
478 if (!Digest)
479 return reportError(E: Digest.takeError(), error, Result: 0);
480 std::uninitialized_copy(first: Digest->begin(), last: Digest->end(), result: bytes);
481 return Digest->size();
482}
483
484bool llcas_digest_print(llcas_cas_t c_cas, llcas_digest_t c_digest,
485 char **printed_id, char **error) {
486 auto &Wrapper = *unwrap(P: c_cas);
487 SmallString<74> PrintDigest;
488 raw_svector_ostream OS(PrintDigest);
489 // Include these for testing purposes.
490 OS << Wrapper.FirstPrefix << Wrapper.SecondPrefix;
491 PluginCASContext::printID(Digest: ArrayRef(c_digest.data, c_digest.size), OS);
492 *printed_id = copyNewMallocString(Str: PrintDigest);
493 return false;
494}
495
496bool llcas_cas_get_objectid(llcas_cas_t c_cas, llcas_digest_t c_digest,
497 llcas_objectid_t *c_id_p, char **error) {
498 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
499 auto ID = CAS.getReference(Hash: ArrayRef(c_digest.data, c_digest.size));
500 if (!ID)
501 return reportError(E: ID.takeError(), error, Result: true);
502
503 *c_id_p = llcas_objectid_t{.opaque: ID->getOpaqueData()};
504 return false;
505}
506
507llcas_digest_t llcas_objectid_get_digest(llcas_cas_t c_cas,
508 llcas_objectid_t c_id) {
509 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
510 ObjectID ID = ObjectID::fromOpaqueData(Opaque: c_id.opaque);
511 ArrayRef<uint8_t> Digest = CAS.getDigest(Ref: ID);
512 return llcas_digest_t{.data: Digest.data(), .size: Digest.size()};
513}
514
515llcas_lookup_result_t llcas_cas_contains_object(llcas_cas_t c_cas,
516 llcas_objectid_t c_id,
517 bool globally, char **error) {
518 ObjectID ID = ObjectID::fromOpaqueData(Opaque: c_id.opaque);
519 return unwrap(P: c_cas)->containsObject(ID, Globally: globally)
520 ? LLCAS_LOOKUP_RESULT_SUCCESS
521 : LLCAS_LOOKUP_RESULT_NOTFOUND;
522}
523
524llcas_lookup_result_t llcas_cas_load_object(llcas_cas_t c_cas,
525 llcas_objectid_t c_id,
526 llcas_loaded_object_t *c_obj_p,
527 char **error) {
528 ObjectID ID = ObjectID::fromOpaqueData(Opaque: c_id.opaque);
529 Expected<std::optional<ondisk::ObjectHandle>> ObjOpt =
530 unwrap(P: c_cas)->loadObject(ID);
531 if (!ObjOpt)
532 return reportError(E: ObjOpt.takeError(), error, Result: LLCAS_LOOKUP_RESULT_ERROR);
533 if (!*ObjOpt)
534 return LLCAS_LOOKUP_RESULT_NOTFOUND;
535
536 ondisk::ObjectHandle Obj = **ObjOpt;
537 *c_obj_p = llcas_loaded_object_t{.opaque: Obj.getOpaqueData()};
538 return LLCAS_LOOKUP_RESULT_SUCCESS;
539}
540
541void llcas_cas_load_object_async(llcas_cas_t c_cas, llcas_objectid_t c_id,
542 void *ctx_cb, llcas_cas_load_object_cb cb,
543 llcas_cancellable_t *c_cancellable) {
544 auto CancelState = std::make_shared<CancellableState>();
545 if (c_cancellable) {
546 *c_cancellable = wrap(P: new CancellableWrap{.State: CancelState});
547 }
548
549 std::string PrintedDigest;
550 {
551 llcas_digest_t c_digest = llcas_objectid_get_digest(c_cas, c_id);
552 char *printed_id;
553 char *c_err;
554 bool failed = llcas_digest_print(c_cas, c_digest, printed_id: &printed_id, error: &c_err);
555 if (failed)
556 report_fatal_error(reason: Twine("digest printing failed: ") + c_err);
557 PrintedDigest = printed_id;
558 llcas_string_dispose(str: printed_id);
559 }
560
561 auto passObject = [ctx_cb,
562 cb](Expected<std::optional<ondisk::ObjectHandle>> Obj) {
563 if (!Obj) {
564 cb(ctx_cb, LLCAS_LOOKUP_RESULT_ERROR, llcas_loaded_object_t(),
565 copyNewMallocString(Str: toString(E: Obj.takeError())));
566 } else if (!*Obj) {
567 cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_loaded_object_t(),
568 nullptr);
569 } else {
570 cb(ctx_cb, LLCAS_LOOKUP_RESULT_SUCCESS,
571 llcas_loaded_object_t{.opaque: (*Obj)->getOpaqueData()}, nullptr);
572 }
573 };
574
575 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
576 ObjectID ID = ObjectID::fromOpaqueData(Opaque: c_id.opaque);
577 if (CAS.containsObject(Ref: ID)) {
578 unwrap(P: c_cas)->syncErrs(Fn: [&](raw_ostream &OS) {
579 OS << "load_object_async existing: " << PrintedDigest << '\n';
580 });
581 return passObject(unwrap(P: c_cas)->loadObject(ID));
582 }
583
584 if (!unwrap(P: c_cas)->UpstreamDB)
585 return passObject(std::nullopt);
586
587 // Try "downloading" the node from upstream.
588
589 unwrap(P: c_cas)->syncErrs(Fn: [&](raw_ostream &OS) {
590 OS << "load_object_async downstream begin: " << PrintedDigest << '\n';
591 });
592 unwrap(P: c_cas)->Pool.async(F: [=] {
593#if LLVM_ENABLE_THREADS
594 // Wait a bit for the caller to proceed.
595 std::this_thread::sleep_for(rtime: std::chrono::milliseconds(100));
596#endif
597 auto &Wrap = *unwrap(P: c_cas);
598 if (CancelState->Cancelled) {
599 Wrap.syncErrs(Fn: [&](raw_ostream &OS) {
600 OS << "load_object_async cancelled: " << PrintedDigest << '\n';
601 });
602 return passObject(std::nullopt);
603 }
604 Wrap.syncErrs(Fn: [&](raw_ostream &OS) {
605 OS << "load_object_async downstream end: " << PrintedDigest << '\n';
606 });
607 if (Wrap.SimulateMissingObjects)
608 return passObject(std::nullopt);
609 passObject(Wrap.loadObject(ID));
610 });
611}
612
613bool llcas_cas_store_object(llcas_cas_t c_cas, llcas_data_t c_data,
614 const llcas_objectid_t *c_refs, size_t c_refs_count,
615 llcas_objectid_t *c_id_p, char **error) {
616 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
617 SmallVector<ObjectID, 64> Refs;
618 Refs.reserve(N: c_refs_count);
619 for (unsigned I = 0; I != c_refs_count; ++I) {
620 Refs.push_back(Elt: ObjectID::fromOpaqueData(Opaque: c_refs[I].opaque));
621 }
622 ArrayRef Data((const char *)c_data.data, c_data.size);
623
624 SmallVector<ArrayRef<uint8_t>, 8> RefHashes;
625 RefHashes.reserve(N: c_refs_count);
626 for (ObjectID Ref : Refs)
627 RefHashes.push_back(Elt: CAS.getDigest(Ref));
628 HashType Digest = BuiltinObjectHasher<HasherT>::hashObject(Refs: RefHashes, Data);
629 auto StoredID = CAS.getReference(Hash: Digest);
630 if (!StoredID)
631 return reportError(E: StoredID.takeError(), error, Result: true);
632
633 if (Error E = CAS.store(ID: *StoredID, Refs, Data))
634 return reportError(E: std::move(E), error, Result: true);
635 *c_id_p = llcas_objectid_t{.opaque: StoredID->getOpaqueData()};
636 return false;
637}
638
639llcas_data_t llcas_loaded_object_get_data(llcas_cas_t c_cas,
640 llcas_loaded_object_t c_obj) {
641 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
642 ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
643 auto Data = CAS.getObjectData(Node: Obj);
644 return llcas_data_t{.data: Data.data(), .size: Data.size()};
645}
646
647/// The \c MemoryBuffer objects handed out by
648/// \c llcas_loaded_object_get_standalone_data, keyed by the bytes the C API
649/// reports, so \c llcas_standalone_data_dispose can find the owner again. The
650/// C API passes back only the buffer, and these outlive the \c llcas_cas_t, so
651/// they cannot be tracked on it.
652/// Intentionally leaked, since a buffer may be disposed of during static
653/// destruction, after a non-leaked map would already be gone.
654static std::mutex StandaloneBuffersLock;
655static auto *StandaloneBuffers =
656 new DenseMap<const void *, std::unique_ptr<MemoryBuffer>>();
657
658llcas_data_t
659llcas_loaded_object_get_standalone_data(llcas_cas_t c_cas,
660 llcas_loaded_object_t c_obj) {
661 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
662 ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
663 // The underlying database already knows how to produce a buffer that does
664 // not reference it, so use that rather than copying the data again. The
665 // plugin API requires a nul terminator, which costs a copy for the objects
666 // whose file has no byte to spare for one.
667 std::unique_ptr<MemoryBuffer> Buffer = CAS.getStandaloneMemoryBuffer(
668 Node: Obj, /*Name=*/"", /*RequiresNullTerminator=*/true);
669 const char *Data = Buffer->getBufferStart();
670 size_t Size = Buffer->getBufferSize();
671 {
672 std::lock_guard<std::mutex> Lock(StandaloneBuffersLock);
673 (*StandaloneBuffers)[Data] = std::move(Buffer);
674 }
675 return llcas_data_t{.data: Data, .size: Size};
676}
677
678void llcas_standalone_data_dispose(llcas_data_t c_data) {
679 std::lock_guard<std::mutex> Lock(StandaloneBuffersLock);
680 StandaloneBuffers->erase(Val: c_data.data);
681}
682
683llcas_object_refs_t llcas_loaded_object_get_refs(llcas_cas_t c_cas,
684 llcas_loaded_object_t c_obj) {
685 auto &CAS = unwrap(P: c_cas)->DB->getGraphDB();
686 ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
687 auto Refs = CAS.getObjectRefs(Node: Obj);
688 return llcas_object_refs_t{.opaque_b: Refs.begin().getOpaqueData(),
689 .opaque_e: Refs.end().getOpaqueData()};
690}
691
692size_t llcas_object_refs_get_count(llcas_cas_t c_cas,
693 llcas_object_refs_t c_refs) {
694 auto B = object_refs_iterator::fromOpaqueData(Opaque: c_refs.opaque_b);
695 auto E = object_refs_iterator::fromOpaqueData(Opaque: c_refs.opaque_e);
696 return E - B;
697}
698
699llcas_objectid_t llcas_object_refs_get_id(llcas_cas_t c_cas,
700 llcas_object_refs_t c_refs,
701 size_t index) {
702 auto RefsI = object_refs_iterator::fromOpaqueData(Opaque: c_refs.opaque_b);
703 ObjectID Ref = *(RefsI + index);
704 return llcas_objectid_t{.opaque: Ref.getOpaqueData()};
705}
706
707llcas_lookup_result_t
708llcas_actioncache_get_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key,
709 llcas_objectid_t *p_value, bool globally,
710 char **error) {
711 auto &Wrap = *unwrap(P: c_cas);
712 auto &DB = *Wrap.DB;
713 ArrayRef Key(c_key.data, c_key.size);
714 std::optional<ObjectID> Value;
715 if (Error E = cacheGet(DB&: DB.getKeyValueDB(), Key).moveInto(Value))
716 return reportError(E: std::move(E), error, Result: LLCAS_LOOKUP_RESULT_ERROR);
717 if (!Value) {
718 if (!globally)
719 return LLCAS_LOOKUP_RESULT_NOTFOUND;
720
721 if (Error E = Wrap.downstreamKey(Key).moveInto(Value))
722 return reportError(E: std::move(E), error, Result: LLCAS_LOOKUP_RESULT_ERROR);
723 if (!Value)
724 return LLCAS_LOOKUP_RESULT_NOTFOUND;
725 }
726 *p_value = llcas_objectid_t{.opaque: Value->getOpaqueData()};
727 return LLCAS_LOOKUP_RESULT_SUCCESS;
728}
729
730void llcas_actioncache_get_for_digest_async(
731 llcas_cas_t c_cas, llcas_digest_t c_key, bool globally, void *ctx_cb,
732 llcas_actioncache_get_cb cb, llcas_cancellable_t *c_cancellable) {
733 auto CancelState = std::make_shared<CancellableState>();
734 if (c_cancellable) {
735 *c_cancellable = wrap(P: new CancellableWrap{.State: CancelState});
736 }
737 bool IsCancellable = c_cancellable != nullptr;
738
739 ArrayRef Key(c_key.data, c_key.size);
740 SmallVector<uint8_t, 32> KeyBuf(Key);
741
742 unwrap(P: c_cas)->Pool.async(F: [=] {
743 if (IsCancellable) {
744#if LLVM_ENABLE_THREADS
745 // Wait a bit for the caller to have a chance to cancel.
746 std::this_thread::sleep_for(rtime: std::chrono::milliseconds(50));
747#endif
748 }
749 auto &Wrap = *unwrap(P: c_cas);
750 if (CancelState->Cancelled) {
751 Wrap.syncErrs(Fn: [&](raw_ostream &OS) {
752 OS << "actioncache_get_for_digest_async cancelled\n";
753 });
754 return cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_objectid_t(),
755 nullptr);
756 }
757 llcas_objectid_t c_value;
758 char *c_err;
759 llcas_lookup_result_t result = llcas_actioncache_get_for_digest(
760 c_cas, c_key: llcas_digest_t{.data: KeyBuf.data(), .size: KeyBuf.size()}, p_value: &c_value, globally,
761 error: &c_err);
762 cb(ctx_cb, result, c_value, c_err);
763 });
764}
765
766bool llcas_actioncache_put_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key,
767 llcas_objectid_t c_value, bool globally,
768 char **error) {
769 auto &Wrap = *unwrap(P: c_cas);
770 auto &DB = *Wrap.DB;
771 ObjectID Value = ObjectID::fromOpaqueData(Opaque: c_value.opaque);
772 ArrayRef Key(c_key.data, c_key.size);
773 Expected<ObjectID> Ret = cachePut(DB&: DB.getKeyValueDB(), Key, ID: Value);
774 if (!Ret)
775 return reportError(E: Ret.takeError(), error, Result: true);
776 if (*Ret != Value)
777 return reportError(
778 E: createStringError(EC: errc::invalid_argument, S: "cache poisoned"), error,
779 Result: true);
780
781 if (globally) {
782 if (Error E = Wrap.upstreamKey(Key, Value))
783 return reportError(E: std::move(E), error, Result: true);
784 }
785
786 return false;
787}
788
789void llcas_actioncache_put_for_digest_async(
790 llcas_cas_t c_cas, llcas_digest_t c_key, llcas_objectid_t c_value,
791 bool globally, void *ctx_cb, llcas_actioncache_put_cb cb,
792 llcas_cancellable_t *c_cancellable) {
793 auto CancelState = std::make_shared<CancellableState>();
794 if (c_cancellable) {
795 *c_cancellable = wrap(P: new CancellableWrap{.State: CancelState});
796 }
797 bool IsCancellable = c_cancellable != nullptr;
798
799 ArrayRef Key(c_key.data, c_key.size);
800 SmallVector<uint8_t, 32> KeyBuf(Key);
801
802 unwrap(P: c_cas)->Pool.async(F: [=] {
803 if (IsCancellable) {
804#if LLVM_ENABLE_THREADS
805 // Wait a bit for the caller to have a chance to cancel.
806 std::this_thread::sleep_for(rtime: std::chrono::milliseconds(50));
807#endif
808 }
809 auto &Wrap = *unwrap(P: c_cas);
810 if (CancelState->Cancelled) {
811 Wrap.syncErrs(Fn: [&](raw_ostream &OS) {
812 OS << "actioncache_put_for_digest_async cancelled\n";
813 });
814 return cb(ctx_cb, false, nullptr);
815 }
816 char *c_err;
817 bool failed = llcas_actioncache_put_for_digest(
818 c_cas, c_key: llcas_digest_t{.data: KeyBuf.data(), .size: KeyBuf.size()}, c_value, globally,
819 error: &c_err);
820 cb(ctx_cb, failed, c_err);
821 });
822}
823