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