ggml-rpc.cpp 43 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155
  1. #include "ggml-rpc.h"
  2. #include "ggml.h"
  3. #include "ggml-backend-impl.h"
  4. #include <cinttypes>
  5. #include <string>
  6. #include <vector>
  7. #include <memory>
  8. #include <unordered_map>
  9. #include <unordered_set>
  10. #ifdef _WIN32
  11. # define WIN32_LEAN_AND_MEAN
  12. # ifndef NOMINMAX
  13. # define NOMINMAX
  14. # endif
  15. # include <windows.h>
  16. # include <winsock2.h>
  17. #else
  18. # include <arpa/inet.h>
  19. # include <sys/socket.h>
  20. # include <sys/types.h>
  21. # include <netinet/in.h>
  22. # include <netinet/tcp.h>
  23. # include <netdb.h>
  24. # include <unistd.h>
  25. #endif
  26. #include <string.h>
  27. #define UNUSED GGML_UNUSED
  28. #define GGML_DEBUG 0
  29. #if (GGML_DEBUG >= 1)
  30. #define GGML_PRINT_DEBUG(...) printf(__VA_ARGS__)
  31. #else
  32. #define GGML_PRINT_DEBUG(...)
  33. #endif
  34. #ifdef _WIN32
  35. typedef SOCKET sockfd_t;
  36. using ssize_t = __int64;
  37. #else
  38. typedef int sockfd_t;
  39. #endif
  40. // cross-platform socket
  41. struct socket_t {
  42. sockfd_t fd;
  43. socket_t(sockfd_t fd) : fd(fd) {}
  44. ~socket_t() {
  45. #ifdef _WIN32
  46. closesocket(this->fd);
  47. #else
  48. close(this->fd);
  49. #endif
  50. }
  51. };
  52. // ggml_tensor is serialized into rpc_tensor
  53. #pragma pack(push, 1)
  54. struct rpc_tensor {
  55. uint64_t id;
  56. uint32_t type;
  57. uint64_t buffer;
  58. uint32_t ne[GGML_MAX_DIMS];
  59. uint32_t nb[GGML_MAX_DIMS];
  60. uint32_t op;
  61. int32_t op_params[GGML_MAX_OP_PARAMS / sizeof(int32_t)];
  62. int32_t flags;
  63. uint64_t src[GGML_MAX_SRC];
  64. uint64_t view_src;
  65. uint64_t view_offs;
  66. uint64_t data;
  67. char name[GGML_MAX_NAME];
  68. };
  69. #pragma pack(pop)
  70. // RPC commands
  71. enum rpc_cmd {
  72. ALLOC_BUFFER = 0,
  73. GET_ALIGNMENT,
  74. GET_MAX_SIZE,
  75. BUFFER_GET_BASE,
  76. FREE_BUFFER,
  77. BUFFER_CLEAR,
  78. SET_TENSOR,
  79. GET_TENSOR,
  80. COPY_TENSOR,
  81. GRAPH_COMPUTE,
  82. GET_DEVICE_MEMORY,
  83. };
  84. // RPC data structures
  85. static ggml_guid_t ggml_backend_rpc_guid() {
  86. static ggml_guid guid = {0x99, 0x68, 0x5b, 0x6c, 0xd2, 0x83, 0x3d, 0x24, 0x25, 0x36, 0x72, 0xe1, 0x5b, 0x0e, 0x14, 0x03};
  87. return &guid;
  88. }
  89. struct ggml_backend_rpc_buffer_type_context {
  90. std::shared_ptr<socket_t> sock;
  91. std::string name;
  92. size_t alignment;
  93. size_t max_size;
  94. };
  95. struct ggml_backend_rpc_context {
  96. std::string endpoint;
  97. std::string name;
  98. std::shared_ptr<socket_t> sock;
  99. ggml_backend_buffer_type_t buft;
  100. };
  101. struct ggml_backend_rpc_buffer_context {
  102. std::shared_ptr<socket_t> sock;
  103. std::unordered_map<ggml_backend_buffer_t, void *> base_cache;
  104. uint64_t remote_ptr;
  105. std::string name;
  106. };
  107. // RPC helper functions
  108. static std::shared_ptr<socket_t> make_socket(sockfd_t fd) {
  109. #ifdef _WIN32
  110. if (fd == INVALID_SOCKET) {
  111. return nullptr;
  112. }
  113. #else
  114. if (fd < 0) {
  115. return nullptr;
  116. }
  117. #endif
  118. return std::make_shared<socket_t>(fd);
  119. }
  120. static bool set_no_delay(sockfd_t sockfd) {
  121. int flag = 1;
  122. // set TCP_NODELAY to disable Nagle's algorithm
  123. int ret = setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(int));
  124. return ret == 0;
  125. }
  126. static bool set_reuse_addr(sockfd_t sockfd) {
  127. int flag = 1;
  128. int ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char *)&flag, sizeof(int));
  129. return ret == 0;
  130. }
  131. static std::shared_ptr<socket_t> socket_connect(const char * host, int port) {
  132. struct sockaddr_in addr;
  133. auto sockfd = socket(AF_INET, SOCK_STREAM, 0);
  134. auto sock_ptr = make_socket(sockfd);
  135. if (sock_ptr == nullptr) {
  136. return nullptr;
  137. }
  138. if (!set_no_delay(sockfd)) {
  139. fprintf(stderr, "Failed to set TCP_NODELAY\n");
  140. return nullptr;
  141. }
  142. addr.sin_family = AF_INET;
  143. addr.sin_port = htons(port);
  144. struct hostent * server = gethostbyname(host);
  145. if (server == NULL) {
  146. fprintf(stderr, "Cannot resolve host '%s'\n", host);
  147. return nullptr;
  148. }
  149. memcpy(&addr.sin_addr.s_addr, server->h_addr, server->h_length);
  150. if (connect(sock_ptr->fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
  151. return nullptr;
  152. }
  153. return sock_ptr;
  154. }
  155. static std::shared_ptr<socket_t> socket_accept(sockfd_t srv_sockfd) {
  156. auto client_socket_fd = accept(srv_sockfd, NULL, NULL);
  157. auto client_socket = make_socket(client_socket_fd);
  158. if (client_socket == nullptr) {
  159. return nullptr;
  160. }
  161. if (!set_no_delay(client_socket_fd)) {
  162. fprintf(stderr, "Failed to set TCP_NODELAY\n");
  163. return nullptr;
  164. }
  165. return client_socket;
  166. }
  167. static std::shared_ptr<socket_t> create_server_socket(const char * host, int port) {
  168. auto sockfd = socket(AF_INET, SOCK_STREAM, 0);
  169. auto sock = make_socket(sockfd);
  170. if (sock == nullptr) {
  171. return nullptr;
  172. }
  173. if (!set_reuse_addr(sockfd)) {
  174. fprintf(stderr, "Failed to set SO_REUSEADDR\n");
  175. return nullptr;
  176. }
  177. struct sockaddr_in serv_addr;
  178. serv_addr.sin_family = AF_INET;
  179. serv_addr.sin_addr.s_addr = inet_addr(host);
  180. serv_addr.sin_port = htons(port);
  181. if (bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
  182. return nullptr;
  183. }
  184. if (listen(sockfd, 1) < 0) {
  185. return nullptr;
  186. }
  187. return sock;
  188. }
  189. static bool send_data(sockfd_t sockfd, const void * data, size_t size) {
  190. size_t bytes_sent = 0;
  191. while (bytes_sent < size) {
  192. ssize_t n = send(sockfd, (const char *)data + bytes_sent, size - bytes_sent, 0);
  193. if (n < 0) {
  194. return false;
  195. }
  196. bytes_sent += n;
  197. }
  198. return true;
  199. }
  200. static bool recv_data(sockfd_t sockfd, void * data, size_t size) {
  201. size_t bytes_recv = 0;
  202. while (bytes_recv < size) {
  203. ssize_t n = recv(sockfd, (char *)data + bytes_recv, size - bytes_recv, 0);
  204. if (n <= 0) {
  205. return false;
  206. }
  207. bytes_recv += n;
  208. }
  209. return true;
  210. }
  211. static bool parse_endpoint(const char * endpoint, std::string & host, int & port) {
  212. std::string str(endpoint);
  213. size_t pos = str.find(':');
  214. if (pos == std::string::npos) {
  215. return false;
  216. }
  217. host = str.substr(0, pos);
  218. port = std::stoi(str.substr(pos + 1));
  219. return true;
  220. }
  221. // RPC request : | rpc_cmd (1 byte) | request_size (8 bytes) | request_data (request_size bytes) |
  222. // RPC response: | response_size (8 bytes) | response_data (response_size bytes) |
  223. static bool send_rpc_cmd(const std::shared_ptr<socket_t> & sock, enum rpc_cmd cmd, const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  224. uint8_t cmd_byte = cmd;
  225. if (!send_data(sock->fd, &cmd_byte, sizeof(cmd_byte))) {
  226. return false;
  227. }
  228. uint64_t input_size = input.size();
  229. if (!send_data(sock->fd, &input_size, sizeof(input_size))) {
  230. return false;
  231. }
  232. if (!send_data(sock->fd, input.data(), input.size())) {
  233. return false;
  234. }
  235. uint64_t output_size;
  236. if (!recv_data(sock->fd, &output_size, sizeof(output_size))) {
  237. return false;
  238. }
  239. if (output_size == 0) {
  240. output.clear();
  241. return true;
  242. }
  243. output.resize(output_size);
  244. if (!recv_data(sock->fd, output.data(), output_size)) {
  245. return false;
  246. }
  247. return true;
  248. }
  249. // RPC client-side implementation
  250. GGML_CALL static const char * ggml_backend_rpc_buffer_get_name(ggml_backend_buffer_t buffer) {
  251. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  252. return ctx->name.c_str();
  253. }
  254. GGML_CALL static void ggml_backend_rpc_buffer_free_buffer(ggml_backend_buffer_t buffer) {
  255. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  256. // input serialization format: | remote_ptr (8 bytes) |
  257. std::vector<uint8_t> input(sizeof(uint64_t), 0);
  258. uint64_t remote_ptr = ctx->remote_ptr;
  259. memcpy(input.data(), &remote_ptr, sizeof(remote_ptr));
  260. std::vector<uint8_t> output;
  261. bool status = send_rpc_cmd(ctx->sock, FREE_BUFFER, input, output);
  262. GGML_ASSERT(status);
  263. GGML_ASSERT(output.empty());
  264. delete ctx;
  265. }
  266. GGML_CALL static void * ggml_backend_rpc_buffer_get_base(ggml_backend_buffer_t buffer) {
  267. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  268. if (ctx->base_cache.find(buffer) != ctx->base_cache.end()) {
  269. return ctx->base_cache[buffer];
  270. }
  271. // input serialization format: | remote_ptr (8 bytes) |
  272. std::vector<uint8_t> input(sizeof(uint64_t), 0);
  273. uint64_t remote_ptr = ctx->remote_ptr;
  274. memcpy(input.data(), &remote_ptr, sizeof(remote_ptr));
  275. std::vector<uint8_t> output;
  276. bool status = send_rpc_cmd(ctx->sock, BUFFER_GET_BASE, input, output);
  277. GGML_ASSERT(status);
  278. GGML_ASSERT(output.size() == sizeof(uint64_t));
  279. // output serialization format: | base_ptr (8 bytes) |
  280. uint64_t base_ptr;
  281. memcpy(&base_ptr, output.data(), sizeof(base_ptr));
  282. void * base = reinterpret_cast<void *>(base_ptr);
  283. ctx->base_cache[buffer] = base;
  284. return base;
  285. }
  286. static rpc_tensor serialize_tensor(const ggml_tensor * tensor) {
  287. rpc_tensor result;
  288. result.id = reinterpret_cast<uint64_t>(tensor);
  289. result.type = tensor->type;
  290. if (tensor->buffer) {
  291. ggml_backend_buffer_t buffer = tensor->buffer;
  292. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  293. result.buffer = ctx->remote_ptr;
  294. } else {
  295. result.buffer = 0;
  296. }
  297. for (uint32_t i = 0; i < GGML_MAX_DIMS; i++) {
  298. result.ne[i] = tensor->ne[i];
  299. result.nb[i] = tensor->nb[i];
  300. }
  301. result.op = tensor->op;
  302. for (uint32_t i = 0; i < GGML_MAX_OP_PARAMS / sizeof(int32_t); i++) {
  303. result.op_params[i] = tensor->op_params[i];
  304. }
  305. result.flags = tensor->flags;
  306. for (uint32_t i = 0; i < GGML_MAX_SRC; i++) {
  307. result.src[i] = reinterpret_cast<uint64_t>(tensor->src[i]);
  308. }
  309. result.view_src = reinterpret_cast<uint64_t>(tensor->view_src);
  310. result.view_offs = tensor->view_offs;
  311. result.data = reinterpret_cast<uint64_t>(tensor->data);
  312. snprintf(result.name, GGML_MAX_NAME, "%s", tensor->name);
  313. return result;
  314. }
  315. GGML_CALL static void ggml_backend_rpc_buffer_init_tensor(ggml_backend_buffer_t buffer, ggml_tensor * tensor) {
  316. UNUSED(buffer);
  317. if (ggml_is_quantized(tensor->type)) {
  318. // TODO: this check is due to MATRIX_ROW_PADDING in CUDA and should be generalized
  319. GGML_ASSERT(tensor->ne[0] % 512 == 0 && "unsupported quantized tensor");
  320. }
  321. }
  322. GGML_CALL static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggml_tensor * tensor, const void * data, size_t offset, size_t size) {
  323. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  324. // input serialization format: | rpc_tensor | offset (8 bytes) | data (size bytes) |
  325. size_t input_size = sizeof(rpc_tensor) + sizeof(uint64_t) + size;
  326. std::vector<uint8_t> input(input_size, 0);
  327. rpc_tensor rpc_tensor = serialize_tensor(tensor);
  328. memcpy(input.data(), &rpc_tensor, sizeof(rpc_tensor));
  329. memcpy(input.data() + sizeof(rpc_tensor), &offset, sizeof(offset));
  330. memcpy(input.data() + sizeof(rpc_tensor) + sizeof(offset), data, size);
  331. std::vector<uint8_t> output;
  332. bool status = send_rpc_cmd(ctx->sock, SET_TENSOR, input, output);
  333. GGML_ASSERT(status);
  334. }
  335. GGML_CALL static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * tensor, void * data, size_t offset, size_t size) {
  336. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  337. // input serialization format: | rpc_tensor | offset (8 bytes) | size (8 bytes) |
  338. int input_size = sizeof(rpc_tensor) + 2*sizeof(uint64_t);
  339. std::vector<uint8_t> input(input_size, 0);
  340. rpc_tensor rpc_tensor = serialize_tensor(tensor);
  341. memcpy(input.data(), &rpc_tensor, sizeof(rpc_tensor));
  342. memcpy(input.data() + sizeof(rpc_tensor), &offset, sizeof(offset));
  343. memcpy(input.data() + sizeof(rpc_tensor) + sizeof(offset), &size, sizeof(size));
  344. std::vector<uint8_t> output;
  345. bool status = send_rpc_cmd(ctx->sock, GET_TENSOR, input, output);
  346. GGML_ASSERT(status);
  347. GGML_ASSERT(output.size() == size);
  348. // output serialization format: | data (size bytes) |
  349. memcpy(data, output.data(), size);
  350. }
  351. GGML_CALL static bool ggml_backend_rpc_buffer_cpy_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * src, ggml_tensor * dst) {
  352. // check if src and dst are on the same server
  353. ggml_backend_buffer_t src_buffer = src->buffer;
  354. ggml_backend_rpc_buffer_context * src_ctx = (ggml_backend_rpc_buffer_context *)src_buffer->context;
  355. ggml_backend_buffer_t dst_buffer = dst->buffer;
  356. ggml_backend_rpc_buffer_context * dst_ctx = (ggml_backend_rpc_buffer_context *)dst_buffer->context;
  357. if (src_ctx->sock != dst_ctx->sock) {
  358. return false;
  359. }
  360. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  361. // input serialization format: | rpc_tensor src | rpc_tensor dst |
  362. int input_size = 2*sizeof(rpc_tensor);
  363. std::vector<uint8_t> input(input_size, 0);
  364. rpc_tensor rpc_src = serialize_tensor(src);
  365. rpc_tensor rpc_dst = serialize_tensor(dst);
  366. memcpy(input.data(), &rpc_src, sizeof(rpc_src));
  367. memcpy(input.data() + sizeof(rpc_src), &rpc_dst, sizeof(rpc_dst));
  368. std::vector<uint8_t> output;
  369. bool status = send_rpc_cmd(ctx->sock, COPY_TENSOR, input, output);
  370. GGML_ASSERT(status);
  371. // output serialization format: | result (1 byte) |
  372. GGML_ASSERT(output.size() == 1);
  373. return output[0];
  374. }
  375. GGML_CALL static void ggml_backend_rpc_buffer_clear(ggml_backend_buffer_t buffer, uint8_t value) {
  376. ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
  377. // serialization format: | bufptr (8 bytes) | value (1 byte) |
  378. int input_size = sizeof(uint64_t) + sizeof(uint8_t);
  379. std::vector<uint8_t> input(input_size, 0);
  380. memcpy(input.data(), &ctx->remote_ptr, sizeof(ctx->remote_ptr));
  381. memcpy(input.data() + sizeof(ctx->remote_ptr), &value, sizeof(value));
  382. std::vector<uint8_t> output;
  383. bool status = send_rpc_cmd(ctx->sock, BUFFER_CLEAR, input, output);
  384. GGML_ASSERT(status);
  385. }
  386. static ggml_backend_buffer_i ggml_backend_rpc_buffer_interface = {
  387. /* .get_name = */ ggml_backend_rpc_buffer_get_name,
  388. /* .free_buffer = */ ggml_backend_rpc_buffer_free_buffer,
  389. /* .get_base = */ ggml_backend_rpc_buffer_get_base,
  390. /* .init_tensor = */ ggml_backend_rpc_buffer_init_tensor,
  391. /* .set_tensor = */ ggml_backend_rpc_buffer_set_tensor,
  392. /* .get_tensor = */ ggml_backend_rpc_buffer_get_tensor,
  393. /* .cpy_tensor = */ ggml_backend_rpc_buffer_cpy_tensor,
  394. /* .clear = */ ggml_backend_rpc_buffer_clear,
  395. /* .reset = */ NULL,
  396. };
  397. GGML_CALL static const char * ggml_backend_rpc_buffer_type_name(ggml_backend_buffer_type_t buft) {
  398. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)buft->context;
  399. return buft_ctx->name.c_str();
  400. }
  401. GGML_CALL static ggml_backend_buffer_t ggml_backend_rpc_buffer_type_alloc_buffer(ggml_backend_buffer_type_t buft, size_t size) {
  402. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)buft->context;
  403. // input serialization format: | size (8 bytes) |
  404. int input_size = sizeof(uint64_t);
  405. std::vector<uint8_t> input(input_size, 0);
  406. memcpy(input.data(), &size, sizeof(size));
  407. std::vector<uint8_t> output;
  408. bool status = send_rpc_cmd(buft_ctx->sock, ALLOC_BUFFER, input, output);
  409. GGML_ASSERT(status);
  410. GGML_ASSERT(output.size() == 2*sizeof(uint64_t));
  411. // output serialization format: | remote_ptr (8 bytes) | remote_size (8 bytes) |
  412. uint64_t remote_ptr;
  413. memcpy(&remote_ptr, output.data(), sizeof(remote_ptr));
  414. size_t remote_size;
  415. memcpy(&remote_size, output.data() + sizeof(uint64_t), sizeof(remote_size));
  416. if (remote_ptr != 0) {
  417. ggml_backend_buffer_t buffer = ggml_backend_buffer_init(buft,
  418. ggml_backend_rpc_buffer_interface,
  419. new ggml_backend_rpc_buffer_context{buft_ctx->sock, {}, remote_ptr, "RPC"},
  420. remote_size);
  421. return buffer;
  422. } else {
  423. return nullptr;
  424. }
  425. }
  426. static size_t get_alignment(const std::shared_ptr<socket_t> & sock) {
  427. // input serialization format: | 0 bytes |
  428. std::vector<uint8_t> input;
  429. std::vector<uint8_t> output;
  430. bool status = send_rpc_cmd(sock, GET_ALIGNMENT, input, output);
  431. GGML_ASSERT(status);
  432. GGML_ASSERT(output.size() == sizeof(uint64_t));
  433. // output serialization format: | alignment (8 bytes) |
  434. uint64_t alignment;
  435. memcpy(&alignment, output.data(), sizeof(alignment));
  436. return alignment;
  437. }
  438. GGML_CALL static size_t ggml_backend_rpc_buffer_type_get_alignment(ggml_backend_buffer_type_t buft) {
  439. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)buft->context;
  440. return buft_ctx->alignment;
  441. }
  442. static size_t get_max_size(const std::shared_ptr<socket_t> & sock) {
  443. // input serialization format: | 0 bytes |
  444. std::vector<uint8_t> input;
  445. std::vector<uint8_t> output;
  446. bool status = send_rpc_cmd(sock, GET_MAX_SIZE, input, output);
  447. GGML_ASSERT(status);
  448. GGML_ASSERT(output.size() == sizeof(uint64_t));
  449. // output serialization format: | max_size (8 bytes) |
  450. uint64_t max_size;
  451. memcpy(&max_size, output.data(), sizeof(max_size));
  452. return max_size;
  453. }
  454. GGML_CALL static size_t ggml_backend_rpc_get_max_size(ggml_backend_buffer_type_t buft) {
  455. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)buft->context;
  456. return buft_ctx->max_size;
  457. }
  458. GGML_CALL static size_t ggml_backend_rpc_buffer_type_get_alloc_size(ggml_backend_buffer_type_t buft, const ggml_tensor * tensor) {
  459. UNUSED(buft);
  460. return ggml_nbytes(tensor);
  461. }
  462. GGML_CALL static bool ggml_backend_rpc_buffer_type_supports_backend(ggml_backend_buffer_type_t buft, ggml_backend_t backend) {
  463. if (!ggml_backend_is_rpc(backend)) {
  464. return false;
  465. }
  466. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)buft->context;
  467. ggml_backend_rpc_context * rpc_ctx = (ggml_backend_rpc_context *)backend->context;
  468. return buft_ctx->sock == rpc_ctx->sock;
  469. }
  470. static ggml_backend_buffer_type_i ggml_backend_rpc_buffer_type_interface = {
  471. /* .get_name = */ ggml_backend_rpc_buffer_type_name,
  472. /* .alloc_buffer = */ ggml_backend_rpc_buffer_type_alloc_buffer,
  473. /* .get_alignment = */ ggml_backend_rpc_buffer_type_get_alignment,
  474. /* .get_max_size = */ ggml_backend_rpc_get_max_size,
  475. /* .get_alloc_size = */ ggml_backend_rpc_buffer_type_get_alloc_size,
  476. /* .supports_backend = */ ggml_backend_rpc_buffer_type_supports_backend,
  477. /* .is_host = */ NULL,
  478. };
  479. GGML_CALL static const char * ggml_backend_rpc_name(ggml_backend_t backend) {
  480. ggml_backend_rpc_context * rpc_ctx = (ggml_backend_rpc_context *)backend->context;
  481. return rpc_ctx->name.c_str();
  482. }
  483. GGML_CALL static void ggml_backend_rpc_free(ggml_backend_t backend) {
  484. ggml_backend_rpc_context * rpc_ctx = (ggml_backend_rpc_context *)backend->context;
  485. ggml_backend_rpc_buffer_type_context * buft_ctx = (ggml_backend_rpc_buffer_type_context *)rpc_ctx->buft->context;
  486. delete buft_ctx;
  487. delete rpc_ctx->buft;
  488. delete rpc_ctx;
  489. delete backend;
  490. }
  491. GGML_CALL static ggml_backend_buffer_type_t ggml_backend_rpc_get_default_buffer_type(ggml_backend_t backend) {
  492. ggml_backend_rpc_context * ctx = (ggml_backend_rpc_context *)backend->context;
  493. return ctx->buft;
  494. }
  495. GGML_CALL static void ggml_backend_rpc_synchronize(ggml_backend_t backend) {
  496. UNUSED(backend);
  497. // this is no-op because we don't have any async operations
  498. }
  499. static void add_tensor(ggml_tensor * tensor, std::vector<rpc_tensor> & tensors, std::unordered_set<ggml_tensor*> & visited) {
  500. if (tensor == nullptr) {
  501. return;
  502. }
  503. if (visited.find(tensor) != visited.end()) {
  504. return;
  505. }
  506. visited.insert(tensor);
  507. for (int i = 0; i < GGML_MAX_SRC; i++) {
  508. add_tensor(tensor->src[i], tensors, visited);
  509. }
  510. add_tensor(tensor->view_src, tensors, visited);
  511. tensors.push_back(serialize_tensor(tensor));
  512. }
  513. static void serialize_graph(const ggml_cgraph * cgraph, std::vector<uint8_t> & output) {
  514. uint32_t n_nodes = cgraph->n_nodes;
  515. std::vector<rpc_tensor> tensors;
  516. std::unordered_set<ggml_tensor*> visited;
  517. for (uint32_t i = 0; i < n_nodes; i++) {
  518. add_tensor(cgraph->nodes[i], tensors, visited);
  519. }
  520. // serialization format:
  521. // | n_nodes (4 bytes) | nodes (n_nodes * sizeof(uint64_t) | n_tensors (4 bytes) | tensors (n_tensors * sizeof(rpc_tensor)) |
  522. uint32_t n_tensors = tensors.size();
  523. int output_size = sizeof(uint32_t) + n_nodes * sizeof(uint64_t) + sizeof(uint32_t) + n_tensors * sizeof(rpc_tensor);
  524. output.resize(output_size, 0);
  525. memcpy(output.data(), &n_nodes, sizeof(n_nodes));
  526. uint64_t * out_nodes = (uint64_t *)(output.data() + sizeof(n_nodes));
  527. for (uint32_t i = 0; i < n_nodes; i++) {
  528. out_nodes[i] = reinterpret_cast<uint64_t>(cgraph->nodes[i]);
  529. }
  530. uint32_t * out_ntensors = (uint32_t *)(output.data() + sizeof(n_nodes) + n_nodes * sizeof(uint64_t));
  531. *out_ntensors = n_tensors;
  532. rpc_tensor * out_tensors = (rpc_tensor *)(output.data() + sizeof(n_nodes) + n_nodes * sizeof(uint64_t) + sizeof(uint32_t));
  533. memcpy(out_tensors, tensors.data(), n_tensors * sizeof(rpc_tensor));
  534. }
  535. GGML_CALL static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, ggml_cgraph * cgraph) {
  536. ggml_backend_rpc_context * rpc_ctx = (ggml_backend_rpc_context *)backend->context;
  537. std::vector<uint8_t> input;
  538. serialize_graph(cgraph, input);
  539. std::vector<uint8_t> output;
  540. bool status = send_rpc_cmd(rpc_ctx->sock, GRAPH_COMPUTE, input, output);
  541. GGML_ASSERT(status);
  542. GGML_ASSERT(output.size() == 1);
  543. return (enum ggml_status)output[0];
  544. }
  545. GGML_CALL static bool ggml_backend_rpc_supports_op(ggml_backend_t backend, const ggml_tensor * op) {
  546. UNUSED(backend);
  547. UNUSED(op);
  548. GGML_ASSERT(false && "not implemented");
  549. return false;
  550. }
  551. static ggml_backend_i ggml_backend_rpc_interface = {
  552. /* .get_name = */ ggml_backend_rpc_name,
  553. /* .free = */ ggml_backend_rpc_free,
  554. /* .get_default_buffer_type = */ ggml_backend_rpc_get_default_buffer_type,
  555. /* .set_tensor_async = */ NULL,
  556. /* .get_tensor_async = */ NULL,
  557. /* .cpy_tensor_async = */ NULL,
  558. /* .synchronize = */ ggml_backend_rpc_synchronize,
  559. /* .graph_plan_create = */ NULL,
  560. /* .graph_plan_free = */ NULL,
  561. /* .graph_plan_compute = */ NULL,
  562. /* .graph_compute = */ ggml_backend_rpc_graph_compute,
  563. /* .supports_op = */ ggml_backend_rpc_supports_op,
  564. /* .offload_op = */ NULL,
  565. /* .event_new = */ NULL,
  566. /* .event_free = */ NULL,
  567. /* .event_record = */ NULL,
  568. /* .event_wait = */ NULL,
  569. /* .event_synchronize = */ NULL,
  570. };
  571. static std::unordered_map<std::string, ggml_backend_t> instances;
  572. GGML_API GGML_CALL ggml_backend_buffer_type_t ggml_backend_rpc_buffer_type(const char * endpoint) {
  573. ggml_backend_t backend = ggml_backend_rpc_init(endpoint);
  574. return backend != nullptr ? ggml_backend_rpc_get_default_buffer_type(backend) : nullptr;
  575. }
  576. GGML_CALL ggml_backend_t ggml_backend_rpc_init(const char * endpoint) {
  577. std::string endpoint_str(endpoint);
  578. if (instances.find(endpoint_str) != instances.end()) {
  579. return instances[endpoint_str];
  580. }
  581. #ifdef _WIN32
  582. {
  583. WSADATA wsaData;
  584. int res = WSAStartup(MAKEWORD(2, 2), &wsaData);
  585. if (res != 0) {
  586. return nullptr;
  587. }
  588. }
  589. #endif
  590. fprintf(stderr, "Connecting to %s\n", endpoint);
  591. std::string host;
  592. int port;
  593. if (!parse_endpoint(endpoint, host, port)) {
  594. return nullptr;
  595. }
  596. auto sock = socket_connect(host.c_str(), port);
  597. if (sock == nullptr) {
  598. return nullptr;
  599. }
  600. size_t alignment = get_alignment(sock);
  601. size_t max_size = get_max_size(sock);
  602. ggml_backend_rpc_buffer_type_context * buft_ctx = new ggml_backend_rpc_buffer_type_context {
  603. /* .sock = */ sock,
  604. /* .name = */ "RPC" + std::to_string(sock->fd),
  605. /* .alignment = */ alignment,
  606. /* .max_size = */ max_size
  607. };
  608. ggml_backend_buffer_type_t buft = new ggml_backend_buffer_type {
  609. /* .iface = */ ggml_backend_rpc_buffer_type_interface,
  610. /* .context = */ buft_ctx
  611. };
  612. ggml_backend_rpc_context * ctx = new ggml_backend_rpc_context {
  613. /* .endpoint = */ endpoint,
  614. /* .name = */ "RPC" + std::to_string(sock->fd),
  615. /* .sock = */ sock,
  616. /* .buft = */ buft
  617. };
  618. instances[endpoint] = new ggml_backend {
  619. /* .guid = */ ggml_backend_rpc_guid(),
  620. /* .interface = */ ggml_backend_rpc_interface,
  621. /* .context = */ ctx
  622. };
  623. return instances[endpoint];
  624. }
  625. GGML_API GGML_CALL bool ggml_backend_is_rpc(ggml_backend_t backend) {
  626. return backend != NULL && ggml_guid_matches(backend->guid, ggml_backend_rpc_guid());
  627. }
  628. static void get_device_memory(const std::shared_ptr<socket_t> & sock, size_t * free, size_t * total) {
  629. // input serialization format: | 0 bytes |
  630. std::vector<uint8_t> input;
  631. std::vector<uint8_t> output;
  632. bool status = send_rpc_cmd(sock, GET_DEVICE_MEMORY, input, output);
  633. GGML_ASSERT(status);
  634. GGML_ASSERT(output.size() == 2*sizeof(uint64_t));
  635. // output serialization format: | free (8 bytes) | total (8 bytes) |
  636. uint64_t free_mem;
  637. memcpy(&free_mem, output.data(), sizeof(free_mem));
  638. uint64_t total_mem;
  639. memcpy(&total_mem, output.data() + sizeof(uint64_t), sizeof(total_mem));
  640. *free = free_mem;
  641. *total = total_mem;
  642. }
  643. GGML_API GGML_CALL void ggml_backend_rpc_get_device_memory(const char * endpoint, size_t * free, size_t * total) {
  644. ggml_backend_t backend = ggml_backend_rpc_init(endpoint);
  645. if (backend == nullptr) {
  646. *free = 0;
  647. *total = 0;
  648. return;
  649. }
  650. ggml_backend_rpc_context * ctx = (ggml_backend_rpc_context *)backend->context;
  651. get_device_memory(ctx->sock, free, total);
  652. }
  653. // RPC server-side implementation
  654. class rpc_server {
  655. public:
  656. rpc_server(ggml_backend_t backend) : backend(backend) {}
  657. ~rpc_server();
  658. bool alloc_buffer(const std::vector<uint8_t> & input, std::vector<uint8_t> & output);
  659. void get_alignment(std::vector<uint8_t> & output);
  660. void get_max_size(std::vector<uint8_t> & output);
  661. bool buffer_get_base(const std::vector<uint8_t> & input, std::vector<uint8_t> & output);
  662. bool free_buffer(const std::vector<uint8_t> & input);
  663. bool buffer_clear(const std::vector<uint8_t> & input);
  664. bool set_tensor(const std::vector<uint8_t> & input);
  665. bool get_tensor(const std::vector<uint8_t> & input, std::vector<uint8_t> & output);
  666. bool copy_tensor(const std::vector<uint8_t> & input, std::vector<uint8_t> & output);
  667. bool graph_compute(const std::vector<uint8_t> & input, std::vector<uint8_t> & output);
  668. private:
  669. ggml_tensor * deserialize_tensor(struct ggml_context * ctx, const rpc_tensor * tensor);
  670. ggml_tensor * create_node(uint64_t id,
  671. struct ggml_context * ctx,
  672. const std::unordered_map<uint64_t, const rpc_tensor*> & tensor_ptrs,
  673. std::unordered_map<uint64_t, struct ggml_tensor*> & tensor_map);
  674. ggml_backend_t backend;
  675. std::unordered_set<ggml_backend_buffer_t> buffers;
  676. };
  677. bool rpc_server::alloc_buffer(const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  678. // input serialization format: | size (8 bytes) |
  679. if (input.size() != sizeof(uint64_t)) {
  680. return false;
  681. }
  682. uint64_t size;
  683. memcpy(&size, input.data(), sizeof(size));
  684. ggml_backend_buffer_type_t buft = ggml_backend_get_default_buffer_type(backend);
  685. ggml_backend_buffer_t buffer = ggml_backend_buft_alloc_buffer(buft, size);
  686. uint64_t remote_ptr = 0;
  687. uint64_t remote_size = 0;
  688. if (buffer != nullptr) {
  689. remote_ptr = reinterpret_cast<uint64_t>(buffer);
  690. remote_size = buffer->size;
  691. GGML_PRINT_DEBUG("[%s] size: %" PRIu64 " -> remote_ptr: %" PRIx64 ", remote_size: %" PRIu64 "\n", __func__, size, remote_ptr, remote_size);
  692. buffers.insert(buffer);
  693. } else {
  694. GGML_PRINT_DEBUG("[%s] size: %" PRIu64 " -> failed\n", __func__, size);
  695. }
  696. // output serialization format: | remote_ptr (8 bytes) | remote_size (8 bytes) |
  697. output.resize(2*sizeof(uint64_t), 0);
  698. memcpy(output.data(), &remote_ptr, sizeof(remote_ptr));
  699. memcpy(output.data() + sizeof(uint64_t), &remote_size, sizeof(remote_size));
  700. return true;
  701. }
  702. void rpc_server::get_alignment(std::vector<uint8_t> & output) {
  703. ggml_backend_buffer_type_t buft = ggml_backend_get_default_buffer_type(backend);
  704. size_t alignment = ggml_backend_buft_get_alignment(buft);
  705. GGML_PRINT_DEBUG("[%s] alignment: %lu\n", __func__, alignment);
  706. // output serialization format: | alignment (8 bytes) |
  707. output.resize(sizeof(uint64_t), 0);
  708. memcpy(output.data(), &alignment, sizeof(alignment));
  709. }
  710. void rpc_server::get_max_size(std::vector<uint8_t> & output) {
  711. ggml_backend_buffer_type_t buft = ggml_backend_get_default_buffer_type(backend);
  712. size_t max_size = ggml_backend_buft_get_max_size(buft);
  713. GGML_PRINT_DEBUG("[%s] max_size: %lu\n", __func__, max_size);
  714. // output serialization format: | max_size (8 bytes) |
  715. output.resize(sizeof(uint64_t), 0);
  716. memcpy(output.data(), &max_size, sizeof(max_size));
  717. }
  718. bool rpc_server::buffer_get_base(const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  719. // input serialization format: | remote_ptr (8 bytes) |
  720. if (input.size() != sizeof(uint64_t)) {
  721. return false;
  722. }
  723. uint64_t remote_ptr;
  724. memcpy(&remote_ptr, input.data(), sizeof(remote_ptr));
  725. GGML_PRINT_DEBUG("[%s] remote_ptr: %" PRIx64 "\n", __func__, remote_ptr);
  726. ggml_backend_buffer_t buffer = reinterpret_cast<ggml_backend_buffer_t>(remote_ptr);
  727. if (buffers.find(buffer) == buffers.end()) {
  728. GGML_PRINT_DEBUG("[%s] buffer not found\n", __func__);
  729. return false;
  730. }
  731. void * base = ggml_backend_buffer_get_base(buffer);
  732. // output serialization format: | base_ptr (8 bytes) |
  733. uint64_t base_ptr = reinterpret_cast<uint64_t>(base);
  734. output.resize(sizeof(uint64_t), 0);
  735. memcpy(output.data(), &base_ptr, sizeof(base_ptr));
  736. return true;
  737. }
  738. bool rpc_server::free_buffer(const std::vector<uint8_t> & input) {
  739. // input serialization format: | remote_ptr (8 bytes) |
  740. if (input.size() != sizeof(uint64_t)) {
  741. return false;
  742. }
  743. uint64_t remote_ptr;
  744. memcpy(&remote_ptr, input.data(), sizeof(remote_ptr));
  745. GGML_PRINT_DEBUG("[%s] remote_ptr: %" PRIx64 "\n", __func__, remote_ptr);
  746. ggml_backend_buffer_t buffer = reinterpret_cast<ggml_backend_buffer_t>(remote_ptr);
  747. if (buffers.find(buffer) == buffers.end()) {
  748. GGML_PRINT_DEBUG("[%s] buffer not found\n", __func__);
  749. return false;
  750. }
  751. ggml_backend_buffer_free(buffer);
  752. buffers.erase(buffer);
  753. return true;
  754. }
  755. bool rpc_server::buffer_clear(const std::vector<uint8_t> & input) {
  756. // input serialization format: | remote_ptr (8 bytes) | value (1 byte) |
  757. if (input.size() != sizeof(uint64_t) + sizeof(uint8_t)) {
  758. return false;
  759. }
  760. uint64_t remote_ptr;
  761. memcpy(&remote_ptr, input.data(), sizeof(remote_ptr));
  762. uint8_t value;
  763. memcpy(&value, input.data() + sizeof(uint64_t), sizeof(value));
  764. GGML_PRINT_DEBUG("[%s] remote_ptr: %" PRIx64 ", value: %u\n", __func__, remote_ptr, value);
  765. ggml_backend_buffer_t buffer = reinterpret_cast<ggml_backend_buffer_t>(remote_ptr);
  766. if (buffers.find(buffer) == buffers.end()) {
  767. GGML_PRINT_DEBUG("[%s] buffer not found\n", __func__);
  768. return false;
  769. }
  770. ggml_backend_buffer_clear(buffer, value);
  771. return true;
  772. }
  773. ggml_tensor * rpc_server::deserialize_tensor(struct ggml_context * ctx, const rpc_tensor * tensor) {
  774. ggml_tensor * result = ggml_new_tensor_4d(ctx, (ggml_type) tensor->type,
  775. tensor->ne[0], tensor->ne[1], tensor->ne[2], tensor->ne[3]);
  776. for (uint32_t i = 0; i < GGML_MAX_DIMS; i++) {
  777. result->nb[i] = tensor->nb[i];
  778. }
  779. result->buffer = reinterpret_cast<ggml_backend_buffer_t>(tensor->buffer);
  780. if (result->buffer && buffers.find(result->buffer) == buffers.end()) {
  781. return nullptr;
  782. }
  783. result->op = (ggml_op) tensor->op;
  784. for (uint32_t i = 0; i < GGML_MAX_OP_PARAMS / sizeof(int32_t); i++) {
  785. result->op_params[i] = tensor->op_params[i];
  786. }
  787. result->flags = tensor->flags;
  788. result->data = reinterpret_cast<void *>(tensor->data);
  789. ggml_set_name(result, tensor->name);
  790. return result;
  791. }
  792. bool rpc_server::set_tensor(const std::vector<uint8_t> & input) {
  793. // serialization format: | rpc_tensor | offset (8 bytes) | data (size bytes) |
  794. if (input.size() < sizeof(rpc_tensor) + sizeof(uint64_t)) {
  795. return false;
  796. }
  797. const rpc_tensor * in_tensor = (const rpc_tensor *)input.data();
  798. uint64_t offset;
  799. memcpy(&offset, input.data() + sizeof(rpc_tensor), sizeof(offset));
  800. size_t size = input.size() - sizeof(rpc_tensor) - sizeof(offset);
  801. struct ggml_init_params params {
  802. /*.mem_size =*/ ggml_tensor_overhead(),
  803. /*.mem_buffer =*/ NULL,
  804. /*.no_alloc =*/ true,
  805. };
  806. struct ggml_context * ctx = ggml_init(params);
  807. ggml_tensor * tensor = deserialize_tensor(ctx, in_tensor);
  808. if (tensor == nullptr) {
  809. GGML_PRINT_DEBUG("[%s] error deserializing tensor\n", __func__);
  810. ggml_free(ctx);
  811. return false;
  812. }
  813. GGML_PRINT_DEBUG("[%s] buffer: %p, data: %p, offset: %" PRIu64 ", size: %zu\n", __func__, (void*)tensor->buffer, tensor->data, offset, size);
  814. const void * data = input.data() + sizeof(rpc_tensor) + sizeof(offset);
  815. ggml_backend_tensor_set(tensor, data, offset, size);
  816. ggml_free(ctx);
  817. return true;
  818. }
  819. bool rpc_server::get_tensor(const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  820. // serialization format: | rpc_tensor | offset (8 bytes) | size (8 bytes) |
  821. if (input.size() != sizeof(rpc_tensor) + 2*sizeof(uint64_t)) {
  822. return false;
  823. }
  824. const rpc_tensor * in_tensor = (const rpc_tensor *)input.data();
  825. uint64_t offset;
  826. memcpy(&offset, input.data() + sizeof(rpc_tensor), sizeof(offset));
  827. uint64_t size;
  828. memcpy(&size, input.data() + sizeof(rpc_tensor) + sizeof(offset), sizeof(size));
  829. struct ggml_init_params params {
  830. /*.mem_size =*/ ggml_tensor_overhead(),
  831. /*.mem_buffer =*/ NULL,
  832. /*.no_alloc =*/ true,
  833. };
  834. struct ggml_context * ctx = ggml_init(params);
  835. ggml_tensor * tensor = deserialize_tensor(ctx, in_tensor);
  836. if (tensor == nullptr) {
  837. GGML_PRINT_DEBUG("[%s] error deserializing tensor\n", __func__);
  838. ggml_free(ctx);
  839. return false;
  840. }
  841. GGML_PRINT_DEBUG("[%s] buffer: %p, data: %p, offset: %" PRIu64 ", size: %" PRIu64 "\n", __func__, (void*)tensor->buffer, tensor->data, offset, size);
  842. // output serialization format: | data (size bytes) |
  843. output.resize(size, 0);
  844. ggml_backend_tensor_get(tensor, output.data(), offset, size);
  845. ggml_free(ctx);
  846. return true;
  847. }
  848. bool rpc_server::copy_tensor(const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  849. // serialization format: | rpc_tensor src | rpc_tensor dst |
  850. if (input.size() != 2*sizeof(rpc_tensor)) {
  851. return false;
  852. }
  853. const rpc_tensor * rpc_src = (const rpc_tensor *)input.data();
  854. const rpc_tensor * rpc_dst = (const rpc_tensor *)(input.data() + sizeof(rpc_src));
  855. struct ggml_init_params params {
  856. /*.mem_size =*/ 2*ggml_tensor_overhead(),
  857. /*.mem_buffer =*/ NULL,
  858. /*.no_alloc =*/ true,
  859. };
  860. struct ggml_context * ctx = ggml_init(params);
  861. ggml_tensor * src = deserialize_tensor(ctx, rpc_src);
  862. ggml_tensor * dst = deserialize_tensor(ctx, rpc_dst);
  863. if (src == nullptr || dst == nullptr) {
  864. GGML_PRINT_DEBUG("[%s] error deserializing tensors\n", __func__);
  865. ggml_free(ctx);
  866. return false;
  867. }
  868. GGML_PRINT_DEBUG("[%s] src->buffer: %p, dst->buffer: %p\n", __func__, (void*)src->buffer, (void*)dst->buffer);
  869. bool result = ggml_backend_buffer_copy_tensor(src, dst);
  870. // output serialization format: | result (1 byte) |
  871. output.resize(1, 0);
  872. output[0] = result;
  873. ggml_free(ctx);
  874. return true;
  875. }
  876. ggml_tensor * rpc_server::create_node(uint64_t id,
  877. struct ggml_context * ctx,
  878. const std::unordered_map<uint64_t, const rpc_tensor*> & tensor_ptrs,
  879. std::unordered_map<uint64_t, struct ggml_tensor*> & tensor_map) {
  880. if (id == 0) {
  881. return nullptr;
  882. }
  883. if (tensor_map.find(id) != tensor_map.end()) {
  884. return tensor_map[id];
  885. }
  886. const rpc_tensor * tensor = tensor_ptrs.at(id);
  887. struct ggml_tensor * result = deserialize_tensor(ctx, tensor);
  888. if (result == nullptr) {
  889. return nullptr;
  890. }
  891. tensor_map[id] = result;
  892. for (int i = 0; i < GGML_MAX_SRC; i++) {
  893. result->src[i] = create_node(tensor->src[i], ctx, tensor_ptrs, tensor_map);
  894. }
  895. result->view_src = create_node(tensor->view_src, ctx, tensor_ptrs, tensor_map);
  896. result->view_offs = tensor->view_offs;
  897. return result;
  898. }
  899. bool rpc_server::graph_compute(const std::vector<uint8_t> & input, std::vector<uint8_t> & output) {
  900. // serialization format:
  901. // | n_nodes (4 bytes) | nodes (n_nodes * sizeof(uint64_t) | n_tensors (4 bytes) | tensors (n_tensors * sizeof(rpc_tensor)) |
  902. if (input.size() < sizeof(uint32_t)) {
  903. return false;
  904. }
  905. uint32_t n_nodes;
  906. memcpy(&n_nodes, input.data(), sizeof(n_nodes));
  907. if (input.size() < sizeof(uint32_t) + n_nodes*sizeof(uint64_t) + sizeof(uint32_t)) {
  908. return false;
  909. }
  910. const uint64_t * nodes = (const uint64_t *)(input.data() + sizeof(n_nodes));
  911. uint32_t n_tensors;
  912. memcpy(&n_tensors, input.data() + sizeof(n_nodes) + n_nodes*sizeof(uint64_t), sizeof(n_tensors));
  913. if (input.size() < sizeof(uint32_t) + n_nodes*sizeof(uint64_t) + sizeof(uint32_t) + n_tensors*sizeof(rpc_tensor)) {
  914. return false;
  915. }
  916. const rpc_tensor * tensors = (const rpc_tensor *)(input.data() + sizeof(n_nodes) + n_nodes*sizeof(uint64_t) + sizeof(n_tensors));
  917. GGML_PRINT_DEBUG("[%s] n_nodes: %u, n_tensors: %u\n", __func__, n_nodes, n_tensors);
  918. static size_t buf_size = ggml_tensor_overhead()*(n_nodes + n_tensors) + ggml_graph_overhead_custom(n_nodes, false);
  919. struct ggml_init_params params = {
  920. /*.mem_size =*/ buf_size,
  921. /*.mem_buffer =*/ NULL,
  922. /*.no_alloc =*/ true,
  923. };
  924. struct ggml_context * ctx = ggml_init(params);
  925. struct ggml_cgraph * graph = ggml_new_graph_custom(ctx, n_nodes, false);
  926. graph->n_nodes = n_nodes;
  927. std::unordered_map<uint64_t, const rpc_tensor*> tensor_ptrs;
  928. for (uint32_t i = 0; i < n_tensors; i++) {
  929. tensor_ptrs[tensors[i].id] = &tensors[i];
  930. }
  931. std::unordered_map<uint64_t, ggml_tensor*> tensor_map;
  932. for (uint32_t i = 0; i < n_nodes; i++) {
  933. graph->nodes[i] = create_node(nodes[i], ctx, tensor_ptrs, tensor_map);
  934. }
  935. ggml_status status = ggml_backend_graph_compute(backend, graph);
  936. // output serialization format: | status (1 byte) |
  937. output.resize(1, 0);
  938. output[0] = status;
  939. ggml_free(ctx);
  940. return true;
  941. }
  942. rpc_server::~rpc_server() {
  943. for (auto buffer : buffers) {
  944. ggml_backend_buffer_free(buffer);
  945. }
  946. }
  947. static void rpc_serve_client(ggml_backend_t backend, sockfd_t sockfd, size_t free_mem, size_t total_mem) {
  948. rpc_server server(backend);
  949. while (true) {
  950. uint8_t cmd;
  951. if (!recv_data(sockfd, &cmd, 1)) {
  952. break;
  953. }
  954. std::vector<uint8_t> input;
  955. std::vector<uint8_t> output;
  956. uint64_t input_size;
  957. if (!recv_data(sockfd, &input_size, sizeof(input_size))) {
  958. break;
  959. }
  960. input.resize(input_size);
  961. if (!recv_data(sockfd, input.data(), input_size)) {
  962. break;
  963. }
  964. bool ok = true;
  965. switch (cmd) {
  966. case ALLOC_BUFFER: {
  967. ok = server.alloc_buffer(input, output);
  968. break;
  969. }
  970. case GET_ALIGNMENT: {
  971. server.get_alignment(output);
  972. break;
  973. }
  974. case GET_MAX_SIZE: {
  975. server.get_max_size(output);
  976. break;
  977. }
  978. case BUFFER_GET_BASE: {
  979. ok = server.buffer_get_base(input, output);
  980. break;
  981. }
  982. case FREE_BUFFER: {
  983. ok = server.free_buffer(input);
  984. break;
  985. }
  986. case BUFFER_CLEAR: {
  987. ok = server.buffer_clear(input);
  988. break;
  989. }
  990. case SET_TENSOR: {
  991. ok = server.set_tensor(input);
  992. break;
  993. }
  994. case GET_TENSOR: {
  995. ok = server.get_tensor(input, output);
  996. break;
  997. }
  998. case COPY_TENSOR: {
  999. ok = server.copy_tensor(input, output);
  1000. break;
  1001. }
  1002. case GRAPH_COMPUTE: {
  1003. ok = server.graph_compute(input, output);
  1004. break;
  1005. }
  1006. case GET_DEVICE_MEMORY: {
  1007. // output serialization format: | free (8 bytes) | total (8 bytes) |
  1008. output.resize(2*sizeof(uint64_t), 0);
  1009. memcpy(output.data(), &free_mem, sizeof(free_mem));
  1010. memcpy(output.data() + sizeof(uint64_t), &total_mem, sizeof(total_mem));
  1011. break;
  1012. }
  1013. default: {
  1014. fprintf(stderr, "Unknown command: %d\n", cmd);
  1015. ok = false;
  1016. }
  1017. }
  1018. if (!ok) {
  1019. break;
  1020. }
  1021. uint64_t output_size = output.size();
  1022. if (!send_data(sockfd, &output_size, sizeof(output_size))) {
  1023. break;
  1024. }
  1025. if (!send_data(sockfd, output.data(), output_size)) {
  1026. break;
  1027. }
  1028. }
  1029. }
  1030. void start_rpc_server(ggml_backend_t backend, const char * endpoint, size_t free_mem, size_t total_mem) {
  1031. std::string host;
  1032. int port;
  1033. if (!parse_endpoint(endpoint, host, port)) {
  1034. return;
  1035. }
  1036. #ifdef _WIN32
  1037. {
  1038. WSADATA wsaData;
  1039. int res = WSAStartup(MAKEWORD(2, 2), &wsaData);
  1040. if (res != 0) {
  1041. fprintf(stderr, "WSAStartup failed: %d\n", res);
  1042. return;
  1043. }
  1044. }
  1045. #endif
  1046. auto server_socket = create_server_socket(host.c_str(), port);
  1047. if (server_socket == nullptr) {
  1048. fprintf(stderr, "Failed to create server socket\n");
  1049. return;
  1050. }
  1051. while (true) {
  1052. auto client_socket = socket_accept(server_socket->fd);
  1053. if (client_socket == nullptr) {
  1054. fprintf(stderr, "Failed to accept client connection\n");
  1055. return;
  1056. }
  1057. printf("Accepted client connection, free_mem=%zu, total_mem=%zu\n", free_mem, total_mem);
  1058. rpc_serve_client(backend, client_socket->fd, free_mem, total_mem);
  1059. printf("Client connection closed\n");
  1060. }
  1061. #ifdef _WIN32
  1062. WSACleanup();
  1063. #endif
  1064. }