Compare commits

...
Author SHA1 Message Date
Xuan Son Nguyen 5e59982160 engine: separate engine and server 2026-10-16 20:17:40 +02:00
41 changed files with 6045 additions and 2383 deletions
+2
View File
@@ -46,6 +46,7 @@ examples:
- examples/**
- tools/**
- all-globs-to-all-files:
- '!tools/engine/**'
- '!tools/server/**'
- '!tools/mtmd/**'
- '!tools/ui/**'
@@ -76,6 +77,7 @@ server/ui:
server:
- changed-files:
- any-glob-to-any-file:
- tools/engine/**
- tools/server/**
mtmd:
- changed-files:
+1
View File
@@ -101,6 +101,7 @@
/tools/batched-bench/ @ggerganov
/tools/cli/ @ngxson
/tools/completion/ @ggerganov
/tools/engine/ @ggml-org/llama-server
/tools/mtmd/ @ggml-org/llama-mtmd
/tools/perplexity/ @ggerganov
/tools/parser/ @pwilkin
+1 -1
View File
@@ -188,7 +188,7 @@ def _load_d1_omni_hparams(dir_model: Path) -> dict[str, Any]:
class D1OmniModel(LFM2Model):
model_arch = gguf.MODEL_ARCH.LFM2
# the server cuts the text to these lengths, see server-decision.cpp
# the server cuts the text to these lengths, see engine-decision.cpp
_MAX_LENGTH = 16384
_IMAGE_TEXT_LENGTH = 896
_AUDIO_TEXT_LENGTH = 15360
+2 -2
View File
@@ -147,11 +147,11 @@ The conversion is the same as above, but a new model needs its own `DecisionType
> [!IMPORTANT]
>
> Most of the logic is handled in `tools/server/server-decision.cpp`, to avoid too many changes to `libllama`.
> Most of the logic is handled in `tools/engine/engine-decision.cpp`, to avoid too many changes to `libllama`.
Note:
- If a new public API is needed in `libllama`, add it to `llama-ext.h`.
- Metadata with a single use case must be hard-coded in `server-decision.cpp` instead of being saved to the GGUF. This avoids bloating the conversion code.
- Metadata with a single use case must be hard-coded in `engine-decision.cpp` instead of being saved to the GGUF. This avoids bloating the conversion code.
- Most importantly, keep your change as small and as self-contained as possible. Reuse the existing infrastructure whenever you can.
For more information, see [PR #29818](https://github.com/ggml-org/llama.cpp/pull/29818).
+1 -1
View File
@@ -214,7 +214,7 @@ Adjacent data-shape rule: empty string is not a valid "unset" initial state - us
- Only add tests that bring meaningful results. Too-trivial tests just bloat the suite and CI.
- No time-sensitive tests (timing thresholds, sleeps, races against wall-clock); they are flaky on CI.
- Think twice about tests that significantly increase CI run time (expensive computation, large inputs, or long sleep/wait delays) or download large amounts of data from the internet (big models, datasets). Flag them and ask whether a smaller model/input or an existing fixture would do.
- Server: small tests of internal components (helpers, `server_tokens`, queues, format converters, ...) belong in `tests/test-server-component.cpp`. The end-to-end pytest suite in `tools/server/tests` is costly in CI time; only add to it when the test requires end-to-end behavior with a real model.
- Server: small tests of internal components (helpers, `engine_tokens`, queues, format converters, ...) belong in `tests/test-server-component.cpp`. The end-to-end pytest suite in `tools/server/tests` is costly in CI time; only add to it when the test requires end-to-end behavior with a real model.
## General (always)
+2 -4
View File
@@ -188,11 +188,9 @@ if (NOT WIN32 OR NOT BUILD_SHARED_LIBS)
llama_build_and_test(test-llama-grammar.cpp LINK llama)
llama_build_and_test(test-batch-alloc.cpp LINK llama)
llama_build_and_test(test-chat.cpp WORKING_DIRECTORY ${PROJECT_SOURCE_DIR})
target_include_directories(test-chat PRIVATE ${PROJECT_SOURCE_DIR}/tools/server)
target_link_libraries(test-chat PRIVATE server-context)
target_link_libraries(test-chat PRIVATE llama-engine)
llama_build_and_test(test-server-component.cpp)
target_include_directories(test-server-component PRIVATE ${PROJECT_SOURCE_DIR}/tools/server)
target_link_libraries(test-server-component PRIVATE server-context)
target_link_libraries(test-server-component PRIVATE llama-engine)
llama_build_and_test(test-json-schema-to-grammar.cpp)
if (NOT GGML_BACKEND_DL)
+9 -9
View File
@@ -7,7 +7,7 @@
//
#include "../src/llama-grammar.h"
#include "../src/unicode.h"
#include "../tools/server/server-chat.h"
#include "../tools/engine/engine-chat.h"
#include "chat-auto-parser.h"
#include "chat.h"
#include "common.h"
@@ -1864,7 +1864,7 @@ static void test_convert_responses_to_chatcmpl() {
}
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
// Verify messages were converted correctly
assert_equals(true, result.contains("messages"));
@@ -1904,7 +1904,7 @@ static void test_convert_responses_to_chatcmpl() {
"model": "test-model"
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
assert_equals((size_t)1, result.at("messages").size());
const auto & msg = result.at("messages")[0];
@@ -1920,7 +1920,7 @@ static void test_convert_responses_to_chatcmpl() {
"model": "test-model"
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
assert_equals((size_t)2, result.at("messages").size());
const auto & sys_msg = result.at("messages")[0];
@@ -1936,7 +1936,7 @@ static void test_convert_responses_to_chatcmpl() {
"max_output_tokens": 100
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
assert_equals(true, result.contains("max_tokens"));
assert_equals(false, result.contains("max_output_tokens"));
@@ -1980,7 +1980,7 @@ static void test_convert_responses_to_chatcmpl() {
]
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
assert_equals(true, result.contains("tools"));
assert_equals(true, result.at("tools").is_array());
@@ -2015,7 +2015,7 @@ static void test_convert_responses_to_chatcmpl() {
]
})");
json result = server_chat_convert_responses_to_chatcmpl(input);
json result = engine_chat_convert_responses_to_chatcmpl(input);
assert_equals(false, result.contains("tools"));
}
@@ -7938,7 +7938,7 @@ static void test_reasoning_budget_tokens_per_request() {
// the reasoning-budget code path in oaicompat_chat_params_parse.
auto tmpls = read_templates("models/templates/Qwen-Qwen3-0.6B.jinja");
server_chat_params opt;
engine_chat_params opt;
opt.tmpls = std::move(tmpls);
opt.use_jinja = true;
opt.enable_thinking = true;
@@ -7970,7 +7970,7 @@ static void test_reasoning_budget_message_per_request() {
// <think>...</think> markers enable the reasoning-budget block in oaicompat_chat_params_parse.
auto tmpls = read_templates("models/templates/Qwen-Qwen3-0.6B.jinja");
server_chat_params opt;
engine_chat_params opt;
opt.tmpls = std::move(tmpls);
opt.use_jinja = true;
opt.enable_thinking = true;
+42 -41
View File
@@ -1,7 +1,8 @@
#include "testing.h"
#include "../tools/server/server-common.h"
#include "../tools/server/server-decision.h"
#include "../tools/engine/engine-common.h"
#include "../tools/server/server-subproc.h"
#include "../tools/engine/engine-decision.h"
#include <cstdlib>
#include <regex>
@@ -15,11 +16,11 @@
// see skills/code-review/SKILL.md for more
//
// server_tokens (text only, has_mtmd = false)
// engine_tokens (text only, has_mtmd = false)
//
static void test_server_tokens_basic(testing & t) {
server_tokens st(llama_tokens{1, 2, 3}, false);
static void test_engine_tokens_basic(testing & t) {
engine_tokens st(llama_tokens{1, 2, 3}, false);
t.assert_equal("size", (size_t) 3, st.size());
t.assert_true("not empty", !st.empty());
t.assert_equal("operator[]", 2, st[1]);
@@ -47,8 +48,8 @@ static void test_server_tokens_basic(testing & t) {
t.assert_true("clear", st.empty());
}
static void test_server_tokens_pos(testing & t) {
server_tokens st(llama_tokens{1, 2, 3, 4, 5}, false);
static void test_engine_tokens_pos(testing & t) {
engine_tokens st(llama_tokens{1, 2, 3, 4, 5}, false);
t.assert_equal("pos_next()", (llama_pos) 5, st.pos_next());
t.assert_equal("pos_next(0)", (llama_pos) 0, st.pos_next(0));
t.assert_equal("pos_next(3)", (llama_pos) 3, st.pos_next(3));
@@ -58,8 +59,8 @@ static void test_server_tokens_pos(testing & t) {
t.assert_equal("size_up_to_pos clamps", (size_t) 5, st.size_up_to_pos(100));
}
static void test_server_tokens_keep_first(testing & t) {
server_tokens st(llama_tokens{1, 2, 3, 4, 5}, false);
static void test_engine_tokens_keep_first(testing & t) {
engine_tokens st(llama_tokens{1, 2, 3, 4, 5}, false);
st.keep_first(5);
t.assert_equal("keep all", (size_t) 5, st.size());
st.keep_first(2);
@@ -68,21 +69,21 @@ static void test_server_tokens_keep_first(testing & t) {
t.assert_true("keep 0", st.empty());
}
static void test_server_tokens_common_prefix(testing & t) {
server_tokens a(llama_tokens{1, 2, 3, 4}, false);
server_tokens empty(llama_tokens{}, false);
static void test_engine_tokens_common_prefix(testing & t) {
engine_tokens a(llama_tokens{1, 2, 3, 4}, false);
engine_tokens empty(llama_tokens{}, false);
t.assert_equal("both empty", (size_t) 0, empty.get_common_prefix(empty));
t.assert_equal("one empty", (size_t) 0, a.get_common_prefix(empty));
t.assert_equal("identical", (size_t) 4, a.get_common_prefix(a));
t.assert_equal("diverge at 2", (size_t) 2, a.get_common_prefix(server_tokens(llama_tokens{1, 2, 9, 4}, false)));
t.assert_equal("diverge at 0", (size_t) 0, a.get_common_prefix(server_tokens(llama_tokens{9, 2, 3, 4}, false)));
t.assert_equal("b is prefix", (size_t) 2, a.get_common_prefix(server_tokens(llama_tokens{1, 2}, false)));
t.assert_equal("a is prefix", (size_t) 4, a.get_common_prefix(server_tokens(llama_tokens{1, 2, 3, 4, 5, 6}, false)));
t.assert_equal("diverge at 2", (size_t) 2, a.get_common_prefix(engine_tokens(llama_tokens{1, 2, 9, 4}, false)));
t.assert_equal("diverge at 0", (size_t) 0, a.get_common_prefix(engine_tokens(llama_tokens{9, 2, 3, 4}, false)));
t.assert_equal("b is prefix", (size_t) 2, a.get_common_prefix(engine_tokens(llama_tokens{1, 2}, false)));
t.assert_equal("a is prefix", (size_t) 4, a.get_common_prefix(engine_tokens(llama_tokens{1, 2, 3, 4, 5, 6}, false)));
}
static void test_server_tokens_serialize(testing & t) {
// same packing as slot save/restore in server-context.cpp
static void test_engine_tokens_serialize(testing & t) {
// same packing as slot save/restore in engine-context.cpp
auto pack = [](const std::vector<char> & data) {
llama_tokens packed(data.size() / sizeof(llama_token));
memcpy(packed.data(), data.data(), data.size());
@@ -90,29 +91,29 @@ static void test_server_tokens_serialize(testing & t) {
};
t.test("roundtrip", [&](testing & t) {
server_tokens st(llama_tokens{10, 20, 30}, false);
engine_tokens st(llama_tokens{10, 20, 30}, false);
std::vector<char> data = st.serialize();
t.assert_equal("aligned to llama_token", (size_t) 0, data.size() % sizeof(llama_token));
server_tokens restored = server_tokens::deserialize(pack(data), false);
engine_tokens restored = engine_tokens::deserialize(pack(data), false);
t.assert_true("tokens", restored.get_tokens() == st.get_tokens());
});
t.test("roundtrip empty", [&](testing & t) {
server_tokens st(llama_tokens{}, false);
server_tokens restored = server_tokens::deserialize(pack(st.serialize()), false);
engine_tokens st(llama_tokens{}, false);
engine_tokens restored = engine_tokens::deserialize(pack(st.serialize()), false);
t.assert_true("empty", restored.empty());
});
t.test("legacy plain token list", [&](testing & t) {
server_tokens restored = server_tokens::deserialize(llama_tokens{7, 8, 9}, false);
engine_tokens restored = engine_tokens::deserialize(llama_tokens{7, 8, 9}, false);
t.assert_true("tokens", restored.get_tokens() == llama_tokens({7, 8, 9}));
});
auto expect_throw = [](testing & t, const std::string & msg, const llama_tokens & packed) {
bool thrown = false;
try {
server_tokens::deserialize(packed, false);
engine_tokens::deserialize(packed, false);
} catch (const std::exception &) {
thrown = true;
}
@@ -120,7 +121,7 @@ static void test_server_tokens_serialize(testing & t) {
};
t.test("corrupted", [&](testing & t) {
llama_tokens good = pack(server_tokens(llama_tokens{10, 20, 30}, false).serialize());
llama_tokens good = pack(engine_tokens(llama_tokens{10, 20, 30}, false).serialize());
llama_tokens bad_version = good;
bad_version[1] = 999;
@@ -142,22 +143,22 @@ static void test_server_tokens_serialize(testing & t) {
});
}
static void test_server_tokens_clone(testing & t) {
server_tokens st(llama_tokens{1, 2, 3}, false);
server_tokens cl = st.clone();
static void test_engine_tokens_clone(testing & t) {
engine_tokens st(llama_tokens{1, 2, 3}, false);
engine_tokens cl = st.clone();
cl.push_back(4);
t.assert_equal("original untouched", (size_t) 3, st.size());
t.assert_equal("clone grew", (size_t) 4, cl.size());
t.assert_true("get_text_tokens", cl.get_text_tokens() == llama_tokens({1, 2, 3, 4}));
}
static void test_server_tokens(testing & t) {
t.test("basic", test_server_tokens_basic);
t.test("pos", test_server_tokens_pos);
t.test("keep_first", test_server_tokens_keep_first);
t.test("common_prefix", test_server_tokens_common_prefix);
t.test("serialize", test_server_tokens_serialize);
t.test("clone", test_server_tokens_clone);
static void test_engine_tokens(testing & t) {
t.test("basic", test_engine_tokens_basic);
t.test("pos", test_engine_tokens_pos);
t.test("keep_first", test_engine_tokens_keep_first);
t.test("common_prefix", test_engine_tokens_common_prefix);
t.test("serialize", test_engine_tokens_serialize);
t.test("clone", test_engine_tokens_clone);
}
//
@@ -374,7 +375,7 @@ static void test_server_pipe(testing & t) {
}
//
// server_decision_escape_special_tokens
// engine_decision_escape_special_tokens
//
// "<|name|>" with U+00A6 in place of "|"
@@ -398,7 +399,7 @@ static void test_decision_escape(testing & t) {
{ "<|abc", "<|abc" },
};
for (const auto & [text, expected] : cases) {
t.assert_equal(text, expected, server_decision_escape_special_tokens(text));
t.assert_equal(text, expected, engine_decision_escape_special_tokens(text));
}
});
@@ -418,7 +419,7 @@ static void test_decision_escape(testing & t) {
for (size_t k = 0, v = idx; k < len; k++, v /= chars.size()) {
text += chars[v % chars.size()];
}
if (server_decision_escape_special_tokens(text) != std::regex_replace(text, re_special, "<\xC2\xA6$1\xC2\xA6>")) {
if (engine_decision_escape_special_tokens(text) != std::regex_replace(text, re_special, "<\xC2\xA6$1\xC2\xA6>")) {
n_diff++;
}
n_strings++;
@@ -430,8 +431,8 @@ static void test_decision_escape(testing & t) {
// std::regex overflowed the stack on these
t.test("long input", [](testing & t) {
const std::string name(1000000, 'a');
t.assert_true("not closed", server_decision_escape_special_tokens("<|" + name) == "<|" + name);
t.assert_true("closed", server_decision_escape_special_tokens("<|" + name + "|>") == escaped(name));
t.assert_true("not closed", engine_decision_escape_special_tokens("<|" + name) == "<|" + name);
t.assert_true("closed", engine_decision_escape_special_tokens("<|" + name + "|>") == escaped(name));
});
}
@@ -447,7 +448,7 @@ int main(int argc, char ** argv) {
t.set_filter(argv[1]);
}
t.test("server_tokens", test_server_tokens);
t.test("engine_tokens", test_engine_tokens);
t.test("utf8", test_utf8);
t.test("sse", test_sse);
t.test("json_helpers", test_json_helpers);
+1
View File
@@ -23,6 +23,7 @@ else()
add_subdirectory(quantize)
if (LLAMA_BUILD_SERVER)
add_subdirectory(ui)
add_subdirectory(engine)
add_subdirectory(cli)
add_subdirectory(server)
endif()
+29
View File
@@ -0,0 +1,29 @@
# llama-engine: inference engine (context, task queue, common utils), used by llama-server and CLI
set(TARGET llama-engine)
add_library(${TARGET}
engine-chat.cpp
engine-chat.h
engine-task.cpp
engine-task.h
engine-queue.cpp
engine-queue.h
engine-common.cpp
engine-common.h
engine-decision.cpp
engine-decision.h
engine-context.cpp
engine-context.h
engine-schema.cpp
engine-schema.h
)
set_target_properties(${TARGET} PROPERTIES WINDOWS_EXPORT_ALL_SYMBOLS ON)
target_include_directories(${TARGET} PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
target_include_directories(${TARGET} PRIVATE ../mtmd ${PROJECT_SOURCE_DIR})
target_link_libraries(${TARGET} PUBLIC llama-common mtmd ${CMAKE_THREAD_LIBS_INIT})
if(LLAMA_TOOLS_INSTALL)
install(TARGETS ${TARGET} LIBRARY)
endif()
@@ -1,9 +1,9 @@
#include "server-chat.h"
#include "server-common.h"
#include "engine-chat.h"
#include "engine-common.h"
#include <sstream>
json server_chat_convert_responses_to_chatcmpl(const json & response_body) {
json engine_chat_convert_responses_to_chatcmpl(const json & response_body) {
if (!response_body.contains("input")) {
throw std::invalid_argument("'input' is required");
}
@@ -345,7 +345,7 @@ static void normalize_anthropic_billing_header(std::string & system_text) {
}
}
json server_chat_convert_anthropic_to_oai(const json & body) {
json engine_chat_convert_anthropic_to_oai(const json & body) {
json oai_body;
// Convert system prompt
@@ -618,7 +618,7 @@ json server_chat_convert_anthropic_to_oai(const json & body) {
return oai_body;
}
json server_chat_msg_diff_to_json_oaicompat(const common_chat_msg_diff & diff) {
json engine_chat_msg_diff_to_json_oaicompat(const common_chat_msg_diff & diff) {
json delta = json::object();
if (!diff.reasoning_content_delta.empty()) {
delta["reasoning_content"] = diff.reasoning_content_delta;
@@ -3,16 +3,15 @@
#pragma once
#include "chat.h"
#include "server-common.h"
#include "server-http.h"
#include "engine-common.h"
#include "json.h"
// Convert OpenAI Responses API format to OpenAI Chat Completions API format
json server_chat_convert_responses_to_chatcmpl(const json & body);
json engine_chat_convert_responses_to_chatcmpl(const json & body);
// Convert Anthropic Messages API format to OpenAI Chat Completions API format
json server_chat_convert_anthropic_to_oai(const json & body);
json engine_chat_convert_anthropic_to_oai(const json & body);
// convert OpenAI transcriptions API format to OpenAI Chat Completions API format
json convert_transcriptions_to_chatcmpl(
@@ -21,4 +20,4 @@ json convert_transcriptions_to_chatcmpl(
const std::map<std::string, uploaded_file> & in_files,
std::vector<raw_buffer> & out_files);
json server_chat_msg_diff_to_json_oaicompat(const common_chat_msg_diff & diff);
json engine_chat_msg_diff_to_json_oaicompat(const common_chat_msg_diff & diff);
File diff suppressed because it is too large Load Diff
@@ -6,7 +6,6 @@
#include "chat.h"
#include "mtmd.h"
#include "mtmd-helper.h"
#include "subproc.h"
#include "json.h"
@@ -39,6 +38,12 @@ using json = common_json;
using raw_buffer = std::vector<uint8_t>;
struct uploaded_file {
raw_buffer data;
std::string filename;
std::string content_type;
};
template <typename T>
static T json_value(const json & body, const std::string & key, const T & default_value) {
// Fallback null to default value
@@ -67,12 +72,12 @@ enum error_type {
};
// thin wrapper around common_grammar_trigger with (de)serialization functions
struct server_grammar_trigger {
struct engine_grammar_trigger {
common_grammar_trigger value;
server_grammar_trigger() = default;
server_grammar_trigger(const common_grammar_trigger & value) : value(value) {}
server_grammar_trigger(const json & in) {
engine_grammar_trigger() = default;
engine_grammar_trigger(const common_grammar_trigger & value) : value(value) {}
engine_grammar_trigger(const json & in) {
value.type = (common_grammar_trigger_type) in.at("type").get<int>();
value.value = in.at("value").get<std::string>();
if (value.type == COMMON_GRAMMAR_TRIGGER_TYPE_TOKEN) {
@@ -110,10 +115,10 @@ const char * get_media_marker();
//
// output modalities for architecture.output_modalities in GET /models
std::vector<std::string> server_model_output_modalities(common_decision_type decision_type);
std::vector<std::string> engine_model_output_modalities(common_decision_type decision_type);
// architecture object of GET /models; shared by the direct server and the router
json server_model_architecture_json(
json engine_model_architecture_json(
bool inp_image,
bool inp_audio,
bool inp_video,
@@ -142,14 +147,14 @@ bool are_lora_equal(
std::vector<size_t> lora_get_enabled_ids(const std::vector<common_adapter_lora_info> & loras);
//
// server_tokens
// engine_tokens
//
/**
* server_tokens is a helper to manage the input tokens and image for the server.
* engine_tokens is a helper to manage the input tokens and image for the server.
* it is made this way to simplify the logic of KV cache management.
*/
struct server_tokens {
struct engine_tokens {
bool has_mtmd = false;
private: // disallow accessing these members directly, risking out-of-sync
@@ -172,24 +177,24 @@ private: // disallow accessing these members directly, risking out-of-sync
// map_idx_to_media will contain: {5, img0}, {8, img1}
public:
server_tokens() = default;
~server_tokens() = default;
engine_tokens() = default;
~engine_tokens() = default;
// Prevent copying
// TODO: server_tokens should be copyable - remove this:
server_tokens(const server_tokens&) = delete;
server_tokens& operator=(const server_tokens&) = delete;
// TODO: engine_tokens should be copyable - remove this:
engine_tokens(const engine_tokens&) = delete;
engine_tokens& operator=(const engine_tokens&) = delete;
// Allow moving (usually implicitly generated if members are movable)
server_tokens(server_tokens&&) = default;
server_tokens& operator=(server_tokens&&) = default;
engine_tokens(engine_tokens&&) = default;
engine_tokens& operator=(engine_tokens&&) = default;
// Allow accessing elements using [] operator
llama_token operator[](size_t index) { return tokens[index]; }
const llama_token& operator[](size_t index) const { return tokens[index]; }
server_tokens(mtmd::input_chunks & mtmd_chunks, bool has_mtmd);
server_tokens(const llama_tokens & tokens, bool has_mtmd);
engine_tokens(mtmd::input_chunks & mtmd_chunks, bool has_mtmd);
engine_tokens(const llama_tokens & tokens, bool has_mtmd);
// for debugging
std::string str() const;
@@ -216,7 +221,7 @@ public:
void push_back_placeholder(const mtmd_input_chunk * chunk);
// appends server tokens, updates the media map. copies media chunks.
void push_back(server_tokens & tokens);
void push_back(engine_tokens & tokens);
// for compatibility with context shift and prompt truncation
void insert(const llama_tokens & inp_tokens);
@@ -227,7 +232,7 @@ public:
llama_tokens get_text_tokens() const;
std::vector<char> serialize() const;
static server_tokens deserialize(const llama_tokens & packed, bool has_mtmd);
static engine_tokens deserialize(const llama_tokens & packed, bool has_mtmd);
// for compatibility with speculative decoding
void set_token(llama_pos pos, llama_token id);
@@ -245,7 +250,7 @@ public:
std::string detokenize(const llama_context * ctx, bool special) const;
size_t get_common_prefix(const server_tokens & b) const;
size_t get_common_prefix(const engine_tokens & b) const;
// split the tokens into message spans, skipping over media chunks
common_chat_msg_spans find_message_spans(const common_chat_msg_delimiters & delims) const;
@@ -253,7 +258,7 @@ public:
// check text token IDs and the mapping between media chunks and token ranges
bool validate(const struct llama_context * ctx) const;
server_tokens clone() const;
engine_tokens clone() const;
};
@@ -290,9 +295,9 @@ void handle_media(
const std::string & url,
const std::string & media_path);
// process mtmd prompt, return the server_tokens containing both text tokens and media chunks
// if is_placeholder is true, the media chunk will be treated as placeholder for counting tokens; the output tokens are not usable for actual inference (e.g. for submitting a task to server_queue)
server_tokens process_mtmd_prompt(
// process mtmd prompt, return the engine_tokens containing both text tokens and media chunks
// if is_placeholder is true, the media chunk will be treated as placeholder for counting tokens; the output tokens are not usable for actual inference (e.g. for submitting a task to engine_queue)
engine_tokens process_mtmd_prompt(
mtmd_context * mctx,
const std::string & prompt,
const std::vector<raw_buffer> & files,
@@ -312,7 +317,7 @@ server_tokens process_mtmd_prompt(
* - "prompt": [[12, 34, 56], [78, 90, 12]]
* - "prompt": [[12, 34, "string", 56, 78], [12, 34, 56], { "prompt_string": "string", "multimodal_data": [ "base64" ]}]
*/
std::vector<server_tokens> tokenize_input_prompts(
std::vector<engine_tokens> tokenize_input_prompts(
const llama_vocab * vocab,
mtmd_context * mctx,
const json & json_prompt,
@@ -321,7 +326,7 @@ std::vector<server_tokens> tokenize_input_prompts(
const mtmd_helper_init_opt & init_opt);
// tokenize a single prompt, see tokenize_input_prompts() for the supported shapes
server_tokens tokenize_input_subprompt(
engine_tokens tokenize_input_subprompt(
const llama_vocab * vocab,
mtmd_context * mctx,
const json & json_prompt,
@@ -334,7 +339,7 @@ server_tokens tokenize_input_subprompt(
//
// global server parameters for chat formatting / parsing
struct server_chat_params {
struct engine_chat_params {
bool use_jinja;
bool prefill_assistant;
common_reasoning_format reasoning_format;
@@ -357,28 +362,28 @@ json oaicompat_completion_params_parse(const json & body);
json oaicompat_chat_params_parse(
const llama_vocab * vocab,
json & body, /* openai api json semantics */
const server_chat_params & opt,
const engine_chat_params & opt,
std::vector<raw_buffer> & out_files,
common_chat_session & out_session);
// used by /embeddings endpoint, content has the same format as a chat message content array
server_tokens tokenize_oai_content_array(
engine_tokens tokenize_oai_content_array(
const llama_vocab * vocab,
mtmd_context * mctx,
const server_chat_params & opt,
const engine_chat_params & opt,
json content,
bool add_special,
bool parse_special,
const mtmd_helper_init_opt & init_opt);
// TODO: move it to server-task.cpp
// TODO: move it to engine-task.cpp
json format_embeddings_response_oaicompat(
const json & request,
const std::string & model_name,
const json & embeddings,
bool use_base64 = false);
// TODO: move it to server-task.cpp
// TODO: move it to engine-task.cpp
json format_response_rerank(
const json & request,
const std::string & model_name,
@@ -391,14 +396,14 @@ json format_response_rerank(
// stats and metrics
//
// shared between server_slot and server_task_result_*
struct server_slot_stats {
// shared between engine_slot and engine_task_result_*
struct engine_slot_stats {
uint64_t n_prompt_cached = 0;
uint64_t n_prompt_processed = 0;
uint64_t n_gen = 0;
// speculative decoding stats
// note: the per-position breakdown lives in server_slot, it is not needed in a task result
// note: the per-position breakdown lives in engine_slot, it is not needed in a task result
uint64_t n_draft_tokens = 0;
uint64_t n_draft_accepted = 0;
uint64_t n_draft_verif_steps = 0;
@@ -478,9 +483,9 @@ struct server_slot_stats {
json to_json() const;
};
// shared between server_context_impl and server_task_result_*
// unlike server_slot_stats, server_metrics is server-global and cumulative, not tied to a slot
struct server_metrics {
// shared between engine_context_impl and engine_task_result_*
// unlike engine_slot_stats, engine_metrics is server-global and cumulative, not tied to a slot
struct engine_metrics {
int64_t t_start = 0;
struct bucket {
@@ -568,7 +573,7 @@ bool is_valid_utf8(const std::string & str);
//
// formatting output responses
// TODO: move these to server-task.cpp
// TODO: move these to engine-task.cpp
//
llama_tokens format_prompt_infill(
@@ -583,126 +588,10 @@ llama_tokens format_prompt_infill(
const llama_tokens & tokens_prompt);
// format rerank task: [BOS]query[EOS][SEP]doc[EOS].
server_tokens format_prompt_rerank(
engine_tokens format_prompt_rerank(
const struct llama_model * model,
const struct llama_vocab * vocab,
mtmd_context * mctx,
const std::string & query,
const std::string & doc,
const mtmd_helper_init_opt & init_opt);
// simple implementation of a pipe
// used for streaming data between threads
template<typename T>
struct server_pipe {
std::mutex mutex;
std::condition_variable cv;
std::queue<T> queue;
std::atomic<bool> writer_closed{false};
std::atomic<bool> reader_closed{false};
// 0 = unbounded (default)
// > 0, write() drops the oldest item once the queue is full
size_t max_size = 0;
void close_write() {
writer_closed.store(true, std::memory_order_relaxed);
cv.notify_all();
}
void close_read() {
reader_closed.store(true, std::memory_order_relaxed);
cv.notify_all();
}
// close_on_stop = true: should_stop means the reader is gone for good, so the writer is told the pipe is broken.
// close_on_stop = false: should_stop is a per-read deadline and further reads still come, so the pipe stays usable.
bool read(T & output, const std::function<bool()> & should_stop, bool close_on_stop = true) {
std::unique_lock<std::mutex> lk(mutex);
constexpr auto poll_interval = std::chrono::milliseconds(500);
while (true) {
if (!queue.empty()) {
output = std::move(queue.front());
queue.pop();
return true;
}
if (writer_closed.load()) {
return false; // clean EOF
}
if (should_stop && should_stop()) { // a null should_stop means "never stop"
if (close_on_stop) {
close_read(); // signal broken pipe to writer
}
return false; // cancelled / deadline reached
}
cv.wait_for(lk, poll_interval);
}
}
bool write(T && data) {
std::lock_guard<std::mutex> lk(mutex);
if (reader_closed.load()) {
return false; // broken pipe
}
if (max_size > 0) {
while (queue.size() >= max_size) {
queue.pop(); // drop oldest to stay bounded
}
}
queue.push(std::move(data));
cv.notify_one();
return true;
}
};
// a child server writes its state commands to stdout and its logs to stderr
enum server_subproc_stream {
SERVER_SUBPROC_STDOUT,
SERVER_SUBPROC_STDERR,
SERVER_SUBPROC_STREAMS,
};
// gives the current stdout to the caller as a stream of its own, and sends everything else written to stdout to stderr
FILE * server_reserve_stdout();
// wrapper around common_subproc to manage a child server process
// mainly used by router mode
struct server_subproc {
common_subproc sproc;
std::atomic<bool> stopped{false}; // set by the monitor once the process exited and was reaped
bool is_alive() { return sproc.alive(); }
void terminate() { sproc.terminate(); }
int join() { return sproc.join(); }
// true if both output pipes of the child are available (call after create())
bool has_output();
// true once both output pipes are closed
bool output_closed() const;
// non-blocking read from one output pipe
// returns the number of bytes read, 0 when nothing is available, -1 when the pipe is closed or broken
int read_output(server_subproc_stream stream, char * buf, size_t len);
// wait until one of a set of children has output, wake() is called, or a timeout passes
struct waiter {
waiter();
~waiter();
// thread-safe; on Windows this is a no-op, wait() returns within 50 ms anyway
void wake();
// timeout_ms < 0 waits until data or wake(); ready[i] is set for each child with data on an open pipe (or a broken pipe)
void wait(const std::vector<server_subproc *> & procs, std::vector<bool> & ready, int64_t timeout_ms);
private:
#ifndef _WIN32
intptr_t wake_fd[2] = { -1, -1 }; // POSIX self-pipe
#endif
};
private:
intptr_t out_handles[SERVER_SUBPROC_STREAMS] = { -1, -1 }; // fd on POSIX, HANDLE on Windows; taken lazily from sproc
bool out_closed [SERVER_SUBPROC_STREAMS] = { false, false };
};
File diff suppressed because it is too large Load Diff
+131
View File
@@ -0,0 +1,131 @@
#pragma once
#include "engine-common.h"
#include "engine-decision.h"
#include "engine-task.h"
#include "engine-queue.h"
#include "json.h"
#include <cstddef>
#include <memory>
#include <mutex>
#include <set>
struct engine_context_impl; // private implementation
constexpr int HTTP_POLLING_SECONDS = 1;
struct engine_context_meta {
std::string build_info;
std::string model_name;
std::set<std::string> model_aliases;
std::set<std::string> model_tags;
std::string model_path;
std::vector<std::string> model_output_modalities; // output modalities for GET /models
bool has_mtmd;
bool has_inp_image;
bool has_inp_audio;
bool has_inp_video;
json json_ui_settings;
int slot_n_ctx;
enum llama_pooling_type pooling_type;
// chat params
engine_chat_params & chat_params;
std::map<std::string, bool> chat_template_caps;
// tokens
std::string bos_token_str;
std::string eos_token_str;
llama_token fim_pre_token;
llama_token fim_sub_token;
llama_token fim_mid_token;
llama_token fim_pad_token;
llama_token fim_rep_token;
llama_token fim_sep_token;
// sampling
std::vector<llama_logit_bias> logit_bias_eog;
// model meta
enum llama_vocab_type model_vocab_type;
int32_t model_vocab_n_tokens;
int32_t model_n_ctx_train;
int32_t model_n_embd_inp;
uint64_t model_n_params;
uint64_t model_size;
std::string model_ftype;
};
enum engine_state {
ENGINE_STATE_DOWNLOADING,
ENGINE_STATE_LOADING,
ENGINE_STATE_READY,
ENGINE_STATE_SLEEPING,
};
static std::string engine_state_to_str(engine_state state) {
switch (state) {
case ENGINE_STATE_DOWNLOADING: return "downloading";
case ENGINE_STATE_LOADING: return "loading";
case ENGINE_STATE_READY: return "ready";
case ENGINE_STATE_SLEEPING: return "sleeping";
default: GGML_ASSERT(false && "invalid engine_state");
}
}
static engine_state engine_state_from_str(const std::string & str) {
if (str == "downloading") return ENGINE_STATE_DOWNLOADING;
if (str == "loading") return ENGINE_STATE_LOADING;
if (str == "ready") return ENGINE_STATE_READY;
if (str == "sleeping") return ENGINE_STATE_SLEEPING;
GGML_ASSERT(false && "invalid engine_state string");
}
using engine_state_callback_t = std::function<void(engine_state, json /* payload */)>;
struct engine_context {
std::unique_ptr<engine_context_impl> impl;
engine_context();
~engine_context();
// load the model and initialize llama_context
// returns true on success
bool load_model(common_params & params);
// this function will block main thread until termination
void start_loop();
// terminate main loop (will unblock start_loop)
void terminate();
// get the underlaying llama_context, can return nullptr if sleeping
// not thread-safe, should only be used from the main thread
llama_context * get_llama_context() const;
// get a new response reader, used by CLI application
engine_response_reader get_response_reader();
// get server metadata (read-only), can only be called after load_model()
// not thread-safe, should only be used from the main thread
engine_context_meta get_meta() const;
// note: must be set before load_model() is called
void set_state_callback(engine_state_callback_t callback);
// used by the HTTP layer
// pointers can only be used when not sleeping, with thread-safe APIs (e.g., tokenizer calls)
llama_model * get_model() const;
const llama_vocab * get_vocab() const;
mtmd_context * get_mtmd_context() const;
const mtmd_helper_init_opt & get_mtmd_init_opt() const;
const engine_decision_context & get_decision() const;
engine_queue & get_queue_tasks();
engine_response & get_queue_results();
engine_metrics get_metrics() const;
void reset_metrics_bucket();
};
File diff suppressed because it is too large Load Diff
+166
View File
@@ -0,0 +1,166 @@
#pragma once
#include "engine-common.h"
#include "engine-task.h"
#include <map>
#include <memory>
#include <string>
#include <vector>
// typed decision models (TypeSafe /v1/systemone API)
// the model answers each question in one forward pass, no token is generated
enum engine_decision_question_type {
ENGINE_DECISION_QUESTION_CHOICE,
ENGINE_DECISION_QUESTION_SCORE,
ENGINE_DECISION_QUESTION_NOUL,
};
struct engine_decision_option {
std::string key;
json description; // null if not provided
};
struct engine_decision_question {
std::string id;
engine_decision_question_type type;
json instructions;
std::vector<engine_decision_option> options; // in the order of the model outputs
};
struct engine_decision_context {
common_decision_type type = COMMON_DECISION_TYPE_NONE;
// read the "<arch>.decision.*" metadata, type stays NONE if the model has none
void init(const llama_model * model);
// true if the questions of a request start with the same tokens, and the model can continue from them
bool can_share_prompt() const {
switch (type) {
case COMMON_DECISION_TYPE_OPENJEV:
case COMMON_DECISION_TYPE_LEV:
case COMMON_DECISION_TYPE_KEV:
case COMMON_DECISION_TYPE_NIMBLE:
case COMMON_DECISION_TYPE_PPLX_DECIDER:
case COMMON_DECISION_TYPE_VLLM_DECISION2:
case COMMON_DECISION_TYPE_LFM2_D1:
return true;
default:
return false;
}
}
// true if all the questions of a request go in one prompt, see fill_task_joint()
bool is_joint() const {
return type == COMMON_DECISION_TYPE_CLEF;
}
// true if the prompt of the model has a place for images
bool can_use_images() const {
switch (type) {
case COMMON_DECISION_TYPE_OPENJEV:
case COMMON_DECISION_TYPE_CLEF:
case COMMON_DECISION_TYPE_PPLX_DECIDER:
case COMMON_DECISION_TYPE_VLLM_D3:
case COMMON_DECISION_TYPE_LFM2_D1:
case COMMON_DECISION_TYPE_LFM2_D1_OMNI:
return true;
default:
return false;
}
}
// throw std::invalid_argument on bad input
std::vector<engine_decision_question> parse_questions(const json & body) const;
// returns the state without its images, they are appended to files in order
// images come from "files" (alias "images") and from the image_url and input_audio parts of a state made of chat messages
// an image can be an audio clip if the model supports it
json parse_state(const json & body, std::vector<raw_buffer> & files) const;
// number of prompts that are evaluated to answer this question, each one shows the options in a different order
size_t n_variants(const engine_decision_question & question) const;
// set the prompt of one variant of this question, and where to read its result
// mctx is only used if there are files
void fill_task(
const json & state,
const std::vector<engine_decision_question> & questions,
const engine_decision_question & question,
size_t variant,
const std::vector<raw_buffer> & files,
mtmd_context * mctx,
const mtmd_helper_init_opt & init_opt,
engine_task & task) const;
// set the prompt of all the questions, the result has the scores of all their options, in order
// mctx is only used if there are files
void fill_task_joint(
const json & state,
const std::vector<engine_decision_question> & questions,
const std::vector<raw_buffer> & files,
mtmd_context * mctx,
const mtmd_helper_init_opt & init_opt,
engine_task & task) const;
// scores: the raw model outputs of each variant
json format_answer(const engine_decision_question & question, const std::vector<std::vector<float>> & scores, bool has_media = false) const;
private:
const llama_vocab * vocab = nullptr;
std::shared_ptr<const common_chat_template> tmpl; // the "systemone" template
std::map<std::string, float> temperatures; // "<type>" or "<type>.<n_options bucket>"
std::map<size_t, std::vector<float>> score_bias; // added to the scores of a score question, by number of levels
size_t n_options_max = 0;
bool noul_true_first = false; // noul options are [true, false] instead of [false, true]
bool noul_criteria_order = false; // noul options are in the order of the criteria keys, if given
bool choice_sorted = false; // choice options are in the order of their keys
bool strict_payload = false; // state, instructions and descriptions must be text, an object or an array
bool canonical_legend = false; // non-string score descriptions are returned as canonical JSON text
// OPENJEV, LEV, NIMBLE, PPLX_DECIDER, VLLM_D3
std::vector<llama_token> labels;
std::vector<std::string> label_texts; // only if the label of an option is given to the template
// LAYA, KEV
llama_token token_marker = LLAMA_TOKEN_NULL;
llama_token token_sep = LLAMA_TOKEN_NULL;
std::string text_marker;
size_t max_head_tokens = 0; // question + options
size_t max_option_tokens = 48;
std::string render(
const json & state,
const std::vector<engine_decision_question> & questions,
const engine_decision_question & question,
size_t variant,
size_t n_images,
bool is_audio = false) const;
json render_options(const engine_decision_question & question, size_t variant) const;
size_t n_outputs(const engine_decision_question & question) const;
// LFM2_D1: label text and tokens of each option
void d1_labels(const engine_decision_question & question, std::vector<std::string> & texts, std::vector<llama_tokens> & groups) const;
void fill_task_laya(llama_tokens & tokens, const engine_decision_question & question, engine_task & task) const;
void fill_task_d1omni(
const json & state,
const std::vector<engine_decision_question> & questions,
const engine_decision_question & question,
size_t variant,
const std::vector<raw_buffer> & files,
mtmd_context * mctx,
const mtmd_helper_init_opt & init_opt,
engine_task & task) const;
float get_temperature(const engine_decision_question & question) const;
double get_decision_confidence(engine_decision_question_type type, const std::vector<double> & probs) const;
};
// group the tasks so that the common prefix of their prompts is evaluated only once
// each group is one parent and its children, it takes at most n_slots slots
// note: the order of the tasks is preserved
std::vector<engine_task> engine_decision_group_tasks(std::vector<engine_task> && tasks, size_t n_slots);
// write "<|name|>" with U+00A6 in place of "|", so text from the request is not parsed as special tokens
std::string engine_decision_escape_special_tokens(const std::string & text);
@@ -1,5 +1,5 @@
#include "server-task.h"
#include "server-queue.h"
#include "engine-task.h"
#include "engine-queue.h"
#include "log.h"
@@ -18,18 +18,18 @@
#define RES_DBG(fmt, ...) LOG_DBG("res %12.*s: " fmt, 12, __func__, __VA_ARGS__)
//
// server_queue
// engine_queue
//
static bool task_resets_idle_timer(server_task_type type) {
return type != SERVER_TASK_TYPE_METRICS;
static bool task_resets_idle_timer(engine_task_type type) {
return type != ENGINE_TASK_TYPE_METRICS;
}
int server_queue::post(server_task && task, bool front) {
int engine_queue::post(engine_task && task, bool front) {
std::unique_lock<std::mutex> lock(mutex_tasks);
GGML_ASSERT(task.id != -1);
// if this is cancel task make sure to clean up pending tasks
if (task.type == SERVER_TASK_TYPE_CANCEL) {
if (task.type == ENGINE_TASK_TYPE_CANCEL) {
cleanup_pending_task(task.id_target);
}
const int task_id = task.id;
@@ -47,7 +47,7 @@ int server_queue::post(server_task && task, bool front) {
return task_id;
}
int server_queue::post(std::vector<server_task> && tasks, bool front) {
int engine_queue::post(std::vector<engine_task> && tasks, bool front) {
std::unique_lock<std::mutex> lock(mutex_tasks);
bool reset_timer = false;
for (auto & task : tasks) {
@@ -55,7 +55,7 @@ int server_queue::post(std::vector<server_task> && tasks, bool front) {
task.id = id++;
}
// if this is cancel task make sure to clean up pending tasks
if (task.type == SERVER_TASK_TYPE_CANCEL) {
if (task.type == ENGINE_TASK_TYPE_CANCEL) {
cleanup_pending_task(task.id_target);
}
reset_timer |= task_resets_idle_timer(task.type);
@@ -73,7 +73,7 @@ int server_queue::post(std::vector<server_task> && tasks, bool front) {
return 0;
}
void server_queue::defer(server_task && task) {
void engine_queue::defer(engine_task && task) {
std::unique_lock<std::mutex> lock(mutex_tasks);
QUE_DBG("defer task, id = %d\n", task.id);
queue_tasks_deferred.push_back(std::move(task));
@@ -81,13 +81,13 @@ void server_queue::defer(server_task && task) {
condition_tasks.notify_one();
}
int server_queue::get_new_id() {
int engine_queue::get_new_id() {
std::unique_lock<std::mutex> lock(mutex_tasks);
int new_id = id++;
return new_id;
}
void server_queue::pop_deferred_task(int id_slot) {
void engine_queue::pop_deferred_task(int id_slot) {
std::unique_lock<std::mutex> lock(mutex_tasks);
if (!queue_tasks_deferred.empty()) {
// try to find a task that uses the specified slot
@@ -112,7 +112,7 @@ void server_queue::pop_deferred_task(int id_slot) {
condition_tasks.notify_one();
}
void server_queue::wait_until_no_sleep() {
void engine_queue::wait_until_no_sleep() {
std::unique_lock<std::mutex> lock(mutex_tasks);
if (!sleeping) {
return;
@@ -129,13 +129,13 @@ void server_queue::wait_until_no_sleep() {
}
}
void server_queue::terminate() {
void engine_queue::terminate() {
std::unique_lock<std::mutex> lock(mutex_tasks);
running = false;
condition_tasks.notify_all();
}
bool server_queue::process_new_tasks(bool is_yielding) {
bool engine_queue::process_new_tasks(bool is_yielding) {
while (true) {
std::unique_lock<std::mutex> lock(mutex_tasks);
if (!running) {
@@ -145,7 +145,7 @@ bool server_queue::process_new_tasks(bool is_yielding) {
if (queue_tasks.empty()) {
return false;
}
server_task task = std::move(queue_tasks.front());
engine_task task = std::move(queue_tasks.front());
queue_tasks.pop_front();
lock.unlock();
@@ -160,7 +160,7 @@ bool server_queue::process_new_tasks(bool is_yielding) {
}
}
void server_queue::worker_loop() {
void engine_queue::worker_loop() {
while (true) {
{
std::unique_lock<std::mutex> lock(mutex_tasks);
@@ -206,7 +206,7 @@ void server_queue::worker_loop() {
}
}
void server_queue::worker_stop() {
void engine_queue::worker_stop() {
if (!worker.thread.joinable()) {
return;
}
@@ -219,7 +219,7 @@ void server_queue::worker_stop() {
worker.thread.join();
}
void server_queue::yield_to_queue(std::function<void()> && work) {
void engine_queue::yield_to_queue(std::function<void()> && work) {
GGML_ASSERT(worker.thread.joinable() && "yield_to_queue() requires start_loop() to be running");
QUE_DBG("%s", "yielding to queue\n");
@@ -275,7 +275,7 @@ void server_queue::yield_to_queue(std::function<void()> && work) {
}
}
void server_queue::start_loop(int64_t idle_sleep_ms) {
void engine_queue::start_loop(int64_t idle_sleep_ms) {
running = true;
time_last_task = ggml_time_ms();
@@ -365,9 +365,9 @@ void server_queue::start_loop(int64_t idle_sleep_ms) {
worker_stop();
}
void server_queue::cleanup_pending_task(int id_target) {
void engine_queue::cleanup_pending_task(int id_target) {
// no need lock because this is called exclusively by post()
auto rm_func = [id_target](const server_task & task) {
auto rm_func = [id_target](const engine_task & task) {
return task.id == id_target;
};
queue_tasks.erase(
@@ -383,17 +383,17 @@ void server_queue::cleanup_pending_task(int id_target) {
}
//
// server_response
// engine_response
//
void server_response::add_waiting_task_id(int id_task) {
void engine_response::add_waiting_task_id(int id_task) {
RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting_task_ids.size());
std::unique_lock<std::mutex> lock(mutex_results);
waiting_task_ids.insert(id_task);
}
void server_response::add_waiting_task_ids(const std::unordered_set<int> & id_tasks) {
void engine_response::add_waiting_task_ids(const std::unordered_set<int> & id_tasks) {
std::unique_lock<std::mutex> lock(mutex_results);
for (const auto & id_task : id_tasks) {
@@ -402,20 +402,20 @@ void server_response::add_waiting_task_ids(const std::unordered_set<int> & id_ta
}
}
void server_response::remove_waiting_task_id(int id_task) {
void engine_response::remove_waiting_task_id(int id_task) {
RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting_task_ids.size());
std::unique_lock<std::mutex> lock(mutex_results);
waiting_task_ids.erase(id_task);
// make sure to clean up all pending results
queue_results.erase(
std::remove_if(queue_results.begin(), queue_results.end(), [id_task](const server_task_result_ptr & res) {
std::remove_if(queue_results.begin(), queue_results.end(), [id_task](const engine_task_result_ptr & res) {
return res->id == id_task;
}),
queue_results.end());
}
void server_response::remove_waiting_task_ids(const std::unordered_set<int> & id_tasks) {
void engine_response::remove_waiting_task_ids(const std::unordered_set<int> & id_tasks) {
std::unique_lock<std::mutex> lock(mutex_results);
for (const auto & id_task : id_tasks) {
@@ -424,7 +424,7 @@ void server_response::remove_waiting_task_ids(const std::unordered_set<int> & id
}
}
server_task_result_ptr server_response::recv(const std::unordered_set<int> & id_tasks) {
engine_task_result_ptr engine_response::recv(const std::unordered_set<int> & id_tasks) {
while (true) {
std::unique_lock<std::mutex> lock(mutex_results);
condition_results.wait(lock, [&]{
@@ -437,7 +437,7 @@ server_task_result_ptr server_response::recv(const std::unordered_set<int> & id_
for (size_t i = 0; i < queue_results.size(); i++) {
if (id_tasks.find(queue_results[i]->id) != id_tasks.end()) {
server_task_result_ptr res = std::move(queue_results[i]);
engine_task_result_ptr res = std::move(queue_results[i]);
queue_results.erase(queue_results.begin() + i);
return res;
}
@@ -447,13 +447,13 @@ server_task_result_ptr server_response::recv(const std::unordered_set<int> & id_
// should never reach here
}
server_task_result_ptr server_response::recv_with_timeout(const std::unordered_set<int> & id_tasks, int timeout) {
engine_task_result_ptr engine_response::recv_with_timeout(const std::unordered_set<int> & id_tasks, int timeout) {
while (true) {
std::unique_lock<std::mutex> lock(mutex_results);
for (int i = 0; i < (int) queue_results.size(); i++) {
if (id_tasks.find(queue_results[i]->id) != id_tasks.end()) {
server_task_result_ptr res = std::move(queue_results[i]);
engine_task_result_ptr res = std::move(queue_results[i]);
queue_results.erase(queue_results.begin() + i);
return res;
}
@@ -472,12 +472,12 @@ server_task_result_ptr server_response::recv_with_timeout(const std::unordered_s
// should never reach here
}
server_task_result_ptr server_response::recv(int id_task) {
engine_task_result_ptr engine_response::recv(int id_task) {
std::unordered_set<int> id_tasks = {id_task};
return recv(id_tasks);
}
void server_response::send(server_task_result_ptr && result) {
void engine_response::send(engine_task_result_ptr && result) {
RES_DBG("sending result for task id = %d\n", result->id);
std::unique_lock<std::mutex> lock(mutex_results);
@@ -492,27 +492,27 @@ void server_response::send(server_task_result_ptr && result) {
}
}
void server_response::broadcast(server_task_result_ptr && result) {
void engine_response::broadcast(engine_task_result_ptr && result) {
std::unique_lock<std::mutex> lock(mutex_results);
for (const auto & id_task : waiting_task_ids) {
RES_DBG("task id = %d pushed to result queue\n", id_task);
server_task_result_ptr res_copy(result->clone());
engine_task_result_ptr res_copy(result->clone());
res_copy->id = id_task; // override id with target task id
queue_results.emplace_back(std::move(res_copy));
}
condition_results.notify_all();
}
void server_response::terminate() {
void engine_response::terminate() {
running = false;
condition_results.notify_all();
}
//
// server_response_reader
// engine_response_reader
//
void server_response_reader::post_task(server_task && task, bool front) {
void engine_response_reader::post_task(engine_task && task, bool front) {
GGML_ASSERT(id_tasks.empty() && "post_task() can only be called once per reader");
GGML_ASSERT(!task.is_parent() && "not supported, use post_tasks() instead");
task.index = 0;
@@ -522,9 +522,9 @@ void server_response_reader::post_task(server_task && task, bool front) {
queue_tasks.post(std::move(task), front);
}
void server_response_reader::post_tasks(std::vector<server_task> && tasks, const common_chat_session & session, bool front) {
void engine_response_reader::post_tasks(std::vector<engine_task> && tasks, const common_chat_session & session, bool front) {
GGML_ASSERT(id_tasks.empty() && "post_tasks() can only be called once per reader");
id_tasks = server_task::get_list_id(tasks);
id_tasks = engine_task::get_list_id(tasks);
states.reserve(tasks.size());
size_t index = 0;
for (auto & task : tasks) {
@@ -541,15 +541,15 @@ void server_response_reader::post_tasks(std::vector<server_task> && tasks, const
queue_tasks.post(std::move(tasks), front);
}
bool server_response_reader::has_next() const {
bool engine_response_reader::has_next() const {
return !cancelled && received_count < id_tasks.size();
}
// return nullptr if should_stop() is true before receiving a result
// note: if one error is received, it will stop further processing and return error result
server_task_result_ptr server_response_reader::next(const std::function<bool()> & should_stop) {
engine_task_result_ptr engine_response_reader::next(const std::function<bool()> & should_stop) {
while (true) {
server_task_result_ptr result = queue_results.recv_with_timeout(id_tasks, polling_interval_seconds);
engine_task_result_ptr result = queue_results.recv_with_timeout(id_tasks, polling_interval_seconds);
if (result == nullptr) {
// timeout, check stop condition
if (should_stop()) {
@@ -577,7 +577,7 @@ server_task_result_ptr server_response_reader::next(const std::function<bool()>
// should not reach here
}
server_response_reader::batch_response server_response_reader::wait_for_all(const std::function<bool()> & should_stop) {
engine_response_reader::batch_response engine_response_reader::wait_for_all(const std::function<bool()> & should_stop) {
batch_response batch_res;
batch_res.results.clear();
batch_res.results.resize(id_tasks.size());
@@ -599,16 +599,16 @@ server_response_reader::batch_response server_response_reader::wait_for_all(cons
return batch_res;
}
void server_response_reader::stop() {
void engine_response_reader::stop() {
queue_results.remove_waiting_task_ids(id_tasks);
if (has_next() && !cancelled) {
// if tasks is not finished yet, cancel them
cancelled = true;
std::vector<server_task> cancel_tasks;
std::vector<engine_task> cancel_tasks;
cancel_tasks.reserve(id_tasks.size());
for (const auto & id_task : id_tasks) {
SRV_WRN("cancel task, id_task = %d\n", id_task);
server_task task(SERVER_TASK_TYPE_CANCEL);
engine_task task(ENGINE_TASK_TYPE_CANCEL);
task.id_target = id_task;
queue_results.remove_waiting_task_id(id_task);
cancel_tasks.push_back(std::move(task));
@@ -1,6 +1,6 @@
#pragma once
#include "server-task.h"
#include "engine-task.h"
#include <condition_variable>
#include <deque>
@@ -11,8 +11,8 @@
#include <unordered_set>
// struct for managing server tasks
// in most cases, use server_response_reader to post new tasks and retrieve results
struct server_queue {
// in most cases, use engine_response_reader to post new tasks and retrieve results
struct engine_queue {
private:
int id = 0;
bool running = false;
@@ -21,11 +21,11 @@ private:
int64_t time_last_task = 0;
// queues
std::deque<server_task> queue_tasks;
std::deque<server_task> queue_tasks_deferred;
std::deque<engine_task> queue_tasks;
std::deque<engine_task> queue_tasks_deferred;
// tasks declined while yielding, put back in queue_tasks once the yield is done
// note: kept as a member so that cleanup_pending_task() can also reach them
std::deque<server_task> queue_tasks_unhandled;
std::deque<engine_task> queue_tasks_unhandled;
std::mutex mutex_tasks;
std::condition_variable condition_tasks;
@@ -42,21 +42,21 @@ private:
worker_t worker;
// callback functions
std::function<bool(server_task &&, bool)> callback_new_task;
std::function<bool(engine_task &&, bool)> callback_new_task;
std::function<void(void)> callback_update_slots;
std::vector<std::function<void(bool)>> callback_sleeping_state;
public:
~server_queue() { worker_stop(); }
~engine_queue() { worker_stop(); }
// Add a new task to the end of the queue
int post(server_task && task, bool front = false);
int post(engine_task && task, bool front = false);
// multi-task version of post()
int post(std::vector<server_task> && tasks, bool front = false);
int post(std::vector<engine_task> && tasks, bool front = false);
// Add a new task, but defer until one slot is available
void defer(server_task && task);
void defer(engine_task && task);
// Get the next id for creating a new task
int get_new_id();
@@ -118,7 +118,7 @@ public:
// only then may the callback return false to decline the task, and it must leave it
// untouched, so that it can be put back in the queue later
// note: while yielding, the callback runs on worker thread, not main thread
void on_new_task(std::function<bool(server_task &&, bool)> callback) {
void on_new_task(std::function<bool(engine_task &&, bool)> callback) {
callback_new_task = std::move(callback);
}
@@ -150,8 +150,8 @@ private:
};
// struct for managing server responses
// in most cases, use server_response_reader to retrieve results
struct server_response {
// in most cases, use engine_response_reader to retrieve results
struct engine_response {
private:
bool running = true;
@@ -159,7 +159,7 @@ private:
std::unordered_set<int> waiting_task_ids;
// the main result queue (using ptr for polymorphism)
std::vector<server_task_result_ptr> queue_results;
std::vector<engine_task_result_ptr> queue_results;
std::mutex mutex_results;
std::condition_variable condition_results;
@@ -177,33 +177,33 @@ public:
void remove_waiting_task_ids(const std::unordered_set<int> & id_tasks);
// This function blocks the thread until there is a response for one of the id_tasks
server_task_result_ptr recv(const std::unordered_set<int> & id_tasks);
engine_task_result_ptr recv(const std::unordered_set<int> & id_tasks);
// same as recv(), but have timeout in seconds
// if timeout is reached, nullptr is returned
server_task_result_ptr recv_with_timeout(const std::unordered_set<int> & id_tasks, int timeout);
engine_task_result_ptr recv_with_timeout(const std::unordered_set<int> & id_tasks, int timeout);
// single-task version of recv()
server_task_result_ptr recv(int id_task);
engine_task_result_ptr recv(int id_task);
// Send a new result to a waiting id_task
void send(server_task_result_ptr && result);
void send(engine_task_result_ptr && result);
// broadcast a new result to all waiting tasks
// (used by router mode)
void broadcast(server_task_result_ptr && result);
void broadcast(engine_task_result_ptr && result);
// terminate the waiting loop
void terminate();
};
// RAII wrapper to make working with server_queue and server_response easier
// RAII wrapper to make working with engine_queue and engine_response easier
// it provides a generator-like API for server responses
// support pooling connection state and aggregating multiple results
struct server_response_reader {
struct engine_response_reader {
std::unordered_set<int> id_tasks;
server_queue & queue_tasks;
server_response & queue_results;
engine_queue & queue_tasks;
engine_response & queue_results;
size_t received_count = 0;
bool cancelled = false;
int polling_interval_seconds;
@@ -213,9 +213,9 @@ struct server_response_reader {
std::vector<task_result_state> states;
// should_stop function will be called each polling_interval_seconds
server_response_reader(server_queue & queue_tasks, server_response & queue_results, int polling_interval_seconds)
engine_response_reader(engine_queue & queue_tasks, engine_response & queue_results, int polling_interval_seconds)
: queue_tasks(queue_tasks), queue_results(queue_results), polling_interval_seconds(polling_interval_seconds) {}
~server_response_reader() {
~engine_response_reader() {
stop();
}
@@ -224,18 +224,18 @@ struct server_response_reader {
}
// if front = true, the task will be posted to the front of the queue (high priority)
void post_task(server_task && task, bool front = false);
void post_tasks(std::vector<server_task> && tasks, const common_chat_session & session = {}, bool front = false);
void post_task(engine_task && task, bool front = false);
void post_tasks(std::vector<engine_task> && tasks, const common_chat_session & session = {}, bool front = false);
bool has_next() const;
// return nullptr if should_stop() is true before receiving a result
// note: if one error is received, it will stop further processing and return error result
server_task_result_ptr next(const std::function<bool()> & should_stop);
engine_task_result_ptr next(const std::function<bool()> & should_stop);
struct batch_response {
bool is_terminated = false; // if true, indicates that processing was stopped before all results were received
std::vector<server_task_result_ptr> results;
server_task_result_ptr error; // nullptr if no error
std::vector<engine_task_result_ptr> results;
engine_task_result_ptr error; // nullptr if no error
};
// aggregate multiple results
batch_response wait_for_all(const std::function<bool()> & should_stop);
@@ -1,8 +1,8 @@
#include "server-schema.h"
#include "engine-schema.h"
#include "json-schema-to-grammar.h"
namespace server_schema {
namespace engine_schema {
//
// llama.cpp-specific completion schema
@@ -324,7 +324,7 @@ std::vector<std::unique_ptr<field>> make_llama_cmpl_schema(const common_params &
->set_handler([&](field_eval_context & ctx, const json & data) {
std::vector<common_grammar_trigger> triggers;
for (const auto & t : data.at("grammar_triggers")) {
triggers.push_back(server_grammar_trigger(t).value);
triggers.push_back(engine_grammar_trigger(t).value);
}
common_sampling_add_grammar_triggers(ctx.params.sampling, ctx.vocab, std::move(triggers));
}));
@@ -600,4 +600,4 @@ void field_nested::eval(field_eval_context & ctx, const json & data) {
}
}
} // namespace server_schema
} // namespace engine_schema
@@ -1,7 +1,7 @@
#pragma once
#include "server-common.h"
#include "server-task.h"
#include "engine-common.h"
#include "engine-task.h"
#include "sampling.h"
#include "speculative.h"
@@ -13,7 +13,7 @@
#include <string>
#include <vector>
namespace server_schema {
namespace engine_schema {
struct field_eval_context {
task_params & params;
@@ -101,4 +101,4 @@ task_params eval_llama_cmpl_schema(
const std::vector<llama_logit_bias> & logit_bias_eog,
const json & data);
} // namespace server_schema
} // namespace engine_schema
@@ -1,14 +1,14 @@
#include "server-task.h"
#include "engine-task.h"
#include "build-info.h"
#include "server-chat.h"
#include "engine-chat.h"
#include "chat.h"
#include "common.h"
#include "json-schema-to-grammar.h"
#include "llama.h"
#include "sampling.h"
#include "speculative.h"
#include "server-common.h"
#include "engine-common.h"
#include <sstream>
@@ -88,7 +88,7 @@ json task_params::to_json(bool only_metrics) const {
auto grammar_triggers = json::array();
for (const auto & trigger : sampling.grammar_triggers) {
server_grammar_trigger ct(trigger);
engine_grammar_trigger ct(trigger);
grammar_triggers.push_back(ct.to_json());
}
@@ -307,9 +307,9 @@ std::vector<unsigned char> completion_token_output::str_to_bytes(const std::stri
}
//
// server_task_result_cmpl_final
// engine_task_result_cmpl_final
//
json server_task_result_cmpl_final::to_json() {
json engine_task_result_cmpl_final::to_json() {
GGML_ASSERT(is_updated && "update() must be called before to_json()");
switch (res_type) {
case TASK_RESPONSE_TYPE_NONE:
@@ -329,7 +329,7 @@ json server_task_result_cmpl_final::to_json() {
}
}
json server_task_result_cmpl_final::to_json_non_oaicompat() {
json engine_task_result_cmpl_final::to_json_non_oaicompat() {
json res = json {
{"index", index},
{"content", content.text},
@@ -354,7 +354,7 @@ json server_task_result_cmpl_final::to_json_non_oaicompat() {
return response_fields.empty() ? res : json_get_nested_values(response_fields, res);
}
json server_task_result_cmpl_final::usage_json_oaicompat() {
json engine_task_result_cmpl_final::usage_json_oaicompat() {
return json {
{"completion_tokens", n_decoded},
{"prompt_tokens", n_prompt_tokens},
@@ -363,7 +363,7 @@ json server_task_result_cmpl_final::usage_json_oaicompat() {
};
}
json server_task_result_cmpl_final::to_json_oaicompat() {
json engine_task_result_cmpl_final::to_json_oaicompat() {
std::time_t t = std::time(0);
json logprobs = json(nullptr); // OAI default to null
if (!stream && probs_output.size() > 0) {
@@ -403,7 +403,7 @@ json server_task_result_cmpl_final::to_json_oaicompat() {
return res;
}
json server_task_result_cmpl_final::to_json_oaicompat_chat() {
json engine_task_result_cmpl_final::to_json_oaicompat_chat() {
std::string finish_reason = "length";
common_chat_msg msg;
if (!oaicompat_msg.empty()) {
@@ -451,7 +451,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_chat() {
return res;
}
json server_task_result_cmpl_final::to_json_oaicompat_chat_stream() {
json engine_task_result_cmpl_final::to_json_oaicompat_chat_stream() {
std::time_t t = std::time(0);
std::string finish_reason = "length";
if (stop == STOP_TYPE_WORD || stop == STOP_TYPE_EOS) {
@@ -465,7 +465,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_chat_stream() {
json {
{"finish_reason", nullptr},
{"index", index},
{"delta", server_chat_msg_diff_to_json_oaicompat(diff)},
{"delta", engine_chat_msg_diff_to_json_oaicompat(diff)},
},
})},
{"created", t},
@@ -517,7 +517,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_chat_stream() {
return deltas;
}
json server_task_result_cmpl_final::to_json_oaicompat_resp() {
json engine_task_result_cmpl_final::to_json_oaicompat_resp() {
common_chat_msg msg;
if (!oaicompat_msg.empty()) {
msg = oaicompat_msg;
@@ -588,7 +588,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_resp() {
return res;
}
json server_task_result_cmpl_final::to_json_oaicompat_resp_stream() {
json engine_task_result_cmpl_final::to_json_oaicompat_resp_stream() {
std::vector<json> server_sent_events;
std::vector<json> output;
@@ -705,7 +705,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_resp_stream() {
return server_sent_events;
}
json server_task_result_cmpl_final::to_json_oaicompat_asr() {
json engine_task_result_cmpl_final::to_json_oaicompat_asr() {
json event = json {
{"type", "transcript.text.done"},
{"text", oaicompat_msg.content},
@@ -720,7 +720,7 @@ json server_task_result_cmpl_final::to_json_oaicompat_asr() {
return event;
}
json server_task_result_cmpl_final::to_json_anthropic() {
json engine_task_result_cmpl_final::to_json_anthropic() {
std::string stop_reason = "max_tokens";
if (stop == STOP_TYPE_WORD || stop == STOP_TYPE_EOS) {
stop_reason = oaicompat_msg.tool_calls.empty() ? "end_turn" : "tool_use";
@@ -786,7 +786,7 @@ json server_task_result_cmpl_final::to_json_anthropic() {
return res;
}
json server_task_result_cmpl_final::to_json_anthropic_stream() {
json engine_task_result_cmpl_final::to_json_anthropic_stream() {
json events = json::array();
std::string stop_reason = "max_tokens";
@@ -975,9 +975,9 @@ json server_task_result_cmpl_final::to_json_anthropic_stream() {
}
//
// server_task_result_cmpl_partial
// engine_task_result_cmpl_partial
//
void server_task_result_cmpl_partial::update(task_result_state & state) {
void engine_task_result_cmpl_partial::update(task_result_state & state) {
is_updated = true;
if (is_begin) {
return; // begin marker only flushes headers, skip parsing
@@ -1015,7 +1015,7 @@ void server_task_result_cmpl_partial::update(task_result_state & state) {
}
}
json server_task_result_cmpl_partial::to_json() {
json engine_task_result_cmpl_partial::to_json() {
GGML_ASSERT(is_updated && "update() must be called before to_json()");
if (is_begin) {
return nullptr; // simply signal to HTTP handler to send the headers and status code
@@ -1038,7 +1038,7 @@ json server_task_result_cmpl_partial::to_json() {
}
}
json server_task_result_cmpl_partial::to_json_non_oaicompat() {
json engine_task_result_cmpl_partial::to_json_non_oaicompat() {
// non-OAI-compat JSON
json res = json {
{"index", index},
@@ -1062,7 +1062,7 @@ json server_task_result_cmpl_partial::to_json_non_oaicompat() {
return res;
}
json server_task_result_cmpl_partial::to_json_oaicompat() {
json engine_task_result_cmpl_partial::to_json_oaicompat() {
std::time_t t = std::time(0);
json logprobs = json(nullptr); // OAI default to null
if (prob_output.probs.size() > 0) {
@@ -1100,7 +1100,7 @@ json server_task_result_cmpl_partial::to_json_oaicompat() {
return res;
}
json server_task_result_cmpl_partial::to_json_oaicompat_chat() {
json engine_task_result_cmpl_partial::to_json_oaicompat_chat() {
bool first = n_decoded == 1;
std::time_t t = std::time(0);
json choices;
@@ -1131,7 +1131,7 @@ json server_task_result_cmpl_partial::to_json_oaicompat_chat() {
}
for (const auto & diff : oaicompat_msg_diffs) {
add_delta(server_chat_msg_diff_to_json_oaicompat(diff));
add_delta(engine_chat_msg_diff_to_json_oaicompat(diff));
}
if (!deltas.empty()) {
@@ -1155,7 +1155,7 @@ json server_task_result_cmpl_partial::to_json_oaicompat_chat() {
return deltas;
}
json server_task_result_cmpl_partial::to_json_oaicompat_resp() {
json engine_task_result_cmpl_partial::to_json_oaicompat_resp() {
std::vector<json> events;
if (!oai_resp_created) {
@@ -1305,7 +1305,7 @@ json server_task_result_cmpl_partial::to_json_oaicompat_resp() {
return events;
}
json server_task_result_cmpl_partial::to_json_oaicompat_asr() {
json engine_task_result_cmpl_partial::to_json_oaicompat_asr() {
json event = json {
{"type", "transcript.text.delta"},
{"delta", content.text},
@@ -1313,7 +1313,7 @@ json server_task_result_cmpl_partial::to_json_oaicompat_asr() {
return event;
}
json server_task_result_cmpl_partial::to_json_anthropic() {
json engine_task_result_cmpl_partial::to_json_anthropic() {
json events = json::array();
bool first = (n_decoded == 1);
// use member variables to track block state across streaming calls
@@ -1453,22 +1453,22 @@ json server_task_result_cmpl_partial::to_json_anthropic() {
}
//
// server_task_result_embd
// engine_task_result_embd
//
json server_task_result_embd::to_json() {
json engine_task_result_embd::to_json() {
return res_type == TASK_RESPONSE_TYPE_OAI_EMBD
? to_json_oaicompat()
: to_json_non_oaicompat();
}
json server_task_result_embd::to_json_non_oaicompat() {
json engine_task_result_embd::to_json_non_oaicompat() {
return json {
{"index", index},
{"embedding", embedding},
};
}
json server_task_result_embd::to_json_oaicompat() {
json engine_task_result_embd::to_json_oaicompat() {
return json {
{"index", index},
{"embedding", embedding[0]},
@@ -1477,9 +1477,9 @@ json server_task_result_embd::to_json_oaicompat() {
}
//
// server_task_result_rerank
// engine_task_result_rerank
//
json server_task_result_rerank::to_json() {
json engine_task_result_rerank::to_json() {
return json {
{"index", index},
{"score", score},
@@ -1488,9 +1488,9 @@ json server_task_result_rerank::to_json() {
}
//
// server_task_result_decision
// engine_task_result_decision
//
json server_task_result_decision::to_json() {
json engine_task_result_decision::to_json() {
return json {
{"index", index},
{"scores", scores},
@@ -1499,9 +1499,9 @@ json server_task_result_decision::to_json() {
}
//
// server_task_result_error
// engine_task_result_error
//
json server_task_result_error::to_json() {
json engine_task_result_error::to_json() {
json res = format_error_response(err_msg, err_type);
if (err_type == ERROR_TYPE_EXCEED_CONTEXT_SIZE) {
res["n_prompt_tokens"] = n_prompt_tokens;
@@ -1511,19 +1511,19 @@ json server_task_result_error::to_json() {
}
//
// server_task_result_metrics
// engine_task_result_metrics
//
json server_task_result_slots::to_json() {
json engine_task_result_slots::to_json() {
return slots_data;
}
json server_task_result_metrics::to_json() {
json engine_task_result_metrics::to_json() {
// not used, /metrics renders prometheus text via to_metrics()
return json{};
}
// metrics definition: https://prometheus.io/docs/practices/naming/#metric-names
std::string server_task_result_metrics::to_metrics() {
std::string engine_task_result_metrics::to_metrics() {
const std::vector<metric_item> counters = {
{
"prompt_tokens_total",
@@ -1620,9 +1620,9 @@ std::string server_task_result_metrics::to_metrics() {
}
//
// server_task_result_slot_save_load
// engine_task_result_slot_save_load
//
json server_task_result_slot_save_load::to_json() {
json engine_task_result_slot_save_load::to_json() {
if (is_save) {
return json {
{ "id_slot", id_slot },
@@ -1647,9 +1647,9 @@ json server_task_result_slot_save_load::to_json() {
}
//
// server_task_result_slot_erase
// engine_task_result_slot_erase
//
json server_task_result_slot_erase::to_json() {
json engine_task_result_slot_erase::to_json() {
return json {
{ "id_slot", id_slot },
{ "n_erased", n_erased },
@@ -1657,10 +1657,10 @@ json server_task_result_slot_erase::to_json() {
}
//
// server_task_result_get_lora
// engine_task_result_get_lora
//
json server_task_result_get_lora::to_json() {
json engine_task_result_get_lora::to_json() {
json result = json::array();
for (size_t i = 0; i < loras.size(); ++i) {
auto & lora = loras[i];
@@ -1681,17 +1681,17 @@ json server_task_result_get_lora::to_json() {
}
//
// server_task_result_apply_lora
// engine_task_result_apply_lora
//
json server_task_result_apply_lora::to_json() {
json engine_task_result_apply_lora::to_json() {
return json {{ "success", true }};
}
//
// server_prompt_cache
// engine_prompt_cache
//
size_t server_prompt_cache::size() const {
size_t engine_prompt_cache::size() const {
size_t res = 0;
for (const auto & state : states) {
@@ -1701,7 +1701,7 @@ size_t server_prompt_cache::size() const {
return res;
}
size_t server_prompt_cache::n_tokens() const {
size_t engine_prompt_cache::n_tokens() const {
size_t res = 0;
for (const auto & state : states) {
@@ -1711,7 +1711,7 @@ size_t server_prompt_cache::n_tokens() const {
return res;
}
server_prompt_cache_state * server_prompt_cache::alloc(const server_prompt & prompt, size_t state_size_tgt, size_t state_size_dft) {
engine_prompt_cache_state * engine_prompt_cache::alloc(const engine_prompt & prompt, size_t state_size_tgt, size_t state_size_dft) {
// first check if the current state is contained fully in the cache
for (auto it = states.begin(); it != states.end(); ++it) {
const int cur_lcp_len = it->prompt.tokens.get_common_prefix(prompt.tokens);
@@ -1793,7 +1793,7 @@ server_prompt_cache_state * server_prompt_cache::alloc(const server_prompt & pro
return &states.back();
}
bool server_prompt_cache::load(server_prompt & prompt, const server_tokens & tokens_new, llama_context * ctx_tgt, llama_context * ctx_dft, int32_t id_slot) {
bool engine_prompt_cache::load(engine_prompt & prompt, const engine_tokens & tokens_new, llama_context * ctx_tgt, llama_context * ctx_dft, int32_t id_slot) {
const int lcp_best = prompt.tokens.get_common_prefix(tokens_new);
float f_keep_best = prompt.tokens.size() > 0 ? float(lcp_best) / prompt.tokens.size() : -1.0f; // empty slot: any cache entry wins
@@ -1870,7 +1870,7 @@ bool server_prompt_cache::load(server_prompt & prompt, const server_tokens & tok
return true;
}
void server_prompt_cache::update() {
void engine_prompt_cache::update() {
if (limit_size > 0) {
while (!states.empty() && size() > limit_size) {
SRV_WRN(" - cache size limit reached, removing oldest entry (size = %.3f MiB)\n", states.front().size() / (1024.0 * 1024.0));
@@ -8,26 +8,26 @@
#include <list>
#include <map>
// TODO: prevent including the whole server-common.h as we only use server_tokens
#include "server-common.h"
// TODO: prevent including the whole engine-common.h as we only use engine_tokens
#include "engine-common.h"
enum server_task_type {
SERVER_TASK_TYPE_COMPLETION,
SERVER_TASK_TYPE_EMBEDDING,
SERVER_TASK_TYPE_RERANK,
SERVER_TASK_TYPE_DECISION,
SERVER_TASK_TYPE_INFILL,
SERVER_TASK_TYPE_CANCEL,
SERVER_TASK_TYPE_CONTROL,
SERVER_TASK_TYPE_NEXT_RESPONSE,
SERVER_TASK_TYPE_METRICS,
SERVER_TASK_TYPE_SLOT_GET,
SERVER_TASK_TYPE_SLOT_SAVE,
SERVER_TASK_TYPE_SLOT_RESTORE,
SERVER_TASK_TYPE_SLOT_ERASE,
SERVER_TASK_TYPE_GET_LORA,
SERVER_TASK_TYPE_SET_LORA,
enum engine_task_type {
ENGINE_TASK_TYPE_COMPLETION,
ENGINE_TASK_TYPE_EMBEDDING,
ENGINE_TASK_TYPE_RERANK,
ENGINE_TASK_TYPE_DECISION,
ENGINE_TASK_TYPE_INFILL,
ENGINE_TASK_TYPE_CANCEL,
ENGINE_TASK_TYPE_CONTROL,
ENGINE_TASK_TYPE_NEXT_RESPONSE,
ENGINE_TASK_TYPE_METRICS,
ENGINE_TASK_TYPE_SLOT_GET,
ENGINE_TASK_TYPE_SLOT_SAVE,
ENGINE_TASK_TYPE_SLOT_RESTORE,
ENGINE_TASK_TYPE_SLOT_ERASE,
ENGINE_TASK_TYPE_GET_LORA,
ENGINE_TASK_TYPE_SET_LORA,
};
// TODO: change this to more generic "response_format" to replace the "format_response_*" in server-common
@@ -85,7 +85,7 @@ struct task_params {
std::string oaicompat_model;
std::string oaicompat_cmpl_id;
// realtime control (SERVER_TASK_TYPE_CONTROL)
// realtime control (ENGINE_TASK_TYPE_CONTROL)
std::string control_action;
std::string control_cmpl_id;
@@ -134,37 +134,37 @@ struct task_result_state {
bool filter_tool_calls = false);
};
struct server_task {
int id = -1; // to be filled by server_queue
struct engine_task {
int id = -1; // to be filled by engine_queue
// TODO @ngxson : remove this field and implement a mapping task_id -> idx in the response_reader
size_t index = 0; // used when there are multiple prompts (batch request)
// used by SERVER_TASK_TYPE_CANCEL
// used by ENGINE_TASK_TYPE_CANCEL
int id_target = -1;
int id_slot = -1;
// used by parallel sampling (multiple completions from same prompt)
int id_parent = -1;
// temporary store of child tasks for scheduling
// note: accessing to elements is invalid after the task is moved to server_slot
std::vector<server_task> child_tasks;
// note: accessing to elements is invalid after the task is moved to engine_slot
std::vector<engine_task> child_tasks;
// if set on a parent, the children have their own prompt and only share its first n_tokens_shared tokens
int32_t n_tokens_shared = 0;
// used by SERVER_TASK_TYPE_INFERENCE
// used by ENGINE_TASK_TYPE_INFERENCE
task_params params;
server_tokens tokens;
engine_tokens tokens;
// only used by CLI, this allow tokenizing CLI inputs on server side
// we need this because mtmd_context and vocab are not accessible outside of server_context
// we need this because mtmd_context and vocab are not accessible outside of engine_context
bool cli = false;
std::string cli_prompt;
std::vector<raw_buffer> cli_files;
server_task_type type;
engine_task_type type;
// used by SERVER_TASK_TYPE_SLOT_SAVE, SERVER_TASK_TYPE_SLOT_RESTORE, SERVER_TASK_TYPE_SLOT_ERASE
// used by ENGINE_TASK_TYPE_SLOT_SAVE, ENGINE_TASK_TYPE_SLOT_RESTORE, ENGINE_TASK_TYPE_SLOT_ERASE
struct slot_action {
int id_slot;
std::string filename;
@@ -172,10 +172,10 @@ struct server_task {
};
slot_action slot_action;
// used by SERVER_TASK_TYPE_METRICS
// used by ENGINE_TASK_TYPE_METRICS
bool metrics_reset_bucket = false;
// used by SERVER_TASK_TYPE_DECISION
// used by ENGINE_TASK_TYPE_DECISION
// where to read the model output of each option, exactly one of the two lists is used
struct decision {
std::vector<llama_token> labels; // logits of these tokens, at the last prompt token
@@ -206,12 +206,12 @@ struct server_task {
};
decision decision;
// used by SERVER_TASK_TYPE_SET_LORA
// used by ENGINE_TASK_TYPE_SET_LORA
std::map<int, float> set_lora; // mapping adapter ID -> scale
server_task() = default;
engine_task() = default;
server_task(server_task_type type) : type(type) {}
engine_task(engine_task_type type) : type(type) {}
int32_t n_tokens() const {
return tokens.size();
@@ -219,10 +219,10 @@ struct server_task {
bool need_embd() const {
switch (type) {
case SERVER_TASK_TYPE_EMBEDDING:
case SERVER_TASK_TYPE_RERANK:
case ENGINE_TASK_TYPE_EMBEDDING:
case ENGINE_TASK_TYPE_RERANK:
return true;
case SERVER_TASK_TYPE_DECISION:
case ENGINE_TASK_TYPE_DECISION:
return !decision.markers.empty() || !decision.order.empty();
default:
return false;
@@ -231,10 +231,10 @@ struct server_task {
bool need_logits() const {
switch (type) {
case SERVER_TASK_TYPE_COMPLETION:
case SERVER_TASK_TYPE_INFILL:
case ENGINE_TASK_TYPE_COMPLETION:
case ENGINE_TASK_TYPE_INFILL:
return true;
case SERVER_TASK_TYPE_DECISION:
case ENGINE_TASK_TYPE_DECISION:
return !decision.labels.empty();
default:
return false;
@@ -243,8 +243,8 @@ struct server_task {
bool need_sampling() const {
switch (type) {
case SERVER_TASK_TYPE_COMPLETION:
case SERVER_TASK_TYPE_INFILL:
case ENGINE_TASK_TYPE_COMPLETION:
case ENGINE_TASK_TYPE_INFILL:
return true;
default:
return false;
@@ -252,7 +252,7 @@ struct server_task {
}
// utility function
static std::unordered_set<int> get_list_id(const std::vector<server_task> & tasks) {
static std::unordered_set<int> get_list_id(const std::vector<engine_task> & tasks) {
std::unordered_set<int> ids(tasks.size());
for (size_t i = 0; i < tasks.size(); i++) {
ids.insert(tasks[i].id);
@@ -275,7 +275,7 @@ struct server_task {
}
void add_child(int id_parent, int id_child) {
server_task copy;
engine_task copy;
copy.id = id_child;
copy.id_parent = id_parent;
@@ -303,7 +303,7 @@ struct server_task {
// embedding and rerank tasks always process their whole prompt, their state is never reused
bool is_stateless() const {
return type == SERVER_TASK_TYPE_EMBEDDING || type == SERVER_TASK_TYPE_RERANK;
return type == ENGINE_TASK_TYPE_EMBEDDING || type == ENGINE_TASK_TYPE_RERANK;
}
};
@@ -316,7 +316,7 @@ struct result_prompt_progress {
json to_json() const;
};
struct server_task_result {
struct engine_task_result {
int id = -1;
int id_slot = -1;
@@ -324,25 +324,25 @@ struct server_task_result {
size_t index = 0; // to be used for batched tasks
virtual bool is_error() {
// only used by server_task_result_error
// only used by engine_task_result_error
return false;
}
virtual bool is_stop() {
// only used by server_task_result_cmpl_*
// only used by engine_task_result_cmpl_*
return true;
}
virtual void update(task_result_state &) {
// only used by server_task_result_cmpl_*
// only used by engine_task_result_cmpl_*
}
virtual json to_json() = 0;
virtual ~server_task_result() = default;
virtual server_task_result * clone() const {
virtual ~engine_task_result() = default;
virtual engine_task_result * clone() const {
GGML_ABORT("not implemented for this task type");
}
};
// using shared_ptr for polymorphism of server_task_result
using server_task_result_ptr = std::unique_ptr<server_task_result>;
// using shared_ptr for polymorphism of engine_task_result
using engine_task_result_ptr = std::unique_ptr<engine_task_result>;
struct completion_token_output {
llama_token tok;
@@ -365,13 +365,13 @@ struct completion_token_output {
};
struct server_task_result_cmpl_final : server_task_result {
struct engine_task_result_cmpl_final : engine_task_result {
common_chat_input content;
llama_tokens tokens;
bool stream;
bool include_usage;
server_slot_stats stats;
engine_slot_stats stats;
std::string prompt;
bool truncated;
@@ -440,7 +440,7 @@ struct server_task_result_cmpl_final : server_task_result {
json to_json_anthropic_stream();
};
struct server_task_result_cmpl_partial : server_task_result {
struct engine_task_result_cmpl_partial : engine_task_result {
common_chat_input content;
llama_tokens tokens;
@@ -453,7 +453,7 @@ struct server_task_result_cmpl_partial : server_task_result {
bool is_begin = false; // whether to send 200 status to HTTP client (begin of SSE stream)
// ref: https://github.com/ggml-org/llama.cpp/pull/23884
completion_token_output prob_output;
server_slot_stats stats;
engine_slot_stats stats;
result_prompt_progress progress;
// response formatting
@@ -499,7 +499,7 @@ struct server_task_result_cmpl_partial : server_task_result {
json to_json_anthropic();
};
struct server_task_result_embd : server_task_result {
struct engine_task_result_embd : engine_task_result {
std::vector<std::vector<float>> embedding;
int32_t n_tokens;
@@ -514,7 +514,7 @@ struct server_task_result_embd : server_task_result {
json to_json_oaicompat();
};
struct server_task_result_rerank : server_task_result {
struct engine_task_result_rerank : engine_task_result {
float score = -1e6;
int32_t n_tokens;
@@ -522,7 +522,7 @@ struct server_task_result_rerank : server_task_result {
virtual json to_json() override;
};
struct server_task_result_decision : server_task_result {
struct engine_task_result_decision : engine_task_result {
std::vector<float> scores; // one raw model output per option
int32_t n_tokens;
@@ -530,7 +530,7 @@ struct server_task_result_decision : server_task_result {
virtual json to_json() override;
};
struct server_task_result_error : server_task_result {
struct engine_task_result_error : engine_task_result {
error_type err_type = ERROR_TYPE_SERVER;
std::string err_msg;
@@ -546,12 +546,12 @@ struct server_task_result_error : server_task_result {
};
// used by /metrics API
struct server_task_result_metrics : server_task_result {
// these are immediate stats, not accumulated (server_metrics is cumulative)
struct engine_task_result_metrics : engine_task_result {
// these are immediate stats, not accumulated (engine_metrics is cumulative)
int n_processing_slots = 0;
int n_tasks_deferred = 0;
server_metrics metrics;
engine_metrics metrics;
virtual json to_json() override;
@@ -564,17 +564,17 @@ struct server_task_result_metrics : server_task_result {
};
// used by /slots API
struct server_task_result_slots : server_task_result {
struct engine_task_result_slots : engine_task_result {
int n_idle_slots = 0;
// while we can also use std::vector<server_slot> this requires copying the slot object which can be quite messy
// while we can also use std::vector<engine_slot> this requires copying the slot object which can be quite messy
// therefore, we use json to temporarily store the slot.to_json() result
json slots_data = json::array();
virtual json to_json() override;
};
struct server_task_result_slot_save_load : server_task_result {
struct engine_task_result_slot_save_load : engine_task_result {
std::string filename;
bool is_save; // true = save, false = load
@@ -585,13 +585,13 @@ struct server_task_result_slot_save_load : server_task_result {
virtual json to_json() override;
};
struct server_task_result_slot_erase : server_task_result {
struct engine_task_result_slot_erase : engine_task_result {
size_t n_erased;
virtual json to_json() override;
};
struct server_task_result_control : server_task_result {
struct engine_task_result_control : engine_task_result {
bool success = false;
std::string message; // optional detail when success is false
@@ -604,7 +604,7 @@ struct server_task_result_control : server_task_result {
}
};
struct server_task_result_get_lora : server_task_result {
struct engine_task_result_get_lora : engine_task_result {
struct lora {
common_adapter_lora_info info;
std::string alora_invocation_string;
@@ -615,12 +615,12 @@ struct server_task_result_get_lora : server_task_result {
virtual json to_json() override;
};
struct server_task_result_apply_lora : server_task_result {
struct engine_task_result_apply_lora : engine_task_result {
virtual json to_json() override;
};
struct server_prompt {
server_tokens tokens;
struct engine_prompt {
engine_tokens tokens;
std::list<common_prompt_checkpoint> checkpoints;
@@ -633,15 +633,15 @@ struct server_prompt {
return tokens.size();
}
server_prompt clone() const {
return server_prompt {
engine_prompt clone() const {
return engine_prompt {
tokens.clone(),
checkpoints,
};
}
};
struct server_prompt_data {
struct engine_prompt_data {
std::vector<uint8_t> main;
std::vector<uint8_t> drft;
@@ -650,9 +650,9 @@ struct server_prompt_data {
}
};
struct server_prompt_cache_state {
server_prompt prompt;
server_prompt_data data;
struct engine_prompt_cache_state {
engine_prompt prompt;
engine_prompt_data data;
size_t size() const {
size_t res = data.size();
@@ -665,13 +665,13 @@ struct server_prompt_cache_state {
}
};
struct server_prompt_cache {
server_prompt_cache(int32_t limit_size_mib, size_t limit_tokens) {
struct engine_prompt_cache {
engine_prompt_cache(int32_t limit_size_mib, size_t limit_tokens) {
this->limit_size = 1024ull*1024ull*(limit_size_mib < 0 ? 0 : limit_size_mib);
this->limit_tokens = limit_tokens;
}
std::list<server_prompt_cache_state> states;
std::list<engine_prompt_cache_state> states;
// in bytes, 0 = no limit
size_t limit_size = 0;
@@ -683,18 +683,18 @@ struct server_prompt_cache {
size_t n_tokens() const;
server_prompt_cache_state * alloc(const server_prompt & prompt, size_t state_size_main, size_t state_size_drft);
engine_prompt_cache_state * alloc(const engine_prompt & prompt, size_t state_size_main, size_t state_size_drft);
bool load(server_prompt & prompt, const server_tokens & tokens_new, llama_context * ctx_tgt, llama_context * ctx_dft, int32_t id_slot);
bool load(engine_prompt & prompt, const engine_tokens & tokens_new, llama_context * ctx_tgt, llama_context * ctx_dft, int32_t id_slot);
void update();
};
// used exclusively by router mode
struct server_task_result_router : server_task_result {
struct engine_task_result_router : engine_task_result {
json data;
virtual json to_json() override { return data; }
virtual server_task_result * clone() const override {
return new server_task_result_router(*this);
virtual engine_task_result * clone() const override {
return new engine_task_result_router(*this);
}
};
+11 -36
View File
@@ -1,40 +1,5 @@
include_directories(${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
# server-context containing the core server logic, used by llama-server and CLI
set(TARGET server-context)
add_library(${TARGET} STATIC
server-chat.cpp
server-chat.h
server-task.cpp
server-task.h
server-queue.cpp
server-queue.h
server-common.cpp
server-common.h
server-decision.cpp
server-decision.h
server-context.cpp
server-context.h
server-stream.cpp
server-stream.h
server-tools.cpp
server-tools.h
server-mcp.cpp
server-mcp.h
server-schema.cpp
server-schema.h
)
if (BUILD_SHARED_LIBS)
set_target_properties(${TARGET} PROPERTIES POSITION_INDEPENDENT_CODE ON)
endif()
target_include_directories(${TARGET} PRIVATE ../mtmd)
target_include_directories(${TARGET} PRIVATE ${PROJECT_SOURCE_DIR})
target_link_libraries(${TARGET} PUBLIC llama-common mtmd ${CMAKE_THREAD_LIBS_INIT})
# llama-server-impl: server logic, reusable by app
set(TARGET llama-server-impl)
@@ -43,6 +8,16 @@ add_library(${TARGET}
server.cpp
server-http.cpp
server-http.h
server-routes.cpp
server-routes.h
server-stream.cpp
server-stream.h
server-subproc.cpp
server-subproc.h
server-tools.cpp
server-tools.h
server-mcp.cpp
server-mcp.h
server-models.cpp
server-models.h
)
@@ -50,7 +25,7 @@ set_target_properties(${TARGET} PROPERTIES WINDOWS_EXPORT_ALL_SYMBOLS ON)
target_include_directories(${TARGET} PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
target_include_directories(${TARGET} PRIVATE ../mtmd ${PROJECT_SOURCE_DIR})
target_link_libraries(${TARGET} PUBLIC server-context llama-ui cpp-httplib ${CMAKE_THREAD_LIBS_INIT})
target_link_libraries(${TARGET} PUBLIC llama-engine llama-ui cpp-httplib ${CMAKE_THREAD_LIBS_INIT})
add_dependencies(${TARGET} llama-ui-assets)
+41 -36
View File
@@ -43,20 +43,25 @@ The server supports two primary operating modes:
- **Inference mode**: The default mode for performing inference with a single loaded GGUF model.
- **Router mode**: Enables management of multiple inference server instances behind a single API endpoint. Requests are automatically routed to the appropriate backend instance based on the requested model.
The code is split into two libraries:
- `libllama-engine` (`tools/engine`): the inference engine, including `engine_context`, task queue, tasks and common utils. It has no HTTP dependency.
- `libllama-server-impl` (`tools/server`): the HTTP layer (`server_http_context`, `server_routes`), router mode, stream replay and MCP.
The core architecture consists of the following components:
- `server_context`: Holds the primary inference state, including the main `llama_context` and all active slots.
- `server_slot`: An abstraction over a single “sequence” in llama.cpp, responsible for managing individual parallel inference requests.
- `server_routes`: Middleware layer between `server_context` and the HTTP interface; handles JSON parsing/formatting and request routing logic.
- `engine_context`: Holds the primary inference state, including the main `llama_context` and all active slots.
- `engine_slot`: An abstraction over a single “sequence” in llama.cpp, responsible for managing individual parallel inference requests.
- `server_routes`: Middleware layer between `engine_context` and the HTTP interface; handles JSON parsing/formatting and request routing logic.
- `server_http_context`: Implements the HTTP server using `cpp-httplib`.
- `server_queue`: Thread-safe queue used by HTTP workers to submit new tasks to `server_context`.
- `server_response`: Thread-safe queue used by `server_context` to return results to HTTP workers.
- `server_response_reader`: Higher-level wrapper around the two queues above for cleaner code.
- `server_task`: Unit of work pushed into `server_queue`.
- `server_task_result`: Unit of result pushed into `server_response`.
- `server_tokens`: Unified representation of token sequences (supports both text and multimodal tokens); used by `server_task` and `server_slot`.
- `engine_queue`: Thread-safe queue used by HTTP workers to submit new tasks to `engine_context`.
- `engine_response`: Thread-safe queue used by `engine_context` to return results to HTTP workers.
- `engine_response_reader`: Higher-level wrapper around the two queues above for cleaner code.
- `engine_task`: Unit of work pushed into `engine_queue`.
- `engine_task_result`: Unit of result pushed into `engine_response`.
- `engine_tokens`: Unified representation of token sequences (supports both text and multimodal tokens); used by `engine_task` and `engine_slot`.
- `server_prompt_checkpoint`: For recurrent (e.g., RWKV) and SWA models, stores snapshots of KV cache state. Enables reuse when subsequent requests share the same prompt prefix, saving redundant computation.
- `server_models`: Standalone component for managing multiple backend instances (used in router mode). It is completely independent of `server_context`.
- `server_models`: Standalone component for managing multiple backend instances (used in router mode). It is completely independent of `engine_context`.
- `stream_session_manager`: process wide owner of resumable SSE stream sessions, keyed by conversation id. A file-static singleton inside `server-stream.cpp`, driven through `server_stream_session_manager_start/stop`. Backs the replay buffer that lets a client reattach to a generation after an HTTP disconnect. See the "Resumable streaming" section below.
```mermaid
@@ -64,13 +69,13 @@ graph TD
API_User <--> server_http_context
server_http_context <-- router mode --> server_models
server_http_context <-- inference mode --> server_routes
server_routes -- server_task --> server_queue
subgraph server_context
server_queue --> server_slot
server_slot -- server_task_result --> server_response
server_slot[multiple server_slot]
server_routes -- engine_task --> engine_queue
subgraph engine_context
engine_queue --> engine_slot
engine_slot -- engine_task_result --> engine_response
engine_slot[multiple engine_slot]
end
server_response --> server_routes
engine_response --> server_routes
```
### Batching
@@ -85,21 +90,21 @@ Following decoding, the system either retrieves embeddings or samples the next t
### Thread Management
`server_context` runs on a dedicated single thread. Because it is single-threaded, heavy post-processing (especially after token generation) should be avoided, as it directly impacts multi-sequence throughput.
`engine_context` runs on a dedicated single thread. Because it is single-threaded, heavy post-processing (especially after token generation) should be avoided, as it directly impacts multi-sequence throughput.
Each incoming HTTP request is handled by its own thread managed by the HTTP library. The following operations are performed in HTTP worker threads:
- JSON request parsing
- Chat template application
- Tokenization
- Conversion of `server_task_result` into final JSON response
- Conversion of `engine_task_result` into final JSON response
- Error formatting into JSON
- Tracking of partial/incremental responses (e.g., streaming tool calls or reasoning steps)
**Best practices to follow:**
- All JSON formatting and chat template logic must stay in the HTTP layer.
- Avoid passing raw JSON between the HTTP layer and `server_slot`. Instead, parse everything into native C++ types as early as possible.
- Avoid passing raw JSON between the HTTP layer and `engine_slot`. Instead, parse everything into native C++ types as early as possible.
### Example trace of a request
@@ -107,13 +112,13 @@ Here is an example trace of an API request for text completion:
- A request arrives at the HTTP layer.
- The request is routed to the corresponding handler inside `server_routes`. In this case, `handle_completions_impl` is invoked.
- The handler parses the input request, constructs a new `server_task`, and passes it to `server_res_generator`.
- The handler parses the input request, constructs a new `engine_task`, and passes it to `server_res_generator`.
- `server_res_generator` creates a new `task_result_state` for each task:
- `task_result_state` stays in the HTTP layer, responsible for keeping track of the current state of the response (e.g., parsing tool calls or thinking messages).
- `server_task` is moved into `server_queue` inside `server_context`.
- `server_context` launches the task by moving it into an available slot (see `launch_slot_with_task()`).
- `engine_task` is moved into `engine_queue` inside `engine_context`.
- `engine_context` launches the task by moving it into an available slot (see `launch_slot_with_task()`).
- `update_slot()` processes the task as described in the "Batching" section above.
- Results may be sent using `send_partial_response` or `send_final_response`, which creates a new `server_task_result` and pushes it to the response queue.
- Results may be sent using `send_partial_response` or `send_final_response`, which creates a new `engine_task_result` and pushes it to the response queue.
- At the same time, `server_res_generator` listens to the response queue and retrieves this response.
- As the response is stateless, `server_res_generator` calls `response->update()` to update the response with the current state.
- `server_res_generator` then calls `response->to_json()` and passes the response to the HTTP layer.
@@ -134,7 +139,7 @@ The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes
Producer side: `server_res_generator` extends `server_res_spipe`, which keeps all spipe logic out of the generic `server_http_res`. `set_req` attaches a producer when the header is present, and the wrapped `next` tees each chunk into the ring before the socket, so a chunk lost to a dead wire is already buffered. While attached, `should_stop` ignores peer disconnect: only a `DELETE` stops generation. On an early peer drop, `on_complete` drains the tail into the ring on the http worker.
Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~server_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind.
Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~engine_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind.
Consumer side: `GET /v1/stream?conv_id=<id>&from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400.
@@ -281,7 +286,7 @@ The main goal of this API is to allow downloading models and/or removing models
Instead of building everything from the ground up (like what most AI agents will do when you ask them to implement a similar feature), we built on top of existing, already well-engineered components inside the codebase:
- Model cache infrastructure as mentioned above (`common/download.h`)
- Server response queue (`server-queue.h`). We use this feature to broadcast events to SSE clients.
- Server response queue (`engine-queue.h`). We use this feature to broadcast events to SSE clients.
- Server router thread management (`server-models.h`). We re-use the same thread model that is used for managing subprocess life cycle, except that we don't create a new subprocess, but launch the download right inside the thread.
The flow for downloading a new model:
@@ -294,27 +299,27 @@ The flow for downloading a new model:
### Sleep mode
Sleep mode was initially introduced in PR [#18228](https://github.com/ggml-org/llama.cpp/pull/18228). The main idea is to have:
- `server_queue` keeping track of the idle timeout
- When the timeout is detected, `server_queue` signals to `server_context_impl` that it should go into sleep
- `server_context_impl` frees all `llama_context` and `mtmd_context`
- `engine_queue` keeping track of the idle timeout
- When the timeout is detected, `engine_queue` signals to `engine_context_impl` that it should go into sleep
- `engine_context_impl` frees all `llama_context` and `mtmd_context`
Compared to simply exiting the whole process, this approach allows accessing some read-only endpoints during sleep, while also handling wakeup-on-request. Any inference request will wake the server up.
Call stack on entering sleeping:
- `server_queue::start_loop` (main thread) sees no task for `idle_sleep_ms` --> `sleeping = true`
- `engine_queue::start_loop` (main thread) sees no task for `idle_sleep_ms` --> `sleeping = true`
- `cb0(true)` --> `server_routes::update_cached_responses`
- snapshots `/props`, `/models` and metrics; the model is still alive here
- `cb1(true)` --> `server_context_impl::handle_sleeping_state`
- `callback_state(SERVER_STATE_SLEEPING)` --> reported to router in child mode
- `cb1(true)` --> `engine_context_impl::handle_sleeping_state`
- `callback_state(ENGINE_STATE_SLEEPING)` --> reported to router in child mode
- `destroy()` --> frees `llama_context` and `mtmd_context`
- `condition_tasks.wait` until `req_stop_sleeping`
Call stack on waking up:
- `server_res_generator` constructor (HTTP thread) --> `server_queue::wait_until_no_sleep`
- `server_res_generator` constructor (HTTP thread) --> `engine_queue::wait_until_no_sleep`
- sets `req_stop_sleeping = true`, then waits until `sleeping == false`
- `server_queue::start_loop` (main thread) wakes up
- `cb1(false)` --> `server_context_impl::handle_sleeping_state`
- `load_model()`, which then emits `callback_state(SERVER_STATE_READY)`
- `engine_queue::start_loop` (main thread) wakes up
- `cb1(false)` --> `engine_context_impl::handle_sleeping_state`
- `load_model()`, which then emits `callback_state(ENGINE_STATE_READY)`
- `cb0(false)` --> `server_routes::update_cached_responses`
- nothing to do, the cache is only read during sleep
- `sleeping = false` --> `notify_all` unblocks the HTTP thread, the request is handled as usual
@@ -325,7 +330,7 @@ Endpoints created with `create_response(true)` (`/health`, `/props`, `/models`,
- Initial server implementation: https://github.com/ggml-org/llama.cpp/pull/1443
- Parallel decoding support: https://github.com/ggml-org/llama.cpp/pull/3228
- Refactor introducing `server_queue` and `server_response`: https://github.com/ggml-org/llama.cpp/pull/5065
- Refactor introducing `server_queue` and `server_response` (now `engine_queue` and `engine_response`): https://github.com/ggml-org/llama.cpp/pull/5065
- Reranking endpoint: https://github.com/ggml-org/llama.cpp/pull/9510
- Multimodal model support (`libmtmd`): https://github.com/ggml-org/llama.cpp/pull/12898
- Unified KV cache handling: https://github.com/ggml-org/llama.cpp/pull/16736
-196
View File
@@ -1,196 +0,0 @@
#pragma once
#include "server-http.h"
#include "server-task.h"
#include "server-queue.h"
#include "json.h"
#include <cstddef>
#include <memory>
#include <mutex>
#include <set>
struct server_context_impl; // private implementation
struct server_context_meta {
std::string build_info;
std::string model_name;
std::set<std::string> model_aliases;
std::set<std::string> model_tags;
std::string model_path;
std::vector<std::string> model_output_modalities; // output modalities for GET /models
bool has_mtmd;
bool has_inp_image;
bool has_inp_audio;
bool has_inp_video;
json json_ui_settings;
int slot_n_ctx;
enum llama_pooling_type pooling_type;
// chat params
server_chat_params & chat_params;
std::map<std::string, bool> chat_template_caps;
// tokens
std::string bos_token_str;
std::string eos_token_str;
llama_token fim_pre_token;
llama_token fim_sub_token;
llama_token fim_mid_token;
llama_token fim_pad_token;
llama_token fim_rep_token;
llama_token fim_sep_token;
// sampling
std::vector<llama_logit_bias> logit_bias_eog;
// model meta
enum llama_vocab_type model_vocab_type;
int32_t model_vocab_n_tokens;
int32_t model_n_ctx_train;
int32_t model_n_embd_inp;
uint64_t model_n_params;
uint64_t model_size;
std::string model_ftype;
};
enum server_state {
SERVER_STATE_DOWNLOADING,
SERVER_STATE_LOADING,
SERVER_STATE_READY,
SERVER_STATE_SLEEPING,
};
static std::string server_state_to_str(server_state state) {
switch (state) {
case SERVER_STATE_DOWNLOADING: return "downloading";
case SERVER_STATE_LOADING: return "loading";
case SERVER_STATE_READY: return "ready";
case SERVER_STATE_SLEEPING: return "sleeping";
default: GGML_ASSERT(false && "invalid server_state");
}
}
static server_state server_state_from_str(const std::string & str) {
if (str == "downloading") return SERVER_STATE_DOWNLOADING;
if (str == "loading") return SERVER_STATE_LOADING;
if (str == "ready") return SERVER_STATE_READY;
if (str == "sleeping") return SERVER_STATE_SLEEPING;
GGML_ASSERT(false && "invalid server_state string");
}
using server_state_callback_t = std::function<void(server_state, json /* payload */)>;
struct server_context {
std::unique_ptr<server_context_impl> impl;
server_context();
~server_context();
// load the model and initialize llama_context
// returns true on success
bool load_model(common_params & params);
// this function will block main thread until termination
void start_loop();
// terminate main loop (will unblock start_loop)
void terminate();
// get the underlaying llama_context, can return nullptr if sleeping
// not thread-safe, should only be used from the main thread
llama_context * get_llama_context() const;
// get a new response reader, used by CLI application
server_response_reader get_response_reader();
// get server metadata (read-only), can only be called after load_model()
// not thread-safe, should only be used from the main thread
server_context_meta get_meta() const;
// note: must be set before load_model() is called
void set_state_callback(server_state_callback_t callback);
};
// forward declarations
struct server_res_generator;
struct server_routes {
server_routes(const common_params & params, server_context & ctx_server);
void init_routes();
// note: this is not thread-safe and can only when ctx_http.is_ready is false
void update_meta(const server_context & ctx_server) {
this->meta = std::make_unique<server_context_meta>(ctx_server.get_meta());
}
// handlers using lambda function, so that they can capture `this` without `std::bind`
// they won't be called until ctx_http.is_ready is set to true
server_http_context::handler_t get_health;
server_http_context::handler_t get_metrics;
server_http_context::handler_t get_slots;
server_http_context::handler_t post_slots;
server_http_context::handler_t get_props;
server_http_context::handler_t post_props;
server_http_context::handler_t post_infill;
server_http_context::handler_t post_completions;
server_http_context::handler_t post_completions_oai;
server_http_context::handler_t post_chat_completions;
server_http_context::handler_t post_chat_completions_tok;
server_http_context::handler_t post_control;
server_http_context::handler_t post_responses_oai;
server_http_context::handler_t post_responses_tok_oai;
server_http_context::handler_t post_transcriptions_oai;
server_http_context::handler_t post_anthropic_messages;
server_http_context::handler_t post_anthropic_count_tokens;
server_http_context::handler_t post_apply_template;
server_http_context::handler_t get_models;
server_http_context::handler_t post_tokenize;
server_http_context::handler_t post_detokenize;
server_http_context::handler_t post_embeddings;
server_http_context::handler_t post_embeddings_oai;
server_http_context::handler_t post_rerank;
server_http_context::handler_t post_systemone;
server_http_context::handler_t get_lora_adapters;
server_http_context::handler_t post_lora_adapters;
// to be used in router mode
json get_model_info() const;
private:
std::unique_ptr<server_res_generator> handle_completions_impl(
const server_http_req & req,
server_task_type type,
const json & data,
const std::vector<raw_buffer> & files,
task_response_type res_type,
const common_chat_session & chat_session = {});
std::unique_ptr<server_res_generator> handle_slots_save(const server_http_req & req, int id_slot);
std::unique_ptr<server_res_generator> handle_slots_restore(const server_http_req & req, int id_slot);
std::unique_ptr<server_res_generator> handle_slots_erase(const server_http_req &, int id_slot);
std::unique_ptr<server_res_generator> handle_embeddings_impl(const server_http_req & req, task_response_type res_type);
std::unique_ptr<server_res_generator> handle_count_tokens(const server_http_req & req, task_response_type res_type);
// using unique_ptr to allow late initialization of const
std::unique_ptr<const server_context_meta> meta;
const common_params & params;
server_context_impl & ctx_server;
server_queue & queue_tasks;
server_response & queue_results;
std::unique_ptr<server_res_generator> create_response(bool bypass_sleep = false);
// cached responses, to be used during sleep
std::mutex mutex_cache;
json cached_models = nullptr;
json cached_props = nullptr;
server_metrics cached_metrics;
// set when a scrape during sleep already reported the throughput buckets
bool should_reset_buckets = false;
// call right before sleep to update the cached responses
void update_cached_responses(bool is_sleeping);
};
+1 -1
View File
@@ -1,7 +1,7 @@
#include "common.h"
#include "http.h"
#include "server-http.h"
#include "server-common.h"
#include "engine-common.h"
#include "ui.h"
#include <cpp-httplib/httplib.h>
+2 -7
View File
@@ -1,5 +1,7 @@
#pragma once
#include "engine-common.h" // raw_buffer, uploaded_file
#include <atomic>
#include <functional>
#include <map>
@@ -38,13 +40,6 @@ struct server_http_res {
// unique pointer, used by set_chunked_content_provider
// httplib requires the stream provider to be stored in heap
using server_http_res_ptr = std::unique_ptr<server_http_res>;
using raw_buffer = std::vector<uint8_t>;
struct uploaded_file {
raw_buffer data;
std::string filename;
std::string content_type;
};
struct server_http_req {
std::map<std::string, std::string> params; // path_params + query_params
+2 -1
View File
@@ -1,6 +1,7 @@
#pragma once
#include "server-common.h"
#include "engine-common.h"
#include "server-subproc.h"
#include <atomic>
#include <chrono>
+19 -19
View File
@@ -1,7 +1,7 @@
#include "server-common.h"
#include "engine-common.h"
#include "http.h"
#include "server-models.h"
#include "server-context.h"
#include "engine-context.h"
#include "server-stream.h"
#include "build-info.h"
@@ -539,7 +539,7 @@ void server_model_meta::update_args(common_preset_context & ctx_preset, std::str
void server_model_meta::update_caps(const common_params & base) {
// reset to the default so a failed refresh cannot keep old values
architecture = server_model_architecture_json(false, false, false, {"text"});
architecture = engine_model_architecture_json(false, false, false, {"text"});
n_ctx_train = 0;
// resolve the model file offline; do not download
@@ -569,7 +569,7 @@ void server_model_meta::update_caps(const common_params & base) {
std::vector<std::string> output_modalities = {"text"};
if (!params.model.path.empty()) {
const common_gguf_info info = common_get_gguf_info(params.model.path);
output_modalities = server_model_output_modalities(info.decision_type);
output_modalities = engine_model_output_modalities(info.decision_type);
n_ctx_train = info.n_ctx_train;
}
@@ -587,7 +587,7 @@ void server_model_meta::update_caps(const common_params & base) {
}
// offline discovery cannot see video; a loaded model reports it
architecture = server_model_architecture_json(inp_image, inp_audio, false, output_modalities);
architecture = engine_model_architecture_json(inp_image, inp_audio, false, output_modalities);
}
//
@@ -687,7 +687,7 @@ void server_models::add_model(server_model_meta && meta) {
}
void server_models::notify_sse(const std::string & event, const std::string & model_id, const json & data) {
std::unique_ptr<server_task_result_router> result = std::make_unique<server_task_result_router>();
std::unique_ptr<engine_task_result_router> result = std::make_unique<engine_task_result_router>();
result->data = {
{"model", model_id},
{"event", event},
@@ -1660,12 +1660,12 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co
}
void server_models::handle_child_state(const std::string & name, const std::string & raw_input) {
server_state state;
engine_state state;
json payload;
try {
json data = json::parse(raw_input.substr(strlen(CMD_CHILD_TO_ROUTER_STATE)));
state = server_state_from_str(json_value(data, "state", std::string()));
state = engine_state_from_str(json_value(data, "state", std::string()));
payload = json_value(data, "payload", json{});
} catch (const std::exception & e) {
SRV_ERR("failed to parse child state update for name=%s: %s\n", name.c_str(), e.what());
@@ -1673,7 +1673,7 @@ void server_models::handle_child_state(const std::string & name, const std::stri
}
switch (state) {
case SERVER_STATE_DOWNLOADING:
case ENGINE_STATE_DOWNLOADING:
{
std::string result = json_value(payload, "result", std::string());
std::string url = json_value(payload, "url", std::string());
@@ -1698,7 +1698,7 @@ void server_models::handle_child_state(const std::string & name, const std::stri
update_download_progress(name, p, false);
}
} break;
case SERVER_STATE_LOADING:
case ENGINE_STATE_LOADING:
{
update_status(name, {
SERVER_MODEL_STATUS_LOADING,
@@ -1707,7 +1707,7 @@ void server_models::handle_child_state(const std::string & name, const std::stri
payload,
});
} break;
case SERVER_STATE_READY:
case ENGINE_STATE_READY:
{
update_status(name, {
SERVER_MODEL_STATUS_LOADED,
@@ -1717,7 +1717,7 @@ void server_models::handle_child_state(const std::string & name, const std::stri
{}, // reset progress info
});
} break;
case SERVER_STATE_SLEEPING:
case ENGINE_STATE_SLEEPING:
{
update_status(name, { SERVER_MODEL_STATUS_SLEEPING });
} break;
@@ -1784,7 +1784,7 @@ struct server_download_state : public common_download_callback {
{"downloaded", p.downloaded},
{"total", p.total},
};
self->notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), data);
self->notify_to_router(engine_state_to_str(ENGINE_STATE_DOWNLOADING), data);
}
void on_start(const common_download_progress & p) override {
on_progress(p);
@@ -1820,7 +1820,7 @@ int server_child::run_download(common_params & params) {
bool ok = dl.run(params);
notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), {
notify_to_router(engine_state_to_str(ENGINE_STATE_DOWNLOADING), {
{"result", ok ? "download_finished" : "download_failed"},
});
@@ -1874,12 +1874,12 @@ void server_child::notify_to_router(const std::string & state, const json & payl
// server_models_routes
//
// RAII wrapper similar to server_response_reader, but doesn't use server_queue
// RAII wrapper similar to engine_response_reader, but doesn't use engine_queue
static std::atomic<int> sse_client_id_counter = 0;
struct server_models_sse_client {
server_response & queue_results;
engine_response & queue_results;
int client_id;
server_models_sse_client(server_response & q)
server_models_sse_client(engine_response & q)
: queue_results(q), client_id(sse_client_id_counter.fetch_add(1, std::memory_order_relaxed)) {
SRV_DBG("new SSE client connected, assigned client_id=%d\n", client_id);
queue_results.add_waiting_task_id(client_id);
@@ -1891,10 +1891,10 @@ struct server_models_sse_client {
// return nullptr if should_stop() is true before receiving a result
// note: if one error is received, it will stop further processing and return error result
server_task_result_ptr next(const std::function<bool()> & should_stop) {
engine_task_result_ptr next(const std::function<bool()> & should_stop) {
while (true) {
static const int http_polling_seconds = 1; // check should_stop every 1 second
server_task_result_ptr result = queue_results.recv_with_timeout({client_id}, http_polling_seconds);
engine_task_result_ptr result = queue_results.recv_with_timeout({client_id}, http_polling_seconds);
if (result == nullptr) {
// timeout, check stop condition
if (should_stop()) {
+5 -4
View File
@@ -3,9 +3,10 @@
#include "common.h"
#include "download.h"
#include "preset.h"
#include "server-common.h"
#include "engine-common.h"
#include "server-http.h"
#include "server-queue.h"
#include "engine-queue.h"
#include "server-subproc.h"
#include <mutex>
#include <condition_variable>
@@ -85,7 +86,7 @@ struct server_model_meta {
int exit_code = 0; // exit code of the model instance process (only valid if status == FAILED)
int stop_timeout = 0; // seconds to wait before force-killing the model instance during shutdown
bool hidden = false; // hidden from GET /models, but still accept if requested
json architecture = server_model_architecture_json(false, false, false, {"text"});
json architecture = engine_model_architecture_json(false, false, false, {"text"});
uint32_t n_ctx_train = 0; // trained context, read from the GGUF metadata; 0 when unknown
bool is_ready() const {
@@ -237,7 +238,7 @@ public:
server_models(const common_params & params, int argc, char ** argv);
~server_models();
server_response sse; // for real-time updates via SSE endpoint
engine_response sse; // for real-time updates via SSE endpoint
// (re-)load the list of models from various sources and prepare the metadata mapping
// - if this is called the first time, simply populate the metadata
File diff suppressed because it is too large Load Diff
+90
View File
@@ -0,0 +1,90 @@
#pragma once
#include "engine-context.h"
#include "server-http.h"
#include "json.h"
#include <memory>
#include <mutex>
// forward declarations
struct server_res_generator;
struct server_routes {
server_routes(const common_params & params, engine_context & ctx_server);
void init_routes();
// note: this is not thread-safe and can only when ctx_http.is_ready is false
void update_meta(const engine_context & ctx_server) {
this->meta = std::make_unique<engine_context_meta>(ctx_server.get_meta());
}
// handlers using lambda function, so that they can capture `this` without `std::bind`
// they won't be called until ctx_http.is_ready is set to true
server_http_context::handler_t get_health;
server_http_context::handler_t get_metrics;
server_http_context::handler_t get_slots;
server_http_context::handler_t post_slots;
server_http_context::handler_t get_props;
server_http_context::handler_t post_props;
server_http_context::handler_t post_infill;
server_http_context::handler_t post_completions;
server_http_context::handler_t post_completions_oai;
server_http_context::handler_t post_chat_completions;
server_http_context::handler_t post_chat_completions_tok;
server_http_context::handler_t post_control;
server_http_context::handler_t post_responses_oai;
server_http_context::handler_t post_responses_tok_oai;
server_http_context::handler_t post_transcriptions_oai;
server_http_context::handler_t post_anthropic_messages;
server_http_context::handler_t post_anthropic_count_tokens;
server_http_context::handler_t post_apply_template;
server_http_context::handler_t get_models;
server_http_context::handler_t post_tokenize;
server_http_context::handler_t post_detokenize;
server_http_context::handler_t post_embeddings;
server_http_context::handler_t post_embeddings_oai;
server_http_context::handler_t post_rerank;
server_http_context::handler_t post_systemone;
server_http_context::handler_t get_lora_adapters;
server_http_context::handler_t post_lora_adapters;
// to be used in router mode
json get_model_info() const;
private:
std::unique_ptr<server_res_generator> handle_completions_impl(
const server_http_req & req,
engine_task_type type,
const json & data,
const std::vector<raw_buffer> & files,
task_response_type res_type,
const common_chat_session & chat_session = {});
std::unique_ptr<server_res_generator> handle_slots_save(const server_http_req & req, int id_slot);
std::unique_ptr<server_res_generator> handle_slots_restore(const server_http_req & req, int id_slot);
std::unique_ptr<server_res_generator> handle_slots_erase(const server_http_req &, int id_slot);
std::unique_ptr<server_res_generator> handle_embeddings_impl(const server_http_req & req, task_response_type res_type);
std::unique_ptr<server_res_generator> handle_count_tokens(const server_http_req & req, task_response_type res_type);
// using unique_ptr to allow late initialization of const
std::unique_ptr<const engine_context_meta> meta;
const common_params & params;
engine_context & ctx_server;
engine_queue & queue_tasks;
engine_response & queue_results;
std::unique_ptr<server_res_generator> create_response(bool bypass_sleep = false);
// cached responses, to be used during sleep
std::mutex mutex_cache;
json cached_models = nullptr;
json cached_props = nullptr;
engine_metrics cached_metrics;
// set when a scrape during sleep already reported the throughput buckets
bool should_reset_buckets = false;
// call right before sleep to update the cached responses
void update_cached_responses(bool is_sleeping);
};
+2 -2
View File
@@ -1,7 +1,7 @@
#include "server-stream.h"
#include "server-common.h"
#include "engine-common.h"
#include "server-http.h"
#include "server-queue.h"
#include "engine-queue.h"
#include <chrono>
#include <memory>
+196
View File
@@ -0,0 +1,196 @@
#include "server-subproc.h"
#include "log.h"
#include <algorithm>
#include <chrono>
#include <cstring>
#include <limits>
#include <thread>
#ifdef _WIN32
// windows.h defines min and max as macros, which breaks std::min and std::max
#define WIN32_LEAN_AND_MEAN
#ifndef NOMINMAX
# define NOMINMAX
#endif
#include <windows.h>
#include <io.h>
#else
#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <unistd.h>
#endif
//
// server_subproc
//
FILE * server_reserve_stdout() {
fflush(stdout);
// the reserved stream is not inherited, grandchildren get stdout and stderr of their own
#ifdef _WIN32
int fd = _dup(_fileno(stdout));
GGML_ASSERT(fd >= 0);
SetHandleInformation((HANDLE) _get_osfhandle(fd), HANDLE_FLAG_INHERIT, 0);
_dup2(_fileno(stderr), _fileno(stdout));
SetStdHandle(STD_OUTPUT_HANDLE, GetStdHandle(STD_ERROR_HANDLE));
FILE * f = _fdopen(fd, "w");
#else
int fd = fcntl(fileno(stdout), F_DUPFD_CLOEXEC, 0);
GGML_ASSERT(fd >= 0);
dup2(fileno(stderr), fileno(stdout));
FILE * f = fdopen(fd, "w");
#endif
GGML_ASSERT(f);
return f;
}
bool server_subproc::has_output() {
FILE * files[SERVER_SUBPROC_STREAMS] = { sproc.stdout_file(), sproc.stderr_file() };
for (int i = 0; i < SERVER_SUBPROC_STREAMS; i++) {
if (out_handles[i] >= 0) {
continue;
}
if (!files[i]) {
return false;
}
#ifdef _WIN32
HANDLE h = (HANDLE) _get_osfhandle(_fileno(files[i]));
if (h == INVALID_HANDLE_VALUE) {
return false;
}
out_handles[i] = (intptr_t) h;
#else
int fd = fileno(files[i]);
if (fd < 0) {
return false;
}
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
out_handles[i] = fd;
#endif
}
return true;
}
bool server_subproc::output_closed() const {
return out_closed[SERVER_SUBPROC_STDOUT] && out_closed[SERVER_SUBPROC_STDERR];
}
int server_subproc::read_output(server_subproc_stream stream, char * buf, size_t len) {
if (out_closed[stream]) {
return -1;
}
if (!has_output()) {
out_closed[stream] = true;
return -1;
}
#ifdef _WIN32
HANDLE h = (HANDLE) out_handles[stream];
DWORD avail = 0;
if (!PeekNamedPipe(h, NULL, 0, NULL, &avail, NULL)) {
out_closed[stream] = true; // pipe broken, child gone
return -1;
}
if (avail == 0) {
return 0;
}
DWORD to_read = avail < (DWORD) len ? avail : (DWORD) len;
DWORD got = 0;
if (!ReadFile(h, buf, to_read, &got, NULL) || got == 0) {
out_closed[stream] = true;
return -1;
}
return (int) got;
#else
while (true) {
ssize_t r = read((int) out_handles[stream], buf, len);
if (r > 0) {
return (int) r;
}
if (r < 0 && errno == EINTR) {
continue;
}
if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
return 0;
}
out_closed[stream] = true; // EOF or error
return -1;
}
#endif
}
server_subproc::waiter::waiter() {
#ifndef _WIN32
int fds[2];
GGML_ASSERT(pipe(fds) == 0);
for (int fd : fds) {
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
}
wake_fd[0] = fds[0];
wake_fd[1] = fds[1];
#endif
}
server_subproc::waiter::~waiter() {
#ifndef _WIN32
close((int) wake_fd[0]);
close((int) wake_fd[1]);
#endif
}
void server_subproc::waiter::wake() {
#ifndef _WIN32
char c = 1;
(void) !write((int) wake_fd[1], &c, 1);
#endif
}
void server_subproc::waiter::wait(const std::vector<server_subproc *> & procs, std::vector<bool> & ready, int64_t timeout_ms) {
ready.assign(procs.size(), false);
#ifdef _WIN32
// no waitable wait exists for anonymous pipes, so poll them in 50 ms steps
bool any = false;
for (size_t i = 0; i < procs.size(); i++) {
server_subproc * p = procs[i];
ready[i] = !p->has_output();
for (int s = 0; s < SERVER_SUBPROC_STREAMS && !ready[i]; s++) {
DWORD avail = 0;
if (!p->out_closed[s] && (!PeekNamedPipe((HANDLE) p->out_handles[s], NULL, 0, NULL, &avail, NULL) || avail > 0)) {
ready[i] = true; // data or broken pipe, read_output() tells which
}
}
any = any || ready[i];
}
if (!any) {
int64_t step = timeout_ms < 0 ? 50 : std::min<int64_t>(timeout_ms, 50);
std::this_thread::sleep_for(std::chrono::milliseconds(step));
}
#else
std::vector<pollfd> pfds;
pfds.reserve(procs.size() * SERVER_SUBPROC_STREAMS + 1);
pfds.push_back({ (int) wake_fd[0], POLLIN, 0 });
for (auto * p : procs) {
const bool open = p->has_output();
for (int s = 0; s < SERVER_SUBPROC_STREAMS; s++) {
pfds.push_back({ open && !p->out_closed[s] ? (int) p->out_handles[s] : -1, POLLIN, 0 }); // poll() skips negative fds
}
}
int timeout = timeout_ms < 0 ? -1 : (int) std::min<int64_t>(timeout_ms, std::numeric_limits<int>::max());
int r = poll(pfds.data(), pfds.size(), timeout);
if (r < 0 && errno != EINTR) {
LOG_ERR("%s: poll() failed: %s\n", __func__, strerror(errno));
}
if (pfds[0].revents) {
char buf[64];
while (read((int) wake_fd[0], buf, sizeof(buf)) > 0) {}
}
for (size_t i = 0; i < procs.size(); i++) {
ready[i] = !procs[i]->has_output();
for (int s = 0; s < SERVER_SUBPROC_STREAMS; s++) {
ready[i] = ready[i] || pfds[1 + i * SERVER_SUBPROC_STREAMS + s].revents != 0;
}
}
#endif
}
+129
View File
@@ -0,0 +1,129 @@
#pragma once
#include "subproc.h"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <cstdio>
#include <functional>
#include <mutex>
#include <queue>
#include <vector>
// simple implementation of a pipe
// used for streaming data between threads
template<typename T>
struct server_pipe {
std::mutex mutex;
std::condition_variable cv;
std::queue<T> queue;
std::atomic<bool> writer_closed{false};
std::atomic<bool> reader_closed{false};
// 0 = unbounded (default)
// > 0, write() drops the oldest item once the queue is full
size_t max_size = 0;
void close_write() {
writer_closed.store(true, std::memory_order_relaxed);
cv.notify_all();
}
void close_read() {
reader_closed.store(true, std::memory_order_relaxed);
cv.notify_all();
}
// close_on_stop = true: should_stop means the reader is gone for good, so the writer is told the pipe is broken.
// close_on_stop = false: should_stop is a per-read deadline and further reads still come, so the pipe stays usable.
bool read(T & output, const std::function<bool()> & should_stop, bool close_on_stop = true) {
std::unique_lock<std::mutex> lk(mutex);
constexpr auto poll_interval = std::chrono::milliseconds(500);
while (true) {
if (!queue.empty()) {
output = std::move(queue.front());
queue.pop();
return true;
}
if (writer_closed.load()) {
return false; // clean EOF
}
if (should_stop && should_stop()) { // a null should_stop means "never stop"
if (close_on_stop) {
close_read(); // signal broken pipe to writer
}
return false; // cancelled / deadline reached
}
cv.wait_for(lk, poll_interval);
}
}
bool write(T && data) {
std::lock_guard<std::mutex> lk(mutex);
if (reader_closed.load()) {
return false; // broken pipe
}
if (max_size > 0) {
while (queue.size() >= max_size) {
queue.pop(); // drop oldest to stay bounded
}
}
queue.push(std::move(data));
cv.notify_one();
return true;
}
};
// a child server writes its state commands to stdout and its logs to stderr
enum server_subproc_stream {
SERVER_SUBPROC_STDOUT,
SERVER_SUBPROC_STDERR,
SERVER_SUBPROC_STREAMS,
};
// gives the current stdout to the caller as a stream of its own, and sends everything else written to stdout to stderr
FILE * server_reserve_stdout();
// wrapper around common_subproc to manage a child server process
// mainly used by router mode
struct server_subproc {
common_subproc sproc;
std::atomic<bool> stopped{false}; // set by the monitor once the process exited and was reaped
bool is_alive() { return sproc.alive(); }
void terminate() { sproc.terminate(); }
int join() { return sproc.join(); }
// true if both output pipes of the child are available (call after create())
bool has_output();
// true once both output pipes are closed
bool output_closed() const;
// non-blocking read from one output pipe
// returns the number of bytes read, 0 when nothing is available, -1 when the pipe is closed or broken
int read_output(server_subproc_stream stream, char * buf, size_t len);
// wait until one of a set of children has output, wake() is called, or a timeout passes
struct waiter {
waiter();
~waiter();
// thread-safe; on Windows this is a no-op, wait() returns within 50 ms anyway
void wake();
// timeout_ms < 0 waits until data or wake(); ready[i] is set for each child with data on an open pipe (or a broken pipe)
void wait(const std::vector<server_subproc *> & procs, std::vector<bool> & ready, int64_t timeout_ms);
private:
#ifndef _WIN32
intptr_t wake_fd[2] = { -1, -1 }; // POSIX self-pipe
#endif
};
private:
intptr_t out_handles[SERVER_SUBPROC_STREAMS] = { -1, -1 }; // fd on POSIX, HANDLE on Windows; taken lazily from sproc
bool out_closed [SERVER_SUBPROC_STREAMS] = { false, false };
};
+2 -2
View File
@@ -1755,7 +1755,7 @@ struct server_tool_get_info : server_tool {
}
};
struct server_tool_stream_result : server_task_result {
struct server_tool_stream_result : engine_task_result {
std::string chunk;
bool done = false;
std::string error_msg;
@@ -1783,7 +1783,7 @@ void server_tool::stream::push(const std::string & chunk) {
struct server_tools_res : server_http_res {
std::thread worker;
server_response * qr = nullptr; // set only for streaming responses
engine_response * qr = nullptr; // set only for streaming responses
int id = -1;
~server_tools_res() override {
+4 -4
View File
@@ -1,8 +1,8 @@
#pragma once
#include "server-common.h"
#include "engine-common.h"
#include "server-http.h"
#include "server-queue.h"
#include "engine-queue.h"
#include "server-mcp.h"
#include <atomic>
@@ -21,7 +21,7 @@ struct server_tool {
virtual std::string type() const { return "server"; }
struct stream {
server_response & qr;
engine_response & qr;
int id;
std::function<bool()> alive;
void push(const std::string & chunk);
@@ -37,7 +37,7 @@ struct server_tools {
std::vector<std::unique_ptr<server_tool>> tools;
// for streaming
server_response queue_res;
engine_response queue_res;
std::atomic<int> res_id{0};
// set when --tools-runtime is configured; routes every tool call through an isolate
+6 -5
View File
@@ -1,4 +1,5 @@
#include "server-context.h"
#include "engine-context.h"
#include "server-routes.h"
#include "server-http.h"
#include "server-models.h"
#include "server-cors-proxy.h"
@@ -195,7 +196,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
server_mcp mcp_mgr;
// struct that contains llama context and inference
server_context ctx_server;
engine_context ctx_server;
//
// Router
@@ -480,8 +481,8 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
// setup communication child --> router if necessary
if (child.is_child()) {
ctx_server.set_state_callback([&](server_state state, json payload) {
child.notify_to_router(server_state_to_str(state), payload);
ctx_server.set_state_callback([&](engine_state state, json payload) {
child.notify_to_router(engine_state_to_str(state), payload);
});
}
@@ -540,7 +541,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
std::thread monitor_thread;
if (child.is_child()) {
monitor_thread = child.setup(shutdown_handler);
child.notify_to_router(server_state_to_str(SERVER_STATE_READY), routes.get_model_info());
child.notify_to_router(engine_state_to_str(ENGINE_STATE_READY), routes.get_model_info());
}
// this call blocks the main thread until queue_tasks.terminate() is called
+1 -1
View File
@@ -96,7 +96,7 @@ def test_slot_restore_legacy_token_list():
assert res.status_code == 200
assert res.body["n_saved"] == 84
# rewrite the token payload into a plain token list, as written by servers that predate the packed server_tokens format
# rewrite the token payload into a plain token list, as written by servers that predate the packed engine_tokens format
path = os.path.join(server.slot_save_path, "slot_legacy.bin")
with open(path, "rb") as f:
data = bytearray(f.read())