blob: 282760073962ca650edec10b3636de3957c6d858 [file] [edit]
//===----------------------------------------------------------------------===//
//
// 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);
});
}