video_in_hal/otherservice/rpc_library/library/src/rpc_thread.c

   1 /*
   2  * @copyright Copyright (c) 2016-2020 TOYOTA MOTOR CORPORATION.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *      http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 /**
  18  * @file rpc_thread.c
  19  * @brief RPC Library Internal Implementation--Processing of Internally Generated Threads
  20  *
  21  */
  22 /** @addtogroup RPClib_in */
  23 /** @{ */
  24
  25 #include <stdio.h>
  26 #include <unistd.h>
  27 #include <string.h>
  28 #include <errno.h>
  29 #include <malloc.h>
  30 #include <fcntl.h>
  31 #include <sys/time.h>
  32 #include <sys/select.h>
  33 #include <sys/poll.h>
  34 #include <sys/prctl.h>
  35
  36 #include <sys/socket.h>
  37 #include <sys/un.h>
  38
  39 #include <sys/inotify.h>
  40
  41 #include <other_service/rpc.h>
  42 #include "rpc_internal.h"
  43
  44 #include <native_service/cl_monitor.h>
  45 #include <native_service/cl_process.h>
  46
  47 /** Sub-threads that wake up in the RPC library */
  48 static RpcThreadInfo *Thread_info[RPC_MAX_THREADS_IN_PROCESS];
  49 static int Num_thread_info;
  50 pthread_t g_rpc_thread = RPC_NO_THREAD;
  51 UINT32 g_rpc_thread_alive;/**< Sub-thread running */
  52
  53 #define RPC_MAGIC_ID (('R'<<24)|('P'<<16)|('C'<<8)|'L')
  54
  55 /** Pipes used for communication with sub-threads
  56  * Read: Main thread, Write: Sub-thread */
  57 int g_rpc_pipe_main_sub[2] = { -1, -1 };
  58
  59 static pthread_mutex_t process_global_mutex = PTHREAD_MUTEX_INITIALIZER;
  60 #define PROCESS_MUTEX_LOCK rpc_mutex_lock(&process_global_mutex)
  61 #define PROCESS_MUTEX_UNLOCK rpc_mutex_unlock(&process_global_mutex)
  62
  63 static void *RpcThreadMain(void *ptr);
  64 static void NotifyMainThread(RpcThreadInfo *th);
  65
  66 static void KillRpcThread(void);
  67 static void NotifyAddServer(RpcThreadInfo *th);
  68 static void NotifyRemoveServer(RpcThreadInfo *th);
  69
  70 static RPC_Result RpcRegistSockName(RpcIdInfo *idinfo, char *client_sock_name, const struct ucred *cr, int wd);
  71 static RPC_Result RpcCheckSockName(RpcIdInfo *idinfo, RPC_ID client_id);
  72 static RPC_Result RpcDeleteSockName(RpcIdInfo *idinfo, int wd);
  73 static RPC_Result RpcAllDeleteSockName(RpcIdInfo *idinfo, int inotify_fd);
  74 static RPC_Result RpcCheckClientCredential(RpcIdInfo *idinfo, const struct ucred *cr);
  75
  76 #define RPC_SUB_THREAD_WAIT_SEC 5
  77
  78 #define WAIT_FOR_SUB_THREAD(loop_cond, sec) \
  79 { \
  80   struct timeval timeout; \
  81   timeout.tv_sec = sec; \
  82   timeout.tv_usec = 0; \
  83   \
  84   int fd = RPC_pipe_sub_main(th)[PIPE_READ]; \
  85   fd_set fds; \
  86   \
  87   while((loop_cond)) { \
  88     FD_ZERO(&fds); \
  89     FD_SET(fd, &fds); \
  90     int sret = select(fd + 1, &fds, NULL, NULL, &timeout); \
  91     if (sret < 0 && errno == EINTR) { \
  92       continue; \
  93     } else if (sret > 0 && FD_ISSET(fd, &fds)) { \
  94       char c; \
  95       read(fd, &c, sizeof(c)); \
  96     } else { \
  97       break; \
  98     } \
  99   } \
 100 }
 101
 102 RUNS_IN_CALLERS_THREAD
 103   RpcThreadInfo *
 104 RpcMyThreadInfo(void) {
 105   RpcThreadInfo *ret = NULL;
 106   int i;
 107   pthread_t me = pthread_self();
 108
 109   PROCESS_MUTEX_LOCK;
 110   for(i = 0; i < RPC_MAX_THREADS_IN_PROCESS ; i++) {
 111     if (Thread_info[i] != NULL
 112         && pthread_equal(Thread_info[i]->thread, me)) {
 113       ret = Thread_info[i];
 114       break;
 115     }
 116   }
 117   PROCESS_MUTEX_UNLOCK;
 118   return ret;
 119 }
 120
 121 RUNS_IN_CALLERS_THREAD
 122   RpcThreadInfo *
 123 RpcCreateThreadInfo(void) {
 124   int i;
 125   pthread_t me = pthread_self();
 126
 127   PROCESS_MUTEX_LOCK;
 128   /* Look for a free slot to store the thread info pointer */
 129   for(i = 0; i < RPC_MAX_THREADS_IN_PROCESS ; i++) {
 130     if (Thread_info[i] != NULL) {
 131       if (pthread_equal(Thread_info[i]->thread, me)) {  // LCOV_EXCL_BR_LINE 6: double check
 132         PROCESS_MUTEX_UNLOCK;  // LCOV_EXCL_START 6: double check
 133         AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 134         return Thread_info[i];
 135       }  // LCOV_EXCL_STOP
 136     } else {
 137       break;
 138     }
 139   }
 140
 141   if (i == RPC_MAX_THREADS_IN_PROCESS) {
 142     PROCESS_MUTEX_UNLOCK;
 143     //CONFIG_ASSERT("Must increase RPC_MAX_THREADS_IN_PROCESS");
 144     RPC_LOG_ERR("Must increase RPC_MAX_THREADS_IN_PROCESS");
 145     return NULL;
 146   }
 147
 148   /* Allocate area for thread info */
 149   // Because there is a timing when the server sub-thread is accessed without being initialized,
 150   // corrected so as to fill up to 0 in the MUTEX.
 151   RpcThreadInfo *th = rpc_malloc(sizeof(*th));
 152   if (th != NULL) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc malloc
 153     memset(th, 0, sizeof(*th));
 154     Thread_info[i] = th;
 155     th->magic = RPC_MAGIC_ID;
 156     Num_thread_info++;
 157   }
 158   PROCESS_MUTEX_UNLOCK;
 159
 160   if (th == NULL) {  // LCOV_EXCL_START 5: fail safe for libc malloc
 161     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 162     RPC_LOG_ERR("Can't alloc thread_info.");
 163     return NULL;
 164   }  // LCOV_EXCL_STOP
 165
 166   /* Initializing Thread Info */
 167   th->thread = me;
 168   pthread_mutex_init(&(th->th_mtx), NULL);
 169   th->sequence_number = RPC_SEQ_NUM_START;
 170
 171   return th;
 172 }
 173
 174 /*
 175  * check if the allocated client ID conflicts with the server ID
 176  * of the same thread
 177  */
 178   static int
 179 RpcCheckIdConflict(RpcThreadInfo *th, RPC_ID id) {
 180   RpcIdInfo *idinfo;
 181   idinfo = RPC_srvr_idinfo(th);
 182   if (idinfo != NULL && RPC_my_id(idinfo) == id) {  // LCOV_EXCL_START 6: double check
 183     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 184     return 1;
 185   }  // LCOV_EXCL_STOP
 186   return 0;
 187 }
 188
 189 /** Adding IDs with RPC_start()
 190  *
 191  * - Main: Create and initialize structures and tell sub-threads to add
 192  * - Sub: Add pointer to thread info (RpcThreadInfo) and notify it to the main process
 193  * - The main process waits for this procedure to finish.
 194  *   Use id_info->thread_info to determine completion (completed if not NULL)
 195  */
 196 RUNS_IN_CALLERS_THREAD
 197   int
 198 RpcCreateIdInfo(RpcThreadInfo *th,
 199     RPC_ID id, RPC_dispatch_func_t dispatch, INT32 secure_check) {
 200   RpcIdInfo *id_info = rpc_malloc(sizeof(*id_info));
 201   if (id_info == NULL) {  // LCOV_EXCL_START 5: fail safe for libc malloc
 202     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 203     return -1;
 204   }  // LCOV_EXCL_STOP
 205   memset(id_info, 0, sizeof(*id_info));
 206
 207   /*
 208    * Creates a Unix domain socket based on a given number of ports
 209    */
 210 #if defined(RPC_USE_UNIX_AUTOBIND)
 211   int sock_un = -1, secure_sock_un = -1;
 212   socklen_t sa_len;
 213   struct sockaddr_un sa_un;
 214
 215 get_id_retry:
 216   sock_un = socket(PF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);  /* Datagram socket for receiving API requests */
 217   if (sock_un < 0) {
 218     RPC_LOG_PERROR("socket");
 219     goto error;
 220   }
 221   SET_NONBLOCK(sock_un);
 222   SET_CLOSE_ON_EXEC(sock_un);
 223
 224   /*
 225    * Naming Rules for Unix domain Sockets
 226    * Server:(ID=50000-59999)
 227    *   sa_un.sun_path[0] = 0x00;
 228    *   sa_un.sun_path[1] = 'S';
 229    *   sa_un.sun_path[2-5] = sprintf("%04x", ID);
 230    *
 231    * Client:(ID=1-0xfffff)
 232    *   sa_un.sun_path[0] = 0x00;
 233    *   sa_un.sun_path[1-5] = sprintf("%05x", ID);
 234    *   ID is autobind by kernel during bind(see linux/net/unix/af_unix.c)
 235    *   ! Since it depends on the unix socket implementations of Linux, be careful when porting to other operating systems.
 236    *
 237    * ID=50000-59999 is duplicated in Server and Client,
 238    * but generated it according to the above rules when sent in the RPClib (see rpc_udp.c)
 239    *
 240    * Because file deletion is leaked when the system is forcibly terminated and abnormal process termination
 241    * by a traditional way to create and bind files under /tmp/RPC/,
 242    * change to the above method(2009.02.04,2012.01.21)
 243    */
 244   memset(&sa_un, 0, sizeof(sa_un));
 245   sa_un.sun_family = AF_UNIX;
 246   if (dispatch != NULL) {  // server
 247     RpcSetServerName(sa_un.sun_path, id);
 248     sa_len = sizeof(sa_un.sun_family) + SOCK_NAME_LEN;
 249   } else {  // client
 250     // Automatically assign name (ID) by unix_autobind()
 251     sa_len = sizeof(sa_un.sun_family);
 252   }
 253
 254 #else /* !AUTOBIND */
 255   int sock_un = -1;
 256   struct sockaddr_un sa_un;
 257   sa_un.sun_family = AF_UNIX;
 258   rpc_set_socket_name(sa_un.sun_path, id);
 259   sock_un = socket(PF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);
 260   if (sock_un < 0) {
 261     RPC_LOG_PERROR("socket(unix)");
 262     goto error;
 263   }
 264   SET_NONBLOCK(sock_un);
 265   SET_CLOSE_ON_EXEC(sock_un);
 266   unlink(sa_un.sun_path);
 267   sa_len = sizeof(sa_un);
 268 #endif  /* !AUTOBIND */
 269   if (bind(sock_un, (struct sockaddr *)&sa_un, sa_len) < 0) {
 270     RPC_LOG_PERROR("DGRAM : bind(unix), ID:%#x", id);
 271     goto error;
 272   }
 273
 274 #if defined(RPC_USE_UNIX_AUTOBIND)
 275   if (dispatch == NULL) {  // client
 276     // Retrieves the assigned name (ID)
 277     socklen_t len = sizeof(sa_un);
 278     if (getsockname(sock_un, (struct sockaddr *)&sa_un, &len) < 0) {
 279       perror("getsockname");
 280       goto error;
 281     }
 282     RpcGetClientName(sa_un.sun_path, &id);
 283     if (RpcCheckIdConflict(th, id)) {  // LCOV_EXCL_BR_LINE 8: dead code, RpcCheckIdConflict always is false
 284       // LCOV_EXCL_START 8: dead code, RpcCheckIdConflict always is false
 285       AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 286       RPC_LOG_STATE("ID %d conflicts with server -- get next one", id);
 287       close(sock_un);
 288       goto get_id_retry;
 289     }  // LCOV_EXCL_STOP
 290     RPC_LOG_DEBUG("client %s", sa_un.sun_path + 1);
 291   }
 292 #endif /* AUTOBIND */
 293
 294   id_info->port = id;
 295   id_info->sock = sock_un;
 296
 297   if (dispatch != NULL) {  /* server */
 298     rpc_assert(th->srvr_id == NULL);  // LCOV_EXCL_BR_LINE 6: double check
 299     RpcApicallInfo *apicall = rpc_malloc(sizeof(*apicall));
 300     if (apicall == NULL) {  // LCOV_EXCL_START 5: fail safe for libc malloc
 301       AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 302       goto error;
 303     }  // LCOV_EXCL_STOP
 304
 305     /* Create Socket for Authentication */
 306     socklen_t secure_sa_len;
 307     struct sockaddr_un secure_sa_un;
 308
 309     secure_sock_un = socket(PF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);  /* stream socket for authentication */
 310     if (secure_sock_un < 0) {
 311       RPC_LOG_PERROR("socket");
 312       rpc_free(apicall);
 313       apicall = NULL;
 314       goto error;
 315     }
 316     SET_NONBLOCK(secure_sock_un);
 317     SET_CLOSE_ON_EXEC(secure_sock_un);
 318
 319     memset(&secure_sa_un, 0, sizeof(secure_sa_un));
 320     secure_sa_un.sun_family = AF_UNIX;
 321     RpcSetServerSecureName(secure_sa_un.sun_path, id);
 322     secure_sa_len = sizeof(secure_sa_un.sun_family) + SECURE_SOCK_NAME_LEN;
 323
 324     /* Bind socket for authentication */
 325     if (bind(secure_sock_un, (struct sockaddr *)&secure_sa_un, secure_sa_len) < 0) {
 326       RPC_LOG_PERROR("STREAM : bind(unix), ID:%#x", id);
 327       rpc_free(apicall);
 328       apicall = NULL;
 329       goto error;
 330     }
 331
 332     id_info->secure_sock = secure_sock_un;
 333
 334     memset(apicall, 0, sizeof(*apicall));
 335     apicall->dispatch_func = dispatch;
 336     apicall->pipe_sub_main[PIPE_READ] = -1;
 337     apicall->pipe_sub_main[PIPE_WRITE] = -1;
 338     apicall->timeout_sec = 30;  /* Server API processing timeout */
 339     apicall->secure_check = secure_check;  /* Authentication check by UID list */
 340     if (NEED_SECURE_CHECK == secure_check) {  /* Initializes the UID list with not-registered if secured given. */
 341       apicall->regist_credential_info = NO_REGISTERED;
 342     }
 343     apicall->sock_info_head = NULL;  /* Leading Node of Source Client Socket Info */
 344     apicall->client_sock_name_num = 0;  /* Number of elements in the source client's socket name list */
 345     apicall->in_process_client = RPC_NO_PORT;  /* Client RPC_ID during API processing */
 346
 347     id_info->apicall = apicall;
 348     th->srvr_id = id_info;
 349
 350     /* Creating a pipe for communication pipe from sub-thread to main-thread direction */
 351     if (pipe(apicall->pipe_sub_main) != 0) {
 352       RPC_LOG_PERROR("pipe");
 353       goto error;
 354     }
 355     SET_NONBLOCK(apicall->pipe_sub_main[PIPE_READ]);
 356     SET_CLOSE_ON_EXEC(apicall->pipe_sub_main[PIPE_READ]);
 357     SET_NONBLOCK(apicall->pipe_sub_main[PIPE_WRITE]);
 358     SET_CLOSE_ON_EXEC(apicall->pipe_sub_main[PIPE_WRITE]);
 359
 360     PROCESS_MUTEX_LOCK;
 361     if (g_rpc_thread == RPC_NO_THREAD) {  /* There are no sub-threads. */
 362       /* Creating a pipe for communication from main-thread to sub-thread direction */
 363       if (pipe(g_rpc_pipe_main_sub) != 0) {
 364         RPC_LOG_PERROR("pipe");
 365         PROCESS_MUTEX_UNLOCK;
 366         goto error;
 367       }
 368
 369       SET_NONBLOCK(g_rpc_pipe_main_sub[PIPE_READ]);
 370       SET_CLOSE_ON_EXEC(g_rpc_pipe_main_sub[PIPE_READ]);
 371       SET_NONBLOCK(g_rpc_pipe_main_sub[PIPE_WRITE]);
 372       SET_CLOSE_ON_EXEC(g_rpc_pipe_main_sub[PIPE_WRITE]);
 373
 374       /* Creating sub-thread */
 375       pthread_t read_th;
 376       if (pthread_create(&read_th, NULL, RpcThreadMain, 0) != 0) {
 377         RPC_LOG_PERROR("pthread_create");
 378         PROCESS_MUTEX_UNLOCK;
 379         goto error;
 380       }
 381       g_rpc_thread = read_th;
 382     }
 383     PROCESS_MUTEX_UNLOCK;
 384
 385     /* Instruct a sub-thread to add and wait for completion */
 386     NotifyAddServer(th);
 387     // LCOV_EXCL_BR_START 15: macro define in rpc_thread.c
 388     WAIT_FOR_SUB_THREAD((th->srvr_id->thread_info == NULL),
 389         RPC_SUB_THREAD_WAIT_SEC);
 390     // LCOV_EXCL_BR_STOP
 391     rpc_assert(th->srvr_id->thread_info != NULL);  // LCOV_EXCL_BR_LINE 6: double check
 392
 393   } else {  /* dispatch == NULL => client */
 394     id_info->count = 1;
 395     id_info->thread_info = th;
 396     th->clnt_id = id_info;
 397   }
 398   return 0;  /* pgr0524 */
 399
 400 error:
 401   if (g_rpc_pipe_main_sub[PIPE_READ] >= 0) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc socket
 402     close(g_rpc_pipe_main_sub[PIPE_READ]);
 403   }
 404   if (g_rpc_pipe_main_sub[PIPE_WRITE] >= 0) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc socket
 405     close(g_rpc_pipe_main_sub[PIPE_WRITE]);
 406   }
 407   if (id_info->apicall != NULL) {
 408     if (id_info->apicall->pipe_sub_main[PIPE_READ] >= 0) {
 409       close(id_info->apicall->pipe_sub_main[PIPE_READ]);
 410     }
 411     if (id_info->apicall->pipe_sub_main[PIPE_WRITE] >= 0) {
 412       close(id_info->apicall->pipe_sub_main[PIPE_WRITE]);
 413     }
 414     rpc_free(id_info->apicall);
 415   }
 416   if (sock_un != -1) {
 417     close(sock_un);
 418 #if !defined(RPC_USE_UNIX_AUTOBIND)
 419     unlink(sa_un.sun_path);
 420 #endif  /* !AUTOBIND */
 421   }
 422   if (secure_sock_un != -1) {
 423     close(secure_sock_un);
 424 #if !defined(RPC_USE_UNIX_AUTOBIND)
 425     unlink(secure_sa_un.sun_path);
 426 #endif  /* !AUTOBIND */
 427   }
 428   rpc_free(id_info);
 429   th->srvr_id = NULL;
 430   return -1;  /* pgr0524 */
 431 }
 432
 433 /*
 434  * Notify an unfinished request of an error at server termination (RPC_end).
 435  */
 436   static void
 437 RpcSendErrorToPendingRequest(RpcThreadInfo *th, RpcIdInfo *idinfo) {
 438   UINT16 api_num;
 439   RPC_ID client;
 440   char *args_string;
 441   unsigned int args_size;
 442   rpc_send_buf sendbuf;
 443   char retcode[10];
 444
 445   do {
 446     api_num = RpcGetAPIRequest(idinfo, &client,
 447         &args_string, &args_size);
 448     if (api_num > 0) {  /* API calls are queued */
 449
 450       sprintf(retcode, "%08x ", RPC_ERR_Fatal);
 451       sendbuf.buf = retcode;
 452       sendbuf.bytes = sizeof(retcode) - 1;
 453       RpcSendUdp2(idinfo, client, RPC_SEND_TO_CLIENT,
 454           RPC_PACKET_APIRETURN, 1, &sendbuf);
 455       RPC_LOG_STATE("sent error result to pending client %05x", client);
 456       RpcFreeAPIArgsString(args_string);
 457     }
 458   } while(api_num > 0);
 459 }
 460
 461 /*
 462  * Notify unfinished request of deadlock when deadlock of the server is detected.
 463  */
 464   static void
 465 RpcSendDeadlockToPendingRequest(RpcThreadInfo *th, RpcIdInfo *idinfo) {
 466   UINT16 api_num;
 467   RPC_ID client;
 468   char *args_string;
 469   unsigned int args_size;
 470   rpc_send_buf sendbuf;
 471   char retcode[10];
 472
 473   do {
 474     api_num = RpcGetAPIRequest(idinfo, &client,
 475         &args_string, &args_size);
 476     if (api_num > 0) {  /* API calls are queued */
 477
 478       sprintf(retcode, "%08x ", RPC_ERR_Server_DeadLock);
 479       sendbuf.buf = retcode;
 480       sendbuf.bytes = sizeof(retcode) - 1;
 481       RpcSendUdp2(idinfo, client, RPC_SEND_TO_CLIENT,
 482           RPC_PACKET_APIRETURN, 1, &sendbuf);
 483       RPC_LOG_STATE("sent deadlock result to pending client %05x", client);
 484       RpcFreeAPIArgsString(args_string);
 485     }
 486   } while(api_num > 0);
 487
 488   if (RPC_NO_PORT != RPC_apicall_in_process_client(idinfo)) {
 489     sprintf(retcode, "%08x ", RPC_ERR_Server_DeadLock);
 490     sendbuf.buf = retcode;
 491     sendbuf.bytes = sizeof(retcode) - 1;
 492     RpcSendUdp2(idinfo, RPC_apicall_in_process_client(idinfo), RPC_SEND_TO_CLIENT,
 493         RPC_PACKET_APIRETURN, 1, &sendbuf);
 494     RPC_LOG_STATE("sent deadlock result to pending client %05x", RPC_apicall_in_process_client(idinfo));
 495     RPC_apicall_in_process_client(idinfo) = RPC_NO_PORT;
 496   }
 497 }
 498
 499 /** Delete RPC_ID Info by RPC_end()
 500  *
 501  * - Main: Notify sub of deletion of RPC_ID info.
 502  * - Sub: Delete a pointer from thread info (RpcThreadInfo) and notify main of that.
 503  * - Main waits for this procedure to finish.
 504  *   Use id_info->thread_info to determine completion (completed if NULL).
 505  *   Then, release the memory related to RPC_ID info and close the socket.
 506  */
 507 RUNS_IN_CALLERS_THREAD
 508   void
 509 RpcDestroyIdInfo(RpcThreadInfo *th, RpcIdInfo *id_info) {
 510   rpc_assert(id_info->count == 0);  // LCOV_EXCL_BR_LINE 6: double check
 511   if (id_info->apicall) {
 512     if (g_rpc_thread_alive != 0) {
 513       NotifyRemoveServer(th);
 514       /* Wait for a sub-thread to recognize IDinfo deletion */
 515       // LCOV_EXCL_BR_LINE 15: macro define in rpc_thread.c
 516       WAIT_FOR_SUB_THREAD((th->srvr_id->thread_info != NULL),
 517           RPC_SUB_THREAD_WAIT_SEC);
 518       // LCOV_EXCL_BR_STOP
 519       rpc_assert(th->srvr_id->thread_info == NULL);  /* not recognized yet */  // LCOV_EXCL_BR_LINE 6: double check
 520     }
 521
 522     if (id_info->apicall->pipe_sub_main[PIPE_READ] >= 0) {
 523       close(id_info->apicall->pipe_sub_main[PIPE_READ]);
 524     }
 525     if (id_info->apicall->pipe_sub_main[PIPE_WRITE] >= 0) {
 526       close(id_info->apicall->pipe_sub_main[PIPE_WRITE]);
 527     }
 528     if (id_info->secure_sock >= 0) {
 529       close(id_info->secure_sock);
 530     }
 531     rpc_free(id_info->apicall);
 532     th->srvr_id = NULL;
 533   } else {
 534     th->clnt_id = NULL;
 535   }
 536   rpc_free(id_info);
 537 }
 538
 539 RUNS_IN_CALLERS_THREAD
 540   void
 541 RpcDestroyThreadInfo(void) {
 542   unsigned int i;
 543   RpcThreadInfo *th = RpcMyThreadInfo();
 544   if (th == NULL) {
 545     return;
 546   }
 547
 548   RPC_THREAD_MUTEX_LOCK(th);
 549   if (th->thread == RPC_NO_THREAD) {  // LCOV_EXCL_BR_LINE 6: double check
 550     RPC_THREAD_MUTEX_UNLOCK(th);  // LCOV_EXCL_START 6: double check
 551     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 552     return;
 553   }  // LCOV_EXCL_STOP
 554   th->thread = RPC_NO_THREAD;
 555   RPC_THREAD_MUTEX_UNLOCK(th);
 556
 557   PROCESS_MUTEX_LOCK;
 558   /*
 559    * Remove the pointer from the global variable.
 560    * Subsequent calls to RpcMyThreadInfo() return NULL
 561    */
 562   for(i = 0; i < RPC_MAX_THREADS_IN_PROCESS ; i++) {
 563     if (th == Thread_info[i]) {
 564       Thread_info[i] = NULL;
 565       Num_thread_info--;
 566       break;
 567     }
 568   }
 569   PROCESS_MUTEX_UNLOCK;
 570   BUG_ASSERT(i < RPC_MAX_THREADS_IN_PROCESS, "No info in Thread_info[]");  // LCOV_EXCL_BR_LINE 15: marco defined in rpc_internal.h
 571
 572   if (Num_thread_info == 0 && g_rpc_thread_alive != 0) {
 573     KillRpcThread();
 574     char name[32];
 575     prctl(PR_GET_NAME, name);
 576     RPC_LOG_DEBUG("[%s]waiting for sub thread to join...", name);
 577     pthread_join(g_rpc_thread, NULL);
 578     RPC_LOG_DEBUG("[%s]sub thread joined.", name);
 579     g_rpc_thread = RPC_NO_THREAD;  /* bug fix */
 580     rpc_assert(g_rpc_thread_alive == 0);  // LCOV_EXCL_BR_LINE 6: double check
 581     close(g_rpc_pipe_main_sub[PIPE_READ]);
 582     close(g_rpc_pipe_main_sub[PIPE_WRITE]);
 583   }
 584
 585   if (th->srvr_id != NULL) {  // LCOV_EXCL_BR_LINE 6: double check
 586     RpcDestroyIdInfo(th, th->srvr_id);  // LCOV_EXCL_START 6: double check
 587     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 588     th->srvr_id = NULL;
 589   }  // LCOV_EXCL_STOP
 590   if (th->clnt_id != NULL) {  // LCOV_EXCL_BR_LINE 6: double check
 591     RpcDestroyIdInfo(th, th->clnt_id);  // LCOV_EXCL_START 6: double check
 592     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 593     th->clnt_id = NULL;
 594   }  // LCOV_EXCL_STOP
 595
 596   pthread_mutex_destroy(&(th->th_mtx));
 597   rpc_free(th);
 598 }
 599
 600 #if !defined(RPC_USE_UNIX_AUTOBIND)
 601 /**
 602  * Sub-function of RPC_end_all()
 603  *
 604  * Assuming releasing memory and closing the socket are processing at immediately after the end of the process,
 605  * Suppress socket file leaks by only deleting socket files because to avoid deadlocks and shorten the time
 606  * by exclusive processing.
 607  */
 608   void
 609 RpcUnlinkSocketFiles(void) {
 610   int i;
 611   char sock_name[16];
 612
 613   for(i = 0; i < RPC_MAX_THREADS_IN_PROCESS ; i++) {
 614     RpcThreadInfo *th = Thread_info[i];
 615     if (th != NULL) {
 616       if (th->srvr_id != NULL) {
 617         rpc_set_socket_name(sock_name, RPC_port(th->srvr_id));
 618         RPC_LOG_STATE("unlink srvr %s", sock_name);
 619         unlink(sock_name);
 620       }
 621       if (th->clnt_id != NULL) {
 622         rpc_set_socket_name(sock_name, RPC_port(th->clnt_id));
 623         RPC_LOG_STATE("unlink clnt %s", sock_name);
 624         unlink(sock_name);
 625       }
 626     }
 627   }
 628 }
 629 #endif /* !AUTOBIND */
 630
 631 /*
 632  * Deadlock detection check for servers in the thread
 633  */
 634   static void
 635 RpcDeadlockCheck(RpcThreadInfo** thread_info, unsigned int num_thread_info) {
 636   struct timespec ts;
 637   clock_gettime(CLOCK_MONOTONIC, &ts);
 638
 639   unsigned int i = 0;
 640   for(i = 0 ; i < num_thread_info ; i++) {
 641     RpcThreadInfo *th = thread_info[i];
 642     RpcIdInfo *idinfo = th->srvr_id;
 643     CL_MonitorEntry_t entry;
 644
 645     if (0 == CL_MonitorGetEntry(CL_MONITOR_TYPE_RPC, RPC_port(idinfo), &entry)) {
 646       if (entry.state == CL_MONITOR_STATE_RUN && entry.timeout <= ts.tv_sec) {
 647         RPC_LOG_ERR("Time Out : RPC_ID = %#x API_NUM = %#x", entry.id, entry.user_data);
 648         fprintf(stderr, "Time Out : RPC_ID = %#x API_NUM = %#x\n", entry.id, entry.user_data);
 649         RpcSendDeadlockToPendingRequest(th, idinfo);
 650       }
 651     }
 652   }
 653 }
 654 /** Main functions of the sub-threads (READING THREAD)
 655 */
 656 RUNS_IN_READING_THREAD
 657   static void *
 658 RpcThreadMain(void *ptr __attribute__((unused))) {
 659   struct pollfd wait_files[RPC_MAX_FD_IN_PROCESS];
 660   RpcThreadInfo *thread_info[RPC_MAX_THREADS_IN_PROCESS];
 661   unsigned int num_thread_info = 0;
 662
 663   unsigned int poll_num;
 664   int need_reset_sockfd = 1;
 665   int normal_exit = 0;
 666   RPC_Result result;
 667   unsigned int i, j;
 668
 669   /* Monitoring for clients process with inotify() *//* Monitoring target filename */
 670   const int inotify_fd = inotify_init1(IN_CLOEXEC);  /* fd for process monitoring with inotify() */
 671
 672   UINT8 readbuf[RPC_UDP_PACKET_SIZE];
 673   memset(readbuf, 0, sizeof(UINT8) * RPC_UDP_PACKET_SIZE);
 674
 675   g_rpc_thread_alive = 1;
 676
 677   CL_MonitorInit(CL_MONITOR_INIT_USER);  /* Using the API for Error Monitoring */
 678
 679   // Name the thread created in the RPClib (append "_R")
 680 #define RPC_APPEND_NAME "_R"
 681 #ifndef PRF_SIZE_PROCESSNAME
 682 #define PRF_SIZE_PROCESSNAME     8  /* Limit name length for Profiler Analysis Tools */
 683 #endif
 684   {
 685     char *p, name[32];
 686     prctl(PR_GET_NAME, name);
 687     name[PRF_SIZE_PROCESSNAME] = '\0';
 688     if (strlen(name) + strlen(RPC_APPEND_NAME) > PRF_SIZE_PROCESSNAME) {
 689       p = name + PRF_SIZE_PROCESSNAME - strlen(RPC_APPEND_NAME);
 690     } else {
 691       p = name + strlen(name);
 692     }
 693     strcpy(p, RPC_APPEND_NAME);
 694     prctl(PR_SET_NAME, name);
 695   }
 696
 697   /* Set the communication pipe with the main thread to poll fd */
 698   poll_num = 1;
 699   wait_files[0].fd = g_rpc_pipe_main_sub[PIPE_READ];
 700   wait_files[0].events = POLLIN;
 701
 702 restart:
 703   for( ; ; ) {
 704     if (need_reset_sockfd) {
 705       /* Set the UDP socket of each RPC_ID to poll fd */
 706       PROCESS_MUTEX_LOCK;
 707       for(i = 0, j = 0 ; i < RPC_MAX_THREADS_IN_PROCESS ; i++) {
 708         if (Thread_info[i] != NULL) {
 709           if (Thread_info[i]->magic != RPC_MAGIC_ID) {  // LCOV_EXCL_BR_LINE 6: double check
 710             AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 711             RPC_LOG_ERR("Someone(me?) destroyed my area!");  // LCOV_EXCL_LINE 6: double check
 712           }
 713           if (Thread_info[i]->srvr_id != NULL
 714               && Thread_info[i]->srvr_id->thread_info != NULL) {
 715             thread_info[j] = Thread_info[i];
 716             j++;
 717           }
 718         }
 719       }
 720       PROCESS_MUTEX_UNLOCK;
 721       num_thread_info = j;
 722
 723       poll_num = 1;
 724       /* Register fd for monitoring with inotify() in poll() */
 725       wait_files[1].fd = inotify_fd;
 726       wait_files[1].events = POLLIN;
 727
 728       poll_num = 2;
 729       for(i = 0 ; i < num_thread_info ; i++) {
 730         /* Datagram socket for API request */
 731         wait_files[poll_num].fd = thread_info[i]->srvr_id->sock;  /* pgr0000 */
 732         wait_files[poll_num].events = POLLIN;
 733         poll_num++;
 734         // LCOV_EXCL_BR_START 5: fail safe for libc listen
 735         /* Authentication stream socket */
 736         if (0 != listen(thread_info[i]->srvr_id->secure_sock, 10)) {  /* Number of queues */
 737           AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 738           RPC_LOG_PERROR("listen(unix)");  // LCOV_EXCL_LINE 5: fail safe for libc listen
 739         }
 740         // LCOV_EXCL_BR_STOP
 741         wait_files[poll_num].fd = thread_info[i]->srvr_id->secure_sock;
 742         wait_files[poll_num].events = POLLIN;
 743         poll_num++;
 744       }
 745       need_reset_sockfd = 0;
 746     }
 747
 748     int pollret;
 749     pollret = poll(wait_files, poll_num, TIMEOUT_FOR_DEADLOCK_CHECK);
 750
 751     int save_errno = errno;
 752
 753     RpcDeadlockCheck(thread_info, num_thread_info);
 754     if (pollret < 0) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc poll
 755       if (save_errno == EINTR) {  // LCOV_EXCL_START 5: fail safe for libc poll
 756         AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 757         continue;
 758       } else {
 759         rpc_assert(pollret > 0);
 760         goto exit_read_thread;
 761       }  // LCOV_EXCL_STOP
 762     }
 763
 764     /* Commands from the main thread (via pipe) */
 765     if ((wait_files[0].revents & POLLIN) == POLLIN) {
 766       char buf[RPC_MAIN_SUB_COMMAND_SIZE];
 767       long ret, arg;
 768       int cmd;
 769       RpcThreadInfo *th;
 770       ret = read(wait_files[0].fd, buf, sizeof(buf));
 771       if (ret == sizeof(buf)) {
 772         sscanf(buf, RPC_MAIN_SUB_COMMANDs, &cmd, &arg);
 773         switch(cmd) {
 774           case RPC_COMMAND_ADD_SERVER:
 775             th = (RpcThreadInfo *)arg;
 776             th->srvr_id->thread_info = th;  /* Indicate the completion of the processing */
 777             NotifyMainThread(th);
 778             need_reset_sockfd = 1;
 779             goto restart;
 780             break;
 781
 782           case RPC_COMMAND_REMOVE_SERVER:
 783             th = (RpcThreadInfo *)arg;
 784             RpcSendErrorToPendingRequest(th, th->srvr_id);
 785
 786             RpcAllDeleteSockName(th->srvr_id, inotify_fd);  /* delete client_sock_name_list */
 787             rpc_free((th->srvr_id)->apicall->uid_list);  /* delete uid_list */
 788             rpc_free((th->srvr_id)->apicall->gid_list);  /* delete gid_list */
 789
 790             th->srvr_id->thread_info = NULL;/* Indicate the completion of the processing */
 791             NotifyMainThread(th);
 792             need_reset_sockfd = 1;
 793             goto restart;
 794             break;
 795
 796           case RPC_COMMAND_EXIT:
 797             /************ Termination request from the parent thread *************/
 798             RPC_LOG_DEBUG("Received exit command from main thread.");
 799             normal_exit = 1;
 800             goto exit_read_thread;
 801             break;
 802         }  /* switch */
 803       }  /* if (ret == sizeof(buf)) */
 804     }  /* Complete the processing of commands from the main thread */
 805
 806     /* Client Monitoring Events with inotify() */
 807     if ((wait_files[1].revents & POLLIN) == POLLIN) {
 808       UINT32 read_len = 0;
 809       int length = 0;
 810       char *buffer;
 811       buffer = (char *)rpc_malloc(BUF_LEN);
 812       if (NULL == buffer) {// LCOV_EXCL_START 5: fail safe for libc malloc
 813         AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 814         RPC_LOG_ERR("rpc_malloc() ERROR.");
 815         goto exit_read_thread;
 816       }  // LCOV_EXCL_STOP
 817
 818       if( (length = (int)read( inotify_fd, buffer, BUF_LEN ) ) < 0 ) {  // LCOV_EXCL_START 5: fail safe for libc read
 819         AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 820         RPC_LOG_PERROR("inotify read() ERROR.");
 821         rpc_free(buffer);
 822         buffer = NULL;
 823         goto exit_read_thread;
 824       }  // LCOV_EXCL_STOP
 825       while ( read_len < length ) {
 826         struct inotify_event *event = ( struct inotify_event * )&buffer[read_len];
 827
 828         if ( event->mask & IN_DELETE_SELF ) {/* Terminating a Client Process */
 829           int i;                  /* Looping variable */
 830
 831           /* Delete the source socket name from all RpcThreadInfo in the received thread */
 832           for(i = 0 ; i < num_thread_info ; i++) {
 833             RpcThreadInfo *th = thread_info[i];
 834             RpcDeleteSockName(th->srvr_id, event->wd);
 835           }
 836         }
 837         read_len += (UINT32)(EVENT_SIZE + event->len);  /* Size of the inotify_event structure */
 838       }
 839       rpc_free(buffer);
 840       goto restart;
 841     }  /* Client Monitoring Events Completed with inotify() */
 842
 843     for(i = 2 ; i < poll_num ; i++) {
 844       /* Event to the API request datagram socket */
 845       if ((i % 2 == 0) && ((wait_files[i].revents & POLLIN) == POLLIN)) {
 846         unsigned int thread_info_num = ((i/2) - 1);  /* Compute thread_info[thread_info_num] with events */
 847         RpcThreadInfo *th = thread_info[thread_info_num];  /* pgr0000 */
 848         RpcIdInfo *idinfo = th->srvr_id;
 849         for(;;) {
 850           /* RPClib packet received */
 851           int readret = RpcReadUdpPacket(idinfo, readbuf);
 852           if (readret < 0) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc recvfrom
 853             rpc_assert(readret >= 0);  // LCOV_EXCL_START 5: fail safe for libc recvfrom
 854             AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 855             goto exit_read_thread;
 856             // LCOV_EXCL_STOP
 857           } else if (readret == 0) {
 858             break;
 859           }
 860           /* successfully read udp packets */
 861           /* parse the packet and queue events */
 862           RPC_ID sender = RPC_NO_ID;
 863           UINT32 seq_num = 0;
 864           UINT32 size = 0;
 865           RPC_packet_type command = RPC_PACKET_NONE;
 866           if (RpcParsePacketHeader((const char *)readbuf, &command, &sender, &seq_num, &size) != RPC_OK) {  // LCOV_EXCL_BR_LINE 11: Unexpected branch // NOLINT(readability/nolint)
 867             goto exit_read_thread;
 868           }
 869
 870           long int api_num;
 871           char *buff;
 872
 873           switch(command) {
 874             case RPC_PACKET_APICALL:
 875               if (RPC_DEBUG != NULL) {  // LCOV_EXCL_BR_LINE 7: debug
 876                 AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 877                 printf("RPC[%s]: received APIcall\n", RPC_DEBUG);  // LCOV_EXCL_LINE 7: debug
 878               }
 879               // Return response without queuing for ALIVE query
 880               api_num = strtol((const char *)(readbuf + RPC_PACKET_HEADER_LEN), NULL, 10);
 881               if (api_num == RPC_API_NUM_RPC_ALIVE) {
 882                 RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 883                     RPC_RESPONSE_APICALL,
 884                     seq_num, "OK", sizeof("OK"));
 885                 break;
 886               }
 887
 888               /* Return BUSY if secure and unregistered in UID-list */
 889               if ((NEED_SECURE_CHECK == RPC_secure_check(idinfo))
 890                     && (NO_REGISTERED == RPC_regist_credential_info(idinfo))) {
 891                 RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 892                 RPC_RESPONSE_APICALL,
 893                 seq_num, "BUSY", sizeof("BUSY"));
 894                 RPC_LOG_ERR("Need UID list register.");
 895                 break;
 896               }
 897
 898               result = RpcQueueAPIRequestBefore(idinfo, size, (char**)&buff);
 899               if (result == RPC_OK) {
 900                 /* Check whether the source has been authenticated */
 901                 if(RPC_OK == RpcCheckSockName(idinfo, sender)) {  /* Registerd the name of the source socket */
 902                   RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 903                       RPC_RESPONSE_APICALL,
 904                       seq_num, "OK", sizeof("OK"));
 905                   RpcQueueAPIRequestAfter(idinfo, sender,
 906                       (const char *)(readbuf + RPC_PACKET_HEADER_LEN),
 907                       size, buff);
 908                 } else {  /* Not registered (in other words, first communication with the source client) */
 909                   /* Authentication request to the client */
 910                   RPC_THREAD_MUTEX_UNLOCK(idinfo->thread_info);
 911                   RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 912                       RPC_RESPONSE_APICALL,
 913                       seq_num, "CERT", sizeof("CERT"));
 914                   rpc_free(buff);
 915                   break;
 916                 }
 917               } else if (result == RPC_ERR_Busy) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc malloc
 918                 RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 919                     RPC_RESPONSE_APICALL,
 920                     seq_num, "BUSY", sizeof("BUSY"));
 921               } else {  // LCOV_EXCL_START 5: fail safe for libc malloc
 922                 AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 923                 RpcSendUdpResponse(idinfo, sender, RPC_SEND_TO_CLIENT,
 924                     RPC_RESPONSE_APICALL,
 925                     seq_num, "ERR", sizeof("ERR"));
 926                 RPC_LOG_ERR("queueing APIcall failed.(%d)", result);
 927                 goto exit_read_thread;
 928               }  // LCOV_EXCL_STOP
 929               NotifyMainThread(th);
 930               if (RPC_DEBUG != NULL) {  // LCOV_EXCL_BR_LINE 7: debug
 931                 AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 932                 printf("RPC[%s]: notified APIcall\n", RPC_DEBUG);  // LCOV_EXCL_LINE 7: debug
 933               }
 934               break;
 935
 936             default:
 937               AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 938               BUG_ASSERT(0, "Unknown UDP packet type");  // LCOV_EXCL_LINE 15: marco defined in rpc_internal.h
 939               goto exit_read_thread;
 940               break;
 941           } /* switch(command) */
 942         }/* for(;;) */
 943
 944         /* Event to the stream socket for authentication */
 945       } else if ((i % 2 != 0) && ((wait_files[i].revents & POLLIN) == POLLIN)) {
 946         unsigned int thread_info_num = ((i-1)/2 - 1);  /* Compute thread_info[thread_info_num] with events */
 947         struct sockaddr_un client_sa_un;
 948         socklen_t client_len = sizeof(struct sockaddr_un);
 949         struct ucred cr;  /* Structure of client credit info */
 950         RpcCertifyResult send_ret;  /* Authentication result to pass to the client */
 951         RpcThreadInfo *th = thread_info[thread_info_num];
 952         RpcIdInfo *idinfo = th->srvr_id;
 953
 954         send_ret.certify_res = CERTIFY_NG;
 955         send_ret.srvr_pid = 0;
 956
 957         /* Obtain client credit info from a connected socket */
 958         int accept_sock = accept4(wait_files[i].fd, (struct sockaddr *)&client_sa_un, &client_len, SOCK_CLOEXEC);
 959         int ret = getsockopt(accept_sock, SOL_SOCKET, SO_PEERCRED, &cr, &client_len);
 960         if (ret == 0) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc getsockopt
 961           client_sa_un = (struct sockaddr_un )client_sa_un;
 962           /* Check if UID of client is allowed to communicate */
 963           if (RPC_OK == RpcCheckClientCredential(idinfo, &cr)) {
 964
 965             /* Obtain the socket name associated with the RPC_ID of the client from the socket info */
 966             char client_sock_name[SOCK_NAME_LEN];
 967             RpcGetClientNameFromSock(client_sa_un.sun_path, client_sock_name);
 968
 969             /* Monitoring client processes with inotify */
 970             char intfy_fname[32];
 971             snprintf(intfy_fname, sizeof(intfy_fname), CL_INTFY_FILENAME_FORMAT, cr.pid);
 972             int wd = inotify_add_watch(inotify_fd, intfy_fname, IN_DELETE_SELF);
 973             if (0 > wd) {  // LCOV_EXCL_BR_LINE 5: fail safe for libc inotify_add_watch
 974               RPC_LOG_STATE("intfy_fname is Not Found [%s].", intfy_fname);
 975             }
 976
 977             /* Register the source socket name in the management table */
 978             RpcRegistSockName(idinfo, client_sock_name, &cr, wd);
 979
 980             /* Send server credit info to the client */
 981             send_ret.certify_res = CERTIFY_OK;
 982             send_ret.srvr_pid = getpid();
 983           }
 984         }
 985
 986         /* Send authentication result to client */
 987         send(accept_sock, (char*)&send_ret, sizeof(RpcCertifyResult), 0);
 988         close(accept_sock);
 989
 990         goto restart;
 991
 992       } else if ((wait_files[i].revents & ~POLLIN) != 0) {  // LCOV_EXCL_START 5: fail safe for libc poll
 993         AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 994         // POLLERR, etc. on UDP port
 995         RPC_LOG_STATE("poll error %x", wait_files[i].revents);
 996
 997         if ((wait_files[i].revents & POLLNVAL) == POLLNVAL) {
 998           need_reset_sockfd = 1;
 999           goto restart;
1000         }
1001       }  /* if ((wait_files[i].revents & POLLIN) == POLLIN) */  // LCOV_EXCL_STOP
1002
1003     }  /* processing UDP packets finished */
1004
1005   }  /* end of forever loop */
1006
1007 exit_read_thread:
1008
1009   g_rpc_thread_alive = 0;
1010   for(i = 0 ; i < num_thread_info ; i++) {  // LCOV_EXCL_BR_LINE 6: double check
1011     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1012     NotifyMainThread(thread_info[i]); /* pgr0000 */  // LCOV_EXCL_LINE 6: double check
1013   }
1014
1015   close(inotify_fd);
1016
1017   if (normal_exit == 0) {  // LCOV_EXCL_BR_LINE 6: double check
1018     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1019     RPC_LOG_CRIT("sub thread ABNORMALLY exited.");  // LCOV_EXCL_LINE 6: double check
1020   } else {
1021     RPC_LOG_DEBUG("sub thread normally exited.");
1022   }
1023   return NULL;
1024 }
1025
1026 /* Notification of sub-thread -> main-thread (via pipe) */
1027 RUNS_IN_READING_THREAD
1028   static void
1029 NotifyMainThread(RpcThreadInfo *th) {
1030   rpc_assert(th->srvr_id->apicall != NULL);  // LCOV_EXCL_BR_LINE 6: double check
1031   char c = 0;
1032   write(th->srvr_id->apicall->pipe_sub_main[PIPE_WRITE], &c, sizeof(c));
1033 }
1034
1035 /* Notification of main-thread -> sub-thread(via pipe) */
1036 /* Termination instruction */
1037 RUNS_IN_CALLERS_THREAD
1038   static void
1039 KillRpcThread(void) {
1040   char buf[RPC_MAIN_SUB_COMMAND_SIZE];
1041   sprintf(buf, RPC_MAIN_SUB_COMMAND, RPC_COMMAND_EXIT, (unsigned long)0);
1042   write(g_rpc_pipe_main_sub[PIPE_WRITE], buf, sizeof(buf));
1043 }
1044
1045 /* AddRPC_ID */
1046 RUNS_IN_CALLERS_THREAD
1047   static void
1048 NotifyAddServer(RpcThreadInfo *th) {
1049   char buf[RPC_MAIN_SUB_COMMAND_SIZE];
1050   sprintf(buf, RPC_MAIN_SUB_COMMAND, RPC_COMMAND_ADD_SERVER,
1051       (unsigned long)th);
1052   write(g_rpc_pipe_main_sub[PIPE_WRITE], buf, sizeof(buf));
1053 }
1054
1055 /* Remove RPC_ID */
1056 RUNS_IN_CALLERS_THREAD
1057   static void
1058 NotifyRemoveServer(RpcThreadInfo *th) {
1059   char buf[RPC_MAIN_SUB_COMMAND_SIZE];
1060   sprintf(buf, RPC_MAIN_SUB_COMMAND, RPC_COMMAND_REMOVE_SERVER,
1061       (unsigned long)th);
1062   write(g_rpc_pipe_main_sub[PIPE_WRITE], buf, sizeof(buf));
1063 }
1064
1065 /* Register the socket name of the source client in the management table. */
1066   static RPC_Result
1067 RpcRegistSockName(RpcIdInfo *idinfo, char *client_sock_name, const struct ucred *cr, int wd) {
1068   if ((NULL == idinfo) || (NULL == client_sock_name) || (NULL == cr) || (0 > cr->pid)) {  // LCOV_EXCL_BR_LINE 6: void *RpcThreadMain(void *ptr)
1069     RPC_LOG_ERR("RpcRegistSockName() : Invalid Param.");  // LCOV_EXCL_START 6: void *RpcThreadMain(void *ptr)
1070     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1071     return RPC_ERR_Fatal;
1072   }  // LCOV_EXCL_STOP
1073
1074   RpcClientSockNameInfo *sock_name_buf, *current;
1075
1076   sock_name_buf = rpc_malloc(sizeof(RpcClientSockNameInfo));
1077   // LCOV_EXCL_BR_START 5: fail safe for libc malloc
1078   if( sock_name_buf == NULL ){
1079     return RPC_ERR_Fatal;
1080   }
1081   // LCOV_EXCL_BR_STOP
1082   strcpy(sock_name_buf->client_sock_name, client_sock_name);  /* Socket name */
1083   sock_name_buf->pid = cr->pid;  /* PID */
1084   sock_name_buf->uid = cr->uid;  /* UID */
1085   sock_name_buf->gid = cr->gid;  /* GID */
1086   sock_name_buf->wd = wd;  /* Non-negative inotify monitored descriptors */
1087   sock_name_buf->next = NULL;  /* Pointer to next node (NULL since last node) */
1088
1089   if (0 == RPC_client_sock_name_num(idinfo)) {
1090     RPC_sock_info_head(idinfo) = sock_name_buf;
1091   } else {
1092     for (current = RPC_sock_info_head(idinfo); current->next != NULL; current = current->next)
1093       ;
1094     current->next = sock_name_buf;
1095   }
1096   RPC_client_sock_name_num_inc(idinfo);
1097   return RPC_OK;
1098 }
1099
1100 /* Check if the socket name of the source client is registered in the management table */
1101   static RPC_Result
1102 RpcCheckSockName(RpcIdInfo *idinfo, RPC_ID client_id) {
1103   if ((NULL == idinfo) || (client_id == RPC_NO_ID)) {  // LCOV_EXCL_BR_LINE 6: void *RpcThreadMain(void *ptr)
1104     RPC_LOG_ERR("RpcCheckSockName() : Invalid Param.");  // LCOV_EXCL_START 6: void *RpcThreadMain(void *ptr)
1105     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1106     return RPC_ERR_Fatal;
1107   }  // LCOV_EXCL_STOP
1108
1109   char buf[7], client_path_name[SOCK_NAME_LEN];  /* Client socket name */
1110
1111   /* Converting client_id to the socket name associated with RPC_ID */
1112   RpcSetClientName(buf, client_id);
1113   memcpy(client_path_name, buf + 1, 5);
1114   client_path_name[5] = '\0';
1115   RpcClientSockNameInfo *current = RPC_sock_info_head(idinfo);
1116
1117   /* Search source socket name in management table */
1118   while (NULL != current) {
1119     if (0 == strcmp(current->client_sock_name, client_path_name)) {  /* Registered socket name (authenticated) */
1120       return RPC_OK;
1121     }
1122     current = current->next;
1123   }
1124   return RPC_ERR_Fatal;  /* Not registerd socket name (unauthenticated) */
1125 }
1126
1127 /* Remove source client socket name from management table */
1128   static RPC_Result
1129 RpcDeleteSockName(RpcIdInfo *idinfo, int wd) {
1130   if ((NULL == idinfo) || (0 > wd)) {
1131     RPC_LOG_ERR("RpcDeleteSockName() : Invalid Param.");
1132     return RPC_ERR_Fatal;
1133   }
1134
1135   RpcClientSockNameInfo *current, *previous;
1136   current = RPC_sock_info_head(idinfo);
1137   previous = current;
1138   int cnt = 0;
1139
1140   /* Remove Source Socket Name in Management Table */
1141   while (NULL != current) {
1142     if (wd == current->wd) {  /* Delete element */
1143       if (0 == cnt) {  /* Delete the start element in the management table */
1144         RPC_sock_info_head(idinfo) = RPC_sock_info_head(idinfo)->next;
1145         rpc_free(current);
1146         current = RPC_sock_info_head(idinfo);
1147         cnt = -1;
1148       } else {  /* Delete other than the start element in the management table */
1149         previous->next = current->next;
1150         rpc_free(current);
1151         current = previous->next;
1152       }
1153       RPC_client_sock_name_num_dec(idinfo);
1154     } else {  /* Refer to the next node without deleting */
1155       previous = current;
1156       current = current->next;
1157     }
1158     cnt ++;
1159   }
1160
1161   return RPC_OK;
1162 }
1163
1164 /* Remove all source client socket names in the management table */
1165   static RPC_Result
1166 RpcAllDeleteSockName(RpcIdInfo *idinfo, int inotify_fd) {
1167   if (NULL == idinfo) {  // LCOV_EXCL_BR_LINE 6: double check in void RpcDestroyThreadInfo(void)
1168     RPC_LOG_ERR("RpcAllDeleteSockName() : Invalid Param.");  // LCOV_EXCL_START 6: void RpcDestroyThreadInfo(void)
1169     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1170     return RPC_ERR_Fatal;
1171   }  // LCOV_EXCL_STOP
1172
1173   RpcClientSockNameInfo *current = RPC_sock_info_head(idinfo);
1174
1175   while (NULL != current) {
1176     RpcClientSockNameInfo *previous = current;
1177     current = current->next;
1178
1179     if (0 <= previous->wd) {
1180       inotify_rm_watch(inotify_fd, previous->wd);
1181     }
1182
1183     rpc_free(previous);
1184     previous = NULL;
1185     RPC_client_sock_name_num_dec(idinfo);
1186   }
1187
1188   return RPC_OK;
1189 }
1190
1191
1192 /* Check if client is allowed to communicate */
1193   static RPC_Result
1194 RpcCheckClientCredential(RpcIdInfo *idinfo, const struct ucred *cr) {
1195   if ((NULL == idinfo) || (NULL == cr)) {  // LCOV_EXCL_BR_LINE 6: double check in void *RpcThreadMain(void *ptr)
1196     RPC_LOG_ERR("RpcCheckClientCredential() : Invalid Param.");  // LCOV_EXCL_START 6: void *RpcThreadMain(void *ptr)
1197     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
1198     return RPC_ERR_Fatal;
1199   }  // LCOV_EXCL_STOP
1200
1201   /* Retern RPC_OK if authentication is not required */
1202   if (NO_SECURE_CHECK == RPC_secure_check(idinfo)) {
1203     return RPC_OK;
1204   }
1205
1206   INT32 i = 0;  /* Loop counter */
1207
1208   /* Search client UID in registered UID list */
1209   for(i = 0; i < RPC_uid_num(idinfo); i++) {
1210     if(RPC_uid_list(idinfo, i) == cr->uid) {  /* Found UID in registered UID list */
1211       return RPC_OK;
1212     }
1213   }
1214
1215   /* Search client GID in registered GID list */
1216   for(i = 0; i < RPC_gid_num(idinfo); i++) {
1217     if(RPC_gid_list(idinfo, i) == cr->gid) {  /* Found GID in registered GID list. */
1218       return RPC_OK;
1219     }
1220   }
1221
1222   RPC_LOG_ERR("[Client isn't authenticated!!!!]");
1223   return RPC_ERR_Fatal;  /* Not found UID in registered UID list */
1224 }