Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 43 additions & 5 deletions src/server/compiler/indexer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
#include <algorithm>
#include <cassert>
#include <format>
#include <memory>
#include <optional>
#include <string>
#include <vector>
Expand Down Expand Up @@ -532,12 +533,48 @@ kota::task<> Indexer::index_one(std::uint32_t server_path_id,
co_return;

workspace.fill_pcm_deps(params.pcms);
params.index_inline_limit = workspace.config.project.index_inline_limit.value;

// Deliberately never committed: the PendingEntry removes the transfer
// file on every master-side exit path, and a write orphaned by
// cancellation is swept with the instance tmp dir.
std::optional<CacheStore::PendingEntry> transfer;
if(workspace.store) {
transfer = workspace.store->begin_store("index", shard_key(file_path));
params.index_output_path = transfer->tmp_path;
}

LOG_INFO("[{}/{}] Indexing {}", index, total, file_path);

ScopedTimer timer;
auto result = co_await pool.send_stateless(params);
if(result.has_value() && result.value().success && !result.value().tu_index_data.empty()) {

llvm::StringRef tu_index_bytes;
std::unique_ptr<llvm::MemoryBuffer> spilled;
if(result.has_value() && result.value().success) {
tu_index_bytes = result.value().tu_index_data;
if(tu_index_bytes.empty() && result.value().tu_index_file_size > 0 && transfer) {
auto buffer = llvm::MemoryBuffer::getFile(transfer->tmp_path);
if(!buffer || (*buffer)->getBufferSize() != result.value().tu_index_file_size ||
llvm::xxh3_64bits((*buffer)->getBuffer()) != result.value().tu_index_hash) {
// A transfer loss is a transport failure, not a build
// verdict: the work itself is fine (a worker died
// mid-write, or the cache dir was wiped mid-flight), so
// requeue like a crash instead of serving the stale shard
// as fresh until the file's next edit.
LOG_WARN("[{}/{}] Requeueing {}: corrupt TUIndex transfer",
index,
total,
file_path);
note_dispatch_failure(server_path_id, ticket, false);
co_return;
}
spilled = std::move(*buffer);
tu_index_bytes = spilled->getBuffer();
}
}

if(result.has_value() && result.value().success && !tu_index_bytes.empty()) {
auto index_ms = timer.ms();
// Merge guard: a newer content-level invalidation during this build
// (or a removal clearing the entry) means this result describes text
Expand All @@ -553,18 +590,19 @@ kota::task<> Indexer::index_one(std::uint32_t server_path_id,
co_return;
}
ScopedTimer merge_timer;
merge(result.value().tu_index_data.data(), result.value().tu_index_data.size());
merge(tu_index_bytes.data(), tu_index_bytes.size());
LOG_PERF("index",
"progress={}/{} file={} bytes={} index_ms={} merge_ms={}",
"progress={}/{} file={} bytes={} transfer={} index_ms={} merge_ms={}",
index,
total,
file_path,
result.value().tu_index_data.size(),
tu_index_bytes.size(),
spilled ? "file" : "inline",
index_ms,
merge_timer.ms());
} else if(result.has_value() && !result.value().success) {
LOG_WARN("[{}/{}] Index failed for {}: {}", index, total, file_path, result.value().error);
} else if(result.has_value() && result.value().tu_index_data.empty()) {
} else if(result.has_value()) {
LOG_WARN("[{}/{}] Index returned empty TUIndex for {}", index, total, file_path);
} else if(result.error().code == worker::dispatch_errc::cancelled ||
result.error().code == worker::dispatch_errc::worker_crashed ||
Expand Down
18 changes: 15 additions & 3 deletions src/server/protocol/worker.h
Original file line number Diff line number Diff line change
Expand Up @@ -185,9 +185,9 @@ enum class BuildKind : uint8_t {
/// Unified parameters for all stateless build/compilation tasks.
/// Fields are used selectively based on `kind`:
/// - All: file, directory, arguments
/// - BuildPCH: + content, preamble_bound, output_path
/// - BuildPCH: + content, preamble_bound, output_path, index_output_path
/// - BuildPCM: + module_name, pcms, output_path
/// - Index: + pcms
/// - Index: + pcms, index_output_path, index_inline_limit
/// - Completion: + text, version, offset, pch, pcms
/// - SignatureHelp: + text, version, offset, pch, pcms
/// - Format: + text, format_range (optional)
Expand All @@ -214,8 +214,14 @@ struct BuildParams {
/// `.pch.idx`), allocated by the master's store alongside output_path.
/// The worker serializes the preamble's index and feature state into
/// it; the master commits both blobs together.
/// Index: tmp path for a serialized TUIndex too large to travel inline
/// in the IPC frame (see BuildResult::tu_index_file_size).
std::string index_output_path;

/// Index: serialized results above this many bytes spill to
/// index_output_path instead of travelling inline.
std::uint64_t index_inline_limit = 8 * 1024 * 1024;

std::string module_name; ///< BuildPCM
uint32_t preamble_bound = UINT32_MAX; ///< BuildPCH
LocalSourceRange format_range; ///< Format (default = full document)
Expand All @@ -242,7 +248,13 @@ struct BuildResult {
/// whose mtime is past this moment may differ from what the build read.
std::int64_t build_at = 0;
std::vector<DepFile> deps;
std::string tu_index_data; ///< Index: serialized TUIndex, merged by the master
std::string tu_index_data; ///< Index: serialized TUIndex, merged by the master
/// Index: set instead of tu_index_data when the serialized TUIndex was
/// spilled to BuildParams::index_output_path (too large for the IPC
/// frame): blob byte size and its xxh3_64, checked by the master before
/// use — the file may be torn if the worker died mid-write.
std::uint64_t tu_index_file_size = 0;
std::uint64_t tu_index_hash = 0;
kota::codec::RawValue result_json; ///< Completion/SignatureHelp result
};

Expand Down
8 changes: 8 additions & 0 deletions src/server/state/config.h
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,14 @@ struct ProjectConfig {
"harness.")
<bool> test_hooks = false;

/// Test-facing, like test_hooks: the default only matters for results
/// large enough that CI cannot produce them.
KOTATSU_ANNOTATE(defaulted = true,
description =
"Serialized TUIndex results above this many bytes are "
"spilled to a tmp file instead of sent inline.")
<std::uint64_t> index_inline_limit = 8 * 1024 * 1024;

KOTATSU_ANNOTATE(defaulted = true, description = "Number of stateful workers.")
<std::uint32_t> stateful_worker_count = 2;

Expand Down
15 changes: 15 additions & 0 deletions src/server/worker/stateless_worker.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
#include "kota/ipc/peer.h"
#include "kota/ipc/transport.h"
#include "llvm/Support/raw_ostream.h"
#include "llvm/Support/xxhash.h"

namespace clice {

Expand Down Expand Up @@ -278,6 +279,20 @@ static worker::BuildResult handle_index(const worker::BuildParams& params,
timer.ms());
worker::BuildResult result;
result.success = true;

// The master verifies size+hash before use, so a torn write is detected
// there and needs no atomic-rename dance here.
if(serialized.size() > params.index_inline_limit && !params.index_output_path.empty()) {
auto written = fs::write(params.index_output_path, serialized);
if(written) {
result.tu_index_file_size = serialized.size();
result.tu_index_hash = llvm::xxh3_64bits(serialized);
return result;
}
LOG_WARN("Index spill to {} failed ({}), sending inline",
params.index_output_path,
written.error().message());
}
result.tu_index_data = std::move(serialized);
return result;
}
Expand Down
8 changes: 8 additions & 0 deletions src/support/filesystem.h
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,14 @@ inline std::expected<void, std::error_code> write(llvm::StringRef path, llvm::St
}
os << content;
os.flush();
// A mid-write failure (disk full, EIO) lands in the stream's error
// flag, and an unchecked flag turns into a fatal error in the
// destructor — surface it as an ordinary error instead.
if(os.has_error()) {
EC = os.error();
os.clear_error();
return std::unexpected(EC);
}
return std::expected<void, std::error_code>();
}

Expand Down
70 changes: 70 additions & 0 deletions tests/integration/server/index_spill.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
/// Oversized TUIndex results travel via a store tmp file instead of the
/// IPC frame. With a tiny inline limit every background-index result
/// spills; cross-file references prove the master verified and merged the
/// spilled bytes, and the transfer files must not outlive their requests.

import * as fs from "node:fs";
import * as path from "node:path";

import { sleep, type CliceClient } from "@clice/tools/client";
import type { StatsResult } from "@clice/tools/protocol";
import { expect, test } from "../fixtures.ts";

/// Poll clice/internal/stats until predicate(stats) holds.
async function waitStats(
client: CliceClient,
predicate: (stats: StatsResult) => boolean,
message = "",
): Promise<StatsResult> {
const deadline = Date.now() + 30_000;
for (;;) {
const stats = await client.stats();
if (predicate(stats)) {
return stats;
}
if (Date.now() > deadline) {
throw new Error(`${message || "stats condition"} not met: ${JSON.stringify(stats)}`);
}
await sleep(200);
}
}

function masterLog(logsDir: string): string {
if (!fs.existsSync(logsDir)) {
return "";
}
return fs
.readdirSync(logsDir, { recursive: true, encoding: "utf8" })
.filter((name) => path.basename(name) === "master.log")
.map((name) => fs.readFileSync(path.join(logsDir, name), "utf8"))
.join("\n");
}

test("index spill roundtrip", async ({ session }) => {
const { client, workspace } = session.tmp();
// The reference probe excludes declarations, so lib.cpp needs a real
// call site for it to show up in the reference list.
workspace.write(
"lib.cpp",
"int shared_fn() { return 42; }\nint call_it() { return shared_fn(); }\n",
);
workspace.write("main.cpp", "int shared_fn();\nint main() { return shared_fn(); }\n");
workspace.writeCDB(["lib.cpp", "main.cpp"]);
workspace.pinCacheDir();
await client.initialize(workspace, {
initializationOptions: {
project: { idle_timeout_ms: 0, index_inline_limit: 1 },
},
});

const [uri] = await client.openAndWait("main.cpp");
expect(
await client.waitForReference(uri, 0, 4, workspace.uri("lib.cpp")),
"cross-file reference through spilled index",
).toBe(true);

expect(masterLog(workspace.path(".clice/logs"))).toContain("transfer=file");

await waitStats(client, (s) => s.pendingTmpFiles === 0, "transfer files leaked");
expect(workspace.tmpFiles(), "tmp directory should be empty after settling").toEqual([]);
});
81 changes: 81 additions & 0 deletions tests/unit/server/stateless_worker_tests.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@
#include "test/test.h"
#include "server/protocol/worker.h"
#include "server/worker_test_helpers.h"
#include "support/filesystem.h"

#include "kota/codec/bincode/bincode.h"
#include "llvm/Support/xxhash.h"

namespace clice::testing {

Expand Down Expand Up @@ -133,9 +135,88 @@ TEST_CASE(IndexRequest) {
params.file = src;
params.directory = "/tmp";
params.arguments = make_args(src);
params.index_output_path = tmp.path("spill.bin");

auto result = co_await w.peer->send_request(params);
EXPECT_TRUE(result.has_value());
if(result.has_value()) {
// A small index stays inline even with a spill path offered.
EXPECT_TRUE(result->success);
EXPECT_FALSE(result->tu_index_data.empty());
EXPECT_EQ(result->tu_index_file_size, std::uint64_t(0));
}
test_done = true;
w.peer->close_output();
});

ASSERT_TRUE(test_done);
}

TEST_CASE(IndexSpillToFile) {
TempDir tmp;
tmp.touch("test_index.cpp", "int indexed_var = 1;\n");
auto src = tmp.path("test_index.cpp");
auto spill = tmp.path("spill.bin");

WorkerHandle w;
ASSERT_TRUE(w.spawn());

bool test_done = false;

w.run([&]() -> kota::task<> {
worker::BuildParams params;
params.kind = worker::BuildKind::Index;
params.file = src;
params.directory = "/tmp";
params.arguments = make_args(src);
params.index_output_path = spill;
params.index_inline_limit = 1;

auto result = co_await w.peer->send_request(params);
EXPECT_TRUE(result.has_value());
if(result.has_value()) {
EXPECT_TRUE(result->success);
EXPECT_TRUE(result->tu_index_data.empty());
auto blob = fs::read(spill);
EXPECT_TRUE(blob.has_value());
if(blob.has_value()) {
EXPECT_EQ(result->tu_index_file_size, blob->size());
EXPECT_EQ(result->tu_index_hash, llvm::xxh3_64bits(*blob));
}
}
test_done = true;
w.peer->close_output();
});

ASSERT_TRUE(test_done);
}

TEST_CASE(SpillFailureFallsInline) {
TempDir tmp;
tmp.touch("test_index.cpp", "int indexed_var = 1;\n");
auto src = tmp.path("test_index.cpp");

WorkerHandle w;
ASSERT_TRUE(w.spawn());

bool test_done = false;

w.run([&]() -> kota::task<> {
worker::BuildParams params;
params.kind = worker::BuildKind::Index;
params.file = src;
params.directory = "/tmp";
params.arguments = make_args(src);
params.index_output_path = tmp.path("no_such_dir/spill.bin");
params.index_inline_limit = 1;

auto result = co_await w.peer->send_request(params);
EXPECT_TRUE(result.has_value());
if(result.has_value()) {
EXPECT_TRUE(result->success);
EXPECT_FALSE(result->tu_index_data.empty());
EXPECT_EQ(result->tu_index_file_size, std::uint64_t(0));
}
test_done = true;
w.peer->close_output();
});
Expand Down
Loading