diff --git a/ggml/src/CMakeLists.txt b/ggml/src/CMakeLists.txt index 82e9480c2f24..d7df54a868ac 100644 --- a/ggml/src/CMakeLists.txt +++ b/ggml/src/CMakeLists.txt @@ -201,6 +201,8 @@ add_library(ggml-base ggml-alloc.c ggml-backend.cpp ggml-backend-meta.cpp + ggml-backend-sanitize.cpp + ggml-backend-sanitize.h ggml-opt.cpp ggml-threading.cpp ggml-threading.h diff --git a/ggml/src/ggml-backend-sanitize.cpp b/ggml/src/ggml-backend-sanitize.cpp new file mode 100644 index 000000000000..1cecb20ce495 --- /dev/null +++ b/ggml/src/ggml-backend-sanitize.cpp @@ -0,0 +1,498 @@ +#include "ggml-backend-sanitize.h" +#include "ggml-backend-impl.h" +#include "ggml-impl.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +int ggml_san_level(void) { + static const int level = []() { + const char * env = getenv("GGML_SCHED_SANITIZE"); + return env ? atoi(env) : 0; + }(); + return level; +} + +namespace { + +constexpr int SAN_MAX_ACTORS = 64; +constexpr size_t SAN_MAX_RANGES = 65536; + +using vclock = std::vector; + +struct san_access { + int actor = -1; + uint64_t clock = 0; + int split = -1; + const char * what = ""; // always a string literal + char tensor[GGML_MAX_NAME] = { 0 }; // copied: source tensor is recycled per graph +}; + +struct san_entry { + size_t end; + san_access last_write; + std::vector reads; // at most one per actor +}; + +struct san_state { + std::mutex mutex; + + std::unordered_map actors; + std::vector actor_names; + std::unordered_map name_counts; + std::vector vc; // vc[a][b] = a's knowledge of b's clock + + std::unordered_map> mem; + + std::unordered_map ev_vc; + std::unordered_map events; +}; + +san_state & state() { + static san_state * s = []() { + san_state * st = new san_state(); + st->actor_names.push_back("HOST"); + st->vc.emplace_back(SAN_MAX_ACTORS, 0); + return st; + }(); + return *s; +} + +// every helper below assumes the caller holds state().mutex + +int actor_id(san_state & s, ggml_backend_t backend) { + if (backend == nullptr) { + return 0; + } + + auto it = s.actors.find(backend); + if (it != s.actors.end()) { + return it->second; + } + + // two backends can share a device and therefore a name, disambiguate + std::string base = ggml_backend_name(backend); + const int idx = s.name_counts[base]++; + + const int id = (int) s.actor_names.size(); + GGML_ASSERT(id < SAN_MAX_ACTORS); + s.actor_names.push_back(idx == 0 ? base : base + "#" + std::to_string(idx)); + s.vc.emplace_back(SAN_MAX_ACTORS, 0); + s.actors[backend] = id; + return id; +} + +const char * actor_name(san_state & s, int id) { + return id >= 0 && id < (int) s.actor_names.size() ? s.actor_names[id].c_str() : "?"; +} + +void join(vclock & dst, const vclock & src) { + for (size_t i = 0; i < dst.size(); i++) { + dst[i] = std::max(dst[i], src[i]); + } +} + +uint64_t tick(san_state & s, int a) { + return ++s.vc[a][a]; +} + +uint64_t issue(san_state & s, int a) { + join(s.vc[a], s.vc[0]); + return tick(s, a); +} + +void maybe_flush(san_state & s) { + if (s.mem.empty()) { + return; + } + + // once the host knows every actor's current clock, the issue rule guarantees that any + // future operation joins vc[0] first, so the shadow state can no longer report a race + for (size_t b = 0; b < s.vc.size(); b++) { + if (s.vc[0][b] < s.vc[b][b]) { + return; + } + } + + s.mem.clear(); +} + +bool is_view_op(enum ggml_op op) { + return op == GGML_OP_VIEW || op == GGML_OP_RESHAPE || op == GGML_OP_PERMUTE || op == GGML_OP_TRANSPOSE; +} + +struct mem_range { + ggml_backend_buffer_t buf; + size_t off; + size_t len; +}; + +bool resolve(const ggml_tensor * t, size_t offset, size_t size, mem_range & out) { + if (t == nullptr || t->data == nullptr) { + return false; + } + + ggml_backend_buffer_t buf = t->view_src ? t->view_src->buffer : t->buffer; + if (buf == nullptr) { + return false; + } + + // weights are written once at load and never recycled by the allocator + if (ggml_backend_buffer_get_usage(buf) == GGML_BACKEND_BUFFER_USAGE_WEIGHTS) { + return false; + } + + void * base = ggml_backend_buffer_get_base(buf); + if (base == nullptr || size == 0) { + return false; + } + + out.buf = buf; + out.off = (size_t) ((const char *) t->data - (const char *) base) + offset; + out.len = size; + return true; +} + +thread_local int tl_split = -1; +thread_local int tl_split_actor = -1; + +void report(san_state & s, const mem_range & mr, const san_access & cur, const san_access & prev, const char * kind) { + fprintf(stderr, "\nggml-sched-sanitize: RACE (%s) on %s[%zu, %zu)\n", + kind, ggml_backend_buffer_name(mr.buf), mr.off, mr.off + mr.len); + fprintf(stderr, "ggml-sched-sanitize: %-5s %-12s @%-4llu split %-4d %s (%s)\n", + strcmp(kind, "write-after-read") == 0 ? "read" : "write", + actor_name(s, prev.actor), (unsigned long long) prev.clock, prev.split, prev.tensor, prev.what); + fprintf(stderr, "ggml-sched-sanitize: %-5s %-12s @%-4llu split %-4d %s (%s)\n", + strcmp(kind, "read-after-write") == 0 ? "read" : "write", + actor_name(s, cur.actor), (unsigned long long) cur.clock, cur.split, cur.tensor, cur.what); + fprintf(stderr, "ggml-sched-sanitize: no happens-before edge: %s knows %s@%llu, needs >=%llu\n\n", + actor_name(s, cur.actor), actor_name(s, prev.actor), + (unsigned long long) s.vc[cur.actor][prev.actor], (unsigned long long) prev.clock); + + GGML_ABORT("ggml-sched-sanitize: race detected"); +} + +// make sure a range boundary exists at pos, so [off,off+len) lands on whole entries +void split_at(std::map & m, size_t pos) { + auto it = m.upper_bound(pos); + if (it == m.begin()) { + return; + } + --it; + if (it->first < pos && it->second.end > pos) { + san_entry tail = it->second; + it->second.end = pos; + m.emplace(pos, std::move(tail)); + } +} + +void access_range(san_state & s, const mem_range & mr, bool write, const san_access & info) { + auto & m = s.mem[mr.buf]; + const int a = info.actor; + + // the shadow state is bounded by the work issued before the host observes every actor, + // not by run length. reaching this means maybe_flush stopped clearing it - + // most likely an operation was enqueued without going through issue(), which breaks + // the argument maybe_flush relies on. that is a sanitizer bug, not a memory limit. + GGML_ASSERT(m.size() <= SAN_MAX_RANGES && "shadow state is not being flushed - see issue()/maybe_flush()"); + + const size_t begin = mr.off; + const size_t end = mr.off + mr.len; + + split_at(m, begin); + split_at(m, end); + + // fill gaps so the whole span is covered by entries + size_t cur = begin; + while (cur < end) { + auto it = m.lower_bound(cur); + if (it == m.end() || it->first >= end) { + san_entry e; + e.end = end; + m.emplace(cur, std::move(e)); + break; + } + if (it->first > cur) { + san_entry e; + e.end = it->first; + m.emplace(cur, std::move(e)); + } + cur = it->second.end; + } + + for (auto it = m.lower_bound(begin); it != m.end() && it->first < end; ++it) { + san_entry & e = it->second; + + if (e.last_write.actor >= 0 && e.last_write.actor != a) { + if (e.last_write.clock > s.vc[a][e.last_write.actor]) { + report(s, mr, info, e.last_write, write ? "write-after-write" : "read-after-write"); + } + } + + if (write) { + for (const san_access & r : e.reads) { + if (r.actor != a && r.clock > s.vc[a][r.actor]) { + report(s, mr, info, r, "write-after-read"); + } + } + } + + if (write) { + e.last_write = info; + e.reads.clear(); + } else { + bool found = false; + for (san_access & r : e.reads) { + if (r.actor == a) { + r = info; + found = true; + break; + } + } + if (!found) { + e.reads.push_back(info); + } + } + } + +} + +// returns false if the tensor is not in tracked memory +bool touch(san_state & s, int a, uint64_t c, const ggml_tensor * t, + size_t offset, size_t size, bool write, const char * what) { + mem_range mr; + if (!resolve(t, offset, size, mr)) { + return false; + } + + san_access info; + info.actor = a; + info.clock = c; + info.what = what; + info.split = tl_split; + snprintf(info.tensor, sizeof(info.tensor), "%s", t->name); + + if (ggml_san_level() >= 2) { + GGML_LOG_DEBUG("san [split %3d %-10s] %-10s %-5s %s[%zu+%zu] %s (%s)\n", + tl_split, tl_split_actor >= 0 ? actor_name(s, tl_split_actor) : "-", actor_name(s, a), + write ? "write" : "read", + ggml_backend_buffer_name(mr.buf), mr.off, mr.len, t->name, what); + } + + access_range(s, mr, write, info); + return true; +} + +void trace(san_state & s, int a, const char * fmt, ...) { + if (ggml_san_level() < 2) { + return; + } + + char body[512]; + va_list ap; + va_start(ap, fmt); + vsnprintf(body, sizeof(body), fmt, ap); + va_end(ap); + + GGML_LOG_DEBUG("san [split %3d %-10s] %-10s %s", + tl_split, tl_split_actor >= 0 ? actor_name(s, tl_split_actor) : "-", actor_name(s, a), body); +} + +} // namespace + +void ggml_san_sync(ggml_backend_t backend) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a = actor_id(s, backend); + join(s.vc[0], s.vc[a]); // host now knows everything this backend knew + trace(s, a, "EDGE synchronize -> HOST\n"); + maybe_flush(s); +} + +void ggml_san_event_record(ggml_backend_event_t event, ggml_backend_t backend) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a = actor_id(s, backend); + s.ev_vc[event] = s.vc[a]; + trace(s, a, "EDGE event_record ev=%d\n", (int) s.events.emplace(event, (int) s.events.size()).first->second); +} + +void ggml_san_event_wait(ggml_backend_t backend, ggml_backend_event_t event) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a = actor_id(s, backend); + auto it = s.ev_vc.find(event); + if (it != s.ev_vc.end()) { + join(s.vc[a], it->second); + } + trace(s, a, "EDGE event_wait ev=%d\n", (int) s.events.emplace(event, (int) s.events.size()).first->second); +} + +void ggml_san_event_sync(ggml_backend_event_t event) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + auto it = s.ev_vc.find(event); + if (it != s.ev_vc.end()) { + join(s.vc[0], it->second); + } + trace(s, 0, "EDGE event_sync ev=%d\n", (int) s.events.emplace(event, (int) s.events.size()).first->second); + maybe_flush(s); +} + +void ggml_san_compute(ggml_backend_t backend, const ggml_cgraph * cgraph) { + if (ggml_san_level() == 0 || cgraph == nullptr) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a = actor_id(s, backend); + + // a backend with no synchronize runs inline on the calling thread, so it both + // inherits and publishes the host's knowledge - without this the CPU backend's + // writes are never ordered against anything and no race can be detected + const bool synchronous = backend->iface.synchronize == nullptr; + + const uint64_t c = issue(s, a); + + size_t n_read = 0; + size_t n_write = 0; + for (int i = 0; i < cgraph->n_nodes; i++) { + ggml_tensor * node = cgraph->nodes[i]; + if (is_view_op(node->op)) { + continue; + } + for (int j = 0; j < GGML_MAX_SRC; j++) { + if (node->op == GGML_OP_CPY && j == 1) { + continue; + } + if (node->src[j] && touch(s, a, c, node->src[j], 0, ggml_nbytes(node->src[j]), false, "compute")) { + n_read++; + } + } + if (touch(s, a, c, node, 0, ggml_nbytes(node), true, "compute")) { + n_write++; + } + } + + if (synchronous) { + join(s.vc[0], s.vc[a]); + maybe_flush(s); + } + + trace(s, a, "compute %d nodes, r=%zu w=%zu ranges\n", cgraph->n_nodes, n_read, n_write); +} + +void ggml_san_access(ggml_backend_t backend, const ggml_tensor * tensor, + size_t offset, size_t size, bool write, const char * what) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a = actor_id(s, backend); + const uint64_t c = issue(s, a); + touch(s, a, c, tensor, offset, size, write, what); + + // a synchronous access has completed by the time the call returns + if (backend == nullptr || backend->iface.synchronize == nullptr) { + join(s.vc[0], s.vc[a]); + maybe_flush(s); + } +} + +void ggml_san_cpy_async(ggml_backend_t src_be, ggml_backend_t dst_be, + const ggml_tensor * src, const ggml_tensor * dst) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + const int a_src = actor_id(s, src_be); + const int a_dst = actor_id(s, dst_be); + + // TODO: clarify in ggml + // which stream actually performs the read of src is backend specific. host memory has + // no queue of its own, so a host->device transfer runs on dst's queue - that is what + // makes the source racy against a later host write. a device->device transfer is + // issued on the src stream (see "copy on src stream" in ggml-cuda), where src's own + // later work is already ordered against it. + ggml_backend_buffer_t src_buf = src->view_src ? src->view_src->buffer : src->buffer; + const int read_actor = (src_buf && ggml_backend_buffer_is_host(src_buf)) ? a_dst : a_src; + + uint64_t read_clock; + uint64_t write_clock; + + if (read_actor == a_dst) { + // the dst queue performs both sides of the copy after src's pending work + join(s.vc[a_dst], s.vc[a_src]); + read_clock = write_clock = issue(s, a_dst); + } else { + // issue the src read first, then carry its completion into the dst write + read_clock = issue(s, read_actor); + join(s.vc[a_dst], s.vc[read_actor]); + write_clock = issue(s, a_dst); + } + + touch(s, read_actor, read_clock, src, 0, ggml_nbytes(src), false, "cpy_async src"); + touch(s, a_dst, write_clock, dst, 0, ggml_nbytes(dst), true, "cpy_async dst"); + + trace(s, a_dst, "cpy_async %s@%s -> %s\n", src->name, actor_name(s, read_actor), dst->name); +} + +void ggml_san_buffer_free(ggml_backend_buffer_t buffer) { + if (ggml_san_level() == 0) { + return; + } + san_state & s = state(); + std::lock_guard lk(s.mutex); + + s.mem.erase(buffer); +} + +void ggml_san_split(int split_id, ggml_backend_t backend, int n_inputs) { + if (ggml_san_level() == 0) { + return; + } + tl_split = split_id; + + if (split_id < 0) { + tl_split_actor = -1; + return; + } + + san_state & s = state(); + std::lock_guard lk(s.mutex); + const int a = actor_id(s, backend); + tl_split_actor = a; + + if (n_inputs == 0) { + trace(s, a, "split 0 inputs (no sync path)\n"); + } +} diff --git a/ggml/src/ggml-backend-sanitize.h b/ggml/src/ggml-backend-sanitize.h new file mode 100644 index 000000000000..9009f50e6492 --- /dev/null +++ b/ggml/src/ggml-backend-sanitize.h @@ -0,0 +1,36 @@ +#pragma once + +// happens-before instrumentation for the ggml backend API +// +// GGML_SCHED_SANITIZE=1 detect happens-before violations +// GGML_SCHED_SANITIZE=2 also trace synchronization edges and memory ranges + +#include "ggml-backend.h" + +#ifdef __cplusplus +extern "C" { +#endif + + int ggml_san_level(void); + + void ggml_san_sync (ggml_backend_t backend); + void ggml_san_event_record(ggml_backend_event_t event, ggml_backend_t backend); + void ggml_san_event_wait (ggml_backend_t backend, ggml_backend_event_t event); + void ggml_san_event_sync (ggml_backend_event_t event); + + void ggml_san_compute (ggml_backend_t backend, const struct ggml_cgraph * cgraph); + + // backend == NULL means the access is performed by the host thread + void ggml_san_access (ggml_backend_t backend, const struct ggml_tensor * tensor, + size_t offset, size_t size, bool write, const char * what); + + void ggml_san_cpy_async (ggml_backend_t src_be, ggml_backend_t dst_be, + const struct ggml_tensor * src, const struct ggml_tensor * dst); + + void ggml_san_buffer_free (ggml_backend_buffer_t buffer); + + void ggml_san_split (int split_id, ggml_backend_t backend, int n_inputs); + +#ifdef __cplusplus +} +#endif diff --git a/ggml/src/ggml-backend.cpp b/ggml/src/ggml-backend.cpp index 7f4e252dca39..6804c1871c34 100644 --- a/ggml/src/ggml-backend.cpp +++ b/ggml/src/ggml-backend.cpp @@ -10,6 +10,7 @@ #include "ggml-backend.h" #include "ggml-backend-impl.h" +#include "ggml-backend-sanitize.h" #include "ggml-alloc.h" #include "ggml-impl.h" @@ -109,6 +110,8 @@ void ggml_backend_buffer_free(ggml_backend_buffer_t buffer) { return; } + ggml_san_buffer_free(buffer); + if (buffer->iface.free_buffer != NULL) { buffer->iface.free_buffer(buffer); } @@ -205,7 +208,12 @@ void ggml_backend_buffer_reset(ggml_backend_buffer_t buffer) { bool ggml_backend_buffer_copy_tensor(const struct ggml_tensor * src, struct ggml_tensor * dst) { ggml_backend_buffer_t dst_buf = dst->view_src ? dst->view_src->buffer : dst->buffer; if (dst_buf->iface.cpy_tensor) { - return dst_buf->iface.cpy_tensor(dst_buf, src, dst); + const bool copied = dst_buf->iface.cpy_tensor(dst_buf, src, dst); + if (copied) { + ggml_san_access(NULL, src, 0, ggml_nbytes(src), false, "buffer_copy_tensor src"); + ggml_san_access(NULL, dst, 0, ggml_nbytes(dst), true, "buffer_copy_tensor dst"); + } + return copied; } return false; } @@ -261,6 +269,7 @@ void ggml_backend_tensor_set_async(ggml_backend_t backend, struct ggml_tensor * ggml_backend_synchronize(backend); ggml_backend_tensor_set(tensor, data, offset, size); } else { + ggml_san_access(backend, tensor, offset, size, true, "set_async"); backend->iface.set_tensor_async(backend, tensor, data, offset, size); } } @@ -275,6 +284,7 @@ void ggml_backend_tensor_get_async(ggml_backend_t backend, const struct ggml_ten ggml_backend_synchronize(backend); ggml_backend_tensor_get(tensor, data, offset, size); } else { + ggml_san_access(backend, tensor, offset, size, false, "get_async"); backend->iface.get_tensor_async(backend, tensor, data, offset, size); } } @@ -297,6 +307,9 @@ void ggml_backend_tensor_set_2d_async(ggml_backend_t backend, struct ggml_tensor GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + (n_copies-1)*stride_tensor + size <= ggml_nbytes(tensor) && "tensor write out of bounds"); + for (size_t i = 0; i < n_copies; i++) { + ggml_san_access(backend, tensor, offset + i*stride_tensor, size, true, "set_2d_async"); + } backend->iface.set_tensor_2d_async(backend, tensor, data, offset, size, n_copies, stride_tensor, stride_data); } @@ -318,6 +331,9 @@ void ggml_backend_tensor_get_2d_async(ggml_backend_t backend, const struct ggml_ GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + (n_copies-1)*stride_tensor + size <= ggml_nbytes(tensor) && "tensor read out of bounds"); + for (size_t i = 0; i < n_copies; i++) { + ggml_san_access(backend, tensor, offset + i*stride_tensor, size, false, "get_2d_async"); + } backend->iface.get_tensor_2d_async(backend, tensor, data, offset, size, n_copies, stride_tensor, stride_data); } @@ -333,6 +349,7 @@ void ggml_backend_tensor_set(struct ggml_tensor * tensor, const void * data, siz GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + size <= ggml_nbytes(tensor) && "tensor write out of bounds"); + ggml_san_access(NULL, tensor, offset, size, true, "tensor_set"); buf->iface.set_tensor(buf, tensor, data, offset, size); } @@ -348,6 +365,7 @@ void ggml_backend_tensor_get(const struct ggml_tensor * tensor, void * data, siz GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + size <= ggml_nbytes(tensor) && "tensor read out of bounds"); + ggml_san_access(NULL, tensor, offset, size, false, "tensor_get"); buf->iface.get_tensor(buf, tensor, data, offset, size); } @@ -370,6 +388,9 @@ void ggml_backend_tensor_set_2d(struct ggml_tensor * tensor, const void * data, GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + (n_copies-1)*stride_tensor + size <= ggml_nbytes(tensor) && "tensor write out of bounds"); + for (size_t i = 0; i < n_copies; i++) { + ggml_san_access(NULL, tensor, offset + i*stride_tensor, size, true, "tensor_set_2d"); + } buf->iface.set_tensor_2d(buf, tensor, data, offset, size, n_copies, stride_tensor, stride_data); } @@ -392,6 +413,9 @@ void ggml_backend_tensor_get_2d(const struct ggml_tensor * tensor, void * data, GGML_ASSERT(tensor->data != NULL && "tensor not allocated"); GGML_ASSERT(offset + (n_copies-1)*stride_tensor + size <= ggml_nbytes(tensor) && "tensor read out of bounds"); + for (size_t i = 0; i < n_copies; i++) { + ggml_san_access(NULL, tensor, offset + i*stride_tensor, size, false, "tensor_get_2d"); + } buf->iface.get_tensor_2d(buf, tensor, data, offset, size, n_copies, stride_tensor, stride_data); } @@ -408,6 +432,7 @@ void ggml_backend_tensor_memset(struct ggml_tensor * tensor, uint8_t value, size GGML_ASSERT(offset + size <= ggml_nbytes(tensor) && "tensor write out of bounds"); GGML_ASSERT(buf->iface.memset_tensor != NULL && "memset not implemented by backend buffer"); + ggml_san_access(NULL, tensor, offset, size, true, "tensor_memset"); buf->iface.memset_tensor(buf, tensor, value, offset, size); } @@ -418,6 +443,7 @@ void ggml_backend_synchronize(ggml_backend_t backend) { } backend->iface.synchronize(backend); + ggml_san_sync(backend); } ggml_backend_graph_plan_t ggml_backend_graph_plan_create(ggml_backend_t backend, struct ggml_cgraph * cgraph) { @@ -449,7 +475,11 @@ enum ggml_status ggml_backend_graph_compute(ggml_backend_t backend, struct ggml_ enum ggml_status ggml_backend_graph_compute_async(ggml_backend_t backend, struct ggml_cgraph * cgraph) { GGML_ASSERT(backend); - return backend->iface.graph_compute(backend, cgraph); + const enum ggml_status status = backend->iface.graph_compute(backend, cgraph); + if (status == GGML_STATUS_SUCCESS) { + ggml_san_compute(backend, cgraph); + } + return status; } bool ggml_backend_supports_op(ggml_backend_t backend, const struct ggml_tensor * op) { @@ -482,9 +512,11 @@ void ggml_backend_tensor_copy(const struct ggml_tensor * src, struct ggml_tensor } if (ggml_backend_buffer_is_host(src->buffer)) { + ggml_san_access(NULL, src, 0, ggml_nbytes(src), false, "tensor_copy src"); ggml_backend_tensor_set(dst, src->data, 0, ggml_nbytes(src)); } else if (ggml_backend_buffer_is_host(dst->buffer)) { ggml_backend_tensor_get(src, dst->data, 0, ggml_nbytes(src)); + ggml_san_access(NULL, dst, 0, ggml_nbytes(src), true, "tensor_copy dst"); } else if (!ggml_backend_buffer_copy_tensor(src, dst)) { #ifndef NDEBUG GGML_LOG_DEBUG("%s: warning: slow copy from %s to %s\n", __func__, ggml_backend_buffer_name(src->buffer), ggml_backend_buffer_name(dst->buffer)); @@ -506,7 +538,9 @@ void ggml_backend_tensor_copy_async(ggml_backend_t backend_src, ggml_backend_t b GGML_ASSERT(backend_dst); if (backend_dst->iface.cpy_tensor_async != NULL) { - if (backend_dst->iface.cpy_tensor_async(backend_src, backend_dst, src, dst)) { + const bool accepted = backend_dst->iface.cpy_tensor_async(backend_src, backend_dst, src, dst); + if (accepted) { + ggml_san_cpy_async(backend_src, backend_dst, src, dst); return; } } @@ -540,6 +574,7 @@ void ggml_backend_event_record(ggml_backend_event_t event, ggml_backend_t backen GGML_ASSERT(backend->iface.event_record != NULL); backend->iface.event_record(backend, event); + ggml_san_event_record(event, backend); } void ggml_backend_event_synchronize(ggml_backend_event_t event) { @@ -547,6 +582,7 @@ void ggml_backend_event_synchronize(ggml_backend_event_t event) { GGML_ASSERT(event->device->iface.event_synchronize); event->device->iface.event_synchronize(event->device, event); + ggml_san_event_sync(event); } void ggml_backend_event_wait(ggml_backend_t backend, ggml_backend_event_t event) { @@ -554,6 +590,7 @@ void ggml_backend_event_wait(ggml_backend_t backend, ggml_backend_event_t event) GGML_ASSERT(backend->iface.event_wait != NULL); backend->iface.event_wait(backend, event); + ggml_san_event_wait(backend, event); } static void ggml_backend_graph_optimize(ggml_backend_t backend, struct ggml_cgraph * cgraph) { @@ -1560,6 +1597,8 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s int split_backend_id = split->backend_id; ggml_backend_t split_backend = sched->backends[split_backend_id]; + ggml_san_split(split_id, split_backend, split->n_inputs); + // copy the input tensors to the split backend for (int input_id = 0; input_id < split->n_inputs; input_id++) { ggml_backend_t input_backend = ggml_backend_sched_get_tensor_backend(sched, split->inputs[input_id]); @@ -1670,7 +1709,14 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s } else { // try async copy, but if not possible, we can still use a sync copy without synchronizing the dst backend, since we handle the synchronization here with multiple copies and events // TODO: add public function to facilitate this, since applications do not have direct access to the backend interface - if (!split_backend->iface.cpy_tensor_async || !split_backend->iface.cpy_tensor_async(input_backend, split_backend, input, input_cpy)) { + bool cpy_async_ok = false; + if (split_backend->iface.cpy_tensor_async) { + cpy_async_ok = split_backend->iface.cpy_tensor_async(input_backend, split_backend, input, input_cpy); + if (cpy_async_ok) { + ggml_san_cpy_async(input_backend, split_backend, input, input_cpy); + } + } + if (!cpy_async_ok) { ggml_backend_synchronize(input_backend); if (sched->events[split_backend_id][sched->cur_copy] != NULL) { ggml_backend_event_synchronize(sched->events[split_backend_id][sched->cur_copy]); @@ -1686,6 +1732,7 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s if (!sched->callback_eval) { enum ggml_status ec = ggml_backend_graph_compute_async(split_backend, &split->graph); if (ec != GGML_STATUS_SUCCESS) { + ggml_san_split(-1, NULL, 0); return ec; } } else { @@ -1708,6 +1755,7 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s enum ggml_status ec = ggml_backend_graph_compute_async(split_backend, &gv); if (ec != GGML_STATUS_SUCCESS) { + ggml_san_split(-1, NULL, 0); return ec; } @@ -1730,6 +1778,8 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s } } + ggml_san_split(-1, NULL, 0); + return GGML_STATUS_SUCCESS; } diff --git a/src/llama-context.cpp b/src/llama-context.cpp index 9b399d6096b1..04337755a6dd 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -1314,6 +1314,41 @@ bool llama_context::set_adapter_cvec( return res; } +// Returns true if any input tensor of the graph resides in a buffer that is +// host-visible but owned by a non-CPU device (ROCm_Host, CUDA_Host, Vulkan_Host). +// +// Such a buffer is created through the CPU buffer interface, so +// ggml_backend_tensor_set on it is a bare memcpy on the calling thread: it never +// reaches the owning backend, and therefore cannot be ordered against an +// in-flight graph_compute_async whose kernels are still reading the tensor. +// The scheduler cannot help either - it inserts no copy or sync for these +// tensors precisely because the backend advertises that it can consume the +// buffer directly (ggml_backend_dev_supports_buft). +static bool graph_has_device_host_inputs(ggml_cgraph * gf) { + for (int i = 0; i < ggml_graph_n_nodes(gf); i++) { + ggml_tensor * node = ggml_graph_node(gf, i); + + for (int j = 0; j < GGML_MAX_SRC; j++) { + ggml_tensor * src = node->src[j]; + + if (src == nullptr || !(src->flags & GGML_TENSOR_FLAG_INPUT) || src->buffer == nullptr) { + continue; + } + + if (!ggml_backend_buffer_is_host(src->buffer)) { + continue; + } + + ggml_backend_dev_t dev = ggml_backend_buft_get_device(ggml_backend_buffer_get_type(src->buffer)); + if (dev != nullptr && ggml_backend_dev_type(dev) != GGML_BACKEND_DEVICE_TYPE_CPU) { + return true; + } + } + } + + return false; +} + llm_graph_result * llama_context::process_ubatch(const llama_ubatch & ubatch, llm_graph_type gtype, llama_memory_context_i * mctx, ggml_status & ret) { if (mctx && !mctx->apply()) { LLAMA_LOG_ERROR("%s: failed to apply memory context\n", __func__); @@ -1362,6 +1397,17 @@ llm_graph_result * llama_context::process_ubatch(const llama_ubatch & ubatch, ll ret = GGML_STATUS_ALLOC_FAILED; return nullptr; } + + gf_host_inputs = graph_has_device_host_inputs(gf); + } + + // the previous graph_compute_async may still be reading the input tensors that + // set_inputs is about to overwrite. this is only a hazard for inputs in a + // device-owned host buffer - see graph_has_device_host_inputs() - and costs + // nothing in practice: during decode the logits readback has already + // synchronized, and during prefill the wait is dwarfed by the ubatch itself. + if (gf_host_inputs) { + ggml_backend_sched_synchronize(sched.get()); } // set the input data for the input tensors diff --git a/src/llama-context.h b/src/llama-context.h index bf91daa8b562..0d68532674a1 100644 --- a/src/llama-context.h +++ b/src/llama-context.h @@ -378,6 +378,13 @@ struct llama_context { // env: LLAMA_GRAPH_REUSE_DISABLE bool graph_reuse_disable = false; + // true if any graph input lives in a device-owned host buffer (ROCm_Host, + // CUDA_Host, Vulkan_Host, ...). such inputs are written by set_inputs with a + // plain CPU memcpy that is not ordered against an in-flight compute, so they + // require an explicit synchronization before they can be overwritten. + // recomputed whenever the graph is rebuilt. + bool gf_host_inputs = false; + // perf mutable int64_t t_start_us = 0; mutable int64_t t_load_us = 0;