process_linker.c 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549
  1. /*
  2. * Copyright (c) 2021-2022 Alibaba Group. All rights reserved.
  3. * License-Identifier: Apache-2.0
  4. *
  5. * Licensed under the Apache License, Version 2.0 (the "License"); you may
  6. * not use this file except in compliance with the License.
  7. * You may obtain a copy of the License at
  8. *
  9. * http://www.apache.org/licenses/LICENSE-2.0
  10. *
  11. * Unless required by applicable law or agreed to in writing, software
  12. * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  13. * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. * See the License for the specific language governing permissions and
  15. * limitations under the License.
  16. *
  17. */
  18. #include <unistd.h>
  19. #include <stdlib.h>
  20. #include <fcntl.h>
  21. #include <sys/socket.h>
  22. #include <sys/un.h>
  23. #include <sys/stat.h>
  24. #include <sys/time.h>
  25. #include <errno.h>
  26. #include <string.h>
  27. #include <stdio.h>
  28. #include <stdint.h>
  29. #include "process_linker.h"
  30. #ifndef NULL
  31. #define NULL ((void *)0)
  32. #endif
  33. #define MAX_CONNECTIONS 3
  34. #define MAX_BUFFER_SIZE (4 * 1024 * 1024)
  35. #define PLINK_PRINT(level, ...) \
  36. { \
  37. if (log_level >= PLINK_LOG_##level) \
  38. { \
  39. struct timeval ts; \
  40. gettimeofday(&ts, 0); \
  41. printf("PLINK[%d][%ld.%06ld] %s: ", pid, ts.tv_sec, ts.tv_usec, #level); \
  42. printf(__VA_ARGS__); \
  43. } \
  44. }
  45. #define PLINK_PRINT_RETURN(retcode, level, ...) \
  46. { \
  47. PLINK_PRINT(level, __VA_ARGS__) \
  48. return retcode; \
  49. }
  50. typedef enum _PlinkLogLevel
  51. {
  52. PLINK_LOG_QUIET = 0,
  53. PLINK_LOG_ERROR,
  54. PLINK_LOG_WARNING,
  55. PLINK_LOG_INFO,
  56. PLINK_LOG_DEBUG,
  57. PLINK_LOG_TRACE,
  58. PLINK_LOG_MAX
  59. } PlinkLogLevel;
  60. typedef struct _PlinkContext
  61. {
  62. PlinkMode mode;
  63. struct sockaddr_un addr;
  64. struct iovec io[PLINK_MAX_DATA_DESCS];
  65. int sockfd;
  66. int cfd[MAX_CONNECTIONS];
  67. int connect[MAX_CONNECTIONS];
  68. int count; // connected client number
  69. char *buffer;
  70. int offset;
  71. int pid;
  72. } PlinkContext;
  73. int log_level = PLINK_LOG_ERROR;
  74. int pid = 0;
  75. static PlinkStatus parseData(PlinkContext *ctx, PlinkPacket *pkt, int total);
  76. static PlinkStatus wait(int sockfd, int timeout_ms);
  77. static int getLogLevel();
  78. PlinkStatus
  79. PLINK_getVersion(PlinkVersion *version)
  80. {
  81. if (version != NULL)
  82. {
  83. version->v.major = PLINK_VERSION_MAJOR;
  84. version->v.minor = PLINK_VERSION_MINOR;
  85. version->v.revision = PLINK_VERSION_REVISION;
  86. version->v.step = 0;
  87. return PLINK_STATUS_OK;
  88. }
  89. else
  90. return PLINK_STATUS_ERROR;
  91. }
  92. PlinkStatus
  93. PLINK_create(PlinkHandle *plink, const char *name, PlinkMode mode)
  94. {
  95. PlinkContext *ctx = NULL;
  96. int sockfd;
  97. log_level = getLogLevel();
  98. pid = getpid();
  99. if (plink == NULL || name == NULL)
  100. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  101. "Wrong parameters: plink = %p, name = %s\n", plink, name);
  102. ctx = (PlinkContext *)malloc(sizeof(*ctx));
  103. if (NULL == ctx)
  104. PLINK_PRINT_RETURN(PLINK_STATUS_NO_MEMORY, ERROR,
  105. "Failed to allocate memory for plink\n");
  106. memset(ctx, 0, sizeof(*ctx));
  107. *plink = (PlinkHandle)ctx;
  108. sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
  109. if (-1 == sockfd)
  110. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  111. "Failed to create socket as AF_UNIX, SOCK_STREAM\n");
  112. ctx->sockfd = sockfd;
  113. ctx->mode = mode;
  114. ctx->addr.sun_family = AF_UNIX;
  115. strncpy(ctx->addr.sun_path, name, sizeof(ctx->addr.sun_path) - 1);
  116. if (mode == PLINK_MODE_SERVER)
  117. {
  118. if (unlink (name) == -1 && errno != ENOENT)
  119. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  120. "Failed to unlink %s\n", name);
  121. if (bind(sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  122. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  123. "Failed to bind to AF_UNIX address: %s\n", ctx->addr.sun_path);
  124. if (listen(sockfd, MAX_CONNECTIONS) == -1)
  125. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  126. "Failed to listen for connection request\n");
  127. }
  128. ctx->buffer = malloc(MAX_BUFFER_SIZE);
  129. if (ctx->buffer == NULL)
  130. PLINK_PRINT_RETURN(PLINK_STATUS_NO_MEMORY, ERROR,
  131. "Failed to allocate memory for internal buffer\n");
  132. ctx->pid = getpid();
  133. return PLINK_STATUS_OK;
  134. }
  135. PlinkStatus
  136. PLINK_connect(PlinkHandle plink, PlinkChannelID *channel)
  137. {
  138. PlinkContext *ctx = (PlinkContext *)plink;
  139. if (ctx == NULL)
  140. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  141. "Wrong parameters: plink = %p\n", plink);
  142. if (ctx->mode == PLINK_MODE_SERVER)
  143. {
  144. if (channel == NULL)
  145. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  146. "Wrong parameters: channel = %p\n", channel);
  147. if (ctx->count >= MAX_CONNECTIONS)
  148. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  149. "Too many connections %d while it is limited to %d\n", ctx->count, MAX_CONNECTIONS);
  150. // find available slot
  151. int i;
  152. for (i = 0; i < MAX_CONNECTIONS; i++)
  153. {
  154. if (ctx->connect[i] == 0)
  155. {
  156. ctx->connect[i] = 1;
  157. *channel = i;
  158. break;
  159. }
  160. }
  161. // wait for connection from client
  162. PLINK_PRINT(INFO, "Waiting for connection...\n");
  163. int fd = accept(ctx->sockfd, NULL, NULL);
  164. if (fd == -1)
  165. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  166. "Failed to accept connection\n");
  167. ctx->cfd[i] = fd;
  168. ctx->count++;
  169. PLINK_PRINT(INFO, "Accepted connection request from client %d (%d/%d): %d\n",
  170. i, ctx->count, MAX_CONNECTIONS, fd);
  171. }
  172. else
  173. {
  174. if (connect(ctx->sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  175. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  176. "Failed to connect to server %s: %s\n", ctx->addr.sun_path, strerror(errno));
  177. PLINK_PRINT(INFO, "Connected to server: %d\n", ctx->sockfd);
  178. }
  179. return PLINK_STATUS_OK;
  180. }
  181. PlinkStatus
  182. PLINK_send(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt)
  183. {
  184. PlinkContext *ctx = (PlinkContext *)plink;
  185. if (ctx == NULL || pkt == NULL)
  186. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  187. "Wrong parameters: plink = %p, pkt = %p\n", plink, pkt);
  188. if (pkt->num > PLINK_MAX_DATA_DESCS)
  189. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  190. "Too many data nodes to send: %d\n", pkt->num);
  191. char buf[CMSG_SPACE(sizeof(int))];
  192. memset(buf, 0, sizeof(buf));
  193. for (int i = 0; i < pkt->num; i++)
  194. {
  195. PlinkDescHdr *hdr = (PlinkDescHdr *)(pkt->list[i]);
  196. ctx->io[i].iov_base = pkt->list[i];
  197. ctx->io[i].iov_len = hdr->size + DATA_HEADER_SIZE;
  198. PLINK_PRINT(INFO, "Sending %ld bytes\n", ctx->io[i].iov_len);
  199. }
  200. struct msghdr msg = {0};
  201. msg.msg_iov = ctx->io;
  202. msg.msg_iovlen = pkt->num;
  203. if (pkt->fd > PLINK_INVALID_FD)
  204. {
  205. msg.msg_control = buf;
  206. msg.msg_controllen = sizeof(buf);
  207. struct cmsghdr *cmsg;
  208. cmsg = CMSG_FIRSTHDR(&msg);
  209. cmsg->cmsg_level = SOL_SOCKET;
  210. cmsg->cmsg_type = SCM_RIGHTS;
  211. cmsg->cmsg_len = CMSG_LEN(sizeof(int));
  212. *((int *)CMSG_DATA(cmsg)) = pkt->fd;
  213. PLINK_PRINT(INFO, "Sent fd %d\n", pkt->fd);
  214. }
  215. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  216. if (sendmsg(sockfd, &msg, 0) == -1)
  217. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  218. "sendmsg() failed: %s\n", strerror(errno));
  219. PLINK_PRINT(INFO, "Sent data to %d\n", sockfd);
  220. return PLINK_STATUS_OK;
  221. }
  222. PlinkStatus
  223. PLINK_recv(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt)
  224. {
  225. PlinkContext *ctx = (PlinkContext *)plink;
  226. if (ctx == NULL || pkt == NULL)
  227. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  228. "Wrong parameters: plink = %p, pkt = %p\n", plink, pkt);
  229. pkt->num = 0;
  230. char buf[CMSG_SPACE(sizeof(int))];
  231. memset(buf, 0, sizeof(buf));
  232. ctx->io[0].iov_base = ctx->buffer + ctx->offset;
  233. ctx->io[0].iov_len = MAX_BUFFER_SIZE - ctx->offset;
  234. struct msghdr msg = {0};
  235. msg.msg_iov = ctx->io;
  236. msg.msg_iovlen = 1;
  237. msg.msg_control = buf;
  238. msg.msg_controllen = sizeof(buf);
  239. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  240. PLINK_PRINT(INFO, "Receiving data from %d\n", sockfd);
  241. int total = recvmsg (sockfd, &msg, 0);
  242. if (total > 0)
  243. PLINK_PRINT(INFO, "Received %d bytes\n", total)
  244. else if (total == 0)
  245. PLINK_PRINT_RETURN(PLINK_STATUS_NO_DATA, WARNING,
  246. "recvmsg() returns %d\n", total)
  247. else
  248. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  249. "Failed to recieve data from %d: %s\n", sockfd, strerror(errno))
  250. if (msg.msg_controllen >= CMSG_SPACE(sizeof(int)))
  251. {
  252. struct cmsghdr *cmsg;
  253. cmsg = CMSG_FIRSTHDR(&msg);
  254. pkt->fd = *((int *)CMSG_DATA(cmsg));
  255. PLINK_PRINT(INFO, "Received fd %d\n", pkt->fd);
  256. }
  257. else
  258. pkt->fd = PLINK_INVALID_FD;
  259. return parseData(ctx, pkt, ctx->offset + total);
  260. }
  261. PlinkStatus
  262. PLINK_wait(PlinkHandle plink, PlinkChannelID channel, int timeout_ms)
  263. {
  264. PlinkContext *ctx = (PlinkContext *)plink;
  265. if (ctx == NULL)
  266. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  267. "Wrong parameters: plink = %p\n", plink);
  268. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  269. return wait(sockfd, timeout_ms);
  270. }
  271. PlinkStatus
  272. PLINK_close(PlinkHandle plink, PlinkChannelID channel)
  273. {
  274. PlinkContext *ctx = (PlinkContext *)plink;
  275. if (ctx == NULL)
  276. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  277. "Wrong parameters: plink = %p\n", plink);
  278. if (ctx->mode == PLINK_MODE_SERVER)
  279. {
  280. if (channel == PLINK_CLOSE_ALL)
  281. {
  282. // close all connections
  283. int i;
  284. for (i = 0; i < MAX_CONNECTIONS; i++)
  285. {
  286. if (ctx->connect[i] != 0)
  287. {
  288. ctx->connect[i] = 0;
  289. close(ctx->cfd[i]);
  290. ctx->count--;
  291. PLINK_PRINT(INFO, "Closed channel %d\n", i);
  292. }
  293. }
  294. unlink(ctx->addr.sun_path);
  295. close(ctx->sockfd);
  296. if (ctx->buffer != NULL)
  297. free(ctx->buffer);
  298. free(ctx);
  299. }
  300. else if (channel < MAX_CONNECTIONS && ctx->connect[channel] != 0)
  301. {
  302. close(ctx->cfd[channel]);
  303. ctx->count--;
  304. PLINK_PRINT(INFO, "Closed channel %d\n", channel);
  305. }
  306. else
  307. {
  308. PLINK_PRINT(ERROR, "Invalid channel: %d\n", channel);
  309. }
  310. unlink(ctx->addr.sun_path);
  311. }
  312. else
  313. {
  314. close(ctx->sockfd);
  315. if (ctx->buffer != NULL)
  316. free(ctx->buffer);
  317. free(ctx);
  318. }
  319. return PLINK_STATUS_OK;
  320. }
  321. PlinkStatus
  322. PLINK_connect_ex(PlinkHandle plink, PlinkChannelID *channel, int timeout_ms)
  323. {
  324. PlinkContext *ctx = (PlinkContext *)plink;
  325. if (ctx == NULL)
  326. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  327. "Wrong parameters: plink = %p\n", plink);
  328. int i = 0;
  329. if (ctx->mode == PLINK_MODE_SERVER)
  330. {
  331. if (channel == NULL)
  332. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  333. "Wrong parameters: channel = %p\n", channel);
  334. if (ctx->count >= MAX_CONNECTIONS)
  335. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  336. "Too many connections %d while it is limited to %d\n", ctx->count, MAX_CONNECTIONS);
  337. // find available slot
  338. for (i = 0; i < MAX_CONNECTIONS; i++)
  339. {
  340. if (ctx->connect[i] == 0)
  341. {
  342. ctx->connect[i] = 1;
  343. *channel = i;
  344. break;
  345. }
  346. }
  347. // wait for connection from client
  348. PLINK_PRINT(INFO, "Waiting for connection...\n");
  349. if (wait(ctx->sockfd, timeout_ms) == PLINK_STATUS_OK)
  350. {
  351. int fd = accept(ctx->sockfd, NULL, NULL);
  352. if (fd == -1)
  353. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  354. "Failed to accept connection\n");
  355. ctx->cfd[i] = fd;
  356. ctx->count++;
  357. PLINK_PRINT(INFO, "Accepted connection request from client %d (%d/%d): %d\n",
  358. i, ctx->count, MAX_CONNECTIONS, fd);
  359. }
  360. else
  361. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  362. "No connection request within %dms\n", timeout_ms);
  363. }
  364. else
  365. {
  366. int max_tries = (timeout_ms + 1000 - 1) / 1000;
  367. for (i = 0; i < max_tries; i++)
  368. {
  369. if (connect(ctx->sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  370. sleep(1);
  371. else
  372. break;
  373. }
  374. if (i >= max_tries)
  375. PLINK_PRINT_RETURN(PLINK_STATUS_TIMEOUT, WARNING,
  376. "Failed to connect to server %s\n", ctx->addr.sun_path);
  377. PLINK_PRINT(INFO, "Connected to server: %d\n", ctx->sockfd);
  378. }
  379. return PLINK_STATUS_OK;
  380. }
  381. PlinkStatus
  382. PLINK_recv_ex(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt, int timeout_ms)
  383. {
  384. int ret = PLINK_wait(plink, channel, timeout_ms);
  385. if (ret == PLINK_STATUS_OK)
  386. {
  387. return PLINK_recv(plink, channel, pkt);
  388. }
  389. return ret;
  390. }
  391. static PlinkStatus
  392. parseData(PlinkContext *ctx, PlinkPacket *pkt, int remaining)
  393. {
  394. PlinkStatus sts = PLINK_STATUS_OK;
  395. if (ctx == NULL)
  396. return PLINK_STATUS_ERROR;
  397. int index = 0;
  398. void *buffer = ctx->buffer;
  399. while (remaining >= DATA_HEADER_SIZE)
  400. {
  401. if (index >= PLINK_MAX_DATA_DESCS)
  402. {
  403. // not enough buffer to store received data, need another recv call
  404. sts = PLINK_STATUS_MORE_DATA;
  405. break;
  406. }
  407. PlinkDescHdr *hdr = (PlinkDescHdr *)buffer;
  408. if (remaining < (unsigned int)hdr->size)
  409. {
  410. PLINK_PRINT(WARNING, "Not enough data received. Expect %d while only %d available\n", hdr->size, remaining);
  411. // return to get more data in next recvmsg call
  412. break;
  413. }
  414. pkt->list[index] = buffer;
  415. buffer += DATA_HEADER_SIZE + hdr->size;
  416. remaining -= DATA_HEADER_SIZE + hdr->size;
  417. index++;
  418. }
  419. // move the remaining data to the beginning of the buffer
  420. if (remaining > 0)
  421. {
  422. memmove(ctx->buffer, buffer, remaining);
  423. ctx->offset = remaining;
  424. }
  425. pkt->num = sts == PLINK_STATUS_MORE_DATA ? index-1 : index;
  426. return sts;
  427. }
  428. static int getLogLevel()
  429. {
  430. char *env = getenv("PLINK_LOG_LEVEL");
  431. if (env == NULL)
  432. return PLINK_LOG_ERROR;
  433. else
  434. {
  435. int level = atoi(env);
  436. if (level >= PLINK_LOG_MAX || level < PLINK_LOG_QUIET)
  437. return PLINK_LOG_ERROR;
  438. else
  439. return level;
  440. }
  441. }
  442. static PlinkStatus
  443. wait(int sockfd, int timeout_ms)
  444. {
  445. fd_set rfds;
  446. struct timeval tv;
  447. int ret;
  448. FD_ZERO(&rfds);
  449. FD_SET(sockfd, &rfds);
  450. tv.tv_sec = timeout_ms / 1000;
  451. tv.tv_usec = (timeout_ms % 1000) * 1000;
  452. PLINK_PRINT(INFO, "Waiting for data from socket %d, timeout %dms\n", sockfd, timeout_ms);
  453. ret = select(sockfd+1, &rfds, NULL, NULL, &tv);
  454. if (ret == -1)
  455. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  456. "Failed to wait for data\n")
  457. else if (ret)
  458. return PLINK_STATUS_OK;
  459. else if (timeout_ms > 0)
  460. PLINK_PRINT_RETURN(PLINK_STATUS_TIMEOUT, WARNING,
  461. "Wait timeout\n");
  462. return PLINK_STATUS_TIMEOUT;
  463. }