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 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <unistd.h>
  25 #include <glib.h>
  26
  27 #ifdef __ANDROID__
  28 #include <linux/time.h>
  29 #endif
  30
  31 #include "caretransmission.h"
  32 #include "caremotehandler.h"
  33 #include "caprotocolmessage.h"
  34 #include "oic_malloc.h"
  35 #include "logger.h"
  36
  37 #define TAG PCF("RET")
  38
  39 typedef struct
  40 {
  41     uint64_t timeStamp;                 /**< last sent time. microseconds */
  42     uint64_t timeout;                   /**< timeout value. microseconds */
  43     uint8_t triedCount;                 /**< retransmission count */
  44     uint16_t messageId;                 /**< coap PDU message id */
  45     CARemoteEndpoint_t *endpoint;       /**< remote endpoint */
  46     void *pdu;                          /**< coap PDU */
  47     uint32_t size;                      /**< coap PDU size */
  48 } CARetransmissionData_t;
  49
  50 /**
  51  * @brief   getCurrent monotonic time
  52  * @return  current time in microseconds
  53  */
  54 uint64_t getCurrentTimeInMicroSeconds();
  55
  56 /**
  57  * @brief   check timeout routine
  58  * @param   currentTime     [IN]microseconds
  59  * @param   timeStamp       [IN]microseconds
  60  * @param   timeoutValue    [IN]microseconds
  61  * @param   triedCount      [IN]
  62  * @return  true if the timeout period has elapsed, false otherwise
  63  */
  64 static bool CACheckTimeout(uint64_t currentTime, uint64_t timeStamp, uint64_t timeoutValue,
  65         uint8_t triedCount)
  66 {
  67     // #1. calculate timeout
  68     uint32_t milliTimeoutValue = timeoutValue * 0.001;
  69     uint64_t timeout = (milliTimeoutValue << triedCount) * (uint64_t) 1000;
  70
  71     if (currentTime >= timeStamp + timeout)
  72     {
  73         OIC_LOG_V(DEBUG, TAG, "%d microseconds time out!!, tried count(%d)", timeout, triedCount);
  74         return true;
  75     }
  76
  77     return false;
  78 }
  79
  80 /**
  81  * @brief   timeout value is
  82  *          between DEFAULT_ACK_TIMEOUT_SEC and
  83  *          (DEFAULT_ACK_TIMEOUT_SEC * DEFAULT_RANDOM_FACTOR) second.
  84  *          DEFAULT_RANDOM_FACTOR       1.5 (CoAP)
  85  * @return  microseconds.
  86  */
  87 static uint64_t CAGetTimeoutValue()
  88 {
  89     return ((DEFAULT_ACK_TIMEOUT_SEC * 1000) + ((1000 * (random() & 0xFF)) >> 8)) *
  90             (uint64_t) 1000;
  91 }
  92
  93 static void CACheckRetransmissionList(CARetransmission_t *context)
  94 {
  95     // mutex lock
  96     u_mutex_lock(context->threadMutex);
  97
  98     int32_t i = 0;
  99     uint32_t len = u_arraylist_length(context->dataList);
 100
 101     for (i = 0; i < len; i++)
 102     {
 103         CARetransmissionData_t *retData = u_arraylist_get(context->dataList, i);
 104
 105         if (retData == NULL)
 106         {
 107             continue;
 108         }
 109
 110         uint64_t currentTime = getCurrentTimeInMicroSeconds();
 111
 112         if (CACheckTimeout(currentTime, retData->timeStamp, retData->timeout, retData->triedCount))
 113         {
 114             // #2. if time's up, send the data.
 115             if (context->dataSendMethod != NULL)
 116             {
 117                 OIC_LOG_V(DEBUG, TAG, "retransmission CON data!!, message id(%d)",
 118                           retData->messageId);
 119                 context->dataSendMethod(retData->endpoint, retData->pdu, retData->size);
 120             }
 121
 122             // #3. increase the retransmission count and update timestamp.
 123             retData->timeStamp = currentTime;
 124             retData->triedCount++;
 125         }
 126
 127         // #4. if tried count is max, remove the retransmission data from list.
 128         if (retData->triedCount >= context->config.tryingCount)
 129         {
 130             CARetransmissionData_t *removedData = u_arraylist_remove(context->dataList, i);
 131
 132             if (removedData != NULL)
 133             {
 134                 OIC_LOG_V(DEBUG, TAG, "max trying count, remove retransmission CON data!!,\
 135                         message id(%d)", removedData->messageId);
 136
 137                 // callback for retransmit timeout
 138                 if (context->timeoutCallback != NULL)
 139                 {
 140                     context->timeoutCallback(removedData->endpoint, removedData->pdu,
 141                             removedData->size);
 142                 }
 143
 144                 CADestroyRemoteEndpointInternal(removedData->endpoint);
 145                 OICFree(removedData->pdu);
 146
 147                 OICFree(removedData);
 148
 149                 // modify loop value.
 150                 len = u_arraylist_length(context->dataList);
 151
 152                 --i;
 153             }
 154             else
 155             {
 156                 OIC_LOG_V(ERROR, TAG, "arraylist remove error");
 157             }
 158
 159         }
 160     }
 161
 162     // mutex unlock
 163     u_mutex_unlock(context->threadMutex);
 164 }
 165
 166 static void CARetransmissionBaseRoutine(void *threadValue)
 167 {
 168     OIC_LOG_V(DEBUG, TAG, "retransmission main thread start..");
 169
 170     CARetransmission_t *context = (CARetransmission_t *) threadValue;
 171
 172     if (context == NULL)
 173     {
 174         OIC_LOG_V(ERROR, TAG, "thread data passing error!!");
 175
 176         return;
 177     }
 178
 179     while (!context->isStop)
 180     {
 181         // mutex lock
 182         u_mutex_lock(context->threadMutex);
 183
 184         if (u_arraylist_length(context->dataList) <= 0)
 185         {
 186             // if list is empty, thread will wait
 187             OIC_LOG_V(DEBUG, TAG, "wait..there is no retransmission data.");
 188
 189             // wait
 190             u_cond_wait(context->threadCond, context->threadMutex);
 191
 192             OIC_LOG_V(DEBUG, TAG, "wake up..");
 193         }
 194         else
 195         {
 196             // check each RETRANSMISSION_CHECK_PERIOD_SEC time.
 197             OIC_LOG_V(DEBUG, TAG, "wait..(%d)microseconds",
 198                         RETRANSMISSION_CHECK_PERIOD_SEC * (uint64_t) 1000000);
 199
 200             // wait
 201             u_cond_wait_until(context->threadCond, context->threadMutex,
 202                               RETRANSMISSION_CHECK_PERIOD_SEC * (uint64_t) 1000000);
 203         }
 204
 205         // mutex unlock
 206         u_mutex_unlock(context->threadMutex);
 207
 208         // check stop flag
 209         if (context->isStop)
 210         {
 211             continue;
 212         }
 213
 214         CACheckRetransmissionList(context);
 215     }
 216
 217     u_cond_signal(context->threadCond);
 218
 219     OIC_LOG_V(DEBUG, TAG, "retransmission main thread end..");
 220
 221 }
 222
 223 CAResult_t CARetransmissionInitialize(CARetransmission_t *context, u_thread_pool_t handle,
 224         CADataSendMethod_t retransmissionSendMethod, CATimeoutCallback_t timeoutCallback,
 225         CARetransmissionConfig_t* config)
 226 {
 227     if (context == NULL)
 228     {
 229         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 230         return CA_STATUS_FAILED;
 231     }
 232
 233     if (handle == NULL)
 234     {
 235         OIC_LOG_V(DEBUG, TAG, "thread pool handle is empty..");
 236         return CA_STATUS_FAILED;
 237     }
 238
 239     OIC_LOG_V(DEBUG, TAG, "thread initialize..");
 240
 241     memset(context, 0, sizeof(CARetransmission_t));
 242
 243     CARetransmissionConfig_t cfg = { 0 };
 244
 245     if (config == NULL)
 246     {
 247         // setDefault
 248         cfg.supportType = DEFAULT_RETRANSMISSION_TYPE;
 249         cfg.tryingCount = DEFAULT_MAX_RETRANSMIT;
 250     }
 251     else
 252     {
 253         cfg = *config;
 254     }
 255
 256     // set send thread data
 257     context->threadPool = handle;
 258     context->threadMutex = u_mutex_new();
 259     context->threadCond = u_cond_new();
 260     context->dataSendMethod = retransmissionSendMethod;
 261     context->timeoutCallback = timeoutCallback;
 262     context->config = cfg;
 263     context->isStop = false;
 264     context->dataList = u_arraylist_create();
 265
 266     return CA_STATUS_OK;
 267 }
 268
 269 CAResult_t CARetransmissionStart(CARetransmission_t *context)
 270 {
 271     if (context == NULL)
 272     {
 273         OIC_LOG_V(DEBUG, TAG, "context is empty..");
 274         return CA_STATUS_FAILED;
 275     }
 276
 277     if (context->threadPool == NULL)
 278     {
 279         OIC_LOG_V(DEBUG, TAG, "thread pool handle is empty..");
 280         return CA_STATUS_FAILED;
 281     }
 282
 283     CAResult_t res = u_thread_pool_add_task(context->threadPool, CARetransmissionBaseRoutine,
 284                                             context);
 285
 286     if (res != CA_STATUS_OK)
 287     {
 288         OIC_LOG_V(ERROR, TAG, "thread pool add task error(send thread).");
 289         return res;
 290     }
 291
 292     return res;
 293 }
 294
 295 CAResult_t CARetransmissionSentData(CARetransmission_t *context,
 296     const CARemoteEndpoint_t* endpoint, const void* pdu, uint32_t size)
 297 {
 298     if (context == NULL || endpoint == NULL || pdu == NULL)
 299     {
 300         OIC_LOG_V(DEBUG, TAG, "invalid parameter..");
 301         return CA_STATUS_INVALID_PARAM;
 302     }
 303
 304     // #0. check support connectivity type
 305     if (!(context->config.supportType & endpoint->connectivityType))
 306     {
 307         OIC_LOG_V(DEBUG, TAG, "not supported connectivity type for retransmission..(%d)",
 308                   endpoint->connectivityType);
 309         return CA_STATUS_OK;
 310     }
 311
 312     // #1. check PDU method type and get message id.
 313     CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
 314     uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
 315
 316     OIC_LOG_V(DEBUG, TAG, "sent pdu, message type(%d), message id(%d)", type, messageId);
 317
 318     if (type != CA_MSG_CONFIRM)
 319     {
 320         return CA_STATUS_OK;
 321     }
 322
 323     // create retransmission data
 324     CARetransmissionData_t *retData = (CARetransmissionData_t *) OICCalloc(
 325                                           1, sizeof(CARetransmissionData_t));
 326
 327     if (retData == NULL)
 328     {
 329         OIC_LOG_V(ERROR, TAG, "memory error!!");
 330         return CA_MEMORY_ALLOC_FAILED;
 331     }
 332
 333     // copy PDU data
 334     void *pduData = (void *) OICMalloc(sizeof(int8_t) * size);
 335     if (pduData == NULL)
 336     {
 337         OICFree(retData);
 338         OIC_LOG_V(ERROR, TAG, "memory error!!");
 339         return CA_MEMORY_ALLOC_FAILED;
 340     }
 341     memcpy(pduData, pdu, sizeof(int8_t) * size);
 342
 343     // clone remote endpoint
 344     CARemoteEndpoint_t *remoteEndpoint = CACloneRemoteEndpoint(endpoint);
 345     if (remoteEndpoint == NULL)
 346     {
 347         OICFree(retData);
 348         OICFree(pduData);
 349         OIC_LOG_V(ERROR, TAG, "memory error!!");
 350         return CA_MEMORY_ALLOC_FAILED;
 351     }
 352
 353     // #2. add additional information. (time stamp, retransmission count...)
 354     retData->timeStamp = getCurrentTimeInMicroSeconds();
 355     retData->timeout = CAGetTimeoutValue();
 356     retData->triedCount = 0;
 357     retData->messageId = messageId;
 358     retData->endpoint = remoteEndpoint;
 359     retData->pdu = pduData;
 360     retData->size = size;
 361
 362     // mutex lock
 363     u_mutex_lock(context->threadMutex);
 364
 365     uint32_t i = 0;
 366     uint32_t len = u_arraylist_length(context->dataList);
 367
 368     // #3. add data into list
 369     for (i = 0; i < len; i++)
 370     {
 371         CARetransmissionData_t *currData = u_arraylist_get(context->dataList, i);
 372
 373         if (currData == NULL)
 374         {
 375             continue;
 376         }
 377
 378         // found index
 379         if (retData != NULL && (currData->endpoint->connectivityType == endpoint->connectivityType)
 380             && currData->messageId == messageId)
 381         {
 382             OIC_LOG_V(ERROR, TAG, "Duplicate message ID");
 383
 384             // mutex unlock
 385             u_mutex_unlock(context->threadMutex);
 386
 387             OICFree(retData);
 388             OICFree(pduData);
 389             OICFree(remoteEndpoint);
 390             return CA_STATUS_FAILED;
 391         }
 392     }
 393
 394     u_arraylist_add(context->dataList, (void *) retData);
 395
 396     // notity the thread
 397     u_cond_signal(context->threadCond);
 398
 399     // mutex unlock
 400     u_mutex_unlock(context->threadMutex);
 401
 402     return CA_STATUS_OK;
 403 }
 404
 405 CAResult_t CARetransmissionReceivedData(CARetransmission_t *context,
 406                                         const CARemoteEndpoint_t *endpoint, const void *pdu,
 407                                         uint32_t size)
 408 {
 409     if (context == NULL || endpoint == NULL || pdu == NULL)
 410     {
 411         OIC_LOG_V(DEBUG, TAG, "invalid parameter..");
 412         return CA_STATUS_INVALID_PARAM;
 413     }
 414
 415     // #0. check support connectivity type
 416     if (!(context->config.supportType & endpoint->connectivityType))
 417     {
 418         OIC_LOG_V(DEBUG, TAG, "not supported connectivity type for retransmission..(%d)",
 419                   endpoint->connectivityType);
 420         return CA_STATUS_OK;
 421     }
 422
 423     // #1. check PDU method type and get message id.
 424     // ACK, RST --> remove the CON data
 425     CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
 426     uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
 427
 428     OIC_LOG_V(DEBUG, TAG, "received pdu, message type(%d), message id(%d)", type, messageId);
 429
 430     if (type != CA_MSG_ACKNOWLEDGE && type != CA_MSG_RESET)
 431     {
 432         return CA_STATUS_OK;
 433     }
 434
 435     // mutex lock
 436     u_mutex_lock(context->threadMutex);
 437
 438     uint32_t i = 0;
 439     uint32_t len = u_arraylist_length(context->dataList);
 440
 441     // find index
 442     for (i = 0; i < len; i++)
 443     {
 444         CARetransmissionData_t *retData = u_arraylist_get(context->dataList, i);
 445
 446         // found index
 447         if (retData != NULL && (retData->endpoint->connectivityType == endpoint->connectivityType)
 448             && retData->messageId == messageId)
 449         {
 450             // #2. remove data from list
 451             CARetransmissionData_t *removedData = u_arraylist_remove(context->dataList, i);
 452
 453             if (removedData != NULL)
 454             {
 455                 OIC_LOG_V(DEBUG, TAG, "remove retransmission CON data!!, message id(%d)",
 456                         messageId);
 457
 458                 CADestroyRemoteEndpointInternal(removedData->endpoint);
 459                 OICFree(removedData->pdu);
 460
 461                 OICFree(removedData);
 462             }
 463             break;
 464         }
 465     }
 466
 467     // mutex unlock
 468     u_mutex_unlock(context->threadMutex);
 469
 470     return CA_STATUS_OK;
 471 }
 472
 473 CAResult_t CARetransmissionStop(CARetransmission_t *context)
 474 {
 475     if (context == NULL)
 476     {
 477         OIC_LOG_V(DEBUG, TAG, "context is empty..");
 478         return CA_STATUS_FAILED;
 479     }
 480
 481     OIC_LOG_V(DEBUG, TAG, "retransmission stop request!!");
 482
 483     // mutex lock
 484     u_mutex_lock(context->threadMutex);
 485
 486     // set stop flag
 487     context->isStop = true;
 488
 489     // notity the thread
 490     u_cond_signal(context->threadCond);
 491
 492     u_cond_wait(context->threadCond, context->threadMutex);
 493
 494     // mutex unlock
 495     u_mutex_unlock(context->threadMutex);
 496
 497     return CA_STATUS_OK;
 498 }
 499
 500 CAResult_t CARetransmissionDestroy(CARetransmission_t *context)
 501 {
 502     if (context == NULL)
 503     {
 504         OIC_LOG_V(DEBUG, TAG, "context is empty..");
 505         return CA_STATUS_FAILED;
 506     }
 507
 508     OIC_LOG_V(DEBUG, TAG, "retransmission context destroy..");
 509
 510     u_mutex_free(context->threadMutex);
 511     context->threadMutex = NULL;
 512     u_cond_free(context->threadCond);
 513     u_arraylist_free(&context->dataList);
 514
 515     return CA_STATUS_OK;
 516 }
 517
 518 uint64_t getCurrentTimeInMicroSeconds()
 519 {
 520     uint64_t currentTime = 0;
 521
 522 #ifdef __ANDROID__
 523     struct timespec getTs;
 524
 525     clock_gettime(CLOCK_MONOTONIC, &getTs);
 526
 527     currentTime = (getTs.tv_sec * (uint64_t)1000000000 + getTs.tv_nsec)/1000;
 528     OIC_LOG_V(DEBUG, TAG, "current time = %d", currentTime);
 529 #else
 530     currentTime = g_get_monotonic_time();
 531 #endif
 532     return currentTime;
 533 }
 534