streamData.c 7.2 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/>.
 */

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

L
Liu Jicong 已提交
18 19
int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData) {
  int32_t blockNum = pReq->blockNum;
20
  SArray* pArray = taosArrayInit_s(sizeof(SSDataBlock), blockNum);
L
Liu Jicong 已提交
21 22 23 24 25 26 27 28
  if (pArray == NULL) {
    return -1;
  }

  ASSERT(pReq->blockNum == taosArrayGetSize(pReq->data));
  ASSERT(pReq->blockNum == taosArrayGetSize(pReq->dataLen));

  for (int32_t i = 0; i < blockNum; i++) {
5
54liuyao 已提交
29
    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*) taosArrayGetP(pReq->data, i);
L
Liu Jicong 已提交
30
    SSDataBlock*       pDataBlock = taosArrayGet(pArray, i);
31
    blockDecode(pDataBlock, pRetrieve->data);
L
Liu Jicong 已提交
32
    // TODO: refactor
33 34
    pDataBlock->info.window.skey = be64toh(pRetrieve->skey);
    pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);
35
    pDataBlock->info.version = be64toh(pRetrieve->version);
36
    pDataBlock->info.watermark = be64toh(pRetrieve->watermark);
37
    memcpy(pDataBlock->info.parTbName, pRetrieve->parTbName, TSDB_TABLE_NAME_LEN);
38

L
Liu Jicong 已提交
39
    pDataBlock->info.type = pRetrieve->streamBlockType;
L
Liu Jicong 已提交
40 41 42 43 44 45 46 47 48 49
    pDataBlock->info.childId = pReq->upstreamChildId;
  }
  pData->blocks = pArray;
  return 0;
}

int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock* pData) {
  SArray* pArray = taosArrayInit(1, sizeof(SSDataBlock));
  if (pArray == NULL) {
    return -1;
L
Liu Jicong 已提交
50
  }
51

H
Haojun Liao 已提交
52
  taosArrayPush(pArray, &(SSDataBlock){0});
L
Liu Jicong 已提交
53
  SRetrieveTableRsp* pRetrieve = pReq->pRetrieve;
54
  SSDataBlock*       pDataBlock = taosArrayGet(pArray, 0);
55
  blockDecode(pDataBlock, pRetrieve->data);
56

57 58 59
  // TODO: refactor
  pDataBlock->info.window.skey = be64toh(pRetrieve->skey);
  pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);
60
  pDataBlock->info.version = be64toh(pRetrieve->version);
61 62 63

  pDataBlock->info.type = pRetrieve->streamBlockType;

L
Liu Jicong 已提交
64
  pData->reqId = pReq->reqId;
L
Liu Jicong 已提交
65
  pData->blocks = pArray;
L
Liu Jicong 已提交
66

L
Liu Jicong 已提交
67 68 69
  return 0;
}

70
SStreamDataSubmit2* streamDataSubmitNew(SPackedData submit, int32_t type) {
H
Haojun Liao 已提交
71
  SStreamDataSubmit2* pDataSubmit = (SStreamDataSubmit2*)taosAllocateQitem(sizeof(SStreamDataSubmit2), DEF_QITEM, submit.msgLen);
72 73 74 75
  if (pDataSubmit == NULL) {
    return NULL;
  }

L
Liu Jicong 已提交
76
  pDataSubmit->dataRef = (int32_t*)taosMemoryMalloc(sizeof(int32_t));
77 78 79 80 81
  if (pDataSubmit->dataRef == NULL) {
    taosFreeQitem(pDataSubmit);
    return NULL;
  }

L
Liu Jicong 已提交
82
  pDataSubmit->submit = submit;
83
  *pDataSubmit->dataRef = 1;   // initialize the reference count to be 1
84
  pDataSubmit->type = type;
85

L
Liu Jicong 已提交
86 87 88
  return pDataSubmit;
}

89 90 91 92 93 94 95 96 97 98
void streamDataSubmitDestroy(SStreamDataSubmit2* pDataSubmit) {
  int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1);
  ASSERT(ref >= 0 && pDataSubmit->type == STREAM_INPUT__DATA_SUBMIT);

  if (ref == 0) {
    taosMemoryFree(pDataSubmit->submit.msgStr);
    taosMemoryFree(pDataSubmit->dataRef);
  }
}

L
Liu Jicong 已提交
99
SStreamMergedSubmit2* streamMergedSubmitNew() {
100
  SStreamMergedSubmit2* pMerged = (SStreamMergedSubmit2*)taosAllocateQitem(sizeof(SStreamMergedSubmit2), DEF_QITEM, 0);
101 102 103
  if (pMerged == NULL) {
    return NULL;
  }
104

L
Liu Jicong 已提交
105
  pMerged->submits = taosArrayInit(0, sizeof(SPackedData));
106
  pMerged->dataRefs = taosArrayInit(0, sizeof(void*));
107 108 109 110 111 112 113 114

  if (pMerged->dataRefs == NULL || pMerged->submits == NULL) {
    taosArrayDestroy(pMerged->submits);
    taosArrayDestroy(pMerged->dataRefs);
    taosFreeQitem(pMerged);
    return NULL;
  }

J
jiacy-jcy 已提交
115
  pMerged->type = STREAM_INPUT__MERGED_SUBMIT;
116 117 118
  return pMerged;
}

L
Liu Jicong 已提交
119
int32_t streamMergeSubmit(SStreamMergedSubmit2* pMerged, SStreamDataSubmit2* pSubmit) {
120
  taosArrayPush(pMerged->dataRefs, &pSubmit->dataRef);
L
Liu Jicong 已提交
121
  taosArrayPush(pMerged->submits, &pSubmit->submit);
122 123 124 125
  pMerged->ver = pSubmit->ver;
  return 0;
}

L
Liu Jicong 已提交
126
static FORCE_INLINE void streamDataSubmitRefInc(SStreamDataSubmit2* pDataSubmit) {
L
Liu Jicong 已提交
127 128 129
  atomic_add_fetch_32(pDataSubmit->dataRef, 1);
}

130
SStreamDataSubmit2* streamSubmitBlockClone(SStreamDataSubmit2* pSubmit) {
131
  SStreamDataSubmit2* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit2), DEF_QITEM, 0);
L
Liu Jicong 已提交
132 133 134
  if (pSubmitClone == NULL) {
    return NULL;
  }
135

L
Liu Jicong 已提交
136
  streamDataSubmitRefInc(pSubmit);
L
Liu Jicong 已提交
137
  memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit2));
L
Liu Jicong 已提交
138 139
  return pSubmitClone;
}
L
Liu Jicong 已提交
140

141 142
SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) {
  if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) {
L
Liu Jicong 已提交
143
    SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;
144
    SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem;
L
Liu Jicong 已提交
145
    taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);
146
    taosArrayDestroy(pBlockSrc->blocks);
147
    taosFreeQitem(pElem);
J
jiacy-jcy 已提交
148
    return dst;
149
  } else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
L
Liu Jicong 已提交
150
    SStreamMergedSubmit2* pMerged = (SStreamMergedSubmit2*)dst;
151
    SStreamDataSubmit2*   pBlockSrc = (SStreamDataSubmit2*)pElem;
152
    streamMergeSubmit(pMerged, pBlockSrc);
153
    taosFreeQitem(pElem);
J
jiacy-jcy 已提交
154
    return dst;
155
  } else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
L
Liu Jicong 已提交
156
    SStreamMergedSubmit2* pMerged = streamMergedSubmitNew();
157
    ASSERT(pMerged);
L
Liu Jicong 已提交
158
    streamMergeSubmit(pMerged, (SStreamDataSubmit2*)dst);
159
    streamMergeSubmit(pMerged, (SStreamDataSubmit2*)pElem);
160
    taosFreeQitem(dst);
161
    taosFreeQitem(pElem);
J
jiacy-jcy 已提交
162
    return (SStreamQueueItem*)pMerged;
L
Liu Jicong 已提交
163
  } else {
J
jiacy-jcy 已提交
164
    return NULL;
L
Liu Jicong 已提交
165 166 167 168 169
  }
}

void streamFreeQitem(SStreamQueueItem* data) {
  int8_t type = data->type;
L
Liu Jicong 已提交
170
  if (type == STREAM_INPUT__GET_RES) {
L
Liu Jicong 已提交
171 172 173
    blockDataDestroy(((SStreamTrigger*)data)->pBlock);
    taosFreeQitem(data);
  } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE) {
174
    taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)blockDataFreeRes);
L
Liu Jicong 已提交
175 176
    taosFreeQitem(data);
  } else if (type == STREAM_INPUT__DATA_SUBMIT) {
177
    streamDataSubmitDestroy((SStreamDataSubmit2*)data);
L
Liu Jicong 已提交
178
    taosFreeQitem(data);
179
  } else if (type == STREAM_INPUT__MERGED_SUBMIT) {
L
Liu Jicong 已提交
180 181
    SStreamMergedSubmit2* pMerge = (SStreamMergedSubmit2*)data;
    int32_t               sz = taosArrayGetSize(pMerge->submits);
182
    for (int32_t i = 0; i < sz; i++) {
183 184 185 186
      int32_t* pRef = taosArrayGetP(pMerge->dataRefs, i);
      int32_t  ref = atomic_sub_fetch_32(pRef, 1);
      ASSERT(ref >= 0);
      if (ref == 0) {
L
Liu Jicong 已提交
187
        SPackedData* pSubmit = (SPackedData*)taosArrayGet(pMerge->submits, i);
L
Liu Jicong 已提交
188
        taosMemoryFree(pSubmit->msgStr);
189
        taosMemoryFree(pRef);
190 191
      }
    }
L
Liu Jicong 已提交
192
    taosArrayDestroy(pMerge->submits);
193 194
    taosArrayDestroy(pMerge->dataRefs);
    taosFreeQitem(pMerge);
L
Liu Jicong 已提交
195 196 197 198 199 200 201 202 203 204
  } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
    SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;

    int32_t ref = atomic_sub_fetch_32(pRefBlock->dataRef, 1);
    ASSERT(ref >= 0);
    if (ref == 0) {
      blockDataDestroy(pRefBlock->pBlock);
      taosMemoryFree(pRefBlock->dataRef);
    }
    taosFreeQitem(pRefBlock);
L
Liu Jicong 已提交
205 206
  }
}