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

20

static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) {

×

21

  if (pMsg->info.handle == NULL) return;

×

22

  SRpcMsg rsp = {

×

24

      .pCont = pMsg->info.rsp,

×

25

      .contLen = pMsg->info.rspLen,

×

28

  tmsgSendRsp(&rsp);

×

31

static void vmProcessMultiMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {

×

32

  SVnodeMgmt     *pMgmt = pInfo->ahandle;

×

33

  int32_t         code = -1;

×

34

  const STraceId *trace = &pMsg->info.traceId;

×

36

  dGTrace("msg:%p, get from vnode-multi-mgmt queue", pMsg);

×

37

  switch (pMsg->msgType) {

×

38

    case TDMT_DND_CREATE_VNODE:

×

39

      code = vmProcessCreateVnodeReq(pMgmt, pMsg);

×

40

      break;

×

43

  if (IsReq(pMsg)) {

×

44

    if (code != 0) {

×

45

      if (terrno != 0) code = terrno;

×

46

      dGError("msg:%p, failed to process since %s, type:%s", pMsg, tstrerror(code), TMSG_INFO(pMsg->msgType));

×

48

    vmSendRsp(pMsg, code);

×

51

  dGTrace("msg:%p, is freed, code:0x%x", pMsg, code);

×

52

  rpcFreeCont(pMsg->pCont);

×

53

  taosFreeQitem(pMsg);

×

54

×

56

static void vmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {

×

57

  SVnodeMgmt     *pMgmt = pInfo->ahandle;

×

58

  int32_t         code = -1;

×

59

  const STraceId *trace = &pMsg->info.traceId;

×

61

  dGTrace("msg:%p, get from vnode-mgmt queue", pMsg);

×

62

  switch (pMsg->msgType) {

×

63

    case TDMT_DND_CREATE_VNODE:

×

64

      code = vmProcessCreateVnodeReq(pMgmt, pMsg);

×

65

      break;

×

66

    case TDMT_DND_DROP_VNODE:

×

67

      code = vmProcessDropVnodeReq(pMgmt, pMsg);

×

68

      break;

×

69

    case TDMT_VND_ALTER_REPLICA:

×

70

      code = vmProcessAlterVnodeReplicaReq(pMgmt, pMsg);

×

71

      break;

×

72

    case TDMT_VND_DISABLE_WRITE:

×

73

      code = vmProcessDisableVnodeWriteReq(pMgmt, pMsg);

×

74

      break;

×

75

    case TDMT_VND_ALTER_HASHRANGE:

×

76

      code = vmProcessAlterHashRangeReq(pMgmt, pMsg);

×

77

      break;

×

78

    case TDMT_DND_ALTER_VNODE_TYPE:

×

79

      code = vmProcessAlterVnodeTypeReq(pMgmt, pMsg);

×

80

      break;

×

81

    case TDMT_DND_CHECK_VNODE_LEARNER_CATCHUP:

×

82

      code = vmProcessCheckLearnCatchupReq(pMgmt, pMsg);

×

83

      break;

×

84

    case TDMT_VND_ARB_HEARTBEAT:

×

85

      code = vmProcessArbHeartBeatReq(pMgmt, pMsg);

×

86

      break;

×

87

    default:

×

88

      terrno = TSDB_CODE_MSG_NOT_PROCESSED;

×

89

      dGError("msg:%p, not processed in vnode-mgmt queue", pMsg);

×

92

  if (IsReq(pMsg)) {

×

93

    if (code != 0) {

×

94

      if (terrno != 0) code = terrno;

×

95

      dGError("msg:%p, failed to process since %s, type:%s", pMsg, tstrerror(code), TMSG_INFO(pMsg->msgType));

×

97

    vmSendRsp(pMsg, code);

×

100

  dGTrace("msg:%p, is freed, code:0x%x", pMsg, code);

×

101

  rpcFreeCont(pMsg->pCont);

×

102

  taosFreeQitem(pMsg);

×

103

×

105

static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {

×

106

  SVnodeObj      *pVnode = pInfo->ahandle;

×

107

  const STraceId *trace = &pMsg->info.traceId;

×

109

  dGTrace("vgId:%d, msg:%p, get from vnode-query queue", pVnode->vgId, pMsg);

×

110

  int32_t code = vnodeProcessQueryMsg(pVnode->pImpl, pMsg, pInfo);

×

111

  if (code != 0) {

×

112

    if (terrno != 0) code = terrno;

×

113

    dGError("vgId:%d, msg:%p, failed to query since %s", pVnode->vgId, pMsg, tstrerror(code));

×

114

    vmSendRsp(pMsg, code);

×

117

  dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

118

  rpcFreeCont(pMsg->pCont);

×

119

  taosFreeQitem(pMsg);

×

120

×

122

static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {

×

124

  SVnodeObj      *pVnode = pInfo->ahandle;

×

125

  const STraceId *trace = &pMsg->info.traceId;

×

127

  dGTrace("vgId:%d, msg:%p, get from vnode-stream queue", pVnode->vgId, pMsg);

×

128

  int32_t code = vnodeProcessStreamMsg(pVnode->pImpl, pMsg, pInfo);

×

129

  if (code != 0) {

×

130

    terrno = code;

×

131

    dGError("vgId:%d, msg:%p, failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),

×

133

    vmSendRsp(pMsg, code);

×

136

  dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

137

  rpcFreeCont(pMsg->pCont);

×

138

  taosFreeQitem(pMsg);

×

140

×

142

static void vmProcessStreamCtrlQueue(SQueueInfo *pInfo, STaosQall* pQall, int32_t numOfItems) {

×

144

  SVnodeObj *pVnode = pInfo->ahandle;

×

145

  void      *pItem = NULL;

×

146

  int32_t    code = 0;

×

148

  while (1) {

×

149

    if (taosGetQitem(pQall, &pItem) == 0) {

×

150

      break;

×

153

    SRpcMsg        *pMsg = pItem;

×

154

    const STraceId *trace = &pMsg->info.traceId;

×

156

    dGTrace("vgId:%d, msg:%p, get from vnode-stream-ctrl queue", pVnode->vgId, pMsg);

×

157

    code = vnodeProcessStreamCtrlMsg(pVnode->pImpl, pMsg, pInfo);

×

158

    if (code != 0) {

×

159

      terrno = code;

×

160

      dGError("vgId:%d, msg:%p, failed to process stream ctrl msg %s since %s", pVnode->vgId, pMsg,

×

162

      vmSendRsp(pMsg, code);

×

165

    dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

166

    rpcFreeCont(pMsg->pCont);

×

167

    taosFreeQitem(pMsg);

×

170

×

172

static void vmProcessStreamChkptQueue(SQueueInfo *pInfo, STaosQall* pQall, int32_t numOfItems) {

×

173

  SVnodeObj *pVnode = pInfo->ahandle;

×

174

  void      *pItem = NULL;

×

175

  int32_t    code = 0;

×

177

  while (1) {

×

178

    if (taosGetQitem(pQall, &pItem) == 0) {

×

179

      break;

×

182

    SRpcMsg        *pMsg = pItem;

×

183

    const STraceId *trace = &pMsg->info.traceId;

×

185

    dGTrace("vgId:%d, msg:%p, get from vnode-stream-chkpt queue", pVnode->vgId, pMsg);

×

186

    code = vnodeProcessStreamChkptMsg(pVnode->pImpl, pMsg, pInfo);

×

187

    if (code != 0) {

×

188

      terrno = code;

×

189

      dGError("vgId:%d, msg:%p, failed to process stream chkpt msg %s since %s", pVnode->vgId, pMsg,

×

191

      vmSendRsp(pMsg, code);

×

194

    dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

195

    rpcFreeCont(pMsg->pCont);

×

196

    taosFreeQitem(pMsg);

×

198

×

200

static void vmProcessStreamLongExecQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {

×

201

  SVnodeObj      *pVnode = pInfo->ahandle;

×

202

  const STraceId *trace = &pMsg->info.traceId;

×

203

  int32_t         code = 0;

×

205

  dGTrace("vgId:%d, msg:%p, get from vnode-stream long-exec queue", pVnode->vgId, pMsg);

×

207

  code = vnodeProcessStreamLongExecMsg(pVnode->pImpl, pMsg, pInfo);

×

208

  if (code != 0) {

×

209

    terrno = code;

×

210

    dGError("vgId:%d, msg:%p, failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),

×

212

    vmSendRsp(pMsg, code);

×

215

  dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

216

  rpcFreeCont(pMsg->pCont);

×

217

  taosFreeQitem(pMsg);

×

218

×

220

static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {

×

221

  SVnodeObj *pVnode = pInfo->ahandle;

×

222

  SRpcMsg   *pMsg = NULL;

×

224

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

×

225

    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;

×

226

    const STraceId *trace = &pMsg->info.traceId;

×

227

    dGTrace("vgId:%d, msg:%p, get from vnode-fetch queue", pVnode->vgId, pMsg);

×

229

    terrno = 0;

×

230

    int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);

×

231

    if (code != 0) {

×

232

      if (code == -1 && terrno != 0) {

×

233

        code = terrno;

×

236

      if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

×

237

        dGDebug("vgId:%d, msg:%p, failed to fetch since %s [vnodeProcessFetchMsg]", pVnode->vgId, pMsg, terrstr());

×

239

        dGError("vgId:%d, msg:%p, failed to fetch since %s [vnodeProcessFetchMsg]", pVnode->vgId, pMsg, terrstr());

×

242

      vmSendRsp(pMsg, code);

×

245

    dGTrace("vgId:%d, msg:%p, is freed, code:0x%x [vnodeProcessFetchMsg]", pVnode->vgId, pMsg, code);

×

246

    rpcFreeCont(pMsg->pCont);

×

247

    taosFreeQitem(pMsg);

×

249

×

251

static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {

×

252

  SVnodeObj *pVnode = pInfo->ahandle;

×

253

  SRpcMsg   *pMsg = NULL;

×

255

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

×

256

    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;

×

257

    const STraceId *trace = &pMsg->info.traceId;

×

258

    dGTrace("vgId:%d, msg:%p, get from vnode-sync queue", pVnode->vgId, pMsg);

×

260

    int32_t code = vnodeProcessSyncMsg(pVnode->pImpl, pMsg, NULL);  // no response here

×

261

    dGTrace("vgId:%d, msg:%p, is freed, code:0x%x", pVnode->vgId, pMsg, code);

×

262

    rpcFreeCont(pMsg->pCont);

×

263

    taosFreeQitem(pMsg);

×

265

×

267

static void vmSendResponse(SRpcMsg *pMsg) {

×

268

  if (pMsg->info.handle) {

×

269

    SRpcMsg rsp = {.info = pMsg->info, .code = terrno};

×

270

    if (rpcSendResponse(&rsp) != 0) {

×

271

      dError("failed to send response since %s", terrstr());

×

274

×

276

static bool vmDataSpaceSufficient(SVnodeObj *pVnode) {

×

277

  STfs *pTfs = pVnode->pImpl->pTfs;

×

278

  if (pTfs) {

×

279

    return tfsDiskSpaceSufficient(pTfs, 0, pVnode->diskPrimary);

×

281

    return osDataSpaceSufficient();

×

285

static int32_t vmAcquireVnodeWrapper(SVnodeMgmt *pMgt, int32_t vgId, SVnodeObj **pNode) {

×

286

  *pNode = vmAcquireVnode(pMgt, vgId);

×

287

  if (*pNode == NULL) {

×

288

    return terrno;

×

290

  return 0;

×

292

static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {

×

293

  int32_t         code = 0;

×

294

  const STraceId *trace = &pMsg->info.traceId;

×

295

  if (pMsg->contLen < sizeof(SMsgHead)) {

×

296

    dGError("invalid rpc msg with no msg head at pCont. pMsg:%p, type:%s, contLen:%d", pMsg, TMSG_INFO(pMsg->msgType),

×

298

    return TSDB_CODE_INVALID_MSG;

×

301

  SMsgHead *pHead = pMsg->pCont;

×

303

  pHead->contLen = ntohl(pHead->contLen);

×

304

  pHead->vgId = ntohl(pHead->vgId);

×

306

  SVnodeObj *pVnode = NULL;

×

307

  code = vmAcquireVnodeWrapper(pMgmt, pHead->vgId, &pVnode);

×

308

  if (code != 0) {

×

309

    dGDebug("vgId:%d, msg:%p, failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,

×

311

    return code;

×

314

  switch (qtype) {

×

315

    case QUERY_QUEUE:

×

316

      code = vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);

×

317

      if (code) {

×

318

        dError("vgId:%d, msg:%p, preprocess query msg failed since %s", pVnode->vgId, pMsg, tstrerror(code));

×

320

        dGTrace("vgId:%d, msg:%p, put into vnode-query queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

321

        code = taosWriteQitem(pVnode->pQueryQ, pMsg);

×

323

      break;

×

324

    case STREAM_QUEUE:

×

325

      dGTrace("vgId:%d, msg:%p, put into vnode-stream queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

326

      code = taosWriteQitem(pVnode->pStreamQ, pMsg);

×

327

      break;

×

328

    case STREAM_CTRL_QUEUE:

×

329

      dGTrace("vgId:%d, msg:%p, put into vnode-stream-ctrl queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

330

      code = taosWriteQitem(pVnode->pStreamCtrlQ, pMsg);

×

331

      break;

×

332

    case STREAM_LONG_EXEC_QUEUE:

×

333

      dGTrace("vgId:%d, msg:%p, put into vnode-stream-long-exec queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

334

      code = taosWriteQitem(pVnode->pStreamLongExecQ, pMsg);

×

335

      break;

×

336

    case STREAM_CHKPT_QUEUE:

×

337

      dGTrace("vgId:%d, msg:%p, put into vnode-stream-chkpt queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

338

      code = taosWriteQitem(pVnode->pStreamChkQ, pMsg);

×

339

      break;

×

340

    case FETCH_QUEUE:

×

341

      dGTrace("vgId:%d, msg:%p, put into vnode-fetch queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

342

      code = taosWriteQitem(pVnode->pFetchQ, pMsg);

×

343

      break;

×

344

    case WRITE_QUEUE:

×

345

      if (!vmDataSpaceSufficient(pVnode)) {

×

346

        code = TSDB_CODE_NO_ENOUGH_DISKSPACE;

×

347

        dError("vgId:%d, msg:%p, failed to put into vnode-write queue since %s, type:%s", pVnode->vgId, pMsg,

×

349

        break;

×

351

      if (pMsg->msgType == TDMT_VND_SUBMIT && (grantCheck(TSDB_GRANT_STORAGE) != TSDB_CODE_SUCCESS)) {

×

352

        code = TSDB_CODE_VND_NO_WRITE_AUTH;

×

353

        dDebug("vgId:%d, msg:%p, failed to put into vnode-write queue since %s, type:%s", pVnode->vgId, pMsg,

×

355

        break;

×

357

      if (pMsg->msgType != TDMT_VND_ALTER_CONFIRM && pVnode->disable) {

×

358

        dDebug("vgId:%d, msg:%p, failed to put into vnode-write queue since its disable, type:%s", pVnode->vgId, pMsg,

×

360

        code = TSDB_CODE_VND_STOPPED;

×

361

        break;

×

363

      dGDebug("vgId:%d, msg:%p, put into vnode-write queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

364

      code = taosWriteQitem(pVnode->pWriteW.queue, pMsg);

×

365

      break;

×

366

    case SYNC_QUEUE:

×

367

      dGDebug("vgId:%d, msg:%p, put into vnode-sync queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

368

      code = taosWriteQitem(pVnode->pSyncW.queue, pMsg);

×

369

      break;

×

370

    case SYNC_RD_QUEUE:

×

371

      dGDebug("vgId:%d, msg:%p, put into vnode-sync-rd queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

372

      code = taosWriteQitem(pVnode->pSyncRdW.queue, pMsg);

×

373

      break;

×

374

    case APPLY_QUEUE:

×

375

      dGDebug("vgId:%d, msg:%p, put into vnode-apply queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));

×

376

      code = taosWriteQitem(pVnode->pApplyW.queue, pMsg);

×

377

      break;

×

378

    default:

×

379

      code = TSDB_CODE_INVALID_MSG;

×

380

      break;

×

383

  vmReleaseVnode(pMgmt, pVnode);

×

384

  return code;

×

387

int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_RD_QUEUE); }

×

389

int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }

×

391

int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); }

×

393

int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); }

×

395

int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }

×

397

int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_QUEUE); }

×

399

int32_t vmPutMsgToStreamCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_CTRL_QUEUE); }

×

401

int32_t vmPutMsgToStreamLongExecQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_LONG_EXEC_QUEUE); }

×

403

int32_t vmPutMsgToStreamChkQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_CHKPT_QUEUE); }

×

405

int32_t vmPutMsgToMultiMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {

×

406

  const STraceId *trace = &pMsg->info.traceId;

×

407

  dGTrace("msg:%p, put into vnode-multi-mgmt queue", pMsg);

×

408

  return taosWriteQitem(pMgmt->mgmtMultiWorker.queue, pMsg);

×

411

int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {

×

412

  const STraceId *trace = &pMsg->info.traceId;

×

413

  dGTrace("msg:%p, put into vnode-mgmt queue", pMsg);

×

414

  return taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg);

×

417

int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {

×

419

  if (pRpc->contLen < sizeof(SMsgHead)) {

×

420

    dError("invalid rpc msg with no msg head at pCont. pRpc:%p, type:%s, len:%d", pRpc, TMSG_INFO(pRpc->msgType),

×

422

    rpcFreeCont(pRpc->pCont);

×

423

    pRpc->pCont = NULL;

×

424

    return TSDB_CODE_INVALID_MSG;

×

427

  EQItype  itype = APPLY_QUEUE == qtype ? APPLY_QITEM : RPC_QITEM;

×

429

  code = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen, (void **)&pMsg);

×

430

  if (code) {

×

431

    rpcFreeCont(pRpc->pCont);

×

432

    pRpc->pCont = NULL;

×

433

    return code;

×

436

  SMsgHead *pHead = pRpc->pCont;

×

437

  dTrace("vgId:%d, msg:%p, is created, type:%s len:%d", pHead->vgId, pMsg, TMSG_INFO(pRpc->msgType), pRpc->contLen);

×

439

  pHead->contLen = htonl(pHead->contLen);

×

440

  pHead->vgId = htonl(pHead->vgId);

×

441

  memcpy(pMsg, pRpc, sizeof(SRpcMsg));

×

442

  pRpc->pCont = NULL;

×

444

  code = vmPutMsgToQueue(pMgmt, pMsg, qtype);

×

445

  if (code != 0) {

×

446

    dTrace("msg:%p, is freed", pMsg);

×

447

    rpcFreeCont(pMsg->pCont);

×

448

    taosFreeQitem(pMsg);

×

451

  return code;

×

454

int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {

×

455

  int32_t    size = -1;

×

456

  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);

×

457

  if (pVnode != NULL) {

×

458

    switch (qtype) {

×

459

      case WRITE_QUEUE:

×

460

        size = taosQueueItemSize(pVnode->pWriteW.queue);

×

461

        break;

×

462

      case SYNC_QUEUE:

×

463

        size = taosQueueItemSize(pVnode->pSyncW.queue);

×

464

        break;

×

465

      case APPLY_QUEUE:

×

466

        size = taosQueueItemSize(pVnode->pApplyW.queue);

×

467

        break;

×

468

      case QUERY_QUEUE:

×

469

        size = taosQueueItemSize(pVnode->pQueryQ);

×

470

        break;

×

471

      case FETCH_QUEUE:

×

472

        size = taosQueueItemSize(pVnode->pFetchQ);

×

473

        break;

×

474

      case STREAM_QUEUE:

×

475

        size = taosQueueItemSize(pVnode->pStreamQ);

×

476

        break;

×

477

      case STREAM_CTRL_QUEUE:

×

478

        size = taosQueueItemSize(pVnode->pStreamCtrlQ);

×

479

        break;

×

480

      case STREAM_LONG_EXEC_QUEUE:

×

481

        size = taosQueueItemSize(pVnode->pStreamLongExecQ);

×

482

        break;

×

483

      case STREAM_CHKPT_QUEUE:

×

484

        size = taosQueueItemSize(pVnode->pStreamChkQ);

×

485

      default:

×

486

        break;

×

489

  if (pVnode) vmReleaseVnode(pMgmt, pVnode);

×

490

  if (size < 0) {

×

491

    dTrace("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype);

×

492

    size = 0;

×

494

  return size;

×

497

int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {

×

498

  int32_t         code = 0;

×

499

  SMultiWorkerCfg wcfg = {.max = 1, .name = "vnode-write", .fp = (FItems)vnodeProposeWriteMsg, .param = pVnode->pImpl};

×

500

  SMultiWorkerCfg scfg = {.max = 1, .name = "vnode-sync", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};

×

501

  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-rd", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};

×

502

  SMultiWorkerCfg acfg = {.max = 1, .name = "vnode-apply", .fp = (FItems)vnodeApplyWriteMsg, .param = pVnode->pImpl};

×

503

  code = tMultiWorkerInit(&pVnode->pWriteW, &wcfg);

×

504

  if (code) {

×

505

    return code;

×

507

  code = tMultiWorkerInit(&pVnode->pSyncW, &scfg);

×

508

  if (code) {

×

509

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

510

    return code;

×

512

  code = tMultiWorkerInit(&pVnode->pSyncRdW, &sccfg);

×

513

  if (code) {

×

514

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

515

    tMultiWorkerCleanup(&pVnode->pSyncW);

×

516

    return code;

×

518

  code = tMultiWorkerInit(&pVnode->pApplyW, &acfg);

×

519

  if (code) {

×

520

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

521

    tMultiWorkerCleanup(&pVnode->pSyncW);

×

522

    tMultiWorkerCleanup(&pVnode->pSyncRdW);

×

523

    return code;

×

526

  pVnode->pQueryQ = tQueryAutoQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);

×

527

  pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);

×

530

  pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue, 2);

×

531

  pVnode->pStreamCtrlQ = tWWorkerAllocQueue(&pMgmt->streamCtrlPool, pVnode, (FItems)vmProcessStreamCtrlQueue);

×

532

  pVnode->pStreamLongExecQ = tAutoQWorkerAllocQueue(&pMgmt->streamLongExecPool, pVnode, (FItem)vmProcessStreamLongExecQueue, 1);

×

533

  pVnode->pStreamChkQ = tWWorkerAllocQueue(&pMgmt->streamChkPool, pVnode, (FItems)vmProcessStreamChkptQueue);

×

535

  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncRdW.queue == NULL ||

×

536

      pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL

×

537

      || pVnode->pStreamCtrlQ == NULL || pVnode->pStreamLongExecQ == NULL || pVnode->pStreamChkQ == NULL) {

×

538

    return TSDB_CODE_OUT_OF_MEMORY;

×

541

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

×

543

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

×

545

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

×

547

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

×

549

  dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);

×

550

  dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,

×

552

  dInfo("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ);

×

553

  dInfo("vgId:%d, stream-long-exec-queue:%p is alloced", pVnode->vgId, pVnode->pStreamLongExecQ);

×

554

  dInfo("vgId:%d, stream-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pStreamCtrlQ,

×

556

  dInfo("vgId:%d, stream-chk-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pStreamChkQ,

×

558

  return 0;

×

561

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

×

562

  tQueryAutoQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);

×

563

  tAutoQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ);

×

564

  tAutoQWorkerFreeQueue(&pMgmt->streamLongExecPool, pVnode->pStreamLongExecQ);

×

565

  tWWorkerFreeQueue(&pMgmt->streamCtrlPool, pVnode->pStreamCtrlQ);

×

566

  tWWorkerFreeQueue(&pMgmt->streamChkPool, pVnode->pStreamChkQ);

×

567

  tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);

×

568

  pVnode->pQueryQ = NULL;

×

569

  pVnode->pFetchQ = NULL;

×

571

  pVnode->pStreamQ = NULL;

×

572

  pVnode->pStreamCtrlQ = NULL;

×

573

  pVnode->pStreamLongExecQ = NULL;

×

575

  pVnode->pStreamChkQ = NULL;

×

576

  pVnode->pFetchQ = NULL;

×

577

  dDebug("vgId:%d, queue is freed", pVnode->vgId);

×

578

×

580

int32_t vmStartWorker(SVnodeMgmt *pMgmt) {

×

581

  int32_t code = 0;

×

583

  SQueryAutoQWorkerPool *pQPool = &pMgmt->queryPool;

×

584

  pQPool->name = "vnode-query";

×

585

  pQPool->min = tsNumOfVnodeQueryThreads;

×

586

  pQPool->max = tsNumOfVnodeQueryThreads;

×

587

  if ((code = tQueryAutoQWorkerInit(pQPool)) != 0) return code;

×

589

  tsNumOfQueryThreads += tsNumOfVnodeQueryThreads;

×

591

  SAutoQWorkerPool *pStreamPool = &pMgmt->streamPool;

×

592

  pStreamPool->name = "vnode-stream";

×

593

  pStreamPool->ratio = tsRatioOfVnodeStreamThreads;

×

594

  if ((code = tAutoQWorkerInit(pStreamPool)) != 0) return code;

×

596

  SAutoQWorkerPool *pLongExecPool = &pMgmt->streamLongExecPool;

×

597

  pLongExecPool->name = "vnode-stream-long-exec";

×

598

  pLongExecPool->ratio = tsRatioOfVnodeStreamThreads/3;

×

599

  if ((code = tAutoQWorkerInit(pLongExecPool)) != 0) return code;

×

601

  SWWorkerPool *pStreamCtrlPool = &pMgmt->streamCtrlPool;

×

602

  pStreamCtrlPool->name = "vnode-stream-ctrl";

×

603

  pStreamCtrlPool->max = 4;

×

604

  if ((code = tWWorkerInit(pStreamCtrlPool)) != 0) return code;

×

606

  SWWorkerPool *pStreamChkPool = &pMgmt->streamChkPool;

×

607

  pStreamChkPool->name = "vnode-stream-chkpt";

×

608

  pStreamChkPool->max = 1;

×

609

  if ((code = tWWorkerInit(pStreamChkPool)) != 0) return code;

×

611

  SWWorkerPool *pFPool = &pMgmt->fetchPool;

×

612

  pFPool->name = "vnode-fetch";

×

613

  pFPool->max = tsNumOfVnodeFetchThreads;

×

614

  if ((code = tWWorkerInit(pFPool)) != 0) return code;

×

616

  SSingleWorkerCfg mgmtCfg = {

×

619

  if ((code = tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg)) != 0) return code;

×

621

  int32_t threadNum = 0;

×

622

  if (tsNumOfCores == 1) {

×

623

    threadNum = 2;

×

625

    threadNum = tsNumOfCores;

×

627

  SSingleWorkerCfg multiMgmtCfg = {.min = threadNum,

×

633

  if ((code = tSingleWorkerInit(&pMgmt->mgmtMultiWorker, &multiMgmtCfg)) != 0) return code;

×

635

  dDebug("vnode workers are initialized");

×

636

  return 0;

×

639

void vmStopWorker(SVnodeMgmt *pMgmt) {

×

640

  tQueryAutoQWorkerCleanup(&pMgmt->queryPool);

×

641

  tAutoQWorkerCleanup(&pMgmt->streamPool);

×

642

  tAutoQWorkerCleanup(&pMgmt->streamLongExecPool);

×

643

  tWWorkerCleanup(&pMgmt->streamCtrlPool);

×

644

  tWWorkerCleanup(&pMgmt->streamChkPool);

×

645

  tWWorkerCleanup(&pMgmt->fetchPool);

×

646

  dDebug("vnode workers are closed");

×

647

×

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