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