streamExec.c 12.1 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

5
54liuyao 已提交
18 19
#define STREAM_EXEC_MAX_BATCH_NUM 100

L
Liu Jicong 已提交
20
static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* pRes) {
21 22 23
  int32_t code = TSDB_CODE_SUCCESS;
  void*   pExecutor = pTask->exec.pExecutor;

24
  while (pTask->taskLevel == TASK_LEVEL__SOURCE) {
25
    int8_t status = atomic_load_8(&pTask->status.taskStatus);
26 27
    if (status != TASK_STATUS__NORMAL && status != TASK_STATUS__RESTORE) {
      qError("stream task wait for the end of fill history, s-task:%s, status:%d", pTask->id.idStr,
28
             atomic_load_8(&pTask->status.taskStatus));
29 30 31 32
      taosMsleep(2);
    } else {
      break;
    }
5
54liuyao 已提交
33
  }
L
Liu Jicong 已提交
34 35

  // set input
L
Liu Jicong 已提交
36
  const SStreamQueueItem* pItem = (const SStreamQueueItem*)data;
L
Liu Jicong 已提交
37
  if (pItem->type == STREAM_INPUT__GET_RES) {
L
Liu Jicong 已提交
38
    const SStreamTrigger* pTrigger = (const SStreamTrigger*)data;
39
    qSetMultiStreamInput(pExecutor, pTrigger->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
40
  } else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
41
    ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
L
Liu Jicong 已提交
42
    const SStreamDataSubmit2* pSubmit = (const SStreamDataSubmit2*)data;
43
    qSetMultiStreamInput(pExecutor, &pSubmit->submit, 1, STREAM_INPUT__DATA_SUBMIT);
44 45
    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);
46
  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
L
Liu Jicong 已提交
47
    const SStreamDataBlock* pBlock = (const SStreamDataBlock*)data;
48 49 50 51 52

    SArray* pBlockList = pBlock->blocks;
    int32_t numOfBlocks = taosArrayGetSize(pBlockList);
    qDebug("s-task:%s set sdata blocks as input num:%d, ver:%"PRId64, pTask->id.idStr, numOfBlocks, pBlock->sourceVer);
    qSetMultiStreamInput(pExecutor, pBlockList->pData, numOfBlocks, STREAM_INPUT__DATA_BLOCK);
53
  } else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {
L
Liu Jicong 已提交
54
    const SStreamMergedSubmit2* pMerged = (const SStreamMergedSubmit2*)data;
55 56 57 58 59

    SArray* pBlockList = pMerged->submits;
    int32_t numOfBlocks = taosArrayGetSize(pBlockList);
    qDebug("st-task:%s %p set submit input (merged), batch num:%d", pTask->id.idStr, pTask, numOfBlocks);
    qSetMultiStreamInput(pExecutor, pBlockList->pData, numOfBlocks, STREAM_INPUT__MERGED_SUBMIT);
L
Liu Jicong 已提交
60 61
  } else if (pItem->type == STREAM_INPUT__REF_DATA_BLOCK) {
    const SStreamRefDataBlock* pRefBlock = (const SStreamRefDataBlock*)data;
62
    qSetMultiStreamInput(pExecutor, pRefBlock->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
63
  } else {
64
    ASSERT(0);
L
Liu Jicong 已提交
65 66
  }

67
  // pExecutor
L
Liu Jicong 已提交
68
  while (1) {
69
    if (pTask->status.taskStatus == TASK_STATUS__DROPPING) {
L
Liu Jicong 已提交
70 71 72
      return 0;
    }

L
Liu Jicong 已提交
73 74
    SSDataBlock* output = NULL;
    uint64_t     ts = 0;
75
    if ((code = qExecTask(pExecutor, &output, &ts)) < 0) {
5
54liuyao 已提交
76
      if (code == TSDB_CODE_QRY_IN_EXEC) {
77
        resetTaskInfo(pExecutor);
5
54liuyao 已提交
78
      }
79 80

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

84
    if (output == NULL) {
5
54liuyao 已提交
85
      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
86 87
        SSDataBlock block = {0};

L
Liu Jicong 已提交
88
        const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)data;
89
        ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
90

L
Liu Jicong 已提交
91
        assignOneDataBlock(&block, taosArrayGet(pRetrieveBlock->blocks, 0));
L
Liu Jicong 已提交
92
        block.info.type = STREAM_PULL_OVER;
L
Liu Jicong 已提交
93 94
        block.info.childId = pTask->selfChildId;
        taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
95

96
        qDebug("task %d(child %d) processed retrieve, reqId %" PRId64, pTask->id.taskId, pTask->selfChildId,
L
Liu Jicong 已提交
97
               pRetrieveBlock->reqId);
98 99 100
      }
      break;
    }
L
Liu Jicong 已提交
101 102 103 104 105 106 107 108

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

109
    qDebug("task %d(child %d) executed and get block", pTask->id.taskId, pTask->selfChildId);
J
jiajingbin 已提交
110

L
Liu Jicong 已提交
111 112 113 114
    SSDataBlock block = {0};
    assignOneDataBlock(&block, output);
    block.info.childId = pTask->selfChildId;
    taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
115
  }
116

L
Liu Jicong 已提交
117 118 119
  return 0;
}

120
int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
121
  ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
122

123
  void* exec = pTask->exec.pExecutor;
124

L
Liu Jicong 已提交
125
  qSetStreamOpOpen(exec);
L
Liu Jicong 已提交
126
  bool finished = false;
L
Liu Jicong 已提交
127

128 129 130 131 132 133 134 135 136
  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) {
137
      if (atomic_load_8(&pTask->status.taskStatus) == TASK_STATUS__DROPPING) {
5
54liuyao 已提交
138
        taosArrayDestroy(pRes);
L
Liu Jicong 已提交
139 140 141
        return 0;
      }

142 143 144
      SSDataBlock* output = NULL;
      uint64_t     ts = 0;
      if (qExecTask(exec, &output, &ts) < 0) {
5
54liuyao 已提交
145
        continue;
146
      }
L
Liu Jicong 已提交
147
      if (output == NULL) {
L
Liu Jicong 已提交
148 149 150 151 152
        if (qStreamRecoverScanFinished(exec)) {
          finished = true;
        } else {
          qSetStreamOpOpen(exec);
        }
L
Liu Jicong 已提交
153 154
        break;
      }
155 156 157 158 159 160

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

L
Liu Jicong 已提交
161 162
      batchCnt++;

163
      qDebug("task %d scan exec block num %d, block limit %d", pTask->id.taskId, batchCnt, batchSz);
L
Liu Jicong 已提交
164 165

      if (batchCnt >= batchSz) break;
166 167
    }
    if (taosArrayGetSize(pRes) == 0) {
168 169
      if (finished) {
        taosArrayDestroy(pRes);
170
        qDebug("task %d finish recover exec task ", pTask->id.taskId);
171 172
        break;
      } else {
173
        qDebug("task %d continue recover exec task ", pTask->id.taskId);
174 175
        continue;
      }
176
    }
S
Shengliang Guan 已提交
177
    SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
178 179 180 181 182 183 184 185 186
    if (qRes == NULL) {
      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }

    qRes->type = STREAM_INPUT__DATA_BLOCK;
    qRes->blocks = pRes;
    streamTaskOutput(pTask, qRes);
L
Liu Jicong 已提交
187 188

    if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
189
      qDebug("task %d scan exec dispatch block num %d", pTask->id.taskId, batchCnt);
L
Liu Jicong 已提交
190 191
      streamDispatch(pTask);
    }
192 193 194 195

    if (finished) {
      break;
    }
196 197 198 199 200
  }
  return 0;
}

#if 0
201 202 203 204
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) {
205
    qDebug("task: %d, stream task exec over, queue empty", pTask->id.taskId);
206 207 208 209 210 211 212 213 214 215
    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;
216
    }
217
  }
218

219 220 221 222 223
  // if drop
  if (pItem->type == STREAM_INPUT__DESTROY) {
    // set status drop
    return -1;
  }
224

225
  if (pTask->taskLevel == TASK_LEVEL__SINK) {
226
    ASSERT(((SStreamQueueItem*)pItem)->type == STREAM_INPUT__DATA_BLOCK);
227
    streamTaskOutput(pTask, (SStreamDataBlock*)pItem);
228 229
  }

230 231 232 233
  // exec impl

  // output
  // try dispatch
234 235
  return 0;
}
236
#endif
L
Liu Jicong 已提交
237

L
Liu Jicong 已提交
238
int32_t streamExecForAll(SStreamTask* pTask) {
L
Liu Jicong 已提交
239
  while (1) {
240 241 242 243
    int32_t batchSize = 1;
    void*   pInput = NULL;

    // merge multiple input data if possible in the input queue.
L
Liu Jicong 已提交
244 245 246
    while (1) {
      SStreamQueueItem* qItem = streamQueueNextItem(pTask->inputQueue);
      if (qItem == NULL) {
247
//        qDebug("s-task:%s extract data from input queue, queue is empty, abort", pTask->id.idStr);
L
Liu Jicong 已提交
248
        break;
L
Liu Jicong 已提交
249
      }
250 251 252

      if (pInput == NULL) {
        pInput = qItem;
253
        streamQueueProcessSuccess(pTask->inputQueue);
254
        if (pTask->taskLevel == TASK_LEVEL__SINK) {
L
Liu Jicong 已提交
255
          break;
L
Liu Jicong 已提交
256
        }
L
Liu Jicong 已提交
257
      } else {
258 259
        void* newRet = NULL;
        if ((newRet = streamMergeQueueItem(pInput, qItem)) == NULL) {
L
Liu Jicong 已提交
260 261 262
          streamQueueProcessFail(pTask->inputQueue);
          break;
        } else {
263 264
          batchSize++;
          pInput = newRet;
L
Liu Jicong 已提交
265
          streamQueueProcessSuccess(pTask->inputQueue);
266
          if (batchSize > STREAM_EXEC_MAX_BATCH_NUM) {
5
54liuyao 已提交
267 268
            break;
          }
L
Liu Jicong 已提交
269
        }
L
Liu Jicong 已提交
270 271
      }
    }
272

273
    if (pTask->status.taskStatus == TASK_STATUS__DROPPING) {
274 275 276
      if (pInput) {
        streamFreeQitem(pInput);
      }
L
Liu Jicong 已提交
277
      return 0;
L
Liu Jicong 已提交
278
    }
L
Liu Jicong 已提交
279

280
    if (pInput == NULL) {
L
Liu Jicong 已提交
281 282 283
      break;
    }

284
    if (pTask->taskLevel == TASK_LEVEL__SINK) {
285
      ASSERT(((SStreamQueueItem*)pInput)->type == STREAM_INPUT__DATA_BLOCK);
286
      qDebug("s-task:%s sink node start to sink result. numOfBlocks:%d", pTask->id.idStr, batchSize);
287
      streamTaskOutput(pTask, pInput);
L
Liu Jicong 已提交
288
      continue;
L
Liu Jicong 已提交
289
    }
L
Liu Jicong 已提交
290

L
Liu Jicong 已提交
291
    SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
292
    qDebug("s-task:%s exec begin, numOfBlocks:%d", pTask->id.idStr, batchSize);
L
Liu Jicong 已提交
293

294
    streamTaskExecImpl(pTask, pInput, pRes);
295

L
liuyao 已提交
296 297 298
    int64_t ckId = 0;
    int64_t dataVer = 0;
    qGetCheckpointVersion(pTask->exec.pExecutor, &dataVer, &ckId);
299
    if (dataVer > pTask->chkInfo.version) {    // save it since the checkpoint is updated
300 301 302 303
      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);

304
      pTask->chkInfo = (SCheckpointInfo) {.version = dataVer, .id = ckId};
305

306 307
      taosWLockLatch(&pTask->pMeta->lock);
      streamMetaSaveTask(pTask->pMeta, pTask);
308
      if (streamMetaCommit(pTask->pMeta) < 0) {
309 310
        taosWUnLockLatch(&pTask->pMeta->lock);
        qError("s-task:%s failed to commit stream meta, since %s", pTask->id.idStr, terrstr());
311
        return -1;
312 313 314
      } else {
        taosWUnLockLatch(&pTask->pMeta->lock);
        qDebug("s-task:%s update checkpoint ver succeed", pTask->id.idStr);
315 316 317 318
      }
    } else {
      qDebug("s-task:%s exec end", pTask->id.idStr);
    }
319

L
Liu Jicong 已提交
320
    if (taosArrayGetSize(pRes) != 0) {
S
Shengliang Guan 已提交
321
      SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
L
Liu Jicong 已提交
322
      if (qRes == NULL) {
L
Liu Jicong 已提交
323
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
324
        streamFreeQitem(pInput);
L
Liu Jicong 已提交
325
        return -1;
L
Liu Jicong 已提交
326
      }
327

L
Liu Jicong 已提交
328 329
      qRes->type = STREAM_INPUT__DATA_BLOCK;
      qRes->blocks = pRes;
L
Liu Jicong 已提交
330

331 332
      if (((SStreamQueueItem*)pInput)->type == STREAM_INPUT__DATA_SUBMIT) {
        SStreamDataSubmit2* pSubmit = (SStreamDataSubmit2*)pInput;
L
Liu Jicong 已提交
333 334
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pSubmit->ver;
335 336
      } else if (((SStreamQueueItem*)pInput)->type == STREAM_INPUT__MERGED_SUBMIT) {
        SStreamMergedSubmit2* pMerged = (SStreamMergedSubmit2*)pInput;
L
Liu Jicong 已提交
337 338
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pMerged->ver;
L
Liu Jicong 已提交
339
      }
L
Liu Jicong 已提交
340 341 342

      if (streamTaskOutput(pTask, qRes) < 0) {
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
343
        streamFreeQitem(pInput);
L
Liu Jicong 已提交
344
        taosFreeQitem(qRes);
L
Liu Jicong 已提交
345 346
        return -1;
      }
L
Liu Jicong 已提交
347 348
    } else {
      taosArrayDestroy(pRes);
L
Liu Jicong 已提交
349
    }
350
    streamFreeQitem(pInput);
L
Liu Jicong 已提交
351
  }
L
Liu Jicong 已提交
352
  return 0;
L
Liu Jicong 已提交
353 354
}

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

L
Liu Jicong 已提交
360 361 362
  if (schedStatus == TASK_SCHED_STATUS__WAITING) {
    int32_t code = streamExecForAll(pTask);
    if (code < 0) {
363
      atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__FAILED);
L
Liu Jicong 已提交
364 365
      return -1;
    }
366

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

371
    if (!taosQueueEmpty(pTask->inputQueue->queue) && (pTask->status.taskStatus != TASK_STATUS__DROPPING)) {
L
Liu Jicong 已提交
372
      streamSchedExec(pTask);
L
Liu Jicong 已提交
373 374
    }
  }
375

L
Liu Jicong 已提交
376
  return 0;
L
Liu Jicong 已提交
377
}