diff --git a/backend/cpp/llama-cpp/CMakeLists.txt b/backend/cpp/llama-cpp/CMakeLists.txt index 190ae66bc..4a6ef1dac 100644 --- a/backend/cpp/llama-cpp/CMakeLists.txt +++ b/backend/cpp/llama-cpp/CMakeLists.txt @@ -120,4 +120,9 @@ if(LLAMA_GRPC_BUILD_TESTS) target_include_directories(tts_request_options_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) target_compile_features(tts_request_options_test PRIVATE cxx_std_17) add_test(NAME tts_request_options_test COMMAND tts_request_options_test) + + add_executable(thread_params_test thread_params_test.cpp thread_params.h) + target_include_directories(thread_params_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) + target_compile_features(thread_params_test PRIVATE cxx_std_17) + add_test(NAME thread_params_test COMMAND thread_params_test) endif() diff --git a/backend/cpp/llama-cpp/Makefile b/backend/cpp/llama-cpp/Makefile index 5d238969a..6a9d98a5e 100644 --- a/backend/cpp/llama-cpp/Makefile +++ b/backend/cpp/llama-cpp/Makefile @@ -1,5 +1,5 @@ -LLAMA_VERSION?=84e908c625fb60992b4cdef8180fb12fa9b4c4bf +LLAMA_VERSION?=0021a77de0a8966059dc94548fb3b96654e0bb12 LLAMA_REPO?=https://github.com/ggerganov/llama.cpp CMAKE_ARGS?= diff --git a/backend/cpp/llama-cpp/grpc-server.cpp b/backend/cpp/llama-cpp/grpc-server.cpp index 82080c644..171ae0483 100644 --- a/backend/cpp/llama-cpp/grpc-server.cpp +++ b/backend/cpp/llama-cpp/grpc-server.cpp @@ -53,6 +53,7 @@ #include "arg.h" #include "chat-auto-parser.h" #include "llama_compat.h" // fork-skew switches, generated by prepare.sh +#include "thread_params.h" #include "message_content.h" #include "passthrough_options.h" #include "tts_request_options.h" @@ -1412,6 +1413,12 @@ static void params_parse(server_context& /*ctx_server*/, const backend::ModelOpt passthrough_draft_gpu_layers); } + // The library initializer now creates both threadpools before the server + // can apply llama_context's fallback for the -1 batch-thread sentinel. + params.cpuparams_batch.n_threads = llama_grpc::resolve_batch_threads( + params.cpuparams_batch.n_threads, + params.cpuparams.n_threads); + #ifndef LOCALAI_LLAMA_CPP_NO_SCORE_TASK // Score-task suffix forking: reserve seq ids (and recurrent-state cells) // beyond the slots so one scoring call decodes all candidate tails in a @@ -3595,9 +3602,15 @@ public: // Populate the response with metrics response->set_slot_id(0); response->set_prompt_json_for_slot(""); +#if LOCALAI_HAS_SERVER_METRICS + response->set_tokens_per_second(res_metrics->metrics.prompt_bucket.n_per_second()); + response->set_tokens_generated(res_metrics->metrics.predict.count); + response->set_prompt_tokens_processed(res_metrics->metrics.prompt.count); +#else response->set_tokens_per_second(res_metrics->n_prompt_tokens_processed ? 1.e3 / res_metrics->t_prompt_processing * res_metrics->n_prompt_tokens_processed : 0.); response->set_tokens_generated(res_metrics->n_tokens_predicted_total); response->set_prompt_tokens_processed(res_metrics->n_prompt_tokens_processed_total); +#endif return grpc::Status::OK; diff --git a/backend/cpp/llama-cpp/patches/0001-add-server-task-type-score.patch b/backend/cpp/llama-cpp/patches/0001-add-server-task-type-score.patch index 905248ecb..f056d47ce 100644 --- a/backend/cpp/llama-cpp/patches/0001-add-server-task-type-score.patch +++ b/backend/cpp/llama-cpp/patches/0001-add-server-task-type-score.patch @@ -96,7 +96,7 @@ index 3b5f6a1..d0e18e6 100644 bool has_next_token = true; bool has_new_line = false; bool truncated = false; -@@ -351,6 +378,10 @@ struct server_slot { +@@ -341,6 +368,10 @@ struct server_slot { } generated_tokens.clear(); generated_token_probs.clear(); @@ -106,7 +106,7 @@ index 3b5f6a1..d0e18e6 100644 + score_divergence = -1; json_schema = json(); - // clear speculative decoding stats + task_prev = std::move(task); @@ -2271,6 +2302,229 @@ private: queue_results.send(std::move(res)); } @@ -391,7 +391,7 @@ index 3b5f6a1..d0e18e6 100644 // make a checkpoint of the parts of the memory that cannot be rolled back. // checkpoints are created only if: -@@ -3523,10 +3799,17 @@ private: +@@ -3444,9 +3720,16 @@ private: // embedding requires all tokens in the batch to be output; // MTP also wants logits at every prompt position so the // streaming hook can mirror t_h_nextn into ctx_dft. @@ -404,16 +404,12 @@ index 3b5f6a1..d0e18e6 100644 + slot.prompt.n_tokens() + 1 < slot.task->n_tokens(); add_ok &= batch.add(slot.id, cur_tok, - slot.prompt.tokens.pos_next(), -- slot.need_embd()); -+ slot.need_embd() || need_score_logit); + /* pos = */ slot.prompt.tokens.pos_next(), +- /* output = */ slot.need_embd(), ++ /* output = */ slot.need_embd() || need_score_logit, + /* is_prompt = */ true); slot.prompt.tokens.push_back(cur_tok); - - slot.n_prompt_tokens_processed++; -@@ -3541,6 +3824,32 @@ private: - } - } - +@@ -3454,2 +3737,28 @@ private: + // score tasks: break at the shared-prompt boundary so the checkpoint + // below lands exactly there — the other candidates of the same + // scoring call re-process only their own tokens. Also break at the @@ -440,9 +436,8 @@ index 3b5f6a1..d0e18e6 100644 + } + } + - // process the last few tokens of the prompt separately in order to allow for a checkpoint to be created. - // create checkpoints that many tokens before the end of the prompt: - // - 4 + n_ubatch + // break at the last user message, or at user messages at least min step past the last checkpoint + if (do_checkpoint && spans.is_user_start(slot.prompt.n_tokens())) { @@ -3573,6 +3882,15 @@ private: const bool is_user_start = spans.is_user_start(n_tokens_start); const bool is_last_user_message = n_tokens_start == last_user_pos; diff --git a/backend/cpp/llama-cpp/patches/0002-add-server-task-type-tts.patch b/backend/cpp/llama-cpp/patches/0002-add-server-task-type-tts.patch index 566a62cfe..6681b04e8 100644 --- a/backend/cpp/llama-cpp/patches/0002-add-server-task-type-tts.patch +++ b/backend/cpp/llama-cpp/patches/0002-add-server-task-type-tts.patch @@ -319,7 +319,7 @@ index 9069463fe..b7fa1e534 100644 } auto result = common_speculative_get_output_limits( -@@ -211,6 +213,30 @@ struct server_slot { +@@ -212,6 +214,30 @@ struct server_slot { mtmd_context * mctx = nullptr; mtmd::batch_ptr mbatch = nullptr; @@ -350,7 +350,7 @@ index 9069463fe..b7fa1e534 100644 // speculative decoding common_speculative * spec; -@@ -400,6 +426,8 @@ struct server_slot { +@@ -391,6 +417,8 @@ struct server_slot { // clear multimodal state mbatch.reset(); @@ -359,7 +359,7 @@ index 9069463fe..b7fa1e534 100644 } void init_sampler() const { -@@ -941,6 +969,14 @@ public: +@@ -829,6 +857,14 @@ public: mtmd_context * mctx = nullptr; const llama_vocab * vocab = nullptr; @@ -374,7 +374,7 @@ index 9069463fe..b7fa1e534 100644 server_queue queue_tasks; server_response queue_results; -@@ -1394,6 +1430,10 @@ private: +@@ -1288,6 +1324,10 @@ private: slot.mctx = mctx; slot.prompt.tokens.has_mtmd = mctx != nullptr; @@ -385,7 +385,7 @@ index 9069463fe..b7fa1e534 100644 SLT_TRC(slot, "new slot, n_ctx = %d\n", slot.n_ctx); slot.callback_on_release = [this](int id_slot) { -@@ -1847,6 +1887,28 @@ private: +@@ -1748,6 +1788,28 @@ private: SLT_DBG(slot, "launching slot : %s\n", safe_json_to_str(slot.to_json()).c_str()); @@ -414,7 +414,7 @@ index 9069463fe..b7fa1e534 100644 // initialize samplers if (task.need_sampling()) { try { -@@ -1864,6 +1926,9 @@ private: +@@ -1765,6 +1827,9 @@ private: // TODO: getting pre sampling logits is not yet supported with backend sampling use_backend_sampling &= !need_pre_sample_logits; @@ -424,7 +424,7 @@ index 9069463fe..b7fa1e534 100644 // TODO: tmp until backend sampling is fully implemented if (use_backend_sampling) { llama_set_sampler(ctx_tgt, slot.id, common_sampler_get(slot.smpl.get())); -@@ -1879,9 +1944,13 @@ private: +@@ -1783,9 +1848,13 @@ private: slot.task = std::make_unique(std::move(task)); @@ -441,7 +441,7 @@ index 9069463fe..b7fa1e534 100644 // reset server kill-switch counter n_empty_consecutive = 0; -@@ -2158,6 +2227,18 @@ private: +@@ -2050,6 +2119,18 @@ private: queue_results.send(std::move(res)); } @@ -460,7 +460,7 @@ index 9069463fe..b7fa1e534 100644 void send_final_response(server_slot & slot) { auto res = std::make_unique(); -@@ -2657,6 +2738,7 @@ private: +@@ -2556,6 +2637,7 @@ private: case SERVER_TASK_TYPE_EMBEDDING: case SERVER_TASK_TYPE_RERANK: case SERVER_TASK_TYPE_SCORE: @@ -468,10 +468,7 @@ index 9069463fe..b7fa1e534 100644 { // special case: if input is provided via CLI, tokenize it first // otherwise, no need to tokenize as it's already done inside the HTTP thread -@@ -3092,6 +3174,14 @@ private: - abort_all_slots("pre_decode() failed: " + std::string(e.what())); - } - +@@ -3007,1 +3089,9 @@ private: + // note: TTS slots bypass the shared batch entirely + try { + process_tts_slots(); @@ -481,9 +478,7 @@ index 9069463fe..b7fa1e534 100644 + } + GGML_ASSERT(batch.slot_batched || batch.size() == 0); - - if (batch.slot_batched) { -@@ -3162,10 +3252,77 @@ private: +@@ -3074,10 +3164,77 @@ private: } } @@ -561,7 +556,7 @@ index 9069463fe..b7fa1e534 100644 if (slot.state == SLOT_STATE_GENERATING && slot.prompt.n_tokens() + 1 >= slot.n_ctx) { if (!params_base.ctx_shift) { // this check is redundant (for good) -@@ -3238,7 +3395,7 @@ private: +@@ -3150,7 +3307,7 @@ private: // determine which slots are generating and drafting iterate(slots, [&](server_slot & slot) { @@ -570,7 +565,7 @@ index 9069463fe..b7fa1e534 100644 return; } -@@ -3370,7 +3527,7 @@ private: +@@ -3284,7 +3441,7 @@ private: return; // batch is full, skip remaining slots } @@ -579,7 +574,7 @@ index 9069463fe..b7fa1e534 100644 return; } -@@ -4379,6 +4536,8 @@ server_context_meta server_context::get_meta() const { +@@ -4433,6 +4590,8 @@ server_context_meta server_context::get_meta() const { /* has_inp_image */ impl->chat_params.allow_image, /* has_inp_audio */ impl->chat_params.allow_audio, /* has_inp_video */ impl->chat_params.allow_video, @@ -588,7 +583,7 @@ index 9069463fe..b7fa1e534 100644 /* json_ui_settings */ impl->json_ui_settings, /* slot_n_ctx */ impl->get_slot_n_ctx(), /* pooling_type */ llama_pooling_type(impl->ctx_tgt), -@@ -4458,6 +4617,11 @@ std::unique_ptr server_routes::handle_completions_impl( +@@ -4512,6 +4671,11 @@ std::unique_ptr server_routes::handle_completions_impl( res->set_req(&req); // will also set spipe if needed @@ -600,7 +595,7 @@ index 9069463fe..b7fa1e534 100644 int32_t sse_ping_interval = params.sse_ping_interval; try { -@@ -5435,6 +5599,150 @@ void server_routes::init_routes() { +@@ -5399,6 +5563,150 @@ void server_routes::init_routes() { return res; }; @@ -776,7 +771,7 @@ diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp index 1ee677553..939630b8b 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp -@@ -1523,6 +1523,17 @@ json server_task_result_rerank::to_json() { +@@ -1497,6 +1497,17 @@ json server_task_result_rerank::to_json() { }; } @@ -832,7 +827,7 @@ index 5bedf1987..e6ca67a65 100644 return true; default: return false; -@@ -514,5 +520,15 @@ struct server_task_result_embd : server_task_result { +@@ -494,5 +500,15 @@ struct server_task_result_embd : server_task_result { json to_json_oaicompat(); }; diff --git a/backend/cpp/llama-cpp/prepare.sh b/backend/cpp/llama-cpp/prepare.sh index 4df276fb8..e658a940a 100644 --- a/backend/cpp/llama-cpp/prepare.sh +++ b/backend/cpp/llama-cpp/prepare.sh @@ -32,6 +32,9 @@ cp -r passthrough_options_test.cpp llama.cpp/tools/grpc-server/ # regression test. cp -r tts_request_options.h llama.cpp/tools/grpc-server/ cp -r tts_request_options_test.cpp llama.cpp/tools/grpc-server/ +# Thread-count default normalization and its standalone regression test. +cp -r thread_params.h llama.cpp/tools/grpc-server/ +cp -r thread_params_test.cpp llama.cpp/tools/grpc-server/ # Parent-death watcher (included by grpc-server.cpp) and its standalone unit # test (run via backend/cpp/run-unit-tests.sh; also buildable under ctest). cp -r parent_watch.h llama.cpp/tools/grpc-server/ @@ -53,10 +56,16 @@ else echo "==> llama.cpp predates the load-mode enum, using the legacy mmap/mlock/direct-io booleans" LEGACY_LOAD_MODE=1 fi +if grep -q "server_metrics metrics;" llama.cpp/tools/server/server-task.h; then + HAS_SERVER_METRICS=1 +else + HAS_SERVER_METRICS=0 +fi cat > llama.cpp/tools/grpc-server/llama_compat.h < + +namespace llama_grpc { + +inline int32_t resolve_batch_threads(int32_t batch_threads, int32_t inference_threads) { + return batch_threads < 0 ? inference_threads : batch_threads; +} + +} // namespace llama_grpc diff --git a/backend/cpp/llama-cpp/thread_params_test.cpp b/backend/cpp/llama-cpp/thread_params_test.cpp new file mode 100644 index 000000000..e4aadd9ed --- /dev/null +++ b/backend/cpp/llama-cpp/thread_params_test.cpp @@ -0,0 +1,15 @@ +#include "thread_params.h" + +#include + +int main() { + if (llama_grpc::resolve_batch_threads(-1, 4) != 4) { + std::fprintf(stderr, "default batch threads did not inherit inference threads\n"); + return 1; + } + if (llama_grpc::resolve_batch_threads(2, 4) != 2) { + std::fprintf(stderr, "explicit batch threads were overwritten\n"); + return 1; + } + return 0; +}