| //===----------------------------------------------------------------------===// | |
| // | |
| // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions. | |
| // See https://llvm.org/LICENSE.txt for license information. | |
| // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception | |
| // | |
| //===----------------------------------------------------------------------===// | |
| /// | |
| /// \file | |
| /// Implementation of the LLVM CAS plugin API, for testing purposes. | |
| /// | |
| /// It is backed by \c UnifiedOnDiskCache and can optionally be given a second | |
| /// on-disk path via the \c upstream-path option, which it uses to simulate | |
| /// "uploading"/"downloading" objects to/from a distributed CAS. | |
| /// | |
| //===----------------------------------------------------------------------===// | |
| #include "llvm-c/CAS/PluginAPI_functions.h" | |
| #include "llvm/CAS/BuiltinObjectHasher.h" | |
| #include "llvm/CAS/CASID.h" | |
| #include "llvm/CAS/OnDiskKeyValueDB.h" | |
| #include "llvm/CAS/UnifiedOnDiskCache.h" | |
| #include "llvm/Support/CBindingWrapping.h" | |
| #include "llvm/Support/Errc.h" | |
| #include "llvm/Support/Error.h" | |
| #include "llvm/Support/MemoryBuffer.h" | |
| #include "llvm/Support/SHA1.h" | |
| #include "llvm/Support/ThreadPool.h" | |
| #include <mutex> | |
| using namespace llvm; | |
| using namespace llvm::cas; | |
| using namespace llvm::cas::ondisk; | |
| namespace llvm::cas::ondisk { | |
| /// Declared in the private "OnDiskCommon.h"; see \c setSmallMaxMappingSize. | |
| void setMaxMappingSize(uint64_t Size); | |
| } // namespace llvm::cas::ondisk | |
| /// This plugin exists only for testing, and a test process can create many | |
| /// instances of it. Keep the on-disk mappings small so that they stay cheap; | |
| /// the default sizes are measured in gigabytes per instance. | |
| /// | |
| /// This has to happen inside the plugin: it links its own copy of LLVMCAS, so | |
| /// the setting the test binary applies to itself does not reach us. | |
| static void setSmallMaxMappingSize() { | |
| static std::once_flag Flag; | |
| std::call_once(Flag, [] { setMaxMappingSize(100 * 1024 * 1024); }); | |
| } | |
| static char *copyNewMallocString(StringRef Str) { | |
| char *c_str = (char *)malloc(Str.size() + 1); | |
| std::uninitialized_copy(Str.begin(), Str.end(), c_str); | |
| c_str[Str.size()] = '\0'; | |
| return c_str; | |
| } | |
| template <typename ResT> | |
| static ResT reportError(Error &&E, char **error, ResT Result = ResT()) { | |
| if (error) | |
| *error = copyNewMallocString(toString(std::move(E))); | |
| return Result; | |
| } | |
| void llcas_get_plugin_version(unsigned *major, unsigned *minor) { | |
| *major = LLCAS_VERSION_MAJOR; | |
| *minor = LLCAS_VERSION_MINOR; | |
| } | |
| void llcas_string_dispose(char *str) { free(str); } | |
| namespace { | |
| struct CancellableState { | |
| std::atomic<bool> Cancelled{false}; | |
| }; | |
| struct CancellableWrap { | |
| std::shared_ptr<CancellableState> State; | |
| }; | |
| DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CancellableWrap, llcas_cancellable_t) | |
| } // namespace | |
| void llcas_cancellable_cancel(llcas_cancellable_t c_cancellable) { | |
| unwrap(c_cancellable)->State->Cancelled = true; | |
| } | |
| void llcas_cancellable_dispose(llcas_cancellable_t c_cancellable) { | |
| delete unwrap(c_cancellable); | |
| } | |
| namespace { | |
| struct CASPluginOptions { | |
| std::string OnDiskPath; | |
| std::string UpstreamPath; | |
| std::string FirstPrefix; | |
| std::string SecondPrefix; | |
| bool SimulateMissingObjects = false; | |
| bool Logging = true; | |
| Error setOption(StringRef Name, StringRef Value); | |
| }; | |
| DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASPluginOptions, llcas_cas_options_t) | |
| } // namespace | |
| Error CASPluginOptions::setOption(StringRef Name, StringRef Value) { | |
| if (Name == "first-prefix") | |
| FirstPrefix = Value; | |
| else if (Name == "second-prefix") | |
| SecondPrefix = Value; | |
| else if (Name == "upstream-path") | |
| UpstreamPath = Value; | |
| else if (Name == "simulate-missing-objects") | |
| SimulateMissingObjects = true; | |
| else if (Name == "no-logging") | |
| Logging = false; | |
| else | |
| return createStringError(errc::invalid_argument, | |
| Twine("unknown option: ") + Name); | |
| return Error::success(); | |
| } | |
| llcas_cas_options_t llcas_cas_options_create(void) { | |
| return wrap(new CASPluginOptions()); | |
| } | |
| void llcas_cas_options_dispose(llcas_cas_options_t c_opts) { | |
| delete unwrap(c_opts); | |
| } | |
| void llcas_cas_options_set_ondisk_path(llcas_cas_options_t c_opts, | |
| const char *path) { | |
| auto &Opts = *unwrap(c_opts); | |
| Opts.OnDiskPath = path; | |
| } | |
| bool llcas_cas_options_set_option(llcas_cas_options_t c_opts, const char *name, | |
| const char *value, char **error) { | |
| auto &Opts = *unwrap(c_opts); | |
| if (Error E = Opts.setOption(name, value)) | |
| return reportError(std::move(E), error, true); | |
| return false; | |
| } | |
| namespace { | |
| using HasherT = SHA1; | |
| using HashType = decltype(HasherT::hash(std::declval<ArrayRef<uint8_t> &>())); | |
| class PluginCASContext : public CASContext { | |
| void printIDImpl(raw_ostream &OS, const CASID &ID) const final { | |
| PluginCASContext::printID(ID.getHash(), OS); | |
| } | |
| public: | |
| static StringRef getHashName() { return "SHA1"; } | |
| StringRef getHashSchemaIdentifier() const final { | |
| static const std::string ID = | |
| ("llvm.cas.builtin.v2[" + getHashName() + "]").str(); | |
| return ID; | |
| } | |
| PluginCASContext() = default; | |
| static Expected<HashType> parseID(StringRef Reference) { | |
| if (!Reference.consume_front("llvmcas://")) | |
| return createStringError( | |
| std::make_error_code(std::errc::invalid_argument), | |
| "invalid cas-id '" + Reference + "'"); | |
| if (Reference.size() != 2 * sizeof(HashType)) | |
| return createStringError( | |
| std::make_error_code(std::errc::invalid_argument), | |
| "wrong size for cas-id hash '" + Reference + "'"); | |
| std::string Binary; | |
| if (!tryGetFromHex(Reference, Binary)) | |
| return createStringError( | |
| std::make_error_code(std::errc::invalid_argument), | |
| "invalid hash in cas-id '" + Reference + "'"); | |
| assert(Binary.size() == sizeof(HashType)); | |
| HashType Digest; | |
| llvm::copy(Binary, Digest.data()); | |
| return Digest; | |
| } | |
| static void printID(ArrayRef<uint8_t> Digest, raw_ostream &OS) { | |
| SmallString<64> Hash; | |
| toHex(Digest, /*LowerCase=*/true, Hash); | |
| OS << "llvmcas://" << Hash; | |
| } | |
| }; | |
| struct CASWrapper { | |
| std::string FirstPrefix; | |
| std::string SecondPrefix; | |
| /// If true, asynchronous "download" of an object will treat it as missing. | |
| bool SimulateMissingObjects = false; | |
| bool Logging = true; | |
| std::unique_ptr<UnifiedOnDiskCache> DB; | |
| /// Used for testing the \c globally parameter of action cache APIs. Simulates | |
| /// "uploading"/"downloading" objects from/to the primary on-disk path. | |
| std::unique_ptr<UnifiedOnDiskCache> UpstreamDB; | |
| DefaultThreadPool Pool{llvm::hardware_concurrency()}; | |
| std::mutex Lock{}; | |
| /// Check if the object is contained, in the "local" CAS only or "globally". | |
| bool containsObject(ObjectID ID, bool Globally); | |
| /// Load the object, potentially "downloading" it from upstream. | |
| Expected<std::optional<ondisk::ObjectHandle>> loadObject(ObjectID ID); | |
| /// "Uploads" a key and the associated full node graph. | |
| Error upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value); | |
| /// "Downloads" the ID associated with the key but not the node data. The node | |
| /// itself and the rest of the nodes in the graph will be "downloaded" lazily | |
| /// as they are visited. | |
| Expected<std::optional<ObjectID>> downstreamKey(ArrayRef<uint8_t> Key); | |
| /// Synchronized access to \c llvm::errs(). | |
| void syncErrs(llvm::function_ref<void(raw_ostream &OS)> Fn) { | |
| if (!Logging) { | |
| // Ignore log output. | |
| SmallString<32> Buf; | |
| raw_svector_ostream OS(Buf); | |
| Fn(OS); | |
| return; | |
| } | |
| std::unique_lock<std::mutex> LockGuard(Lock); | |
| Fn(errs()); | |
| errs().flush(); | |
| } | |
| private: | |
| /// "Uploads" the full object node graph. | |
| Expected<ObjectID> upstreamNode(ObjectID Node); | |
| /// "Downloads" only a single object node. The rest of the nodes in the graph | |
| /// will be "downloaded" lazily as they are visited. | |
| Expected<ObjectID> downstreamNode(ObjectID Node); | |
| }; | |
| DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASWrapper, llcas_cas_t) | |
| } // namespace | |
| bool CASWrapper::containsObject(ObjectID ID, bool Globally) { | |
| if (DB->getGraphDB().containsObject(ID)) | |
| return true; | |
| if (!Globally || !UpstreamDB) | |
| return false; | |
| auto UpstreamID = expectedToOptional( | |
| UpstreamDB->getGraphDB().getReference(DB->getGraphDB().getDigest(ID))); | |
| if (!UpstreamID) | |
| return false; | |
| return UpstreamDB->getGraphDB().containsObject(*UpstreamID); | |
| } | |
| Expected<std::optional<ondisk::ObjectHandle>> | |
| CASWrapper::loadObject(ObjectID ID) { | |
| std::optional<ondisk::ObjectHandle> Obj; | |
| if (Error E = DB->getGraphDB().load(ID).moveInto(Obj)) | |
| return std::move(E); | |
| if (Obj) | |
| return Obj; | |
| if (!UpstreamDB) | |
| return std::nullopt; | |
| // Try "downloading" the node from upstream. | |
| auto UpstreamID = | |
| UpstreamDB->getGraphDB().getReference(DB->getGraphDB().getDigest(ID)); | |
| if (!UpstreamID) | |
| return UpstreamID.takeError(); | |
| std::optional<ObjectID> Ret; | |
| if (Error E = downstreamNode(*UpstreamID).moveInto(Ret)) | |
| return std::move(E); | |
| return DB->getGraphDB().load(ID); | |
| } | |
| /// Imports a single object node. | |
| static Expected<ObjectID> importNode(ObjectID FromID, OnDiskGraphDB &FromDB, | |
| OnDiskGraphDB &ToDB) { | |
| auto ToID = ToDB.getReference(FromDB.getDigest(FromID)); | |
| if (!ToID) | |
| return ToID.takeError(); | |
| if (ToDB.containsObject(*ToID)) | |
| return ToID; | |
| std::optional<ondisk::ObjectHandle> FromH; | |
| if (Error E = FromDB.load(FromID).moveInto(FromH)) | |
| return std::move(E); | |
| if (!FromH) | |
| return ToID; | |
| auto Data = FromDB.getObjectData(*FromH); | |
| auto FromRefs = FromDB.getObjectRefs(*FromH); | |
| SmallVector<ObjectID> Refs; | |
| for (ObjectID FromRef : FromRefs) { | |
| auto Ref = ToDB.getReference(FromDB.getDigest(FromRef)); | |
| if (!Ref) | |
| return Ref.takeError(); | |
| Refs.push_back(*Ref); | |
| } | |
| if (Error E = ToDB.store(*ToID, Refs, Data)) | |
| return std::move(E); | |
| return ToID; | |
| } | |
| Expected<ObjectID> CASWrapper::upstreamNode(ObjectID Node) { | |
| OnDiskGraphDB &FromDB = DB->getGraphDB(); | |
| OnDiskGraphDB &ToDB = UpstreamDB->getGraphDB(); | |
| std::optional<ondisk::ObjectHandle> FromH; | |
| if (Error E = FromDB.load(Node).moveInto(FromH)) | |
| return std::move(E); | |
| if (!FromH) | |
| return createStringError(errc::invalid_argument, "node doesn't exist"); | |
| for (ObjectID Ref : FromDB.getObjectRefs(*FromH)) { | |
| std::optional<ObjectID> ID; | |
| if (Error E = upstreamNode(Ref).moveInto(ID)) | |
| return std::move(E); | |
| } | |
| return importNode(Node, FromDB, ToDB); | |
| } | |
| Expected<ObjectID> CASWrapper::downstreamNode(ObjectID Node) { | |
| OnDiskGraphDB &FromDB = UpstreamDB->getGraphDB(); | |
| OnDiskGraphDB &ToDB = DB->getGraphDB(); | |
| return importNode(Node, FromDB, ToDB); | |
| } | |
| static Expected<ObjectID> cachePut(OnDiskKeyValueDB &DB, ArrayRef<uint8_t> Key, | |
| ObjectID ID) { | |
| auto Value = UnifiedOnDiskCache::getValueFromObjectID(ID); | |
| auto Result = DB.put(Key, Value); | |
| if (!Result) | |
| return Result.takeError(); | |
| return UnifiedOnDiskCache::getObjectIDFromValue(*Result); | |
| } | |
| static Expected<std::optional<ObjectID>> cacheGet(OnDiskKeyValueDB &DB, | |
| ArrayRef<uint8_t> Key) { | |
| auto Result = DB.get(Key); | |
| if (!Result) | |
| return Result.takeError(); | |
| if (!*Result) | |
| return std::nullopt; | |
| return UnifiedOnDiskCache::getObjectIDFromValue(**Result); | |
| } | |
| Error CASWrapper::upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value) { | |
| if (!UpstreamDB) | |
| return Error::success(); | |
| Expected<ObjectID> UpstreamVal = upstreamNode(Value); | |
| if (!UpstreamVal) | |
| return UpstreamVal.takeError(); | |
| Expected<ObjectID> PutValue = | |
| cachePut(UpstreamDB->getKeyValueDB(), Key, *UpstreamVal); | |
| if (!PutValue) | |
| return PutValue.takeError(); | |
| assert(*PutValue == *UpstreamVal); | |
| return Error::success(); | |
| } | |
| Expected<std::optional<ObjectID>> | |
| CASWrapper::downstreamKey(ArrayRef<uint8_t> Key) { | |
| if (!UpstreamDB) | |
| return std::nullopt; | |
| std::optional<ObjectID> UpstreamValue; | |
| if (Error E = | |
| cacheGet(UpstreamDB->getKeyValueDB(), Key).moveInto(UpstreamValue)) | |
| return std::move(E); | |
| if (!UpstreamValue) | |
| return std::nullopt; | |
| auto Value = DB->getGraphDB().getReference( | |
| UpstreamDB->getGraphDB().getDigest(*UpstreamValue)); | |
| if (!Value) | |
| return Value.takeError(); | |
| Expected<ObjectID> PutValue = cachePut(DB->getKeyValueDB(), Key, *Value); | |
| if (!PutValue) | |
| return PutValue.takeError(); | |
| assert(*PutValue == *Value); | |
| return PutValue; | |
| } | |
| llcas_cas_t llcas_cas_create(llcas_cas_options_t c_opts, char **error) { | |
| auto &Opts = *unwrap(c_opts); | |
| setSmallMaxMappingSize(); | |
| Expected<std::unique_ptr<UnifiedOnDiskCache>> DB = UnifiedOnDiskCache::open( | |
| Opts.OnDiskPath, /*SizeLimit=*/std::nullopt, | |
| PluginCASContext::getHashName(), sizeof(HashType)); | |
| if (!DB) | |
| return reportError<llcas_cas_t>(DB.takeError(), error); | |
| std::unique_ptr<UnifiedOnDiskCache> UpstreamDB; | |
| if (!Opts.UpstreamPath.empty()) { | |
| if (Error E = UnifiedOnDiskCache::open( | |
| Opts.UpstreamPath, /*SizeLimit=*/std::nullopt, | |
| PluginCASContext::getHashName(), sizeof(HashType)) | |
| .moveInto(UpstreamDB)) | |
| return reportError<llcas_cas_t>(std::move(E), error); | |
| } | |
| return wrap(new CASWrapper{Opts.FirstPrefix, Opts.SecondPrefix, | |
| Opts.SimulateMissingObjects, Opts.Logging, | |
| std::move(*DB), std::move(UpstreamDB)}); | |
| } | |
| void llcas_cas_dispose(llcas_cas_t c_cas) { delete unwrap(c_cas); } | |
| int64_t llcas_cas_get_ondisk_size(llcas_cas_t c_cas, char **error) { | |
| return unwrap(c_cas)->DB->getStorageSize(); | |
| } | |
| bool llcas_cas_set_ondisk_size_limit(llcas_cas_t c_cas, int64_t size_limit, | |
| char **error) { | |
| std::optional<uint64_t> SizeLimit; | |
| if (size_limit < 0) { | |
| return reportError( | |
| llvm::createStringError( | |
| llvm::inconvertibleErrorCode(), | |
| "invalid size limit passed to llcas_cas_set_ondisk_size_limit"), | |
| error, true); | |
| } | |
| if (size_limit > 0) { | |
| SizeLimit = size_limit; | |
| } | |
| unwrap(c_cas)->DB->setSizeLimit(SizeLimit); | |
| return false; | |
| } | |
| bool llcas_cas_prune_ondisk_data(llcas_cas_t c_cas, char **error) { | |
| if (Error E = unwrap(c_cas)->DB->collectGarbage()) | |
| return reportError(std::move(E), error, true); | |
| return false; | |
| } | |
| void llcas_cas_options_set_client_version(llcas_cas_options_t, unsigned major, | |
| unsigned minor) { | |
| // Ignore for now. | |
| } | |
| char *llcas_cas_get_hash_schema_name(llcas_cas_t) { | |
| // Using same name as builtin CAS so that it's interchangeable for testing | |
| // purposes. | |
| return copyNewMallocString("llvm.cas.builtin.v2[BLAKE3]"); | |
| } | |
| unsigned llcas_digest_parse(llcas_cas_t c_cas, const char *printed_digest, | |
| uint8_t *bytes, size_t bytes_size, char **error) { | |
| auto &Wrapper = *unwrap(c_cas); | |
| if (bytes_size < sizeof(HashType)) | |
| return sizeof(HashType); | |
| StringRef PrintedDigest = printed_digest; | |
| bool Consumed = PrintedDigest.consume_front(Wrapper.FirstPrefix); | |
| assert(Consumed); | |
| (void)Consumed; | |
| Consumed = PrintedDigest.consume_front(Wrapper.SecondPrefix); | |
| assert(Consumed); | |
| (void)Consumed; | |
| Expected<HashType> Digest = PluginCASContext::parseID(PrintedDigest); | |
| if (!Digest) | |
| return reportError(Digest.takeError(), error, 0); | |
| std::uninitialized_copy(Digest->begin(), Digest->end(), bytes); | |
| return Digest->size(); | |
| } | |
| bool llcas_digest_print(llcas_cas_t c_cas, llcas_digest_t c_digest, | |
| char **printed_id, char **error) { | |
| auto &Wrapper = *unwrap(c_cas); | |
| SmallString<74> PrintDigest; | |
| raw_svector_ostream OS(PrintDigest); | |
| // Include these for testing purposes. | |
| OS << Wrapper.FirstPrefix << Wrapper.SecondPrefix; | |
| PluginCASContext::printID(ArrayRef(c_digest.data, c_digest.size), OS); | |
| *printed_id = copyNewMallocString(PrintDigest); | |
| return false; | |
| } | |
| bool llcas_cas_get_objectid(llcas_cas_t c_cas, llcas_digest_t c_digest, | |
| llcas_objectid_t *c_id_p, char **error) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| auto ID = CAS.getReference(ArrayRef(c_digest.data, c_digest.size)); | |
| if (!ID) | |
| return reportError(ID.takeError(), error, true); | |
| *c_id_p = llcas_objectid_t{ID->getOpaqueData()}; | |
| return false; | |
| } | |
| llcas_digest_t llcas_objectid_get_digest(llcas_cas_t c_cas, | |
| llcas_objectid_t c_id) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque); | |
| ArrayRef<uint8_t> Digest = CAS.getDigest(ID); | |
| return llcas_digest_t{Digest.data(), Digest.size()}; | |
| } | |
| llcas_lookup_result_t llcas_cas_contains_object(llcas_cas_t c_cas, | |
| llcas_objectid_t c_id, | |
| bool globally, char **error) { | |
| ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque); | |
| return unwrap(c_cas)->containsObject(ID, globally) | |
| ? LLCAS_LOOKUP_RESULT_SUCCESS | |
| : LLCAS_LOOKUP_RESULT_NOTFOUND; | |
| } | |
| llcas_lookup_result_t llcas_cas_load_object(llcas_cas_t c_cas, | |
| llcas_objectid_t c_id, | |
| llcas_loaded_object_t *c_obj_p, | |
| char **error) { | |
| ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque); | |
| Expected<std::optional<ondisk::ObjectHandle>> ObjOpt = | |
| unwrap(c_cas)->loadObject(ID); | |
| if (!ObjOpt) | |
| return reportError(ObjOpt.takeError(), error, LLCAS_LOOKUP_RESULT_ERROR); | |
| if (!*ObjOpt) | |
| return LLCAS_LOOKUP_RESULT_NOTFOUND; | |
| ondisk::ObjectHandle Obj = **ObjOpt; | |
| *c_obj_p = llcas_loaded_object_t{Obj.getOpaqueData()}; | |
| return LLCAS_LOOKUP_RESULT_SUCCESS; | |
| } | |
| void llcas_cas_load_object_async(llcas_cas_t c_cas, llcas_objectid_t c_id, | |
| void *ctx_cb, llcas_cas_load_object_cb cb, | |
| llcas_cancellable_t *c_cancellable) { | |
| auto CancelState = std::make_shared<CancellableState>(); | |
| if (c_cancellable) { | |
| *c_cancellable = wrap(new CancellableWrap{CancelState}); | |
| } | |
| std::string PrintedDigest; | |
| { | |
| llcas_digest_t c_digest = llcas_objectid_get_digest(c_cas, c_id); | |
| char *printed_id; | |
| char *c_err; | |
| bool failed = llcas_digest_print(c_cas, c_digest, &printed_id, &c_err); | |
| if (failed) | |
| report_fatal_error(Twine("digest printing failed: ") + c_err); | |
| PrintedDigest = printed_id; | |
| llcas_string_dispose(printed_id); | |
| } | |
| auto passObject = [ctx_cb, | |
| cb](Expected<std::optional<ondisk::ObjectHandle>> Obj) { | |
| if (!Obj) { | |
| cb(ctx_cb, LLCAS_LOOKUP_RESULT_ERROR, llcas_loaded_object_t(), | |
| copyNewMallocString(toString(Obj.takeError()))); | |
| } else if (!*Obj) { | |
| cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_loaded_object_t(), | |
| nullptr); | |
| } else { | |
| cb(ctx_cb, LLCAS_LOOKUP_RESULT_SUCCESS, | |
| llcas_loaded_object_t{(*Obj)->getOpaqueData()}, nullptr); | |
| } | |
| }; | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque); | |
| if (CAS.containsObject(ID)) { | |
| unwrap(c_cas)->syncErrs([&](raw_ostream &OS) { | |
| OS << "load_object_async existing: " << PrintedDigest << '\n'; | |
| }); | |
| return passObject(unwrap(c_cas)->loadObject(ID)); | |
| } | |
| if (!unwrap(c_cas)->UpstreamDB) | |
| return passObject(std::nullopt); | |
| // Try "downloading" the node from upstream. | |
| unwrap(c_cas)->syncErrs([&](raw_ostream &OS) { | |
| OS << "load_object_async downstream begin: " << PrintedDigest << '\n'; | |
| }); | |
| unwrap(c_cas)->Pool.async([=] { | |
| #if LLVM_ENABLE_THREADS | |
| // Wait a bit for the caller to proceed. | |
| std::this_thread::sleep_for(std::chrono::milliseconds(100)); | |
| #endif | |
| auto &Wrap = *unwrap(c_cas); | |
| if (CancelState->Cancelled) { | |
| Wrap.syncErrs([&](raw_ostream &OS) { | |
| OS << "load_object_async cancelled: " << PrintedDigest << '\n'; | |
| }); | |
| return passObject(std::nullopt); | |
| } | |
| Wrap.syncErrs([&](raw_ostream &OS) { | |
| OS << "load_object_async downstream end: " << PrintedDigest << '\n'; | |
| }); | |
| if (Wrap.SimulateMissingObjects) | |
| return passObject(std::nullopt); | |
| passObject(Wrap.loadObject(ID)); | |
| }); | |
| } | |
| bool llcas_cas_store_object(llcas_cas_t c_cas, llcas_data_t c_data, | |
| const llcas_objectid_t *c_refs, size_t c_refs_count, | |
| llcas_objectid_t *c_id_p, char **error) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| SmallVector<ObjectID, 64> Refs; | |
| Refs.reserve(c_refs_count); | |
| for (unsigned I = 0; I != c_refs_count; ++I) { | |
| Refs.push_back(ObjectID::fromOpaqueData(c_refs[I].opaque)); | |
| } | |
| ArrayRef Data((const char *)c_data.data, c_data.size); | |
| SmallVector<ArrayRef<uint8_t>, 8> RefHashes; | |
| RefHashes.reserve(c_refs_count); | |
| for (ObjectID Ref : Refs) | |
| RefHashes.push_back(CAS.getDigest(Ref)); | |
| HashType Digest = BuiltinObjectHasher<HasherT>::hashObject(RefHashes, Data); | |
| auto StoredID = CAS.getReference(Digest); | |
| if (!StoredID) | |
| return reportError(StoredID.takeError(), error, true); | |
| if (Error E = CAS.store(*StoredID, Refs, Data)) | |
| return reportError(std::move(E), error, true); | |
| *c_id_p = llcas_objectid_t{StoredID->getOpaqueData()}; | |
| return false; | |
| } | |
| llcas_data_t llcas_loaded_object_get_data(llcas_cas_t c_cas, | |
| llcas_loaded_object_t c_obj) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque); | |
| auto Data = CAS.getObjectData(Obj); | |
| return llcas_data_t{Data.data(), Data.size()}; | |
| } | |
| /// The \c MemoryBuffer objects handed out by | |
| /// \c llcas_loaded_object_get_standalone_data, keyed by the bytes the C API | |
| /// reports, so \c llcas_standalone_data_dispose can find the owner again. The | |
| /// C API passes back only the buffer, and these outlive the \c llcas_cas_t, so | |
| /// they cannot be tracked on it. | |
| /// Intentionally leaked, since a buffer may be disposed of during static | |
| /// destruction, after a non-leaked map would already be gone. | |
| static std::mutex StandaloneBuffersLock; | |
| static auto *StandaloneBuffers = | |
| new DenseMap<const void *, std::unique_ptr<MemoryBuffer>>(); | |
| llcas_data_t | |
| llcas_loaded_object_get_standalone_data(llcas_cas_t c_cas, | |
| llcas_loaded_object_t c_obj) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque); | |
| // The underlying database already knows how to produce a buffer that does | |
| // not reference it, so use that rather than copying the data again. The | |
| // plugin API requires a nul terminator, which costs a copy for the objects | |
| // whose file has no byte to spare for one. | |
| std::unique_ptr<MemoryBuffer> Buffer = CAS.getStandaloneMemoryBuffer( | |
| Obj, /*Name=*/"", /*RequiresNullTerminator=*/true); | |
| const char *Data = Buffer->getBufferStart(); | |
| size_t Size = Buffer->getBufferSize(); | |
| { | |
| std::lock_guard<std::mutex> Lock(StandaloneBuffersLock); | |
| (*StandaloneBuffers)[Data] = std::move(Buffer); | |
| } | |
| return llcas_data_t{Data, Size}; | |
| } | |
| void llcas_standalone_data_dispose(llcas_data_t c_data) { | |
| std::lock_guard<std::mutex> Lock(StandaloneBuffersLock); | |
| StandaloneBuffers->erase(c_data.data); | |
| } | |
| llcas_object_refs_t llcas_loaded_object_get_refs(llcas_cas_t c_cas, | |
| llcas_loaded_object_t c_obj) { | |
| auto &CAS = unwrap(c_cas)->DB->getGraphDB(); | |
| ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque); | |
| auto Refs = CAS.getObjectRefs(Obj); | |
| return llcas_object_refs_t{Refs.begin().getOpaqueData(), | |
| Refs.end().getOpaqueData()}; | |
| } | |
| size_t llcas_object_refs_get_count(llcas_cas_t c_cas, | |
| llcas_object_refs_t c_refs) { | |
| auto B = object_refs_iterator::fromOpaqueData(c_refs.opaque_b); | |
| auto E = object_refs_iterator::fromOpaqueData(c_refs.opaque_e); | |
| return E - B; | |
| } | |
| llcas_objectid_t llcas_object_refs_get_id(llcas_cas_t c_cas, | |
| llcas_object_refs_t c_refs, | |
| size_t index) { | |
| auto RefsI = object_refs_iterator::fromOpaqueData(c_refs.opaque_b); | |
| ObjectID Ref = *(RefsI + index); | |
| return llcas_objectid_t{Ref.getOpaqueData()}; | |
| } | |
| llcas_lookup_result_t | |
| llcas_actioncache_get_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key, | |
| llcas_objectid_t *p_value, bool globally, | |
| char **error) { | |
| auto &Wrap = *unwrap(c_cas); | |
| auto &DB = *Wrap.DB; | |
| ArrayRef Key(c_key.data, c_key.size); | |
| std::optional<ObjectID> Value; | |
| if (Error E = cacheGet(DB.getKeyValueDB(), Key).moveInto(Value)) | |
| return reportError(std::move(E), error, LLCAS_LOOKUP_RESULT_ERROR); | |
| if (!Value) { | |
| if (!globally) | |
| return LLCAS_LOOKUP_RESULT_NOTFOUND; | |
| if (Error E = Wrap.downstreamKey(Key).moveInto(Value)) | |
| return reportError(std::move(E), error, LLCAS_LOOKUP_RESULT_ERROR); | |
| if (!Value) | |
| return LLCAS_LOOKUP_RESULT_NOTFOUND; | |
| } | |
| *p_value = llcas_objectid_t{Value->getOpaqueData()}; | |
| return LLCAS_LOOKUP_RESULT_SUCCESS; | |
| } | |
| void llcas_actioncache_get_for_digest_async( | |
| llcas_cas_t c_cas, llcas_digest_t c_key, bool globally, void *ctx_cb, | |
| llcas_actioncache_get_cb cb, llcas_cancellable_t *c_cancellable) { | |
| auto CancelState = std::make_shared<CancellableState>(); | |
| if (c_cancellable) { | |
| *c_cancellable = wrap(new CancellableWrap{CancelState}); | |
| } | |
| bool IsCancellable = c_cancellable != nullptr; | |
| ArrayRef Key(c_key.data, c_key.size); | |
| SmallVector<uint8_t, 32> KeyBuf(Key); | |
| unwrap(c_cas)->Pool.async([=] { | |
| if (IsCancellable) { | |
| #if LLVM_ENABLE_THREADS | |
| // Wait a bit for the caller to have a chance to cancel. | |
| std::this_thread::sleep_for(std::chrono::milliseconds(50)); | |
| #endif | |
| } | |
| auto &Wrap = *unwrap(c_cas); | |
| if (CancelState->Cancelled) { | |
| Wrap.syncErrs([&](raw_ostream &OS) { | |
| OS << "actioncache_get_for_digest_async cancelled\n"; | |
| }); | |
| return cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_objectid_t(), | |
| nullptr); | |
| } | |
| llcas_objectid_t c_value; | |
| char *c_err; | |
| llcas_lookup_result_t result = llcas_actioncache_get_for_digest( | |
| c_cas, llcas_digest_t{KeyBuf.data(), KeyBuf.size()}, &c_value, globally, | |
| &c_err); | |
| cb(ctx_cb, result, c_value, c_err); | |
| }); | |
| } | |
| bool llcas_actioncache_put_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key, | |
| llcas_objectid_t c_value, bool globally, | |
| char **error) { | |
| auto &Wrap = *unwrap(c_cas); | |
| auto &DB = *Wrap.DB; | |
| ObjectID Value = ObjectID::fromOpaqueData(c_value.opaque); | |
| ArrayRef Key(c_key.data, c_key.size); | |
| Expected<ObjectID> Ret = cachePut(DB.getKeyValueDB(), Key, Value); | |
| if (!Ret) | |
| return reportError(Ret.takeError(), error, true); | |
| if (*Ret != Value) | |
| return reportError( | |
| createStringError(errc::invalid_argument, "cache poisoned"), error, | |
| true); | |
| if (globally) { | |
| if (Error E = Wrap.upstreamKey(Key, Value)) | |
| return reportError(std::move(E), error, true); | |
| } | |
| return false; | |
| } | |
| void llcas_actioncache_put_for_digest_async( | |
| llcas_cas_t c_cas, llcas_digest_t c_key, llcas_objectid_t c_value, | |
| bool globally, void *ctx_cb, llcas_actioncache_put_cb cb, | |
| llcas_cancellable_t *c_cancellable) { | |
| auto CancelState = std::make_shared<CancellableState>(); | |
| if (c_cancellable) { | |
| *c_cancellable = wrap(new CancellableWrap{CancelState}); | |
| } | |
| bool IsCancellable = c_cancellable != nullptr; | |
| ArrayRef Key(c_key.data, c_key.size); | |
| SmallVector<uint8_t, 32> KeyBuf(Key); | |
| unwrap(c_cas)->Pool.async([=] { | |
| if (IsCancellable) { | |
| #if LLVM_ENABLE_THREADS | |
| // Wait a bit for the caller to have a chance to cancel. | |
| std::this_thread::sleep_for(std::chrono::milliseconds(50)); | |
| #endif | |
| } | |
| auto &Wrap = *unwrap(c_cas); | |
| if (CancelState->Cancelled) { | |
| Wrap.syncErrs([&](raw_ostream &OS) { | |
| OS << "actioncache_put_for_digest_async cancelled\n"; | |
| }); | |
| return cb(ctx_cb, false, nullptr); | |
| } | |
| char *c_err; | |
| bool failed = llcas_actioncache_put_for_digest( | |
| c_cas, llcas_digest_t{KeyBuf.data(), KeyBuf.size()}, c_value, globally, | |
| &c_err); | |
| cb(ctx_cb, failed, c_err); | |
| }); | |
| } |