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