streamMeta.c 14.0 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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/>.
 */

#include "executor.h"
17
#include "streamBackendRocksdb.h"
18
#include "streamInt.h"
dengyihao's avatar
dengyihao 已提交
19
#include "tref.h"
L
Liu Jicong 已提交
20
#include "ttimer.h"
L
Liu Jicong 已提交
21

dengyihao's avatar
dengyihao 已提交
22
static TdThreadOnce streamMetaModuleInit = PTHREAD_ONCE_INIT;
dengyihao's avatar
dengyihao 已提交
23
int32_t             streamBackendId = 0;
Y
yihaoDeng 已提交
24
int32_t             streamBackendCfWrapperId = 0;
Y
yihaoDeng 已提交
25 26

static void streamMetaEnvInit() {
Y
yihaoDeng 已提交
27
  streamBackendId = taosOpenRef(64, streamBackendCleanup);
Y
yihaoDeng 已提交
28
  streamBackendCfWrapperId = taosOpenRef(64, streamBackendHandleCleanup);
Y
yihaoDeng 已提交
29
}
dengyihao's avatar
dengyihao 已提交
30 31

void streamMetaInit() { taosThreadOnce(&streamMetaModuleInit, streamMetaEnvInit); }
Y
yihaoDeng 已提交
32 33
void streamMetaCleanup() {
  taosCloseRef(streamBackendId);
Y
yihaoDeng 已提交
34
  taosCloseRef(streamBackendCfWrapperId);
Y
yihaoDeng 已提交
35
}
dengyihao's avatar
dengyihao 已提交
36

37
SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId) {
dengyihao's avatar
dengyihao 已提交
38
  int32_t      code = -1;
L
Liu Jicong 已提交
39 40 41 42 43
  SStreamMeta* pMeta = taosMemoryCalloc(1, sizeof(SStreamMeta));
  if (pMeta == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
44

L
Liu Jicong 已提交
45 46
  int32_t len = strlen(path) + 20;
  char*   streamPath = taosMemoryCalloc(1, len);
47
  sprintf(streamPath, "%s/%s", path, "stream");
48
  pMeta->path = taosStrdup(streamPath);
49
  if (tdbOpen(pMeta->path, 16 * 1024, 1, &pMeta->db, 0) < 0) {
L
Liu Jicong 已提交
50 51
    goto _err;
  }
dengyihao's avatar
dengyihao 已提交
52
  memset(streamPath, 0, len);
L
Liu Jicong 已提交
53

L
Liu Jicong 已提交
54
  sprintf(streamPath, "%s/%s", pMeta->path, "checkpoints");
dengyihao's avatar
dengyihao 已提交
55
  code = taosMulModeMkDir(streamPath, 0755);
dengyihao's avatar
dengyihao 已提交
56 57 58 59
  if (code != 0) {
    terrno = TAOS_SYSTEM_ERROR(code);
    goto _err;
  }
60

61
  if (tdbTbOpen("task.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pTaskDb, 0) < 0) {
L
Liu Jicong 已提交
62 63 64
    goto _err;
  }

65
  if (tdbTbOpen("checkpoint.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pCheckpointDb, 0) < 0) {
L
Liu Jicong 已提交
66 67 68
    goto _err;
  }

69
  _hash_fn_t fp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR);
H
Haojun Liao 已提交
70
  pMeta->pTasks = taosHashInit(64, fp, true, HASH_NO_LOCK);
L
Liu Jicong 已提交
71 72 73 74
  if (pMeta->pTasks == NULL) {
    goto _err;
  }

75
  // task list
76
  pMeta->pTaskList = taosArrayInit(4, sizeof(SStreamId));
77 78 79 80 81
  if (pMeta->pTaskList == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

L
Liu Jicong 已提交
82 83 84 85
  if (streamMetaBegin(pMeta) < 0) {
    goto _err;
  }

H
Haojun Liao 已提交
86
  pMeta->walScanCounter = 0;
87
  pMeta->vgId = vgId;
L
Liu Jicong 已提交
88 89
  pMeta->ahandle = ahandle;
  pMeta->expandFunc = expandFunc;
dengyihao's avatar
dengyihao 已提交
90

dengyihao's avatar
dengyihao 已提交
91 92 93
  memset(streamPath, 0, len);
  sprintf(streamPath, "%s/%s", pMeta->path, "state");
  code = taosMulModeMkDir(streamPath, 0755);
dengyihao's avatar
dengyihao 已提交
94 95 96 97 98
  if (code != 0) {
    terrno = TAOS_SYSTEM_ERROR(code);
    goto _err;
  }

dengyihao's avatar
dengyihao 已提交
99
  pMeta->streamBackend = streamBackendInit(streamPath);
dengyihao's avatar
dengyihao 已提交
100 101 102
  if (pMeta->streamBackend == NULL) {
    goto _err;
  }
dengyihao's avatar
dengyihao 已提交
103
  pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
Y
yihaoDeng 已提交
104 105
  pMeta->pTaskBackendUnique =
      taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
dengyihao's avatar
dengyihao 已提交
106

dengyihao's avatar
dengyihao 已提交
107
  taosMemoryFree(streamPath);
dengyihao's avatar
dengyihao 已提交
108

109
  taosInitRWLatch(&pMeta->lock);
Y
yihaoDeng 已提交
110 111
  taosThreadMutexInit(&pMeta->backendMutex, NULL);

112
  return pMeta;
L
Liu Jicong 已提交
113

L
Liu Jicong 已提交
114
_err:
dengyihao's avatar
dengyihao 已提交
115
  taosMemoryFree(streamPath);
L
Liu Jicong 已提交
116
  taosMemoryFree(pMeta->path);
L
Liu Jicong 已提交
117
  if (pMeta->pTasks) taosHashCleanup(pMeta->pTasks);
118
  if (pMeta->pTaskList) taosArrayDestroy(pMeta->pTaskList);
L
Liu Jicong 已提交
119
  if (pMeta->pTaskDb) tdbTbClose(pMeta->pTaskDb);
120
  if (pMeta->pCheckpointDb) tdbTbClose(pMeta->pCheckpointDb);
L
Liu Jicong 已提交
121
  if (pMeta->db) tdbClose(pMeta->db);
dengyihao's avatar
dengyihao 已提交
122
  // if (pMeta->streamBackend) streamBackendCleanup(pMeta->streamBackend);
L
Liu Jicong 已提交
123
  taosMemoryFree(pMeta);
dengyihao's avatar
dengyihao 已提交
124
  qError("failed to open stream meta");
L
Liu Jicong 已提交
125 126 127 128
  return NULL;
}

void streamMetaClose(SStreamMeta* pMeta) {
129
  tdbAbort(pMeta->db, pMeta->txn);
130
  tdbTbClose(pMeta->pTaskDb);
131
  tdbTbClose(pMeta->pCheckpointDb);
132
  tdbClose(pMeta->db);
L
Liu Jicong 已提交
133 134 135 136

  void* pIter = NULL;
  while (1) {
    pIter = taosHashIterate(pMeta->pTasks, pIter);
137 138 139 140
    if (pIter == NULL) {
      break;
    }

L
Liu Jicong 已提交
141
    SStreamTask* pTask = *(SStreamTask**)pIter;
142 143 144 145 146 147 148 149
    if (pTask->schedTimer) {
      taosTmrStop(pTask->schedTimer);
      pTask->schedTimer = NULL;
    }

    if (pTask->launchTaskTimer) {
      taosTmrStop(pTask->launchTaskTimer);
      pTask->launchTaskTimer = NULL;
L
Liu Jicong 已提交
150
    }
151

152
    tFreeStreamTask(pTask);
L
Liu Jicong 已提交
153
  }
154

L
Liu Jicong 已提交
155
  taosHashCleanup(pMeta->pTasks);
dengyihao's avatar
dengyihao 已提交
156
  taosRemoveRef(streamBackendId, pMeta->streamBackendRid);
157
  pMeta->pTaskList = taosArrayDestroy(pMeta->pTaskList);
L
Liu Jicong 已提交
158
  taosMemoryFree(pMeta->path);
Y
yihaoDeng 已提交
159 160
  taosThreadMutexDestroy(&pMeta->backendMutex);
  taosHashCleanup(pMeta->pTaskBackendUnique);
L
Liu Jicong 已提交
161 162 163
  taosMemoryFree(pMeta);
}

L
Liu Jicong 已提交
164 165
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
  void*   buf = NULL;
L
Liu Jicong 已提交
166 167
  int32_t len;
  int32_t code;
Y
yihaoDeng 已提交
168
  pTask->ver = SSTREAM_TASK_VER;
169
  tEncodeSize(tEncodeStreamTask, pTask, len, code);
L
Liu Jicong 已提交
170 171 172
  if (code < 0) {
    return -1;
  }
L
Liu Jicong 已提交
173
  buf = taosMemoryCalloc(1, len);
L
Liu Jicong 已提交
174 175 176 177
  if (buf == NULL) {
    return -1;
  }

178
  SEncoder encoder = {0};
L
Liu Jicong 已提交
179
  tEncoderInit(&encoder, buf, len);
180
  tEncodeStreamTask(&encoder, pTask);
181
  tEncoderClear(&encoder);
L
Liu Jicong 已提交
182

183
  if (tdbTbUpsert(pMeta->pTaskDb, &pTask->id.taskId, sizeof(int32_t), buf, len, pMeta->txn) < 0) {
H
Haojun Liao 已提交
184
    qError("s-task:%s save to disk failed, code:%s", pTask->id.idStr, tstrerror(terrno));
L
Liu Jicong 已提交
185 186 187
    return -1;
  }

L
Liu Jicong 已提交
188
  taosMemoryFree(buf);
L
Liu Jicong 已提交
189 190 191
  return 0;
}

192 193 194 195 196 197 198 199 200 201 202
int32_t streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId) {
  int32_t code = tdbTbDelete(pMeta->pTaskDb, &taskId, sizeof(taskId), pMeta->txn);
  if (code != 0) {
    qError("vgId:%d failed to remove task:0x%x from metastore, code:%s", pMeta->vgId, taskId, tstrerror(terrno));
  } else {
    qDebug("vgId:%d remove task:0x%x from metastore", pMeta->vgId, taskId);
  }

  return code;
}

203
// add to the ready tasks hash map, not the restored tasks hash map
204 205 206
int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask, bool* pAdded) {
  *pAdded = false;

H
Haojun Liao 已提交
207 208
  int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId};
  void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys));
209
  if (p == NULL) {
210 211 212 213 214
    if (pMeta->expandFunc(pMeta->ahandle, pTask, ver) < 0) {
      tFreeStreamTask(pTask);
      return -1;
    }

215
    taosArrayPush(pMeta->pTaskList, &pTask->id);
H
Haojun Liao 已提交
216

217 218 219 220 221 222 223 224 225 226 227
    if (streamMetaSaveTask(pMeta, pTask) < 0) {
      tFreeStreamTask(pTask);
      return -1;
    }

    if (streamMetaCommit(pMeta) < 0) {
      tFreeStreamTask(pTask);
      return -1;
    }
  } else {
    return 0;
228 229
  }

H
Haojun Liao 已提交
230
  taosHashPut(pMeta->pTasks, keys, sizeof(keys), &pTask, POINTER_BYTES);
231
  *pAdded = true;
L
Liu Jicong 已提交
232 233
  return 0;
}
234

235
int32_t streamMetaGetNumOfTasks(SStreamMeta* pMeta) {
236 237
  size_t size = taosHashGetSize(pMeta->pTasks);
  ASSERT(taosArrayGetSize(pMeta->pTaskList) == taosHashGetSize(pMeta->pTasks));
238
  return (int32_t)size;
239
}
L
Liu Jicong 已提交
240

241
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId) {
L
Liu Jicong 已提交
242 243
  taosRLockLatch(&pMeta->lock);

244 245
  int64_t keys[2] = {streamId, taskId};
  SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
246
  if (ppTask != NULL) {
247
    if (!streamTaskShouldStop(&(*ppTask)->status)) {
H
Haojun Liao 已提交
248
      int32_t ref = atomic_add_fetch_32(&(*ppTask)->refCnt, 1);
249
      taosRUnLockLatch(&pMeta->lock);
250
      qTrace("s-task:%s acquire task, ref:%d", (*ppTask)->id.idStr, ref);
251 252
      return *ppTask;
    }
L
Liu Jicong 已提交
253
  }
254

L
Liu Jicong 已提交
255 256 257 258 259
  taosRUnLockLatch(&pMeta->lock);
  return NULL;
}

void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask) {
H
Haojun Liao 已提交
260 261
  int32_t ref = atomic_sub_fetch_32(&pTask->refCnt, 1);
  if (ref > 0) {
262
    qTrace("s-task:%s release task, ref:%d", pTask->id.idStr, ref);
H
Haojun Liao 已提交
263
  } else if (ref == 0) {
264
    ASSERT(streamTaskShouldStop(&pTask->status));
H
Haojun Liao 已提交
265
    qTrace("s-task:%s all refs are gone, free it", pTask->id.idStr);
266
    tFreeStreamTask(pTask);
H
Haojun Liao 已提交
267 268 269 270 271
  } else if (ref < 0) {
    qError("task ref is invalid, ref:%d, %s", ref, pTask->id.idStr);
  }
}

272
static void doRemoveIdFromList(SStreamMeta* pMeta, int32_t num, SStreamId* id) {
H
Haojun Liao 已提交
273
  for (int32_t i = 0; i < num; ++i) {
274 275
    SStreamId* pTaskId = taosArrayGet(pMeta->pTaskList, i);
    if (pTaskId->streamId == id->streamId && pTaskId->taskId == id->taskId) {
H
Haojun Liao 已提交
276 277 278
      taosArrayRemove(pMeta->pTaskList, i);
      break;
    }
L
Liu Jicong 已提交
279 280 281
  }
}

H
Haojun Liao 已提交
282
int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId) {
H
Haojun Liao 已提交
283
  SStreamTask* pTask = NULL;
H
Haojun Liao 已提交
284

H
Haojun Liao 已提交
285 286
  // pre-delete operation
  taosWLockLatch(&pMeta->lock);
H
Haojun Liao 已提交
287 288 289

  int64_t keys[2] = {streamId, taskId};
  SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
L
Liu Jicong 已提交
290
  if (ppTask) {
H
Haojun Liao 已提交
291 292 293 294 295
    pTask = *ppTask;
    atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__DROPPING);
  } else {
    qDebug("vgId:%d failed to find the task:0x%x, it may be dropped already", pMeta->vgId, taskId);
    taosWUnLockLatch(&pMeta->lock);
296
    return 0;
H
Haojun Liao 已提交
297 298 299
  }
  taosWUnLockLatch(&pMeta->lock);

300 301
  qDebug("s-task:0x%x set task status:%s and start to unregister it", taskId,
         streamGetTaskStatusStr(TASK_STATUS__DROPPING));
H
Haojun Liao 已提交
302

Y
yihaoDeng 已提交
303
  while (1) {
H
Haojun Liao 已提交
304
    taosRLockLatch(&pMeta->lock);
H
Haojun Liao 已提交
305
    ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
H
Haojun Liao 已提交
306 307 308 309 310 311

    if (ppTask) {
      if ((*ppTask)->status.timerActive == 0) {
        taosRUnLockLatch(&pMeta->lock);
        break;
      }
312

H
Haojun Liao 已提交
313 314 315 316 317 318 319 320 321 322 323
      taosMsleep(10);
      qDebug("s-task:%s wait for quit from timer", (*ppTask)->id.idStr);
      taosRUnLockLatch(&pMeta->lock);
    } else {
      taosRUnLockLatch(&pMeta->lock);
      break;
    }
  }

  // let's do delete of stream task
  taosWLockLatch(&pMeta->lock);
H
Haojun Liao 已提交
324
  ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
H
Haojun Liao 已提交
325
  if (ppTask) {
326
    taosHashRemove(pMeta->pTasks, keys, sizeof(keys));
dengyihao's avatar
dengyihao 已提交
327
    atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__DROPPING);
328

H
Haojun Liao 已提交
329
    ASSERT(pTask->status.timerActive == 0);
330
    doRemoveIdFromList(pMeta, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id);
H
Haojun Liao 已提交
331 332 333 334

    // remove the ref by timer
    if (pTask->triggerParam != 0) {
      taosTmrStop(pTask->schedTimer);
335
    }
H
Haojun Liao 已提交
336 337 338

    streamMetaRemoveTask(pMeta, taskId);
    streamMetaReleaseTask(pMeta, pTask);
339
  } else {
H
Haojun Liao 已提交
340
    qDebug("vgId:%d failed to find the task:0x%x, it may have been dropped already", pMeta->vgId, taskId);
L
Liu Jicong 已提交
341
  }
H
Haojun Liao 已提交
342 343

  taosWUnLockLatch(&pMeta->lock);
344
  return 0;
L
Liu Jicong 已提交
345 346
}

L
Liu Jicong 已提交
347
int32_t streamMetaBegin(SStreamMeta* pMeta) {
348 349
  if (tdbBegin(pMeta->db, &pMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL,
               TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
L
Liu Jicong 已提交
350 351 352 353 354
    return -1;
  }
  return 0;
}

355
// todo add error log
L
Liu Jicong 已提交
356
int32_t streamMetaCommit(SStreamMeta* pMeta) {
357
  if (tdbCommit(pMeta->db, pMeta->txn) < 0) {
358
    qError("failed to commit stream meta");
L
Liu Jicong 已提交
359 360
    return -1;
  }
361

362
  if (tdbPostCommit(pMeta->db, pMeta->txn) < 0) {
363
    qError("failed to commit stream meta");
364 365
    return -1;
  }
366 367 368

  if (tdbBegin(pMeta->db, &pMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL,
               TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
369 370
    return -1;
  }
371

L
Liu Jicong 已提交
372 373 374
  return 0;
}

375
int32_t streamMetaAbort(SStreamMeta* pMeta) {
376
  if (tdbAbort(pMeta->db, pMeta->txn) < 0) {
377 378
    return -1;
  }
379 380 381

  if (tdbBegin(pMeta->db, &pMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL,
               TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
382 383
    return -1;
  }
L
Liu Jicong 已提交
384 385
  return 0;
}
386

L
Liu Jicong 已提交
387
int32_t streamLoadTasks(SStreamMeta* pMeta, int64_t ver) {
L
Liu Jicong 已提交
388
  TBC* pCur = NULL;
389

L
Liu Jicong 已提交
390
  if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) {
391
    qError("vgId:%d failed to open stream meta, code:%s", pMeta->vgId, tstrerror(terrno));
L
Liu Jicong 已提交
392 393 394 395 396 397 398 399
    return -1;
  }

  void*    pKey = NULL;
  int32_t  kLen = 0;
  void*    pVal = NULL;
  int32_t  vLen = 0;
  SDecoder decoder;
400
  SArray*  pRecycleList = taosArrayInit(4, sizeof(int32_t));
L
Liu Jicong 已提交
401 402 403 404 405 406

  tdbTbcMoveToFirst(pCur);

  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
    SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
    if (pTask == NULL) {
L
Liu Jicong 已提交
407 408
      tdbFree(pKey);
      tdbFree(pVal);
5
54liuyao 已提交
409
      tdbTbcClose(pCur);
410
      taosArrayDestroy(pRecycleList);
L
Liu Jicong 已提交
411 412 413
      return -1;
    }
    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
Y
yihaoDeng 已提交
414 415 416 417 418 419 420
    if (tDecodeStreamTask(&decoder, pTask) < 0) {
      tDecoderClear(&decoder);
      tdbFree(pKey);
      tdbFree(pVal);
      tdbTbcClose(pCur);
      taosArrayDestroy(pRecycleList);
      tFreeStreamTask(pTask);
Y
yihaoDeng 已提交
421 422 423 424
      qError(
          "stream read incompatible data, rm %s/vnode/vnode*/tq/stream if taosd cannot start, and rebuild stream "
          "manually",
          tsDataDir);
Y
yihaoDeng 已提交
425 426
      return -1;
    }
Y
yihaoDeng 已提交
427
    tDecoderClear(&decoder);
428

429 430 431 432 433 434 435 436 437 438 439 440
    if (pTask->status.taskStatus == TASK_STATUS__DROPPING) {
      int32_t taskId = pTask->id.taskId;
      tFreeStreamTask(pTask);

      taosArrayPush(pRecycleList, &taskId);

      int32_t total = taosArrayGetSize(pRecycleList);
      qDebug("s-task:0x%x is already dropped, add into recycle list, total:%d", taskId, total);
      continue;
    }

    // do duplicate task check.
H
Haojun Liao 已提交
441 442
    int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId};
    void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys));
443
    if (p == NULL) {
444 445 446 447
      if (pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.version) < 0) {
        tdbFree(pKey);
        tdbFree(pVal);
        tdbTbcClose(pCur);
448 449
        tFreeStreamTask(pTask);
        taosArrayDestroy(pRecycleList);
450 451
        return -1;
      }
452

453
      taosArrayPush(pMeta->pTaskList, &pTask->id);
454
    } else {
455 456 457 458
      tdbFree(pKey);
      tdbFree(pVal);
      tdbTbcClose(pCur);
      taosMemoryFree(pTask);
459
      continue;
460
    }
461

H
Haojun Liao 已提交
462
    if (taosHashPut(pMeta->pTasks, keys, sizeof(keys), &pTask, sizeof(void*)) < 0) {
L
Liu Jicong 已提交
463 464
      tdbFree(pKey);
      tdbFree(pVal);
5
54liuyao 已提交
465
      tdbTbcClose(pCur);
466 467
      tFreeStreamTask(pTask);
      taosArrayDestroy(pRecycleList);
468 469
      return -1;
    }
470

471
    ASSERT(pTask->status.downstreamReady == 0);
472 473
  }

474 475
  tdbFree(pKey);
  tdbFree(pVal);
476
  if (tdbTbcClose(pCur) < 0) {
477
    taosArrayDestroy(pRecycleList);
478
    return -1;
L
Liu Jicong 已提交
479 480
  }

481
  if (taosArrayGetSize(pRecycleList) > 0) {
Y
yihaoDeng 已提交
482 483
    for (int32_t i = 0; i < taosArrayGetSize(pRecycleList); ++i) {
      int32_t taskId = *(int32_t*)taosArrayGet(pRecycleList, i);
484 485 486 487
      streamMetaRemoveTask(pMeta, taskId);
    }
  }

Y
yihaoDeng 已提交
488
  qDebug("vgId:%d load %d task from disk", pMeta->vgId, (int32_t)taosArrayGetSize(pMeta->pTaskList));
489
  taosArrayDestroy(pRecycleList);
L
Liu Jicong 已提交
490 491
  return 0;
}