streamExec.c 13.4 KB
Newer Older
L
Liu Jicong 已提交
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/>.
 */

L
Liu Jicong 已提交
16
#include "streamInc.h"
L
Liu Jicong 已提交
17

H
Haojun Liao 已提交
18 19
// maximum allowed processed block batches. One block may include several submit blocks
#define MAX_STREAM_EXEC_BATCH_NUM 128
L
liuyao 已提交
20
#define MIN_STREAM_EXEC_BATCH_NUM 16
5
54liuyao 已提交
21

22
bool streamTaskShouldStop(const SStreamStatus* pStatus) {
dengyihao's avatar
dengyihao 已提交
23
  int32_t status = atomic_load_8((int8_t*)&pStatus->taskStatus);
24 25 26
  return (status == TASK_STATUS__STOP) || (status == TASK_STATUS__DROPPING);
}

L
liuyao 已提交
27
bool streamTaskShouldPause(const SStreamStatus* pStatus) {
dengyihao's avatar
dengyihao 已提交
28
  int32_t status = atomic_load_8((int8_t*)&pStatus->taskStatus);
L
liuyao 已提交
29 30 31
  return (status == TASK_STATUS__PAUSE);
}

L
Liu Jicong 已提交
32
static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* pRes) {
33 34 35
  int32_t code = TSDB_CODE_SUCCESS;
  void*   pExecutor = pTask->exec.pExecutor;

36
  while (pTask->taskLevel == TASK_LEVEL__SOURCE) {
37
    int8_t status = atomic_load_8(&pTask->status.taskStatus);
38
    if (status != TASK_STATUS__NORMAL) {
39
      qError("stream task wait for the end of fill history, s-task:%s, status:%d", pTask->id.idStr,
40
             atomic_load_8(&pTask->status.taskStatus));
41 42 43 44
      taosMsleep(2);
    } else {
      break;
    }
5
54liuyao 已提交
45
  }
L
Liu Jicong 已提交
46 47

  // set input
L
Liu Jicong 已提交
48
  const SStreamQueueItem* pItem = (const SStreamQueueItem*)data;
L
Liu Jicong 已提交
49
  if (pItem->type == STREAM_INPUT__GET_RES) {
L
Liu Jicong 已提交
50
    const SStreamTrigger* pTrigger = (const SStreamTrigger*)data;
51
    qSetMultiStreamInput(pExecutor, pTrigger->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
52
  } else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
53
    ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
L
Liu Jicong 已提交
54
    const SStreamDataSubmit2* pSubmit = (const SStreamDataSubmit2*)data;
55
    qSetMultiStreamInput(pExecutor, &pSubmit->submit, 1, STREAM_INPUT__DATA_SUBMIT);
dengyihao's avatar
dengyihao 已提交
56 57
    qDebug("s-task:%s set submit blocks as source block completed, %p %p len:%d ver:%" PRId64, pTask->id.idStr, pSubmit,
           pSubmit->submit.msgStr, pSubmit->submit.msgLen, pSubmit->submit.ver);
58
  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
L
Liu Jicong 已提交
59
    const SStreamDataBlock* pBlock = (const SStreamDataBlock*)data;
60 61 62

    SArray* pBlockList = pBlock->blocks;
    int32_t numOfBlocks = taosArrayGetSize(pBlockList);
dengyihao's avatar
dengyihao 已提交
63
    qDebug("s-task:%s set sdata blocks as input num:%d, ver:%" PRId64, pTask->id.idStr, numOfBlocks, pBlock->sourceVer);
64
    qSetMultiStreamInput(pExecutor, pBlockList->pData, numOfBlocks, STREAM_INPUT__DATA_BLOCK);
65
  } else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {
L
Liu Jicong 已提交
66
    const SStreamMergedSubmit2* pMerged = (const SStreamMergedSubmit2*)data;
67 68 69

    SArray* pBlockList = pMerged->submits;
    int32_t numOfBlocks = taosArrayGetSize(pBlockList);
70
    qDebug("s-task:%s %p set submit input (merged), numOfblocks:%d", pTask->id.idStr, pTask, numOfBlocks);
71
    qSetMultiStreamInput(pExecutor, pBlockList->pData, numOfBlocks, STREAM_INPUT__MERGED_SUBMIT);
L
Liu Jicong 已提交
72 73
  } else if (pItem->type == STREAM_INPUT__REF_DATA_BLOCK) {
    const SStreamRefDataBlock* pRefBlock = (const SStreamRefDataBlock*)data;
74
    qSetMultiStreamInput(pExecutor, pRefBlock->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
75
  } else {
76
    ASSERT(0);
L
Liu Jicong 已提交
77 78
  }

79
  // pExecutor
L
Liu Jicong 已提交
80
  while (1) {
81
    if (streamTaskShouldStop(&pTask->status)) {
L
Liu Jicong 已提交
82 83 84
      return 0;
    }

L
Liu Jicong 已提交
85 86
    SSDataBlock* output = NULL;
    uint64_t     ts = 0;
87
    if ((code = qExecTask(pExecutor, &output, &ts)) < 0) {
5
54liuyao 已提交
88
      if (code == TSDB_CODE_QRY_IN_EXEC) {
89
        resetTaskInfo(pExecutor);
5
54liuyao 已提交
90
      }
91 92

      qError("unexpected stream execution, s-task:%s since %s", pTask->id.idStr, terrstr());
L
Liu Jicong 已提交
93
      continue;
L
Liu Jicong 已提交
94
    }
95

96
    if (output == NULL) {
5
54liuyao 已提交
97
      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
98 99
        SSDataBlock block = {0};

L
Liu Jicong 已提交
100
        const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)data;
101
        ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
102

L
Liu Jicong 已提交
103
        assignOneDataBlock(&block, taosArrayGet(pRetrieveBlock->blocks, 0));
L
Liu Jicong 已提交
104
        block.info.type = STREAM_PULL_OVER;
L
Liu Jicong 已提交
105 106
        block.info.childId = pTask->selfChildId;
        taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
107

108
        qDebug("task %d(child %d) processed retrieve, reqId %" PRId64, pTask->id.taskId, pTask->selfChildId,
L
Liu Jicong 已提交
109
               pRetrieveBlock->reqId);
110 111 112
      }
      break;
    }
L
Liu Jicong 已提交
113 114 115 116 117 118 119 120

    if (output->info.type == STREAM_RETRIEVE) {
      if (streamBroadcastToChildren(pTask, output) < 0) {
        // TODO
      }
      continue;
    }

121
    qDebug("s-task:%s (child %d) executed and get block", pTask->id.idStr, pTask->selfChildId);
J
jiajingbin 已提交
122

L
Liu Jicong 已提交
123 124 125 126
    SSDataBlock block = {0};
    assignOneDataBlock(&block, output);
    block.info.childId = pTask->selfChildId;
    taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
127
  }
128

L
Liu Jicong 已提交
129 130 131
  return 0;
}

132
int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
dengyihao's avatar
dengyihao 已提交
133
  int32_t code = 0;
134

dengyihao's avatar
dengyihao 已提交
135
  ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
136

137
  void* exec = pTask->exec.pExecutor;
138

L
Liu Jicong 已提交
139
  qSetStreamOpOpen(exec);
L
Liu Jicong 已提交
140
  bool finished = false;
L
Liu Jicong 已提交
141

142 143 144 145 146 147 148 149 150
  while (1) {
    SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
    if (pRes == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }

    int32_t batchCnt = 0;
    while (1) {
L
liuyao 已提交
151
      if (streamTaskShouldStop(&pTask->status) || streamTaskShouldPause(&pTask->status)) {
5
54liuyao 已提交
152
        taosArrayDestroy(pRes);
L
Liu Jicong 已提交
153 154 155
        return 0;
      }

156 157 158
      SSDataBlock* output = NULL;
      uint64_t     ts = 0;
      if (qExecTask(exec, &output, &ts) < 0) {
5
54liuyao 已提交
159
        continue;
160
      }
L
Liu Jicong 已提交
161
      if (output == NULL) {
L
Liu Jicong 已提交
162 163 164 165 166
        if (qStreamRecoverScanFinished(exec)) {
          finished = true;
        } else {
          qSetStreamOpOpen(exec);
        }
L
Liu Jicong 已提交
167 168
        break;
      }
169 170 171 172 173 174

      SSDataBlock block = {0};
      assignOneDataBlock(&block, output);
      block.info.childId = pTask->selfChildId;
      taosArrayPush(pRes, &block);

L
Liu Jicong 已提交
175 176
      batchCnt++;

177
      qDebug("s-task:%s scan exec numOfBlocks:%d, limit:%d", pTask->id.idStr, batchCnt, batchSz);
H
Haojun Liao 已提交
178 179 180
      if (batchCnt >= batchSz) {
        break;
      }
181
    }
H
Haojun Liao 已提交
182

183
    if (taosArrayGetSize(pRes) == 0) {
184 185
      if (finished) {
        taosArrayDestroy(pRes);
H
Haojun Liao 已提交
186
        qDebug("s-task:%s finish recover exec task ", pTask->id.idStr);
187 188
        break;
      } else {
H
Haojun Liao 已提交
189
        qDebug("s-task:%s continue recover exec task ", pTask->id.idStr);
190 191
        continue;
      }
192
    }
H
Haojun Liao 已提交
193

S
Shengliang Guan 已提交
194
    SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
195 196 197 198 199 200 201 202
    if (qRes == NULL) {
      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }

    qRes->type = STREAM_INPUT__DATA_BLOCK;
    qRes->blocks = pRes;
dengyihao's avatar
dengyihao 已提交
203 204
    code = streamTaskOutput(pTask, qRes);
    if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {
dengyihao's avatar
dengyihao 已提交
205
      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
dengyihao's avatar
dengyihao 已提交
206
      taosFreeQitem(qRes);
dengyihao's avatar
dengyihao 已提交
207 208
      return code;
    }
L
Liu Jicong 已提交
209 210

    if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
211
      qDebug("s-task:%s scan exec dispatch blocks:%d", pTask->id.idStr, batchCnt);
L
Liu Jicong 已提交
212 213
      streamDispatch(pTask);
    }
214 215 216 217

    if (finished) {
      break;
    }
218 219 220 221 222
  }
  return 0;
}

#if 0
223 224 225 226
int32_t streamBatchExec(SStreamTask* pTask, int32_t batchLimit) {
  // fetch all queue item, merge according to batchLimit
  int32_t numOfItems = taosReadAllQitems(pTask->inputQueue1, pTask->inputQall);
  if (numOfItems == 0) {
227
    qDebug("task: %d, stream task exec over, queue empty", pTask->id.taskId);
228 229 230 231 232 233 234 235 236 237
    return 0;
  }
  SStreamQueueItem* pMerged = NULL;
  SStreamQueueItem* pItem = NULL;
  taosGetQitem(pTask->inputQall, (void**)&pItem);
  if (pItem == NULL) {
    if (pMerged != NULL) {
      // process merged item
    } else {
      return 0;
238
    }
239
  }
240

241 242 243 244 245
  // if drop
  if (pItem->type == STREAM_INPUT__DESTROY) {
    // set status drop
    return -1;
  }
246

247
  if (pTask->taskLevel == TASK_LEVEL__SINK) {
248
    ASSERT(((SStreamQueueItem*)pItem)->type == STREAM_INPUT__DATA_BLOCK);
249
    streamTaskOutput(pTask, (SStreamDataBlock*)pItem);
250 251
  }

252 253 254 255
  // exec impl

  // output
  // try dispatch
256 257
  return 0;
}
258
#endif
L
Liu Jicong 已提交
259

L
Liu Jicong 已提交
260
int32_t streamExecForAll(SStreamTask* pTask) {
dengyihao's avatar
dengyihao 已提交
261
  int32_t code = 0;
L
Liu Jicong 已提交
262
  while (1) {
263
    int32_t batchSize = 1;
L
liuyao 已提交
264
    int16_t times = 0;
265

266 267
    SStreamQueueItem* pInput = NULL;

268
    // merge multiple input data if possible in the input queue.
H
Haojun Liao 已提交
269 270
    qDebug("s-task:%s start to extract data block from inputQ", pTask->id.idStr);

L
Liu Jicong 已提交
271 272 273
    while (1) {
      SStreamQueueItem* qItem = streamQueueNextItem(pTask->inputQueue);
      if (qItem == NULL) {
L
liuyao 已提交
274 275 276
        if (pTask->taskLevel == TASK_LEVEL__SOURCE && batchSize < MIN_STREAM_EXEC_BATCH_NUM && times < 5) {
          times++;
          taosMsleep(1);
277
          qDebug("===stream===try again batchSize:%d", batchSize);
L
liuyao 已提交
278 279
          continue;
        }
280

L
liuyao 已提交
281
        qDebug("===stream===break batchSize:%d", batchSize);
L
Liu Jicong 已提交
282
        break;
L
Liu Jicong 已提交
283
      }
284 285 286

      if (pInput == NULL) {
        pInput = qItem;
287
        streamQueueProcessSuccess(pTask->inputQueue);
288
        if (pTask->taskLevel == TASK_LEVEL__SINK) {
L
Liu Jicong 已提交
289
          break;
L
Liu Jicong 已提交
290
        }
L
Liu Jicong 已提交
291
      } else {
292
        // todo we need to sort the data block, instead of just appending into the array list.
293 294
        void* newRet = NULL;
        if ((newRet = streamMergeQueueItem(pInput, qItem)) == NULL) {
L
Liu Jicong 已提交
295 296 297
          streamQueueProcessFail(pTask->inputQueue);
          break;
        } else {
298 299
          batchSize++;
          pInput = newRet;
L
Liu Jicong 已提交
300
          streamQueueProcessSuccess(pTask->inputQueue);
L
liuyao 已提交
301
          if (batchSize > MAX_STREAM_EXEC_BATCH_NUM) {
H
Haojun Liao 已提交
302
            qDebug("maximum batch limit:%d reached, processing, %s", MAX_STREAM_EXEC_BATCH_NUM, pTask->id.idStr);
5
54liuyao 已提交
303 304
            break;
          }
L
Liu Jicong 已提交
305
        }
L
Liu Jicong 已提交
306 307
      }
    }
308

309
    if (streamTaskShouldStop(&pTask->status)) {
310 311 312
      if (pInput) {
        streamFreeQitem(pInput);
      }
313

L
Liu Jicong 已提交
314
      return 0;
L
Liu Jicong 已提交
315
    }
L
Liu Jicong 已提交
316

317
    if (pInput == NULL) {
L
Liu Jicong 已提交
318 319 320
      break;
    }

321
    if (pTask->taskLevel == TASK_LEVEL__SINK) {
322
      ASSERT(pInput->type == STREAM_INPUT__DATA_BLOCK);
323
      qDebug("s-task:%s sink node start to sink result. numOfBlocks:%d", pTask->id.idStr, batchSize);
324
      streamTaskOutput(pTask, (SStreamDataBlock*)pInput);
L
Liu Jicong 已提交
325
      continue;
L
Liu Jicong 已提交
326
    }
L
Liu Jicong 已提交
327

L
Liu Jicong 已提交
328
    SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
329
    qDebug("s-task:%s start to execute, block batches:%d", pTask->id.idStr, batchSize);
L
Liu Jicong 已提交
330

331
    streamTaskExecImpl(pTask, pInput, pRes);
332

L
liuyao 已提交
333 334 335
    int64_t ckId = 0;
    int64_t dataVer = 0;
    qGetCheckpointVersion(pTask->exec.pExecutor, &dataVer, &ckId);
dengyihao's avatar
dengyihao 已提交
336
    if (ckId > pTask->chkInfo.id) {  // save it since the checkpoint is updated
337 338 339
      qDebug("s-task:%s exec end, start to update check point, ver from %" PRId64 " to %" PRId64
             ", checkPoint id:%" PRId64 " -> %" PRId64,
             pTask->id.idStr, pTask->chkInfo.version, dataVer, pTask->chkInfo.id, ckId);
L
Liu Jicong 已提交
340

dengyihao's avatar
dengyihao 已提交
341
      pTask->chkInfo = (SCheckpointInfo){.version = dataVer, .id = ckId, .currentVer = pTask->chkInfo.currentVer};
342

343
      taosWLockLatch(&pTask->pMeta->lock);
344

345
      streamMetaSaveTask(pTask->pMeta, pTask);
346
      if (streamMetaCommit(pTask->pMeta) < 0) {
347 348
        taosWUnLockLatch(&pTask->pMeta->lock);
        qError("s-task:%s failed to commit stream meta, since %s", pTask->id.idStr, terrstr());
349
        return -1;
350 351 352
      } else {
        taosWUnLockLatch(&pTask->pMeta->lock);
        qDebug("s-task:%s update checkpoint ver succeed", pTask->id.idStr);
353 354 355 356
      }
    } else {
      qDebug("s-task:%s exec end", pTask->id.idStr);
    }
357

L
Liu Jicong 已提交
358
    if (taosArrayGetSize(pRes) != 0) {
S
Shengliang Guan 已提交
359
      SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
L
Liu Jicong 已提交
360
      if (qRes == NULL) {
L
Liu Jicong 已提交
361
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
362
        streamFreeQitem(pInput);
L
Liu Jicong 已提交
363
        return -1;
L
Liu Jicong 已提交
364
      }
365

L
Liu Jicong 已提交
366 367
      qRes->type = STREAM_INPUT__DATA_BLOCK;
      qRes->blocks = pRes;
L
Liu Jicong 已提交
368

369 370
      if (((SStreamQueueItem*)pInput)->type == STREAM_INPUT__DATA_SUBMIT) {
        SStreamDataSubmit2* pSubmit = (SStreamDataSubmit2*)pInput;
L
Liu Jicong 已提交
371 372
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pSubmit->ver;
373 374
      } else if (((SStreamQueueItem*)pInput)->type == STREAM_INPUT__MERGED_SUBMIT) {
        SStreamMergedSubmit2* pMerged = (SStreamMergedSubmit2*)pInput;
L
Liu Jicong 已提交
375 376
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pMerged->ver;
L
Liu Jicong 已提交
377
      }
L
Liu Jicong 已提交
378

dengyihao's avatar
dengyihao 已提交
379 380 381
      code = streamTaskOutput(pTask, qRes);
      if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {
        // backpressure and record position
L
Liu Jicong 已提交
382
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
383
        streamFreeQitem(pInput);
L
Liu Jicong 已提交
384
        taosFreeQitem(qRes);
L
Liu Jicong 已提交
385 386
        return -1;
      }
L
Liu Jicong 已提交
387 388
    } else {
      taosArrayDestroy(pRes);
L
Liu Jicong 已提交
389
    }
390
    streamFreeQitem(pInput);
L
Liu Jicong 已提交
391
  }
L
Liu Jicong 已提交
392
  return 0;
L
Liu Jicong 已提交
393 394
}

L
Liu Jicong 已提交
395
int32_t streamTryExec(SStreamTask* pTask) {
396
  // this function may be executed by multi-threads, so status check is required.
L
Liu Jicong 已提交
397
  int8_t schedStatus =
398
      atomic_val_compare_exchange_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__WAITING, TASK_SCHED_STATUS__ACTIVE);
399

L
Liu Jicong 已提交
400 401 402
  if (schedStatus == TASK_SCHED_STATUS__WAITING) {
    int32_t code = streamExecForAll(pTask);
    if (code < 0) {
403
      atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__FAILED);
L
Liu Jicong 已提交
404 405
      return -1;
    }
406

407
    // todo the task should be commit here
408
    atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
409
    qDebug("s-task:%s exec completed", pTask->id.idStr);
L
Liu Jicong 已提交
410

411
    if (!taosQueueEmpty(pTask->inputQueue->queue) && (!streamTaskShouldStop(&pTask->status))) {
L
Liu Jicong 已提交
412
      streamSchedExec(pTask);
L
Liu Jicong 已提交
413 414
    }
  }
415

L
Liu Jicong 已提交
416
  return 0;
L
Liu Jicong 已提交
417
}