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

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

  // set input
L
Liu Jicong 已提交
23
  const SStreamQueueItem* pItem = (const SStreamQueueItem*)data;
L
Liu Jicong 已提交
24
  if (pItem->type == STREAM_INPUT__GET_RES) {
L
Liu Jicong 已提交
25
    const SStreamTrigger* pTrigger = (const SStreamTrigger*)data;
L
Liu Jicong 已提交
26
    qSetMultiStreamInput(exec, pTrigger->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
27
  } else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
28
    ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
L
Liu Jicong 已提交
29
    const SStreamDataSubmit* pSubmit = (const SStreamDataSubmit*)data;
J
jiajingbin 已提交
30
    qDebug("task %d %p set submit input %p %p %d 1", pTask->taskId, pTask, pSubmit, pSubmit->data, *pSubmit->dataRef);
L
Liu Jicong 已提交
31
    qSetMultiStreamInput(exec, pSubmit->data, 1, STREAM_INPUT__DATA_SUBMIT);
32
  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
L
Liu Jicong 已提交
33 34
    const SStreamDataBlock* pBlock = (const SStreamDataBlock*)data;
    SArray*                 blocks = pBlock->blocks;
L
Liu Jicong 已提交
35
    qDebug("task %d %p set ssdata input", pTask->taskId, pTask);
L
Liu Jicong 已提交
36
    qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_INPUT__DATA_BLOCK);
37
  } else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {
L
Liu Jicong 已提交
38 39
    const SStreamMergedSubmit* pMerged = (const SStreamMergedSubmit*)data;
    SArray*                    blocks = pMerged->reqs;
L
Liu Jicong 已提交
40
    qDebug("task %d %p set submit input (merged), batch num: %d", pTask->taskId, pTask, (int32_t)blocks->size);
L
Liu Jicong 已提交
41
    qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_INPUT__MERGED_SUBMIT);
L
Liu Jicong 已提交
42 43 44
  } else if (pItem->type == STREAM_INPUT__REF_DATA_BLOCK) {
    const SStreamRefDataBlock* pRefBlock = (const SStreamRefDataBlock*)data;
    qSetMultiStreamInput(exec, pRefBlock->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
45
  } else {
46
    ASSERT(0);
L
Liu Jicong 已提交
47 48 49 50 51 52
  }

  // exec
  while (1) {
    SSDataBlock* output = NULL;
    uint64_t     ts = 0;
L
Liu Jicong 已提交
53
    if ((code = qExecTask(exec, &output, &ts)) < 0) {
54
      /*ASSERT(false);*/
L
Liu Jicong 已提交
55 56
      qError("unexpected stream execution, stream %" PRId64 " task: %d,  since %s", pTask->streamId, pTask->taskId,
             terrstr());
L
Liu Jicong 已提交
57
      continue;
L
Liu Jicong 已提交
58
    }
59
    if (output == NULL) {
5
54liuyao 已提交
60
      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
L
Liu Jicong 已提交
61 62
        SSDataBlock             block = {0};
        const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)data;
63
        ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
L
Liu Jicong 已提交
64
        assignOneDataBlock(&block, taosArrayGet(pRetrieveBlock->blocks, 0));
L
Liu Jicong 已提交
65
        block.info.type = STREAM_PULL_OVER;
L
Liu Jicong 已提交
66 67
        block.info.childId = pTask->selfChildId;
        taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
68

S
Shengliang Guan 已提交
69
        qDebug("task %d(child %d) processed retrieve, reqId %" PRId64, pTask->taskId, pTask->selfChildId,
L
Liu Jicong 已提交
70
               pRetrieveBlock->reqId);
71 72 73
      }
      break;
    }
L
Liu Jicong 已提交
74 75 76 77 78 79 80 81

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

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

L
Liu Jicong 已提交
84 85 86 87
    SSDataBlock block = {0};
    assignOneDataBlock(&block, output);
    block.info.childId = pTask->selfChildId;
    taosArrayPush(pRes, &block);
L
Liu Jicong 已提交
88 89 90 91
  }
  return 0;
}

92
int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
93
  ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
94 95 96

  void* exec = pTask->exec.executor;

L
Liu Jicong 已提交
97
  qSetStreamOpOpen(exec);
L
Liu Jicong 已提交
98
  bool finished = false;
L
Liu Jicong 已提交
99

100 101 102 103 104 105 106 107 108 109 110 111
  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) {
      SSDataBlock* output = NULL;
      uint64_t     ts = 0;
      if (qExecTask(exec, &output, &ts) < 0) {
112
        ASSERT(0);
113
      }
L
Liu Jicong 已提交
114 115 116 117
      if (output == NULL) {
        finished = true;
        break;
      }
118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139

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

      if (++batchCnt >= batchSz) break;
    }
    if (taosArrayGetSize(pRes) == 0) {
      taosArrayDestroy(pRes);
      break;
    }
    SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
    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 已提交
140 141 142 143

    if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
      streamDispatch(pTask);
    }
L
Liu Jicong 已提交
144
    if (finished) break;
145 146 147 148 149
  }
  return 0;
}

#if 0
150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
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) {
    qDebug("task: %d, stream task exec over, queue empty", pTask->taskId);
    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;
165
    }
166
  }
167

168 169 170 171 172
  // if drop
  if (pItem->type == STREAM_INPUT__DESTROY) {
    // set status drop
    return -1;
  }
173

174
  if (pTask->taskLevel == TASK_LEVEL__SINK) {
175
    ASSERT(((SStreamQueueItem*)pItem)->type == STREAM_INPUT__DATA_BLOCK);
176
    streamTaskOutput(pTask, (SStreamDataBlock*)pItem);
177 178
  }

179 180 181 182
  // exec impl

  // output
  // try dispatch
183 184
  return 0;
}
185
#endif
L
Liu Jicong 已提交
186

L
Liu Jicong 已提交
187
int32_t streamExecForAll(SStreamTask* pTask) {
L
Liu Jicong 已提交
188
  while (1) {
L
Liu Jicong 已提交
189
    int32_t batchCnt = 1;
190
    void*   input = NULL;
L
Liu Jicong 已提交
191 192 193
    while (1) {
      SStreamQueueItem* qItem = streamQueueNextItem(pTask->inputQueue);
      if (qItem == NULL) {
L
Liu Jicong 已提交
194
        qDebug("stream task exec over, queue empty, task: %d", pTask->taskId);
L
Liu Jicong 已提交
195
        break;
L
Liu Jicong 已提交
196
      }
197 198
      if (input == NULL) {
        input = qItem;
199
        streamQueueProcessSuccess(pTask->inputQueue);
200
        if (pTask->taskLevel == TASK_LEVEL__SINK) {
L
Liu Jicong 已提交
201
          break;
L
Liu Jicong 已提交
202
        }
L
Liu Jicong 已提交
203
      } else {
J
jiacy-jcy 已提交
204
        void* newRet;
205
        if ((newRet = streamMergeQueueItem(input, qItem)) == NULL) {
L
Liu Jicong 已提交
206 207 208
          streamQueueProcessFail(pTask->inputQueue);
          break;
        } else {
L
Liu Jicong 已提交
209
          batchCnt++;
210
          input = newRet;
L
Liu Jicong 已提交
211 212
          streamQueueProcessSuccess(pTask->inputQueue);
        }
L
Liu Jicong 已提交
213 214
      }
    }
215

L
Liu Jicong 已提交
216
    if (pTask->taskStatus == TASK_STATUS__DROPPING) {
217
      if (input) streamFreeQitem(input);
L
Liu Jicong 已提交
218
      return 0;
L
Liu Jicong 已提交
219
    }
L
Liu Jicong 已提交
220

221
    if (input == NULL) {
L
Liu Jicong 已提交
222 223 224
      break;
    }

225
    if (pTask->taskLevel == TASK_LEVEL__SINK) {
226
      ASSERT(((SStreamQueueItem*)input)->type == STREAM_INPUT__DATA_BLOCK);
227
      streamTaskOutput(pTask, input);
L
Liu Jicong 已提交
228
      continue;
L
Liu Jicong 已提交
229
    }
L
Liu Jicong 已提交
230

L
Liu Jicong 已提交
231 232
    SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));

L
Liu Jicong 已提交
233
    qDebug("stream task %d exec begin, msg batch: %d", pTask->taskId, batchCnt);
234
    streamTaskExecImpl(pTask, input, pRes);
L
Liu Jicong 已提交
235
    qDebug("stream task %d exec end", pTask->taskId);
236

L
Liu Jicong 已提交
237 238 239
    if (taosArrayGetSize(pRes) != 0) {
      SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
      if (qRes == NULL) {
L
Liu Jicong 已提交
240
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
241
        streamFreeQitem(input);
L
Liu Jicong 已提交
242
        return -1;
L
Liu Jicong 已提交
243 244 245
      }
      qRes->type = STREAM_INPUT__DATA_BLOCK;
      qRes->blocks = pRes;
L
Liu Jicong 已提交
246

247 248
      if (((SStreamQueueItem*)input)->type == STREAM_INPUT__DATA_SUBMIT) {
        SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)input;
L
Liu Jicong 已提交
249 250
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pSubmit->ver;
251 252
      } else if (((SStreamQueueItem*)input)->type == STREAM_INPUT__MERGED_SUBMIT) {
        SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)input;
L
Liu Jicong 已提交
253 254
        qRes->childId = pTask->selfChildId;
        qRes->sourceVer = pMerged->ver;
L
Liu Jicong 已提交
255
      }
L
Liu Jicong 已提交
256 257 258

      if (streamTaskOutput(pTask, qRes) < 0) {
        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
259
        streamFreeQitem(input);
L
Liu Jicong 已提交
260
        taosFreeQitem(qRes);
L
Liu Jicong 已提交
261 262
        return -1;
      }
L
Liu Jicong 已提交
263 264
    } else {
      taosArrayDestroy(pRes);
L
Liu Jicong 已提交
265
    }
266
    streamFreeQitem(input);
L
Liu Jicong 已提交
267
  }
L
Liu Jicong 已提交
268
  return 0;
L
Liu Jicong 已提交
269 270
}

L
Liu Jicong 已提交
271 272 273 274 275 276 277 278 279 280
int32_t streamTryExec(SStreamTask* pTask) {
  int8_t schedStatus =
      atomic_val_compare_exchange_8(&pTask->schedStatus, TASK_SCHED_STATUS__WAITING, TASK_SCHED_STATUS__ACTIVE);
  if (schedStatus == TASK_SCHED_STATUS__WAITING) {
    int32_t code = streamExecForAll(pTask);
    if (code < 0) {
      atomic_store_8(&pTask->schedStatus, TASK_SCHED_STATUS__FAILED);
      return -1;
    }
    atomic_store_8(&pTask->schedStatus, TASK_SCHED_STATUS__INACTIVE);
L
Liu Jicong 已提交
281

L
Liu Jicong 已提交
282 283
    if (!taosQueueEmpty(pTask->inputQueue->queue)) {
      streamSchedExec(pTask);
L
Liu Jicong 已提交
284 285
    }
  }
L
Liu Jicong 已提交
286
  return 0;
L
Liu Jicong 已提交
287
}