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

×

123

  SVnodeObj      *pVnode = pInfo->ahandle;

×

124

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

×

126

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

×

127

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

×

128

  if (code != 0) {

×

129

    terrno = code;

×

130

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

×

132

    vmSendRsp(pMsg, code);

×

135

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

×

136

  rpcFreeCont(pMsg->pCont);

×

137

  taosFreeQitem(pMsg);

×

138

×

140

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

×

141

  SVnodeObj *pVnode = pInfo->ahandle;

×

142

  void      *pItem = NULL;

×

143

  int32_t    code = 0;

×

145

  while (1) {

×

146

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

×

147

      break;

×

150

    SRpcMsg        *pMsg = pItem;

×

151

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

×

153

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

×

154

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

×

155

    if (code != 0) {

×

156

      terrno = code;

×

157

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

×

159

      vmSendRsp(pMsg, code);

×

162

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

×

163

    rpcFreeCont(pMsg->pCont);

×

164

    taosFreeQitem(pMsg);

×

166

×

168

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

×

169

  SVnodeObj *pVnode = pInfo->ahandle;

×

170

  void      *pItem = NULL;

×

171

  int32_t    code = 0;

×

173

  while (1) {

×

174

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

×

175

      break;

×

178

    SRpcMsg        *pMsg = pItem;

×

179

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

×

181

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

×

182

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

×

183

    if (code != 0) {

×

184

      terrno = code;

×

185

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

×

187

      vmSendRsp(pMsg, code);

×

190

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

×

191

    rpcFreeCont(pMsg->pCont);

×

192

    taosFreeQitem(pMsg);

×

194

×

196

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

×

197

  SVnodeObj      *pVnode = pInfo->ahandle;

×

198

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

×

199

  int32_t         code = 0;

×

201

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

×

203

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

×

204

  if (code != 0) {

×

205

    terrno = code;

×

206

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

×

208

    vmSendRsp(pMsg, code);

×

211

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

×

212

  rpcFreeCont(pMsg->pCont);

×

213

  taosFreeQitem(pMsg);

×

214

×

216

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

×

217

  SVnodeObj *pVnode = pInfo->ahandle;

×

218

  SRpcMsg   *pMsg = NULL;

×

220

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

×

221

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

×

222

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

×

223

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

×

225

    terrno = 0;

×

226

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

×

227

    if (code != 0) {

×

228

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

×

229

        code = terrno;

×

232

      if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

×

233

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

×

235

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

×

238

      vmSendRsp(pMsg, code);

×

241

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

×

242

    rpcFreeCont(pMsg->pCont);

×

243

    taosFreeQitem(pMsg);

×

245

×

247

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

×

248

  SVnodeObj *pVnode = pInfo->ahandle;

×

249

  SRpcMsg   *pMsg = NULL;

×

251

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

×

252

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

×

253

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

×

254

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

×

256

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

×

257

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

×

258

    rpcFreeCont(pMsg->pCont);

×

259

    taosFreeQitem(pMsg);

×

261

×

263

static void vmSendResponse(SRpcMsg *pMsg) {

×

264

  if (pMsg->info.handle) {

×

265

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

×

266

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

×

267

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

×

270

×

272

static bool vmDataSpaceSufficient(SVnodeObj *pVnode) {

×

273

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

×

274

  if (pTfs) {

×

275

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

×

277

    return osDataSpaceSufficient();

×

281

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

×

282

  *pNode = vmAcquireVnode(pMgt, vgId);

×

283

  if (*pNode == NULL) {

×

284

    return terrno;

×

286

  return 0;

×

288

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

×

289

  int32_t         code = 0;

×

290

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

×

291

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

×

292

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

×

294

    return TSDB_CODE_INVALID_MSG;

×

297

  SMsgHead *pHead = pMsg->pCont;

×

299

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

×

300

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

×

302

  SVnodeObj *pVnode = NULL;

×

303

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

×

304

  if (code != 0) {

×

305

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

×

307

    return code;

×

310

  switch (qtype) {

×

311

    case QUERY_QUEUE:

×

312

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

×

313

      if (code) {

×

314

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

×

316

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

×

317

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

×

319

      break;

×

320

    case STREAM_QUEUE:

×

321

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

×

322

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

×

323

      break;

×

324

    case STREAM_CTRL_QUEUE:

×

325

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

×

326

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

×

327

      break;

×

328

    case STREAM_LONG_EXEC_QUEUE:

×

329

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

×

330

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

×

331

      break;

×

332

    case STREAM_CHKPT_QUEUE:

×

333

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

×

334

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

×

335

      break;

×

336

    case FETCH_QUEUE:

×

337

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

×

338

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

×

339

      break;

×

340

    case WRITE_QUEUE:

×

341

      if (!vmDataSpaceSufficient(pVnode)) {

×

342

        code = TSDB_CODE_NO_ENOUGH_DISKSPACE;

×

343

        dError("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, tstrerror(code));

×

344

        break;

×

346

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

×

347

        code = TSDB_CODE_VND_NO_WRITE_AUTH;

×

348

        dDebug("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, tstrerror(code));

×

349

        break;

×

351

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

×

352

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

×

353

        code = TSDB_CODE_VND_STOPPED;

×

354

        break;

×

356

      dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);

×

357

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

×

358

      break;

×

359

    case SYNC_QUEUE:

×

360

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

×

361

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

×

362

      break;

×

363

    case SYNC_RD_QUEUE:

×

364

      dGTrace("vgId:%d, msg:%p put into vnode-sync-rd queue", pVnode->vgId, pMsg);

×

365

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

×

366

      break;

×

367

    case APPLY_QUEUE:

×

368

      dGTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);

×

369

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

×

370

      break;

×

371

    default:

×

372

      code = TSDB_CODE_INVALID_MSG;

×

373

      break;

×

376

  vmReleaseVnode(pMgmt, pVnode);

×

377

  return code;

×

380

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

×

382

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

×

384

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

×

386

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

×

388

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

×

390

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

×

392

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

×

394

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

×

396

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

×

398

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

×

399

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

×

400

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

×

401

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

×

404

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

×

405

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

×

406

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

×

407

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

×

410

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

×

412

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

×

413

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

×

415

    rpcFreeCont(pRpc->pCont);

×

416

    pRpc->pCont = NULL;

×

417

    return TSDB_CODE_INVALID_MSG;

×

420

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

×

422

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

×

423

  if (code) {

×

424

    rpcFreeCont(pRpc->pCont);

×

425

    pRpc->pCont = NULL;

×

426

    return code;

×

429

  SMsgHead *pHead = pRpc->pCont;

×

430

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

×

432

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

×

433

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

×

434

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

×

435

  pRpc->pCont = NULL;

×

437

  code = vmPutMsgToQueue(pMgmt, pMsg, qtype);

×

438

  if (code != 0) {

×

439

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

×

440

    rpcFreeCont(pMsg->pCont);

×

441

    taosFreeQitem(pMsg);

×

444

  return code;

×

447

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

×

448

  int32_t    size = -1;

×

449

  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);

×

450

  if (pVnode != NULL) {

×

451

    switch (qtype) {

×

452

      case WRITE_QUEUE:

×

453

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

×

454

        break;

×

455

      case SYNC_QUEUE:

×

456

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

×

457

        break;

×

458

      case APPLY_QUEUE:

×

459

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

×

460

        break;

×

461

      case QUERY_QUEUE:

×

462

        size = taosQueueItemSize(pVnode->pQueryQ);

×

463

        break;

×

464

      case FETCH_QUEUE:

×

465

        size = taosQueueItemSize(pVnode->pFetchQ);

×

466

        break;

×

467

      case STREAM_QUEUE:

×

468

        size = taosQueueItemSize(pVnode->pStreamQ);

×

469

        break;

×

470

      case STREAM_CTRL_QUEUE:

×

471

        size = taosQueueItemSize(pVnode->pStreamCtrlQ);

×

472

        break;

×

473

      case STREAM_LONG_EXEC_QUEUE:

×

474

        size = taosQueueItemSize(pVnode->pStreamLongExecQ);

×

475

        break;

×

476

      case STREAM_CHKPT_QUEUE:

×

477

        size = taosQueueItemSize(pVnode->pStreamChkQ);

×

478

      default:

×

479

        break;

×

482

  if (pVnode) vmReleaseVnode(pMgmt, pVnode);

×

483

  if (size < 0) {

×

484

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

×

485

    size = 0;

×

487

  return size;

×

490

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

×

491

  int32_t         code = 0;

×

492

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

×

493

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

×

494

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

×

495

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

×

496

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

×

497

  if (code) {

×

498

    return code;

×

500

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

×

501

  if (code) {

×

502

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

503

    return code;

×

505

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

×

506

  if (code) {

×

507

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

508

    tMultiWorkerCleanup(&pVnode->pSyncW);

×

509

    return code;

×

511

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

×

512

  if (code) {

×

513

    tMultiWorkerCleanup(&pVnode->pWriteW);

×

514

    tMultiWorkerCleanup(&pVnode->pSyncW);

×

515

    tMultiWorkerCleanup(&pVnode->pSyncRdW);

×

516

    return code;

×

519

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

×

520

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

×

523

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

×

524

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

×

525

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

×

526

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

×

528

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

×

529

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

×

530

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

×

531

    return TSDB_CODE_OUT_OF_MEMORY;

×

534

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

×

536

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

×

538

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

×

540

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

×

542

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

×

543

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

×

545

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

×

546

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

×

547

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

×

549

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

×

551

  return 0;

×

554

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

×

555

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

×

556

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

×

557

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

×

558

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

×

559

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

×

560

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

×

561

  pVnode->pQueryQ = NULL;

×

562

  pVnode->pFetchQ = NULL;

×

564

  pVnode->pStreamQ = NULL;

×

565

  pVnode->pStreamCtrlQ = NULL;

×

566

  pVnode->pStreamLongExecQ = NULL;

×

568

  pVnode->pStreamChkQ = NULL;

×

569

  pVnode->pFetchQ = NULL;

×

570

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

×

571

×

573

int32_t vmStartWorker(SVnodeMgmt *pMgmt) {

×

574

  int32_t code = 0;

×

576

  SQueryAutoQWorkerPool *pQPool = &pMgmt->queryPool;

×

577

  pQPool->name = "vnode-query";

×

578

  pQPool->min = tsNumOfVnodeQueryThreads;

×

579

  pQPool->max = tsNumOfVnodeQueryThreads;

×

580

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

×

582

  tsNumOfQueryThreads += tsNumOfVnodeQueryThreads;

×

584

  SAutoQWorkerPool *pStreamPool = &pMgmt->streamPool;

×

585

  pStreamPool->name = "vnode-stream";

×

586

  pStreamPool->ratio = tsRatioOfVnodeStreamThreads;

×

587

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

×

589

  SAutoQWorkerPool *pLongExecPool = &pMgmt->streamLongExecPool;

×

590

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

×

591

  pLongExecPool->ratio = tsRatioOfVnodeStreamThreads/3;

×

592

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

×

594

  SWWorkerPool *pStreamCtrlPool = &pMgmt->streamCtrlPool;

×

595

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

×

596

  pStreamCtrlPool->max = 1;

×

597

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

×

599

  SWWorkerPool *pStreamChkPool = &pMgmt->streamChkPool;

×

600

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

×

601

  pStreamChkPool->max = 1;

×

602

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

×

604

  SWWorkerPool *pFPool = &pMgmt->fetchPool;

×

605

  pFPool->name = "vnode-fetch";

×

606

  pFPool->max = tsNumOfVnodeFetchThreads;

×

607

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

×

609

  SSingleWorkerCfg mgmtCfg = {

×

612

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

×

614

  int32_t threadNum = 0;

×

615

  if (tsNumOfCores == 1) {

×

616

    threadNum = 2;

×

618

    threadNum = tsNumOfCores;

×

620

  SSingleWorkerCfg multiMgmtCfg = {.min = threadNum,

×

626

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

×

628

  dDebug("vnode workers are initialized");

×

629

  return 0;

×

632

void vmStopWorker(SVnodeMgmt *pMgmt) {

×

633

  tQueryAutoQWorkerCleanup(&pMgmt->queryPool);

×

634

  tAutoQWorkerCleanup(&pMgmt->streamPool);

×

635

  tAutoQWorkerCleanup(&pMgmt->streamLongExecPool);

×

636

  tWWorkerCleanup(&pMgmt->streamCtrlPool);

×

637

  tWWorkerCleanup(&pMgmt->streamChkPool);

×

638

  tWWorkerCleanup(&pMgmt->fetchPool);

×

639

  dDebug("vnode workers are closed");

×

640

×

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