owen/longhaul.cpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Vulkan longhaul support

Browse Source
This commit is contained in:
Owen Qwen
2026-07-30 15:09:43 -05:00
parent a673afad85
commit fa404a9d97
13 changed files with 262 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/llama-longhaul.cpp
+66 -40
View File
@@ -89,8 +89,14 @@ void llama_longhaul_cache::io_worker() {
try {
const auto & source = *job->source;
void * destination =
static_cast<uint8_t *>(source.tensor->data) + size_t(job->slot) * source.tensor->nb[2];
void * destination = nullptr;
if (job->direct) {
destination =
static_cast<uint8_t *>(source.tensor->data) + size_t(job->slot) * source.tensor->nb[2];
} else {
job->staging.resize(source.expert_size);
destination = job->staging.data();
}
files.at(source.file_idx)->read_at(
destination, source.expert_size,
source.offset + size_t(job->expert_id) * source.expert_size);
@@ -107,20 +113,45 @@ void llama_longhaul_cache::io_worker() {
}
}
ggml_backend_t llama_longhaul_cache::upload_backend(
const llama_model_loader::longhaul_source & source) {
ggml_backend_buffer_type_t buft = ggml_backend_buffer_get_type(source.tensor->buffer);
ggml_backend_dev_t dev = ggml_backend_buft_get_device(buft);
if (dev == nullptr || buft != ggml_backend_dev_buffer_type(dev)) {
return nullptr;
}
auto it = upload_backends.find(dev);
if (it != upload_backends.end()) {
return it->second.get();
}
ggml_backend_ptr backend(ggml_backend_dev_init(dev, nullptr));
if (!backend) {
return nullptr;
}
ggml_backend_t result = backend.get();
upload_backends.emplace(dev, std::move(backend));
return result;
}
bool llama_longhaul_cache::load_plan_sources() {
const size_t n_direct_sources = std::count_if(
sources_by_layer.at(locked_layer).begin(),
sources_by_layer.at(locked_layer).end(),
[&](const auto * source) { return source_is_direct(*source); });
const size_t n_jobs = load_plan.size() * n_direct_sources;
const size_t n_jobs = load_plan.size() * sources_by_layer.at(locked_layer).size();
io_jobs.clear();
io_jobs.reserve(n_jobs);
for (const auto & item : load_plan) {
for (const auto * source : sources_by_layer.at(locked_layer)) {
if (source_is_direct(*source)) {
io_jobs.push_back({ source, item.first, item.second, false, {} });
}
io_jobs.push_back({
source,
item.first,
item.second,
source_is_direct(*source),
{},
false,
{},
});
}
}
@@ -133,35 +164,6 @@ bool llama_longhaul_cache::load_plan_sources() {
io_ready.notify_all();
}
bool staged_ok = true;
for (const auto & item : load_plan) {
const int32_t expert_id = item.first;
const int slot = item.second;
for (const auto * source : sources_by_layer.at(locked_layer)) {
if (source_is_direct(*source)) {
continue;
}
read_buffer.resize(source->expert_size);
try {
files.at(source->file_idx)->read_at(
read_buffer.data(), read_buffer.size(),
source->offset + size_t(expert_id) * source->expert_size);
ggml_backend_tensor_set(
source->tensor, read_buffer.data(),
size_t(slot) * source->tensor->nb[2], read_buffer.size());
bytes_read += read_buffer.size();
} catch (const std::exception & e) {
last_error = e.what();
staged_ok = false;
break;
}
}
if (!staged_ok) {
break;
}
}
if (!io_jobs.empty()) {
std::unique_lock<std::mutex> lock(io_mutex);
io_done.wait(lock, [&] { return io_pending == 0; });
@@ -174,7 +176,31 @@ bool llama_longhaul_cache::load_plan_sources() {
}
bytes_read += job.source->expert_size;
}
return staged_ok;
std::vector<ggml_backend_t> pending_backends;
for (const auto & job : io_jobs) {
if (job.direct) {
continue;
}
const size_t offset = size_t(job.slot) * job.source->tensor->nb[2];
ggml_backend_t backend = upload_backend(*job.source);
if (backend != nullptr) {
ggml_backend_tensor_set_async(
backend, job.source->tensor, job.staging.data(), offset, job.staging.size());
if (std::find(pending_backends.begin(), pending_backends.end(), backend) == pending_backends.end()) {
pending_backends.push_back(backend);
}
} else {
ggml_backend_tensor_set(
job.source->tensor, job.staging.data(), offset, job.staging.size());
}
}
for (ggml_backend_t backend : pending_backends) {
ggml_backend_synchronize(backend);
}
return true;
}
void llama_longhaul_cache::invalidate_plan(int layer) {
src/llama-longhaul.h
+6 -1
View File
@@ -6,6 +6,7 @@
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <string>
@@ -43,6 +44,8 @@ private:
const llama_model_loader::longhaul_source * source;
int32_t expert_id;
int slot;
bool direct;
std::vector<uint8_t> staging;
bool ok = false;
std::string error;
};
@@ -74,7 +77,6 @@ private:
std::vector<int32_t> available_slots;
std::vector<std::pair<int32_t, int32_t>> load_plan;
std::vector<int32_t> id_buffer;
std::vector<uint8_t> read_buffer;
std::vector<std::thread> io_workers;
std::vector<io_job> io_jobs;
@@ -85,7 +87,10 @@ private:
size_t io_pending = 0;
bool io_stopping = false;
std::map<ggml_backend_dev_t, ggml_backend_ptr> upload_backends;
bool source_is_direct(const llama_model_loader::longhaul_source & source) const;
ggml_backend_t upload_backend(const llama_model_loader::longhaul_source & source);
bool load_plan_sources();
void invalidate_plan(int layer);
void io_worker();
src/llama-mmap.cpp
+16
View File
@@ -66,6 +66,8 @@ static std::string llama_format_win_err(DWORD err) {
// llama_file
struct llama_file::impl {
bool no_cache = false;
#if defined(_WIN32)
HANDLE fp_win32;
std::string GetErrorMessageWin32(DWORD error_code) const {
@@ -448,6 +450,14 @@ void llama_file::read_at(void * ptr, size_t len, size_t offset) {
}
n_read += result;
}
#if defined(__linux__) && defined(POSIX_FADV_DONTNEED)
if (pimpl->no_cache) {
// Longhaul retains useful expert data in its own bounded cache. Drop
// completed file reads so the kernel page cache does not become a
// second, unbounded copy of the streamed weights.
(void) posix_fadvise(file_id(), (off_t) offset, (off_t) len, POSIX_FADV_DONTNEED);
}
#endif
#endif
}
@@ -456,6 +466,12 @@ void llama_file::set_no_cache() const {
if (fcntl(file_id(), F_NOCACHE, 1) != 0) {
throw std::runtime_error(format("failed to disable file cache: %s", strerror(errno)));
}
#elif defined(__linux__) && defined(POSIX_FADV_RANDOM)
pimpl->no_cache = true;
const int err = posix_fadvise(file_id(), 0, 0, POSIX_FADV_RANDOM);
if (err != 0) {
LLAMA_LOG_WARN("%s: failed to set random-access file advice: %s\n", __func__, strerror(err));
}
#endif
}
src/llama-model.cpp
+12 -6
View File
@@ -1356,9 +1356,6 @@ bool llama_model_base::load_tensors(llama_model_loader & ml) {
}
if (params.load_mode == LLAMA_LOAD_MODE_LONGHAUL) {
#if !defined(__APPLE__)
throw std::runtime_error("longhaul is only supported on macOS");
#endif
if (arch != LLM_ARCH_QWEN35MOE && arch != LLM_ARCH_LAGUNA) {
throw std::runtime_error("longhaul currently requires a qwen35moe or laguna model");
}
@@ -1368,12 +1365,21 @@ bool llama_model_base::load_tensors(llama_model_loader & ml) {
if (params.check_tensors || params.no_alloc || params.vocab_only) {
throw std::runtime_error("longhaul does not support check-tensors, no-alloc, or vocab-only loading");
}
ggml_backend_dev_t longhaul_dev = nullptr;
for (int il = 0; il < n_layer_all; ++il) {
ggml_backend_dev_t dev = pimpl->dev_layer[il].dev;
if (ggml_backend_dev_type(dev) != GGML_BACKEND_DEVICE_TYPE_GPU ||
strcmp(ggml_backend_reg_name(ggml_backend_dev_backend_reg(dev)), "MTL") != 0) {
throw std::runtime_error("longhaul requires every model layer on Metal");
const char * backend = ggml_backend_reg_name(ggml_backend_dev_backend_reg(dev));
const bool supported_backend =
strcmp(backend, "MTL") == 0 || strcmp(backend, "Vulkan") == 0;
if (ggml_backend_dev_type(dev) != GGML_BACKEND_DEVICE_TYPE_GPU || !supported_backend) {
throw std::runtime_error(
"longhaul requires every model layer on a Metal or Vulkan GPU");
}
if (longhaul_dev != nullptr && dev != longhaul_dev) {
throw std::runtime_error(
"longhaul requires every model layer on one GPU; select it with --device");
}
longhaul_dev = dev;
}
ml.configure_longhaul(params.longhaul_cache_bytes, n_expert, n_layer_all);
if (ml.longhaul_slots < (size_t) n_expert_used) {