resource/csdk/connectivity/src/caretransmission.c

   1 /******************************************************************
   2  *
   3  * Copyright 2014 Samsung Electronics All Rights Reserved.
   4  *
   5  *
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  ******************************************************************/
  20
  21 // Defining _BSD_SOURCE or _DEFAULT_SOURCE causes header files to expose
  22 // definitions that may otherwise be skipped. Skipping can cause implicit
  23 // declaration warnings and/or bugs and subtle problems in code execution.
  24 // For glibc information on feature test macros,
  25 // Refer http://www.gnu.org/software/libc/manual/html_node/Feature-Test-Macros.html
  26 //
  27 // This file requires #define use due to random()
  28 // For details on compatibility and glibc support,
  29 // Refer http://www.gnu.org/software/libc/manual/html_node/BSD-Random.html
  30 #define _DEFAULT_SOURCE
  31
  32 // Defining _POSIX_C_SOURCE macro with 199309L (or greater) as value
  33 // causes header files to expose definitions
  34 // corresponding to the POSIX.1b, Real-time extensions
  35 // (IEEE Std 1003.1b-1993) specification
  36 //
  37 // For this specific file, see use of clock_gettime,
  38 // Refer to http://pubs.opengroup.org/stage7tc1/functions/clock_gettime.html
  39 // and to http://man7.org/linux/man-pages/man2/clock_gettime.2.html
  40 #ifndef _POSIX_C_SOURCE
  41 #define _POSIX_C_SOURCE 200809L
  42 #endif
  43
  44 #include <stdio.h>
  45 #include <stdlib.h>
  46 #include <string.h>
  47
  48 #ifndef SINGLE_THREAD
  49 #include <unistd.h>
  50 #include <time.h>
  51 #include <sys/time.h>
  52 #endif
  53
  54 #if defined(__ANDROID__)
  55 #include <linux/time.h>
  56 #endif
  57
  58 #include "caretransmission.h"
  59 #include "caremotehandler.h"
  60 #include "caprotocolmessage.h"
  61 #include "oic_malloc.h"
  62 #include "logger.h"
  63
  64 #define TAG "OIC_CA_RETRANS"
  65
  66 typedef struct
  67 {
  68     uint64_t timeStamp;                 /**< last sent time. microseconds */
  69 #ifndef SINGLE_THREAD
  70     uint64_t timeout;                   /**< timeout value. microseconds */
  71 #endif
  72     uint8_t triedCount;                 /**< retransmission count */
  73     uint16_t messageId;                 /**< coap PDU message id */
  74     CAEndpoint_t *endpoint;             /**< remote endpoint */
  75     void *pdu;                          /**< coap PDU */
  76     uint32_t size;                      /**< coap PDU size */
  77 } CARetransmissionData_t;
  78
  79 static const uint64_t USECS_PER_SEC = 1000000;
  80
  81 /**
  82  * @brief   getCurrent monotonic time
  83  * @return  current time in microseconds
  84  */
  85 uint64_t getCurrentTimeInMicroSeconds();
  86
  87 #ifndef SINGLE_THREAD
  88 /**
  89  * @brief   timeout value is
  90  *          between DEFAULT_ACK_TIMEOUT_SEC and
  91  *          (DEFAULT_ACK_TIMEOUT_SEC * DEFAULT_RANDOM_FACTOR) second.
  92  *          DEFAULT_RANDOM_FACTOR       1.5 (CoAP)
  93  * @return  microseconds.
  94  */
  95 static uint64_t CAGetTimeoutValue()
  96 {
  97 #if !defined(__msys_nt__)
  98     return ((DEFAULT_ACK_TIMEOUT_SEC * 1000) + ((1000 * (random() & 0xFF)) >> 8)) *
  99             (uint64_t) 1000;
 100 #else
 101     return ((DEFAULT_ACK_TIMEOUT_SEC * 1000) + ((1000 * (rand() & 0xFF)) >> 8)) *
 102             (uint64_t) 1000;
 103 #endif
 104 }
 105
 106 CAResult_t CARetransmissionStart(CARetransmission_t *context)
 107 {
 108     if (NULL == context)
 109     {
 110         OIC_LOG(ERROR, TAG, "context is empty");
 111         return CA_STATUS_INVALID_PARAM;
 112     }
 113
 114     if (NULL == context->threadPool)
 115     {
 116         OIC_LOG(ERROR, TAG, "thread pool handle is empty..");
 117         return CA_STATUS_INVALID_PARAM;
 118     }
 119
 120     CAResult_t res = ca_thread_pool_add_task(context->threadPool, CARetransmissionBaseRoutine,
 121                                              context);
 122
 123     if (CA_STATUS_OK != res)
 124     {
 125         OIC_LOG(ERROR, TAG, "thread pool add task error(send thread).");
 126         return res;
 127     }
 128
 129     return res;
 130 }
 131 #endif
 132
 133 /**
 134  * @brief   check timeout routine
 135  * @param   currentTime     [IN]microseconds
 136  * @param   retData         [IN]retransmission data
 137  * @return  true if the timeout period has elapsed, false otherwise
 138  */
 139 static bool CACheckTimeout(uint64_t currentTime, CARetransmissionData_t *retData)
 140 {
 141 #ifndef SINGLE_THREAD
 142     // #1. calculate timeout
 143     uint32_t milliTimeoutValue = retData->timeout * 0.001;
 144     uint64_t timeout = (milliTimeoutValue << retData->triedCount) * (uint64_t) 1000;
 145
 146     if (currentTime >= retData->timeStamp + timeout)
 147     {
 148         OIC_LOG_V(DEBUG, TAG, "%llu microseconds time out!!, tried count(%d)",
 149                   timeout, retData->triedCount);
 150         return true;
 151     }
 152 #else
 153     // #1. calculate timeout
 154     uint64_t timeOut = (2 << retData->triedCount) * 1000000;
 155
 156     if (currentTime >= retData->timeStamp + timeOut)
 157     {
 158         OIC_LOG_V(DEBUG, TAG, "timeout=%d, tried cnt=%d",
 159                   (2 << retData->triedCount), retData->triedCount);
 160         return true;
 161     }
 162 #endif
 163     return false;
 164 }
 165
 166 static void CACheckRetransmissionList(CARetransmission_t *context)
 167 {
 168     if (NULL == context)
 169     {
 170         OIC_LOG(ERROR, TAG, "context is null");
 171         return;
 172     }
 173
 174     // mutex lock
 175     ca_mutex_lock(context->threadMutex);
 176
 177     uint32_t i = 0;
 178     uint32_t len = u_arraylist_length(context->dataList);
 179
 180     for (i = 0; i < len; i++)
 181     {
 182         CARetransmissionData_t *retData = u_arraylist_get(context->dataList, i);
 183
 184         if (NULL == retData)
 185         {
 186             continue;
 187         }
 188
 189         uint64_t currentTime = getCurrentTimeInMicroSeconds();
 190
 191         if (CACheckTimeout(currentTime, retData))
 192         {
 193             // #2. if time's up, send the data.
 194             if (NULL != context->dataSendMethod)
 195             {
 196                 OIC_LOG_V(DEBUG, TAG, "retransmission CON data!!, msgid=%d",
 197                           retData->messageId);
 198                 context->dataSendMethod(retData->endpoint, retData->pdu, retData->size);
 199             }
 200
 201             // #3. increase the retransmission count and update timestamp.
 202             retData->timeStamp = currentTime;
 203             retData->triedCount++;
 204         }
 205
 206         // #4. if tried count is max, remove the retransmission data from list.
 207         if (retData->triedCount >= context->config.tryingCount)
 208         {
 209             CARetransmissionData_t *removedData = u_arraylist_remove(context->dataList, i);
 210             if (NULL == removedData)
 211             {
 212                 OIC_LOG(ERROR, TAG, "Removed data is NULL");
 213                 // mutex unlock
 214                 ca_mutex_unlock(context->threadMutex);
 215                 return;
 216             }
 217             OIC_LOG_V(DEBUG, TAG, "max trying count, remove RTCON data,"
 218                       "msgid=%d", removedData->messageId);
 219
 220             // callback for retransmit timeout
 221             if (NULL != context->timeoutCallback)
 222             {
 223                 context->timeoutCallback(removedData->endpoint, removedData->pdu,
 224                                          removedData->size);
 225             }
 226
 227             CAFreeEndpoint(removedData->endpoint);
 228             OICFree(removedData->pdu);
 229
 230             OICFree(removedData);
 231
 232             // modify loop value.
 233             len = u_arraylist_length(context->dataList);
 234             --i;
 235         }
 236     }
 237
 238     // mutex unlock
 239     ca_mutex_unlock(context->threadMutex);
 240 }
 241
 242 void CARetransmissionBaseRoutine(void *threadValue)
 243 {
 244     OIC_LOG(DEBUG, TAG, "retransmission main thread start");
 245
 246     CARetransmission_t *context = (CARetransmission_t *) threadValue;
 247
 248     if (NULL == context)
 249     {
 250         OIC_LOG(ERROR, TAG, "thread data passing error");
 251
 252         return;
 253     }
 254
 255 #ifdef SINGLE_THREAD
 256     if (true == context->isStop)
 257     {
 258         OIC_LOG(DEBUG, TAG, "thread stopped");
 259         return;
 260     }
 261     CACheckRetransmissionList(context);
 262 #else
 263
 264     while (!context->isStop)
 265     {
 266         // mutex lock
 267         ca_mutex_lock(context->threadMutex);
 268
 269         if (!context->isStop && u_arraylist_length(context->dataList) <= 0)
 270         {
 271             // if list is empty, thread will wait
 272             OIC_LOG(DEBUG, TAG, "wait..there is no retransmission data.");
 273
 274             // wait
 275             ca_cond_wait(context->threadCond, context->threadMutex);
 276
 277             OIC_LOG(DEBUG, TAG, "wake up..");
 278         }
 279         else if (!context->isStop)
 280         {
 281             // check each RETRANSMISSION_CHECK_PERIOD_SEC time.
 282             OIC_LOG_V(DEBUG, TAG, "wait..(%lld)microseconds",
 283                       RETRANSMISSION_CHECK_PERIOD_SEC * (uint64_t) USECS_PER_SEC);
 284
 285             // wait
 286             uint64_t absTime = RETRANSMISSION_CHECK_PERIOD_SEC * (uint64_t) USECS_PER_SEC;
 287             ca_cond_wait_for(context->threadCond, context->threadMutex, absTime );
 288         }
 289         else
 290         {
 291             // we are stopping, so we want to unlock and finish stopping
 292         }
 293
 294         // mutex unlock
 295         ca_mutex_unlock(context->threadMutex);
 296
 297         // check stop flag
 298         if (context->isStop)
 299         {
 300             continue;
 301         }
 302
 303         CACheckRetransmissionList(context);
 304     }
 305
 306     ca_mutex_lock(context->threadMutex);
 307     ca_cond_signal(context->threadCond);
 308     ca_mutex_unlock(context->threadMutex);
 309
 310 #endif
 311     OIC_LOG(DEBUG, TAG, "retransmission main thread end");
 312
 313 }
 314
 315 CAResult_t CARetransmissionInitialize(CARetransmission_t *context,
 316                                       ca_thread_pool_t handle,
 317                                       CADataSendMethod_t retransmissionSendMethod,
 318                                       CATimeoutCallback_t timeoutCallback,
 319                                       CARetransmissionConfig_t* config)
 320 {
 321     if (NULL == context)
 322     {
 323         OIC_LOG(ERROR, TAG, "thread instance is empty");
 324         return CA_STATUS_INVALID_PARAM;
 325     }
 326 #ifndef SINGLE_THREAD
 327     if (NULL == handle)
 328     {
 329         OIC_LOG(ERROR, TAG, "thread pool handle is empty");
 330         return CA_STATUS_INVALID_PARAM;
 331     }
 332 #endif
 333     OIC_LOG(DEBUG, TAG, "thread initialize");
 334
 335     memset(context, 0, sizeof(CARetransmission_t));
 336
 337     CARetransmissionConfig_t cfg = { .supportType = DEFAULT_RETRANSMISSION_TYPE,
 338                                      .tryingCount = DEFAULT_RETRANSMISSION_COUNT };
 339
 340     if (config)
 341     {
 342         cfg = *config;
 343     }
 344
 345     // set send thread data
 346     context->threadPool = handle;
 347     context->threadMutex = ca_mutex_new();
 348     context->threadCond = ca_cond_new();
 349     context->dataSendMethod = retransmissionSendMethod;
 350     context->timeoutCallback = timeoutCallback;
 351     context->config = cfg;
 352     context->isStop = false;
 353     context->dataList = u_arraylist_create();
 354
 355     return CA_STATUS_OK;
 356 }
 357
 358 CAResult_t CARetransmissionSentData(CARetransmission_t *context,
 359                                     const CAEndpoint_t *endpoint,
 360                                     const void *pdu, uint32_t size)
 361 {
 362     if (NULL == context || NULL == endpoint || NULL == pdu)
 363     {
 364         OIC_LOG(ERROR, TAG, "invalid parameter");
 365         return CA_STATUS_INVALID_PARAM;
 366     }
 367
 368     // #0. check support transport type
 369     if (!(context->config.supportType & endpoint->adapter))
 370     {
 371         OIC_LOG_V(DEBUG, TAG, "not supported transport type=%d", endpoint->adapter);
 372         return CA_NOT_SUPPORTED;
 373     }
 374
 375     // #1. check PDU method type and get message id.
 376     CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
 377     uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
 378
 379     OIC_LOG_V(DEBUG, TAG, "sent pdu, msgtype=%d, msgid=%d", type, messageId);
 380
 381     if (CA_MSG_CONFIRM != type)
 382     {
 383         OIC_LOG(DEBUG, TAG, "not supported message type");
 384         return CA_NOT_SUPPORTED;
 385     }
 386
 387     // create retransmission data
 388     CARetransmissionData_t *retData = (CARetransmissionData_t *) OICCalloc(
 389                                           1, sizeof(CARetransmissionData_t));
 390
 391     if (NULL == retData)
 392     {
 393         OIC_LOG(ERROR, TAG, "memory error");
 394         return CA_MEMORY_ALLOC_FAILED;
 395     }
 396
 397     // copy PDU data
 398     void *pduData = (void *) OICMalloc(size);
 399     if (NULL == pduData)
 400     {
 401         OICFree(retData);
 402         OIC_LOG(ERROR, TAG, "memory error");
 403         return CA_MEMORY_ALLOC_FAILED;
 404     }
 405     memcpy(pduData, pdu, size);
 406
 407     // clone remote endpoint
 408     CAEndpoint_t *remoteEndpoint = CACloneEndpoint(endpoint);
 409     if (NULL == remoteEndpoint)
 410     {
 411         OICFree(retData);
 412         OICFree(pduData);
 413         OIC_LOG(ERROR, TAG, "memory error");
 414         return CA_MEMORY_ALLOC_FAILED;
 415     }
 416
 417     // #2. add additional information. (time stamp, retransmission count...)
 418     retData->timeStamp = getCurrentTimeInMicroSeconds();
 419 #ifndef SINGLE_THREAD
 420     retData->timeout = CAGetTimeoutValue();
 421 #endif
 422     retData->triedCount = 0;
 423     retData->messageId = messageId;
 424     retData->endpoint = remoteEndpoint;
 425     retData->pdu = pduData;
 426     retData->size = size;
 427 #ifndef SINGLE_THREAD
 428     // mutex lock
 429     ca_mutex_lock(context->threadMutex);
 430
 431     uint32_t i = 0;
 432     uint32_t len = u_arraylist_length(context->dataList);
 433
 434     // #3. add data into list
 435     for (i = 0; i < len; i++)
 436     {
 437         CARetransmissionData_t *currData = u_arraylist_get(context->dataList, i);
 438
 439         if (NULL == currData)
 440         {
 441             continue;
 442         }
 443
 444         // found index
 445         if (NULL != currData->endpoint && currData->messageId == messageId
 446             && (currData->endpoint->adapter == endpoint->adapter))
 447         {
 448             OIC_LOG(ERROR, TAG, "Duplicate message ID");
 449
 450             // mutex unlock
 451             ca_mutex_unlock(context->threadMutex);
 452
 453             OICFree(retData);
 454             OICFree(pduData);
 455             OICFree(remoteEndpoint);
 456             return CA_STATUS_FAILED;
 457         }
 458     }
 459
 460     u_arraylist_add(context->dataList, (void *) retData);
 461
 462     // notify the thread
 463     ca_cond_signal(context->threadCond);
 464
 465     // mutex unlock
 466     ca_mutex_unlock(context->threadMutex);
 467
 468 #else
 469     u_arraylist_add(context->dataList, (void *) retData);
 470
 471     CACheckRetransmissionList(context);
 472 #endif
 473     return CA_STATUS_OK;
 474 }
 475
 476 CAResult_t CARetransmissionReceivedData(CARetransmission_t *context,
 477                                         const CAEndpoint_t *endpoint, const void *pdu,
 478                                         uint32_t size, void **retransmissionPdu)
 479 {
 480     OIC_LOG(DEBUG, TAG, "IN");
 481     if (NULL == context || NULL == endpoint || NULL == pdu || NULL == retransmissionPdu)
 482     {
 483         OIC_LOG(ERROR, TAG, "invalid parameter");
 484         return CA_STATUS_INVALID_PARAM;
 485     }
 486
 487     // #0. check support transport type
 488     if (!(context->config.supportType & endpoint->adapter))
 489     {
 490         OIC_LOG_V(DEBUG, TAG, "not supported transport type=%d", endpoint->adapter);
 491         return CA_STATUS_OK;
 492     }
 493
 494     // #1. check PDU method type and get message id.
 495     // ACK, RST --> remove the CON data
 496     CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
 497     uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
 498     CAResponseResult_t code = CAGetCodeFromPduBinaryData(pdu, size);
 499
 500     OIC_LOG_V(DEBUG, TAG, "received pdu, msgtype=%d, msgid=%d, code=%d",
 501               type, messageId, code);
 502
 503     if (((CA_MSG_ACKNOWLEDGE != type) && (CA_MSG_RESET != type))
 504         || (CA_MSG_RESET == type && CA_EMPTY != code))
 505     {
 506         return CA_STATUS_OK;
 507     }
 508
 509     // mutex lock
 510     ca_mutex_lock(context->threadMutex);
 511     uint32_t len = u_arraylist_length(context->dataList);
 512
 513     // find index
 514     uint32_t i;
 515     for (i = 0; i < len; i++)
 516     {
 517         CARetransmissionData_t *retData = (CARetransmissionData_t *) u_arraylist_get(
 518                 context->dataList, i);
 519
 520         if (NULL == retData)
 521         {
 522             continue;
 523         }
 524
 525         // found index
 526         if (NULL != retData->endpoint && retData->messageId == messageId
 527             && (retData->endpoint->adapter == endpoint->adapter))
 528         {
 529             // get pdu data for getting token when CA_EMPTY(RST/ACK) is received from remote device
 530             // if retransmission was finish..token will be unavailable.
 531             if (CA_EMPTY == CAGetCodeFromPduBinaryData(pdu, size))
 532             {
 533                 OIC_LOG(DEBUG, TAG, "code is CA_EMPTY");
 534
 535                 if (NULL == retData->pdu)
 536                 {
 537                     OIC_LOG(ERROR, TAG, "retData->pdu is null");
 538                     OICFree(retData);
 539                     // mutex unlock
 540                     ca_mutex_unlock(context->threadMutex);
 541
 542                     return CA_STATUS_FAILED;
 543                 }
 544
 545                 // copy PDU data
 546                 (*retransmissionPdu) = (void *) OICCalloc(1, retData->size);
 547                 if ((*retransmissionPdu) == NULL)
 548                 {
 549                     OICFree(retData);
 550                     OIC_LOG(ERROR, TAG, "memory error");
 551
 552                     // mutex unlock
 553                     ca_mutex_unlock(context->threadMutex);
 554
 555                     return CA_MEMORY_ALLOC_FAILED;
 556                 }
 557                 memcpy((*retransmissionPdu), retData->pdu, retData->size);
 558             }
 559
 560             // #2. remove data from list
 561             CARetransmissionData_t *removedData = u_arraylist_remove(context->dataList, i);
 562             if (NULL == removedData)
 563             {
 564                 OIC_LOG(ERROR, TAG, "Removed data is NULL");
 565
 566                 // mutex unlock
 567                 ca_mutex_unlock(context->threadMutex);
 568
 569                 return CA_STATUS_FAILED;
 570             }
 571
 572             OIC_LOG_V(DEBUG, TAG, "remove RTCON data!!, msgid=%d", messageId);
 573
 574             CAFreeEndpoint(removedData->endpoint);
 575             OICFree(removedData->pdu);
 576             OICFree(removedData);
 577
 578             break;
 579         }
 580     }
 581
 582     // mutex unlock
 583     ca_mutex_unlock(context->threadMutex);
 584
 585     OIC_LOG(DEBUG, TAG, "OUT");
 586     return CA_STATUS_OK;
 587 }
 588
 589 CAResult_t CARetransmissionStop(CARetransmission_t *context)
 590 {
 591     if (NULL == context)
 592     {
 593         OIC_LOG(ERROR, TAG, "context is empty..");
 594         return CA_STATUS_INVALID_PARAM;
 595     }
 596
 597     OIC_LOG(DEBUG, TAG, "retransmission stop request!!");
 598
 599     // mutex lock
 600     ca_mutex_lock(context->threadMutex);
 601
 602     // set stop flag
 603     context->isStop = true;
 604
 605     // notify the thread
 606     ca_cond_signal(context->threadCond);
 607
 608     ca_cond_wait(context->threadCond, context->threadMutex);
 609
 610     // mutex unlock
 611     ca_mutex_unlock(context->threadMutex);
 612
 613     return CA_STATUS_OK;
 614 }
 615
 616 CAResult_t CARetransmissionDestroy(CARetransmission_t *context)
 617 {
 618     if (NULL == context)
 619     {
 620         OIC_LOG(ERROR, TAG, "context is empty..");
 621         return CA_STATUS_INVALID_PARAM;
 622     }
 623
 624     OIC_LOG(DEBUG, TAG, "retransmission context destroy..");
 625
 626     ca_mutex_free(context->threadMutex);
 627     context->threadMutex = NULL;
 628     ca_cond_free(context->threadCond);
 629     u_arraylist_free(&context->dataList);
 630
 631     return CA_STATUS_OK;
 632 }
 633
 634 uint64_t getCurrentTimeInMicroSeconds()
 635 {
 636     OIC_LOG(DEBUG, TAG, "IN");
 637     uint64_t currentTime = 0;
 638
 639 #ifdef __ANDROID__
 640     struct timespec getTs;
 641
 642     clock_gettime(CLOCK_MONOTONIC, &getTs);
 643
 644     currentTime = (getTs.tv_sec * (uint64_t)1000000000 + getTs.tv_nsec)/1000;
 645     OIC_LOG_V(DEBUG, TAG, "current time = %lld", currentTime);
 646 #elif defined __ARDUINO__
 647     currentTime = millis() * 1000;
 648     OIC_LOG_V(DEBUG, TAG, "currtime=%lu", currentTime);
 649 #else
 650 #if _POSIX_TIMERS > 0
 651     struct timespec ts;
 652     clock_gettime(CLOCK_MONOTONIC, &ts);
 653     currentTime = ts.tv_sec * USECS_PER_SEC + ts.tv_nsec / 1000;
 654 #else
 655     struct timeval tv;
 656     gettimeofday(&tv, NULL);
 657     currentTime = tv.tv_sec * USECS_PER_SEC + tv.tv_usec;
 658 #endif
 659 #endif
 660
 661     OIC_LOG(DEBUG, TAG, "OUT");
 662     return currentTime;
 663 }