server.cpp 139 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460
  1. #include "utils.hpp"
  2. #include "arg.h"
  3. #include "common.h"
  4. #include "json-schema-to-grammar.h"
  5. #include "llama.h"
  6. #include "log.h"
  7. #include "sampling.h"
  8. #include "speculative.h"
  9. // Change JSON_ASSERT from assert() to GGML_ASSERT:
  10. #define JSON_ASSERT GGML_ASSERT
  11. #include "json.hpp"
  12. // mime type for sending response
  13. #define MIMETYPE_JSON "application/json; charset=utf-8"
  14. // auto generated files (update with ./deps.sh)
  15. #include "index.html.hpp"
  16. #include "completion.js.hpp"
  17. #include "loading.html.hpp"
  18. #include "deps_daisyui.min.css.hpp"
  19. #include "deps_markdown-it.js.hpp"
  20. #include "deps_tailwindcss.js.hpp"
  21. #include "deps_vue.esm-browser.js.hpp"
  22. #include <atomic>
  23. #include <condition_variable>
  24. #include <cstddef>
  25. #include <cinttypes>
  26. #include <deque>
  27. #include <memory>
  28. #include <mutex>
  29. #include <signal.h>
  30. #include <thread>
  31. #include <unordered_map>
  32. #include <unordered_set>
  33. using json = nlohmann::ordered_json;
  34. enum stop_type {
  35. STOP_TYPE_FULL,
  36. STOP_TYPE_PARTIAL,
  37. };
  38. // state diagram: https://github.com/ggerganov/llama.cpp/pull/9283
  39. enum slot_state {
  40. SLOT_STATE_IDLE,
  41. SLOT_STATE_STARTED, // TODO: this state is only used for setting up the initial prompt processing; maybe merge it with launch_slot_with_task in the future
  42. SLOT_STATE_PROCESSING_PROMPT,
  43. SLOT_STATE_DONE_PROMPT,
  44. SLOT_STATE_GENERATING,
  45. };
  46. enum server_state {
  47. SERVER_STATE_LOADING_MODEL, // Server is starting up, model not fully loaded yet
  48. SERVER_STATE_READY, // Server is ready and model is loaded
  49. };
  50. enum server_task_type {
  51. SERVER_TASK_TYPE_INFERENCE,
  52. SERVER_TASK_TYPE_CANCEL,
  53. SERVER_TASK_TYPE_NEXT_RESPONSE,
  54. SERVER_TASK_TYPE_METRICS,
  55. SERVER_TASK_TYPE_SLOT_SAVE,
  56. SERVER_TASK_TYPE_SLOT_RESTORE,
  57. SERVER_TASK_TYPE_SLOT_ERASE,
  58. SERVER_TASK_TYPE_SET_LORA,
  59. };
  60. enum server_task_inf_type {
  61. SERVER_TASK_INF_TYPE_COMPLETION,
  62. SERVER_TASK_INF_TYPE_EMBEDDING,
  63. SERVER_TASK_INF_TYPE_RERANK,
  64. SERVER_TASK_INF_TYPE_INFILL,
  65. };
  66. struct server_task {
  67. int id = -1; // to be filled by server_queue
  68. int id_target = -1; // used by SERVER_TASK_TYPE_CANCEL
  69. llama_tokens prompt_tokens;
  70. server_task_type type;
  71. json data;
  72. server_task_inf_type inf_type = SERVER_TASK_INF_TYPE_COMPLETION;
  73. // utility function
  74. static std::unordered_set<int> get_list_id(const std::vector<server_task> & tasks) {
  75. std::unordered_set<int> ids(tasks.size());
  76. for (size_t i = 0; i < tasks.size(); i++) {
  77. ids.insert(tasks[i].id);
  78. }
  79. return ids;
  80. }
  81. };
  82. struct server_task_result {
  83. int id = -1;
  84. json data;
  85. bool stop;
  86. bool error;
  87. };
  88. struct server_static_file {
  89. const unsigned char * data;
  90. unsigned int size;
  91. const char * mime_type;
  92. };
  93. struct slot_params {
  94. bool stream = true;
  95. bool cache_prompt = true; // remember the prompt to avoid reprocessing all prompt
  96. int32_t n_keep = 0; // number of tokens to keep from initial prompt
  97. int32_t n_discard = 0; // number of tokens after n_keep that may be discarded when shifting context, 0 defaults to half
  98. int32_t n_predict = -1; // new tokens to predict
  99. int32_t n_indent = 0; // mininum line indentation for the generated text in number of whitespace characters
  100. int64_t t_max_prompt_ms = -1; // TODO: implement
  101. int64_t t_max_predict_ms = -1; // if positive, limit the generation phase to this time limit
  102. std::vector<std::string> antiprompt;
  103. struct common_params_sampling sampling;
  104. struct common_params_speculative speculative;
  105. };
  106. struct server_slot {
  107. int id;
  108. int id_task = -1;
  109. llama_batch batch_spec;
  110. llama_context * ctx_dft = nullptr;
  111. common_speculative * spec = nullptr;
  112. // the index relative to completion multi-task request
  113. size_t index = 0;
  114. struct slot_params params;
  115. slot_state state = SLOT_STATE_IDLE;
  116. // used to determine the slot that has been used the longest
  117. int64_t t_last_used = -1;
  118. // generation props
  119. int32_t n_ctx = 0; // context size per slot
  120. int32_t n_past = 0;
  121. int32_t n_decoded = 0;
  122. int32_t n_remaining = -1;
  123. int32_t i_batch = -1;
  124. int32_t n_predict = -1; // TODO: disambiguate from params.n_predict
  125. // n_prompt_tokens may not be equal to prompt_tokens.size(), because prompt maybe truncated
  126. int32_t n_prompt_tokens = 0;
  127. int32_t n_prompt_tokens_processed = 0;
  128. // input prompt tokens
  129. llama_tokens prompt_tokens;
  130. size_t last_nl_pos = 0;
  131. std::string generated_text;
  132. llama_tokens cache_tokens;
  133. std::vector<completion_token_output> generated_token_probs;
  134. server_task_inf_type inf_type = SERVER_TASK_INF_TYPE_COMPLETION;
  135. bool has_next_token = true;
  136. bool has_new_line = false;
  137. bool truncated = false;
  138. bool stopped_eos = false;
  139. bool stopped_word = false;
  140. bool stopped_limit = false;
  141. bool timings_per_token = false;
  142. bool oaicompat = false;
  143. std::string oaicompat_model;
  144. std::string stopping_word;
  145. // sampling
  146. json json_schema;
  147. struct common_sampler * smpl = nullptr;
  148. llama_token sampled;
  149. // stats
  150. size_t n_sent_text = 0; // number of sent text character
  151. size_t n_sent_token_probs = 0;
  152. int64_t t_start_process_prompt;
  153. int64_t t_start_generation;
  154. double t_prompt_processing; // ms
  155. double t_token_generation; // ms
  156. std::function<void(int)> callback_on_release;
  157. void reset() {
  158. SLT_DBG(*this, "%s", "\n");
  159. n_prompt_tokens = 0;
  160. last_nl_pos = 0;
  161. generated_text = "";
  162. has_new_line = false;
  163. truncated = false;
  164. stopped_eos = false;
  165. stopped_word = false;
  166. stopped_limit = false;
  167. stopping_word = "";
  168. n_past = 0;
  169. n_sent_text = 0;
  170. n_sent_token_probs = 0;
  171. inf_type = SERVER_TASK_INF_TYPE_COMPLETION;
  172. generated_token_probs.clear();
  173. }
  174. bool has_budget(const common_params & global_params) {
  175. if (params.n_predict == -1 && global_params.n_predict == -1) {
  176. return true; // limitless
  177. }
  178. n_remaining = -1;
  179. if (params.n_predict != -1) {
  180. n_remaining = params.n_predict - n_decoded;
  181. } else if (global_params.n_predict != -1) {
  182. n_remaining = global_params.n_predict - n_decoded;
  183. }
  184. return n_remaining > 0; // no budget
  185. }
  186. bool is_processing() const {
  187. return state != SLOT_STATE_IDLE;
  188. }
  189. bool can_speculate() const {
  190. return ctx_dft && params.speculative.n_max > 0 && params.cache_prompt;
  191. }
  192. void add_token(const completion_token_output & token) {
  193. if (!is_processing()) {
  194. SLT_WRN(*this, "%s", "slot is not processing\n");
  195. return;
  196. }
  197. generated_token_probs.push_back(token);
  198. }
  199. void release() {
  200. if (is_processing()) {
  201. SLT_INF(*this, "stop processing: n_past = %d, truncated = %d\n", n_past, truncated);
  202. t_last_used = ggml_time_us();
  203. t_token_generation = (ggml_time_us() - t_start_generation) / 1e3;
  204. state = SLOT_STATE_IDLE;
  205. callback_on_release(id);
  206. }
  207. }
  208. json get_formated_timings() const {
  209. return json {
  210. {"prompt_n", n_prompt_tokens_processed},
  211. {"prompt_ms", t_prompt_processing},
  212. {"prompt_per_token_ms", t_prompt_processing / n_prompt_tokens_processed},
  213. {"prompt_per_second", 1e3 / t_prompt_processing * n_prompt_tokens_processed},
  214. {"predicted_n", n_decoded},
  215. {"predicted_ms", t_token_generation},
  216. {"predicted_per_token_ms", t_token_generation / n_decoded},
  217. {"predicted_per_second", 1e3 / t_token_generation * n_decoded},
  218. };
  219. }
  220. size_t find_stopping_strings(const std::string & text, const size_t last_token_size, const stop_type type) {
  221. size_t stop_pos = std::string::npos;
  222. for (const std::string & word : params.antiprompt) {
  223. size_t pos;
  224. if (type == STOP_TYPE_FULL) {
  225. const size_t tmp = word.size() + last_token_size;
  226. const size_t from_pos = text.size() > tmp ? text.size() - tmp : 0;
  227. pos = text.find(word, from_pos);
  228. } else {
  229. pos = find_partial_stop_string(word, text);
  230. }
  231. if (pos != std::string::npos && (stop_pos == std::string::npos || pos < stop_pos)) {
  232. if (type == STOP_TYPE_FULL) {
  233. stopped_word = true;
  234. stopping_word = word;
  235. has_next_token = false;
  236. }
  237. stop_pos = pos;
  238. }
  239. }
  240. return stop_pos;
  241. }
  242. void print_timings() const {
  243. const double t_prompt = t_prompt_processing / n_prompt_tokens_processed;
  244. const double n_prompt_second = 1e3 / t_prompt_processing * n_prompt_tokens_processed;
  245. const double t_gen = t_token_generation / n_decoded;
  246. const double n_gen_second = 1e3 / t_token_generation * n_decoded;
  247. SLT_INF(*this,
  248. "\n"
  249. "\rprompt eval time = %10.2f ms / %5d tokens (%8.2f ms per token, %8.2f tokens per second)\n"
  250. "\r eval time = %10.2f ms / %5d tokens (%8.2f ms per token, %8.2f tokens per second)\n"
  251. "\r total time = %10.2f ms / %5d tokens\n",
  252. t_prompt_processing, n_prompt_tokens_processed, t_prompt, n_prompt_second,
  253. t_token_generation, n_decoded, t_gen, n_gen_second,
  254. t_prompt_processing + t_token_generation, n_prompt_tokens_processed + n_decoded);
  255. }
  256. };
  257. struct server_metrics {
  258. int64_t t_start = 0;
  259. uint64_t n_prompt_tokens_processed_total = 0;
  260. uint64_t t_prompt_processing_total = 0;
  261. uint64_t n_tokens_predicted_total = 0;
  262. uint64_t t_tokens_generation_total = 0;
  263. uint64_t n_prompt_tokens_processed = 0;
  264. uint64_t t_prompt_processing = 0;
  265. uint64_t n_tokens_predicted = 0;
  266. uint64_t t_tokens_generation = 0;
  267. uint64_t n_decode_total = 0;
  268. uint64_t n_busy_slots_total = 0;
  269. void init() {
  270. t_start = ggml_time_us();
  271. }
  272. void on_prompt_eval(const server_slot & slot) {
  273. n_prompt_tokens_processed_total += slot.n_prompt_tokens_processed;
  274. n_prompt_tokens_processed += slot.n_prompt_tokens_processed;
  275. t_prompt_processing += slot.t_prompt_processing;
  276. t_prompt_processing_total += slot.t_prompt_processing;
  277. }
  278. void on_prediction(const server_slot & slot) {
  279. n_tokens_predicted_total += slot.n_decoded;
  280. n_tokens_predicted += slot.n_decoded;
  281. t_tokens_generation += slot.t_token_generation;
  282. t_tokens_generation_total += slot.t_token_generation;
  283. }
  284. void on_decoded(const std::vector<server_slot> & slots) {
  285. n_decode_total++;
  286. for (const auto & slot : slots) {
  287. if (slot.is_processing()) {
  288. n_busy_slots_total++;
  289. }
  290. }
  291. }
  292. void reset_bucket() {
  293. n_prompt_tokens_processed = 0;
  294. t_prompt_processing = 0;
  295. n_tokens_predicted = 0;
  296. t_tokens_generation = 0;
  297. }
  298. };
  299. struct server_queue {
  300. int id = 0;
  301. bool running;
  302. // queues
  303. std::deque<server_task> queue_tasks;
  304. std::deque<server_task> queue_tasks_deferred;
  305. std::mutex mutex_tasks;
  306. std::condition_variable condition_tasks;
  307. // callback functions
  308. std::function<void(server_task)> callback_new_task;
  309. std::function<void(void)> callback_update_slots;
  310. // Add a new task to the end of the queue
  311. int post(server_task task, bool front = false) {
  312. std::unique_lock<std::mutex> lock(mutex_tasks);
  313. if (task.id == -1) {
  314. task.id = id++;
  315. }
  316. QUE_DBG("new task, id = %d, front = %d\n", task.id, front);
  317. if (front) {
  318. queue_tasks.push_front(std::move(task));
  319. } else {
  320. queue_tasks.push_back(std::move(task));
  321. }
  322. condition_tasks.notify_one();
  323. return task.id;
  324. }
  325. // multi-task version of post()
  326. int post(std::vector<server_task> & tasks, bool front = false) {
  327. std::unique_lock<std::mutex> lock(mutex_tasks);
  328. for (auto & task : tasks) {
  329. if (task.id == -1) {
  330. task.id = id++;
  331. }
  332. QUE_DBG("new task, id = %d/%d, front = %d\n", task.id, (int) tasks.size(), front);
  333. if (front) {
  334. queue_tasks.push_front(std::move(task));
  335. } else {
  336. queue_tasks.push_back(std::move(task));
  337. }
  338. }
  339. condition_tasks.notify_one();
  340. return 0;
  341. }
  342. // Add a new task, but defer until one slot is available
  343. void defer(server_task task) {
  344. std::unique_lock<std::mutex> lock(mutex_tasks);
  345. QUE_DBG("defer task, id = %d\n", task.id);
  346. queue_tasks_deferred.push_back(std::move(task));
  347. condition_tasks.notify_one();
  348. }
  349. // Get the next id for creating a new task
  350. int get_new_id() {
  351. std::unique_lock<std::mutex> lock(mutex_tasks);
  352. int new_id = id++;
  353. return new_id;
  354. }
  355. // Register function to process a new task
  356. void on_new_task(std::function<void(server_task)> callback) {
  357. callback_new_task = std::move(callback);
  358. }
  359. // Register the function to be called when all slots data is ready to be processed
  360. void on_update_slots(std::function<void(void)> callback) {
  361. callback_update_slots = std::move(callback);
  362. }
  363. // Call when the state of one slot is changed, it will move one task from deferred to main queue
  364. void pop_deferred_task() {
  365. std::unique_lock<std::mutex> lock(mutex_tasks);
  366. if (!queue_tasks_deferred.empty()) {
  367. queue_tasks.emplace_back(std::move(queue_tasks_deferred.front()));
  368. queue_tasks_deferred.pop_front();
  369. }
  370. condition_tasks.notify_one();
  371. }
  372. // end the start_loop routine
  373. void terminate() {
  374. std::unique_lock<std::mutex> lock(mutex_tasks);
  375. running = false;
  376. condition_tasks.notify_all();
  377. }
  378. /**
  379. * Main loop consists of these steps:
  380. * - Wait until a new task arrives
  381. * - Process the task (i.e. maybe copy data into slot)
  382. * - Check if multitask is finished
  383. * - Update all slots
  384. */
  385. void start_loop() {
  386. running = true;
  387. while (true) {
  388. QUE_DBG("%s", "processing new tasks\n");
  389. while (true) {
  390. std::unique_lock<std::mutex> lock(mutex_tasks);
  391. if (queue_tasks.empty()) {
  392. lock.unlock();
  393. break;
  394. }
  395. server_task task = queue_tasks.front();
  396. queue_tasks.pop_front();
  397. lock.unlock();
  398. QUE_DBG("processing task, id = %d\n", task.id);
  399. callback_new_task(std::move(task));
  400. }
  401. // all tasks in the current loop is processed, slots data is now ready
  402. QUE_DBG("%s", "update slots\n");
  403. callback_update_slots();
  404. QUE_DBG("%s", "waiting for new tasks\n");
  405. {
  406. std::unique_lock<std::mutex> lock(mutex_tasks);
  407. if (queue_tasks.empty()) {
  408. if (!running) {
  409. QUE_DBG("%s", "terminate\n");
  410. return;
  411. }
  412. condition_tasks.wait(lock, [&]{
  413. return (!queue_tasks.empty() || !running);
  414. });
  415. }
  416. }
  417. }
  418. }
  419. };
  420. struct server_response {
  421. // for keeping track of all tasks waiting for the result
  422. std::unordered_set<int> waiting_task_ids;
  423. // the main result queue
  424. std::vector<server_task_result> queue_results;
  425. std::mutex mutex_results;
  426. std::condition_variable condition_results;
  427. // add the id_task to the list of tasks waiting for response
  428. void add_waiting_task_id(int id_task) {
  429. SRV_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting_task_ids.size());
  430. std::unique_lock<std::mutex> lock(mutex_results);
  431. waiting_task_ids.insert(id_task);
  432. }
  433. void add_waiting_tasks(const std::vector<server_task> & tasks) {
  434. std::unique_lock<std::mutex> lock(mutex_results);
  435. for (const auto & task : tasks) {
  436. SRV_DBG("add task %d to waiting list. current waiting = %d (before add)\n", task.id, (int) waiting_task_ids.size());
  437. waiting_task_ids.insert(task.id);
  438. }
  439. }
  440. // when the request is finished, we can remove task associated with it
  441. void remove_waiting_task_id(int id_task) {
  442. SRV_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting_task_ids.size());
  443. std::unique_lock<std::mutex> lock(mutex_results);
  444. waiting_task_ids.erase(id_task);
  445. }
  446. void remove_waiting_task_ids(const std::unordered_set<int> & id_tasks) {
  447. std::unique_lock<std::mutex> lock(mutex_results);
  448. for (const auto & id_task : id_tasks) {
  449. SRV_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting_task_ids.size());
  450. waiting_task_ids.erase(id_task);
  451. }
  452. }
  453. // This function blocks the thread until there is a response for one of the id_tasks
  454. server_task_result recv(const std::unordered_set<int> & id_tasks) {
  455. while (true) {
  456. std::unique_lock<std::mutex> lock(mutex_results);
  457. condition_results.wait(lock, [&]{
  458. return !queue_results.empty();
  459. });
  460. for (int i = 0; i < (int) queue_results.size(); i++) {
  461. if (id_tasks.find(queue_results[i].id) != id_tasks.end()) {
  462. server_task_result res = queue_results[i];
  463. queue_results.erase(queue_results.begin() + i);
  464. return res;
  465. }
  466. }
  467. }
  468. // should never reach here
  469. }
  470. // single-task version of recv()
  471. server_task_result recv(int id_task) {
  472. std::unordered_set<int> id_tasks = {id_task};
  473. return recv(id_tasks);
  474. }
  475. // Send a new result to a waiting id_task
  476. void send(server_task_result & result) {
  477. SRV_DBG("sending result for task id = %d\n", result.id);
  478. std::unique_lock<std::mutex> lock(mutex_results);
  479. for (const auto & id_task : waiting_task_ids) {
  480. if (result.id == id_task) {
  481. SRV_DBG("task id = %d moved to result queue\n", result.id);
  482. queue_results.push_back(std::move(result));
  483. condition_results.notify_all();
  484. return;
  485. }
  486. }
  487. }
  488. };
  489. struct server_context {
  490. common_params params_base;
  491. llama_model * model = nullptr;
  492. llama_context * ctx = nullptr;
  493. std::vector<common_lora_adapter_container> loras;
  494. llama_model * model_dft = nullptr;
  495. llama_context_params cparams_dft;
  496. llama_batch batch = {};
  497. bool clean_kv_cache = true;
  498. bool add_bos_token = true;
  499. bool has_eos_token = false;
  500. int32_t n_ctx; // total context for all clients / slots
  501. // slots / clients
  502. std::vector<server_slot> slots;
  503. json default_generation_settings_for_props;
  504. server_queue queue_tasks;
  505. server_response queue_results;
  506. server_metrics metrics;
  507. // Necessary similarity of prompt for slot selection
  508. float slot_prompt_similarity = 0.0f;
  509. ~server_context() {
  510. if (ctx) {
  511. llama_free(ctx);
  512. ctx = nullptr;
  513. }
  514. if (model) {
  515. llama_free_model(model);
  516. model = nullptr;
  517. }
  518. if (model_dft) {
  519. llama_free_model(model_dft);
  520. model_dft = nullptr;
  521. }
  522. // Clear any sampling context
  523. for (server_slot & slot : slots) {
  524. common_sampler_free(slot.smpl);
  525. slot.smpl = nullptr;
  526. llama_free(slot.ctx_dft);
  527. slot.ctx_dft = nullptr;
  528. common_speculative_free(slot.spec);
  529. slot.spec = nullptr;
  530. llama_batch_free(slot.batch_spec);
  531. }
  532. llama_batch_free(batch);
  533. }
  534. bool load_model(const common_params & params) {
  535. SRV_INF("loading model '%s'\n", params.model.c_str());
  536. params_base = params;
  537. common_init_result llama_init = common_init_from_params(params_base);
  538. model = llama_init.model;
  539. ctx = llama_init.context;
  540. loras = llama_init.lora_adapters;
  541. if (model == nullptr) {
  542. SRV_ERR("failed to load model, '%s'\n", params_base.model.c_str());
  543. return false;
  544. }
  545. n_ctx = llama_n_ctx(ctx);
  546. add_bos_token = llama_add_bos_token(model);
  547. has_eos_token = !llama_add_eos_token(model);
  548. if (!params_base.speculative.model.empty()) {
  549. SRV_INF("loading draft model '%s'\n", params_base.speculative.model.c_str());
  550. auto params_dft = params_base;
  551. params_dft.devices = params_base.speculative.devices;
  552. params_dft.model = params_base.speculative.model;
  553. params_dft.n_ctx = params_base.speculative.n_ctx == 0 ? params_base.n_ctx / params_base.n_parallel : params_base.speculative.n_ctx;
  554. params_dft.n_gpu_layers = params_base.speculative.n_gpu_layers;
  555. params_dft.n_parallel = 1;
  556. common_init_result llama_init_dft = common_init_from_params(params_dft);
  557. model_dft = llama_init_dft.model;
  558. if (model_dft == nullptr) {
  559. SRV_ERR("failed to load draft model, '%s'\n", params_base.speculative.model.c_str());
  560. return false;
  561. }
  562. if (!common_speculative_are_compatible(ctx, llama_init_dft.context)) {
  563. SRV_ERR("the draft model '%s' is not compatible with the target model '%s'\n", params_base.speculative.model.c_str(), params_base.model.c_str());
  564. llama_free (llama_init_dft.context);
  565. llama_free_model(llama_init_dft.model);
  566. return false;
  567. }
  568. const int n_ctx_dft = llama_n_ctx(llama_init_dft.context);
  569. cparams_dft = common_context_params_to_llama(params_dft);
  570. cparams_dft.n_batch = n_ctx_dft;
  571. // force F16 KV cache for the draft model for extra performance
  572. cparams_dft.type_k = GGML_TYPE_F16;
  573. cparams_dft.type_v = GGML_TYPE_F16;
  574. // the context is not needed - we will create one for each slot
  575. llama_free(llama_init_dft.context);
  576. }
  577. return true;
  578. }
  579. bool validate_model_chat_template() const {
  580. std::vector<char> model_template(2048, 0); // longest known template is about 1200 bytes
  581. std::string template_key = "tokenizer.chat_template";
  582. int32_t res = llama_model_meta_val_str(model, template_key.c_str(), model_template.data(), model_template.size());
  583. if (res >= 0) {
  584. llama_chat_message chat[] = {{"user", "test"}};
  585. std::string tmpl = std::string(model_template.data(), model_template.size());
  586. int32_t chat_res = llama_chat_apply_template(model, tmpl.c_str(), chat, 1, true, nullptr, 0);
  587. return chat_res > 0;
  588. }
  589. return false;
  590. }
  591. void init() {
  592. const int32_t n_ctx_slot = n_ctx / params_base.n_parallel;
  593. SRV_INF("initializing slots, n_slots = %d\n", params_base.n_parallel);
  594. for (int i = 0; i < params_base.n_parallel; i++) {
  595. server_slot slot;
  596. slot.id = i;
  597. slot.n_ctx = n_ctx_slot;
  598. slot.n_predict = params_base.n_predict;
  599. if (model_dft) {
  600. slot.batch_spec = llama_batch_init(params_base.speculative.n_max + 1, 0, 1);
  601. slot.ctx_dft = llama_new_context_with_model(model_dft, cparams_dft);
  602. if (slot.ctx_dft == nullptr) {
  603. SRV_ERR("%s", "failed to create draft context\n");
  604. return;
  605. }
  606. slot.spec = common_speculative_init(slot.ctx_dft);
  607. if (slot.spec == nullptr) {
  608. SRV_ERR("%s", "failed to create speculator\n");
  609. return;
  610. }
  611. }
  612. SLT_INF(slot, "new slot n_ctx_slot = %d\n", slot.n_ctx);
  613. slot.params.sampling = params_base.sampling;
  614. slot.callback_on_release = [this](int) {
  615. queue_tasks.pop_deferred_task();
  616. };
  617. slot.reset();
  618. slots.push_back(slot);
  619. }
  620. default_generation_settings_for_props = get_formated_generation(slots.front());
  621. default_generation_settings_for_props["seed"] = -1;
  622. // the update_slots() logic will always submit a maximum of n_batch or n_parallel tokens
  623. // note that n_batch can be > n_ctx (e.g. for non-causal attention models such as BERT where the KV cache is not used)
  624. {
  625. const int32_t n_batch = llama_n_batch(ctx);
  626. // only a single seq_id per token is needed
  627. batch = llama_batch_init(std::max(n_batch, params_base.n_parallel), 0, 1);
  628. }
  629. metrics.init();
  630. }
  631. server_slot * get_slot_by_id(int id) {
  632. for (server_slot & slot : slots) {
  633. if (slot.id == id) {
  634. return &slot;
  635. }
  636. }
  637. return nullptr;
  638. }
  639. server_slot * get_available_slot(const server_task & task) {
  640. server_slot * ret = nullptr;
  641. // find the slot that has at least n% prompt similarity
  642. if (ret == nullptr && slot_prompt_similarity != 0.0f) {
  643. int lcs_len = 0;
  644. float similarity = 0;
  645. for (server_slot & slot : slots) {
  646. // skip the slot if it is not available
  647. if (slot.is_processing()) {
  648. continue;
  649. }
  650. // skip the slot if it does not contains cached tokens
  651. if (slot.cache_tokens.empty()) {
  652. continue;
  653. }
  654. // length of the Longest Common Subsequence between the current slot's prompt and the input prompt
  655. int cur_lcs_len = common_lcs(slot.cache_tokens, task.prompt_tokens);
  656. // fraction of the common subsequence length compared to the current slot's prompt length
  657. float cur_similarity = static_cast<float>(cur_lcs_len) / static_cast<int>(slot.cache_tokens.size());
  658. // select the current slot if the criteria match
  659. if (cur_lcs_len > lcs_len && cur_similarity > slot_prompt_similarity) {
  660. lcs_len = cur_lcs_len;
  661. similarity = cur_similarity;
  662. ret = &slot;
  663. }
  664. }
  665. if (ret != nullptr) {
  666. SLT_DBG(*ret, "selected slot by lcs similarity, lcs_len = %d, similarity = %f\n", lcs_len, similarity);
  667. }
  668. }
  669. // find the slot that has been least recently used
  670. if (ret == nullptr) {
  671. int64_t t_last = ggml_time_us();
  672. for (server_slot & slot : slots) {
  673. // skip the slot if it is not available
  674. if (slot.is_processing()) {
  675. continue;
  676. }
  677. // select the current slot if the criteria match
  678. if (slot.t_last_used < t_last) {
  679. t_last = slot.t_last_used;
  680. ret = &slot;
  681. }
  682. }
  683. if (ret != nullptr) {
  684. SLT_DBG(*ret, "selected slot by lru, t_last = %" PRId64 "\n", t_last);
  685. }
  686. }
  687. return ret;
  688. }
  689. bool launch_slot_with_task(server_slot & slot, const server_task & task) {
  690. // Sampling parameter defaults are loaded from the global server context (but individual requests can still override them)
  691. slot_params defaults;
  692. defaults.sampling = params_base.sampling;
  693. defaults.speculative = params_base.speculative;
  694. const auto & data = task.data;
  695. if (data.count("__oaicompat") != 0) {
  696. slot.oaicompat = true;
  697. slot.oaicompat_model = json_value(data, "model", std::string(DEFAULT_OAICOMPAT_MODEL));
  698. } else {
  699. slot.oaicompat = false;
  700. slot.oaicompat_model = "";
  701. }
  702. slot.timings_per_token = json_value(data, "timings_per_token", false);
  703. slot.params.stream = json_value(data, "stream", false);
  704. slot.params.cache_prompt = json_value(data, "cache_prompt", true);
  705. slot.params.n_predict = json_value(data, "n_predict", json_value(data, "max_tokens", defaults.n_predict));
  706. slot.params.n_indent = json_value(data, "n_indent", defaults.n_indent);
  707. slot.params.n_keep = json_value(data, "n_keep", defaults.n_keep);
  708. slot.params.n_discard = json_value(data, "n_discard", defaults.n_discard);
  709. //slot.params.t_max_prompt_ms = json_value(data, "t_max_prompt_ms", defaults.t_max_prompt_ms); // TODO: implement
  710. slot.params.t_max_predict_ms = json_value(data, "t_max_predict_ms", defaults.t_max_predict_ms);
  711. slot.params.sampling.top_k = json_value(data, "top_k", defaults.sampling.top_k);
  712. slot.params.sampling.top_p = json_value(data, "top_p", defaults.sampling.top_p);
  713. slot.params.sampling.min_p = json_value(data, "min_p", defaults.sampling.min_p);
  714. slot.params.sampling.xtc_probability = json_value(data, "xtc_probability", defaults.sampling.xtc_probability);
  715. slot.params.sampling.xtc_threshold = json_value(data, "xtc_threshold", defaults.sampling.xtc_threshold);
  716. slot.params.sampling.typ_p = json_value(data, "typical_p", defaults.sampling.typ_p);
  717. slot.params.sampling.temp = json_value(data, "temperature", defaults.sampling.temp);
  718. slot.params.sampling.dynatemp_range = json_value(data, "dynatemp_range", defaults.sampling.dynatemp_range);
  719. slot.params.sampling.dynatemp_exponent = json_value(data, "dynatemp_exponent", defaults.sampling.dynatemp_exponent);
  720. slot.params.sampling.penalty_last_n = json_value(data, "repeat_last_n", defaults.sampling.penalty_last_n);
  721. slot.params.sampling.penalty_repeat = json_value(data, "repeat_penalty", defaults.sampling.penalty_repeat);
  722. slot.params.sampling.penalty_freq = json_value(data, "frequency_penalty", defaults.sampling.penalty_freq);
  723. slot.params.sampling.penalty_present = json_value(data, "presence_penalty", defaults.sampling.penalty_present);
  724. slot.params.sampling.dry_multiplier = json_value(data, "dry_multiplier", defaults.sampling.dry_multiplier);
  725. slot.params.sampling.dry_base = json_value(data, "dry_base", defaults.sampling.dry_base);
  726. slot.params.sampling.dry_allowed_length = json_value(data, "dry_allowed_length", defaults.sampling.dry_allowed_length);
  727. slot.params.sampling.dry_penalty_last_n = json_value(data, "dry_penalty_last_n", defaults.sampling.dry_penalty_last_n);
  728. slot.params.sampling.mirostat = json_value(data, "mirostat", defaults.sampling.mirostat);
  729. slot.params.sampling.mirostat_tau = json_value(data, "mirostat_tau", defaults.sampling.mirostat_tau);
  730. slot.params.sampling.mirostat_eta = json_value(data, "mirostat_eta", defaults.sampling.mirostat_eta);
  731. slot.params.sampling.penalize_nl = json_value(data, "penalize_nl", defaults.sampling.penalize_nl);
  732. slot.params.sampling.seed = json_value(data, "seed", defaults.sampling.seed);
  733. slot.params.sampling.n_probs = json_value(data, "n_probs", defaults.sampling.n_probs);
  734. slot.params.sampling.min_keep = json_value(data, "min_keep", defaults.sampling.min_keep);
  735. slot.params.speculative.n_min = json_value(data, "speculative.n_min", defaults.speculative.n_min);
  736. slot.params.speculative.n_max = json_value(data, "speculative.n_max", defaults.speculative.n_max);
  737. slot.params.speculative.p_min = json_value(data, "speculative.p_min", defaults.speculative.p_min);
  738. slot.params.speculative.n_min = std::min(slot.params.speculative.n_max, slot.params.speculative.n_min);
  739. if (slot.params.sampling.dry_base < 1.0f) {
  740. slot.params.sampling.dry_base = defaults.sampling.dry_base;
  741. }
  742. // sequence breakers for DRY
  743. {
  744. // Currently, this is not compatible with TextGen WebUI, Koboldcpp and SillyTavern format
  745. // Ref: https://github.com/oobabooga/text-generation-webui/blob/d1af7a41ade7bd3c3a463bfa640725edb818ebaf/extensions/openai/typing.py#L39
  746. if (data.contains("dry_sequence_breakers")) {
  747. slot.params.sampling.dry_sequence_breakers = json_value(data, "dry_sequence_breakers", std::vector<std::string>());
  748. if (slot.params.sampling.dry_sequence_breakers.empty()) {
  749. send_error(task, "Error: dry_sequence_breakers must be a non-empty array of strings", ERROR_TYPE_INVALID_REQUEST);
  750. return false;
  751. }
  752. }
  753. }
  754. // process "json_schema" and "grammar"
  755. if (data.contains("json_schema") && !data.at("json_schema").is_null() && data.contains("grammar") && !data.at("grammar").is_null()) {
  756. send_error(task, "Either \"json_schema\" or \"grammar\" can be specified, but not both", ERROR_TYPE_INVALID_REQUEST);
  757. return false;
  758. }
  759. if (data.contains("json_schema") && !data.contains("grammar")) {
  760. try {
  761. auto schema = json_value(data, "json_schema", json::object());
  762. slot.params.sampling.grammar = json_schema_to_grammar(schema);
  763. } catch (const std::exception & e) {
  764. send_error(task, std::string("\"json_schema\": ") + e.what(), ERROR_TYPE_INVALID_REQUEST);
  765. return false;
  766. }
  767. } else {
  768. slot.params.sampling.grammar = json_value(data, "grammar", defaults.sampling.grammar);
  769. }
  770. if (slot.n_predict > 0 && slot.params.n_predict > slot.n_predict) {
  771. // Might be better to reject the request with a 400 ?
  772. slot.params.n_predict = slot.n_predict;
  773. SLT_WRN(slot, "n_predict = %d exceeds server configuration, setting to %d", slot.n_predict, slot.n_predict);
  774. }
  775. {
  776. slot.params.sampling.logit_bias.clear();
  777. if (json_value(data, "ignore_eos", false) && has_eos_token) {
  778. slot.params.sampling.logit_bias.push_back({llama_token_eos(model), -INFINITY});
  779. }
  780. const auto & logit_bias = data.find("logit_bias");
  781. if (logit_bias != data.end() && logit_bias->is_array()) {
  782. const int n_vocab = llama_n_vocab(model);
  783. for (const auto & el : *logit_bias) {
  784. // TODO: we may want to throw errors here, in case "el" is incorrect
  785. if (el.is_array() && el.size() == 2) {
  786. float bias;
  787. if (el[1].is_number()) {
  788. bias = el[1].get<float>();
  789. } else if (el[1].is_boolean() && !el[1].get<bool>()) {
  790. bias = -INFINITY;
  791. } else {
  792. continue;
  793. }
  794. if (el[0].is_number_integer()) {
  795. llama_token tok = el[0].get<llama_token>();
  796. if (tok >= 0 && tok < n_vocab) {
  797. slot.params.sampling.logit_bias.push_back({tok, bias});
  798. }
  799. } else if (el[0].is_string()) {
  800. auto toks = common_tokenize(model, el[0].get<std::string>(), false);
  801. for (auto tok : toks) {
  802. slot.params.sampling.logit_bias.push_back({tok, bias});
  803. }
  804. }
  805. }
  806. }
  807. }
  808. }
  809. {
  810. slot.params.antiprompt.clear();
  811. const auto & stop = data.find("stop");
  812. if (stop != data.end() && stop->is_array()) {
  813. for (const auto & word : *stop) {
  814. if (!word.empty()) {
  815. slot.params.antiprompt.push_back(word);
  816. }
  817. }
  818. }
  819. }
  820. {
  821. const auto & samplers = data.find("samplers");
  822. if (samplers != data.end()) {
  823. if (samplers->is_array()) {
  824. std::vector<std::string> sampler_names;
  825. for (const auto & name : *samplers) {
  826. if (name.is_string()) {
  827. sampler_names.emplace_back(name);
  828. }
  829. }
  830. slot.params.sampling.samplers = common_sampler_types_from_names(sampler_names, false);
  831. } else if (samplers->is_string()){
  832. std::string sampler_string;
  833. for (const auto & name : *samplers) {
  834. sampler_string += name;
  835. }
  836. slot.params.sampling.samplers = common_sampler_types_from_chars(sampler_string);
  837. }
  838. } else {
  839. slot.params.sampling.samplers = defaults.sampling.samplers;
  840. }
  841. }
  842. {
  843. if (slot.smpl != nullptr) {
  844. common_sampler_free(slot.smpl);
  845. }
  846. slot.smpl = common_sampler_init(model, slot.params.sampling);
  847. if (slot.smpl == nullptr) {
  848. // for now, the only error that may happen here is invalid grammar
  849. send_error(task, "Failed to parse grammar", ERROR_TYPE_INVALID_REQUEST);
  850. return false;
  851. }
  852. }
  853. if (slot.ctx_dft) {
  854. llama_batch_free(slot.batch_spec);
  855. slot.batch_spec = llama_batch_init(slot.params.speculative.n_max + 1, 0, 1);
  856. }
  857. slot.state = SLOT_STATE_STARTED;
  858. SLT_INF(slot, "%s", "processing task\n");
  859. return true;
  860. }
  861. void kv_cache_clear() {
  862. SRV_DBG("%s", "clearing KV cache\n");
  863. // clear the entire KV cache
  864. llama_kv_cache_clear(ctx);
  865. clean_kv_cache = false;
  866. }
  867. bool process_token(completion_token_output & result, server_slot & slot) {
  868. // remember which tokens were sampled - used for repetition penalties during sampling
  869. const std::string token_str = common_token_to_piece(ctx, result.tok, params_base.special);
  870. slot.sampled = result.tok;
  871. // search stop word and delete it
  872. slot.generated_text += token_str;
  873. slot.has_next_token = true;
  874. // check if there is incomplete UTF-8 character at the end
  875. bool incomplete = false;
  876. for (unsigned i = 1; i < 5 && i <= slot.generated_text.size(); ++i) {
  877. unsigned char c = slot.generated_text[slot.generated_text.size() - i];
  878. if ((c & 0xC0) == 0x80) {
  879. // continuation byte: 10xxxxxx
  880. continue;
  881. }
  882. if ((c & 0xE0) == 0xC0) {
  883. // 2-byte character: 110xxxxx ...
  884. incomplete = i < 2;
  885. } else if ((c & 0xF0) == 0xE0) {
  886. // 3-byte character: 1110xxxx ...
  887. incomplete = i < 3;
  888. } else if ((c & 0xF8) == 0xF0) {
  889. // 4-byte character: 11110xxx ...
  890. incomplete = i < 4;
  891. }
  892. // else 1-byte character or invalid byte
  893. break;
  894. }
  895. if (!incomplete) {
  896. size_t pos = std::min(slot.n_sent_text, slot.generated_text.size());
  897. const std::string str_test = slot.generated_text.substr(pos);
  898. bool send_text = true;
  899. size_t stop_pos = slot.find_stopping_strings(str_test, token_str.size(), STOP_TYPE_FULL);
  900. if (stop_pos != std::string::npos) {
  901. slot.generated_text.erase(
  902. slot.generated_text.begin() + pos + stop_pos,
  903. slot.generated_text.end());
  904. pos = std::min(slot.n_sent_text, slot.generated_text.size());
  905. } else if (slot.has_next_token) {
  906. stop_pos = slot.find_stopping_strings(str_test, token_str.size(), STOP_TYPE_PARTIAL);
  907. send_text = stop_pos == std::string::npos;
  908. }
  909. // check if there is any token to predict
  910. if (send_text) {
  911. // no send the stop word in the response
  912. result.text_to_send = slot.generated_text.substr(pos, std::string::npos);
  913. slot.n_sent_text += result.text_to_send.size();
  914. // add the token to slot queue and cache
  915. }
  916. slot.add_token(result);
  917. if (slot.params.stream) {
  918. send_partial_response(slot, result);
  919. }
  920. }
  921. if (incomplete) {
  922. slot.has_next_token = true;
  923. }
  924. // check the limits
  925. if (slot.n_decoded > 0 && slot.has_next_token && !slot.has_budget(params_base)) {
  926. slot.stopped_limit = true;
  927. slot.has_next_token = false;
  928. SLT_DBG(slot, "stopped by limit, n_decoded = %d, n_predict = %d\n", slot.n_decoded, slot.params.n_predict);
  929. }
  930. if (slot.has_new_line) {
  931. // if we have already seen a new line, we stop after a certain time limit
  932. if (slot.params.t_max_predict_ms > 0 && (ggml_time_us() - slot.t_start_generation > 1000.0f*slot.params.t_max_predict_ms)) {
  933. slot.stopped_limit = true;
  934. slot.has_next_token = false;
  935. SLT_DBG(slot, "stopped by time limit, n_decoded = %d, t_max_predict_ms = %d ms\n", slot.n_decoded, (int) slot.params.t_max_predict_ms);
  936. }
  937. // require that each new line has a whitespace prefix (i.e. indentation) of at least slot.params.n_indent
  938. if (slot.params.n_indent > 0) {
  939. // check the current indentation
  940. // TODO: improve by not doing it more than once for each new line
  941. if (slot.last_nl_pos > 0) {
  942. size_t pos = slot.last_nl_pos;
  943. int n_indent = 0;
  944. while (pos < slot.generated_text.size() && (slot.generated_text[pos] == ' ' || slot.generated_text[pos] == '\t')) {
  945. n_indent++;
  946. pos++;
  947. }
  948. if (pos < slot.generated_text.size() && n_indent < slot.params.n_indent) {
  949. slot.stopped_limit = true;
  950. slot.has_next_token = false;
  951. // cut the last line
  952. slot.generated_text.erase(pos, std::string::npos);
  953. SLT_DBG(slot, "stopped by indentation limit, n_decoded = %d, n_indent = %d\n", slot.n_decoded, n_indent);
  954. }
  955. }
  956. // find the next new line
  957. {
  958. const size_t pos = slot.generated_text.find('\n', slot.last_nl_pos);
  959. if (pos != std::string::npos) {
  960. slot.last_nl_pos = pos + 1;
  961. }
  962. }
  963. }
  964. }
  965. // check if there is a new line in the generated text
  966. if (result.text_to_send.find('\n') != std::string::npos) {
  967. slot.has_new_line = true;
  968. }
  969. // if context shift is disabled, we stop when it reaches the context limit
  970. if (slot.n_past >= slot.n_ctx) {
  971. slot.truncated = true;
  972. slot.stopped_limit = true;
  973. slot.has_next_token = false;
  974. SLT_DBG(slot, "stopped due to running out of context capacity, n_past = %d, n_prompt_tokens = %d, n_decoded = %d, n_ctx = %d\n",
  975. slot.n_decoded, slot.n_prompt_tokens, slot.n_past, slot.n_ctx);
  976. }
  977. if (llama_token_is_eog(model, result.tok)) {
  978. slot.stopped_eos = true;
  979. slot.has_next_token = false;
  980. SLT_DBG(slot, "%s", "stopped by EOS\n");
  981. }
  982. const auto n_ctx_train = llama_n_ctx_train(model);
  983. if (slot.params.n_predict < 1 && slot.n_predict < 1 && slot.n_prompt_tokens + slot.n_decoded >= n_ctx_train) {
  984. slot.truncated = true;
  985. slot.stopped_limit = true;
  986. slot.has_next_token = false; // stop prediction
  987. SLT_WRN(slot,
  988. "n_predict (%d) is set for infinite generation. "
  989. "Limiting generated tokens to n_ctx_train (%d) to avoid EOS-less generation infinite loop\n",
  990. slot.params.n_predict, n_ctx_train);
  991. }
  992. SLT_DBG(slot, "n_decoded = %d, n_remaining = %d, next token: %5d '%s'\n", slot.n_decoded, slot.n_remaining, result.tok, token_str.c_str());
  993. return slot.has_next_token; // continue
  994. }
  995. json get_formated_generation(const server_slot & slot) const {
  996. std::vector<std::string> samplers;
  997. samplers.reserve(slot.params.sampling.samplers.size());
  998. for (const auto & sampler : slot.params.sampling.samplers) {
  999. samplers.emplace_back(common_sampler_type_to_str(sampler));
  1000. }
  1001. return json {
  1002. {"n_ctx", slot.n_ctx},
  1003. {"n_predict", slot.n_predict}, // Server configured n_predict
  1004. {"model", params_base.model_alias},
  1005. {"seed", slot.params.sampling.seed},
  1006. {"seed_cur", slot.smpl ? common_sampler_get_seed(slot.smpl) : 0},
  1007. {"temperature", slot.params.sampling.temp},
  1008. {"dynatemp_range", slot.params.sampling.dynatemp_range},
  1009. {"dynatemp_exponent", slot.params.sampling.dynatemp_exponent},
  1010. {"top_k", slot.params.sampling.top_k},
  1011. {"top_p", slot.params.sampling.top_p},
  1012. {"min_p", slot.params.sampling.min_p},
  1013. {"xtc_probability", slot.params.sampling.xtc_probability},
  1014. {"xtc_threshold", slot.params.sampling.xtc_threshold},
  1015. {"typical_p", slot.params.sampling.typ_p},
  1016. {"repeat_last_n", slot.params.sampling.penalty_last_n},
  1017. {"repeat_penalty", slot.params.sampling.penalty_repeat},
  1018. {"presence_penalty", slot.params.sampling.penalty_present},
  1019. {"frequency_penalty", slot.params.sampling.penalty_freq},
  1020. {"dry_multiplier", slot.params.sampling.dry_multiplier},
  1021. {"dry_base", slot.params.sampling.dry_base},
  1022. {"dry_allowed_length", slot.params.sampling.dry_allowed_length},
  1023. {"dry_penalty_last_n", slot.params.sampling.dry_penalty_last_n},
  1024. {"dry_sequence_breakers", slot.params.sampling.dry_sequence_breakers},
  1025. {"mirostat", slot.params.sampling.mirostat},
  1026. {"mirostat_tau", slot.params.sampling.mirostat_tau},
  1027. {"mirostat_eta", slot.params.sampling.mirostat_eta},
  1028. {"penalize_nl", slot.params.sampling.penalize_nl},
  1029. {"stop", slot.params.antiprompt},
  1030. {"max_tokens", slot.params.n_predict}, // User configured n_predict
  1031. {"n_keep", slot.params.n_keep},
  1032. {"n_discard", slot.params.n_discard},
  1033. {"ignore_eos", slot.params.sampling.ignore_eos},
  1034. {"stream", slot.params.stream},
  1035. //{"logit_bias", slot.params.sampling.logit_bias},
  1036. {"n_probs", slot.params.sampling.n_probs},
  1037. {"min_keep", slot.params.sampling.min_keep},
  1038. {"grammar", slot.params.sampling.grammar},
  1039. {"samplers", samplers},
  1040. {"speculative", slot.can_speculate()},
  1041. {"speculative.n_max", slot.params.speculative.n_max},
  1042. {"speculative.n_min", slot.params.speculative.n_min},
  1043. {"speculative.p_min", slot.params.speculative.p_min},
  1044. {"timings_per_token", slot.timings_per_token},
  1045. };
  1046. }
  1047. void send_error(const server_task & task, const std::string & error, const enum error_type type = ERROR_TYPE_SERVER) {
  1048. send_error(task.id, error, type);
  1049. }
  1050. void send_error(const server_slot & slot, const std::string & error, const enum error_type type = ERROR_TYPE_SERVER) {
  1051. send_error(slot.id_task, error, type);
  1052. }
  1053. void send_error(const int id_task, const std::string & error, const enum error_type type = ERROR_TYPE_SERVER) {
  1054. SRV_ERR("task id = %d, error: %s\n", id_task, error.c_str());
  1055. server_task_result res;
  1056. res.id = id_task;
  1057. res.stop = false;
  1058. res.error = true;
  1059. res.data = format_error_response(error, type);
  1060. queue_results.send(res);
  1061. }
  1062. void send_partial_response(server_slot & slot, completion_token_output tkn) {
  1063. server_task_result res;
  1064. res.id = slot.id_task;
  1065. res.error = false;
  1066. res.stop = false;
  1067. res.data = json {
  1068. {"content", tkn.text_to_send},
  1069. {"stop", false},
  1070. {"id_slot", slot.id},
  1071. {"multimodal", false},
  1072. {"index", slot.index},
  1073. };
  1074. if (slot.params.sampling.n_probs > 0) {
  1075. const llama_tokens to_send_toks = common_tokenize(ctx, tkn.text_to_send, false);
  1076. const size_t probs_pos = std::min(slot.n_sent_token_probs, slot.generated_token_probs.size());
  1077. const size_t probs_stop_pos = std::min(slot.n_sent_token_probs + to_send_toks.size(), slot.generated_token_probs.size());
  1078. std::vector<completion_token_output> probs_output;
  1079. if (probs_pos < probs_stop_pos) {
  1080. probs_output = std::vector<completion_token_output>(
  1081. slot.generated_token_probs.begin() + probs_pos,
  1082. slot.generated_token_probs.begin() + probs_stop_pos);
  1083. }
  1084. slot.n_sent_token_probs = probs_stop_pos;
  1085. res.data["completion_probabilities"] = probs_vector_to_json(ctx, probs_output);
  1086. }
  1087. if (slot.oaicompat) {
  1088. res.data["oaicompat_token_ctr"] = slot.n_decoded;
  1089. res.data["model"] = slot.oaicompat_model;
  1090. }
  1091. if (slot.timings_per_token) {
  1092. res.data["timings"] = slot.get_formated_timings();
  1093. }
  1094. queue_results.send(res);
  1095. }
  1096. void send_final_response(const server_slot & slot) {
  1097. server_task_result res;
  1098. res.id = slot.id_task;
  1099. res.error = false;
  1100. res.stop = true;
  1101. res.data = json {
  1102. {"content", !slot.params.stream ? slot.generated_text : ""},
  1103. {"id_slot", slot.id},
  1104. {"stop", true},
  1105. {"model", params_base.model_alias},
  1106. {"tokens_predicted", slot.n_decoded},
  1107. {"tokens_evaluated", slot.n_prompt_tokens},
  1108. {"generation_settings", get_formated_generation(slot)},
  1109. {"prompt", common_detokenize(ctx, slot.prompt_tokens)},
  1110. {"has_new_line", slot.has_new_line},
  1111. {"truncated", slot.truncated},
  1112. {"stopped_eos", slot.stopped_eos},
  1113. {"stopped_word", slot.stopped_word},
  1114. {"stopped_limit", slot.stopped_limit},
  1115. {"stopping_word", slot.stopping_word},
  1116. {"tokens_cached", slot.n_past},
  1117. {"timings", slot.get_formated_timings()},
  1118. {"index", slot.index},
  1119. };
  1120. if (slot.params.sampling.n_probs > 0) {
  1121. std::vector<completion_token_output> probs;
  1122. if (!slot.params.stream && slot.stopped_word) {
  1123. const llama_tokens stop_word_toks = common_tokenize(ctx, slot.stopping_word, false);
  1124. size_t safe_offset = std::min(slot.generated_token_probs.size(), stop_word_toks.size());
  1125. probs = std::vector<completion_token_output>(
  1126. slot.generated_token_probs.begin(),
  1127. slot.generated_token_probs.end() - safe_offset);
  1128. } else {
  1129. probs = std::vector<completion_token_output>(
  1130. slot.generated_token_probs.begin(),
  1131. slot.generated_token_probs.end());
  1132. }
  1133. res.data["completion_probabilities"] = probs_vector_to_json(ctx, probs);
  1134. }
  1135. if (slot.oaicompat) {
  1136. res.data["oaicompat_token_ctr"] = slot.n_decoded;
  1137. res.data["model"] = slot.oaicompat_model;
  1138. }
  1139. queue_results.send(res);
  1140. }
  1141. void send_embedding(const server_slot & slot, const llama_batch & batch) {
  1142. server_task_result res;
  1143. res.id = slot.id_task;
  1144. res.error = false;
  1145. res.stop = true;
  1146. const int n_embd = llama_n_embd(model);
  1147. std::vector<float> embd_res(n_embd, 0.0f);
  1148. for (int i = 0; i < batch.n_tokens; ++i) {
  1149. if (!batch.logits[i] || batch.seq_id[i][0] != slot.id) {
  1150. continue;
  1151. }
  1152. const float * embd = llama_get_embeddings_seq(ctx, batch.seq_id[i][0]);
  1153. if (embd == NULL) {
  1154. embd = llama_get_embeddings_ith(ctx, i);
  1155. }
  1156. if (embd == NULL) {
  1157. SLT_ERR(slot, "failed to get embeddings, token = %d, seq_id = %d\n", batch.token[i], batch.seq_id[i][0]);
  1158. res.data = json {
  1159. {"embedding", std::vector<float>(n_embd, 0.0f)},
  1160. {"index", slot.index},
  1161. };
  1162. continue;
  1163. }
  1164. common_embd_normalize(embd, embd_res.data(), n_embd);
  1165. res.data = json {
  1166. {"embedding", embd_res},
  1167. {"index", slot.index},
  1168. };
  1169. }
  1170. SLT_DBG(slot, "%s", "sending embeddings\n");
  1171. queue_results.send(res);
  1172. }
  1173. void send_rerank(const server_slot & slot, const llama_batch & batch) {
  1174. server_task_result res;
  1175. res.id = slot.id_task;
  1176. res.error = false;
  1177. res.stop = true;
  1178. for (int i = 0; i < batch.n_tokens; ++i) {
  1179. if (!batch.logits[i] || batch.seq_id[i][0] != slot.id) {
  1180. continue;
  1181. }
  1182. const float * embd = llama_get_embeddings_seq(ctx, batch.seq_id[i][0]);
  1183. if (embd == NULL) {
  1184. embd = llama_get_embeddings_ith(ctx, i);
  1185. }
  1186. if (embd == NULL) {
  1187. SLT_ERR(slot, "failed to get embeddings, token = %d, seq_id = %d\n", batch.token[i], batch.seq_id[i][0]);
  1188. res.data = json {
  1189. {"index", slot.index},
  1190. {"score", -1e6},
  1191. };
  1192. continue;
  1193. }
  1194. res.data = json {
  1195. {"index", slot.index},
  1196. {"score", embd[0]},
  1197. };
  1198. }
  1199. SLT_DBG(slot, "sending rerank result, res = '%s'\n", res.data.dump().c_str());
  1200. queue_results.send(res);
  1201. }
  1202. //
  1203. // Functions to create new task(s) and receive result(s)
  1204. //
  1205. // break the input "prompt" into multiple tasks if needed, then format and tokenize the input prompt(s)
  1206. std::vector<server_task> create_tasks_inference(json data, server_task_inf_type inf_type) {
  1207. std::vector<server_task> tasks;
  1208. auto create_task = [&](json & task_data, llama_tokens & prompt_tokens) {
  1209. SRV_DBG("create task, n_tokens = %d\n", (int) prompt_tokens.size());
  1210. server_task task;
  1211. task.id = queue_tasks.get_new_id();
  1212. task.inf_type = inf_type;
  1213. task.type = SERVER_TASK_TYPE_INFERENCE;
  1214. task.data = task_data;
  1215. task.prompt_tokens = std::move(prompt_tokens);
  1216. tasks.push_back(std::move(task));
  1217. };
  1218. static constexpr const char * error_msg = "\"prompt\" must be a string, an array of token ids or an array of prompts";
  1219. if (!data.contains("prompt")) {
  1220. throw std::runtime_error(error_msg);
  1221. }
  1222. // because llama_tokenize api is thread-safe, we can tokenize the prompt from HTTP thread
  1223. bool add_special = inf_type != SERVER_TASK_INF_TYPE_RERANK && inf_type != SERVER_TASK_INF_TYPE_INFILL;
  1224. std::vector<llama_tokens> tokenized_prompts = tokenize_input_prompts(ctx, data.at("prompt"), add_special, true);
  1225. switch (inf_type) {
  1226. case SERVER_TASK_INF_TYPE_RERANK:
  1227. {
  1228. // prompts[0] is the question
  1229. // the rest are the answers/documents
  1230. GGML_ASSERT(tokenized_prompts.size() > 1);
  1231. SRV_DBG("creating rerank tasks, n_prompts = %d\n", (int) tokenized_prompts.size() - 1);
  1232. for (size_t i = 1; i < tokenized_prompts.size(); i++) {
  1233. data["index"] = i - 1;
  1234. auto tokens = format_rerank(model, tokenized_prompts[0], tokenized_prompts[i]);
  1235. create_task(data, tokens);
  1236. }
  1237. } break;
  1238. case SERVER_TASK_INF_TYPE_INFILL:
  1239. {
  1240. SRV_DBG("creating infill tasks, n_prompts = %d\n", (int) tokenized_prompts.size());
  1241. for (size_t i = 0; i < tokenized_prompts.size(); i++) {
  1242. data["index"] = i;
  1243. auto tokens = format_infill(
  1244. ctx,
  1245. data.at("input_prefix"),
  1246. data.at("input_suffix"),
  1247. data.at("input_extra"),
  1248. params_base.n_batch,
  1249. params_base.n_predict,
  1250. slots[0].n_ctx, // TODO: there should be a better way
  1251. params_base.spm_infill,
  1252. tokenized_prompts[i]
  1253. );
  1254. create_task(data, tokens);
  1255. }
  1256. } break;
  1257. default:
  1258. {
  1259. SRV_DBG("creating multi-prompt tasks, n_prompts = %d\n", (int) tokenized_prompts.size());
  1260. for (size_t i = 0; i < tokenized_prompts.size(); i++) {
  1261. data["index"] = i;
  1262. create_task(data, tokenized_prompts[i]);
  1263. }
  1264. }
  1265. }
  1266. return tasks;
  1267. }
  1268. void cancel_tasks(const std::unordered_set<int> & id_tasks) {
  1269. std::vector<server_task> cancel_tasks;
  1270. cancel_tasks.reserve(id_tasks.size());
  1271. for (const auto & id_task : id_tasks) {
  1272. SRV_WRN("cancel task, id_task = %d\n", id_task);
  1273. server_task task;
  1274. task.type = SERVER_TASK_TYPE_CANCEL;
  1275. task.id_target = id_task;
  1276. cancel_tasks.push_back(task);
  1277. queue_results.remove_waiting_task_id(id_task);
  1278. }
  1279. // push to beginning of the queue, so it has highest priority
  1280. queue_tasks.post(cancel_tasks, true);
  1281. }
  1282. // receive the results from task(s) created by create_tasks_inference
  1283. void receive_cmpl_results(
  1284. const std::unordered_set<int> & id_tasks,
  1285. const std::function<void(std::vector<server_task_result>&)> & result_handler,
  1286. const std::function<void(json)> & error_handler) {
  1287. // TODO: currently, there is no way to detect the client has cancelled the request
  1288. std::vector<server_task_result> results(id_tasks.size());
  1289. for (size_t i = 0; i < id_tasks.size(); i++) {
  1290. server_task_result result = queue_results.recv(id_tasks);
  1291. if (result.error) {
  1292. error_handler(result.data);
  1293. cancel_tasks(id_tasks);
  1294. return;
  1295. }
  1296. const size_t idx = result.data["index"];
  1297. GGML_ASSERT(idx < results.size() && "index out of range");
  1298. results[idx] = result;
  1299. }
  1300. result_handler(results);
  1301. }
  1302. // receive the results from task(s) created by create_tasks_inference, in stream mode
  1303. void receive_cmpl_results_stream(
  1304. const std::unordered_set<int> & id_tasks, const
  1305. std::function<bool(server_task_result&)> & result_handler, const
  1306. std::function<void(json)> & error_handler) {
  1307. size_t n_finished = 0;
  1308. while (true) {
  1309. server_task_result result = queue_results.recv(id_tasks);
  1310. if (!result_handler(result)) {
  1311. cancel_tasks(id_tasks);
  1312. break;
  1313. }
  1314. if (result.error) {
  1315. error_handler(result.data);
  1316. cancel_tasks(id_tasks);
  1317. break;
  1318. }
  1319. if (result.stop) {
  1320. if (++n_finished == id_tasks.size()) {
  1321. break;
  1322. }
  1323. }
  1324. }
  1325. }
  1326. //
  1327. // Functions to process the task
  1328. //
  1329. void process_single_task(server_task task) {
  1330. switch (task.type) {
  1331. case SERVER_TASK_TYPE_INFERENCE:
  1332. {
  1333. const int id_slot = json_value(task.data, "id_slot", -1);
  1334. server_slot * slot = id_slot != -1 ? get_slot_by_id(id_slot) : get_available_slot(task);
  1335. if (slot == nullptr) {
  1336. // if no slot is available, we defer this task for processing later
  1337. SRV_DBG("no slot is available, defer task, id_task = %d\n", task.id);
  1338. queue_tasks.defer(task);
  1339. break;
  1340. }
  1341. if (slot->is_processing()) {
  1342. // if requested slot is unavailable, we defer this task for processing later
  1343. SRV_DBG("requested slot is unavailable, defer task, id_task = %d\n", task.id);
  1344. queue_tasks.defer(task);
  1345. break;
  1346. }
  1347. slot->reset();
  1348. slot->id_task = task.id;
  1349. slot->inf_type = task.inf_type;
  1350. slot->index = json_value(task.data, "index", 0);
  1351. slot->prompt_tokens = std::move(task.prompt_tokens);
  1352. if (!launch_slot_with_task(*slot, task)) {
  1353. SRV_ERR("failed to launch slot with task, id_task = %d\n", task.id);
  1354. break;
  1355. }
  1356. } break;
  1357. case SERVER_TASK_TYPE_CANCEL:
  1358. {
  1359. // release slot linked with the task id
  1360. for (auto & slot : slots) {
  1361. if (slot.id_task == task.id_target) {
  1362. slot.release();
  1363. break;
  1364. }
  1365. }
  1366. } break;
  1367. case SERVER_TASK_TYPE_NEXT_RESPONSE:
  1368. {
  1369. // do nothing
  1370. } break;
  1371. case SERVER_TASK_TYPE_METRICS:
  1372. {
  1373. json slots_data = json::array();
  1374. int n_idle_slots = 0;
  1375. int n_processing_slots = 0;
  1376. for (server_slot & slot : slots) {
  1377. json slot_data = get_formated_generation(slot);
  1378. slot_data["id"] = slot.id;
  1379. slot_data["id_task"] = slot.id_task;
  1380. slot_data["is_processing"] = slot.is_processing();
  1381. slot_data["prompt"] = common_detokenize(ctx, slot.prompt_tokens);
  1382. slot_data["next_token"] = {
  1383. {"has_next_token", slot.has_next_token},
  1384. {"has_new_line", slot.has_new_line},
  1385. {"n_remain", slot.n_remaining},
  1386. {"n_decoded", slot.n_decoded},
  1387. {"stopped_eos", slot.stopped_eos},
  1388. {"stopped_word", slot.stopped_word},
  1389. {"stopped_limit", slot.stopped_limit},
  1390. {"stopping_word", slot.stopping_word},
  1391. };
  1392. if (slot.is_processing()) {
  1393. n_processing_slots++;
  1394. } else {
  1395. n_idle_slots++;
  1396. }
  1397. slots_data.push_back(slot_data);
  1398. }
  1399. SRV_DBG("n_idle_slots = %d, n_processing_slots = %d\n", n_idle_slots, n_processing_slots);
  1400. server_task_result res;
  1401. res.id = task.id;
  1402. res.stop = true;
  1403. res.error = false;
  1404. res.data = {
  1405. { "idle", n_idle_slots },
  1406. { "processing", n_processing_slots },
  1407. { "deferred", queue_tasks.queue_tasks_deferred.size() },
  1408. { "t_start", metrics.t_start},
  1409. { "n_prompt_tokens_processed_total", metrics.n_prompt_tokens_processed_total},
  1410. { "t_tokens_generation_total", metrics.t_tokens_generation_total},
  1411. { "n_tokens_predicted_total", metrics.n_tokens_predicted_total},
  1412. { "t_prompt_processing_total", metrics.t_prompt_processing_total},
  1413. { "n_prompt_tokens_processed", metrics.n_prompt_tokens_processed},
  1414. { "t_prompt_processing", metrics.t_prompt_processing},
  1415. { "n_tokens_predicted", metrics.n_tokens_predicted},
  1416. { "t_tokens_generation", metrics.t_tokens_generation},
  1417. { "n_decode_total", metrics.n_decode_total},
  1418. { "n_busy_slots_total", metrics.n_busy_slots_total},
  1419. { "kv_cache_tokens_count", llama_get_kv_cache_token_count(ctx)},
  1420. { "kv_cache_used_cells", llama_get_kv_cache_used_cells(ctx)},
  1421. { "slots", slots_data },
  1422. };
  1423. if (json_value(task.data, "reset_bucket", false)) {
  1424. metrics.reset_bucket();
  1425. }
  1426. queue_results.send(res);
  1427. } break;
  1428. case SERVER_TASK_TYPE_SLOT_SAVE:
  1429. {
  1430. int id_slot = task.data.at("id_slot");
  1431. server_slot * slot = get_slot_by_id(id_slot);
  1432. if (slot == nullptr) {
  1433. send_error(task, "Invalid slot ID", ERROR_TYPE_INVALID_REQUEST);
  1434. break;
  1435. }
  1436. if (slot->is_processing()) {
  1437. // if requested slot is unavailable, we defer this task for processing later
  1438. SRV_DBG("requested slot is unavailable, defer task, id_task = %d\n", task.id);
  1439. queue_tasks.defer(task);
  1440. break;
  1441. }
  1442. const size_t token_count = slot->cache_tokens.size();
  1443. const int64_t t_start = ggml_time_us();
  1444. std::string filename = task.data.at("filename");
  1445. std::string filepath = task.data.at("filepath");
  1446. const size_t nwrite = llama_state_seq_save_file(ctx, filepath.c_str(), slot->id, slot->cache_tokens.data(), token_count);
  1447. const int64_t t_end = ggml_time_us();
  1448. const double t_save_ms = (t_end - t_start) / 1000.0;
  1449. server_task_result result;
  1450. result.id = task.id;
  1451. result.stop = true;
  1452. result.error = false;
  1453. result.data = json {
  1454. { "id_slot", id_slot },
  1455. { "filename", filename },
  1456. { "n_saved", token_count }, // tokens saved
  1457. { "n_written", nwrite }, // bytes written
  1458. { "timings", {
  1459. { "save_ms", t_save_ms }
  1460. } }
  1461. };
  1462. queue_results.send(result);
  1463. } break;
  1464. case SERVER_TASK_TYPE_SLOT_RESTORE:
  1465. {
  1466. int id_slot = task.data.at("id_slot");
  1467. server_slot * slot = get_slot_by_id(id_slot);
  1468. if (slot == nullptr) {
  1469. send_error(task, "Invalid slot ID", ERROR_TYPE_INVALID_REQUEST);
  1470. break;
  1471. }
  1472. if (slot->is_processing()) {
  1473. // if requested slot is unavailable, we defer this task for processing later
  1474. SRV_DBG("requested slot is unavailable, defer task, id_task = %d\n", task.id);
  1475. queue_tasks.defer(task);
  1476. break;
  1477. }
  1478. const int64_t t_start = ggml_time_us();
  1479. std::string filename = task.data.at("filename");
  1480. std::string filepath = task.data.at("filepath");
  1481. slot->cache_tokens.resize(slot->n_ctx);
  1482. size_t token_count = 0;
  1483. size_t nread = llama_state_seq_load_file(ctx, filepath.c_str(), slot->id, slot->cache_tokens.data(), slot->cache_tokens.size(), &token_count);
  1484. if (nread == 0) {
  1485. slot->cache_tokens.resize(0);
  1486. send_error(task, "Unable to restore slot, no available space in KV cache or invalid slot save file", ERROR_TYPE_INVALID_REQUEST);
  1487. break;
  1488. }
  1489. slot->cache_tokens.resize(token_count);
  1490. const int64_t t_end = ggml_time_us();
  1491. const double t_restore_ms = (t_end - t_start) / 1000.0;
  1492. server_task_result result;
  1493. result.id = task.id;
  1494. result.stop = true;
  1495. result.error = false;
  1496. result.data = json {
  1497. { "id_slot", id_slot },
  1498. { "filename", filename },
  1499. { "n_restored", token_count }, // tokens restored
  1500. { "n_read", nread }, // bytes read
  1501. { "timings", {
  1502. { "restore_ms", t_restore_ms }
  1503. } }
  1504. };
  1505. queue_results.send(result);
  1506. } break;
  1507. case SERVER_TASK_TYPE_SLOT_ERASE:
  1508. {
  1509. int id_slot = task.data.at("id_slot");
  1510. server_slot * slot = get_slot_by_id(id_slot);
  1511. if (slot == nullptr) {
  1512. send_error(task, "Invalid slot ID", ERROR_TYPE_INVALID_REQUEST);
  1513. break;
  1514. }
  1515. if (slot->is_processing()) {
  1516. // if requested slot is unavailable, we defer this task for processing later
  1517. SRV_DBG("requested slot is unavailable, defer task, id_task = %d\n", task.id);
  1518. queue_tasks.defer(task);
  1519. break;
  1520. }
  1521. // Erase token cache
  1522. const size_t n_erased = slot->cache_tokens.size();
  1523. llama_kv_cache_seq_rm(ctx, slot->id, -1, -1);
  1524. slot->cache_tokens.clear();
  1525. server_task_result result;
  1526. result.id = task.id;
  1527. result.stop = true;
  1528. result.error = false;
  1529. result.data = json {
  1530. { "id_slot", id_slot },
  1531. { "n_erased", n_erased }
  1532. };
  1533. queue_results.send(result);
  1534. } break;
  1535. case SERVER_TASK_TYPE_SET_LORA:
  1536. {
  1537. common_lora_adapters_apply(ctx, loras);
  1538. server_task_result result;
  1539. result.id = task.id;
  1540. result.stop = true;
  1541. result.error = false;
  1542. result.data = json{{ "success", true }};
  1543. queue_results.send(result);
  1544. } break;
  1545. }
  1546. }
  1547. void update_slots() {
  1548. // check if all slots are idle
  1549. {
  1550. bool all_idle = true;
  1551. for (auto & slot : slots) {
  1552. if (slot.is_processing()) {
  1553. all_idle = false;
  1554. break;
  1555. }
  1556. }
  1557. if (all_idle) {
  1558. SRV_INF("%s", "all slots are idle\n");
  1559. if (clean_kv_cache) {
  1560. kv_cache_clear();
  1561. }
  1562. return;
  1563. }
  1564. }
  1565. {
  1566. SRV_DBG("%s", "posting NEXT_RESPONSE\n");
  1567. server_task task;
  1568. task.type = SERVER_TASK_TYPE_NEXT_RESPONSE;
  1569. task.id_target = -1;
  1570. queue_tasks.post(task);
  1571. }
  1572. // apply context-shift if needed
  1573. // TODO: simplify and improve
  1574. for (server_slot & slot : slots) {
  1575. if (slot.is_processing() && slot.n_past + 1 >= slot.n_ctx) {
  1576. if (!params_base.ctx_shift) {
  1577. // this check is redundant (for good)
  1578. // we should never get here, because generation should already stopped in process_token()
  1579. slot.release();
  1580. send_error(slot, "context shift is disabled", ERROR_TYPE_SERVER);
  1581. continue;
  1582. }
  1583. // Shift context
  1584. const int n_keep = slot.params.n_keep + add_bos_token;
  1585. const int n_left = slot.n_past - n_keep;
  1586. const int n_discard = slot.params.n_discard ? slot.params.n_discard : (n_left / 2);
  1587. SLT_WRN(slot, "slot context shift, n_keep = %d, n_left = %d, n_discard = %d\n", n_keep, n_left, n_discard);
  1588. llama_kv_cache_seq_rm (ctx, slot.id, n_keep , n_keep + n_discard);
  1589. llama_kv_cache_seq_add(ctx, slot.id, n_keep + n_discard, slot.n_past, -n_discard);
  1590. if (slot.params.cache_prompt) {
  1591. for (size_t i = n_keep + n_discard; i < slot.cache_tokens.size(); i++) {
  1592. slot.cache_tokens[i - n_discard] = slot.cache_tokens[i];
  1593. }
  1594. slot.cache_tokens.resize(slot.cache_tokens.size() - n_discard);
  1595. }
  1596. slot.n_past -= n_discard;
  1597. slot.truncated = true;
  1598. }
  1599. }
  1600. // start populating the batch for this iteration
  1601. common_batch_clear(batch);
  1602. // frist, add sampled tokens from any ongoing sequences
  1603. for (auto & slot : slots) {
  1604. if (slot.state != SLOT_STATE_GENERATING) {
  1605. continue;
  1606. }
  1607. slot.i_batch = batch.n_tokens;
  1608. common_batch_add(batch, slot.sampled, slot.n_past, { slot.id }, true);
  1609. slot.n_past += 1;
  1610. if (slot.params.cache_prompt) {
  1611. slot.cache_tokens.push_back(slot.sampled);
  1612. }
  1613. SLT_DBG(slot, "slot decode token, n_ctx = %d, n_past = %d, n_cache_tokens = %d, truncated = %d\n",
  1614. slot.n_ctx, slot.n_past, (int) slot.cache_tokens.size(), slot.truncated);
  1615. }
  1616. // process in chunks of params.n_batch
  1617. int32_t n_batch = llama_n_batch(ctx);
  1618. int32_t n_ubatch = llama_n_ubatch(ctx);
  1619. // track if this is an embedding or non-embedding batch
  1620. // if we've added sampled tokens above, we are in non-embedding mode
  1621. // -1: none, 0: non-embedding, 1: embedding
  1622. // TODO: make enum
  1623. int32_t batch_type = batch.n_tokens > 0 ? 0 : -1;
  1624. // next, batch any pending prompts without exceeding n_batch
  1625. if (params_base.cont_batching || batch.n_tokens == 0) {
  1626. for (auto & slot : slots) {
  1627. // this slot still has a prompt to be processed
  1628. if (slot.state == SLOT_STATE_PROCESSING_PROMPT || slot.state == SLOT_STATE_STARTED) {
  1629. auto & prompt_tokens = slot.prompt_tokens;
  1630. // TODO: maybe move branch to outside of this loop in the future
  1631. if (slot.state == SLOT_STATE_STARTED) {
  1632. slot.t_start_process_prompt = ggml_time_us();
  1633. slot.t_start_generation = 0;
  1634. slot.n_past = 0;
  1635. slot.n_prompt_tokens = prompt_tokens.size();
  1636. slot.state = SLOT_STATE_PROCESSING_PROMPT;
  1637. SLT_INF(slot, "new prompt, n_ctx_slot = %d, n_keep = %d, n_prompt_tokens = %d\n", slot.n_ctx, slot.params.n_keep, slot.n_prompt_tokens);
  1638. // print prompt tokens (for debugging)
  1639. if (1) {
  1640. // first 16 tokens (avoid flooding logs)
  1641. for (int i = 0; i < std::min<int>(16, prompt_tokens.size()); i++) {
  1642. SLT_DBG(slot, "prompt token %3d: %6d '%s'\n", i, prompt_tokens[i], common_token_to_piece(ctx, prompt_tokens[i]).c_str());
  1643. }
  1644. } else {
  1645. // all
  1646. for (int i = 0; i < (int) prompt_tokens.size(); i++) {
  1647. SLT_DBG(slot, "prompt token %3d: %6d '%s'\n", i, prompt_tokens[i], common_token_to_piece(ctx, prompt_tokens[i]).c_str());
  1648. }
  1649. }
  1650. // empty prompt passed -> release the slot and send empty response
  1651. if (prompt_tokens.empty()) {
  1652. SLT_WRN(slot, "%s", "empty prompt - releasing slot\n");
  1653. slot.release();
  1654. slot.print_timings();
  1655. send_final_response(slot);
  1656. continue;
  1657. }
  1658. if (slot.inf_type == SERVER_TASK_INF_TYPE_EMBEDDING || slot.inf_type == SERVER_TASK_INF_TYPE_RERANK) {
  1659. if (slot.n_prompt_tokens > n_ubatch) {
  1660. slot.release();
  1661. send_error(slot, "input is too large to process. increase the physical batch size", ERROR_TYPE_SERVER);
  1662. continue;
  1663. }
  1664. if (slot.n_prompt_tokens > slot.n_ctx) {
  1665. slot.release();
  1666. send_error(slot, "input is larger than the max context size. skipping", ERROR_TYPE_SERVER);
  1667. continue;
  1668. }
  1669. } else {
  1670. if (!params_base.ctx_shift) {
  1671. // if context shift is disabled, we make sure prompt size is smaller than KV size
  1672. // TODO: there should be a separate parameter that control prompt truncation
  1673. // context shift should be applied only during the generation phase
  1674. if (slot.n_prompt_tokens >= slot.n_ctx) {
  1675. slot.release();
  1676. send_error(slot, "the request exceeds the available context size. try increasing the context size or enable context shift", ERROR_TYPE_INVALID_REQUEST);
  1677. continue;
  1678. }
  1679. }
  1680. if (slot.params.n_keep < 0) {
  1681. slot.params.n_keep = slot.n_prompt_tokens;
  1682. }
  1683. slot.params.n_keep = std::min(slot.n_ctx - 4, slot.params.n_keep);
  1684. // if input prompt is too big, truncate it
  1685. if (slot.n_prompt_tokens >= slot.n_ctx) {
  1686. const int n_left = slot.n_ctx - slot.params.n_keep;
  1687. const int n_block_size = n_left / 2;
  1688. const int erased_blocks = (slot.n_prompt_tokens - slot.params.n_keep - n_block_size) / n_block_size;
  1689. llama_tokens new_tokens(
  1690. prompt_tokens.begin(),
  1691. prompt_tokens.begin() + slot.params.n_keep);
  1692. new_tokens.insert(
  1693. new_tokens.end(),
  1694. prompt_tokens.begin() + slot.params.n_keep + erased_blocks * n_block_size,
  1695. prompt_tokens.end());
  1696. prompt_tokens = std::move(new_tokens);
  1697. slot.truncated = true;
  1698. slot.n_prompt_tokens = prompt_tokens.size();
  1699. SLT_WRN(slot, "input truncated, n_ctx = %d, n_keep = %d, n_left = %d, n_prompt_tokens = %d\n", slot.n_ctx, slot.params.n_keep, n_left, slot.n_prompt_tokens);
  1700. GGML_ASSERT(slot.n_prompt_tokens < slot.n_ctx);
  1701. }
  1702. if (slot.params.cache_prompt) {
  1703. // reuse any previously computed tokens that are common with the new prompt
  1704. slot.n_past = common_lcp(slot.cache_tokens, prompt_tokens);
  1705. // reuse chunks from the cached prompt by shifting their KV cache in the new position
  1706. if (params_base.n_cache_reuse > 0) {
  1707. size_t head_c = slot.n_past; // cache
  1708. size_t head_p = slot.n_past; // current prompt
  1709. SLT_DBG(slot, "trying to reuse chunks with size > %d, slot.n_past = %d\n", params_base.n_cache_reuse, slot.n_past);
  1710. while (head_c < slot.cache_tokens.size() &&
  1711. head_p < prompt_tokens.size()) {
  1712. size_t n_match = 0;
  1713. while (head_c + n_match < slot.cache_tokens.size() &&
  1714. head_p + n_match < prompt_tokens.size() &&
  1715. slot.cache_tokens[head_c + n_match] == prompt_tokens[head_p + n_match]) {
  1716. n_match++;
  1717. }
  1718. if (n_match >= (size_t) params_base.n_cache_reuse) {
  1719. SLT_INF(slot, "reusing chunk with size %zu, shifting KV cache [%zu, %zu) -> [%zu, %zu)\n", n_match, head_c, head_c + n_match, head_p, head_p + n_match);
  1720. //for (size_t i = head_p; i < head_p + n_match; i++) {
  1721. // SLT_DBG(slot, "cache token %3zu: %6d '%s'\n", i, prompt_tokens[i], common_token_to_piece(ctx, prompt_tokens[i]).c_str());
  1722. //}
  1723. const int64_t kv_shift = (int64_t) head_p - (int64_t) head_c;
  1724. llama_kv_cache_seq_rm (ctx, slot.id, head_p, head_c);
  1725. llama_kv_cache_seq_add(ctx, slot.id, head_c, -1, kv_shift);
  1726. for (size_t i = 0; i < n_match; i++) {
  1727. slot.cache_tokens[head_p + i] = slot.cache_tokens[head_c + i];
  1728. slot.n_past++;
  1729. }
  1730. head_c += n_match;
  1731. head_p += n_match;
  1732. } else {
  1733. head_c += 1;
  1734. }
  1735. }
  1736. SLT_DBG(slot, "after context reuse, new slot.n_past = %d\n", slot.n_past);
  1737. }
  1738. }
  1739. }
  1740. if (slot.n_past == slot.n_prompt_tokens && slot.n_past > 0) {
  1741. // we have to evaluate at least 1 token to generate logits.
  1742. SLT_WRN(slot, "need to evaluate at least 1 token to generate logits, n_past = %d, n_prompt_tokens = %d\n", slot.n_past, slot.n_prompt_tokens);
  1743. slot.n_past--;
  1744. }
  1745. slot.n_prompt_tokens_processed = 0;
  1746. }
  1747. // non-causal tasks require to fit the entire prompt in the physical batch
  1748. if (slot.inf_type == SERVER_TASK_INF_TYPE_EMBEDDING || slot.inf_type == SERVER_TASK_INF_TYPE_RERANK) {
  1749. // cannot fit the prompt in the current batch - will try next iter
  1750. if (batch.n_tokens + slot.n_prompt_tokens > n_batch) {
  1751. continue;
  1752. }
  1753. }
  1754. // check that we are in the right batch_type, if not defer the slot
  1755. const bool slot_type =
  1756. slot.inf_type == SERVER_TASK_INF_TYPE_EMBEDDING ||
  1757. slot.inf_type == SERVER_TASK_INF_TYPE_RERANK ? 1 : 0;
  1758. if (batch_type == -1) {
  1759. batch_type = slot_type;
  1760. } else if (batch_type != slot_type) {
  1761. continue;
  1762. }
  1763. // keep only the common part
  1764. if (!llama_kv_cache_seq_rm(ctx, slot.id, slot.n_past, -1)) {
  1765. // could not partially delete (likely using a non-Transformer model)
  1766. llama_kv_cache_seq_rm(ctx, slot.id, -1, -1);
  1767. // there is no common part left
  1768. slot.n_past = 0;
  1769. }
  1770. SLT_INF(slot, "kv cache rm [%d, end)\n", slot.n_past);
  1771. // remove the non-common part from the cache
  1772. slot.cache_tokens.resize(slot.n_past);
  1773. // add prompt tokens for processing in the current batch
  1774. while (slot.n_past < slot.n_prompt_tokens && batch.n_tokens < n_batch) {
  1775. common_batch_add(batch, prompt_tokens[slot.n_past], slot.n_past, { slot.id }, false);
  1776. if (slot.params.cache_prompt) {
  1777. slot.cache_tokens.push_back(prompt_tokens[slot.n_past]);
  1778. }
  1779. slot.n_prompt_tokens_processed++;
  1780. slot.n_past++;
  1781. }
  1782. SLT_INF(slot, "prompt processing progress, n_past = %d, n_tokens = %d, progress = %f\n", slot.n_past, batch.n_tokens, (float) slot.n_prompt_tokens_processed / slot.n_prompt_tokens);
  1783. // entire prompt has been processed
  1784. if (slot.n_past == slot.n_prompt_tokens) {
  1785. slot.state = SLOT_STATE_DONE_PROMPT;
  1786. GGML_ASSERT(batch.n_tokens > 0);
  1787. common_sampler_reset(slot.smpl);
  1788. // Process all prompt tokens through sampler system
  1789. for (int i = 0; i < slot.n_prompt_tokens; ++i) {
  1790. common_sampler_accept(slot.smpl, prompt_tokens[i], false);
  1791. }
  1792. // extract the logits only for the last token
  1793. batch.logits[batch.n_tokens - 1] = true;
  1794. slot.n_decoded = 0;
  1795. slot.i_batch = batch.n_tokens - 1;
  1796. SLT_INF(slot, "prompt done, n_past = %d, n_tokens = %d\n", slot.n_past, batch.n_tokens);
  1797. }
  1798. }
  1799. if (batch.n_tokens >= n_batch) {
  1800. break;
  1801. }
  1802. }
  1803. }
  1804. if (batch.n_tokens == 0) {
  1805. SRV_WRN("%s", "no tokens to decode\n");
  1806. return;
  1807. }
  1808. SRV_DBG("decoding batch, n_tokens = %d\n", batch.n_tokens);
  1809. // make sure we're in the right embedding mode
  1810. llama_set_embeddings(ctx, batch_type == 1);
  1811. // process the created batch of tokens
  1812. for (int32_t i = 0; i < batch.n_tokens; i += n_batch) {
  1813. const int32_t n_tokens = std::min(n_batch, batch.n_tokens - i);
  1814. llama_batch batch_view = {
  1815. n_tokens,
  1816. batch.token + i,
  1817. nullptr,
  1818. batch.pos + i,
  1819. batch.n_seq_id + i,
  1820. batch.seq_id + i,
  1821. batch.logits + i,
  1822. };
  1823. const int ret = llama_decode(ctx, batch_view);
  1824. metrics.on_decoded(slots);
  1825. if (ret != 0) {
  1826. if (n_batch == 1 || ret < 0) {
  1827. // if you get here, it means the KV cache is full - try increasing it via the context size
  1828. SRV_ERR("failed to decode the batch: KV cache is full - try increasing it via the context size, i = %d, n_batch = %d, ret = %d\n", i, n_batch, ret);
  1829. for (auto & slot : slots) {
  1830. slot.release();
  1831. send_error(slot, "Input prompt is too big compared to KV size. Please try increasing KV size.");
  1832. }
  1833. break; // break loop of n_batch
  1834. }
  1835. // retry with half the batch size to try to find a free slot in the KV cache
  1836. n_batch /= 2;
  1837. i -= n_batch;
  1838. SRV_WRN("failed to find free space in the KV cache, retrying with smaller batch size - try increasing it via the context size or enable defragmentation, i = %d, n_batch = %d, ret = %d\n", i, n_batch, ret);
  1839. continue; // continue loop of n_batch
  1840. }
  1841. for (auto & slot : slots) {
  1842. if (slot.i_batch < (int) i || slot.i_batch >= (int) (i + n_tokens)) {
  1843. continue; // continue loop of slots
  1844. }
  1845. if (slot.state == SLOT_STATE_DONE_PROMPT) {
  1846. if (slot.inf_type == SERVER_TASK_INF_TYPE_EMBEDDING) {
  1847. // prompt evaluated for embedding
  1848. send_embedding(slot, batch_view);
  1849. slot.release();
  1850. slot.i_batch = -1;
  1851. continue; // continue loop of slots
  1852. }
  1853. if (slot.inf_type == SERVER_TASK_INF_TYPE_RERANK) {
  1854. send_rerank(slot, batch_view);
  1855. slot.release();
  1856. slot.i_batch = -1;
  1857. continue; // continue loop of slots
  1858. }
  1859. // prompt evaluated for next-token prediction
  1860. slot.state = SLOT_STATE_GENERATING;
  1861. } else if (slot.state != SLOT_STATE_GENERATING) {
  1862. continue; // continue loop of slots
  1863. }
  1864. llama_token id = common_sampler_sample(slot.smpl, ctx, slot.i_batch - i);
  1865. slot.i_batch = -1;
  1866. common_sampler_accept(slot.smpl, id, true);
  1867. slot.n_decoded += 1;
  1868. const int64_t t_current = ggml_time_us();
  1869. if (slot.n_decoded == 1) {
  1870. slot.t_start_generation = t_current;
  1871. slot.t_prompt_processing = (slot.t_start_generation - slot.t_start_process_prompt) / 1e3;
  1872. metrics.on_prompt_eval(slot);
  1873. }
  1874. slot.t_token_generation = (t_current - slot.t_start_generation) / 1e3;
  1875. completion_token_output result;
  1876. result.tok = id;
  1877. const auto * cur_p = common_sampler_get_candidates(slot.smpl);
  1878. for (size_t i = 0; i < (size_t) slot.params.sampling.n_probs; ++i) {
  1879. result.probs.push_back({
  1880. cur_p->data[i].id,
  1881. i >= cur_p->size ? 0.0f : cur_p->data[i].p,
  1882. });
  1883. }
  1884. if (!process_token(result, slot)) {
  1885. // release slot because of stop condition
  1886. slot.release();
  1887. slot.print_timings();
  1888. send_final_response(slot);
  1889. metrics.on_prediction(slot);
  1890. continue;
  1891. }
  1892. }
  1893. // do speculative decoding
  1894. for (auto & slot : slots) {
  1895. if (!slot.is_processing() || !slot.can_speculate()) {
  1896. continue;
  1897. }
  1898. if (slot.state != SLOT_STATE_GENERATING) {
  1899. continue;
  1900. }
  1901. llama_token id = slot.sampled;
  1902. struct common_speculative_params params_spec;
  1903. params_spec.n_draft = slot.params.speculative.n_max;
  1904. params_spec.n_reuse = llama_n_ctx(slot.ctx_dft) - slot.params.speculative.n_max;
  1905. params_spec.p_min = slot.params.speculative.p_min;
  1906. llama_tokens draft = common_speculative_gen_draft(slot.spec, params_spec, slot.cache_tokens, id);
  1907. // ignore small drafts
  1908. if (slot.params.speculative.n_min > (int) draft.size()) {
  1909. continue;
  1910. }
  1911. // construct the speculation batch
  1912. common_batch_clear(slot.batch_spec);
  1913. common_batch_add (slot.batch_spec, id, slot.n_past, { slot.id }, true);
  1914. for (size_t i = 0; i < draft.size(); ++i) {
  1915. common_batch_add(slot.batch_spec, draft[i], slot.n_past + 1 + i, { slot.id }, true);
  1916. }
  1917. llama_decode(ctx, slot.batch_spec);
  1918. // the accepted tokens from the speculation
  1919. const auto ids = common_sampler_sample_and_accept_n(slot.smpl, ctx, draft);
  1920. slot.n_past += ids.size();
  1921. slot.n_decoded += ids.size();
  1922. slot.cache_tokens.push_back(id);
  1923. slot.cache_tokens.insert(slot.cache_tokens.end(), ids.begin(), ids.end() - 1);
  1924. llama_kv_cache_seq_rm(ctx, slot.id, slot.n_past, -1);
  1925. for (size_t i = 0; i < ids.size(); ++i) {
  1926. completion_token_output result;
  1927. result.tok = ids[i];
  1928. if (!process_token(result, slot)) {
  1929. // release slot because of stop condition
  1930. slot.release();
  1931. slot.print_timings();
  1932. send_final_response(slot);
  1933. metrics.on_prediction(slot);
  1934. break;
  1935. }
  1936. }
  1937. SRV_DBG("accepted %d/%d draft tokens\n", (int) ids.size() - 1, (int) draft.size());
  1938. }
  1939. }
  1940. SRV_DBG("%s", "run slots completed\n");
  1941. }
  1942. json model_meta() const {
  1943. return json {
  1944. {"vocab_type", llama_vocab_type (model)},
  1945. {"n_vocab", llama_n_vocab (model)},
  1946. {"n_ctx_train", llama_n_ctx_train (model)},
  1947. {"n_embd", llama_n_embd (model)},
  1948. {"n_params", llama_model_n_params(model)},
  1949. {"size", llama_model_size (model)},
  1950. };
  1951. }
  1952. };
  1953. static void log_server_request(const httplib::Request & req, const httplib::Response & res) {
  1954. // skip GH copilot requests when using default port
  1955. if (req.path == "/v1/health" || req.path == "/v1/completions") {
  1956. return;
  1957. }
  1958. LOG_INF("request: %s %s %s %d\n", req.method.c_str(), req.path.c_str(), req.remote_addr.c_str(), res.status);
  1959. LOG_DBG("request: %s\n", req.body.c_str());
  1960. LOG_DBG("response: %s\n", res.body.c_str());
  1961. }
  1962. std::function<void(int)> shutdown_handler;
  1963. std::atomic_flag is_terminating = ATOMIC_FLAG_INIT;
  1964. inline void signal_handler(int signal) {
  1965. if (is_terminating.test_and_set()) {
  1966. // in case it hangs, we can force terminate the server by hitting Ctrl+C twice
  1967. // this is for better developer experience, we can remove when the server is stable enough
  1968. fprintf(stderr, "Received second interrupt, terminating immediately.\n");
  1969. exit(1);
  1970. }
  1971. shutdown_handler(signal);
  1972. }
  1973. int main(int argc, char ** argv) {
  1974. // own arguments required by this example
  1975. common_params params;
  1976. if (!common_params_parse(argc, argv, params, LLAMA_EXAMPLE_SERVER)) {
  1977. return 1;
  1978. }
  1979. common_init();
  1980. // enabling this will output extra debug information in the HTTP responses from the server
  1981. // see format_final_response_oaicompat()
  1982. const bool verbose = params.verbosity > 9;
  1983. // struct that contains llama context and inference
  1984. server_context ctx_server;
  1985. if (params.model_alias == "unknown") {
  1986. params.model_alias = params.model;
  1987. }
  1988. llama_backend_init();
  1989. llama_numa_init(params.numa);
  1990. LOG_INF("system info: n_threads = %d, n_threads_batch = %d, total_threads = %d\n", params.cpuparams.n_threads, params.cpuparams_batch.n_threads, std::thread::hardware_concurrency());
  1991. LOG_INF("\n");
  1992. LOG_INF("%s\n", common_params_get_system_info(params).c_str());
  1993. LOG_INF("\n");
  1994. // static files
  1995. std::map<std::string, server_static_file> static_files = {
  1996. { "/", { index_html, index_html_len, "text/html; charset=utf-8" }},
  1997. { "/completion.js", { completion_js, completion_js_len, "text/javascript; charset=utf-8" }},
  1998. { "/deps_daisyui.min.css", { deps_daisyui_min_css, deps_daisyui_min_css_len, "text/css; charset=utf-8" }},
  1999. { "/deps_markdown-it.js", { deps_markdown_it_js, deps_markdown_it_js_len, "text/javascript; charset=utf-8" }},
  2000. { "/deps_tailwindcss.js", { deps_tailwindcss_js, deps_tailwindcss_js_len, "text/javascript; charset=utf-8" }},
  2001. { "/deps_vue.esm-browser.js", { deps_vue_esm_browser_js, deps_vue_esm_browser_js_len, "text/javascript; charset=utf-8" }},
  2002. };
  2003. std::unique_ptr<httplib::Server> svr;
  2004. #ifdef CPPHTTPLIB_OPENSSL_SUPPORT
  2005. if (params.ssl_file_key != "" && params.ssl_file_cert != "") {
  2006. LOG_INF("Running with SSL: key = %s, cert = %s\n", params.ssl_file_key.c_str(), params.ssl_file_cert.c_str());
  2007. svr.reset(
  2008. new httplib::SSLServer(params.ssl_file_cert.c_str(), params.ssl_file_key.c_str())
  2009. );
  2010. } else {
  2011. LOG_INF("Running without SSL\n");
  2012. svr.reset(new httplib::Server());
  2013. }
  2014. #else
  2015. if (params.ssl_file_key != "" && params.ssl_file_cert != "") {
  2016. LOG_ERR("Server is built without SSL support\n");
  2017. return 1;
  2018. }
  2019. svr.reset(new httplib::Server());
  2020. #endif
  2021. std::atomic<server_state> state{SERVER_STATE_LOADING_MODEL};
  2022. svr->set_default_headers({{"Server", "llama.cpp"}});
  2023. svr->set_logger(log_server_request);
  2024. auto res_error = [](httplib::Response & res, const json & error_data) {
  2025. json final_response {{"error", error_data}};
  2026. res.set_content(final_response.dump(-1, ' ', false, json::error_handler_t::replace), MIMETYPE_JSON);
  2027. res.status = json_value(error_data, "code", 500);
  2028. };
  2029. auto res_ok = [](httplib::Response & res, const json & data) {
  2030. res.set_content(data.dump(-1, ' ', false, json::error_handler_t::replace), MIMETYPE_JSON);
  2031. res.status = 200;
  2032. };
  2033. svr->set_exception_handler([&res_error](const httplib::Request &, httplib::Response & res, std::exception_ptr ep) {
  2034. std::string message;
  2035. try {
  2036. std::rethrow_exception(ep);
  2037. } catch (std::exception & e) {
  2038. message = e.what();
  2039. } catch (...) {
  2040. message = "Unknown Exception";
  2041. }
  2042. json formatted_error = format_error_response(message, ERROR_TYPE_SERVER);
  2043. LOG_WRN("got exception: %s\n", formatted_error.dump().c_str());
  2044. res_error(res, formatted_error);
  2045. });
  2046. svr->set_error_handler([&res_error](const httplib::Request &, httplib::Response & res) {
  2047. if (res.status == 404) {
  2048. res_error(res, format_error_response("File Not Found", ERROR_TYPE_NOT_FOUND));
  2049. }
  2050. // for other error codes, we skip processing here because it's already done by res_error()
  2051. });
  2052. // set timeouts and change hostname and port
  2053. svr->set_read_timeout (params.timeout_read);
  2054. svr->set_write_timeout(params.timeout_write);
  2055. std::unordered_map<std::string, std::string> log_data;
  2056. log_data["hostname"] = params.hostname;
  2057. log_data["port"] = std::to_string(params.port);
  2058. if (params.api_keys.size() == 1) {
  2059. auto key = params.api_keys[0];
  2060. log_data["api_key"] = "api_key: ****" + key.substr(std::max((int)(key.length() - 4), 0));
  2061. } else if (params.api_keys.size() > 1) {
  2062. log_data["api_key"] = "api_key: " + std::to_string(params.api_keys.size()) + " keys loaded";
  2063. }
  2064. // Necessary similarity of prompt for slot selection
  2065. ctx_server.slot_prompt_similarity = params.slot_prompt_similarity;
  2066. //
  2067. // Middlewares
  2068. //
  2069. auto middleware_validate_api_key = [&params, &res_error, &static_files](const httplib::Request & req, httplib::Response & res) {
  2070. static const std::unordered_set<std::string> public_endpoints = {
  2071. "/health",
  2072. "/models",
  2073. "/v1/models",
  2074. };
  2075. // If API key is not set, skip validation
  2076. if (params.api_keys.empty()) {
  2077. return true;
  2078. }
  2079. // If path is public or is static file, skip validation
  2080. if (public_endpoints.find(req.path) != public_endpoints.end() || static_files.find(req.path) != static_files.end()) {
  2081. return true;
  2082. }
  2083. // Check for API key in the header
  2084. auto auth_header = req.get_header_value("Authorization");
  2085. std::string prefix = "Bearer ";
  2086. if (auth_header.substr(0, prefix.size()) == prefix) {
  2087. std::string received_api_key = auth_header.substr(prefix.size());
  2088. if (std::find(params.api_keys.begin(), params.api_keys.end(), received_api_key) != params.api_keys.end()) {
  2089. return true; // API key is valid
  2090. }
  2091. }
  2092. // API key is invalid or not provided
  2093. res_error(res, format_error_response("Invalid API Key", ERROR_TYPE_AUTHENTICATION));
  2094. LOG_WRN("Unauthorized: Invalid API Key\n");
  2095. return false;
  2096. };
  2097. auto middleware_server_state = [&res_error, &state](const httplib::Request & req, httplib::Response & res) {
  2098. server_state current_state = state.load();
  2099. if (current_state == SERVER_STATE_LOADING_MODEL) {
  2100. auto tmp = string_split<std::string>(req.path, '.');
  2101. if (req.path == "/" || tmp.back() == "html") {
  2102. res.set_content(reinterpret_cast<const char*>(loading_html), loading_html_len, "text/html; charset=utf-8");
  2103. res.status = 503;
  2104. } else {
  2105. res_error(res, format_error_response("Loading model", ERROR_TYPE_UNAVAILABLE));
  2106. }
  2107. return false;
  2108. }
  2109. return true;
  2110. };
  2111. // register server middlewares
  2112. svr->set_pre_routing_handler([&middleware_validate_api_key, &middleware_server_state](const httplib::Request & req, httplib::Response & res) {
  2113. res.set_header("Access-Control-Allow-Origin", req.get_header_value("Origin"));
  2114. // If this is OPTIONS request, skip validation because browsers don't include Authorization header
  2115. if (req.method == "OPTIONS") {
  2116. res.set_header("Access-Control-Allow-Credentials", "true");
  2117. res.set_header("Access-Control-Allow-Methods", "GET, POST");
  2118. res.set_header("Access-Control-Allow-Headers", "*");
  2119. res.set_content("", "text/html"); // blank response, no data
  2120. return httplib::Server::HandlerResponse::Handled; // skip further processing
  2121. }
  2122. if (!middleware_server_state(req, res)) {
  2123. return httplib::Server::HandlerResponse::Handled;
  2124. }
  2125. if (!middleware_validate_api_key(req, res)) {
  2126. return httplib::Server::HandlerResponse::Handled;
  2127. }
  2128. return httplib::Server::HandlerResponse::Unhandled;
  2129. });
  2130. //
  2131. // Route handlers (or controllers)
  2132. //
  2133. const auto handle_health = [&](const httplib::Request &, httplib::Response & res) {
  2134. // error and loading states are handled by middleware
  2135. json health = {{"status", "ok"}};
  2136. res_ok(res, health);
  2137. };
  2138. const auto handle_slots = [&](const httplib::Request & req, httplib::Response & res) {
  2139. if (!params.endpoint_slots) {
  2140. res_error(res, format_error_response("This server does not support slots endpoint. Start it with `--slots`", ERROR_TYPE_NOT_SUPPORTED));
  2141. return;
  2142. }
  2143. // request slots data using task queue
  2144. server_task task;
  2145. task.id = ctx_server.queue_tasks.get_new_id();
  2146. task.type = SERVER_TASK_TYPE_METRICS;
  2147. ctx_server.queue_results.add_waiting_task_id(task.id);
  2148. ctx_server.queue_tasks.post(task, true); // high-priority task
  2149. // get the result
  2150. server_task_result result = ctx_server.queue_results.recv(task.id);
  2151. ctx_server.queue_results.remove_waiting_task_id(task.id);
  2152. // optionally return "fail_on_no_slot" error
  2153. const int n_idle_slots = result.data.at("idle");
  2154. if (req.has_param("fail_on_no_slot")) {
  2155. if (n_idle_slots == 0) {
  2156. res_error(res, format_error_response("no slot available", ERROR_TYPE_UNAVAILABLE));
  2157. return;
  2158. }
  2159. }
  2160. res_ok(res, result.data.at("slots"));
  2161. };
  2162. const auto handle_metrics = [&](const httplib::Request &, httplib::Response & res) {
  2163. if (!params.endpoint_metrics) {
  2164. res_error(res, format_error_response("This server does not support metrics endpoint. Start it with `--metrics`", ERROR_TYPE_NOT_SUPPORTED));
  2165. return;
  2166. }
  2167. // request slots data using task queue
  2168. server_task task;
  2169. task.id = ctx_server.queue_tasks.get_new_id();
  2170. task.id_target = -1;
  2171. task.type = SERVER_TASK_TYPE_METRICS;
  2172. task.data.push_back({{"reset_bucket", true}});
  2173. ctx_server.queue_results.add_waiting_task_id(task.id);
  2174. ctx_server.queue_tasks.post(task, true); // high-priority task
  2175. // get the result
  2176. server_task_result result = ctx_server.queue_results.recv(task.id);
  2177. ctx_server.queue_results.remove_waiting_task_id(task.id);
  2178. json data = result.data;
  2179. const uint64_t n_prompt_tokens_processed = data.at("n_prompt_tokens_processed");
  2180. const uint64_t t_prompt_processing = data.at("t_prompt_processing");
  2181. const uint64_t n_tokens_predicted = data.at("n_tokens_predicted");
  2182. const uint64_t t_tokens_generation = data.at("t_tokens_generation");
  2183. const uint64_t n_decode_total = data.at("n_decode_total");
  2184. const uint64_t n_busy_slots_total = data.at("n_busy_slots_total");
  2185. const int32_t kv_cache_used_cells = data.at("kv_cache_used_cells");
  2186. // metrics definition: https://prometheus.io/docs/practices/naming/#metric-names
  2187. json all_metrics_def = json {
  2188. {"counter", {{
  2189. {"name", "prompt_tokens_total"},
  2190. {"help", "Number of prompt tokens processed."},
  2191. {"value", (uint64_t) data.at("n_prompt_tokens_processed_total")}
  2192. }, {
  2193. {"name", "prompt_seconds_total"},
  2194. {"help", "Prompt process time"},
  2195. {"value", (uint64_t) data.at("t_prompt_processing_total") / 1.e3}
  2196. }, {
  2197. {"name", "tokens_predicted_total"},
  2198. {"help", "Number of generation tokens processed."},
  2199. {"value", (uint64_t) data.at("n_tokens_predicted_total")}
  2200. }, {
  2201. {"name", "tokens_predicted_seconds_total"},
  2202. {"help", "Predict process time"},
  2203. {"value", (uint64_t) data.at("t_tokens_generation_total") / 1.e3}
  2204. }, {
  2205. {"name", "n_decode_total"},
  2206. {"help", "Total number of llama_decode() calls"},
  2207. {"value", n_decode_total}
  2208. }, {
  2209. {"name", "n_busy_slots_per_decode"},
  2210. {"help", "Average number of busy slots per llama_decode() call"},
  2211. {"value", (float) n_busy_slots_total / (float) n_decode_total}
  2212. }}},
  2213. {"gauge", {{
  2214. {"name", "prompt_tokens_seconds"},
  2215. {"help", "Average prompt throughput in tokens/s."},
  2216. {"value", n_prompt_tokens_processed ? 1.e3 / t_prompt_processing * n_prompt_tokens_processed : 0.}
  2217. },{
  2218. {"name", "predicted_tokens_seconds"},
  2219. {"help", "Average generation throughput in tokens/s."},
  2220. {"value", n_tokens_predicted ? 1.e3 / t_tokens_generation * n_tokens_predicted : 0.}
  2221. },{
  2222. {"name", "kv_cache_usage_ratio"},
  2223. {"help", "KV-cache usage. 1 means 100 percent usage."},
  2224. {"value", 1. * kv_cache_used_cells / params.n_ctx}
  2225. },{
  2226. {"name", "kv_cache_tokens"},
  2227. {"help", "KV-cache tokens."},
  2228. {"value", (uint64_t) data.at("kv_cache_tokens_count")}
  2229. },{
  2230. {"name", "requests_processing"},
  2231. {"help", "Number of request processing."},
  2232. {"value", (uint64_t) data.at("processing")}
  2233. },{
  2234. {"name", "requests_deferred"},
  2235. {"help", "Number of request deferred."},
  2236. {"value", (uint64_t) data.at("deferred")}
  2237. }}}
  2238. };
  2239. std::stringstream prometheus;
  2240. for (const auto & el : all_metrics_def.items()) {
  2241. const auto & type = el.key();
  2242. const auto & metrics_def = el.value();
  2243. for (const auto & metric_def : metrics_def) {
  2244. const std::string name = metric_def.at("name");
  2245. const std::string help = metric_def.at("help");
  2246. auto value = json_value(metric_def, "value", 0.);
  2247. prometheus << "# HELP llamacpp:" << name << " " << help << "\n"
  2248. << "# TYPE llamacpp:" << name << " " << type << "\n"
  2249. << "llamacpp:" << name << " " << value << "\n";
  2250. }
  2251. }
  2252. const int64_t t_start = data.at("t_start");
  2253. res.set_header("Process-Start-Time-Unix", std::to_string(t_start));
  2254. res.set_content(prometheus.str(), "text/plain; version=0.0.4");
  2255. res.status = 200; // HTTP OK
  2256. };
  2257. const auto handle_slots_save = [&ctx_server, &res_error, &res_ok, &params](const httplib::Request & req, httplib::Response & res, int id_slot) {
  2258. json request_data = json::parse(req.body);
  2259. std::string filename = request_data.at("filename");
  2260. if (!fs_validate_filename(filename)) {
  2261. res_error(res, format_error_response("Invalid filename", ERROR_TYPE_INVALID_REQUEST));
  2262. return;
  2263. }
  2264. std::string filepath = params.slot_save_path + filename;
  2265. server_task task;
  2266. task.type = SERVER_TASK_TYPE_SLOT_SAVE;
  2267. task.data = {
  2268. { "id_slot", id_slot },
  2269. { "filename", filename },
  2270. { "filepath", filepath },
  2271. };
  2272. const int id_task = ctx_server.queue_tasks.post(task);
  2273. ctx_server.queue_results.add_waiting_task_id(id_task);
  2274. server_task_result result = ctx_server.queue_results.recv(id_task);
  2275. ctx_server.queue_results.remove_waiting_task_id(id_task);
  2276. if (result.error) {
  2277. res_error(res, result.data);
  2278. } else {
  2279. res_ok(res, result.data);
  2280. }
  2281. };
  2282. const auto handle_slots_restore = [&ctx_server, &res_error, &res_ok, &params](const httplib::Request & req, httplib::Response & res, int id_slot) {
  2283. json request_data = json::parse(req.body);
  2284. std::string filename = request_data.at("filename");
  2285. if (!fs_validate_filename(filename)) {
  2286. res_error(res, format_error_response("Invalid filename", ERROR_TYPE_INVALID_REQUEST));
  2287. return;
  2288. }
  2289. std::string filepath = params.slot_save_path + filename;
  2290. server_task task;
  2291. task.type = SERVER_TASK_TYPE_SLOT_RESTORE;
  2292. task.data = {
  2293. { "id_slot", id_slot },
  2294. { "filename", filename },
  2295. { "filepath", filepath },
  2296. };
  2297. const int id_task = ctx_server.queue_tasks.post(task);
  2298. ctx_server.queue_results.add_waiting_task_id(id_task);
  2299. server_task_result result = ctx_server.queue_results.recv(id_task);
  2300. ctx_server.queue_results.remove_waiting_task_id(id_task);
  2301. if (result.error) {
  2302. res_error(res, result.data);
  2303. } else {
  2304. res_ok(res, result.data);
  2305. }
  2306. };
  2307. const auto handle_slots_erase = [&ctx_server, &res_error, &res_ok](const httplib::Request & /* req */, httplib::Response & res, int id_slot) {
  2308. server_task task;
  2309. task.type = SERVER_TASK_TYPE_SLOT_ERASE;
  2310. task.data = {
  2311. { "id_slot", id_slot },
  2312. };
  2313. const int id_task = ctx_server.queue_tasks.post(task);
  2314. ctx_server.queue_results.add_waiting_task_id(id_task);
  2315. server_task_result result = ctx_server.queue_results.recv(id_task);
  2316. ctx_server.queue_results.remove_waiting_task_id(id_task);
  2317. if (result.error) {
  2318. res_error(res, result.data);
  2319. } else {
  2320. res_ok(res, result.data);
  2321. }
  2322. };
  2323. const auto handle_slots_action = [&params, &res_error, &handle_slots_save, &handle_slots_restore, &handle_slots_erase](const httplib::Request & req, httplib::Response & res) {
  2324. if (params.slot_save_path.empty()) {
  2325. res_error(res, format_error_response("This server does not support slots action. Start it with `--slot-save-path`", ERROR_TYPE_NOT_SUPPORTED));
  2326. return;
  2327. }
  2328. std::string id_slot_str = req.path_params.at("id_slot");
  2329. int id_slot;
  2330. try {
  2331. id_slot = std::stoi(id_slot_str);
  2332. } catch (const std::exception &) {
  2333. res_error(res, format_error_response("Invalid slot ID", ERROR_TYPE_INVALID_REQUEST));
  2334. return;
  2335. }
  2336. std::string action = req.get_param_value("action");
  2337. if (action == "save") {
  2338. handle_slots_save(req, res, id_slot);
  2339. } else if (action == "restore") {
  2340. handle_slots_restore(req, res, id_slot);
  2341. } else if (action == "erase") {
  2342. handle_slots_erase(req, res, id_slot);
  2343. } else {
  2344. res_error(res, format_error_response("Invalid action", ERROR_TYPE_INVALID_REQUEST));
  2345. }
  2346. };
  2347. const auto handle_props = [&ctx_server, &res_ok](const httplib::Request &, httplib::Response & res) {
  2348. json data = {
  2349. { "default_generation_settings", ctx_server.default_generation_settings_for_props },
  2350. { "total_slots", ctx_server.params_base.n_parallel },
  2351. { "chat_template", llama_get_chat_template(ctx_server.model) },
  2352. };
  2353. res_ok(res, data);
  2354. };
  2355. const auto handle_props_change = [&ctx_server, &res_error, &res_ok](const httplib::Request & req, httplib::Response & res) {
  2356. if (!ctx_server.params_base.endpoint_props) {
  2357. res_error(res, format_error_response("This server does not support changing global properties. Start it with `--props`", ERROR_TYPE_NOT_SUPPORTED));
  2358. return;
  2359. }
  2360. json data = json::parse(req.body);
  2361. // update any props here
  2362. res_ok(res, {{ "success", true }});
  2363. };
  2364. const auto handle_completions_generic = [&ctx_server, &res_error, &res_ok](server_task_inf_type inf_type, json & data, httplib::Response & res) {
  2365. if (ctx_server.params_base.embedding) {
  2366. res_error(res, format_error_response("This server does not support completions. Start it without `--embeddings`", ERROR_TYPE_NOT_SUPPORTED));
  2367. return;
  2368. }
  2369. std::vector<server_task> tasks = ctx_server.create_tasks_inference(data, inf_type);
  2370. ctx_server.queue_results.add_waiting_tasks(tasks);
  2371. ctx_server.queue_tasks.post(tasks);
  2372. bool stream = json_value(data, "stream", false);
  2373. const auto task_ids = server_task::get_list_id(tasks);
  2374. if (!stream) {
  2375. ctx_server.receive_cmpl_results(task_ids, [&](std::vector<server_task_result> & results) {
  2376. if (results.size() == 1) {
  2377. // single result
  2378. res_ok(res, results[0].data);
  2379. } else {
  2380. // multiple results (multitask)
  2381. json arr = json::array();
  2382. for (const auto & res : results) {
  2383. arr.push_back(res.data);
  2384. }
  2385. res_ok(res, arr);
  2386. }
  2387. }, [&](const json & error_data) {
  2388. res_error(res, error_data);
  2389. });
  2390. ctx_server.queue_results.remove_waiting_task_ids(task_ids);
  2391. } else {
  2392. const auto chunked_content_provider = [task_ids, &ctx_server](size_t, httplib::DataSink & sink) {
  2393. ctx_server.receive_cmpl_results_stream(task_ids, [&](const server_task_result & result) -> bool {
  2394. return server_sent_event(sink, "data", result.data);
  2395. }, [&](const json & error_data) {
  2396. server_sent_event(sink, "error", error_data);
  2397. });
  2398. sink.done();
  2399. return false;
  2400. };
  2401. auto on_complete = [task_ids, &ctx_server] (bool) {
  2402. ctx_server.queue_results.remove_waiting_task_ids(task_ids);
  2403. };
  2404. res.set_chunked_content_provider("text/event-stream", chunked_content_provider, on_complete);
  2405. }
  2406. };
  2407. const auto handle_completions = [&handle_completions_generic](const httplib::Request & req, httplib::Response & res) {
  2408. json data = json::parse(req.body);
  2409. return handle_completions_generic(SERVER_TASK_INF_TYPE_COMPLETION, data, res);
  2410. };
  2411. const auto handle_infill = [&ctx_server, &res_error, &handle_completions_generic](const httplib::Request & req, httplib::Response & res) {
  2412. // check model compatibility
  2413. std::string err;
  2414. if (llama_token_fim_pre(ctx_server.model) == LLAMA_TOKEN_NULL) {
  2415. err += "prefix token is missing. ";
  2416. }
  2417. if (llama_token_fim_suf(ctx_server.model) == LLAMA_TOKEN_NULL) {
  2418. err += "suffix token is missing. ";
  2419. }
  2420. if (llama_token_fim_mid(ctx_server.model) == LLAMA_TOKEN_NULL) {
  2421. err += "middle token is missing. ";
  2422. }
  2423. if (!err.empty()) {
  2424. res_error(res, format_error_response(string_format("Infill is not supported by this model: %s", err.c_str()), ERROR_TYPE_NOT_SUPPORTED));
  2425. return;
  2426. }
  2427. json data = json::parse(req.body);
  2428. // validate input
  2429. if (!data.contains("input_prefix")) {
  2430. res_error(res, format_error_response("\"input_prefix\" is required", ERROR_TYPE_INVALID_REQUEST));
  2431. }
  2432. if (!data.contains("input_suffix")) {
  2433. res_error(res, format_error_response("\"input_suffix\" is required", ERROR_TYPE_INVALID_REQUEST));
  2434. }
  2435. if (data.contains("input_extra") && !data.at("input_extra").is_array()) {
  2436. res_error(res, format_error_response("\"input_extra\" must be an array of {\"filename\": string, \"text\": string}", ERROR_TYPE_INVALID_REQUEST));
  2437. return;
  2438. }
  2439. json input_extra = json_value(data, "input_extra", json::array());
  2440. for (const auto & chunk : input_extra) {
  2441. // { "text": string, "filename": string }
  2442. if (!chunk.contains("text") || !chunk.at("text").is_string()) {
  2443. res_error(res, format_error_response("extra_context chunk must contain a \"text\" field with a string value", ERROR_TYPE_INVALID_REQUEST));
  2444. return;
  2445. }
  2446. // filename is optional
  2447. if (chunk.contains("filename") && !chunk.at("filename").is_string()) {
  2448. res_error(res, format_error_response("extra_context chunk's \"filename\" field must be a string", ERROR_TYPE_INVALID_REQUEST));
  2449. return;
  2450. }
  2451. }
  2452. data["input_extra"] = input_extra; // default to empty array if it's not exist
  2453. return handle_completions_generic(SERVER_TASK_INF_TYPE_INFILL, data, res);
  2454. };
  2455. // TODO: maybe merge this function with "handle_completions_generic"
  2456. const auto handle_chat_completions = [&ctx_server, &params, &res_error, &res_ok, verbose](const httplib::Request & req, httplib::Response & res) {
  2457. if (ctx_server.params_base.embedding) {
  2458. res_error(res, format_error_response("This server does not support completions. Start it without `--embeddings`", ERROR_TYPE_NOT_SUPPORTED));
  2459. return;
  2460. }
  2461. json data = oaicompat_completion_params_parse(ctx_server.model, json::parse(req.body), params.chat_template);
  2462. std::vector<server_task> tasks = ctx_server.create_tasks_inference(data, SERVER_TASK_INF_TYPE_COMPLETION);
  2463. ctx_server.queue_results.add_waiting_tasks(tasks);
  2464. ctx_server.queue_tasks.post(tasks);
  2465. bool stream = json_value(data, "stream", false);
  2466. const auto task_ids = server_task::get_list_id(tasks);
  2467. const auto completion_id = gen_chatcmplid();
  2468. if (!stream) {
  2469. ctx_server.receive_cmpl_results(task_ids, [&](const std::vector<server_task_result> & results) {
  2470. // multitask is never support in chat completion, there is only one result
  2471. json result_oai = format_final_response_oaicompat(data, results[0].data, completion_id, /*.streaming =*/ false, verbose);
  2472. res_ok(res, result_oai);
  2473. }, [&](const json & error_data) {
  2474. res_error(res, error_data);
  2475. });
  2476. ctx_server.queue_results.remove_waiting_task_ids(task_ids);
  2477. } else {
  2478. const auto chunked_content_provider = [task_ids, &ctx_server, completion_id](size_t, httplib::DataSink & sink) {
  2479. ctx_server.receive_cmpl_results_stream(task_ids, [&](const server_task_result & result) -> bool {
  2480. std::vector<json> result_array = format_partial_response_oaicompat(result.data, completion_id);
  2481. for (auto & event_data : result_array) {
  2482. if (event_data.empty()) {
  2483. continue; // skip the stop token
  2484. }
  2485. if (!server_sent_event(sink, "data", event_data)) {
  2486. return false; // connection is closed
  2487. }
  2488. }
  2489. return true; // ok
  2490. }, [&](const json & error_data) {
  2491. server_sent_event(sink, "error", error_data);
  2492. });
  2493. static const std::string ev_done = "data: [DONE]\n\n";
  2494. sink.write(ev_done.data(), ev_done.size());
  2495. sink.done();
  2496. return true;
  2497. };
  2498. auto on_complete = [task_ids, &ctx_server] (bool) {
  2499. ctx_server.queue_results.remove_waiting_task_ids(task_ids);
  2500. };
  2501. res.set_chunked_content_provider("text/event-stream", chunked_content_provider, on_complete);
  2502. }
  2503. };
  2504. const auto handle_models = [&params, &ctx_server](const httplib::Request &, httplib::Response & res) {
  2505. json models = {
  2506. {"object", "list"},
  2507. {"data", {
  2508. {
  2509. {"id", params.model_alias},
  2510. {"object", "model"},
  2511. {"created", std::time(0)},
  2512. {"owned_by", "llamacpp"},
  2513. {"meta", ctx_server.model_meta()}
  2514. },
  2515. }}
  2516. };
  2517. res.set_content(models.dump(), MIMETYPE_JSON);
  2518. };
  2519. const auto handle_tokenize = [&ctx_server, &res_ok](const httplib::Request & req, httplib::Response & res) {
  2520. const json body = json::parse(req.body);
  2521. json tokens_response = json::array();
  2522. if (body.count("content") != 0) {
  2523. const bool add_special = json_value(body, "add_special", false);
  2524. const bool with_pieces = json_value(body, "with_pieces", false);
  2525. llama_tokens tokens = tokenize_mixed(ctx_server.ctx, body.at("content"), add_special, true);
  2526. if (with_pieces) {
  2527. for (const auto& token : tokens) {
  2528. std::string piece = common_token_to_piece(ctx_server.ctx, token);
  2529. json piece_json;
  2530. // Check if the piece is valid UTF-8
  2531. if (is_valid_utf8(piece)) {
  2532. piece_json = piece;
  2533. } else {
  2534. // If not valid UTF-8, store as array of byte values
  2535. piece_json = json::array();
  2536. for (unsigned char c : piece) {
  2537. piece_json.push_back(static_cast<int>(c));
  2538. }
  2539. }
  2540. tokens_response.push_back({
  2541. {"id", token},
  2542. {"piece", piece_json}
  2543. });
  2544. }
  2545. } else {
  2546. tokens_response = tokens;
  2547. }
  2548. }
  2549. const json data = format_tokenizer_response(tokens_response);
  2550. res_ok(res, data);
  2551. };
  2552. const auto handle_detokenize = [&ctx_server, &res_ok](const httplib::Request & req, httplib::Response & res) {
  2553. const json body = json::parse(req.body);
  2554. std::string content;
  2555. if (body.count("tokens") != 0) {
  2556. const llama_tokens tokens = body.at("tokens");
  2557. content = tokens_to_str(ctx_server.ctx, tokens.cbegin(), tokens.cend());
  2558. }
  2559. const json data = format_detokenized_response(content);
  2560. res_ok(res, data);
  2561. };
  2562. const auto handle_embeddings = [&ctx_server, &res_error, &res_ok](const httplib::Request & req, httplib::Response & res) {
  2563. const json body = json::parse(req.body);
  2564. bool is_openai = false;
  2565. // an input prompt can be a string or a list of tokens (integer)
  2566. json prompt;
  2567. if (body.count("input") != 0) {
  2568. is_openai = true;
  2569. prompt = body.at("input");
  2570. } else if (body.count("content") != 0) {
  2571. // with "content", we only support single prompt
  2572. prompt = std::vector<std::string>{body.at("content")};
  2573. } else {
  2574. res_error(res, format_error_response("\"input\" or \"content\" must be provided", ERROR_TYPE_INVALID_REQUEST));
  2575. return;
  2576. }
  2577. // create and queue the task
  2578. json responses = json::array();
  2579. bool error = false;
  2580. {
  2581. std::vector<server_task> tasks = ctx_server.create_tasks_inference({{"prompt", prompt}}, SERVER_TASK_INF_TYPE_EMBEDDING);
  2582. ctx_server.queue_results.add_waiting_tasks(tasks);
  2583. ctx_server.queue_tasks.post(tasks);
  2584. // get the result
  2585. std::unordered_set<int> task_ids = server_task::get_list_id(tasks);
  2586. ctx_server.receive_cmpl_results(task_ids, [&](std::vector<server_task_result> & results) {
  2587. for (const auto & res : results) {
  2588. responses.push_back(res.data);
  2589. }
  2590. }, [&](const json & error_data) {
  2591. res_error(res, error_data);
  2592. error = true;
  2593. });
  2594. ctx_server.queue_results.remove_waiting_task_ids(task_ids);
  2595. }
  2596. if (error) {
  2597. return;
  2598. }
  2599. // write JSON response
  2600. json root = is_openai
  2601. ? format_embeddings_response_oaicompat(body, responses)
  2602. : responses[0];
  2603. res_ok(res, root);
  2604. };
  2605. const auto handle_rerank = [&ctx_server, &res_error, &res_ok](const httplib::Request & req, httplib::Response & res) {
  2606. if (!ctx_server.params_base.reranking || ctx_server.params_base.embedding) {
  2607. res_error(res, format_error_response("This server does not support reranking. Start it with `--reranking` and without `--embedding`", ERROR_TYPE_NOT_SUPPORTED));
  2608. return;
  2609. }
  2610. const json body = json::parse(req.body);
  2611. // TODO: implement
  2612. //int top_n = 1;
  2613. //if (body.count("top_n") != 1) {
  2614. // top_n = body.at("top_n");
  2615. //} else {
  2616. // res_error(res, format_error_response("\"top_n\" must be provided", ERROR_TYPE_INVALID_REQUEST));
  2617. // return;
  2618. //}
  2619. json query;
  2620. if (body.count("query") == 1) {
  2621. query = body.at("query");
  2622. if (!query.is_string()) {
  2623. res_error(res, format_error_response("\"query\" must be a string", ERROR_TYPE_INVALID_REQUEST));
  2624. return;
  2625. }
  2626. } else {
  2627. res_error(res, format_error_response("\"query\" must be provided", ERROR_TYPE_INVALID_REQUEST));
  2628. return;
  2629. }
  2630. std::vector<std::string> documents = json_value(body, "documents", std::vector<std::string>());
  2631. if (documents.empty()) {
  2632. res_error(res, format_error_response("\"documents\" must be a non-empty string array", ERROR_TYPE_INVALID_REQUEST));
  2633. return;
  2634. }
  2635. // construct prompt object: array of ["query", "doc0", "doc1", ...]
  2636. json prompt;
  2637. prompt.push_back(query);
  2638. for (const auto & doc : documents) {
  2639. prompt.push_back(doc);
  2640. }
  2641. LOG_DBG("rerank prompt: %s\n", prompt.dump().c_str());
  2642. // create and queue the task
  2643. json responses = json::array();
  2644. bool error = false;
  2645. {
  2646. std::vector<server_task> tasks = ctx_server.create_tasks_inference({{"prompt", prompt}}, SERVER_TASK_INF_TYPE_RERANK);
  2647. ctx_server.queue_results.add_waiting_tasks(tasks);
  2648. ctx_server.queue_tasks.post(tasks);
  2649. // get the result
  2650. std::unordered_set<int> task_ids = server_task::get_list_id(tasks);
  2651. ctx_server.receive_cmpl_results(task_ids, [&](std::vector<server_task_result> & results) {
  2652. for (const auto & res : results) {
  2653. responses.push_back(res.data);
  2654. }
  2655. }, [&](const json & error_data) {
  2656. res_error(res, error_data);
  2657. error = true;
  2658. });
  2659. }
  2660. if (error) {
  2661. return;
  2662. }
  2663. // write JSON response
  2664. json root = format_response_rerank(body, responses);
  2665. res_ok(res, root);
  2666. };
  2667. const auto handle_lora_adapters_list = [&](const httplib::Request &, httplib::Response & res) {
  2668. json result = json::array();
  2669. for (size_t i = 0; i < ctx_server.loras.size(); ++i) {
  2670. auto & lora = ctx_server.loras[i];
  2671. result.push_back({
  2672. {"id", i},
  2673. {"path", lora.path},
  2674. {"scale", lora.scale},
  2675. });
  2676. }
  2677. res_ok(res, result);
  2678. res.status = 200; // HTTP OK
  2679. };
  2680. const auto handle_lora_adapters_apply = [&](const httplib::Request & req, httplib::Response & res) {
  2681. const std::vector<json> body = json::parse(req.body);
  2682. int max_idx = ctx_server.loras.size();
  2683. // clear existing value
  2684. for (auto & lora : ctx_server.loras) {
  2685. lora.scale = 0.0f;
  2686. }
  2687. // set value
  2688. for (auto entry : body) {
  2689. int id = entry.at("id");
  2690. float scale = entry.at("scale");
  2691. if (0 <= id && id < max_idx) {
  2692. ctx_server.loras[id].scale = scale;
  2693. } else {
  2694. throw std::runtime_error("invalid adapter id");
  2695. }
  2696. }
  2697. server_task task;
  2698. task.type = SERVER_TASK_TYPE_SET_LORA;
  2699. const int id_task = ctx_server.queue_tasks.post(task);
  2700. ctx_server.queue_results.add_waiting_task_id(id_task);
  2701. server_task_result result = ctx_server.queue_results.recv(id_task);
  2702. ctx_server.queue_results.remove_waiting_task_id(id_task);
  2703. res_ok(res, result.data);
  2704. res.status = 200; // HTTP OK
  2705. };
  2706. //
  2707. // Router
  2708. //
  2709. // register static assets routes
  2710. if (!params.public_path.empty()) {
  2711. // Set the base directory for serving static files
  2712. bool is_found = svr->set_mount_point("/", params.public_path);
  2713. if (!is_found) {
  2714. LOG_ERR("%s: static assets path not found: %s\n", __func__, params.public_path.c_str());
  2715. return 1;
  2716. }
  2717. } else {
  2718. // using embedded static files
  2719. for (const auto & it : static_files) {
  2720. const server_static_file & static_file = it.second;
  2721. svr->Get(it.first.c_str(), [&static_file](const httplib::Request &, httplib::Response & res) {
  2722. res.set_content(reinterpret_cast<const char*>(static_file.data), static_file.size, static_file.mime_type);
  2723. return false;
  2724. });
  2725. }
  2726. }
  2727. // register API routes
  2728. svr->Get ("/health", handle_health); // public endpoint (no API key check)
  2729. svr->Get ("/metrics", handle_metrics);
  2730. svr->Get ("/props", handle_props);
  2731. svr->Post("/props", handle_props_change);
  2732. svr->Get ("/models", handle_models); // public endpoint (no API key check)
  2733. svr->Get ("/v1/models", handle_models); // public endpoint (no API key check)
  2734. svr->Post("/completion", handle_completions); // legacy
  2735. svr->Post("/completions", handle_completions);
  2736. svr->Post("/v1/completions", handle_completions);
  2737. svr->Post("/chat/completions", handle_chat_completions);
  2738. svr->Post("/v1/chat/completions", handle_chat_completions);
  2739. svr->Post("/infill", handle_infill);
  2740. svr->Post("/embedding", handle_embeddings); // legacy
  2741. svr->Post("/embeddings", handle_embeddings);
  2742. svr->Post("/v1/embeddings", handle_embeddings);
  2743. svr->Post("/rerank", handle_rerank);
  2744. svr->Post("/reranking", handle_rerank);
  2745. svr->Post("/v1/rerank", handle_rerank);
  2746. svr->Post("/v1/reranking", handle_rerank);
  2747. svr->Post("/tokenize", handle_tokenize);
  2748. svr->Post("/detokenize", handle_detokenize);
  2749. // LoRA adapters hotswap
  2750. svr->Get ("/lora-adapters", handle_lora_adapters_list);
  2751. svr->Post("/lora-adapters", handle_lora_adapters_apply);
  2752. // Save & load slots
  2753. svr->Get ("/slots", handle_slots);
  2754. svr->Post("/slots/:id_slot", handle_slots_action);
  2755. //
  2756. // Start the server
  2757. //
  2758. if (params.n_threads_http < 1) {
  2759. // +2 threads for monitoring endpoints
  2760. params.n_threads_http = std::max(params.n_parallel + 2, (int32_t) std::thread::hardware_concurrency() - 1);
  2761. }
  2762. log_data["n_threads_http"] = std::to_string(params.n_threads_http);
  2763. svr->new_task_queue = [&params] { return new httplib::ThreadPool(params.n_threads_http); };
  2764. // clean up function, to be called before exit
  2765. auto clean_up = [&svr]() {
  2766. svr->stop();
  2767. llama_backend_free();
  2768. };
  2769. // bind HTTP listen port
  2770. bool was_bound = false;
  2771. if (params.port == 0) {
  2772. int bound_port = svr->bind_to_any_port(params.hostname);
  2773. if ((was_bound = (bound_port >= 0))) {
  2774. params.port = bound_port;
  2775. }
  2776. } else {
  2777. was_bound = svr->bind_to_port(params.hostname, params.port);
  2778. }
  2779. if (!was_bound) {
  2780. //LOG_ERROR("couldn't bind HTTP server socket", {
  2781. // {"hostname", params.hostname},
  2782. // {"port", params.port},
  2783. //});
  2784. LOG_ERR("%s: couldn't bind HTTP server socket, hostname: %s, port: %d\n", __func__, params.hostname.c_str(), params.port);
  2785. clean_up();
  2786. return 1;
  2787. }
  2788. // run the HTTP server in a thread
  2789. std::thread t([&]() { svr->listen_after_bind(); });
  2790. svr->wait_until_ready();
  2791. LOG_INF("%s: HTTP server is listening, hostname: %s, port: %d, http threads: %d\n", __func__, params.hostname.c_str(), params.port, params.n_threads_http);
  2792. // load the model
  2793. LOG_INF("%s: loading model\n", __func__);
  2794. if (!ctx_server.load_model(params)) {
  2795. clean_up();
  2796. t.join();
  2797. LOG_ERR("%s: exiting due to model loading error\n", __func__);
  2798. return 1;
  2799. }
  2800. ctx_server.init();
  2801. state.store(SERVER_STATE_READY);
  2802. LOG_INF("%s: model loaded\n", __func__);
  2803. // if a custom chat template is not supplied, we will use the one that comes with the model (if any)
  2804. if (params.chat_template.empty()) {
  2805. if (!ctx_server.validate_model_chat_template()) {
  2806. LOG_WRN("%s: The chat template that comes with this model is not yet supported, falling back to chatml. This may cause the model to output suboptimal responses\n", __func__);
  2807. params.chat_template = "chatml";
  2808. }
  2809. }
  2810. // print sample chat example to make it clear which template is used
  2811. LOG_INF("%s: chat template, built_in: %d, chat_example: '%s'\n", __func__, params.chat_template.empty(), common_chat_format_example(ctx_server.model, params.chat_template).c_str());
  2812. ctx_server.queue_tasks.on_new_task(std::bind(
  2813. &server_context::process_single_task, &ctx_server, std::placeholders::_1));
  2814. ctx_server.queue_tasks.on_update_slots(std::bind(
  2815. &server_context::update_slots, &ctx_server));
  2816. shutdown_handler = [&](int) {
  2817. ctx_server.queue_tasks.terminate();
  2818. };
  2819. LOG_INF("%s: server is listening on http://%s:%d - starting the main loop\n", __func__, params.hostname.c_str(), params.port);
  2820. ctx_server.queue_tasks.start_loop();
  2821. #if defined (__unix__) || (defined (__APPLE__) && defined (__MACH__))
  2822. struct sigaction sigint_action;
  2823. sigint_action.sa_handler = signal_handler;
  2824. sigemptyset (&sigint_action.sa_mask);
  2825. sigint_action.sa_flags = 0;
  2826. sigaction(SIGINT, &sigint_action, NULL);
  2827. sigaction(SIGTERM, &sigint_action, NULL);
  2828. #elif defined (_WIN32)
  2829. auto console_ctrl_handler = +[](DWORD ctrl_type) -> BOOL {
  2830. return (ctrl_type == CTRL_C_EVENT) ? (signal_handler(SIGINT), true) : false;
  2831. };
  2832. SetConsoleCtrlHandler(reinterpret_cast<PHANDLER_ROUTINE>(console_ctrl_handler), true);
  2833. #endif
  2834. clean_up();
  2835. t.join();
  2836. return 0;
  2837. }