process_linker.c 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552
  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 ioIn[PLINK_MAX_DATA_DESCS];
  65. struct iovec ioOut[PLINK_MAX_DATA_DESCS];
  66. int sockfd;
  67. int cfd[MAX_CONNECTIONS];
  68. int connect[MAX_CONNECTIONS];
  69. int count; // connected client number
  70. char *buffer;
  71. int offset;
  72. int pid;
  73. } PlinkContext;
  74. int log_level = PLINK_LOG_ERROR;
  75. int pid = 0;
  76. static PlinkStatus parseData(PlinkContext *ctx, PlinkPacket *pkt, int total);
  77. static PlinkStatus wait(int sockfd, int timeout_ms);
  78. static int getLogLevel();
  79. PlinkStatus
  80. PLINK_getVersion(PlinkVersion *version)
  81. {
  82. if (version != NULL)
  83. {
  84. version->v.major = PLINK_VERSION_MAJOR;
  85. version->v.minor = PLINK_VERSION_MINOR;
  86. version->v.revision = PLINK_VERSION_REVISION;
  87. version->v.step = 0;
  88. return PLINK_STATUS_OK;
  89. }
  90. else
  91. return PLINK_STATUS_ERROR;
  92. }
  93. PlinkStatus
  94. PLINK_create(PlinkHandle *plink, const char *name, PlinkMode mode)
  95. {
  96. PlinkContext *ctx = NULL;
  97. int sockfd;
  98. log_level = getLogLevel();
  99. pid = getpid();
  100. if (plink == NULL || name == NULL)
  101. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  102. "Wrong parameters: plink = %p, name = %s\n", plink, name);
  103. ctx = (PlinkContext *)malloc(sizeof(*ctx));
  104. if (NULL == ctx)
  105. PLINK_PRINT_RETURN(PLINK_STATUS_NO_MEMORY, ERROR,
  106. "Failed to allocate memory for plink\n");
  107. memset(ctx, 0, sizeof(*ctx));
  108. *plink = (PlinkHandle)ctx;
  109. sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
  110. if (-1 == sockfd)
  111. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  112. "Failed to create socket as AF_UNIX, SOCK_STREAM\n");
  113. ctx->sockfd = sockfd;
  114. ctx->mode = mode;
  115. ctx->addr.sun_family = AF_UNIX;
  116. strncpy(ctx->addr.sun_path, name, sizeof(ctx->addr.sun_path) - 1);
  117. if (mode == PLINK_MODE_SERVER)
  118. {
  119. if (unlink (name) == -1 && errno != ENOENT)
  120. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  121. "Failed to unlink %s\n", name);
  122. if (bind(sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  123. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  124. "Failed to bind to AF_UNIX address: %s\n", ctx->addr.sun_path);
  125. if (listen(sockfd, MAX_CONNECTIONS) == -1)
  126. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  127. "Failed to listen for connection request\n");
  128. }
  129. ctx->buffer = malloc(MAX_BUFFER_SIZE);
  130. if (ctx->buffer == NULL)
  131. PLINK_PRINT_RETURN(PLINK_STATUS_NO_MEMORY, ERROR,
  132. "Failed to allocate memory for internal buffer\n");
  133. ctx->pid = getpid();
  134. return PLINK_STATUS_OK;
  135. }
  136. PlinkStatus
  137. PLINK_connect(PlinkHandle plink, PlinkChannelID *channel)
  138. {
  139. PlinkContext *ctx = (PlinkContext *)plink;
  140. if (ctx == NULL)
  141. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  142. "Wrong parameters: plink = %p\n", plink);
  143. if (ctx->mode == PLINK_MODE_SERVER)
  144. {
  145. if (channel == NULL)
  146. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  147. "Wrong parameters: channel = %p\n", channel);
  148. if (ctx->count >= MAX_CONNECTIONS)
  149. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  150. "Too many connections %d while it is limited to %d\n", ctx->count, MAX_CONNECTIONS);
  151. // find available slot
  152. int i;
  153. for (i = 0; i < MAX_CONNECTIONS; i++)
  154. {
  155. if (ctx->connect[i] == 0)
  156. {
  157. ctx->connect[i] = 1;
  158. *channel = i;
  159. break;
  160. }
  161. }
  162. // wait for connection from client
  163. PLINK_PRINT(INFO, "Waiting for connection...\n");
  164. int fd = accept(ctx->sockfd, NULL, NULL);
  165. if (fd == -1)
  166. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  167. "Failed to accept connection\n");
  168. ctx->cfd[i] = fd;
  169. ctx->count++;
  170. PLINK_PRINT(INFO, "Accepted connection request from client %d (%d/%d): %d\n",
  171. i, ctx->count, MAX_CONNECTIONS, fd);
  172. }
  173. else
  174. {
  175. if (connect(ctx->sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  176. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  177. "Failed to connect to server %s: %s\n", ctx->addr.sun_path, strerror(errno));
  178. PLINK_PRINT(INFO, "Connected to server: %d\n", ctx->sockfd);
  179. }
  180. return PLINK_STATUS_OK;
  181. }
  182. PlinkStatus
  183. PLINK_send(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt)
  184. {
  185. PlinkContext *ctx = (PlinkContext *)plink;
  186. if (ctx == NULL || pkt == NULL)
  187. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  188. "Wrong parameters: plink = %p, pkt = %p\n", plink, pkt);
  189. if (pkt->num > PLINK_MAX_DATA_DESCS)
  190. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  191. "Too many data nodes to send: %d\n", pkt->num);
  192. char buf[CMSG_SPACE(sizeof(int))];
  193. memset(buf, 0, sizeof(buf));
  194. for (int i = 0; i < pkt->num; i++)
  195. {
  196. PlinkDescHdr *hdr = (PlinkDescHdr *)(pkt->list[i]);
  197. ctx->ioOut[i].iov_base = pkt->list[i];
  198. ctx->ioOut[i].iov_len = hdr->size + DATA_HEADER_SIZE;
  199. PLINK_PRINT(INFO, "Sending Out %ld bytes\n", ctx->ioOut[i].iov_len);
  200. }
  201. struct msghdr msg = {0};
  202. msg.msg_iov = ctx->ioOut;
  203. msg.msg_iovlen = pkt->num;
  204. if (pkt->fd > PLINK_INVALID_FD)
  205. {
  206. msg.msg_control = buf;
  207. msg.msg_controllen = sizeof(buf);
  208. struct cmsghdr *cmsg;
  209. cmsg = CMSG_FIRSTHDR(&msg);
  210. cmsg->cmsg_level = SOL_SOCKET;
  211. cmsg->cmsg_type = SCM_RIGHTS;
  212. cmsg->cmsg_len = CMSG_LEN(sizeof(int));
  213. *((int *)CMSG_DATA(cmsg)) = pkt->fd;
  214. PLINK_PRINT(INFO, "Sent fd %d\n", pkt->fd);
  215. }
  216. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  217. if (sendmsg(sockfd, &msg, 0) == -1)
  218. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  219. "sendmsg() failed: %s\n", strerror(errno));
  220. PLINK_PRINT(INFO, "Sent data to %d\n", sockfd);
  221. return PLINK_STATUS_OK;
  222. }
  223. PlinkStatus
  224. PLINK_recv(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt)
  225. {
  226. PlinkContext *ctx = (PlinkContext *)plink;
  227. if (ctx == NULL || pkt == NULL)
  228. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  229. "Wrong parameters: plink = %p, pkt = %p\n", plink, pkt);
  230. pkt->num = 0;
  231. char buf[CMSG_SPACE(sizeof(int))];
  232. memset(buf, 0, sizeof(buf));
  233. ctx->ioIn[0].iov_base = ctx->buffer + ctx->offset;
  234. ctx->ioIn[0].iov_len = MAX_BUFFER_SIZE - ctx->offset;
  235. struct msghdr msg = {0};
  236. msg.msg_iov = ctx->ioIn;
  237. msg.msg_iovlen = 1;
  238. msg.msg_control = buf;
  239. msg.msg_controllen = sizeof(buf);
  240. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  241. PLINK_PRINT(INFO, "Receiving data from %d\n", sockfd);
  242. int total = recvmsg (sockfd, &msg, 0);
  243. if (total > 0)
  244. PLINK_PRINT(INFO, "Received %d bytes\n", total)
  245. else if (total == 0)
  246. PLINK_PRINT_RETURN(PLINK_STATUS_NO_DATA, WARNING,
  247. "recvmsg() returns %d\n", total)
  248. else
  249. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  250. "Failed to recieve data from %d: %s\n", sockfd, strerror(errno))
  251. if (msg.msg_controllen >= CMSG_SPACE(sizeof(int)))
  252. {
  253. struct cmsghdr *cmsg;
  254. cmsg = CMSG_FIRSTHDR(&msg);
  255. pkt->fd = *((int *)CMSG_DATA(cmsg));
  256. PLINK_PRINT(INFO, "Received fd %d\n", pkt->fd);
  257. }
  258. else
  259. pkt->fd = PLINK_INVALID_FD;
  260. return parseData(ctx, pkt, ctx->offset + total);
  261. }
  262. PlinkStatus
  263. PLINK_wait(PlinkHandle plink, PlinkChannelID channel, int timeout_ms)
  264. {
  265. PlinkContext *ctx = (PlinkContext *)plink;
  266. if (ctx == NULL)
  267. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  268. "Wrong parameters: plink = %p\n", plink);
  269. int sockfd = ctx->mode == PLINK_MODE_SERVER ? ctx->cfd[channel] : ctx->sockfd;
  270. return wait(sockfd, timeout_ms);
  271. }
  272. PlinkStatus
  273. PLINK_close(PlinkHandle plink, PlinkChannelID channel)
  274. {
  275. PlinkContext *ctx = (PlinkContext *)plink;
  276. if (ctx == NULL)
  277. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  278. "Wrong parameters: plink = %p\n", plink);
  279. if (ctx->mode == PLINK_MODE_SERVER)
  280. {
  281. if (channel == PLINK_CLOSE_ALL)
  282. {
  283. // close all connections
  284. int i;
  285. for (i = 0; i < MAX_CONNECTIONS; i++)
  286. {
  287. if (ctx->connect[i] != 0)
  288. {
  289. ctx->connect[i] = 0;
  290. close(ctx->cfd[i]);
  291. ctx->count--;
  292. PLINK_PRINT(INFO, "Closed channel %d\n", i);
  293. }
  294. }
  295. unlink(ctx->addr.sun_path);
  296. close(ctx->sockfd);
  297. if (ctx->buffer != NULL)
  298. free(ctx->buffer);
  299. free(ctx);
  300. }
  301. else if (channel < MAX_CONNECTIONS && ctx->connect[channel] != 0)
  302. {
  303. close(ctx->cfd[channel]);
  304. ctx->count--;
  305. PLINK_PRINT(INFO, "Closed channel %d\n", channel);
  306. }
  307. else
  308. {
  309. PLINK_PRINT(ERROR, "Invalid channel: %d\n", channel);
  310. }
  311. unlink(ctx->addr.sun_path);
  312. }
  313. else
  314. {
  315. close(ctx->sockfd);
  316. if (ctx->buffer != NULL)
  317. free(ctx->buffer);
  318. free(ctx);
  319. }
  320. return PLINK_STATUS_OK;
  321. }
  322. PlinkStatus
  323. PLINK_connect_ex(PlinkHandle plink, PlinkChannelID *channel, int timeout_ms)
  324. {
  325. PlinkContext *ctx = (PlinkContext *)plink;
  326. if (ctx == NULL)
  327. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  328. "Wrong parameters: plink = %p\n", plink);
  329. int i = 0;
  330. if (ctx->mode == PLINK_MODE_SERVER)
  331. {
  332. if (channel == NULL)
  333. PLINK_PRINT_RETURN(PLINK_STATUS_WRONG_PARAMS, ERROR,
  334. "Wrong parameters: channel = %p\n", channel);
  335. if (ctx->count >= MAX_CONNECTIONS)
  336. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  337. "Too many connections %d while it is limited to %d\n", ctx->count, MAX_CONNECTIONS);
  338. // find available slot
  339. for (i = 0; i < MAX_CONNECTIONS; i++)
  340. {
  341. if (ctx->connect[i] == 0)
  342. {
  343. ctx->connect[i] = 1;
  344. *channel = i;
  345. break;
  346. }
  347. }
  348. // wait for connection from client
  349. PLINK_PRINT(INFO, "Waiting for connection...\n");
  350. if (wait(ctx->sockfd, timeout_ms) == PLINK_STATUS_OK)
  351. {
  352. int fd = accept(ctx->sockfd, NULL, NULL);
  353. if (fd == -1)
  354. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  355. "Failed to accept connection\n");
  356. ctx->cfd[i] = fd;
  357. ctx->count++;
  358. PLINK_PRINT(INFO, "Accepted connection request from client %d (%d/%d): %d\n",
  359. i, ctx->count, MAX_CONNECTIONS, fd);
  360. }
  361. else
  362. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  363. "No connection request within %dms\n", timeout_ms);
  364. }
  365. else
  366. {
  367. int max_tries = (timeout_ms + 1000 - 1) / 1000;
  368. for (i = 0; i < max_tries; i++)
  369. {
  370. if (connect(ctx->sockfd, (struct sockaddr *)&ctx->addr, sizeof(struct sockaddr_un)) == -1)
  371. sleep(1);
  372. else
  373. break;
  374. }
  375. if (i >= max_tries)
  376. PLINK_PRINT_RETURN(PLINK_STATUS_TIMEOUT, WARNING,
  377. "Failed to connect to server %s\n", ctx->addr.sun_path);
  378. PLINK_PRINT(INFO, "Connected to server: %d\n", ctx->sockfd);
  379. }
  380. return PLINK_STATUS_OK;
  381. }
  382. PlinkStatus
  383. PLINK_recv_ex(PlinkHandle plink, PlinkChannelID channel, PlinkPacket *pkt, int timeout_ms)
  384. {
  385. int ret = PLINK_wait(plink, channel, timeout_ms);
  386. if (ret == PLINK_STATUS_OK)
  387. {
  388. return PLINK_recv(plink, channel, pkt);
  389. }
  390. return ret;
  391. }
  392. static PlinkStatus
  393. parseData(PlinkContext *ctx, PlinkPacket *pkt, int remaining)
  394. {
  395. PlinkStatus sts = PLINK_STATUS_OK;
  396. if (ctx == NULL)
  397. return PLINK_STATUS_ERROR;
  398. int index = 0;
  399. void *buffer = ctx->buffer;
  400. while (remaining >= DATA_HEADER_SIZE)
  401. {
  402. if (index >= PLINK_MAX_DATA_DESCS)
  403. {
  404. // not enough buffer to store received data, need another recv call
  405. sts = PLINK_STATUS_MORE_DATA;
  406. PLINK_PRINT(ERROR, "sts:%d Received %d bytes,index exceed max:%d!\n",
  407. sts, remaining, PLINK_MAX_DATA_DESCS);
  408. break;
  409. }
  410. PlinkDescHdr *hdr = (PlinkDescHdr *)buffer;
  411. if (remaining < (unsigned int)hdr->size)
  412. {
  413. PLINK_PRINT(WARNING, "Not enough data received. Expect %d while only %d available\n", hdr->size, remaining);
  414. // return to get more data in next recvmsg call
  415. break;
  416. }
  417. pkt->list[index] = buffer;
  418. buffer += DATA_HEADER_SIZE + hdr->size;
  419. remaining -= DATA_HEADER_SIZE + hdr->size;
  420. index++;
  421. }
  422. // move the remaining data to the beginning of the buffer
  423. if (remaining > 0)
  424. {
  425. memmove(ctx->buffer, buffer, remaining);
  426. ctx->offset = remaining;
  427. }
  428. pkt->num = sts == PLINK_STATUS_MORE_DATA ? index-1 : index;
  429. return sts;
  430. }
  431. static int getLogLevel()
  432. {
  433. char *env = getenv("PLINK_LOG_LEVEL");
  434. if (env == NULL)
  435. return PLINK_LOG_ERROR;
  436. else
  437. {
  438. int level = atoi(env);
  439. if (level >= PLINK_LOG_MAX || level < PLINK_LOG_QUIET)
  440. return PLINK_LOG_ERROR;
  441. else
  442. return level;
  443. }
  444. }
  445. static PlinkStatus
  446. wait(int sockfd, int timeout_ms)
  447. {
  448. fd_set rfds;
  449. struct timeval tv;
  450. int ret;
  451. FD_ZERO(&rfds);
  452. FD_SET(sockfd, &rfds);
  453. tv.tv_sec = timeout_ms / 1000;
  454. tv.tv_usec = (timeout_ms % 1000) * 1000;
  455. PLINK_PRINT(INFO, "Waiting for data from socket %d, timeout %dms\n", sockfd, timeout_ms);
  456. ret = select(sockfd+1, &rfds, NULL, NULL, &tv);
  457. if (ret == -1)
  458. PLINK_PRINT_RETURN(PLINK_STATUS_ERROR, ERROR,
  459. "Failed to wait for data\n")
  460. else if (ret)
  461. return PLINK_STATUS_OK;
  462. else if (timeout_ms > 0)
  463. PLINK_PRINT_RETURN(PLINK_STATUS_TIMEOUT, WARNING,
  464. "Wait timeout\n");
  465. return PLINK_STATUS_TIMEOUT;
  466. }