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