taosdata/TDengine | Build 3798 | source/dnode/mgmt/mgmt_vnode/src/vmInt.c | Coveralls

23

int32_t vmGetPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {

×

24

  int32_t    diskId = -1;

×

25

  SVnodeObj *pVnode = NULL;

×

27

  (void)taosThreadRwlockRdlock(&pMgmt->hashLock);

×

28

  int32_t r = taosHashGetDup(pMgmt->runngingHash, &vgId, sizeof(int32_t), (void *)&pVnode);

×

29

  if (pVnode != NULL) {

×

30

    diskId = pVnode->diskPrimary;

×

32

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

33

  return diskId;

×

36

static void vmFreeVnodeObj(SVnodeObj **ppVnode) {

×

37

  if (!ppVnode || !(*ppVnode)) return;

×

39

  SVnodeObj *pVnode = *ppVnode;

×

41

  int32_t refCount = atomic_load_32(&pVnode->refCount);

×

42

  while (refCount > 0) {

×

43

    dWarn("vgId:%d, vnode is refenced, retry to free in 200ms, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);

×

44

    taosMsleep(200);

×

45

    refCount = atomic_load_32(&pVnode->refCount);

×

48

  taosMemoryFree(pVnode->path);

×

49

  taosMemoryFree(pVnode);

×

50

  ppVnode[0] = NULL;

×

53

static int32_t vmRegisterCreatingState(SVnodeMgmt *pMgmt, int32_t vgId, int32_t diskId) {

×

54

  int32_t    code = 0;

×

55

  SVnodeObj *pCreatingVnode = taosMemoryCalloc(1, sizeof(SVnodeObj));

×

56

  if (pCreatingVnode == NULL) {

×

57

    dError("failed to alloc vnode since %s", terrstr());

×

58

    return terrno;

×

60

  (void)memset(pCreatingVnode, 0, sizeof(SVnodeObj));

×

62

  pCreatingVnode->vgId = vgId;

×

63

  pCreatingVnode->diskPrimary = diskId;

×

65

  code = taosThreadRwlockWrlock(&pMgmt->hashLock);

×

66

  if (code != 0) {

×

67

    taosMemoryFree(pCreatingVnode);

×

68

    return code;

×

71

  dTrace("vgId:%d, put vnode into creating hash, pCreatingVnode:%p", vgId, pCreatingVnode);

×

72

  code = taosHashPut(pMgmt->creatingHash, &vgId, sizeof(int32_t), &pCreatingVnode, sizeof(SVnodeObj *));

×

73

  if (code != 0) {

×

74

    dError("vgId:%d, failed to put vnode to creatingHash", vgId);

×

75

    taosMemoryFree(pCreatingVnode);

×

78

  int32_t r = taosThreadRwlockUnlock(&pMgmt->hashLock);

×

79

  if (r != 0) {

×

80

    dError("vgId:%d, failed to unlock since %s", vgId, tstrerror(r));

×

83

  return code;

×

86

static void vmUnRegisterCreatingState(SVnodeMgmt *pMgmt, int32_t vgId) {

×

87

  SVnodeObj *pOld = NULL;

×

89

  (void)taosThreadRwlockWrlock(&pMgmt->hashLock);

×

90

  int32_t r = taosHashGetDup(pMgmt->creatingHash, &vgId, sizeof(int32_t), (void *)&pOld);

×

91

  if (r != 0) {

×

92

    dError("vgId:%d, failed to get vnode from creating Hash", vgId);

×

94

  dTrace("vgId:%d, remove from creating Hash", vgId);

×

95

  r = taosHashRemove(pMgmt->creatingHash, &vgId, sizeof(int32_t));

×

96

  if (r != 0) {

×

97

    dError("vgId:%d, failed to remove vnode from creatingHash", vgId);

×

99

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

101

  if (pOld) {

×

102

    dTrace("vgId:%d, free vnode pOld:%p", vgId, &pOld);

×

103

    vmFreeVnodeObj(&pOld);

×

106

_OVER:

×

107

  if (r != 0) {

×

108

    dError("vgId:%d, failed to remove vnode from creatingHash since %s", vgId, tstrerror(r));

×

110

×

112

int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {

×

113

  int32_t code = 0;

×

114

  STfs   *pTfs = pMgmt->pTfs;

×

115

  int32_t diskId = 0;

×

116

  if (!pTfs) {

×

117

    return diskId;

×

121

  char vnodePath[TSDB_FILENAME_LEN] = {0};

×

122

  snprintf(vnodePath, TSDB_FILENAME_LEN - 1, "vnode%svnode%d", TD_DIRSEP, vgId);

×

123

  char fname[TSDB_FILENAME_LEN] = {0};

×

124

  char fnameTmp[TSDB_FILENAME_LEN] = {0};

×

125

  snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME);

×

126

  snprintf(fnameTmp, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME_TMP);

×

128

  diskId = tfsSearch(pTfs, 0, fname);

×

129

  if (diskId >= 0) {

×

130

    return diskId;

×

132

  diskId = tfsSearch(pTfs, 0, fnameTmp);

×

133

  if (diskId >= 0) {

×

134

    return diskId;

×

138

  int32_t     disks[TFS_MAX_DISKS_PER_TIER] = {0};

×

139

  int32_t     numOfVnodes = 0;

×

140

  SVnodeObj **ppVnodes = NULL;

×

142

  code = taosThreadMutexLock(&pMgmt->mutex);

×

143

  if (code != 0) {

×

144

    return code;

×

147

  code = vmGetAllVnodeListFromHashWithCreating(pMgmt, &numOfVnodes, &ppVnodes);

×

148

  if (code != 0) {

×

149

    int32_t r = taosThreadMutexUnlock(&pMgmt->mutex);

×

150

    if (r != 0) {

×

151

      dError("vgId:%d, failed to unlock mutex since %s", vgId, tstrerror(r));

×

153

    return code;

×

156

  for (int32_t v = 0; v < numOfVnodes; v++) {

×

157

    SVnodeObj *pVnode = ppVnodes[v];

×

158

    disks[pVnode->diskPrimary] += 1;

×

161

  int32_t minVal = INT_MAX;

×

162

  int32_t ndisk = tfsGetDisksAtLevel(pTfs, 0);

×

163

  diskId = 0;

×

164

  for (int32_t id = 0; id < ndisk; id++) {

×

165

    if (minVal > disks[id]) {

×

166

      minVal = disks[id];

×

167

      diskId = id;

×

170

  code = vmRegisterCreatingState(pMgmt, vgId, diskId);

×

171

  if (code != 0) {

×

172

    int32_t r = taosThreadMutexUnlock(&pMgmt->mutex);

×

173

    if (r != 0) {

×

174

      dError("vgId:%d, failed to unlock mutex since %s", vgId, tstrerror(r));

×

176

    goto _OVER;

×

179

  code = taosThreadMutexUnlock(&pMgmt->mutex);

×

180

  if (code != 0) {

×

181

    goto _OVER;

×

184

_OVER:

×

186

  for (int32_t i = 0; i < numOfVnodes; ++i) {

×

187

    if (ppVnodes == NULL || ppVnodes[i] == NULL) continue;

×

188

    vmReleaseVnode(pMgmt, ppVnodes[i]);

×

190

  if (ppVnodes != NULL) {

×

191

    taosMemoryFree(ppVnodes);

×

194

  if (code != 0) {

×

195

    dError("vgId:%d, failed to alloc disk since %s", vgId, tstrerror(code));

×

196

    return code;

×

198

    dInfo("vgId:%d, alloc disk:%d of level 0. ndisk:%d, vnodes: %d", vgId, diskId, ndisk, numOfVnodes);

×

199

    return diskId;

×

203

void vmCleanPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) { vmUnRegisterCreatingState(pMgmt, vgId); }

×

205

SVnodeObj *vmAcquireVnodeImpl(SVnodeMgmt *pMgmt, int32_t vgId, bool strict) {

×

206

  SVnodeObj *pVnode = NULL;

×

208

  (void)taosThreadRwlockRdlock(&pMgmt->hashLock);

×

209

  int32_t r = taosHashGetDup(pMgmt->runngingHash, &vgId, sizeof(int32_t), (void *)&pVnode);

×

210

  if (pVnode == NULL || strict && (pVnode->dropped || pVnode->failed)) {

×

211

    terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;

×

212

    pVnode = NULL;

×

214

    int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);

×

215

    dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);

×

217

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

219

  return pVnode;

×

222

SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) { return vmAcquireVnodeImpl(pMgmt, vgId, true); }

×

224

void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {

×

225

  if (pVnode == NULL) return;

×

228

  int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);

×

229

  dTrace("vgId:%d, release vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);

×

233

static int32_t vmRegisterRunningState(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {

×

234

  SVnodeObj *pOld = NULL;

×

236

  int32_t r = taosHashGetDup(pMgmt->runngingHash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);

×

237

  if (r != 0) {

×

238

    dError("vgId:%d, failed to get vnode from hash", pVnode->vgId);

×

240

  if (pOld) {

×

241

    vmFreeVnodeObj(&pOld);

×

243

  int32_t code = taosHashPut(pMgmt->runngingHash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));

×

245

  return code;

×

248

static void vmUnRegisterRunningState(SVnodeMgmt *pMgmt, int32_t vgId) {

×

249

  dInfo("vgId:%d, remove from hash", vgId);

×

250

  int32_t r = taosHashRemove(pMgmt->runngingHash, &vgId, sizeof(int32_t));

×

251

  if (r != 0) {

×

252

    dError("vgId:%d, failed to remove vnode since %s", vgId, tstrerror(r));

×

254

×

256

static int32_t vmRegisterClosedState(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {

×

257

  int32_t    code = 0;

×

258

  SVnodeObj *pClosedVnode = taosMemoryCalloc(1, sizeof(SVnodeObj));

×

259

  if (pClosedVnode == NULL) {

×

260

    dError("failed to alloc vnode since %s", terrstr());

×

261

    return terrno;

×

263

  (void)memset(pClosedVnode, 0, sizeof(SVnodeObj));

×

265

  pClosedVnode->vgId = pVnode->vgId;

×

266

  pClosedVnode->dropped = pVnode->dropped;

×

267

  pClosedVnode->vgVersion = pVnode->vgVersion;

×

268

  pClosedVnode->diskPrimary = pVnode->diskPrimary;

×

269

  pClosedVnode->toVgId = pVnode->toVgId;

×

271

  SVnodeObj *pOld = NULL;

×

272

  int32_t    r = taosHashGetDup(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);

×

273

  if (r != 0) {

×

274

    dError("vgId:%d, failed to get vnode from closedHash", pVnode->vgId);

×

276

  if (pOld) {

×

277

    vmFreeVnodeObj(&pOld);

×

279

  dInfo("vgId:%d, put vnode to closedHash", pVnode->vgId);

×

280

  r = taosHashPut(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), &pClosedVnode, sizeof(SVnodeObj *));

×

281

  if (r != 0) {

×

282

    dError("vgId:%d, failed to put vnode to closedHash", pVnode->vgId);

×

285

  return code;

×

288

static void vmUnRegisterClosedState(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {

×

289

  SVnodeObj *pOld = NULL;

×

290

  int32_t    r = taosHashGetDup(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);

×

291

  if (r != 0) {

×

292

    dError("vgId:%d, failed to get vnode from closedHash", pVnode->vgId);

×

294

  if (pOld != NULL) {

×

295

    vmFreeVnodeObj(&pOld);

×

296

    dInfo("vgId:%d, remove from closedHash", pVnode->vgId);

×

297

    r = taosHashRemove(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t));

×

298

    if (r != 0) {

×

299

      dError("vgId:%d, failed to remove vnode from hash", pVnode->vgId);

×

302

×

304

int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {

×

305

  SVnodeObj *pVnode = taosMemoryCalloc(1, sizeof(SVnodeObj));

×

306

  if (pVnode == NULL) {

×

307

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

308

    return -1;

×

311

  pVnode->vgId = pCfg->vgId;

×

312

  pVnode->vgVersion = pCfg->vgVersion;

×

313

  pVnode->diskPrimary = pCfg->diskPrimary;

×

314

  pVnode->refCount = 0;

×

315

  pVnode->dropped = 0;

×

316

  pVnode->failed = 0;

×

317

  pVnode->path = taosStrdup(pCfg->path);

×

318

  pVnode->pImpl = pImpl;

×

320

  if (pVnode->path == NULL) {

×

321

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

322

    taosMemoryFree(pVnode);

×

323

    return -1;

×

326

  if (pImpl) {

×

327

    if (vmAllocQueue(pMgmt, pVnode) != 0) {

×

328

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

329

      taosMemoryFree(pVnode->path);

×

330

      taosMemoryFree(pVnode);

×

331

      return -1;

×

334

    pVnode->failed = 1;

×

337

  (void)taosThreadRwlockWrlock(&pMgmt->hashLock);

×

338

  int32_t code = vmRegisterRunningState(pMgmt, pVnode);

×

339

  vmUnRegisterClosedState(pMgmt, pVnode);

×

340

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

342

  return code;

×

345

void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal, bool keepClosed) {

×

346

  char path[TSDB_FILENAME_LEN] = {0};

×

347

  bool atExit = true;

×

349

  if (pVnode->pImpl && vnodeIsLeader(pVnode->pImpl)) {

×

350

    vnodeProposeCommitOnNeed(pVnode->pImpl, atExit);

×

353

  (void)taosThreadRwlockWrlock(&pMgmt->hashLock);

×

354

  vmUnRegisterRunningState(pMgmt, pVnode->vgId);

×

355

  if (keepClosed) {

×

356

    if (vmRegisterClosedState(pMgmt, pVnode) != 0) {

×

357

      (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

358

      return;

×

361

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

×

363

  vmReleaseVnode(pMgmt, pVnode);

×

365

  if (pVnode->failed) {

×

366

    goto _closed;

×

368

  dInfo("vgId:%d, pre close", pVnode->vgId);

×

369

  vnodePreClose(pVnode->pImpl);

×

371

  dInfo("vgId:%d, wait for vnode ref become 0", pVnode->vgId);

×

372

  while (pVnode->refCount > 0) taosMsleep(10);

×

374

  dInfo("vgId:%d, wait for vnode write queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue,

×

376

  tMultiWorkerCleanup(&pVnode->pWriteW);

×

378

  dInfo("vgId:%d, wait for vnode sync queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,

×

380

  tMultiWorkerCleanup(&pVnode->pSyncW);

×

382

  dInfo("vgId:%d, wait for vnode sync rd queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,

×

384

  tMultiWorkerCleanup(&pVnode->pSyncRdW);

×

386

  dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,

×

388

  tMultiWorkerCleanup(&pVnode->pApplyW);

×

390

  dInfo("vgId:%d, wait for vnode fetch queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,

×

392

  while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);

×

394

  dInfo("vgId:%d, wait for vnode query queue:%p is empty", pVnode->vgId, pVnode->pQueryQ);

×

395

  while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10);

×

397

  tqNotifyClose(pVnode->pImpl->pTq);

×

399

  dInfo("vgId:%d, wait for vnode stream queue:%p is empty, %d remains", pVnode->vgId,

×

401

  while (!taosQueueEmpty(pVnode->pStreamQ)) taosMsleep(50);

×

403

  dInfo("vgId:%d, wait for vnode stream ctrl queue:%p is empty", pVnode->vgId, pVnode->pStreamCtrlQ);

×

404

  while (!taosQueueEmpty(pVnode->pStreamCtrlQ)) taosMsleep(50);

×

406

  dInfo("vgId:%d, wait for vnode stream long-exec queue:%p is empty, %d remains", pVnode->vgId,

×

408

  while (!taosQueueEmpty(pVnode->pStreamLongExecQ)) taosMsleep(50);

×

410

  dInfo("vgId:%d, wait for vnode stream chkpt queue:%p is empty", pVnode->vgId, pVnode->pStreamChkQ);

×

411

  while (!taosQueueEmpty(pVnode->pStreamChkQ)) taosMsleep(10);

×

413

  dInfo("vgId:%d, all vnode queues is empty", pVnode->vgId);

×

415

  dInfo("vgId:%d, post close", pVnode->vgId);

×

416

  vnodePostClose(pVnode->pImpl);

×

418

  vmFreeQueue(pMgmt, pVnode);

×

420

  if (commitAndRemoveWal) {

×

421

    dInfo("vgId:%d, commit data for vnode split", pVnode->vgId);

×

422

    if (vnodeSyncCommit(pVnode->pImpl) != 0) {

×

423

      dError("vgId:%d, failed to commit data", pVnode->vgId);

×

425

    if (vnodeBegin(pVnode->pImpl) != 0) {

×

426

      dError("vgId:%d, failed to begin", pVnode->vgId);

×

428

    dInfo("vgId:%d, commit data finished", pVnode->vgId);

×

431

  int32_t nodeId = vnodeNodeId(pVnode->pImpl);

×

432

  vnodeClose(pVnode->pImpl);

×

433

  pVnode->pImpl = NULL;

×

435

_closed:

×

436

  dInfo("vgId:%d, vnode is closed", pVnode->vgId);

×

438

  if (commitAndRemoveWal) {

×

439

    snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d%swal", TD_DIRSEP, pVnode->vgId, TD_DIRSEP);

×

440

    dInfo("vgId:%d, remove all wals, path:%s", pVnode->vgId, path);

×

441

    if (tfsRmdir(pMgmt->pTfs, path) != 0) {

×

442

      dTrace("vgId:%d, failed to remove wals, path:%s", pVnode->vgId, path);

×

444

    if (tfsMkdir(pMgmt->pTfs, path) != 0) {

×

445

      dTrace("vgId:%d, failed to create wals, path:%s", pVnode->vgId, path);

×

449

  if (pVnode->dropped) {

×

450

    dInfo("vgId:%d, vnode is destroyed, dropped:%d", pVnode->vgId, pVnode->dropped);

×

451

    snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pVnode->vgId);

×

452

    vnodeDestroy(pVnode->vgId, path, pMgmt->pTfs, nodeId);

×

455

  vmFreeVnodeObj(&pVnode);

×

458

void vmCloseFailedVnode(SVnodeMgmt *pMgmt, int32_t vgId) {

×

459

  int32_t r = 0;

×

460

  r = taosThreadRwlockWrlock(&pMgmt->hashLock);

×

461

  if (r != 0) {

×

462

    dError("vgId:%d, failed to lock since %s", vgId, tstrerror(r));

×

464

  if (r == 0) {

×

465

    vmUnRegisterRunningState(pMgmt, vgId);

×

467

  r = taosThreadRwlockUnlock(&pMgmt->hashLock);

×

468

  if (r != 0) {

×

469

    dError("vgId:%d, failed to unlock since %s", vgId, tstrerror(r));

×

471

×

473

static int32_t vmRestoreVgroupId(SWrapperCfg *pCfg, STfs *pTfs) {

×

474

  int32_t srcVgId = pCfg->vgId;

×

475

  int32_t dstVgId = pCfg->toVgId;

×

476

  if (dstVgId == 0) return 0;

×

481

  snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);

×

482

  snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);

×

484

  int32_t diskPrimary = pCfg->diskPrimary;

×

485

  int32_t vgId = vnodeRestoreVgroupId(srcPath, dstPath, srcVgId, dstVgId, diskPrimary, pTfs);

×

486

  if (vgId <= 0) {

×

487

    dError("vgId:%d, failed to restore vgroup id. srcPath: %s", pCfg->vgId, srcPath);

×

488

    return -1;

×

491

  pCfg->vgId = vgId;

×

492

  pCfg->toVgId = 0;

×

493

  return 0;

×

496

static void *vmOpenVnodeInThread(void *param) {

×

497

  SVnodeThread *pThread = param;

×

498

  SVnodeMgmt   *pMgmt = pThread->pMgmt;

×

501

  dInfo("thread:%d, start to open or destroy %d vnodes", pThread->threadIndex, pThread->vnodeNum);

×

502

  setThreadName("open-vnodes");

×

504

  for (int32_t v = 0; v < pThread->vnodeNum; ++v) {

×

505

    SWrapperCfg *pCfg = &pThread->pCfgs[v];

×

506

    if (pCfg->dropped) {

×

507

      char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

508

      snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to destroy, %d of %d have been dropped", pCfg->vgId,

×

510

      tmsgReportStartup("vnode-destroy", stepDesc);

×

512

      snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId);

×

513

      vnodeDestroy(pCfg->vgId, path, pMgmt->pTfs, 0);

×

514

      pThread->updateVnodesList = true;

×

515

      pThread->dropped++;

×

516

      (void)atomic_add_fetch_32(&pMgmt->state.dropVnodes, 1);

×

517

      continue;

×

520

    char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

521

    snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to restore, %d of %d have been opened", pCfg->vgId,

×

523

    tmsgReportStartup("vnode-open", stepDesc);

×

525

    if (pCfg->toVgId) {

×

526

      if (vmRestoreVgroupId(pCfg, pMgmt->pTfs) != 0) {

×

527

        dError("vgId:%d, failed to restore vgroup id by thread:%d", pCfg->vgId, pThread->threadIndex);

×

528

        pThread->failed++;

×

529

        continue;

×

531

      pThread->updateVnodesList = true;

×

534

    int32_t diskPrimary = pCfg->diskPrimary;

×

535

    snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId);

×

537

    SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb, false);

×

539

    if (pImpl == NULL) {

×

540

      dError("vgId:%d, failed to open vnode by thread:%d since %s", pCfg->vgId, pThread->threadIndex, terrstr());

×

541

      if (terrno != TSDB_CODE_NEED_RETRY) {

×

542

        pThread->failed++;

×

543

        continue;

×

547

    if (vmOpenVnode(pMgmt, pCfg, pImpl) != 0) {

×

548

      dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);

×

549

      pThread->failed++;

×

550

      continue;

×

553

    dInfo("vgId:%d, is opened by thread:%d", pCfg->vgId, pThread->threadIndex);

×

554

    pThread->opened++;

×

555

    (void)atomic_add_fetch_32(&pMgmt->state.openVnodes, 1);

×

558

  dInfo("thread:%d, numOfVnodes:%d, opened:%d dropped:%d failed:%d", pThread->threadIndex, pThread->vnodeNum,

×

560

  return NULL;

×

563

static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {

×

564

  pMgmt->runngingHash =

×

565

      taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

566

  if (pMgmt->runngingHash == NULL) {

×

567

    dError("failed to init vnode hash since %s", terrstr());

×

568

    return TSDB_CODE_OUT_OF_MEMORY;

×

571

  pMgmt->closedHash =

×

572

      taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

573

  if (pMgmt->closedHash == NULL) {

×

574

    dError("failed to init vnode closed hash since %s", terrstr());

×

575

    return TSDB_CODE_OUT_OF_MEMORY;

×

578

  pMgmt->creatingHash =

×

579

      taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

580

  if (pMgmt->creatingHash == NULL) {

×

581

    dError("failed to init vnode creatingHash hash since %s", terrstr());

×

582

    return TSDB_CODE_OUT_OF_MEMORY;

×

585

  SWrapperCfg *pCfgs = NULL;

×

586

  int32_t      numOfVnodes = 0;

×

587

  int32_t      code = 0;

×

588

  if ((code = vmGetVnodeListFromFile(pMgmt, &pCfgs, &numOfVnodes)) != 0) {

×

589

    dInfo("failed to get vnode list from disk since %s", tstrerror(code));

×

590

    return code;

×

593

  pMgmt->state.totalVnodes = numOfVnodes;

×

595

  int32_t threadNum = tsNumOfCores / 2;

×

596

  if (threadNum < 1) threadNum = 1;

×

597

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

599

  SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread));

×

600

  if (threads == NULL) {

×

601

    dError("failed to allocate memory for threads since %s", terrstr());

×

602

    taosMemoryFree(pCfgs);

×

603

    return terrno;

×

606

  for (int32_t t = 0; t < threadNum; ++t) {

×

607

    threads[t].threadIndex = t;

×

608

    threads[t].pMgmt = pMgmt;

×

609

    threads[t].pCfgs = taosMemoryCalloc(vnodesPerThread, sizeof(SWrapperCfg));

×

612

  for (int32_t v = 0; v < numOfVnodes; ++v) {

×

613

    int32_t       t = v % threadNum;

×

614

    SVnodeThread *pThread = &threads[t];

×

615

    pThread->pCfgs[pThread->vnodeNum++] = pCfgs[v];

×

618

  dInfo("open %d vnodes with %d threads", numOfVnodes, threadNum);

×

620

  for (int32_t t = 0; t < threadNum; ++t) {

×

621

    SVnodeThread *pThread = &threads[t];

×

622

    if (pThread->vnodeNum == 0) continue;

×

625

    (void)taosThreadAttrInit(&thAttr);

×

626

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

630

    if (taosThreadCreate(&pThread->thread, &thAttr, vmOpenVnodeInThread, pThread) != 0) {

×

631

      dError("thread:%d, failed to create thread to open vnode, reason:%s", pThread->threadIndex, strerror(ERRNO));

×

634

    (void)taosThreadAttrDestroy(&thAttr);

×

637

  bool updateVnodesList = false;

×

639

  for (int32_t t = 0; t < threadNum; ++t) {

×

640

    SVnodeThread *pThread = &threads[t];

×

641

    if (pThread->vnodeNum > 0 && taosCheckPthreadValid(pThread->thread)) {

×

642

      (void)taosThreadJoin(pThread->thread, NULL);

×

643

      taosThreadClear(&pThread->thread);

×

645

    taosMemoryFree(pThread->pCfgs);

×

646

    if (pThread->updateVnodesList) updateVnodesList = true;

×

648

  taosMemoryFree(threads);

×

649

  taosMemoryFree(pCfgs);

×

651

  if ((pMgmt->state.openVnodes + pMgmt->state.dropVnodes) != pMgmt->state.totalVnodes) {

×

652

    dError("there are total vnodes:%d, opened:%d", pMgmt->state.totalVnodes, pMgmt->state.openVnodes);

×

653

    return terrno = TSDB_CODE_VND_INIT_FAILED;

×

656

  if (updateVnodesList && (code = vmWriteVnodeListToFile(pMgmt)) != 0) {

×

657

    dError("failed to write vnode list since %s", tstrerror(code));

×

658

    return code;

×

661

  dInfo("successfully opened %d vnodes", pMgmt->state.totalVnodes);

×

662

  return 0;

×

665

static void *vmCloseVnodeInThread(void *param) {

×

666

  SVnodeThread *pThread = param;

×

667

  SVnodeMgmt   *pMgmt = pThread->pMgmt;

×

669

  dInfo("thread:%d, start to close %d vnodes", pThread->threadIndex, pThread->vnodeNum);

×

670

  setThreadName("close-vnodes");

×

672

  for (int32_t v = 0; v < pThread->vnodeNum; ++v) {

×

673

    SVnodeObj *pVnode = pThread->ppVnodes[v];

×

675

    char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

676

    snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to close, %d of %d have been closed", pVnode->vgId,

×

678

    tmsgReportStartup("vnode-close", stepDesc);

×

680

    vmCloseVnode(pMgmt, pVnode, false, false);

×

683

  dInfo("thread:%d, numOfVnodes:%d is closed", pThread->threadIndex, pThread->vnodeNum);

×

684

  return NULL;

×

687

static void vmCloseVnodes(SVnodeMgmt *pMgmt) {

×

688

  int32_t code = 0;

×

689

  dInfo("start to close all vnodes");

×

690

  tSingleWorkerCleanup(&pMgmt->mgmtWorker);

×

691

  dInfo("vnodes mgmt worker is stopped");

×

692

  tSingleWorkerCleanup(&pMgmt->mgmtMultiWorker);

×

693

  dInfo("vnodes multiple mgmt worker is stopped");

×

695

  int32_t     numOfVnodes = 0;

×

696

  SVnodeObj **ppVnodes = NULL;

×

697

  code = vmGetVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes);

×

698

  if (code != 0) {

×

699

    dError("failed to get vnode list since %s", tstrerror(code));

×

700

    return;

×

703

  int32_t threadNum = tsNumOfCores / 2;

×

704

  if (threadNum < 1) threadNum = 1;

×

705

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

707

  SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread));

×

708

  for (int32_t t = 0; t < threadNum; ++t) {

×

709

    threads[t].threadIndex = t;

×

710

    threads[t].pMgmt = pMgmt;

×

711

    threads[t].ppVnodes = taosMemoryCalloc(vnodesPerThread, sizeof(SVnode *));

×

714

  for (int32_t v = 0; v < numOfVnodes; ++v) {

×

715

    int32_t       t = v % threadNum;

×

716

    SVnodeThread *pThread = &threads[t];

×

717

    if (pThread->ppVnodes != NULL && ppVnodes != NULL) {

×

718

      pThread->ppVnodes[pThread->vnodeNum++] = ppVnodes[v];

×

722

  pMgmt->state.openVnodes = 0;

×

723

  dInfo("close %d vnodes with %d threads", numOfVnodes, threadNum);

×

725

  int64_t st = taosGetTimestampMs();

×

726

  dInfo("notify all streams closed in all %d vnodes, ts:%" PRId64, numOfVnodes, st);

×

727

  if (ppVnodes != NULL) {

×

728

    for (int32_t i = 0; i < numOfVnodes; ++i) {

×

729

      if (ppVnodes[i] != NULL) {

×

730

        if (ppVnodes[i]->pImpl != NULL) {

×

731

          tqNotifyClose(ppVnodes[i]->pImpl->pTq);

×

737

  int64_t et = taosGetTimestampMs();

×

738

  dInfo("notify close stream completed in %d vnodes, elapsed time: %" PRId64 "ms", numOfVnodes, et - st);

×

740

  for (int32_t t = 0; t < threadNum; ++t) {

×

741

    SVnodeThread *pThread = &threads[t];

×

742

    if (pThread->vnodeNum == 0) continue;

×

745

    (void)taosThreadAttrInit(&thAttr);

×

746

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

750

    if (taosThreadCreate(&pThread->thread, &thAttr, vmCloseVnodeInThread, pThread) != 0) {

×

751

      dError("thread:%d, failed to create thread to close vnode since %s", pThread->threadIndex, strerror(ERRNO));

×

754

    (void)taosThreadAttrDestroy(&thAttr);

×

757

  for (int32_t t = 0; t < threadNum; ++t) {

×

758

    SVnodeThread *pThread = &threads[t];

×

759

    if (pThread->vnodeNum > 0 && taosCheckPthreadValid(pThread->thread)) {

×

760

      (void)taosThreadJoin(pThread->thread, NULL);

×

761

      taosThreadClear(&pThread->thread);

×

763

    taosMemoryFree(pThread->ppVnodes);

×

765

  taosMemoryFree(threads);

×

767

  if (ppVnodes != NULL) {

×

768

    taosMemoryFree(ppVnodes);

×

771

  if (pMgmt->runngingHash != NULL) {

×

772

    taosHashCleanup(pMgmt->runngingHash);

×

773

    pMgmt->runngingHash = NULL;

×

776

  void *pIter = taosHashIterate(pMgmt->closedHash, NULL);

×

777

  while (pIter) {

×

778

    SVnodeObj **ppVnode = pIter;

×

779

    vmFreeVnodeObj(ppVnode);

×

780

    pIter = taosHashIterate(pMgmt->closedHash, pIter);

×

783

  if (pMgmt->closedHash != NULL) {

×

784

    taosHashCleanup(pMgmt->closedHash);

×

785

    pMgmt->closedHash = NULL;

×

788

  pIter = taosHashIterate(pMgmt->creatingHash, NULL);

×

789

  while (pIter) {

×

790

    SVnodeObj **ppVnode = pIter;

×

791

    vmFreeVnodeObj(ppVnode);

×

792

    pIter = taosHashIterate(pMgmt->creatingHash, pIter);

×

795

  if (pMgmt->creatingHash != NULL) {

×

796

    taosHashCleanup(pMgmt->creatingHash);

×

797

    pMgmt->creatingHash = NULL;

×

800

  dInfo("total vnodes:%d are all closed", numOfVnodes);

×

803

static void vmCleanup(SVnodeMgmt *pMgmt) {

×

804

  vmCloseVnodes(pMgmt);

×

805

  vmStopWorker(pMgmt);

×

806

  vnodeCleanup();

×

807

  (void)taosThreadRwlockDestroy(&pMgmt->hashLock);

×

808

  (void)taosThreadMutexDestroy(&pMgmt->mutex);

×

809

  (void)taosThreadMutexDestroy(&pMgmt->fileLock);

×

810

  taosMemoryFree(pMgmt);

×

811

×

813

static void vmCheckSyncTimeout(SVnodeMgmt *pMgmt) {

×

814

  int32_t     code = 0;

×

815

  int32_t     numOfVnodes = 0;

×

816

  SVnodeObj **ppVnodes = NULL;

×

817

  code = vmGetVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes);

×

818

  if (code != 0) {

×

819

    dError("failed to get vnode list since %s", tstrerror(code));

×

820

    return;

×

823

  if (ppVnodes != NULL) {

×

824

    for (int32_t i = 0; i < numOfVnodes; ++i) {

×

825

      SVnodeObj *pVnode = ppVnodes[i];

×

826

      if (!pVnode->failed) {

×

827

        vnodeSyncCheckTimeout(pVnode->pImpl);

×

829

      vmReleaseVnode(pMgmt, pVnode);

×

831

    taosMemoryFree(ppVnodes);

×

835

static void *vmThreadFp(void *param) {

×

836

  SVnodeMgmt *pMgmt = param;

×

837

  int64_t     lastTime = 0;

×

838

  setThreadName("vnode-timer");

×

840

  while (1) {

×

841

    lastTime++;

×

842

    taosMsleep(100);

×

843

    if (pMgmt->stop) break;

×

844

    if (lastTime % 10 != 0) continue;

×

846

    int64_t sec = lastTime / 10;

×

847

    if (sec % (VNODE_TIMEOUT_SEC / 2) == 0) {

×

848

      vmCheckSyncTimeout(pMgmt);

×

852

  return NULL;

×

855

static int32_t vmInitTimer(SVnodeMgmt *pMgmt) {

×

856

  int32_t      code = 0;

×

858

  (void)taosThreadAttrInit(&thAttr);

×

859

  (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

863

  if (taosThreadCreate(&pMgmt->thread, &thAttr, vmThreadFp, pMgmt) != 0) {

×

864

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

865

    dError("failed to create vnode timer thread since %s", tstrerror(code));

×

866

    return code;

×

869

  (void)taosThreadAttrDestroy(&thAttr);

×

870

  return 0;

×

873

static void vmCleanupTimer(SVnodeMgmt *pMgmt) {

×

874

  pMgmt->stop = true;

×

875

  if (taosCheckPthreadValid(pMgmt->thread)) {

×

876

    (void)taosThreadJoin(pMgmt->thread, NULL);

×

877

    taosThreadClear(&pMgmt->thread);

×

879

×

881

static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {

×

882

  int32_t code = -1;

×

884

  SVnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SVnodeMgmt));

×

885

  if (pMgmt == NULL) {

×

886

    code = terrno;

×

887

    goto _OVER;

×

890

  pMgmt->pData = pInput->pData;

×

891

  pMgmt->path = pInput->path;

×

892

  pMgmt->name = pInput->name;

×

893

  pMgmt->msgCb = pInput->msgCb;

×

894

  pMgmt->msgCb.putToQueueFp = (PutToQueueFp)vmPutRpcMsgToQueue;

×

895

  pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize;

×

896

  pMgmt->msgCb.mgmt = pMgmt;

×

898

  code = taosThreadRwlockInit(&pMgmt->hashLock, NULL);

×

899

  if (code != 0) {

×

900

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

901

    goto _OVER;

×

904

  code = taosThreadMutexInit(&pMgmt->mutex, NULL);

×

905

  if (code != 0) {

×

906

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

907

    goto _OVER;

×

910

  code = taosThreadMutexInit(&pMgmt->fileLock, NULL);

×

911

  if (code != 0) {

×

912

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

913

    goto _OVER;

×

916

  pMgmt->pTfs = pInput->pTfs;

×

917

  if (pMgmt->pTfs == NULL) {

×

918

    dError("tfs is null.");

×

919

    goto _OVER;

×

921

  tmsgReportStartup("vnode-tfs", "initialized");

×

922

  if ((code = walInit(pInput->stopDnodeFp)) != 0) {

×

923

    dError("failed to init wal since %s", tstrerror(code));

×

924

    goto _OVER;

×

927

  tmsgReportStartup("vnode-wal", "initialized");

×

929

  if ((code = syncInit()) != 0) {

×

930

    dError("failed to open sync since %s", tstrerror(code));

×

931

    goto _OVER;

×

933

  tmsgReportStartup("vnode-sync", "initialized");

×

935

  if ((code = vnodeInit(pInput->stopDnodeFp)) != 0) {

×

936

    dError("failed to init vnode since %s", tstrerror(code));

×

937

    goto _OVER;

×

939

  tmsgReportStartup("vnode-commit", "initialized");

×

941

  if ((code = vmStartWorker(pMgmt)) != 0) {

×

942

    dError("failed to init workers since %s", tstrerror(code));

×

943

    goto _OVER;

×

945

  tmsgReportStartup("vnode-worker", "initialized");

×

947

  if ((code = vmOpenVnodes(pMgmt)) != 0) {

×

948

    dError("failed to open all vnodes since %s", tstrerror(code));

×

949

    goto _OVER;

×

951

  tmsgReportStartup("vnode-vnodes", "initialized");

×

953

  if ((code = udfcOpen()) != 0) {

×

954

    dError("failed to open udfc in vnode since %s", tstrerror(code));

×

955

    goto _OVER;

×

958

  code = 0;

×

960

_OVER:

×

961

  if (code == 0) {

×

962

    pOutput->pMgmt = pMgmt;

×

964

    dError("failed to init vnodes-mgmt since %s", tstrerror(code));

×

965

    vmCleanup(pMgmt);

×

968

  return code;

×

971

static int32_t vmRequire(const SMgmtInputOpt *pInput, bool *required) {

×

972

  *required = tsNumOfSupportVnodes > 0;

×

973

  return 0;

×

976

static void *vmRestoreVnodeInThread(void *param) {

×

977

  SVnodeThread *pThread = param;

×

978

  SVnodeMgmt   *pMgmt = pThread->pMgmt;

×

980

  dInfo("thread:%d, start to restore %d vnodes", pThread->threadIndex, pThread->vnodeNum);

×

981

  setThreadName("restore-vnodes");

×

983

  for (int32_t v = 0; v < pThread->vnodeNum; ++v) {

×

984

    SVnodeObj *pVnode = pThread->ppVnodes[v];

×

985

    if (pVnode->failed) {

×

986

      dError("vgId:%d, cannot restore a vnode in failed mode.", pVnode->vgId);

×

987

      continue;

×

990

    char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

991

    snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to restore, %d of %d have been restored", pVnode->vgId,

×

993

    tmsgReportStartup("vnode-restore", stepDesc);

×

995

    int32_t code = vnodeStart(pVnode->pImpl);

×

996

    if (code != 0) {

×

997

      dError("vgId:%d, failed to restore vnode by thread:%d", pVnode->vgId, pThread->threadIndex);

×

998

      pThread->failed++;

×

1000

      dInfo("vgId:%d, is restored by thread:%d", pVnode->vgId, pThread->threadIndex);

×

1001

      pThread->opened++;

×

1002

      (void)atomic_add_fetch_32(&pMgmt->state.openVnodes, 1);

×

1006

  dInfo("thread:%d, numOfVnodes:%d, restored:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened,

×

1008

  return NULL;

×

1011

static int32_t vmStartVnodes(SVnodeMgmt *pMgmt) {

×

1012

  int32_t     code = 0;

×

1013

  int32_t     numOfVnodes = 0;

×

1014

  SVnodeObj **ppVnodes = NULL;

×

1015

  code = vmGetVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes);

×

1016

  if (code != 0) {

×

1017

    dError("failed to get vnode list since %s", tstrerror(code));

×

1018

    return code;

×

1021

  int32_t threadNum = tsNumOfCores / 2;

×

1022

  if (threadNum < 1) threadNum = 1;

×

1023

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

1025

  SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread));

×

1026

  if (threads == NULL) {

×

1027

    return terrno;

×

1030

  for (int32_t t = 0; t < threadNum; ++t) {

×

1031

    threads[t].threadIndex = t;

×

1032

    threads[t].pMgmt = pMgmt;

×

1033

    threads[t].ppVnodes = taosMemoryCalloc(vnodesPerThread, sizeof(SVnode *));

×

1034

    if (threads[t].ppVnodes == NULL) {

×

1035

      code = terrno;

×

1036

      break;

×

1040

  for (int32_t v = 0; v < numOfVnodes; ++v) {

×

1041

    int32_t       t = v % threadNum;

×

1042

    SVnodeThread *pThread = &threads[t];

×

1043

    if (pThread->ppVnodes != NULL && ppVnodes != NULL) {

×

1044

      pThread->ppVnodes[pThread->vnodeNum++] = ppVnodes[v];

×

1048

  pMgmt->state.openVnodes = 0;

×

1049

  pMgmt->state.dropVnodes = 0;

×

1050

  dInfo("restore %d vnodes with %d threads", numOfVnodes, threadNum);

×

1052

  for (int32_t t = 0; t < threadNum; ++t) {

×

1053

    SVnodeThread *pThread = &threads[t];

×

1054

    if (pThread->vnodeNum == 0) continue;

×

1057

    (void)taosThreadAttrInit(&thAttr);

×

1058

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

1059

    if (taosThreadCreate(&pThread->thread, &thAttr, vmRestoreVnodeInThread, pThread) != 0) {

×

1060

      dError("thread:%d, failed to create thread to restore vnode since %s", pThread->threadIndex, strerror(ERRNO));

×

1063

    (void)taosThreadAttrDestroy(&thAttr);

×

1066

  for (int32_t t = 0; t < threadNum; ++t) {

×

1067

    SVnodeThread *pThread = &threads[t];

×

1068

    if (pThread->vnodeNum > 0 && taosCheckPthreadValid(pThread->thread)) {

×

1069

      (void)taosThreadJoin(pThread->thread, NULL);

×

1070

      taosThreadClear(&pThread->thread);

×

1072

    taosMemoryFree(pThread->ppVnodes);

×

1074

  taosMemoryFree(threads);

×

1076

  for (int32_t i = 0; i < numOfVnodes; ++i) {

×

1077

    if (ppVnodes == NULL || ppVnodes[i] == NULL) continue;

×

1078

    vmReleaseVnode(pMgmt, ppVnodes[i]);

×

1081

  if (ppVnodes != NULL) {

×

1082

    taosMemoryFree(ppVnodes);

×

1085

  return vmInitTimer(pMgmt);

×

1096

static void vmStop(SVnodeMgmt *pMgmt) { vmCleanupTimer(pMgmt); }

×

1098

SMgmtFunc vmGetMgmtFunc() {

×

1099

  SMgmtFunc mgmtFunc = {0};

×

1100

  mgmtFunc.openFp = vmInit;

×

1101

  mgmtFunc.closeFp = (NodeCloseFp)vmCleanup;

×

1102

  mgmtFunc.startFp = (NodeStartFp)vmStartVnodes;

×

1103

  mgmtFunc.stopFp = (NodeStopFp)vmStop;

×

1104

  mgmtFunc.requiredFp = vmRequire;

×

1105

  mgmtFunc.getHandlesFp = vmGetMsgHandles;

×

1107

  return mgmtFunc;

×

taosdata / TDengine / #3798

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3798

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous