taosdata/TDengine | Build 3638 | 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

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

×

588

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

×

589

    return -1;

×

592

  pMgmt->state.totalVnodes = numOfVnodes;

×

594

  int32_t threadNum = tsNumOfCores / 2;

×

595

  if (threadNum < 1) threadNum = 1;

×

596

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

598

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

×

599

  if (threads == NULL) {

×

600

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

×

601

    taosMemoryFree(pCfgs);

×

602

    return terrno;

×

605

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

×

606

    threads[t].threadIndex = t;

×

607

    threads[t].pMgmt = pMgmt;

×

608

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

×

611

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

×

612

    int32_t       t = v % threadNum;

×

613

    SVnodeThread *pThread = &threads[t];

×

614

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

×

617

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

×

619

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

×

620

    SVnodeThread *pThread = &threads[t];

×

621

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

×

624

    (void)taosThreadAttrInit(&thAttr);

×

625

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

626

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

×

627

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

×

630

    (void)taosThreadAttrDestroy(&thAttr);

×

633

  bool updateVnodesList = false;

×

635

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

×

636

    SVnodeThread *pThread = &threads[t];

×

637

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

×

638

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

×

639

      taosThreadClear(&pThread->thread);

×

641

    taosMemoryFree(pThread->pCfgs);

×

642

    if (pThread->updateVnodesList) updateVnodesList = true;

×

644

  taosMemoryFree(threads);

×

645

  taosMemoryFree(pCfgs);

×

647

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

×

648

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

×

649

    terrno = TSDB_CODE_VND_INIT_FAILED;

×

650

    return -1;

×

653

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

×

654

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

×

655

    return -1;

×

658

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

×

659

  return 0;

×

662

static void *vmCloseVnodeInThread(void *param) {

×

663

  SVnodeThread *pThread = param;

×

664

  SVnodeMgmt   *pMgmt = pThread->pMgmt;

×

666

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

×

667

  setThreadName("close-vnodes");

×

669

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

×

670

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

×

672

    char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

673

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

×

675

    tmsgReportStartup("vnode-close", stepDesc);

×

677

    vmCloseVnode(pMgmt, pVnode, false, false);

×

680

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

×

681

  return NULL;

×

684

static void vmCloseVnodes(SVnodeMgmt *pMgmt) {

×

685

  int32_t code = 0;

×

686

  dInfo("start to close all vnodes");

×

687

  tSingleWorkerCleanup(&pMgmt->mgmtWorker);

×

688

  dInfo("vnodes mgmt worker is stopped");

×

689

  tSingleWorkerCleanup(&pMgmt->mgmtMultiWorker);

×

690

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

×

692

  int32_t     numOfVnodes = 0;

×

693

  SVnodeObj **ppVnodes = NULL;

×

694

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

×

695

  if (code != 0) {

×

696

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

×

697

    return;

×

700

  int32_t threadNum = tsNumOfCores / 2;

×

701

  if (threadNum < 1) threadNum = 1;

×

702

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

704

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

×

705

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

×

706

    threads[t].threadIndex = t;

×

707

    threads[t].pMgmt = pMgmt;

×

708

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

×

711

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

×

712

    int32_t       t = v % threadNum;

×

713

    SVnodeThread *pThread = &threads[t];

×

714

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

×

715

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

×

719

  pMgmt->state.openVnodes = 0;

×

720

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

×

722

  int64_t st = taosGetTimestampMs();

×

723

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

×

724

  if (ppVnodes != NULL) {

×

725

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

×

726

      if (ppVnodes[i] != NULL) {

×

727

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

×

728

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

×

734

  int64_t et = taosGetTimestampMs();

×

735

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

×

737

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

×

738

    SVnodeThread *pThread = &threads[t];

×

739

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

×

742

    (void)taosThreadAttrInit(&thAttr);

×

743

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

745

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

×

746

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

×

749

    (void)taosThreadAttrDestroy(&thAttr);

×

752

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

×

753

    SVnodeThread *pThread = &threads[t];

×

754

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

×

755

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

×

756

      taosThreadClear(&pThread->thread);

×

758

    taosMemoryFree(pThread->ppVnodes);

×

760

  taosMemoryFree(threads);

×

762

  if (ppVnodes != NULL) {

×

763

    taosMemoryFree(ppVnodes);

×

766

  if (pMgmt->runngingHash != NULL) {

×

767

    taosHashCleanup(pMgmt->runngingHash);

×

768

    pMgmt->runngingHash = NULL;

×

771

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

×

772

  while (pIter) {

×

773

    SVnodeObj **ppVnode = pIter;

×

774

    vmFreeVnodeObj(ppVnode);

×

775

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

×

778

  if (pMgmt->closedHash != NULL) {

×

779

    taosHashCleanup(pMgmt->closedHash);

×

780

    pMgmt->closedHash = NULL;

×

783

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

×

784

  while (pIter) {

×

785

    SVnodeObj **ppVnode = pIter;

×

786

    vmFreeVnodeObj(ppVnode);

×

787

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

×

790

  if (pMgmt->creatingHash != NULL) {

×

791

    taosHashCleanup(pMgmt->creatingHash);

×

792

    pMgmt->creatingHash = NULL;

×

795

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

×

798

static void vmCleanup(SVnodeMgmt *pMgmt) {

×

799

  vmCloseVnodes(pMgmt);

×

800

  vmStopWorker(pMgmt);

×

801

  vnodeCleanup();

×

802

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

×

803

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

×

804

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

×

805

  taosMemoryFree(pMgmt);

×

806

×

808

static void vmCheckSyncTimeout(SVnodeMgmt *pMgmt) {

×

809

  int32_t     code = 0;

×

810

  int32_t     numOfVnodes = 0;

×

811

  SVnodeObj **ppVnodes = NULL;

×

812

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

×

813

  if (code != 0) {

×

814

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

×

815

    return;

×

818

  if (ppVnodes != NULL) {

×

819

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

×

820

      SVnodeObj *pVnode = ppVnodes[i];

×

821

      if (!pVnode->failed) {

×

822

        vnodeSyncCheckTimeout(pVnode->pImpl);

×

824

      vmReleaseVnode(pMgmt, pVnode);

×

826

    taosMemoryFree(ppVnodes);

×

830

static void *vmThreadFp(void *param) {

×

831

  SVnodeMgmt *pMgmt = param;

×

832

  int64_t     lastTime = 0;

×

833

  setThreadName("vnode-timer");

×

835

  while (1) {

×

836

    lastTime++;

×

837

    taosMsleep(100);

×

838

    if (pMgmt->stop) break;

×

839

    if (lastTime % 10 != 0) continue;

×

841

    int64_t sec = lastTime / 10;

×

842

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

×

843

      vmCheckSyncTimeout(pMgmt);

×

847

  return NULL;

×

850

static int32_t vmInitTimer(SVnodeMgmt *pMgmt) {

×

851

  int32_t      code = 0;

×

853

  (void)taosThreadAttrInit(&thAttr);

×

854

  (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

855

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

×

856

    code = TAOS_SYSTEM_ERROR(errno);

×

857

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

×

858

    return code;

×

861

  (void)taosThreadAttrDestroy(&thAttr);

×

862

  return 0;

×

865

static void vmCleanupTimer(SVnodeMgmt *pMgmt) {

×

866

  pMgmt->stop = true;

×

867

  if (taosCheckPthreadValid(pMgmt->thread)) {

×

868

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

×

869

    taosThreadClear(&pMgmt->thread);

×

871

×

873

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

×

874

  int32_t code = -1;

×

876

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

×

877

  if (pMgmt == NULL) {

×

878

    code = terrno;

×

879

    goto _OVER;

×

882

  pMgmt->pData = pInput->pData;

×

883

  pMgmt->path = pInput->path;

×

884

  pMgmt->name = pInput->name;

×

885

  pMgmt->msgCb = pInput->msgCb;

×

886

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

×

887

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

×

888

  pMgmt->msgCb.mgmt = pMgmt;

×

890

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

×

891

  if (code != 0) {

×

892

    code = TAOS_SYSTEM_ERROR(errno);

×

893

    goto _OVER;

×

896

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

×

897

  if (code != 0) {

×

898

    code = TAOS_SYSTEM_ERROR(errno);

×

899

    goto _OVER;

×

902

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

×

903

  if (code != 0) {

×

904

    code = TAOS_SYSTEM_ERROR(errno);

×

905

    goto _OVER;

×

908

  pMgmt->pTfs = pInput->pTfs;

×

909

  if (pMgmt->pTfs == NULL) {

×

910

    dError("tfs is null.");

×

911

    goto _OVER;

×

913

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

×

914

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

×

915

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

×

916

    goto _OVER;

×

919

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

×

921

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

×

922

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

×

923

    goto _OVER;

×

925

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

×

927

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

×

928

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

×

929

    goto _OVER;

×

931

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

×

933

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

×

934

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

×

935

    goto _OVER;

×

937

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

×

939

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

×

940

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

×

941

    goto _OVER;

×

943

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

×

945

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

×

946

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

×

947

    goto _OVER;

×

950

  code = 0;

×

952

_OVER:

×

953

  if (code == 0) {

×

954

    pOutput->pMgmt = pMgmt;

×

956

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

×

957

    vmCleanup(pMgmt);

×

960

  return code;

×

963

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

×

964

  *required = tsNumOfSupportVnodes > 0;

×

965

  return 0;

×

968

static void *vmRestoreVnodeInThread(void *param) {

×

969

  SVnodeThread *pThread = param;

×

970

  SVnodeMgmt   *pMgmt = pThread->pMgmt;

×

972

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

×

973

  setThreadName("restore-vnodes");

×

975

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

×

976

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

×

977

    if (pVnode->failed) {

×

978

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

×

979

      continue;

×

982

    char stepDesc[TSDB_STEP_DESC_LEN] = {0};

×

983

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

×

985

    tmsgReportStartup("vnode-restore", stepDesc);

×

987

    int32_t code = vnodeStart(pVnode->pImpl);

×

988

    if (code != 0) {

×

989

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

×

990

      pThread->failed++;

×

992

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

×

993

      pThread->opened++;

×

994

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

×

998

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

×

1000

  return NULL;

×

1003

static int32_t vmStartVnodes(SVnodeMgmt *pMgmt) {

×

1004

  int32_t     code = 0;

×

1005

  int32_t     numOfVnodes = 0;

×

1006

  SVnodeObj **ppVnodes = NULL;

×

1007

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

×

1008

  if (code != 0) {

×

1009

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

×

1010

    return code;

×

1013

  int32_t threadNum = tsNumOfCores / 2;

×

1014

  if (threadNum < 1) threadNum = 1;

×

1015

  int32_t vnodesPerThread = numOfVnodes / threadNum + 1;

×

1017

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

×

1018

  if (threads == NULL) {

×

1019

    return terrno;

×

1022

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

×

1023

    threads[t].threadIndex = t;

×

1024

    threads[t].pMgmt = pMgmt;

×

1025

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

×

1026

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

×

1027

      code = terrno;

×

1028

      break;

×

1032

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

×

1033

    int32_t       t = v % threadNum;

×

1034

    SVnodeThread *pThread = &threads[t];

×

1035

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

×

1036

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

×

1040

  pMgmt->state.openVnodes = 0;

×

1041

  pMgmt->state.dropVnodes = 0;

×

1042

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

×

1044

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

×

1045

    SVnodeThread *pThread = &threads[t];

×

1046

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

×

1049

    (void)taosThreadAttrInit(&thAttr);

×

1050

    (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

×

1051

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

×

1052

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

×

1055

    (void)taosThreadAttrDestroy(&thAttr);

×

1058

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

×

1059

    SVnodeThread *pThread = &threads[t];

×

1060

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

×

1061

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

×

1062

      taosThreadClear(&pThread->thread);

×

1064

    taosMemoryFree(pThread->ppVnodes);

×

1066

  taosMemoryFree(threads);

×

1068

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

×

1069

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

×

1070

    vmReleaseVnode(pMgmt, ppVnodes[i]);

×

1073

  if (ppVnodes != NULL) {

×

1074

    taosMemoryFree(ppVnodes);

×

1077

  return vmInitTimer(pMgmt);

×

1088

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

×

1090

SMgmtFunc vmGetMgmtFunc() {

×

1091

  SMgmtFunc mgmtFunc = {0};

×

1092

  mgmtFunc.openFp = vmInit;

×

1093

  mgmtFunc.closeFp = (NodeCloseFp)vmCleanup;

×

1094

  mgmtFunc.startFp = (NodeStartFp)vmStartVnodes;

×

1095

  mgmtFunc.stopFp = (NodeStopFp)vmStop;

×

1096

  mgmtFunc.requiredFp = vmRequire;

×

1097

  mgmtFunc.getHandlesFp = vmGetMsgHandles;

×

1099

  return mgmtFunc;

×

taosdata / TDengine / #3638

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

taosdata / TDengine / #3638

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