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