streamTask.c 7.2 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
/*
 * 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"
#include "tstream.h"

L
Liu Jicong 已提交
19
SStreamTask* tNewSStreamTask(int64_t streamId) {
L
Liu Jicong 已提交
20 21 22 23 24 25
  SStreamTask* pTask = (SStreamTask*)taosMemoryCalloc(1, sizeof(SStreamTask));
  if (pTask == NULL) {
    return NULL;
  }
  pTask->taskId = tGenIdPI32();
  pTask->streamId = streamId;
L
Liu Jicong 已提交
26
  pTask->execStatus = TASK_EXEC_STATUS__IDLE;
L
Liu Jicong 已提交
27 28 29 30 31 32
  pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
  pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;

  return pTask;
}

L
Liu Jicong 已提交
33 34 35 36
int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo) {
  if (tEncodeI32(pEncoder, pInfo->taskId) < 0) return -1;
  if (tEncodeI32(pEncoder, pInfo->nodeId) < 0) return -1;
  if (tEncodeI32(pEncoder, pInfo->childId) < 0) return -1;
L
Liu Jicong 已提交
37
  if (tEncodeI64(pEncoder, pInfo->processedVer) < 0) return -1;
L
Liu Jicong 已提交
38 39 40 41 42 43 44 45
  if (tEncodeSEpSet(pEncoder, &pInfo->epSet) < 0) return -1;
  return 0;
}

int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo) {
  if (tDecodeI32(pDecoder, &pInfo->taskId) < 0) return -1;
  if (tDecodeI32(pDecoder, &pInfo->nodeId) < 0) return -1;
  if (tDecodeI32(pDecoder, &pInfo->childId) < 0) return -1;
L
Liu Jicong 已提交
46
  if (tDecodeI64(pDecoder, &pInfo->processedVer) < 0) return -1;
L
Liu Jicong 已提交
47 48 49 50
  if (tDecodeSEpSet(pDecoder, &pInfo->epSet) < 0) return -1;
  return 0;
}

L
Liu Jicong 已提交
51 52 53 54
int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
  /*if (tStartEncode(pEncoder) < 0) return -1;*/
  if (tEncodeI64(pEncoder, pTask->streamId) < 0) return -1;
  if (tEncodeI32(pEncoder, pTask->taskId) < 0) return -1;
55
  if (tEncodeI8(pEncoder, pTask->isDataScan) < 0) return -1;
L
Liu Jicong 已提交
56 57 58 59
  if (tEncodeI8(pEncoder, pTask->execType) < 0) return -1;
  if (tEncodeI8(pEncoder, pTask->sinkType) < 0) return -1;
  if (tEncodeI8(pEncoder, pTask->dispatchType) < 0) return -1;
  if (tEncodeI16(pEncoder, pTask->dispatchMsgType) < 0) return -1;
60 61 62

  if (tEncodeI8(pEncoder, pTask->taskStatus) < 0) return -1;
  if (tEncodeI8(pEncoder, pTask->execStatus) < 0) return -1;
L
Liu Jicong 已提交
63

L
Liu Jicong 已提交
64
  if (tEncodeI32(pEncoder, pTask->selfChildId) < 0) return -1;
L
Liu Jicong 已提交
65 66 67
  if (tEncodeI32(pEncoder, pTask->nodeId) < 0) return -1;
  if (tEncodeSEpSet(pEncoder, &pTask->epSet) < 0) return -1;

L
Liu Jicong 已提交
68 69 70 71 72 73 74
  int32_t epSz = taosArrayGetSize(pTask->childEpInfo);
  if (tEncodeI32(pEncoder, epSz) < 0) return -1;
  for (int32_t i = 0; i < epSz; i++) {
    SStreamChildEpInfo* pInfo = taosArrayGetP(pTask->childEpInfo, i);
    if (tEncodeStreamEpInfo(pEncoder, pInfo) < 0) return -1;
  }

L
Liu Jicong 已提交
75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90
  if (pTask->execType != TASK_EXEC__NONE) {
    if (tEncodeCStr(pEncoder, pTask->exec.qmsg) < 0) return -1;
  }

  if (pTask->sinkType == TASK_SINK__TABLE) {
    if (tEncodeI64(pEncoder, pTask->tbSink.stbUid) < 0) return -1;
    if (tEncodeCStr(pEncoder, pTask->tbSink.stbFullName) < 0) return -1;
    if (tEncodeSSchemaWrapper(pEncoder, pTask->tbSink.pSchemaWrapper) < 0) return -1;
  } else if (pTask->sinkType == TASK_SINK__SMA) {
    if (tEncodeI64(pEncoder, pTask->smaSink.smaId) < 0) return -1;
  } else if (pTask->sinkType == TASK_SINK__FETCH) {
    if (tEncodeI8(pEncoder, pTask->fetchSink.reserved) < 0) return -1;
  } else {
    ASSERT(pTask->sinkType == TASK_SINK__NONE);
  }

L
Liu Jicong 已提交
91
  if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
L
Liu Jicong 已提交
92 93 94 95 96
    if (tEncodeI32(pEncoder, pTask->fixedEpDispatcher.taskId) < 0) return -1;
    if (tEncodeI32(pEncoder, pTask->fixedEpDispatcher.nodeId) < 0) return -1;
    if (tEncodeSEpSet(pEncoder, &pTask->fixedEpDispatcher.epSet) < 0) return -1;
  } else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
    if (tSerializeSUseDbRspImp(pEncoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
L
Liu Jicong 已提交
97
    if (tEncodeCStr(pEncoder, pTask->shuffleDispatcher.stbFullName) < 0) return -1;
L
Liu Jicong 已提交
98
  }
99
  if (tEncodeI64(pEncoder, pTask->triggerParam) < 0) return -1;
L
Liu Jicong 已提交
100 101 102 103 104 105 106 107 108

  /*tEndEncode(pEncoder);*/
  return pEncoder->pos;
}

int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
  /*if (tStartDecode(pDecoder) < 0) return -1;*/
  if (tDecodeI64(pDecoder, &pTask->streamId) < 0) return -1;
  if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1;
109
  if (tDecodeI8(pDecoder, &pTask->isDataScan) < 0) return -1;
L
Liu Jicong 已提交
110 111 112 113
  if (tDecodeI8(pDecoder, &pTask->execType) < 0) return -1;
  if (tDecodeI8(pDecoder, &pTask->sinkType) < 0) return -1;
  if (tDecodeI8(pDecoder, &pTask->dispatchType) < 0) return -1;
  if (tDecodeI16(pDecoder, &pTask->dispatchMsgType) < 0) return -1;
114 115 116

  if (tDecodeI8(pDecoder, &pTask->taskStatus) < 0) return -1;
  if (tDecodeI8(pDecoder, &pTask->execStatus) < 0) return -1;
L
Liu Jicong 已提交
117

L
Liu Jicong 已提交
118
  if (tDecodeI32(pDecoder, &pTask->selfChildId) < 0) return -1;
L
Liu Jicong 已提交
119 120 121
  if (tDecodeI32(pDecoder, &pTask->nodeId) < 0) return -1;
  if (tDecodeSEpSet(pDecoder, &pTask->epSet) < 0) return -1;

L
Liu Jicong 已提交
122 123 124 125 126 127 128 129 130 131
  int32_t epSz;
  if (tDecodeI32(pDecoder, &epSz) < 0) return -1;
  pTask->childEpInfo = taosArrayInit(epSz, sizeof(void*));
  for (int32_t i = 0; i < epSz; i++) {
    SStreamChildEpInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamChildEpInfo));
    if (pInfo == NULL) return -1;
    if (tDecodeStreamEpInfo(pDecoder, pInfo) < 0) return -1;
    taosArrayPush(pTask->childEpInfo, &pInfo);
  }

L
Liu Jicong 已提交
132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149
  if (pTask->execType != TASK_EXEC__NONE) {
    if (tDecodeCStrAlloc(pDecoder, &pTask->exec.qmsg) < 0) return -1;
  }

  if (pTask->sinkType == TASK_SINK__TABLE) {
    if (tDecodeI64(pDecoder, &pTask->tbSink.stbUid) < 0) return -1;
    if (tDecodeCStrTo(pDecoder, pTask->tbSink.stbFullName) < 0) return -1;
    pTask->tbSink.pSchemaWrapper = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
    if (pTask->tbSink.pSchemaWrapper == NULL) return -1;
    if (tDecodeSSchemaWrapper(pDecoder, pTask->tbSink.pSchemaWrapper) < 0) return -1;
  } else if (pTask->sinkType == TASK_SINK__SMA) {
    if (tDecodeI64(pDecoder, &pTask->smaSink.smaId) < 0) return -1;
  } else if (pTask->sinkType == TASK_SINK__FETCH) {
    if (tDecodeI8(pDecoder, &pTask->fetchSink.reserved) < 0) return -1;
  } else {
    ASSERT(pTask->sinkType == TASK_SINK__NONE);
  }

L
Liu Jicong 已提交
150
  if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
L
Liu Jicong 已提交
151 152 153 154 155
    if (tDecodeI32(pDecoder, &pTask->fixedEpDispatcher.taskId) < 0) return -1;
    if (tDecodeI32(pDecoder, &pTask->fixedEpDispatcher.nodeId) < 0) return -1;
    if (tDecodeSEpSet(pDecoder, &pTask->fixedEpDispatcher.epSet) < 0) return -1;
  } else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
    if (tDeserializeSUseDbRspImp(pDecoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
L
Liu Jicong 已提交
156
    if (tDecodeCStrTo(pDecoder, pTask->shuffleDispatcher.stbFullName) < 0) return -1;
L
Liu Jicong 已提交
157
  }
158
  if (tDecodeI64(pDecoder, &pTask->triggerParam) < 0) return -1;
L
Liu Jicong 已提交
159 160 161 162 163 164 165 166 167

  /*tEndDecode(pDecoder);*/
  return 0;
}

void tFreeSStreamTask(SStreamTask* pTask) {
  streamQueueClose(pTask->inputQueue);
  streamQueueClose(pTask->outputQueue);
  if (pTask->exec.qmsg) taosMemoryFree(pTask->exec.qmsg);
168
  if (pTask->exec.executor) qDestroyTask(pTask->exec.executor);
L
Liu Jicong 已提交
169 170
  taosMemoryFree(pTask);
}