• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4965

09 Feb 2026 01:16AM UTC coverage: 66.884% (-0.02%) from 66.908%
#4965

push

travis-ci

web-flow
docs: add support for recording STMT to CSV files (#34276)

* docs: add support for recording STMT to CSV files

* docs: update version for STMT recording feature in CSV files

205798 of 307696 relevant lines covered (66.88%)

126382200.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.22
/source/util/src/ttimer.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#define _DEFAULT_SOURCE
17
#include "ttimer.h"
18
#include "taoserror.h"
19
#include "tdef.h"
20
#include "tlog.h"
21
#include "tsched.h"
22

23
// clang-format off
24
#define tmrFatal(...) { if (tmrDebugFlag & DEBUG_FATAL) { taosPrintLog("TMR FATAL ", DEBUG_FATAL, tmrDebugFlag, __VA_ARGS__); }}
25
#define tmrError(...) { if (tmrDebugFlag & DEBUG_ERROR) { taosPrintLog("TMR ERROR ", DEBUG_ERROR, tmrDebugFlag, __VA_ARGS__); }}
26
#define tmrWarn(...)  { if (tmrDebugFlag & DEBUG_WARN)  { taosPrintLog("TMR WARN  ", DEBUG_WARN,  tmrDebugFlag, __VA_ARGS__); }}
27
#define tmrInfo(...)  { if (tmrDebugFlag & DEBUG_INFO)  { taosPrintLog("TMR INFO  ", DEBUG_INFO,  tmrDebugFlag, __VA_ARGS__); }}
28
#define tmrDebug(...) { if (tmrDebugFlag & DEBUG_DEBUG) { taosPrintLog("TMR DEBUG ", DEBUG_DEBUG, tmrDebugFlag, __VA_ARGS__); }}
29
#define tmrTrace(...) { if (tmrDebugFlag & DEBUG_TRACE) { taosPrintLog("TMR TRACE ", DEBUG_TRACE, tmrDebugFlag, __VA_ARGS__); }}
30
// clang-format on
31

32
#define TIMER_STATE_WAITING  0
33
#define TIMER_STATE_EXPIRED  1
34
#define TIMER_STATE_STOPPED  2
35
#define TIMER_STATE_CANCELED 3
36

37
typedef union _tmr_ctrl_t {
38
  char label[16];
39
  struct {
40
    // pad to ensure 'next' is the end of this union
41
    char               padding[16 - sizeof(union _tmr_ctrl_t*)];
42
    union _tmr_ctrl_t* next;
43
  };
44
} tmr_ctrl_t;
45

46
typedef struct tmr_obj_t {
47
  uintptr_t         id;
48
  tmr_ctrl_t*       ctrl;
49
  struct tmr_obj_t* mnext;
50
  struct tmr_obj_t* prev;
51
  struct tmr_obj_t* next;
52
  uint16_t          slot;
53
  uint8_t           wheel;
54
  uint8_t           state;
55
  uint8_t           refCount;
56
  uint8_t           reserved1;
57
  uint16_t          reserved2;
58
  union {
59
    int64_t expireAt;
60
    int64_t executedBy;
61
  };
62
  TAOS_TMR_CALLBACK fp;
63
  void*             param;
64
  uint8_t           priority;
65
} tmr_obj_t;
66

67
typedef struct timer_list_t {
68
  int64_t    lockedBy;
69
  tmr_obj_t* timers;
70
} timer_list_t;
71

72
typedef struct timer_map_t {
73
  uint32_t      size;
74
  uint32_t      count;
75
  timer_list_t* slots;
76
} timer_map_t;
77

78
typedef struct time_wheel_t {
79
  TdThreadMutex mutex;
80
  int64_t       nextScanAt;
81
  uint32_t      resolution;
82
  uint16_t      size;
83
  uint16_t      index;
84
  tmr_obj_t**   slots;
85
} time_wheel_t;
86

87
static int32_t tsMaxTmrCtrl = TSDB_MAX_VNODES_PER_DB + 100;
88

89
static int32_t       tmrModuleInit = 0;
90
static TdThreadMutex tmrCtrlMutex;
91
static tmr_ctrl_t*   tmrCtrls;
92
static tmr_ctrl_t*   unusedTmrCtrl = NULL;
93
static void*         tmrQhandle;
94
static void*         tmrQhandleHigh;
95
static int32_t       numOfTmrCtrl = 0;
96

97
int32_t          taosTmrThreads = 1;
98
static uintptr_t nextTimerId = 0;
99

100
static time_wheel_t wheels[] = {
101
    {.resolution = MSECONDS_PER_TICK, .size = 4096},
102
    {.resolution = 1000, .size = 1024},
103
    {.resolution = 60000, .size = 1024},
104
};
105
static timer_map_t timerMap;
106

107
static uintptr_t getNextTimerId() {
343,582,825✔
108
  uintptr_t id;
109
  do {
110
    id = (uintptr_t)atomic_add_fetch_ptr((void**)&nextTimerId, 1);
343,582,825✔
111
  } while (id == 0);
343,581,496✔
112
  return id;
343,579,925✔
113
}
114

115
static void timerAddRef(tmr_obj_t* timer) {
1,024,376,988✔
116
  (void)atomic_add_fetch_8(&timer->refCount, 1);
1,024,376,988✔
117
}
1,024,381,876✔
118

119
static void timerDecRef(tmr_obj_t* timer) {
1,023,073,485✔
120
  if (atomic_sub_fetch_8(&timer->refCount, 1) == 0) {
1,023,073,485✔
121
    taosMemoryFree(timer);
342,926,318✔
122
  }
123
}
1,023,073,920✔
124

125
static void lockTimerList(timer_list_t* list) {
1,030,199,167✔
126
  int64_t tid = taosGetSelfPthreadId();
1,030,199,167✔
127
  int32_t i = 0;
1,030,199,872✔
128
  while (atomic_val_compare_exchange_64(&(list->lockedBy), 0, tid) != 0) {
1,030,199,872✔
129
    if (++i % 1000 == 0) {
×
130
      (void)sched_yield();
×
131
    }
132
  }
133
}
1,030,198,812✔
134

135
static void unlockTimerList(timer_list_t* list) {
1,030,199,601✔
136
  int64_t tid = taosGetSelfPthreadId();
1,030,199,601✔
137
  if (atomic_val_compare_exchange_64(&(list->lockedBy), tid, 0) != tid) {
1,030,199,414✔
138
    uError("%" PRId64 " trying to unlock a timer list not locked by current thread.", tid);
×
139
  }
140
}
1,030,197,650✔
141

142
static void addTimer(tmr_obj_t* timer) {
343,580,864✔
143
  timerAddRef(timer);
343,580,864✔
144
  timer->wheel = tListLen(wheels);
343,580,976✔
145

146
  uint32_t      idx = (uint32_t)(timer->id % timerMap.size);
343,580,976✔
147
  timer_list_t* list = timerMap.slots + idx;
343,581,335✔
148

149
  lockTimerList(list);
343,580,996✔
150
  timer->mnext = list->timers;
343,579,975✔
151
  list->timers = timer;
343,580,976✔
152
  unlockTimerList(list);
343,581,663✔
153
}
343,581,385✔
154

155
static tmr_obj_t* findTimer(uintptr_t id) {
487,133,514✔
156
  tmr_obj_t* timer = NULL;
487,133,514✔
157
  if (id > 0) {
487,133,514✔
158
    uint32_t      idx = (uint32_t)(id % timerMap.size);
343,692,354✔
159
    timer_list_t* list = timerMap.slots + idx;
343,692,354✔
160
    lockTimerList(list);
343,691,930✔
161
    for (timer = list->timers; timer != NULL; timer = timer->mnext) {
343,692,048✔
162
      if (timer->id == id) {
337,219,596✔
163
        timerAddRef(timer);
337,219,736✔
164
        break;
337,220,421✔
165
      }
166
    }
167
    unlockTimerList(list);
343,693,161✔
168
  }
169
  return timer;
487,132,256✔
170
}
171

172
static void removeTimer(uintptr_t id) {
342,927,709✔
173
  tmr_obj_t*    prev = NULL;
342,927,709✔
174
  uint32_t      idx = (uint32_t)(id % timerMap.size);
342,927,709✔
175
  timer_list_t* list = timerMap.slots + idx;
342,927,709✔
176
  lockTimerList(list);
342,927,709✔
177
  for (tmr_obj_t* p = list->timers; p != NULL; p = p->mnext) {
342,927,709✔
178
    if (p->id == id) {
342,927,709✔
179
      if (prev == NULL) {
342,927,438✔
180
        list->timers = p->mnext;
342,927,438✔
181
      } else {
182
        prev->mnext = p->mnext;
×
183
      }
184
      timerDecRef(p);
342,927,709✔
185
      break;
342,927,709✔
186
    }
187
    prev = p;
271✔
188
  }
189
  unlockTimerList(list);
342,927,709✔
190
}
342,927,438✔
191

192
static void addToWheel(tmr_obj_t* timer, uint32_t delay) {
343,579,767✔
193
  timerAddRef(timer);
343,579,767✔
194
  // select a wheel for the timer, we are not an accurate timer,
195
  // but the inaccuracy should not be too large.
196
  timer->wheel = tListLen(wheels) - 1;
343,581,663✔
197
  for (uint8_t i = 0; i < tListLen(wheels); i++) {
350,305,828✔
198
    time_wheel_t* wheel = wheels + i;
350,304,341✔
199
    if (delay < wheel->resolution * wheel->size) {
350,304,074✔
200
      timer->wheel = i;
343,580,854✔
201
      break;
343,581,126✔
202
    }
203
  }
204

205
  time_wheel_t* wheel = wheels + timer->wheel;
343,582,613✔
206
  timer->prev = NULL;
343,580,930✔
207
  timer->expireAt = taosGetMonotonicMs() + delay;
343,579,877✔
208

209
  (void)taosThreadMutexLock(&wheel->mutex);
343,581,419✔
210

211
  uint32_t idx = 0;
343,581,663✔
212
  if (timer->expireAt > wheel->nextScanAt) {
343,581,663✔
213
    // adjust delay according to next scan time of this wheel
214
    // so that the timer is not fired earlier than desired.
215
    delay = (uint32_t)(timer->expireAt - wheel->nextScanAt);
343,581,663✔
216
    idx = (delay + wheel->resolution - 1) / wheel->resolution;
343,581,663✔
217
  }
218

219
  timer->slot = (uint16_t)((wheel->index + idx + 1) % wheel->size);
343,581,663✔
220
  tmr_obj_t* p = wheel->slots[timer->slot];
343,581,663✔
221
  wheel->slots[timer->slot] = timer;
343,581,663✔
222
  timer->next = p;
343,581,663✔
223
  if (p != NULL) {
343,581,663✔
224
    p->prev = timer;
24,945,617✔
225
  }
226

227
  (void)taosThreadMutexUnlock(&wheel->mutex);
343,581,663✔
228
}
343,581,663✔
229

230
static bool removeFromWheel(tmr_obj_t* timer) {
127,937,921✔
231
  uint8_t wheelIdx = timer->wheel;
127,937,921✔
232
  if (wheelIdx >= tListLen(wheels)) {
127,938,886✔
233
    return false;
2,202✔
234
  }
235
  time_wheel_t* wheel = wheels + wheelIdx;
127,936,684✔
236

237
  bool removed = false;
127,935,455✔
238
  (void)taosThreadMutexLock(&wheel->mutex);
127,935,455✔
239
  // other thread may modify timer->wheel, check again.
240
  if (timer->wheel < tListLen(wheels)) {
127,937,579✔
241
    if (timer->prev != NULL) {
127,937,579✔
242
      timer->prev->next = timer->next;
1,140,947✔
243
    }
244
    if (timer->next != NULL) {
127,937,579✔
245
      timer->next->prev = timer->prev;
2,286,194✔
246
    }
247
    if (timer == wheel->slots[timer->slot]) {
127,937,579✔
248
      wheel->slots[timer->slot] = timer->next;
126,796,632✔
249
    }
250
    timer->wheel = tListLen(wheels);
127,937,579✔
251
    timer->next = NULL;
127,937,579✔
252
    timer->prev = NULL;
127,937,579✔
253
    timerDecRef(timer);
127,937,579✔
254
    removed = true;
127,937,579✔
255
  }
256
  (void)taosThreadMutexUnlock(&wheel->mutex);
127,937,579✔
257

258
  return removed;
127,937,579✔
259
}
260

261
static void processExpiredTimer(void* handle, void* arg) {
214,988,374✔
262
  tmr_obj_t* timer = (tmr_obj_t*)handle;
214,988,374✔
263
  timer->executedBy = taosGetSelfPthreadId();
214,988,374✔
264
  uint8_t state = atomic_val_compare_exchange_8(&timer->state, TIMER_STATE_WAITING, TIMER_STATE_EXPIRED);
214,986,196✔
265
  if (state == TIMER_STATE_WAITING) {
214,986,154✔
266
    const char* fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] execution start.";
214,985,818✔
267
    tmrDebug(fmt, timer->ctrl->label, timer->id, timer->fp, timer->param);
214,985,818✔
268

269
    (*timer->fp)(timer->param, (tmr_h)timer->id);
214,985,818✔
270
    atomic_store_8(&timer->state, TIMER_STATE_STOPPED);
214,987,928✔
271

272
    fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] execution end.";
214,987,928✔
273
    tmrDebug(fmt, timer->ctrl->label, timer->id, timer->fp, timer->param);
214,987,928✔
274
  }
275
  removeTimer(timer->id);
214,988,264✔
276
  timerDecRef(timer);
214,990,130✔
277
}
214,990,130✔
278

279
static void addToExpired(tmr_obj_t* head) {
2,147,483,647✔
280
  const char* fmt = "%s adding expired timer[id=%" PRIuPTR ", fp=%p, param=%p] to queue.";
2,147,483,647✔
281

282
  while (head != NULL) {
2,147,483,647✔
283
    uintptr_t  id = head->id;
214,990,130✔
284
    tmr_obj_t* next = head->next;
214,990,130✔
285
    tmrDebug(fmt, head->ctrl->label, id, head->fp, head->param);
214,990,130✔
286

287
    SSchedMsg schedMsg;
214,724,112✔
288
    schedMsg.fp = NULL;
214,990,130✔
289
    schedMsg.tfp = processExpiredTimer;
214,990,130✔
290
    schedMsg.msg = NULL;
214,990,130✔
291
    schedMsg.ahandle = head;
214,990,130✔
292
    schedMsg.thandle = NULL;
214,990,130✔
293
    uint8_t priority = head->priority;
214,990,130✔
294

295
    if (priority == 1) {
214,990,130✔
296
      if (taosScheduleTask(tmrQhandle, &schedMsg) != 0) {
188,947,178✔
297
        tmrError("%s failed to add expired timer[id=%" PRIuPTR "] to queue.", head->ctrl->label, id);
×
298
      }
299
    } else if (priority == 2) {
26,042,952✔
300
      if (taosScheduleTask(tmrQhandleHigh, &schedMsg) != 0) {
26,042,952✔
301
        tmrError("%s failed to add expired timer[id=%" PRIuPTR "] to high level queue.", head->ctrl->label, id);
×
302
      }
303
    }
304
    else{
305
      tmrError("%s invalid priority level %d for timer[id=%" PRIuPTR "].", head->ctrl->label, priority, id);
×
306
    }
307

308
    tmrDebug("timer[id=%" PRIuPTR "] has been added to queue priority:%d.", id, priority);
214,990,130✔
309
    head = next;
214,990,130✔
310
  }
311
}
2,147,483,647✔
312

313
static uintptr_t doStartTimer(tmr_obj_t* timer, TAOS_TMR_CALLBACK fp, int32_t mseconds, void* param, tmr_ctrl_t* ctrl,
343,580,972✔
314
                              uint8_t priority) {
315
  uintptr_t id = getNextTimerId();
343,580,972✔
316
  timer->id = id;
343,579,925✔
317
  timer->state = TIMER_STATE_WAITING;
343,579,925✔
318
  timer->fp = fp;
343,580,648✔
319
  timer->param = param;
343,581,531✔
320
  timer->ctrl = ctrl;
343,580,504✔
321
  timer->priority = priority;
343,581,203✔
322
  addTimer(timer);
343,580,273✔
323

324
  const char* fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] started";
343,581,385✔
325
  tmrDebug(fmt, ctrl->label, timer->id, timer->fp, timer->param);
343,581,385✔
326

327
  if (mseconds == 0) {
343,579,713✔
328
    timer->wheel = tListLen(wheels);
×
329
    timerAddRef(timer);
×
330
    addToExpired(timer);
×
331
  } else {
332
    addToWheel(timer, mseconds);
343,579,713✔
333
  }
334

335
  // note: use `timer->id` here is unsafe as `timer` may already be freed
336
  return id;
343,581,663✔
337
}
338

339
tmr_h taosTmrStartPriority(TAOS_TMR_CALLBACK fp, int32_t mseconds, void* param, void* handle, uint8_t priority) {
343,579,979✔
340
  tmr_ctrl_t* ctrl = (tmr_ctrl_t*)handle;
343,579,979✔
341
  if (ctrl == NULL || ctrl->label[0] == 0) {
343,579,979✔
342
    return NULL;
×
343
  }
344

345
  tmr_obj_t* timer = (tmr_obj_t*)taosMemoryCalloc(1, sizeof(tmr_obj_t));
343,580,272✔
346
  if (timer == NULL) {
343,577,967✔
347
    tmrError("%s failed to allocated memory for new timer object.", ctrl->label);
×
348
    return NULL;
×
349
  }
350

351
  return (tmr_h)doStartTimer(timer, fp, mseconds, param, ctrl, priority);
343,577,967✔
352
}
353

354
tmr_h taosTmrStart(TAOS_TMR_CALLBACK fp, int32_t mseconds, void* param, void* handle) {
11,100,070✔
355
  return taosTmrStartPriority(fp, mseconds, param, handle, 1);
11,100,070✔
356
}
357

358
static void taosTimerLoopFunc(int32_t signo) {
2,147,483,647✔
359
  int64_t now = taosGetMonotonicMs();
2,147,483,647✔
360

361
  for (int32_t i = 0; i < tListLen(wheels); i++) {
2,147,483,647✔
362
    // `expried` is a temporary expire list.
363
    // expired timers are first add to this list, then move
364
    // to expired queue as a batch to improve performance.
365
    // note this list is used as a stack in this function.
366
    tmr_obj_t* expired = NULL;
2,147,483,647✔
367

368
    time_wheel_t* wheel = wheels + i;
2,147,483,647✔
369
    while (now >= wheel->nextScanAt) {
2,147,483,647✔
370
      (void)taosThreadMutexLock(&wheel->mutex);
2,147,483,647✔
371
      wheel->index = (wheel->index + 1) % wheel->size;
2,147,483,647✔
372
      tmr_obj_t* timer = wheel->slots[wheel->index];
2,147,483,647✔
373
      while (timer != NULL) {
2,147,483,647✔
374
        tmr_obj_t* next = timer->next;
214,990,130✔
375
        if (now < timer->expireAt) {
214,990,130✔
376
          timer = next;
×
377
          continue;
×
378
        }
379

380
        // remove from the wheel
381
        if (timer->prev == NULL) {
214,990,130✔
382
          wheel->slots[wheel->index] = next;
214,990,130✔
383
          if (next != NULL) {
214,990,130✔
384
            next->prev = NULL;
21,450,496✔
385
          }
386
        } else {
387
          timer->prev->next = next;
×
388
          if (next != NULL) {
×
389
            next->prev = timer->prev;
×
390
          }
391
        }
392
        timer->wheel = tListLen(wheels);
214,990,130✔
393

394
        // add to temporary expire list
395
        timer->next = expired;
214,990,130✔
396
        timer->prev = NULL;
214,990,130✔
397
        if (expired != NULL) {
214,990,130✔
398
          expired->prev = timer;
21,720,851✔
399
        }
400
        expired = timer;
214,990,130✔
401

402
        timer = next;
214,990,130✔
403
      }
404
      wheel->nextScanAt += wheel->resolution;
2,147,483,647✔
405
      (void)taosThreadMutexUnlock(&wheel->mutex);
2,147,483,647✔
406
    }
407

408
    addToExpired(expired);
2,147,483,647✔
409
  }
410
}
2,147,483,647✔
411

412
static bool doStopTimer(tmr_obj_t* timer, uint8_t state) {
337,218,561✔
413
  if (state == TIMER_STATE_WAITING) {
337,218,561✔
414
    bool reusable = false;
127,937,921✔
415
    if (removeFromWheel(timer)) {
127,937,921✔
416
      removeTimer(timer->id);
127,937,579✔
417
      // only safe to reuse the timer when timer is removed from the wheel.
418
      // we cannot guarantee the thread safety of the timr in all other cases.
419
      reusable = true;
127,937,308✔
420
    }
421
    const char* fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] is cancelled.";
127,939,510✔
422
    tmrDebug(fmt, timer->ctrl->label, timer->id, timer->fp, timer->param);
127,939,510✔
423
    return reusable;
127,939,510✔
424
  }
425

426
  if (state != TIMER_STATE_EXPIRED) {
209,280,640✔
427
    // timer already stopped or cancelled, has nothing to do in this case
428
    return false;
335✔
429
  }
430

431
  if (timer->executedBy == taosGetSelfPthreadId()) {
209,280,305✔
432
    // taosTmrReset is called in the timer callback, should do nothing in this
433
    // case to avoid dead lock. note taosTmrReset must be the last statement
434
    // of the callback funtion, will be a bug otherwise.
435
    return false;
209,276,457✔
436
  }
437

438
  // timer callback is executing in another thread, we SHOULD wait it to stop,
439
  // BUT this may result in dead lock if current thread are holding a lock which
440
  // the timer callback need to acquire. so, we HAVE TO return directly.
441
  const char* fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] is executing and cannot be stopped.";
3,848✔
442
  tmrDebug(fmt, timer->ctrl->label, timer->id, timer->fp, timer->param);
3,848✔
443
  return false;
3,848✔
444
}
445

446
bool taosTmrStop(tmr_h timerId) {
154,631,191✔
447
  uintptr_t id = (uintptr_t)timerId;
154,631,191✔
448

449
  tmr_obj_t* timer = findTimer(id);
154,631,191✔
450
  if (timer == NULL) {
154,629,984✔
451
    tmrDebug("timer[id=%" PRIuPTR "] does not exist", id);
26,692,130✔
452
    return false;
26,691,318✔
453
  }
454

455
  uint8_t state = atomic_val_compare_exchange_8(&timer->state, TIMER_STATE_WAITING, TIMER_STATE_CANCELED);
127,937,854✔
456
  (void)doStopTimer(timer, state);
127,940,363✔
457
  timerDecRef(timer);
127,940,092✔
458

459
  return state == TIMER_STATE_WAITING;
127,940,363✔
460
}
461

462
bool taosTmrStopA(tmr_h* timerId) {
4,106,182✔
463
  bool ret = taosTmrStop(*timerId);
4,106,182✔
464
  *timerId = NULL;
4,106,182✔
465
  return ret;
4,106,182✔
466
}
467

468
bool taosTmrIsStopped(tmr_h* timerId) {
20,564✔
469
  uintptr_t id = (uintptr_t)*timerId;
20,564✔
470

471
  tmr_obj_t* timer = findTimer(id);
20,564✔
472
  if (timer == NULL) {
20,564✔
473
    tmrDebug("timer[id=%" PRIuPTR "] does not exist", id);
20,564✔
474
    return true;
20,564✔
475
  }
476

477
  uint8_t state = atomic_load_8(&timer->state);
×
478

479
  return (state == TIMER_STATE_CANCELED) || (state == TIMER_STATE_STOPPED);
×
480
}
481

482
bool taosTmrResetPriority(TAOS_TMR_CALLBACK fp, int32_t mseconds, void* param, void* handle, tmr_h* pTmrId,
332,481,143✔
483
                          uint8_t priority) {
484
  tmr_ctrl_t* ctrl = (tmr_ctrl_t*)handle;
332,481,143✔
485
  if (ctrl == NULL || ctrl->label[0] == 0) {
332,481,143✔
486
    return false;
10✔
487
  }
488

489
  uintptr_t  id = (uintptr_t)*pTmrId;
332,481,214✔
490
  bool       stopped = false;
332,481,290✔
491
  tmr_obj_t* timer = findTimer(id);
332,481,290✔
492
  if (timer == NULL) {
332,480,686✔
493
    tmrDebug("%s timer[id=%" PRIuPTR "] does not exist", ctrl->label, id);
123,200,628✔
494
  } else {
495
    uint8_t state = atomic_val_compare_exchange_8(&timer->state, TIMER_STATE_WAITING, TIMER_STATE_CANCELED);
209,280,058✔
496
    if (!doStopTimer(timer, state)) {
209,280,058✔
497
      timerDecRef(timer);
209,278,667✔
498
      timer = NULL;
209,278,667✔
499
    }
500
    stopped = state == TIMER_STATE_WAITING;
209,280,058✔
501
  }
502

503
  if (timer == NULL) {
332,481,184✔
504
    *pTmrId = taosTmrStartPriority(fp, mseconds, param, handle, priority);
332,479,793✔
505
    if (NULL == *pTmrId) {
332,480,202✔
506
      stopped = true;
×
507
    }
508
    return stopped;
332,480,202✔
509
  }
510

511
  tmrDebug("%s timer[id=%" PRIuPTR "] is reused", ctrl->label, timer->id);
1,391✔
512

513
  // wait until there's no other reference to this timer,
514
  // so that we can reuse this timer safely.
515
  for (int32_t i = 1; atomic_load_8(&timer->refCount) > 1; ++i) {
1,391✔
516
    if (i % 1000 == 0) {
×
517
      (void)sched_yield();
×
518
    }
519
  }
520

521
  if (timer->refCount != 1) {
1,391✔
522
    uError("timer refCount=%d not expected 1", timer->refCount);
×
523
  }
524
  memset(timer, 0, sizeof(*timer));
1,391✔
525
  *pTmrId = (tmr_h)doStartTimer(timer, fp, mseconds, param, ctrl, priority);
1,391✔
526

527
  return stopped;
1,391✔
528
}
529

530
bool taosTmrReset(TAOS_TMR_CALLBACK fp, int32_t mseconds, void* param, void* handle, tmr_h* pTmrId) {
303,980,759✔
531
  return taosTmrResetPriority(fp, mseconds, param, handle, pTmrId, 1);
303,980,759✔
532
}
533

534
static int32_t taosTmrModuleInit(void) {
2,174,560✔
535
  tmrCtrls = taosMemoryMalloc(sizeof(tmr_ctrl_t) * tsMaxTmrCtrl);
2,174,560✔
536
  if (tmrCtrls == NULL) {
2,174,560✔
537
    tmrError("failed to allocate memory for timer controllers.");
×
538
    return terrno;
×
539
  }
540

541
  memset(&timerMap, 0, sizeof(timerMap));
2,174,560✔
542

543
  for (uint32_t i = 0; i < tsMaxTmrCtrl - 1; ++i) {
2,147,483,647✔
544
    tmr_ctrl_t* ctrl = tmrCtrls + i;
2,147,483,647✔
545
    ctrl->next = ctrl + 1;
2,147,483,647✔
546
  }
547
  (tmrCtrls + tsMaxTmrCtrl - 1)->next = NULL;
2,174,560✔
548
  unusedTmrCtrl = tmrCtrls;
2,174,560✔
549

550
  (void)taosThreadMutexInit(&tmrCtrlMutex, NULL);
2,174,560✔
551

552
  int64_t now = taosGetMonotonicMs();
2,174,560✔
553
  for (int32_t i = 0; i < tListLen(wheels); i++) {
8,698,240✔
554
    time_wheel_t* wheel = wheels + i;
6,523,680✔
555
    if (taosThreadMutexInit(&wheel->mutex, NULL) != 0) {
6,523,680✔
556
      tmrError("failed to create the mutex for wheel, reason:%s", strerror(ERRNO));
×
557
      return terrno;
×
558
    }
559
    wheel->nextScanAt = now + wheel->resolution;
6,523,680✔
560
    wheel->index = 0;
6,523,680✔
561
    wheel->slots = (tmr_obj_t**)taosMemoryCalloc(wheel->size, sizeof(tmr_obj_t*));
6,523,680✔
562
    if (wheel->slots == NULL) {
6,523,680✔
563
      tmrError("failed to allocate wheel slots");
×
564
      return terrno;
×
565
    }
566
    timerMap.size += wheel->size;
6,523,680✔
567
  }
568

569
  timerMap.count = 0;
2,174,560✔
570
  timerMap.slots = (timer_list_t*)taosMemoryCalloc(timerMap.size, sizeof(timer_list_t));
2,174,560✔
571
  if (timerMap.slots == NULL) {
2,174,560✔
572
    tmrError("failed to allocate hash map");
×
573
    return terrno;
×
574
  }
575

576
  tmrQhandle = taosInitScheduler(10000, taosTmrThreads, "tmr", NULL);
2,174,560✔
577
  tmrQhandleHigh = taosInitScheduler(10000, taosTmrThreads, "high-tmr", NULL);
2,174,560✔
578
  if (taosInitTimer(taosTimerLoopFunc, MSECONDS_PER_TICK) != 0) {
2,174,560✔
579
    tmrError("failed to initialize timer");
×
580
  }
581

582
  tmrDebug("timer module is initialized, number of threads: %d", taosTmrThreads);
2,174,560✔
583

584
  return 2;
2,174,560✔
585
}
586

587
static int32_t taosTmrInitModule(void) {
12,457,433✔
588
  if (atomic_load_32(&tmrModuleInit) == 2) {
12,457,433✔
589
    return 0;
10,282,873✔
590
  }
591

592
  if (atomic_load_32(&tmrModuleInit) < 0) {
2,174,560✔
593
    return -1;
×
594
  }
595

596
  while (true) {
597
    if (0 == atomic_val_compare_exchange_32(&tmrModuleInit, 0, 1)) {
4,349,120✔
598
      atomic_store_32(&tmrModuleInit, taosTmrModuleInit());
2,174,560✔
599
    } else if (atomic_load_32(&tmrModuleInit) < 0) {
2,174,560✔
600
      return -1;
×
601
    } else if (atomic_load_32(&tmrModuleInit) == 2) {
2,174,560✔
602
      return 0;
2,174,560✔
603
    } else {
604
      taosMsleep(1);
×
605
    }
606
  }
607

608
  return -1;
609
}
610

611
void* taosTmrInit(int32_t maxNumOfTmrs, int32_t resolution, int32_t longest, const char* label) {
12,457,433✔
612
  const char* ret = taosMonotonicInit();
12,457,433✔
613
  tmrDebug("ttimer monotonic clock source:%s", ret);
12,457,433✔
614

615
  if (taosTmrInitModule() < 0) {
12,457,433✔
616
    return NULL;
×
617
  }
618

619
  (void)taosThreadMutexLock(&tmrCtrlMutex);
12,457,433✔
620
  tmr_ctrl_t* ctrl = unusedTmrCtrl;
12,457,433✔
621
  if (ctrl != NULL) {
12,457,433✔
622
    unusedTmrCtrl = ctrl->next;
12,457,433✔
623
    numOfTmrCtrl++;
12,457,433✔
624
  }
625
  (void)taosThreadMutexUnlock(&tmrCtrlMutex);
12,457,433✔
626

627
  if (ctrl == NULL) {
12,457,433✔
628
    tmrError("%s too many timer controllers, failed to create timer controller.", label);
×
629
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
630
    return NULL;
×
631
  }
632

633
  tstrncpy(ctrl->label, label, sizeof(ctrl->label));
12,457,433✔
634

635
  tmrDebug("%s timer controller is initialized, number of timer controllers: %d.", label, numOfTmrCtrl);
12,456,745✔
636
  return ctrl;
12,456,745✔
637
}
638

639
void taosTmrCleanUp(void* handle) {
10,849,125✔
640
  tmr_ctrl_t* ctrl = (tmr_ctrl_t*)handle;
10,849,125✔
641
  if (ctrl == NULL || ctrl->label[0] == 0) {
10,849,125✔
642
    return;
214✔
643
  }
644

645
  tmrDebug("%s timer controller is cleaned up.", ctrl->label);
10,848,911✔
646
  ctrl->label[0] = 0;
10,848,911✔
647

648
  (void)taosThreadMutexLock(&tmrCtrlMutex);
10,848,911✔
649
  ctrl->next = unusedTmrCtrl;
10,848,911✔
650
  numOfTmrCtrl--;
10,848,911✔
651
  unusedTmrCtrl = ctrl;
10,848,911✔
652
  (void)taosThreadMutexUnlock(&tmrCtrlMutex);
10,848,911✔
653

654
  tmrDebug("time controller's tmr ctrl size:  %d", numOfTmrCtrl);
10,848,911✔
655
  if (numOfTmrCtrl <= 0) {
10,848,911✔
656
    taosUninitTimer();
566,095✔
657

658
    taosCleanUpScheduler(tmrQhandle);
566,095✔
659
    taosMemoryFreeClear(tmrQhandle);
566,095✔
660

661
    taosCleanUpScheduler(tmrQhandleHigh);
566,095✔
662
    taosMemoryFreeClear(tmrQhandleHigh);
566,095✔
663

664
    for (int32_t i = 0; i < tListLen(wheels); i++) {
2,264,380✔
665
      time_wheel_t* wheel = wheels + i;
1,698,285✔
666
      (void)taosThreadMutexDestroy(&wheel->mutex);
1,698,285✔
667
      taosMemoryFree(wheel->slots);
1,698,285✔
668
    }
669

670
    (void)taosThreadMutexDestroy(&tmrCtrlMutex);
566,095✔
671

672
    for (size_t i = 0; i < timerMap.size; i++) {
2,147,483,647✔
673
      timer_list_t* list = timerMap.slots + i;
2,147,483,647✔
674
      tmr_obj_t*    t = list->timers;
2,147,483,647✔
675
      while (t != NULL) {
2,147,483,647✔
676
        tmr_obj_t* next = t->mnext;
279,054✔
677
        taosMemoryFree(t);
279,054✔
678
        t = next;
279,054✔
679
      }
680
    }
681
    taosMemoryFree(timerMap.slots);
566,095✔
682
    taosMemoryFree(tmrCtrls);
566,095✔
683

684
    tmrCtrls = NULL;
566,095✔
685
    unusedTmrCtrl = NULL;
566,095✔
686
    atomic_store_32(&tmrModuleInit, 0);
566,095✔
687
  }
688
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc