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

S
Shengliang Guan 已提交
16 17
#define _DEFAULT_SOURCE
#include "tsched.h"
18
#include "tdef.h"
S
log  
Shengliang Guan 已提交
19
#include "tlog.h"
20
#include "ttimer.h"
S
Shengliang Guan 已提交
21
#include "tutil.h"
22

S
Shengliang Guan 已提交
23
#define DUMP_SCHEDULER_TIME_WINDOW 30000  // every 30sec, take a snap shot of task queue.
H
hzcheng 已提交
24

25
static void *taosProcessSchedQueue(void *param);
S
Shengliang Guan 已提交
26
static void  taosDumpSchedulerStatus(void *qhandle, void *tmrId);
H
hzcheng 已提交
27

D
dapan1121 已提交
28 29 30 31 32 33 34
void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *label, SSchedQueue *pSched) {
  if (NULL == pSched) {
    pSched = (SSchedQueue *)taosMemoryCalloc(sizeof(SSchedQueue), 1);
    if (pSched == NULL) {
      uError("%s: no enough memory for pSched", label);
      return NULL;
    }
J
Jeff Tao 已提交
35 36
  }

wafwerar's avatar
wafwerar 已提交
37
  pSched->queue = (SSchedMsg *)taosMemoryCalloc(sizeof(SSchedMsg), queueSize);
J
Jeff Tao 已提交
38
  if (pSched->queue == NULL) {
S
slguan 已提交
39
    uError("%s: no enough memory for queue", label);
J
Jeff Tao 已提交
40 41 42 43
    taosCleanUpScheduler(pSched);
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
44
  pSched->qthread = taosMemoryCalloc(sizeof(TdThread), numOfThreads);
J
Jeff Tao 已提交
45
  if (pSched->qthread == NULL) {
S
slguan 已提交
46
    uError("%s: no enough memory for qthread", label);
J
Jeff Tao 已提交
47 48
    taosCleanUpScheduler(pSched);
    return NULL;
49
  }
H
hzcheng 已提交
50 51

  pSched->queueSize = queueSize;
S
Shengliang Guan 已提交
52
  tstrncpy(pSched->label, label, sizeof(pSched->label));  // fix buffer overflow
H
hzcheng 已提交
53

J
Jeff Tao 已提交
54 55 56
  pSched->fullSlot = 0;
  pSched->emptySlot = 0;

wafwerar's avatar
wafwerar 已提交
57
  if (taosThreadMutexInit(&pSched->queueMutex, NULL) < 0) {
S
slguan 已提交
58
    uError("init %s:queueMutex failed(%s)", label, strerror(errno));
J
Jeff Tao 已提交
59 60
    taosCleanUpScheduler(pSched);
    return NULL;
H
hzcheng 已提交
61 62
  }

S
Shengliang Guan 已提交
63
  if (tsem_init(&pSched->emptySem, 0, (uint32_t)pSched->queueSize) != 0) {
S
slguan 已提交
64
    uError("init %s:empty semaphore failed(%s)", label, strerror(errno));
J
Jeff Tao 已提交
65 66
    taosCleanUpScheduler(pSched);
    return NULL;
H
hzcheng 已提交
67 68
  }

S
slguan 已提交
69
  if (tsem_init(&pSched->fullSem, 0, 0) != 0) {
S
slguan 已提交
70
    uError("init %s:full semaphore failed(%s)", label, strerror(errno));
J
Jeff Tao 已提交
71 72
    taosCleanUpScheduler(pSched);
    return NULL;
H
hzcheng 已提交
73 74
  }

D
dapan1121 已提交
75
  atomic_store_8(&pSched->stop, 0);
S
Shengliang Guan 已提交
76
  for (int32_t i = 0; i < numOfThreads; ++i) {
wafwerar's avatar
wafwerar 已提交
77 78 79 80 81
    TdThreadAttr attr;
    taosThreadAttrInit(&attr);
    taosThreadAttrSetDetachState(&attr, PTHREAD_CREATE_JOINABLE);
    int32_t code = taosThreadCreate(pSched->qthread + i, &attr, taosProcessSchedQueue, (void *)pSched);
    taosThreadAttrDestroy(&attr);
J
Jeff Tao 已提交
82
    if (code != 0) {
S
slguan 已提交
83
      uError("%s: failed to create rpc thread(%s)", label, strerror(errno));
J
Jeff Tao 已提交
84 85
      taosCleanUpScheduler(pSched);
      return NULL;
H
hzcheng 已提交
86
    }
87
    ++pSched->numOfThreads;
H
hzcheng 已提交
88 89
  }

90
  uDebug("%s scheduler is initialized, numOfThreads:%d", label, pSched->numOfThreads);
H
hzcheng 已提交
91 92 93 94

  return (void *)pSched;
}

S
Shengliang Guan 已提交
95
void *taosInitSchedulerWithInfo(int32_t queueSize, int32_t numOfThreads, const char *label, void *tmrCtrl) {
D
dapan1121 已提交
96
  SSchedQueue *pSched = taosInitScheduler(queueSize, numOfThreads, label, NULL);
97

98 99 100 101
  if (tmrCtrl != NULL && pSched != NULL) {
    pSched->pTmrCtrl = tmrCtrl;
    taosTmrReset(taosDumpSchedulerStatus, DUMP_SCHEDULER_TIME_WINDOW, pSched, pSched->pTmrCtrl, &pSched->pTimer);
  }
102

103 104 105
  return pSched;
}

106
void *taosProcessSchedQueue(void *scheduler) {
H
hzcheng 已提交
107
  SSchedMsg    msg;
108
  SSchedQueue *pSched = (SSchedQueue *)scheduler;
S
Shengliang Guan 已提交
109
  int32_t      ret = 0;
H
hzcheng 已提交
110

H
Haojun Liao 已提交
111 112 113
  char name[16] = {0};
  snprintf(name, tListLen(name), "%s-taskQ", pSched->label);
  setThreadName(name);
114

H
hzcheng 已提交
115
  while (1) {
116 117
    if ((ret = tsem_wait(&pSched->fullSem)) != 0) {
      uFatal("wait %s fullSem failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
118
      ASSERT(0);
H
hzcheng 已提交
119
    }
D
dapan1121 已提交
120
    if (atomic_load_8(&pSched->stop)) {
121 122
      break;
    }
H
hzcheng 已提交
123

wafwerar's avatar
wafwerar 已提交
124
    if ((ret = taosThreadMutexLock(&pSched->queueMutex)) != 0) {
125
      uFatal("lock %s queueMutex failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
126
      ASSERT(0);
127
    }
H
hzcheng 已提交
128 129 130 131 132

    msg = pSched->queue[pSched->fullSlot];
    memset(pSched->queue + pSched->fullSlot, 0, sizeof(SSchedMsg));
    pSched->fullSlot = (pSched->fullSlot + 1) % pSched->queueSize;

wafwerar's avatar
wafwerar 已提交
133
    if ((ret = taosThreadMutexUnlock(&pSched->queueMutex)) != 0) {
134
      uFatal("unlock %s queueMutex failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
135
      ASSERT(0);
136
    }
H
hzcheng 已提交
137

138 139
    if ((ret = tsem_post(&pSched->emptySem)) != 0) {
      uFatal("post %s emptySem failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
140
      ASSERT(0);
141
    }
H
hzcheng 已提交
142 143 144 145 146 147

    if (msg.fp)
      (*(msg.fp))(&msg);
    else if (msg.tfp)
      (*(msg.tfp))(msg.ahandle, msg.thandle);
  }
S
slguan 已提交
148 149

  return NULL;
H
hzcheng 已提交
150 151
}

dengyihao's avatar
dengyihao 已提交
152
int taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg) {
153
  SSchedQueue *pSched = (SSchedQueue *)queueScheduler;
S
Shengliang Guan 已提交
154
  int32_t      ret = 0;
155

H
hzcheng 已提交
156
  if (pSched == NULL) {
S
slguan 已提交
157
    uError("sched is not ready, msg:%p is dropped", pMsg);
dengyihao's avatar
dengyihao 已提交
158
    return -1;
H
hzcheng 已提交
159 160
  }

D
dapan1121 已提交
161 162
  if (atomic_load_8(&pSched->stop)) {
    uError("sched is already stopped, msg:%p is dropped", pMsg);
dengyihao's avatar
dengyihao 已提交
163
    return -1;
D
dapan1121 已提交
164 165
  }

166 167
  if ((ret = tsem_wait(&pSched->emptySem)) != 0) {
    uFatal("wait %s emptySem failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
168
    ASSERT(0);
169
  }
H
hzcheng 已提交
170

wafwerar's avatar
wafwerar 已提交
171
  if ((ret = taosThreadMutexLock(&pSched->queueMutex)) != 0) {
172
    uFatal("lock %s queueMutex failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
173
    ASSERT(0);
174
  }
H
hzcheng 已提交
175 176 177 178

  pSched->queue[pSched->emptySlot] = *pMsg;
  pSched->emptySlot = (pSched->emptySlot + 1) % pSched->queueSize;

wafwerar's avatar
wafwerar 已提交
179
  if ((ret = taosThreadMutexUnlock(&pSched->queueMutex)) != 0) {
180
    uFatal("unlock %s queueMutex failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
181
    ASSERT(0);
182
  }
H
hzcheng 已提交
183

184 185
  if ((ret = tsem_post(&pSched->fullSem)) != 0) {
    uFatal("post %s fullSem failed(%s)", pSched->label, strerror(errno));
D
dapan1121 已提交
186
    ASSERT(0);
187
  }
dengyihao's avatar
dengyihao 已提交
188
  return ret;
H
hzcheng 已提交
189 190 191 192 193 194
}

void taosCleanUpScheduler(void *param) {
  SSchedQueue *pSched = (SSchedQueue *)param;
  if (pSched == NULL) return;

D
dapan1121 已提交
195
  uDebug("start to cleanup %s schedQsueue", pSched->label);
dengyihao's avatar
dengyihao 已提交
196

D
dapan1121 已提交
197 198 199
  atomic_store_8(&pSched->stop, 1);

  taosMsleep(200);
dengyihao's avatar
dengyihao 已提交
200

S
Shengliang Guan 已提交
201 202
  for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
    if (taosCheckPthreadValid(pSched->qthread[i])) {
203 204
      tsem_post(&pSched->fullSem);
    }
205
  }
S
Shengliang Guan 已提交
206
  for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
S
TD-1037  
Shengliang Guan 已提交
207
    if (taosCheckPthreadValid(pSched->qthread[i])) {
wafwerar's avatar
wafwerar 已提交
208
      taosThreadJoin(pSched->qthread[i], NULL);
209
      taosThreadClear(&pSched->qthread[i]);
210
    }
H
hzcheng 已提交
211 212
  }

S
slguan 已提交
213 214
  tsem_destroy(&pSched->emptySem);
  tsem_destroy(&pSched->fullSem);
wafwerar's avatar
wafwerar 已提交
215
  taosThreadMutexDestroy(&pSched->queueMutex);
S
Shengliang Guan 已提交
216

217
  if (pSched->pTimer) {
L
Liu Jicong 已提交
218 219
    taosTmrStop(pSched->pTimer);
    pSched->pTimer = NULL;
220
  }
H
hzcheng 已提交
221

wafwerar's avatar
wafwerar 已提交
222 223
  if (pSched->queue) taosMemoryFree(pSched->queue);
  if (pSched->qthread) taosMemoryFree(pSched->qthread);
dengyihao's avatar
dengyihao 已提交
224
  // taosMemoryFree(pSched);
H
hzcheng 已提交
225
}
226 227 228 229 230 231 232

// for debug purpose, dump the scheduler status every 1min.
void taosDumpSchedulerStatus(void *qhandle, void *tmrId) {
  SSchedQueue *pSched = (SSchedQueue *)qhandle;
  if (pSched == NULL || pSched->pTimer == NULL || pSched->pTimer != tmrId) {
    return;
  }
S
Shengliang Guan 已提交
233

234 235
  int32_t size = ((pSched->emptySlot - pSched->fullSlot) + pSched->queueSize) % pSched->queueSize;
  if (size > 0) {
236
    uDebug("scheduler:%s, current tasks in queue:%d, task thread:%d", pSched->label, size, pSched->numOfThreads);
237
  }
S
Shengliang Guan 已提交
238

239
  taosTmrReset(taosDumpSchedulerStatus, DUMP_SCHEDULER_TIME_WINDOW, pSched, pSched->pTmrCtrl, &pSched->pTimer);
240
}