service/native/framework_unified/client/NS_MessageQueue/src/ns_msg_queue.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 /// \ingroup  tag_NSMessageQueue
  19 /// \brief    API Header for Zone Player Service APIs to be used by senders and
  20 ///           receivers.
  21 ///
  22 /// APIs to register/unregister notifications and add/remove recievers to the notifications.
  23 ///
  24 ///////////////////////////////////////////////////////////////////////////////
  25
  26 #include <unistd.h>
  27 #include <stdio.h>
  28 #include <string.h>
  29 #include "ns_msg_queue.h"
  30 #include <malloc.h>
  31 #include <mqueue.h>
  32 #include <ns_mq_internal.h>
  33 #include <errno.h>
  34 #include "ns_mq_anomaly.h"
  35
  36 #include <other_service/strlcpy.h>
  37 #include <other_service/strlcat.h>
  38
  39 const UI_32 DEFAULT_MSG_PRIORITY  = 10;
  40 const UI_32 MAX_MESSAGES_STORED_IN_QUEUE = 256;
  41 const UI_32 MAX_SYNC_RESPONSE_STORED_IN_QUEUE = 2;
  42
  43 ////////////////////////////////////////////////////////////////////////////////////////////////////
  44 //  Function : TranslateError
  45 //  Translates global error variables into FW EFrameworkunifiedStatus
  46 ////////////////////////////////////////////////////////////////////////////////////////////////////
  47 EFrameworkunifiedStatus TranslateError(int error) {
  48   EFrameworkunifiedStatus eStatus = eFrameworkunifiedStatusFail;
  49
  50   switch (error) {
  51     case EOK:
  52       eStatus = eFrameworkunifiedStatusOK;
  53       break;
  54     case EBUSY:
  55       eStatus = eFrameworkunifiedStatusThreadBusy;
  56       break;
  57     case EDEADLK:
  58       eStatus = eFrameworkunifiedStatusThreadSelfJoin;
  59       break;
  60     case EFAULT:
  61       eStatus = eFrameworkunifiedStatusFault;
  62       break;
  63     case EINVAL:
  64       eStatus = eFrameworkunifiedStatusInvldParam;
  65       break;
  66     case ESRCH:
  67       eStatus = eFrameworkunifiedStatusThreadNotExist;
  68       break;
  69     case EBADF:
  70       eStatus = eFrameworkunifiedStatusErrNoEBADF;
  71       break;
  72     case EAGAIN:
  73       eStatus = eFrameworkunifiedStatusErrNoEAGAIN;
  74       break;
  75     case EINTR:
  76       eStatus = eFrameworkunifiedStatusErrNoEINTR;
  77       break;
  78     case EMSGSIZE:
  79       eStatus = eFrameworkunifiedStatusInvldBufSize;
  80       break;
  81     case ENOTSUP:
  82       eStatus = eFrameworkunifiedStatusNotImplemented;
  83       break;
  84     case EPERM:
  85       eStatus = eFrameworkunifiedStatusAccessError;
  86       break;
  87     default:
  88       eStatus = eFrameworkunifiedStatusFail;
  89       break;
  90   }
  91
  92   return eStatus;
  93 }
  94
  95 static UI_8 GetNormalizedMqName(PSTR normalized_mqname , PCSTR name, size_t size) {
  96   if ((NULL != name) && (NULL != normalized_mqname)) {
  97     if (name[0] != '/') {
  98       strlcpy(normalized_mqname, "/", size);
  99       strlcat(normalized_mqname, name, size);
 100     } else {
 101       strlcpy(normalized_mqname, name, size);
 102     }
 103
 104     return 1;
 105   } else {
 106     return 0;
 107   }
 108 }
 109
 110 HANDLE OpenReceiverNotBlocked(PCSTR name) {
 111   HANDLE rtnHandle = INVALID_HANDLE;  // invalid until made valid.
 112   PSTR normalized_qname = NULL;
 113   SI_32 q_fd = -1;  // fd to mqueue
 114   SQhandle *rcvHndl = NULL;
 115   struct mq_attr mqattr;
 116   size_t norm_qname_size = 0;
 117
 118   // Check for invalid name
 119   if (name == NULL) {
 120     return INVALID_HANDLE;
 121   }
 122
 123   if (strlen(name) >= MAX_QUEUE_NAME_SIZE) {
 124     return INVALID_HANDLE;
 125   }
 126
 127   norm_qname_size = strlen(name) + 2;
 128   normalized_qname = (PSTR)malloc(norm_qname_size);
 129
 130   if (0 != GetNormalizedMqName(normalized_qname, name, norm_qname_size)) {
 131     if (strlen(normalized_qname) > LIMIT_QUEUE_NAME_SIZE) {
 132       free(normalized_qname);
 133       return INVALID_HANDLE;
 134     }
 135
 136     mqattr.mq_flags = 0;
 137     mqattr.mq_maxmsg = (__syscall_slong_t)MAX_MESSAGES_STORED_IN_QUEUE;
 138     mqattr.mq_msgsize = MAX_QUEUE_MSG_SIZE;
 139
 140     q_fd = mq_open(normalized_qname, O_RDONLY | O_CREAT | O_NONBLOCK | O_CLOEXEC , 0666, &mqattr);
 141
 142     if (q_fd != -1) {
 143       rcvHndl = (SQhandle *)malloc(sizeof(SQhandle));
 144       if (rcvHndl != NULL) {
 145         rcvHndl->check_code = MQ_CHECK_CODE;
 146         rcvHndl->fd = q_fd;
 147         rcvHndl->q_name = normalized_qname;
 148         rcvHndl->q_type = eQTypeReveiver;
 149
 150         /////////////////////////////////////////
 151         rcvHndl->threadid = 0;
 152         /////////////////////////////////////////
 153
 154         // Set the return handle to rcvHndl
 155         rtnHandle = rcvHndl;
 156       }
 157     }
 158   }
 159
 160   if (INVALID_HANDLE == rtnHandle) {
 161     // couldn't connect the queue,
 162     // release the memory to normalized queue name
 163     if (normalized_qname != NULL) {
 164       free(normalized_qname);
 165       normalized_qname = NULL;  // mb20110108  item 11
 166     }
 167   }
 168
 169   return rtnHandle;
 170 }
 171
 172 static HANDLE openReceiverInternal(PCSTR name, UI_32 mq_maxmsg, UI_32 mq_msgsize, BOOL is_internal) {
 173   HANDLE rtnHandle = INVALID_HANDLE;  // invalid until made valid.
 174   PSTR normalized_qname = NULL;
 175   SI_32 q_fd = -1;  // fd to mqueue
 176   SQhandle *rcvHndl = NULL;
 177   struct mq_attr mqattr;
 178   size_t norm_qname_size = 0;
 179
 180   // Check for invalid name
 181   if (name == NULL) {
 182     return INVALID_HANDLE;
 183   }
 184
 185   if (strlen(name) >= MAX_QUEUE_NAME_SIZE) {
 186     return INVALID_HANDLE;
 187   }
 188
 189   norm_qname_size = strlen(name) + 2;
 190   normalized_qname = (PSTR)malloc(norm_qname_size);
 191
 192   if (0 != GetNormalizedMqName(normalized_qname, name, norm_qname_size)) {
 193     int i;
 194
 195     if (!is_internal && strlen(normalized_qname) > LIMIT_QUEUE_NAME_SIZE) {
 196       free(normalized_qname);
 197       return INVALID_HANDLE;
 198     }
 199
 200     mqattr.mq_flags = 0;
 201     mqattr.mq_maxmsg = (__syscall_slong_t)mq_maxmsg;
 202     mqattr.mq_msgsize = (__syscall_slong_t)mq_msgsize;
 203
 204     for (i = 0; i < (sizeof(mq_anomaly_list) / sizeof(mq_anomaly_list[0])); i++) {
 205       if (strcmp(normalized_qname, mq_anomaly_list[i].name) == 0) {
 206         mqattr.mq_maxmsg = (__syscall_slong_t)mq_anomaly_list[i].maxMsg;
 207         break;
 208       }
 209     }
 210
 211
 212     q_fd = mq_open(normalized_qname, O_RDONLY | O_CREAT | O_CLOEXEC, 0666, &mqattr);
 213
 214     if (q_fd != -1) {
 215       rcvHndl = (SQhandle *)malloc(sizeof(SQhandle));
 216       if (rcvHndl != NULL) {
 217         rcvHndl->check_code = MQ_CHECK_CODE;
 218         rcvHndl->fd = q_fd;
 219         rcvHndl->q_name = normalized_qname;
 220         rcvHndl->q_type = eQTypeReveiver;
 221
 222         /////////////////////////////////////////
 223         rcvHndl->threadid = 0;
 224         /////////////////////////////////////////
 225
 226         // Set the return handle to rcvHndl
 227         rtnHandle = rcvHndl;
 228       }
 229
 230     }
 231   }
 232
 233   if (INVALID_HANDLE == rtnHandle) {
 234     // couldn't connect the queue,
 235     // release the memory to normalized queue name
 236     if (normalized_qname != NULL) {
 237       free(normalized_qname);
 238       normalized_qname = NULL;  // mb20110108  item 11
 239     }
 240   }
 241
 242   return rtnHandle;
 243 }
 244
 245 HANDLE OpenReceiver(PCSTR name) {
 246   return openReceiverInternal(name, MAX_MESSAGES_STORED_IN_QUEUE, MAX_QUEUE_MSG_SIZE, FALSE);
 247 }
 248
 249 HANDLE openSyncReceiver(PCSTR name) {
 250   return openReceiverInternal(name, MAX_SYNC_RESPONSE_STORED_IN_QUEUE , MAX_QUEUE_MSG_SIZE, TRUE);
 251 }
 252
 253 static HANDLE openSenderInternal(PCSTR name, UI_32 mq_maxmsg, UI_32 mq_msgsize, BOOL is_internal, BOOL zero_copy) {
 254   HANDLE rtnHandle = INVALID_HANDLE;  // invalid until made valid.
 255   SI_32 q_fd = -1;  // fd to queue
 256   SQhandle *sndHndl = NULL;
 257   PSTR normalized_qname = NULL;
 258   size_t norm_qname_size = 0;
 259
 260   struct mq_attr mqattr;
 261
 262   // Check for invalid name
 263   if (name == NULL) {
 264     return rtnHandle;
 265   }
 266
 267   if (strlen(name) >= MAX_QUEUE_NAME_SIZE) {
 268     return rtnHandle;
 269   }
 270
 271   norm_qname_size = strlen(name) + 2;
 272   normalized_qname = (PSTR)malloc(norm_qname_size);
 273
 274   if (NULL == normalized_qname) {
 275     return rtnHandle;
 276   }
 277
 278   if (0 != GetNormalizedMqName(normalized_qname, name, norm_qname_size)) {
 279     int i;
 280
 281
 282     if (!is_internal && strlen(normalized_qname) > LIMIT_QUEUE_NAME_SIZE) {
 283       free(normalized_qname);
 284       return INVALID_HANDLE;
 285     }
 286
 287     mqattr.mq_flags = 0;
 288     mqattr.mq_maxmsg = (__syscall_slong_t)mq_maxmsg;
 289     mqattr.mq_msgsize = (__syscall_slong_t)mq_msgsize;
 290
 291     for (i = 0; i < (sizeof(mq_anomaly_list) / sizeof(mq_anomaly_list[0])); i++) {
 292       if (strcmp(normalized_qname, mq_anomaly_list[i].name) == 0) {
 293         mqattr.mq_maxmsg = (__syscall_slong_t)mq_anomaly_list[i].maxMsg;
 294         break;
 295       }
 296     }
 297
 298     q_fd = mq_open(normalized_qname, O_WRONLY | O_CREAT | O_NONBLOCK | O_CLOEXEC, 0666, &mqattr);
 299
 300     if (q_fd != -1) {
 301       sndHndl = (SQhandle *)malloc(sizeof(SQhandle));
 302       if (NULL != sndHndl) {
 303         sndHndl->check_code = MQ_CHECK_CODE;
 304         sndHndl->fd = q_fd;
 305         sndHndl->q_name = normalized_qname;
 306         sndHndl->q_type = eQTypeSender;
 307
 308         sndHndl->threadid = 0;
 309
 310
 311         // Set the return handle to sndHndl
 312         rtnHandle = sndHndl;
 313
 314         if (zero_copy) {
 315           sndHndl->sendbuf = malloc(MAX_QUEUE_MSG_SIZE);
 316           if (sndHndl->sendbuf == NULL) {
 317             // LCOV_EXCL_START 5: malloc's error case
 318             AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 319             free(sndHndl);
 320             rtnHandle = INVALID_HANDLE;
 321             // LCOV_EXCL_STOP
 322           }
 323         } else {
 324           sndHndl->sendbuf = NULL;
 325         }
 326       }
 327     }
 328   }
 329
 330   if (INVALID_HANDLE == rtnHandle) {
 331     // couldn't connect the queue,
 332     // release the memory to normalized queue name
 333     if (normalized_qname != NULL) {
 334       free(normalized_qname);
 335       normalized_qname = NULL;  // mb20110108  item 11
 336     }
 337     if (q_fd != -1) {
 338       mq_close(q_fd);
 339     }
 340   }
 341
 342   return rtnHandle;  // invalid until made valid.
 343 }
 344
 345 HANDLE OpenSender(PCSTR name) {
 346   return openSenderInternal(name, MAX_MESSAGES_STORED_IN_QUEUE, MAX_QUEUE_MSG_SIZE, FALSE, FALSE);
 347 }
 348
 349
 350 HANDLE OpenSenderChild(PCSTR name, pthread_t threadid) {
 351   HANDLE h = OpenSender(name);
 352   if (INVALID_HANDLE != h) {
 353     SQhandle *sq = (SQhandle *)h;
 354     sq->threadid = threadid;
 355   }
 356   return h;
 357 }
 358
 359
 360 HANDLE openSyncSender(PCSTR name) {
 361   return openSenderInternal(name, MAX_SYNC_RESPONSE_STORED_IN_QUEUE, MAX_QUEUE_MSG_SIZE, TRUE, FALSE);
 362 }
 363
 364 HANDLE openSenderZc(PCSTR name) {
 365   return openSenderInternal(name, MAX_MESSAGES_STORED_IN_QUEUE, MAX_QUEUE_MSG_SIZE, FALSE, TRUE);
 366 }
 367
 368
 369 EFrameworkunifiedStatus JoinChild(HANDLE hChildApp) {
 370   // mb20110108 Code re-structured per  comments 24 & 25
 371   if (mqCheckValidHandle(hChildApp)) {
 372     SQhandle *sq = (SQhandle *)hChildApp;
 373     return TranslateError(pthread_join(sq->threadid, NULL));
 374   } else {
 375     return eFrameworkunifiedStatusFail;
 376   }
 377 }
 378
 379 EFrameworkunifiedStatus GetChildThreadPriority(HANDLE hChildApp, PSI_32 threadPrio) {
 380   EFrameworkunifiedStatus eStatus = eFrameworkunifiedStatusFail;
 381   // LCOV_EXCL_BR_START 6:GetChildThreadPriority is called by McGetChildThreadPriority, and hChildApp will check there.
 382   if (mqCheckValidHandle(hChildApp)) {
 383   // LCOV_EXCL_BR_STOP
 384     SQhandle *sq = (SQhandle *)hChildApp;
 385     SI_32 schedPolicy;  // this is needed according to syntax of pthread_getschedparam. api not available to get prio directly.
 386     struct sched_param schedParam;
 387     eStatus = TranslateError(pthread_getschedparam(sq->threadid, &schedPolicy, &schedParam));
 388
 389     *threadPrio = schedParam.sched_priority;
 390   } else {
 391     // LCOV_EXCL_START 6:GetChildThreadPriority is called by McGetChildThreadPriority, and hChildApp will check there.
 392     AGL_ASSERT_NOT_TESTED();  // LCOV_EXCL_LINE 200: test assert
 393     eStatus = eFrameworkunifiedStatusInvldHandle;
 394     // LCOV_EXCL_STOP
 395   }
 396   return eStatus;
 397 }
 398
 399 /// endhack...
 400 /////////////////////////////////////////
 401
 402 EFrameworkunifiedStatus SendMessage(HANDLE hMessage, UI_32 length, PVOID data) {
 403   return SendMessageWithPriority(hMessage, length, data, eFrameworkunifiedMsgPrioNormal);
 404 }
 405
 406 EFrameworkunifiedStatus SendMessageWithPriority(HANDLE hMessage, UI_32 length, PVOID data, EFrameworkunifiedMessagePriorties priority) {
 407   // define a Q handle structure
 408   SQhandle *sndHndl = NULL;
 409
 410   // mb20110108 Added per  comments 27 & 28
 411   if (length > 0 && NULL == data) {
 412     return eFrameworkunifiedStatusInvldBuf;
 413   }
 414
 415   // check handle for null case...
 416   if (mqCheckValidHandle(hMessage) == FALSE) {
 417     return eFrameworkunifiedStatusInvldHandle;
 418   }
 419
 420   sndHndl = (SQhandle *)hMessage;
 421
 422   // check to see if this is a sender handle
 423   if (sndHndl->q_type != eQTypeSender) {
 424     return eFrameworkunifiedStatusInvldHndlType;
 425   }
 426
 427   if (-1 == mq_send((mqd_t)sndHndl->fd, (PCSTR)data, (size_t)length, (unsigned int)priority)) {
 428     if (errno == EAGAIN) {
 429       return eFrameworkunifiedStatusMsgQFull;
 430     } else {
 431       return TranslateError(errno);
 432     }
 433   }
 434   return eFrameworkunifiedStatusOK;
 435 }
 436
 437 SI_32  ReceiveMessage(HANDLE hMessage, UI_32 length, PVOID data) {
 438   // define a Q handle structure
 439   SQhandle *rcvHndl = NULL;
 440
 441   // check handle for null case...
 442   if (mqCheckValidHandle(hMessage) == FALSE) {
 443     errno = ENODATA;
 444     return -1;
 445   }
 446
 447   rcvHndl = (SQhandle *)hMessage;
 448
 449   // check to see if this is a receiver handle
 450   if (rcvHndl->q_type != eQTypeReveiver) {
 451     errno = ENODATA;
 452     return -1;
 453   }
 454
 455
 456   return (SI_32)mq_receive((mqd_t)rcvHndl->fd, (char *)data, (size_t)length, NULL);
 457 }
 458
 459 EFrameworkunifiedStatus CloseReceiver(HANDLE handle) {
 460   SQhandle *rcvHndl = NULL;
 461   SI_32 q_fd;
 462
 463   // check handle for null case...
 464   if (mqCheckValidHandle(handle) == FALSE) {
 465     return eFrameworkunifiedStatusInvldHandle;
 466   }
 467
 468   rcvHndl = (SQhandle *)handle;
 469
 470   // check to see if this is a receiver handle
 471   if (rcvHndl->q_type != eQTypeReveiver) {
 472     return eFrameworkunifiedStatusInvldHndlType;
 473   }
 474
 475   rcvHndl->check_code = 0;
 476   q_fd = rcvHndl->fd;
 477
 478   if (NULL != rcvHndl->q_name) {
 479     free(rcvHndl->q_name);    // remove the memory to the name
 480     rcvHndl->q_name = NULL;
 481   }
 482   free((void *)handle);    // remove handle.. now..
 483   handle = INVALID_HANDLE;
 484
 485   return ((-1 == mq_close((mqd_t)q_fd)) ? eFrameworkunifiedStatusInvldHandle : eFrameworkunifiedStatusOK);
 486 }
 487
 488 EFrameworkunifiedStatus CloseSender(HANDLE handle) {
 489   SQhandle *sndHndl = NULL;
 490   SI_32 q_fd;
 491
 492   // check handle for null case...
 493   if (mqCheckValidHandle(handle) == FALSE) {
 494     return eFrameworkunifiedStatusInvldHandle;
 495   }
 496
 497   sndHndl = (SQhandle *)handle;
 498
 499   // check to see if this is a sender handle
 500   if (sndHndl->q_type != eQTypeSender) {
 501     return eFrameworkunifiedStatusInvldHndlType;
 502   }
 503
 504   sndHndl->check_code = 0;
 505   q_fd = sndHndl->fd;  // copy the fd, need it to close the queues....
 506
 507   if (NULL != sndHndl->q_name) {
 508     free(sndHndl->q_name);  // remove the memory to the name
 509     sndHndl->q_name = NULL;
 510   }
 511   if (NULL != sndHndl->sendbuf) {
 512     free(sndHndl->sendbuf);
 513     sndHndl->sendbuf = NULL;
 514   }
 515   free((void *)handle);  // remove handle.. now..
 516   handle = INVALID_HANDLE;  // invalidate handle so user doesn't reuse...
 517
 518   return (-1 == mq_close((mqd_t)q_fd)) ? eFrameworkunifiedStatusInvldHandle : eFrameworkunifiedStatusOK;
 519 }
 520
 521 static UI_8 IsMessageAvailable(SI_32 fd) {
 522   struct mq_attr sMqStatus;
 523   if (-1 == mq_getattr(fd, &sMqStatus)) {
 524     // Error Detected.
 525     return 0;
 526   } else {
 527     if (0 < sMqStatus.mq_curmsgs) {
 528       return 1;
 529     } else {
 530       return 0;
 531     }
 532   }
 533 }
 534
 535 void Flush(HANDLE hMessage) {
 536   SQhandle *rcvHndl = NULL;
 537
 538   if (mqCheckValidHandle(hMessage)) {
 539     rcvHndl = (SQhandle *)hMessage;
 540
 541     if (rcvHndl->q_type != eQTypeReveiver) {
 542       return;
 543     }
 544
 545     // check there is anything on the queue before going into the loop....
 546     if (IsMessageAvailable(rcvHndl->fd)) {
 547       CHAR l_pReceiveBuffer_o[MAX_QUEUE_MSG_SIZE];
 548
 549       // read till there isn't anything on the queue.
 550       while (IsMessageAvailable(rcvHndl->fd)) {
 551         mq_receive((mqd_t)rcvHndl->fd, l_pReceiveBuffer_o, (size_t)MAX_QUEUE_MSG_SIZE, NULL);
 552       }
 553     }
 554   }
 555 }
 556
 557 EQType GetQueueType(HANDLE hMessage) {
 558   EQType qType = eQTypeInvld;
 559
 560   if (mqCheckValidHandle(hMessage)) {
 561     SQhandle *handle = (SQhandle *)hMessage;
 562     qType = handle->q_type;
 563   }
 564
 565   return qType;
 566 }
 567
 568 PCSTR GetQueueName(HANDLE hMessage) {
 569   PCSTR name = NULL;
 570
 571   if (mqCheckValidHandle(hMessage)) {
 572     SQhandle *hMsgQ = (SQhandle *)hMessage;
 573
 574     if (hMsgQ->q_type == eQTypeSender ||
 575         hMsgQ->q_type == eQTypeReveiver) {
 576       name = hMsgQ->q_name;
 577     }
 578   }
 579
 580   return name;
 581 }
 582
 583 int GetQueueFD(HANDLE hMessage) {
 584   int fd = -1;
 585
 586   if (mqCheckValidHandle(hMessage)) {
 587     SQhandle *hMsgQ = (SQhandle *)hMessage;
 588
 589     if (hMsgQ->q_type == eQTypeSender ||
 590         hMsgQ->q_type == eQTypeReveiver) {
 591       fd = hMsgQ->fd;
 592     }
 593   }
 594
 595   return fd;
 596 }