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