| //===----------------------------------------------------------------------===// |
| // |
| // 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); |
| }); |
| } |