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