vnodeCommit.c 15.2 KB
Newer Older
H
refact  
Hongze Cheng 已提交
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/>.
 */

H
Hongze Cheng 已提交
16
#include "vnd.h"
H
Hongze Cheng 已提交
17
#include "vnodeInt.h"
H
Hongze Cheng 已提交
18

H
Hongze Cheng 已提交
19 20
#define VND_INFO_FNAME_TMP "vnode_tmp.json"

H
Hongze Cheng 已提交
21 22
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData);
static int vnodeCommitImpl(SCommitInfo *pInfo);
H
refact  
Hongze Cheng 已提交
23

H
Hongze Cheng 已提交
24 25
#define WAIT_TIME_MILI_SEC 10  // miliseconds

H
Hongze Cheng 已提交
26 27
static int32_t vnodeTryRecycleBufPool(SVnode *pVnode) {
  int32_t code = 0;
H
Hongze Cheng 已提交
28

H
Hongze Cheng 已提交
29 30
  if (pVnode->onRecycle == NULL) {
    if (pVnode->recycleHead == NULL) {
31
      vDebug("vgId:%d, no recyclable buffer pool", TD_VID(pVnode));
H
Hongze Cheng 已提交
32
      goto _exit;
H
Hongze Cheng 已提交
33
    } else {
34
      vDebug("vgId:%d, buffer pool %p of id %d on recycle queue, try to recycle", TD_VID(pVnode), pVnode->recycleHead,
H
Hongze Cheng 已提交
35 36 37 38 39 40 41 42 43 44
             pVnode->recycleHead->id);

      pVnode->onRecycle = pVnode->recycleHead;
      if (pVnode->recycleHead == pVnode->recycleTail) {
        pVnode->recycleHead = pVnode->recycleTail = NULL;
      } else {
        pVnode->recycleHead = pVnode->recycleHead->recycleNext;
        pVnode->recycleHead->recyclePrev = NULL;
      }
      pVnode->onRecycle->recycleNext = pVnode->onRecycle->recyclePrev = NULL;
H
Hongze Cheng 已提交
45
    }
H
Hongze Cheng 已提交
46
  }
H
Hongze Cheng 已提交
47

H
Hongze Cheng 已提交
48 49
  code = vnodeBufPoolRecycle(pVnode->onRecycle);
  if (code) goto _exit;
H
Hongze Cheng 已提交
50

H
Hongze Cheng 已提交
51
_exit:
H
Hongze Cheng 已提交
52
  if (code) {
53
    vError("vgId:%d, %s failed since %s", TD_VID(pVnode), __func__, tstrerror(code));
H
Hongze Cheng 已提交
54
  }
H
Hongze Cheng 已提交
55 56
  return code;
}
H
Hongze Cheng 已提交
57 58
static int32_t vnodeGetBufPoolToUse(SVnode *pVnode) {
  int32_t code = 0;
H
Hongze Cheng 已提交
59
  int32_t lino = 0;
H
Hongze Cheng 已提交
60

H
Hongze Cheng 已提交
61
  taosThreadMutexLock(&pVnode->mutex);
H
Hongze Cheng 已提交
62

H
Hongze Cheng 已提交
63
  int32_t nTry = 0;
H
Hongze Cheng 已提交
64 65 66
  for (;;) {
    ++nTry;

H
Hongze Cheng 已提交
67
    if (pVnode->freeList) {
68
      vDebug("vgId:%d, allocate free buffer pool on %d try, pPool:%p id:%d", TD_VID(pVnode), nTry, pVnode->freeList,
H
Hongze Cheng 已提交
69
             pVnode->freeList->id);
H
Hongze Cheng 已提交
70

H
Hongze Cheng 已提交
71 72 73 74 75 76
      pVnode->inUse = pVnode->freeList;
      pVnode->inUse->nRef = 1;
      pVnode->freeList = pVnode->inUse->freeNext;
      pVnode->inUse->freeNext = NULL;
      break;
    } else {
77
      vDebug("vgId:%d, no free buffer pool on %d try, try to recycle...", TD_VID(pVnode), nTry);
78

H
Hongze Cheng 已提交
79 80
      code = vnodeTryRecycleBufPool(pVnode);
      TSDB_CHECK_CODE(code, lino, _exit);
81

H
Hongze Cheng 已提交
82
      if (pVnode->freeList == NULL) {
83
        vDebug("vgId:%d, no free buffer pool on %d try, wait %d ms...", TD_VID(pVnode), nTry, WAIT_TIME_MILI_SEC);
H
Hongze Cheng 已提交
84 85 86 87 88

        struct timeval  tv;
        struct timespec ts;
        taosGetTimeOfDay(&tv);
        ts.tv_nsec = tv.tv_usec * 1000 + WAIT_TIME_MILI_SEC * 1000000;
H
Hongze Cheng 已提交
89 90 91 92 93 94
        if (ts.tv_nsec > 999999999l) {
          ts.tv_sec = tv.tv_sec + 1;
          ts.tv_nsec -= 1000000000l;
        } else {
          ts.tv_sec = tv.tv_sec;
        }
H
Hongze Cheng 已提交
95 96 97

        int32_t rc = taosThreadCondTimedWait(&pVnode->poolNotEmpty, &pVnode->mutex, &ts);
        if (rc && rc != ETIMEDOUT) {
H
Hongze Cheng 已提交
98 99
          code = TAOS_SYSTEM_ERROR(rc);
          TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
100 101
        }
      }
H
Hongze Cheng 已提交
102 103
    }
  }
H
Hongze Cheng 已提交
104

H
Hongze Cheng 已提交
105
_exit:
H
Hongze Cheng 已提交
106
  taosThreadMutexUnlock(&pVnode->mutex);
H
Hongze Cheng 已提交
107
  if (code) {
108
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
H
Hongze Cheng 已提交
109
  }
H
Hongze Cheng 已提交
110 111 112 113 114
  return code;
}
int vnodeBegin(SVnode *pVnode) {
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
115

H
Hongze Cheng 已提交
116 117 118 119
  // alloc buffer pool
  code = vnodeGetBufPoolToUse(pVnode);
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
120
  // begin meta
121
  if (metaBegin(pVnode->pMeta, META_BEGIN_HEAP_BUFFERPOOL) < 0) {
H
Hongze Cheng 已提交
122 123
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
124 125 126
  }

  // begin tsdb
H
Hongze Cheng 已提交
127
  if (tsdbBegin(pVnode->pTsdb) < 0) {
H
Hongze Cheng 已提交
128 129
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
130
  }
C
Cary Xu 已提交
131

C
Cary Xu 已提交
132
  // begin sma
C
Cary Xu 已提交
133
  if (VND_IS_RSMA(pVnode) && smaBegin(pVnode->pSma) < 0) {
H
Hongze Cheng 已提交
134 135
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
136
  }
C
Cary Xu 已提交
137

H
Hongze Cheng 已提交
138 139
_exit:
  if (code) {
H
Hongze Cheng 已提交
140
    terrno = code;
141
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
H
Hongze Cheng 已提交
142 143
  }
  return code;
H
Hongze Cheng 已提交
144 145
}

146
int vnodeShouldCommit(SVnode *pVnode, bool atExit) {
147 148
  bool diskAvail = osDataSpaceAvailable();
  bool needCommit = false;
B
Benguang Zhao 已提交
149

150
  taosThreadMutexLock(&pVnode->mutex);
B
Benguang Zhao 已提交
151
  if (pVnode->inUse && diskAvail) {
152
    needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) || (pVnode->inUse->size > 0 && atExit);
C
Cary Xu 已提交
153
  }
154 155
  taosThreadMutexUnlock(&pVnode->mutex);
  return needCommit;
C
Cary Xu 已提交
156
}
H
Hongze Cheng 已提交
157

H
Hongze Cheng 已提交
158 159 160
int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
  char      fname[TSDB_FILENAME_LEN];
  TdFilePtr pFile;
H
Hongze Cheng 已提交
161
  char     *data;
H
Hongze Cheng 已提交
162 163 164 165 166 167

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);

  // encode info
  data = NULL;

H
Hongze Cheng 已提交
168
  if (vnodeEncodeInfo(pInfo, &data) < 0) {
169
    vError("failed to encode json info.");
H
Hongze Cheng 已提交
170 171 172 173 174 175
    return -1;
  }

  // save info to a vnode_tmp.json
  pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
  if (pFile == NULL) {
S
Shengliang Guan 已提交
176
    vError("failed to open info file:%s for write:%s", fname, terrstr());
H
Hongze Cheng 已提交
177 178 179 180
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

H
Hongze Cheng 已提交
181
  if (taosWriteFile(pFile, data, strlen(data)) < 0) {
182
    vError("failed to write info file:%s error:%s", fname, terrstr());
H
Hongze Cheng 已提交
183 184 185 186 187
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosFsyncFile(pFile) < 0) {
S
Shengliang Guan 已提交
188
    vError("failed to fsync info file:%s error:%s", fname, terrstr());
H
Hongze Cheng 已提交
189 190 191 192 193 194 195 196 197
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  taosCloseFile(&pFile);

  // free info binary
  taosMemoryFree(data);

198 199
  vInfo("vgId:%d, vnode info is saved, fname:%s replica:%d selfIndex:%d", pInfo->config.vgId, fname,
        pInfo->config.syncCfg.replicaNum, pInfo->config.syncCfg.myIndex);
H
Hongze Cheng 已提交
200 201 202 203 204 205 206 207 208

  return 0;

_err:
  taosCloseFile(&pFile);
  taosMemoryFree(data);
  return -1;
}

H
Hongze Cheng 已提交
209
int vnodeCommitInfo(const char *dir) {
H
Hongze Cheng 已提交
210 211 212 213 214 215 216 217 218 219 220
  char fname[TSDB_FILENAME_LEN];
  char tfname[TSDB_FILENAME_LEN];

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);
  snprintf(tfname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);

  if (taosRenameFile(tfname, fname) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

H
Hongze Cheng 已提交
221
  vInfo("vnode info is committed, dir:%s", dir);
H
Hongze Cheng 已提交
222 223 224
  return 0;
}

H
Hongze Cheng 已提交
225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244
int vnodeLoadInfo(const char *dir, SVnodeInfo *pInfo) {
  char      fname[TSDB_FILENAME_LEN];
  TdFilePtr pFile = NULL;
  char     *pData = NULL;
  int64_t   size;

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);

  // read info
  pFile = taosOpenFile(fname, TD_FILE_READ);
  if (pFile == NULL) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  if (taosFStatFile(pFile, &size, NULL) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
245
  pData = taosMemoryMalloc(size + 1);
H
Hongze Cheng 已提交
246 247 248 249 250 251 252 253 254 255
  if (pData == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  if (taosReadFile(pFile, pData, size) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
256 257
  pData[size] = '\0';

H
Hongze Cheng 已提交
258 259 260 261 262 263 264 265 266 267
  taosCloseFile(&pFile);

  // decode info
  if (vnodeDecodeInfo(pData, pInfo) < 0) {
    taosMemoryFree(pData);
    return -1;
  }

  taosMemoryFree(pData);

H
Hongze Cheng 已提交
268
  return 0;
H
Hongze Cheng 已提交
269 270 271 272 273

_err:
  taosCloseFile(&pFile);
  taosMemoryFree(pData);
  return -1;
H
Hongze Cheng 已提交
274 275
}

H
Hongze Cheng 已提交
276
static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
277
  int32_t code = 0;
278 279 280
  int32_t lino = 0;
  char    dir[TSDB_FILENAME_LEN] = {0};

H
Hongze Cheng 已提交
281 282
  tsem_wait(&pVnode->canCommit);

283 284 285 286 287
  pVnode->state.commitTerm = pVnode->state.applyTerm;

  pInfo->info.config = pVnode->config;
  pInfo->info.state.committed = pVnode->state.applied;
  pInfo->info.state.commitTerm = pVnode->state.applyTerm;
H
Hongze Cheng 已提交
288
  pInfo->info.state.commitID = ++pVnode->state.commitID;
289 290 291 292
  pInfo->pVnode = pVnode;
  pInfo->txn = metaGetTxn(pVnode->pMeta);

  // save info
293
  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
294 295

  vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
296 297 298 299 300
  if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

301
  tsdbPrepareCommit(pVnode->pTsdb);
K
kailixu 已提交
302

303
  metaPrepareAsyncCommit(pVnode->pMeta);
304

305 306
  code = smaPrepareAsyncCommit(pVnode->pSma);
  if (code) goto _exit;
K
kailixu 已提交
307

K
kailixu 已提交
308 309 310 311 312 313
  taosThreadMutexLock(&pVnode->mutex);
  ASSERT(pVnode->onCommit == NULL);
  pVnode->onCommit = pVnode->inUse;
  pVnode->inUse = NULL;
  taosThreadMutexUnlock(&pVnode->mutex);

314 315 316 317 318
_exit:
  if (code) {
    vError("vgId:%d, %s failed at line %d since %s, commit id:%" PRId64, TD_VID(pVnode), __func__, lino,
           tstrerror(code), pVnode->state.commitID);
  } else {
H
Hongze Cheng 已提交
319
    vDebug("vgId:%d, %s done, commit id:%" PRId64, TD_VID(pVnode), __func__, pInfo->info.state.commitID);
320
  }
321

322
  return code;
H
Hongze Cheng 已提交
323
}
H
Hongze Cheng 已提交
324
static void vnodeReturnBufPool(SVnode *pVnode) {
H
Hongze Cheng 已提交
325
  taosThreadMutexLock(&pVnode->mutex);
326

H
Hongze Cheng 已提交
327 328 329
  SVBufPool *pPool = pVnode->onCommit;
  int32_t    nRef = atomic_sub_fetch_32(&pPool->nRef, 1);

H
Hongze Cheng 已提交
330 331
  pVnode->onCommit = NULL;
  if (nRef == 0) {
H
Hongze Cheng 已提交
332
    vnodeBufPoolAddToFreeList(pPool);
H
Hongze Cheng 已提交
333
  } else if (nRef > 0) {
334
    vDebug("vgId:%d, buffer pool %p of id %d is added to recycle queue", TD_VID(pVnode), pPool, pPool->id);
H
Hongze Cheng 已提交
335 336 337 338 339 340 341 342 343 344

    if (pVnode->recycleTail == NULL) {
      pPool->recyclePrev = pPool->recycleNext = NULL;
      pVnode->recycleHead = pVnode->recycleTail = pPool;
    } else {
      pPool->recyclePrev = pVnode->recycleTail;
      pPool->recycleNext = NULL;
      pVnode->recycleTail->recycleNext = pPool;
      pVnode->recycleTail = pPool;
    }
H
Hongze Cheng 已提交
345 346
  } else {
    ASSERT(0);
H
Hongze Cheng 已提交
347 348 349
  }

  taosThreadMutexUnlock(&pVnode->mutex);
H
Hongze Cheng 已提交
350
}
H
Hongze Cheng 已提交
351 352 353
static int32_t vnodeCommitTask(void *arg) {
  int32_t code = 0;

H
Hongze Cheng 已提交
354
  SCommitInfo *pInfo = (SCommitInfo *)arg;
H
Hongze Cheng 已提交
355
  SVnode      *pVnode = pInfo->pVnode;
H
Hongze Cheng 已提交
356

H
Hongze Cheng 已提交
357 358
  // commit
  code = vnodeCommitImpl(pInfo);
H
Hongze Cheng 已提交
359 360
  if (code) goto _exit;

H
Hongze Cheng 已提交
361
  vnodeReturnBufPool(pVnode);
H
Hongze Cheng 已提交
362

363
_exit:
H
Hongze Cheng 已提交
364
  // end commit
H
Hongze Cheng 已提交
365
  tsem_post(&pVnode->canCommit);
H
Hongze Cheng 已提交
366
  taosMemoryFree(pInfo);
H
Hongze Cheng 已提交
367 368
  return code;
}
B
Benguang Zhao 已提交
369

H
refact  
Hongze Cheng 已提交
370
int vnodeAsyncCommit(SVnode *pVnode) {
H
Hongze Cheng 已提交
371
  int32_t code = 0;
H
Hongze Cheng 已提交
372

H
Hongze Cheng 已提交
373
  SCommitInfo *pInfo = (SCommitInfo *)taosMemoryCalloc(1, sizeof(*pInfo));
H
Hongze Cheng 已提交
374 375 376 377
  if (NULL == pInfo) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
378 379 380 381 382 383 384 385

  // prepare to commit
  code = vnodePrepareCommit(pVnode, pInfo);
  if (TSDB_CODE_SUCCESS != code) {
    goto _exit;
  }

  // schedule the task
386
  code = vnodeScheduleTask(vnodeCommitTask, pInfo);
H
Hongze Cheng 已提交
387

H
Hongze Cheng 已提交
388 389
_exit:
  if (code) {
390 391 392
    if (NULL != pInfo) {
      taosMemoryFree(pInfo);
    }
393 394
    tsem_post(&pVnode->canCommit);
    vError("vgId:%d, %s failed since %s, commit id:%" PRId64, TD_VID(pVnode), __func__, tstrerror(code),
H
Hongze Cheng 已提交
395 396
           pVnode->state.commitID);
  } else {
397 398
    vInfo("vgId:%d, vnode async commit done, commitId:%" PRId64 " term:%" PRId64 " applied:%" PRId64, TD_VID(pVnode),
          pVnode->state.commitID, pVnode->state.applyTerm, pVnode->state.applied);
H
Hongze Cheng 已提交
399 400
  }
  return code;
H
Hongze Cheng 已提交
401
}
H
refact  
Hongze Cheng 已提交
402

H
Hongze Cheng 已提交
403 404
int vnodeSyncCommit(SVnode *pVnode) {
  vnodeAsyncCommit(pVnode);
H
Hongze Cheng 已提交
405 406
  tsem_wait(&pVnode->canCommit);
  tsem_post(&pVnode->canCommit);
H
Hongze Cheng 已提交
407 408 409
  return 0;
}

H
Hongze Cheng 已提交
410 411 412 413 414 415
static int vnodeCommitImpl(SCommitInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;

  char    dir[TSDB_FILENAME_LEN] = {0};
  SVnode *pVnode = pInfo->pVnode;
H
Hongze Cheng 已提交
416

417
  vInfo("vgId:%d, start to commit, commitId:%" PRId64 " version:%" PRId64 " term: %" PRId64, TD_VID(pVnode),
B
Benguang Zhao 已提交
418 419
        pInfo->info.state.commitID, pInfo->info.state.committed, pInfo->info.state.commitTerm);

420 421 422 423 424 425
  // persist wal before starting
  if (walPersist(pVnode->pWal) < 0) {
    vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), terrstr());
    return -1;
  }

426
  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
427

428
  syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);
H
Hongze Cheng 已提交
429 430

  // commit each sub-system
H
Hongze Cheng 已提交
431
  code = tsdbCommit(pVnode->pTsdb, pInfo);
C
Cary Xu 已提交
432 433
  TSDB_CHECK_CODE(code, lino, _exit);

434 435 436 437
  if (!TSDB_CACHE_NO(pVnode->config)) {
    code = tsdbCacheCommit(pVnode->pTsdb);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
438

C
Cary Xu 已提交
439
  if (VND_IS_RSMA(pVnode)) {
H
Hongze Cheng 已提交
440
    code = smaCommit(pVnode->pSma, pInfo);
H
Hongze Cheng 已提交
441
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
442
  }
C
Cary Xu 已提交
443

H
Hongze Cheng 已提交
444
  if (tqCommit(pVnode->pTq) < 0) {
H
Hongze Cheng 已提交
445 446
    code = TSDB_CODE_FAILED;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
447 448 449
  }

  // commit info
H
Hongze Cheng 已提交
450
  if (vnodeCommitInfo(dir) < 0) {
H
Hongze Cheng 已提交
451 452
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
453
  }
H
Hongze Cheng 已提交
454

C
Cary Xu 已提交
455 456 457
  code = tsdbFinishCommit(pVnode->pTsdb);
  TSDB_CHECK_CODE(code, lino, _exit);

C
Cary Xu 已提交
458 459 460 461
  if (VND_IS_RSMA(pVnode)) {
    code = smaFinishCommit(pVnode->pSma);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
462

463
  if (metaFinishCommit(pVnode->pMeta, pInfo->txn) < 0) {
464 465 466 467
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

H
Hongze Cheng 已提交
468
  pVnode->state.committed = pInfo->info.state.committed;
H
Hongze Cheng 已提交
469

C
Cary Xu 已提交
470 471
  if (smaPostCommit(pVnode->pSma) < 0) {
    vError("vgId:%d, failed to post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
C
Cary Xu 已提交
472 473
    return -1;
  }
H
Hongze Cheng 已提交
474

475
  syncEndSnapshot(pVnode->sync);
H
Hongze Cheng 已提交
476

H
Hongze Cheng 已提交
477 478
_exit:
  if (code) {
S
Shengliang Guan 已提交
479
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
H
Hongze Cheng 已提交
480 481 482
  } else {
    vInfo("vgId:%d, commit end", TD_VID(pVnode));
  }
H
Hongze Cheng 已提交
483 484 485
  return 0;
}

H
Hongze Cheng 已提交
486 487
bool vnodeShouldRollback(SVnode *pVnode) {
  char tFName[TSDB_FILENAME_LEN] = {0};
488 489
  int32_t offset = 0;

490
  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
491 492
  offset = strlen(tFName);
  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
H
Hongze Cheng 已提交
493 494 495 496 497 498

  return taosCheckExistFile(tFName);
}

void vnodeRollback(SVnode *pVnode) {
  char tFName[TSDB_FILENAME_LEN] = {0};
499 500
  int32_t offset = 0;

501
  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
502 503
  offset = strlen(tFName);
  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
H
Hongze Cheng 已提交
504 505 506 507

  (void)taosRemoveFile(tFName);
}

H
Hongze Cheng 已提交
508 509 510 511
static int vnodeEncodeState(const void *pObj, SJson *pJson) {
  const SVState *pState = (SVState *)pObj;

  if (tjsonAddIntegerToObject(pJson, "commit version", pState->committed) < 0) return -1;
H
Hongze Cheng 已提交
512
  if (tjsonAddIntegerToObject(pJson, "commit ID", pState->commitID) < 0) return -1;
H
Hongze Cheng 已提交
513
  if (tjsonAddIntegerToObject(pJson, "commit term", pState->commitTerm) < 0) return -1;
H
Hongze Cheng 已提交
514 515 516 517 518 519 520

  return 0;
}

static int vnodeDecodeState(const SJson *pJson, void *pObj) {
  SVState *pState = (SVState *)pObj;

521 522
  int32_t code;
  tjsonGetNumberValue(pJson, "commit version", pState->committed, code);
H
Hongze Cheng 已提交
523
  if (code < 0) return -1;
H
Hongze Cheng 已提交
524 525
  tjsonGetNumberValue(pJson, "commit ID", pState->commitID, code);
  if (code < 0) return -1;
H
Hongze Cheng 已提交
526 527
  tjsonGetNumberValue(pJson, "commit term", pState->commitTerm, code);
  if (code < 0) return -1;
H
Hongze Cheng 已提交
528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558

  return 0;
}

static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData) {
  SJson *pJson;
  char  *pData;

  *ppData = NULL;

  pJson = tjsonCreateObject();
  if (pJson == NULL) {
    return -1;
  }

  if (tjsonAddObject(pJson, "config", vnodeEncodeConfig, (void *)&pInfo->config) < 0) {
    goto _err;
  }

  if (tjsonAddObject(pJson, "state", vnodeEncodeState, (void *)&pInfo->state) < 0) {
    goto _err;
  }

  pData = tjsonToString(pJson);
  if (pData == NULL) {
    goto _err;
  }

  tjsonDelete(pJson);

  *ppData = pData;
H
Hongze Cheng 已提交
559
  return 0;
H
Hongze Cheng 已提交
560 561 562 563

_err:
  tjsonDelete(pJson);
  return -1;
H
Hongze Cheng 已提交
564 565
}

H
Hongze Cheng 已提交
566
int vnodeDecodeInfo(uint8_t *pData, SVnodeInfo *pInfo) {
H
Hongze Cheng 已提交
567 568
  SJson *pJson = NULL;

H
fix bug  
Hongze Cheng 已提交
569
  pJson = tjsonParse(pData);
H
Hongze Cheng 已提交
570 571 572 573 574 575 576 577 578 579 580 581 582 583
  if (pJson == NULL) {
    return -1;
  }

  if (tjsonToObject(pJson, "config", vnodeDecodeConfig, (void *)&pInfo->config) < 0) {
    goto _err;
  }

  if (tjsonToObject(pJson, "state", vnodeDecodeState, (void *)&pInfo->state) < 0) {
    goto _err;
  }

  tjsonDelete(pJson);

H
Hongze Cheng 已提交
584
  return 0;
H
Hongze Cheng 已提交
585 586 587 588

_err:
  tjsonDelete(pJson);
  return -1;
H
Hongze Cheng 已提交
589
}