sma.h 10.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
/*
 * 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/>.
 */

#ifndef _TD_VNODE_SMA_H_
#define _TD_VNODE_SMA_H_

#include "vnodeInt.h"

#ifdef __cplusplus
extern "C" {
#endif

// smaDebug ================
// clang-format off
#define smaFatal(...) do { if (smaDebugFlag & DEBUG_FATAL) { taosPrintLog("SMA FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
#define smaError(...) do { if (smaDebugFlag & DEBUG_ERROR) { taosPrintLog("SMA ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}     while(0)
#define smaWarn(...)  do { if (smaDebugFlag & DEBUG_WARN)  { taosPrintLog("SMA WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}       while(0)
#define smaInfo(...)  do { if (smaDebugFlag & DEBUG_INFO)  { taosPrintLog("SMA ", DEBUG_INFO, 255, __VA_ARGS__); }}            while(0)
C
Cary Xu 已提交
31
#define smaDebug(...) do { if (smaDebugFlag & DEBUG_DEBUG) { taosPrintLog("SMA ", DEBUG_DEBUG, tsdbDebugFlag, __VA_ARGS__); }} while(0)
32 33 34
#define smaTrace(...) do { if (smaDebugFlag & DEBUG_TRACE) { taosPrintLog("SMA ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on

C
Cary Xu 已提交
35
#define RSMA_TASK_INFO_HASH_SLOT (8)
C
Cary Xu 已提交
36

37 38
typedef struct SSmaEnv       SSmaEnv;
typedef struct SSmaStat      SSmaStat;
C
Cary Xu 已提交
39 40
typedef struct STSmaStat     STSmaStat;
typedef struct SRSmaStat     SRSmaStat;
41 42 43
typedef struct SSmaKey       SSmaKey;
typedef struct SRSmaInfo     SRSmaInfo;
typedef struct SRSmaInfoItem SRSmaInfoItem;
C
Cary Xu 已提交
44 45 46
typedef struct SQTaskFile    SQTaskFile;
typedef struct SQTaskFReader SQTaskFReader;
typedef struct SQTaskFWriter SQTaskFWriter;
47 48

struct SSmaEnv {
C
Cary Xu 已提交
49 50
  SRWLatch  lock;
  int8_t    type;
C
Cary Xu 已提交
51
  int8_t    flag;  // 0x01 inClose
C
Cary Xu 已提交
52
  SSmaStat *pStat;
53 54
};

C
Cary Xu 已提交
55 56
#define SMA_ENV_FLG_CLOSE ((int8_t)0x1)

57
typedef struct {
C
Cary Xu 已提交
58 59
  int8_t  inited;
  int32_t rsetId;
60
  void   *tmrHandle;  // shared by all fetch tasks
61 62
} SSmaMgmt;

C
Cary Xu 已提交
63 64 65 66
#define SMA_ENV_LOCK(env)  (&(env)->lock)
#define SMA_ENV_TYPE(env)  ((env)->type)
#define SMA_ENV_STAT(env)  ((env)->pStat)
#define SMA_RSMA_STAT(sma) ((SRSmaStat *)SMA_ENV_STAT((SSmaEnv *)(sma)->pRSmaEnv))
67

C
Cary Xu 已提交
68
struct STSmaStat {
C
Cary Xu 已提交
69 70 71
  int8_t    state;  // ETsdbSmaStat
  STSma    *pTSma;  // cache schema
  STSchema *pTSchema;
72 73
};

C
Cary Xu 已提交
74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91
struct SQTaskFile {
  volatile int32_t nRef;
  int64_t          commitID;
  int64_t          size;
};

struct SQTaskFReader {
  SSma      *pSma;
  SQTaskFile fTask;
  TdFilePtr  pReadH;
};
struct SQTaskFWriter {
  SSma      *pSma;
  SQTaskFile fTask;
  TdFilePtr  pWriteH;
  char      *fname;
};

C
Cary Xu 已提交
92
struct SRSmaStat {
C
Cary Xu 已提交
93 94 95
  SSma            *pSma;
  int64_t          commitAppliedVer;  // vnode applied version for async commit
  int64_t          refId;             // shared by fetch tasks
C
Cary Xu 已提交
96
  volatile int64_t nBufItems;         // number of items in queue buffer
C
Cary Xu 已提交
97 98 99 100
  SRWLatch         lock;              // r/w lock for rsma fs(e.g. qtaskinfo)
  int8_t           triggerStat;       // shared by fetch tasks
  int8_t           commitStat;        // 0 not in committing, 1 in committing
  SArray          *aTaskFile;         // qTaskFiles committed recently(for recovery/snapshot r/w)
C
Cary Xu 已提交
101
  SHashObj        *infoHash;          // key: suid, value: SRSmaInfo
C
Cary Xu 已提交
102
  tsem_t           notEmpty;          // has items in queue buffer
C
Cary Xu 已提交
103 104
};

105 106
struct SSmaStat {
  union {
C
Cary Xu 已提交
107 108
    STSmaStat tsmaStat;  // time-range-wise sma
    SRSmaStat rsmaStat;  // rollup sma
109 110
  };
  T_REF_DECLARE()
C
Cary Xu 已提交
111
  char data[];
112
};
113

C
Cary Xu 已提交
114 115 116 117 118 119 120
#define SMA_STAT_TSMA(s)     (&(s)->tsmaStat)
#define SMA_STAT_RSMA(s)     (&(s)->rsmaStat)
#define RSMA_INFO_HASH(r)    ((r)->infoHash)
#define RSMA_TRIGGER_STAT(r) (&(r)->triggerStat)
#define RSMA_COMMIT_STAT(r)  (&(r)->commitStat)
#define RSMA_REF_ID(r)       ((r)->refId)
#define RSMA_FS_LOCK(r)      (&(r)->lock)
C
Cary Xu 已提交
121 122

struct SRSmaInfoItem {
C
Cary Xu 已提交
123 124 125
  int8_t   level : 4;
  int8_t   fetchLevel : 4;
  int8_t   triggerStat;
C
Cary Xu 已提交
126
  uint16_t nSkipped;
C
Cary Xu 已提交
127 128
  int32_t  maxDelay;  // ms
  tmr_h    tmrId;
C
Cary Xu 已提交
129 130 131
};

struct SRSmaInfo {
C
Cary Xu 已提交
132 133
  STSchema *pTSchema;
  int64_t   suid;
C
Cary Xu 已提交
134 135 136
  int64_t   refId;     // refId of SRSmaStat
  int64_t   lastRecv;  // ms
  int8_t    assigned;  // 0 idle, 1 assgined for exec
C
Cary Xu 已提交
137
  int8_t    delFlag;
C
Cary Xu 已提交
138
  int16_t   padding;
C
Cary Xu 已提交
139
  T_REF_DECLARE()
C
Cary Xu 已提交
140
  SRSmaInfoItem items[TSDB_RETENTION_L2];
C
Cary Xu 已提交
141
  void         *taskInfo[TSDB_RETENTION_L2];   // qTaskInfo_t
C
Cary Xu 已提交
142 143 144 145 146
  STaosQueue   *queue;                         // buffer queue of SubmitReq
  STaosQall    *qall;                          // buffer qall of SubmitReq
  void         *iTaskInfo[TSDB_RETENTION_L2];  // immutable qTaskInfo_t
  STaosQueue   *iQueue;                        // immutable buffer queue of SubmitReq
  STaosQall    *iQall;                         // immutable buffer qall of SubmitReq
C
Cary Xu 已提交
147
};
C
Cary Xu 已提交
148

C
Cary Xu 已提交
149
#define RSMA_INFO_HEAD_LEN     offsetof(SRSmaInfo, items)
C
Cary Xu 已提交
150 151 152
#define RSMA_INFO_IS_DEL(r)    ((r)->delFlag == 1)
#define RSMA_INFO_SET_DEL(r)   ((r)->delFlag = 1)
#define RSMA_INFO_QTASK(r, i)  ((r)->taskInfo[i])
C
Cary Xu 已提交
153
#define RSMA_INFO_IQTASK(r, i) ((r)->iTaskInfo[i])
C
Cary Xu 已提交
154
#define RSMA_INFO_ITEM(r, i)   (&(r)->items[i])
C
Cary Xu 已提交
155 156 157 158 159

enum {
  TASK_TRIGGER_STAT_INIT = 0,
  TASK_TRIGGER_STAT_ACTIVE = 1,
  TASK_TRIGGER_STAT_INACTIVE = 2,
C
Cary Xu 已提交
160 161
  TASK_TRIGGER_STAT_PAUSED = 3,
  TASK_TRIGGER_STAT_CANCELLED = 4,
162
  TASK_TRIGGER_STAT_DROPPED = 5,
C
Cary Xu 已提交
163
};
C
Cary Xu 已提交
164

C
Cary Xu 已提交
165 166 167
enum {
  RSMA_ROLE_CREATE = 0,
  RSMA_ROLE_DROP = 1,
C
Cary Xu 已提交
168 169
  RSMA_ROLE_SUBMIT = 2,
  RSMA_ROLE_FETCH = 3,
C
Cary Xu 已提交
170 171
  RSMA_ROLE_ITERATE = 4,
};
C
Cary Xu 已提交
172

173 174 175 176 177
enum {
  RSMA_RESTORE_REBOOT = 1,
  RSMA_RESTORE_SYNC = 2,
};

C
Cary Xu 已提交
178 179 180 181 182 183
typedef enum {
  RSMA_EXEC_OVERFLOW = 1,  // triggered by queue buf overflow
  RSMA_EXEC_TIMEOUT = 2,   // triggered by timer
  RSMA_EXEC_COMMIT = 3,    // triggered by commit
} ERsmaExecType;

184 185 186 187 188 189 190 191 192
void  tdDestroySmaEnv(SSmaEnv *pSmaEnv);
void *tdFreeSmaEnv(SSmaEnv *pSmaEnv);

int32_t tdDropTSma(SSma *pSma, char *pMsg);
int32_t tdDropTSmaData(SSma *pSma, int64_t indexUid);
int32_t tdInsertRSmaData(SSma *pSma, char *msg);

int32_t tdRefSmaStat(SSma *pSma, SSmaStat *pStat);
int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat);
C
Cary Xu 已提交
193 194
int32_t tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
C
Cary Xu 已提交
195

C
Cary Xu 已提交
196 197
void   *tdAcquireSmaRef(int32_t rsetId, int64_t refId);
int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId);
C
Cary Xu 已提交
198

C
Cary Xu 已提交
199
int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
200 201 202 203

int32_t tdLockSma(SSma *pSma);
int32_t tdUnLockSma(SSma *pSma);

C
Cary Xu 已提交
204 205 206
static FORCE_INLINE int8_t tdSmaStat(STSmaStat *pTStat) {
  if (pTStat) {
    return atomic_load_8(&pTStat->state);
207 208 209 210
  }
  return TSDB_SMA_STAT_UNKNOWN;
}

C
Cary Xu 已提交
211 212
static FORCE_INLINE bool tdSmaStatIsOK(STSmaStat *pTStat, int8_t *state) {
  if (!pTStat) {
213 214 215 216
    return false;
  }

  if (state) {
C
Cary Xu 已提交
217
    *state = atomic_load_8(&pTStat->state);
218 219
    return *state == TSDB_SMA_STAT_OK;
  }
C
Cary Xu 已提交
220
  return atomic_load_8(&pTStat->state) == TSDB_SMA_STAT_OK;
221 222
}

C
Cary Xu 已提交
223 224
static FORCE_INLINE bool tdSmaStatIsExpired(STSmaStat *pTStat) {
  return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_EXPIRED) : true;
225 226
}

C
Cary Xu 已提交
227 228
static FORCE_INLINE bool tdSmaStatIsDropped(STSmaStat *pTStat) {
  return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_DROPPED) : true;
229 230
}

C
Cary Xu 已提交
231 232 233
static FORCE_INLINE void tdSmaStatSetOK(STSmaStat *pTStat) {
  if (pTStat) {
    atomic_store_8(&pTStat->state, TSDB_SMA_STAT_OK);
234 235 236
  }
}

C
Cary Xu 已提交
237 238 239
static FORCE_INLINE void tdSmaStatSetExpired(STSmaStat *pTStat) {
  if (pTStat) {
    atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_EXPIRED);
240 241 242
  }
}

C
Cary Xu 已提交
243 244 245
static FORCE_INLINE void tdSmaStatSetDropped(STSmaStat *pTStat) {
  if (pTStat) {
    atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_DROPPED);
246 247
  }
}
C
Cary Xu 已提交
248

C
Cary Xu 已提交
249 250
void           tdRSmaQTaskInfoGetFileName(int32_t vid, int64_t version, char *outputName);
void           tdRSmaQTaskInfoGetFullName(int32_t vid, int64_t version, const char *path, char *outputName);
C
Cary Xu 已提交
251
int32_t        tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
C
Cary Xu 已提交
252
void           tdFreeQTaskInfo(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
C
Cary Xu 已提交
253 254
static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
void          *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
C
Cary Xu 已提交
255 256
void          *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
int32_t        tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
C
Cary Xu 已提交
257
int32_t        tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
258

259
int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
260 261
int32_t tdProcessRSmaRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer);
int32_t tdRsmaRestore(SSma *pSma, int8_t type, int64_t committedVer);
C
Cary Xu 已提交
262

C
Cary Xu 已提交
263
int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
C
Cary Xu 已提交
264 265
int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
C
Cary Xu 已提交
266

C
Cary Xu 已提交
267 268
// smaFileUtil ================

269 270
#define TD_FILE_HEAD_SIZE 512

C
Cary Xu 已提交
271 272 273 274
typedef struct STFInfo STFInfo;
typedef struct STFile  STFile;

struct STFInfo {
275
  // common fields
C
Cary Xu 已提交
276
  uint32_t magic;
C
Cary Xu 已提交
277
  uint32_t ftype;
C
Cary Xu 已提交
278
  uint32_t fver;
C
Cary Xu 已提交
279
  int64_t  fsize;
C
Cary Xu 已提交
280 281
};

282 283 284 285
enum {
  TD_FTYPE_RSMA_QTASKINFO = 0,
};

C
Cary Xu 已提交
286
struct STFile {
C
Cary Xu 已提交
287
  uint8_t   state;
C
Cary Xu 已提交
288
  STFInfo   info;
C
Cary Xu 已提交
289
  char     *fname;
C
Cary Xu 已提交
290 291 292
  TdFilePtr pFile;
};

C
Cary Xu 已提交
293 294
#define TD_TFILE_PFILE(tf)        ((tf)->pFile)
#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
C
Cary Xu 已提交
295
#define TD_TFILE_FULL_NAME(tf)    ((tf)->fname)
C
Cary Xu 已提交
296 297 298 299
#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
#define TD_TFILE_CLOSED(tf)       (!TD_TFILE_OPENED(tf))
#define TD_TFILE_SET_CLOSED(f)    (TD_TFILE_PFILE(f) = NULL)
#define TD_TFILE_SET_STATE(tf, s) ((tf)->state = (s))
C
Cary Xu 已提交
300

C
Cary Xu 已提交
301 302
int32_t tdInitTFile(STFile *pTFile, const char *dname, const char *fname);
int32_t tdCreateTFile(STFile *pTFile, bool updateHeader, int8_t fType);
C
Cary Xu 已提交
303 304 305 306 307
int32_t tdOpenTFile(STFile *pTFile, int flags);
int64_t tdReadTFile(STFile *pTFile, void *buf, int64_t nbyte);
int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence);
int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte);
int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset);
308
int64_t tdGetTFileSize(STFile *pTFile, int64_t *size);
C
Cary Xu 已提交
309 310 311 312 313
int32_t tdRemoveTFile(STFile *pTFile);
int32_t tdLoadTFileHeader(STFile *pTFile, STFInfo *pInfo);
int32_t tdUpdateTFileHeader(STFile *pTFile);
void    tdUpdateTFileMagic(STFile *pTFile, void *pCksm);
void    tdCloseTFile(STFile *pTFile);
C
Cary Xu 已提交
314
void    tdDestroyTFile(STFile *pTFile);
315

C
Cary Xu 已提交
316 317 318
void tdGetVndFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t version,
                      char *outputName);
void tdGetVndDirName(int32_t vgId, const char *pdname, const char *dname, bool endWithSep, char *outputName);
C
Cary Xu 已提交
319

320 321 322 323
#ifdef __cplusplus
}
#endif

L
Liu Jicong 已提交
324
#endif /*_TD_VNODE_SMA_H_*/