taosdata/TDengine | Build 4452 | source/dnode/mnode/sdb/src/sdbHash.c | Coveralls

21

const char *sdbTableName(ESdbType type) {

4,773,930✔

23

    case SDB_TRANS:

289,128✔

24

      return "trans";

289,128✔

25

    case SDB_CLUSTER:

34,818✔

26

      return "cluster";

34,818✔

27

    case SDB_MNODE:

37,237✔

28

      return "mnode";

37,237✔

29

    case SDB_QNODE:

42,973✔

30

      return "qnode";

42,973✔

31

    case SDB_SNODE:

10,616✔

32

      return "snode";

10,616✔

33

    case SDB_DNODE:

434,473✔

34

      return "dnode";

434,473✔

35

    case SDB_USER:

331,355✔

36

      return "user";

331,355✔

37

    case SDB_AUTH:

4✔

38

      return "auth";

4✔

39

    case SDB_ACCT:

19,752✔

40

      return "acct";

19,752✔

41

    case SDB_STREAM_CK:

4✔

42

      return "stream_ck";

4✔

43

    case SDB_STREAM:

26,761✔

44

      return "stream";

26,761✔

45

    case SDB_OFFSET:

4✔

46

      return "offset";

4✔

47

    case SDB_SUBSCRIBE:

14,626✔

48

      return "subscribe";

14,626✔

49

    case SDB_CONSUMER:

18,754✔

50

      return "consumer";

18,754✔

51

    case SDB_TOPIC:

13,187✔

52

      return "topic";

13,187✔

53

    case SDB_VGROUP:

2,054,559✔

54

      return "vgroup";

2,054,559✔

55

    case SDB_SMA:

18,755✔

56

      return "sma";

18,755✔

57

    case SDB_STB:

161,954✔

58

      return "stb";

161,954✔

59

    case SDB_DB:

370,186✔

60

      return "db";

370,186✔

61

    case SDB_FUNC:

13,557✔

62

      return "func";

13,557✔

63

    case SDB_IDX:

70,055✔

64

      return "idx";

70,055✔

65

    case SDB_VIEW:

13,364✔

66

      return "view";

13,364✔

67

    case SDB_STREAM_SEQ:

10,388✔

68

      return "stream_seq";

10,388✔

69

    case SDB_COMPACT:

10,457✔

70

      return "compact";

10,457✔

71

    case SDB_COMPACT_DETAIL:

10,691✔

72

      return "compact_detail";

10,691✔

73

    case SDB_GRANT:

23,830✔

74

      return "grant";

23,830✔

75

    case SDB_ARBGROUP:

10,564✔

76

      return "arb_group";

10,564✔

77

    case SDB_ANODE:

10,401✔

78

      return "anode";

10,401✔

79

    case SDB_BNODE:

11,442✔

80

      return "bnode";

11,442✔

81

    case SDB_CFG:

710,262✔

82

      return "config";

710,262✔

88

const char *sdbStatusName(ESdbStatus status) {

3,831,140✔

90

    case SDB_STATUS_CREATING:

36,421✔

91

      return "creating";

36,421✔

92

    case SDB_STATUS_DROPPING:

9,493✔

93

      return "dropping";

9,493✔

94

    case SDB_STATUS_READY:

3,709,639✔

95

      return "ready";

3,709,639✔

96

    case SDB_STATUS_DROPPED:

66,503✔

97

      return "dropped";

66,503✔

98

    case SDB_STATUS_INIT:

9,078✔

99

      return "init";

9,078✔

102

    default:

6✔

103

      return "undefine";

6✔

107

void sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper) {

48,874,100✔

109

  EKeyType keyType = pSdb->keyTypes[pRow->type];

48,874,100✔

111

  if (keyType == SDB_KEY_BINARY) {

48,874,100✔

112

    mTrace("%s:%s, ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type), (char *)pRow->pObj,

7,755,791✔

114

  } else if (keyType == SDB_KEY_INT32) {

41,118,309✔

115

    mTrace("%s:%d, ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type), *(int32_t *)pRow->pObj,

40,804,367✔

118

    mTrace("%s:%" PRId64 ", ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type),

320,482✔

123

48,874,262✔

125

static SHashObj *sdbGetHash(SSdb *pSdb, int32_t type) {

27,307,745✔

128

    return NULL;

1✔

131

  SHashObj *hash = pSdb->hashObjs[type];

27,307,909✔

132

  if (hash == NULL) {

27,307,909✔

133

    terrno = TSDB_CODE_APP_ERROR;

4✔

134

    return NULL;

4✔

137

  return hash;

27,307,905✔

140

static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) {

5,819,980✔

141

  int32_t  keySize = 0;

5,819,980✔

142

  EKeyType keyType = pSdb->keyTypes[type];

5,819,980✔

144

  if (keyType == SDB_KEY_INT32) {

5,819,980✔

145

    keySize = sizeof(int32_t);

3,345,746✔

146

  } else if (keyType == SDB_KEY_BINARY) {

2,474,234✔

147

    keySize = strlen(pKey) + 1;

2,450,894✔

149

    keySize = sizeof(int64_t);

23,340✔

152

  return keySize;

5,819,980✔

155

static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {

369,908✔

156

  int32_t type = pRow->type;

369,908✔

157

  sdbWriteLock(pSdb, type);

369,908✔

159

  SSdbRow *pOldRow = taosHashGet(hash, pRow->pObj, keySize);

369,908✔

167

  pRow->refCount = 0;

369,908✔

168

  pRow->status = pRaw->status;

369,908✔

169

  sdbPrintOper(pSdb, pRow, "insert");

369,908✔

171

  int32_t code = 0;

369,908✔

178

  SdbInsertFp insertFp = pSdb->insertFps[pRow->type];

369,908✔

180

    code = (*insertFp)(pSdb, pRow->pObj);

369,908✔

192

  sdbUnLock(pSdb, type);

369,908✔

194

  if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT32) {

369,908✔

195

    pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], *((int32_t *)pRow->pObj));

86,205✔

197

  if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT64) {

369,908✔

198

    pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], *((int64_t *)pRow->pObj));

3,204✔

200

  pSdb->tableVer[pRow->type]++;

369,908✔

202

  return 0;

369,908✔

205

static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) {

485,821✔

206

  int32_t type = pNewRow->type;

485,821✔

207

  sdbWriteLock(pSdb, type);

485,821✔

209

  SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize);

485,821✔

211

    sdbUnLock(pSdb, type);

351,570✔

212

    return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize);

351,570✔

215

  SSdbRow *pOldRow = *ppOldRow;

134,251✔

216

  pOldRow->status = pRaw->status;

134,251✔

217

  sdbPrintOper(pSdb, pOldRow, "update");

134,251✔

219

  int32_t     code = 0;

134,251✔

220

  SdbUpdateFp updateFp = pSdb->updateFps[type];

134,251✔

222

    code = (*updateFp)(pSdb, pOldRow->pObj, pNewRow->pObj);

134,251✔

224

  sdbUnLock(pSdb, type);

134,251✔

227

  sdbFreeRow(pSdb, pNewRow, false);

134,251✔

229

  pSdb->tableVer[pOldRow->type]++;

134,251✔

230

  return code;

134,251✔

233

static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {

81,979✔

234

  int32_t type = pRow->type;

81,979✔

235

  sdbWriteLock(pSdb, type);

81,979✔

237

  SSdbRow **ppOldRow = taosHashGet(hash, pRow->pObj, keySize);

81,979✔

239

    sdbUnLock(pSdb, type);

39✔

240

    sdbFreeRow(pSdb, pRow, false);

39✔

241

    terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;

39✔

242

    return terrno;

39✔

244

  SSdbRow *pOldRow = *ppOldRow;

81,940✔

245

  pOldRow->status = pRaw->status;

81,940✔

247

  (void)atomic_add_fetch_32(&pOldRow->refCount, 1);

81,940✔

248

  sdbPrintOper(pSdb, pOldRow, "delete");

81,940✔

256

  pSdb->tableVer[pOldRow->type]++;

81,940✔

257

  sdbUnLock(pSdb, type);

81,940✔

259

  sdbFreeRow(pSdb, pRow, false);

81,940✔

261

  sdbCheckRow(pSdb, pOldRow);

81,940✔

262

  return 0;

81,940✔

265

int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw) {

586,140✔

266

  if (pRaw->type == SDB_CFG) {

586,140✔

267

    mTrace("sdb write cfg");

240,602✔

269

  SHashObj *hash = sdbGetHash(pSdb, pRaw->type);

586,140✔

270

  if (hash == NULL) return terrno;

586,140✔

272

  SdbDecodeFp decodeFp = pSdb->decodeFps[pRaw->type];

586,138✔

273

  SSdbRow    *pRow = (*decodeFp)(pRaw);

586,138✔

276

  pRow->type = pRaw->type;

586,138✔

278

  int32_t keySize = sdbGetkeySize(pSdb, pRow->type, pRow->pObj);

586,138✔

279

  int32_t code = TSDB_CODE_SDB_INVALID_ACTION_TYPE;

586,138✔

282

    case SDB_STATUS_CREATING:

18,338✔

283

      code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize);

18,338✔

284

      break;

18,338✔

285

    case SDB_STATUS_READY:

485,821✔

288

      code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize);

485,821✔

289

      break;

485,821✔

290

    case SDB_STATUS_DROPPED:

81,979✔

291

      code = sdbDeleteRow(pSdb, hash, pRaw, pRow, keySize);

81,979✔

292

      break;

81,979✔

295

  return code;

586,138✔

298

int32_t sdbWrite(SSdb *pSdb, SSdbRaw *pRaw) {

64,542✔

299

  int32_t code = sdbWriteWithoutFree(pSdb, pRaw);

64,542✔

300

  sdbFreeRaw(pRaw);

64,542✔

301

  return code;

64,542✔

304

void *sdbAcquireAll(SSdb *pSdb, ESdbType type, const void *pKey, bool onlyReady) {

5,234,080✔

305

  terrno = 0;

5,234,080✔

307

  SHashObj *hash = sdbGetHash(pSdb, type);

5,233,983✔

308

  if (hash == NULL) return NULL;

5,233,861✔

310

  void   *pRet = NULL;

5,233,860✔

311

  int32_t keySize = sdbGetkeySize(pSdb, type, pKey);

5,233,860✔

313

  sdbReadLock(pSdb, type);

5,233,809✔

315

  SSdbRow **ppRow = taosHashGet(hash, pKey, keySize);

5,234,192✔

316

  if (ppRow == NULL || *ppRow == NULL) {

5,234,191✔

317

    sdbUnLock(pSdb, type);

618,806✔

318

    terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;

618,856✔

319

    return NULL;

618,856✔

322

  SSdbRow *pRow = *ppRow;

4,615,385✔

324

    case SDB_STATUS_READY:

4,607,183✔

325

      (void)atomic_add_fetch_32(&pRow->refCount, 1);

4,607,183✔

326

      pRet = pRow->pObj;

4,607,228✔

327

      sdbPrintOper(pSdb, pRow, "acquire");

4,607,228✔

328

      break;

4,607,112✔

329

    case SDB_STATUS_CREATING:

5,768✔

330

      terrno = TSDB_CODE_SDB_OBJ_CREATING;

5,768✔

331

      break;

5,768✔

332

    case SDB_STATUS_DROPPING:

2,441✔

333

      terrno = TSDB_CODE_SDB_OBJ_DROPPING;

2,441✔

334

      break;

2,441✔

340

  if (pRet == NULL) {

4,615,321✔

349

  sdbUnLock(pSdb, type);

4,615,321✔

350

  return pRet;

4,615,333✔

353

void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) { return sdbAcquireAll(pSdb, type, pKey, true); }

5,230,844✔

354

void *sdbAcquireNotReadyObj(SSdb *pSdb, ESdbType type, const void *pKey) {

3,243✔

355

  return sdbAcquireAll(pSdb, type, pKey, false);

3,243✔

358

static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {

81,940✔

359

  int32_t type = pRow->type;

81,940✔

360

  sdbWriteLock(pSdb, type);

81,940✔

362

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

81,940✔

363

  sdbPrintOper(pSdb, pRow, "check");

81,940✔

365

    sdbFreeRow(pSdb, pRow, true);

16,261✔

368

  sdbUnLock(pSdb, type);

81,940✔

369

81,940✔

371

void sdbReleaseLock(SSdb *pSdb, void *pObj, bool lock) {

23,657,047✔

372

  if (pObj == NULL) return;

23,657,047✔

374

  SSdbRow *pRow = (SSdbRow *)((char *)pObj - sizeof(SSdbRow));

23,548,690✔

377

  int32_t type = pRow->type;

23,548,690✔

378

  if (lock) {

23,548,690✔

379

    sdbWriteLock(pSdb, type);

23,545,031✔

382

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

23,553,068✔

383

  sdbPrintOper(pSdb, pRow, "release");

23,555,573✔

384

  if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) {

23,554,609✔

385

    sdbFreeRow(pSdb, pRow, true);

65,679✔

387

  if (ref < 0) {

23,554,609✔

391

  if (lock) {

23,554,560✔

392

    sdbUnLock(pSdb, type);

23,549,632✔

396

void sdbRelease(SSdb *pSdb, void *pObj) { sdbReleaseLock(pSdb, pObj, true); }

23,651,763✔

398

void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {

19,303,436✔

399

  *ppObj = NULL;

19,303,436✔

401

  SHashObj *hash = sdbGetHash(pSdb, type);

19,303,436✔

402

  if (hash == NULL) return NULL;

19,303,038✔

404

  sdbReadLock(pSdb, type);

19,303,037✔

406

  SSdbRow **ppRow = taosHashIterate(hash, pIter);

19,304,811✔

407

  while (ppRow != NULL) {

19,340,660✔

408

    SSdbRow *pRow = *ppRow;

18,590,782✔

410

      ppRow = taosHashIterate(hash, ppRow);

34,927✔

411

      continue;

34,899✔

414

    (void) atomic_add_fetch_32(&pRow->refCount, 1);

18,555,855✔

416

    sdbPrintOper(pSdb, pRow, "fetch");

18,554,984✔

417

    *ppObj = pRow->pObj;

18,552,819✔

418

    break;

18,552,819✔

420

  sdbUnLock(pSdb, type);

19,302,697✔

422

  return ppRow;

19,304,072✔

425

void *sdbFetchAll(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj, ESdbStatus *status, bool lock) {

497,883✔

426

  *ppObj = NULL;

497,883✔

428

  SHashObj *hash = sdbGetHash(pSdb, type);

497,883✔

431

  if (lock) {

497,885✔

432

    sdbReadLock(pSdb, type);

489,322✔

435

  SSdbRow **ppRow = taosHashIterate(hash, pIter);

497,889✔

436

  while (ppRow != NULL) {

497,891✔

437

    SSdbRow *pRow = *ppRow;

392,577✔

443

    (void)atomic_add_fetch_32(&pRow->refCount, 1);

392,577✔

444

    sdbPrintOper(pSdb, pRow, "fetch");

392,578✔

445

    *ppObj = pRow->pObj;

392,577✔

446

    *status = pRow->status;

392,577✔

447

    break;

392,577✔

449

  if (lock) {

497,891✔

450

    sdbUnLock(pSdb, type);

489,326✔

453

  return ppRow;

497,888✔

456

void sdbCancelFetch(SSdb *pSdb, void *pIter) {

1,011,314✔

457

  if (pIter == NULL) return;

1,011,314✔

458

  SSdbRow *pRow = *(SSdbRow **)pIter;

1,009,055✔

459

  mTrace("cancel fetch row:%p", pRow);

1,009,055✔

460

  SHashObj *hash = sdbGetHash(pSdb, pRow->type);

1,009,055✔

463

  int32_t type = pRow->type;

1,009,055✔

464

  sdbReadLock(pSdb, type);

1,009,055✔

465

  taosHashCancelIterate(hash, pIter);

1,009,047✔

466

  sdbUnLock(pSdb, type);

1,009,055✔

469

void sdbCancelFetchByType(SSdb *pSdb, void *pIter, ESdbType type) {

1✔

472

  SHashObj *hash = sdbGetHash(pSdb, type);

1✔

475

  sdbReadLock(pSdb, type);

1✔

476

  taosHashCancelIterate(hash, pIter);

1✔

477

  sdbUnLock(pSdb, type);

1✔

480

void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2, void *p3) {

41,784✔

481

  SHashObj *hash = sdbGetHash(pSdb, type);

41,784✔

484

  sdbReadLock(pSdb, type);

41,785✔

486

  SSdbRow **ppRow = taosHashIterate(hash, NULL);

41,787✔

487

  while (ppRow != NULL) {

454,806✔

488

    SSdbRow *pRow = *ppRow;

413,028✔

489

    if (pRow->status == SDB_STATUS_READY) {

413,028✔

490

      bool isContinue = (*fp)(pSdb->pMnode, pRow->pObj, p1, p2, p3);

402,157✔

491

      if (!isContinue) {

401,747✔

492

        taosHashCancelIterate(hash, ppRow);

2✔

493

        break;

2✔

497

    ppRow = taosHashIterate(hash, ppRow);

412,616✔

500

  sdbUnLock(pSdb, type);

41,780✔

503

int32_t sdbGetSize(SSdb *pSdb, ESdbType type) {

568,365✔

504

  SHashObj *hash = sdbGetHash(pSdb, type);

568,365✔

507

  sdbReadLock(pSdb, type);

568,363✔

508

  int32_t size = taosHashGetSize(hash);

568,367✔

509

  sdbUnLock(pSdb, type);

568,363✔

511

  return size;

568,366✔

514

int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {

73,587✔

515

  SHashObj *hash = sdbGetHash(pSdb, type);

73,587✔

516

  if (hash == NULL) return -1;

73,587✔

518

  if (pSdb->keyTypes[type] != SDB_KEY_INT32) return -1;

73,586✔

520

  int32_t maxId = 0;

73,580✔

521

  sdbReadLock(pSdb, type);

73,580✔

523

  SSdbRow **ppRow = taosHashIterate(hash, NULL);

73,580✔

524

  while (ppRow != NULL) {

146,033✔

525

    SSdbRow *pRow = *ppRow;

72,453✔

526

    int32_t  id = *(int32_t *)pRow->pObj;

72,453✔

527

    maxId = TMAX(id, maxId);

72,453✔

528

    ppRow = taosHashIterate(hash, ppRow);

72,453✔

531

  sdbUnLock(pSdb, type);

73,580✔

532

  maxId = TMAX(maxId, pSdb->maxId[type]);

73,580✔

533

  return maxId + 1;

73,580✔

536

int64_t sdbGetTableVer(SSdb *pSdb, ESdbType type) {

371,585✔

537

  if (type >= SDB_MAX || type < 0) {

371,585✔

538

    terrno = TSDB_CODE_SDB_INVALID_TABLE_TYPE;

1✔

539

    return -1;

1✔

542

  return pSdb->tableVer[type];

371,584✔

taosdata / TDengine / #4452

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

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	#include "sdb.h"
18
19	static void sdbCheckRow(SSdb pSdb, SSdbRow pRow);
20
21	const char *sdbTableName(ESdbType type) {	4,773,930✔
22	switch (type) {	4,773,930!
23	case SDB_TRANS:	289,128✔
24	return "trans";	289,128✔
25	case SDB_CLUSTER:	34,818✔
26	return "cluster";	34,818✔
27	case SDB_MNODE:	37,237✔
28	return "mnode";	37,237✔
29	case SDB_QNODE:	42,973✔
30	return "qnode";	42,973✔
31	case SDB_SNODE:	10,616✔
32	return "snode";	10,616✔
33	case SDB_DNODE:	434,473✔
34	return "dnode";	434,473✔
35	case SDB_USER:	331,355✔
36	return "user";	331,355✔
37	case SDB_AUTH:	4✔
38	return "auth";	4✔
39	case SDB_ACCT:	19,752✔
40	return "acct";	19,752✔
41	case SDB_STREAM_CK:	4✔
42	return "stream_ck";	4✔
43	case SDB_STREAM:	26,761✔
44	return "stream";	26,761✔
45	case SDB_OFFSET:	4✔
46	return "offset";	4✔
47	case SDB_SUBSCRIBE:	14,626✔
48	return "subscribe";	14,626✔
49	case SDB_CONSUMER:	18,754✔
50	return "consumer";	18,754✔
51	case SDB_TOPIC:	13,187✔
52	return "topic";	13,187✔
53	case SDB_VGROUP:	2,054,559✔
54	return "vgroup";	2,054,559✔
55	case SDB_SMA:	18,755✔
56	return "sma";	18,755✔
57	case SDB_STB:	161,954✔
58	return "stb";	161,954✔
59	case SDB_DB:	370,186✔
60	return "db";	370,186✔
61	case SDB_FUNC:	13,557✔
62	return "func";	13,557✔
63	case SDB_IDX:	70,055✔
64	return "idx";	70,055✔
65	case SDB_VIEW:	13,364✔
66	return "view";	13,364✔
67	case SDB_STREAM_SEQ:	10,388✔
68	return "stream_seq";	10,388✔
69	case SDB_COMPACT:	10,457✔
70	return "compact";	10,457✔
71	case SDB_COMPACT_DETAIL:	10,691✔
72	return "compact_detail";	10,691✔
73	case SDB_GRANT:	23,830✔
74	return "grant";	23,830✔
75	case SDB_ARBGROUP:	10,564✔
76	return "arb_group";	10,564✔
77	case SDB_ANODE:	10,401✔
78	return "anode";	10,401✔
79	case SDB_BNODE:	11,442✔
80	return "bnode";	11,442✔
81	case SDB_CFG:	710,262✔
82	return "config";	710,262✔
83	default:	×
84	return "undefine";	×
85	}
86	}
87
88	const char *sdbStatusName(ESdbStatus status) {	3,831,140✔
89	switch (status) {	3,831,140!
90	case SDB_STATUS_CREATING:	36,421✔
91	return "creating";	36,421✔
92	case SDB_STATUS_DROPPING:	9,493✔
93	return "dropping";	9,493✔
94	case SDB_STATUS_READY:	3,709,639✔
95	return "ready";	3,709,639✔
96	case SDB_STATUS_DROPPED:	66,503✔
97	return "dropped";	66,503✔
98	case SDB_STATUS_INIT:	9,078✔
99	return "init";	9,078✔
100	case SDB_STATUS_UPDATE:	×
101	return "update";	×
102	default:	6✔
103	return "undefine";	6✔
104	}
105	}
106
107	void sdbPrintOper(SSdb pSdb, SSdbRow pRow, const char *oper) {	48,874,100✔
108	#if 1
109	EKeyType keyType = pSdb->keyTypes[pRow->type];	48,874,100✔
110
111	if (keyType == SDB_KEY_BINARY) {	48,874,100✔
112	mTrace("%s:%s, ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type), (char *)pRow->pObj,	7,755,791✔
113	pRow->refCount, oper, pRow, pRow->pObj, sdbStatusName(pRow->status));
114	} else if (keyType == SDB_KEY_INT32) {	41,118,309✔
115	mTrace("%s:%d, ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type), (int32_t )pRow->pObj,	40,804,367✔
116	pRow->refCount, oper, pRow, pRow->pObj, sdbStatusName(pRow->status));
117	} else if (keyType == SDB_KEY_INT64) {	313,942!
118	mTrace("%s:%" PRId64 ", ref:%d oper:%s row:%p row->pObj:%p status:%s", sdbTableName(pRow->type),	320,482✔
119	(int64_t )pRow->pObj, pRow->refCount, oper, pRow, pRow->pObj, sdbStatusName(pRow->status));
120	} else {
121	}
122	#endif
123	}	48,874,262✔
124
125	static SHashObj sdbGetHash(SSdb pSdb, int32_t type) {	27,307,745✔
126	if (type >= SDB_MAX \|\| type < 0) {	27,307,745!
127	terrno = TSDB_CODE_SDB_INVALID_TABLE_TYPE;	×
128	return NULL;	1✔
129	}
130
131	SHashObj *hash = pSdb->hashObjs[type];	27,307,909✔
132	if (hash == NULL) {	27,307,909✔
133	terrno = TSDB_CODE_APP_ERROR;	4✔
134	return NULL;	4✔
135	}
136
137	return hash;	27,307,905✔
138	}
139
140	static int32_t sdbGetkeySize(SSdb pSdb, ESdbType type, const void pKey) {	5,819,980✔
141	int32_t keySize = 0;	5,819,980✔
142	EKeyType keyType = pSdb->keyTypes[type];	5,819,980✔
143
144	if (keyType == SDB_KEY_INT32) {	5,819,980✔
145	keySize = sizeof(int32_t);	3,345,746✔
146	} else if (keyType == SDB_KEY_BINARY) {	2,474,234✔
147	keySize = strlen(pKey) + 1;	2,450,894✔
148	} else {
149	keySize = sizeof(int64_t);	23,340✔
150	}
151
152	return keySize;	5,819,980✔
153	}
154
155	static int32_t sdbInsertRow(SSdb pSdb, SHashObj hash, SSdbRaw pRaw, SSdbRow pRow, int32_t keySize) {	369,908✔
156	int32_t type = pRow->type;	369,908✔
157	sdbWriteLock(pSdb, type);	369,908✔
158
159	SSdbRow *pOldRow = taosHashGet(hash, pRow->pObj, keySize);	369,908✔
160	if (pOldRow != NULL) {	369,908!
161	sdbUnLock(pSdb, type);	×
162	sdbFreeRow(pSdb, pRow, false);	×
163	terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE;	×
164	return terrno;	×
165	}
166
167	pRow->refCount = 0;	369,908✔
168	pRow->status = pRaw->status;	369,908✔
169	sdbPrintOper(pSdb, pRow, "insert");	369,908✔
170
171	int32_t code = 0;	369,908✔
172	if ((code = taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *))) != 0) {	369,908!
173	sdbUnLock(pSdb, type);	×
174	sdbFreeRow(pSdb, pRow, false);	×
175	return code;	×
176	}
177
178	SdbInsertFp insertFp = pSdb->insertFps[pRow->type];	369,908✔
179	if (insertFp != NULL) {	369,908!
180	code = (*insertFp)(pSdb, pRow->pObj);	369,908✔
181	if (code != 0) {	369,908!
182	if (taosHashRemove(hash, pRow->pObj, keySize) != 0) {	×
183	mError("failed to remove row from hash");	×
184	}
185	sdbFreeRow(pSdb, pRow, false);	×
186	sdbUnLock(pSdb, type);	×
187	terrno = code;	×
188	return terrno;	×
189	}
190	}
191
192	sdbUnLock(pSdb, type);	369,908✔
193
194	if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT32) {	369,908✔
195	pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], ((int32_t )pRow->pObj));	86,205✔
196	}
197	if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT64) {	369,908✔
198	pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], ((int64_t )pRow->pObj));	3,204✔
199	}
200	pSdb->tableVer[pRow->type]++;	369,908✔
201
202	return 0;	369,908✔
203	}
204
205	static int32_t sdbUpdateRow(SSdb pSdb, SHashObj hash, SSdbRaw pRaw, SSdbRow pNewRow, int32_t keySize) {	485,821✔
206	int32_t type = pNewRow->type;	485,821✔
207	sdbWriteLock(pSdb, type);	485,821✔
208
209	SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize);	485,821✔
210	if (ppOldRow == NULL \|\| *ppOldRow == NULL) {	485,821!
211	sdbUnLock(pSdb, type);	351,570✔
212	return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize);	351,570✔
213	}
214
215	SSdbRow pOldRow = ppOldRow;	134,251✔
216	pOldRow->status = pRaw->status;	134,251✔
217	sdbPrintOper(pSdb, pOldRow, "update");	134,251✔
218
219	int32_t code = 0;	134,251✔
220	SdbUpdateFp updateFp = pSdb->updateFps[type];	134,251✔
221	if (updateFp != NULL) {	134,251!
222	code = (*updateFp)(pSdb, pOldRow->pObj, pNewRow->pObj);	134,251✔
223	}
224	sdbUnLock(pSdb, type);	134,251✔
225
226	// sdbUnLock(pSdb, type);
227	sdbFreeRow(pSdb, pNewRow, false);	134,251✔
228
229	pSdb->tableVer[pOldRow->type]++;	134,251✔
230	return code;	134,251✔
231	}
232
233	static int32_t sdbDeleteRow(SSdb pSdb, SHashObj hash, SSdbRaw pRaw, SSdbRow pRow, int32_t keySize) {	81,979✔
234	int32_t type = pRow->type;	81,979✔
235	sdbWriteLock(pSdb, type);	81,979✔
236
237	SSdbRow **ppOldRow = taosHashGet(hash, pRow->pObj, keySize);	81,979✔
238	if (ppOldRow == NULL \|\| *ppOldRow == NULL) {	81,979!
239	sdbUnLock(pSdb, type);	39✔
240	sdbFreeRow(pSdb, pRow, false);	39✔
241	terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;	39✔
242	return terrno;	39✔
243	}
244	SSdbRow pOldRow = ppOldRow;	81,940✔
245	pOldRow->status = pRaw->status;	81,940✔
246
247	(void)atomic_add_fetch_32(&pOldRow->refCount, 1);	81,940✔
248	sdbPrintOper(pSdb, pOldRow, "delete");	81,940✔
249
250	if (taosHashRemove(hash, pOldRow->pObj, keySize) != 0) {	81,940!
251	sdbUnLock(pSdb, type);	×
252	sdbFreeRow(pSdb, pRow, false);	×
253	terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;	×
254	return terrno;	×
255	}
256	pSdb->tableVer[pOldRow->type]++;	81,940✔
257	sdbUnLock(pSdb, type);	81,940✔
258
259	sdbFreeRow(pSdb, pRow, false);	81,940✔
260
261	sdbCheckRow(pSdb, pOldRow);	81,940✔
262	return 0;	81,940✔
263	}
264
265	int32_t sdbWriteWithoutFree(SSdb pSdb, SSdbRaw pRaw) {	586,140✔
266	if (pRaw->type == SDB_CFG) {	586,140✔
267	mTrace("sdb write cfg");	240,602✔
268	}
269	SHashObj *hash = sdbGetHash(pSdb, pRaw->type);	586,140✔
270	if (hash == NULL) return terrno;	586,140✔
271
272	SdbDecodeFp decodeFp = pSdb->decodeFps[pRaw->type];	586,138✔
273	SSdbRow pRow = (decodeFp)(pRaw);	586,138✔
274	if (pRow == NULL) return terrno;	586,138!
275
276	pRow->type = pRaw->type;	586,138✔
277
278	int32_t keySize = sdbGetkeySize(pSdb, pRow->type, pRow->pObj);	586,138✔
279	int32_t code = TSDB_CODE_SDB_INVALID_ACTION_TYPE;	586,138✔
280
281	switch (pRaw->status) {	586,138!
282	case SDB_STATUS_CREATING:	18,338✔
283	code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize);	18,338✔
284	break;	18,338✔
285	case SDB_STATUS_READY:	485,821✔
286	case SDB_STATUS_UPDATE:
287	case SDB_STATUS_DROPPING:
288	code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize);	485,821✔
289	break;	485,821✔
290	case SDB_STATUS_DROPPED:	81,979✔
291	code = sdbDeleteRow(pSdb, hash, pRaw, pRow, keySize);	81,979✔
292	break;	81,979✔
293	}
294
295	return code;	586,138✔
296	}
297
298	int32_t sdbWrite(SSdb pSdb, SSdbRaw pRaw) {	64,542✔
299	int32_t code = sdbWriteWithoutFree(pSdb, pRaw);	64,542✔
300	sdbFreeRaw(pRaw);	64,542✔
301	return code;	64,542✔
302	}
303
304	void sdbAcquireAll(SSdb pSdb, ESdbType type, const void *pKey, bool onlyReady) {	5,234,080✔
305	terrno = 0;	5,234,080✔
306
307	SHashObj *hash = sdbGetHash(pSdb, type);	5,233,983✔
308	if (hash == NULL) return NULL;	5,233,861✔
309
310	void *pRet = NULL;	5,233,860✔
311	int32_t keySize = sdbGetkeySize(pSdb, type, pKey);	5,233,860✔
312
313	sdbReadLock(pSdb, type);	5,233,809✔
314
315	SSdbRow **ppRow = taosHashGet(hash, pKey, keySize);	5,234,192✔
316	if (ppRow == NULL \|\| *ppRow == NULL) {	5,234,191✔
317	sdbUnLock(pSdb, type);	618,806✔
318	terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;	618,856✔
319	return NULL;	618,856✔
320	}
321
322	SSdbRow pRow = ppRow;	4,615,385✔
323	switch (pRow->status) {	4,615,385!
324	case SDB_STATUS_READY:	4,607,183✔
325	(void)atomic_add_fetch_32(&pRow->refCount, 1);	4,607,183✔
326	pRet = pRow->pObj;	4,607,228✔
327	sdbPrintOper(pSdb, pRow, "acquire");	4,607,228✔
328	break;	4,607,112✔
329	case SDB_STATUS_CREATING:	5,768✔
330	terrno = TSDB_CODE_SDB_OBJ_CREATING;	5,768✔
331	break;	5,768✔
332	case SDB_STATUS_DROPPING:	2,441✔
333	terrno = TSDB_CODE_SDB_OBJ_DROPPING;	2,441✔
334	break;	2,441✔
335	default:	×
336	terrno = TSDB_CODE_APP_ERROR;	×
337	break;	×
338	}
339
340	if (pRet == NULL) {	4,615,321✔
341	if (!onlyReady) {	8,209!
342	terrno = 0;	×
343	(void)atomic_add_fetch_32(&pRow->refCount, 1);	×
344	pRet = pRow->pObj;	×
345	sdbPrintOper(pSdb, pRow, "acquire");	×
346	}
347	}
348
349	sdbUnLock(pSdb, type);	4,615,321✔
350	return pRet;	4,615,333✔
351	}
352
353	void sdbAcquire(SSdb pSdb, ESdbType type, const void *pKey) { return sdbAcquireAll(pSdb, type, pKey, true); }	5,230,844✔
354	void sdbAcquireNotReadyObj(SSdb pSdb, ESdbType type, const void *pKey) {	3,243✔
355	return sdbAcquireAll(pSdb, type, pKey, false);	3,243✔
356	}
357
358	static void sdbCheckRow(SSdb pSdb, SSdbRow pRow) {	81,940✔
359	int32_t type = pRow->type;	81,940✔
360	sdbWriteLock(pSdb, type);	81,940✔
361
362	int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);	81,940✔
363	sdbPrintOper(pSdb, pRow, "check");	81,940✔
364	if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) {	81,940!
365	sdbFreeRow(pSdb, pRow, true);	16,261✔
366	}
367
368	sdbUnLock(pSdb, type);	81,940✔
369	}	81,940✔
370
371	void sdbReleaseLock(SSdb pSdb, void pObj, bool lock) {	23,657,047✔
372	if (pObj == NULL) return;	23,657,047✔
373
374	SSdbRow pRow = (SSdbRow )((char *)pObj - sizeof(SSdbRow));	23,548,690✔
375	if (pRow->type >= SDB_MAX) return;	23,548,690!
376
377	int32_t type = pRow->type;	23,548,690✔
378	if (lock) {	23,548,690✔
379	sdbWriteLock(pSdb, type);	23,545,031✔
380	}
381
382	int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);	23,553,068✔
383	sdbPrintOper(pSdb, pRow, "release");	23,555,573✔
384	if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) {	23,554,609✔
385	sdbFreeRow(pSdb, pRow, true);	65,679✔
386	}
387	if (ref < 0) {	23,554,609✔
388	mError("row:%p, ref:%d, type:%s", pRow, ref, sdbTableName(type));	4,458!
389	}
390
391	if (lock) {	23,554,560✔
392	sdbUnLock(pSdb, type);	23,549,632✔
393	}
394	}
395
396	void sdbRelease(SSdb pSdb, void pObj) { sdbReleaseLock(pSdb, pObj, true); }	23,651,763✔
397
398	void sdbFetch(SSdb pSdb, ESdbType type, void pIter, void *ppObj) {	19,303,436✔
399	*ppObj = NULL;	19,303,436✔
400
401	SHashObj *hash = sdbGetHash(pSdb, type);	19,303,436✔
402	if (hash == NULL) return NULL;	19,303,038✔
403
404	sdbReadLock(pSdb, type);	19,303,037✔
405
406	SSdbRow **ppRow = taosHashIterate(hash, pIter);	19,304,811✔
407	while (ppRow != NULL) {	19,340,660✔
408	SSdbRow pRow = ppRow;	18,590,782✔
409	if (pRow == NULL \|\| pRow->status != SDB_STATUS_READY) {	18,590,782!
410	ppRow = taosHashIterate(hash, ppRow);	34,927✔
411	continue;	34,899✔
412	}
413
414	(void) atomic_add_fetch_32(&pRow->refCount, 1);	18,555,855✔
415
416	sdbPrintOper(pSdb, pRow, "fetch");	18,554,984✔
417	*ppObj = pRow->pObj;	18,552,819✔
418	break;	18,552,819✔
419	}
420	sdbUnLock(pSdb, type);	19,302,697✔
421
422	return ppRow;	19,304,072✔
423	}
424
425	void sdbFetchAll(SSdb pSdb, ESdbType type, void pIter, void ppObj, ESdbStatus status, bool lock) {	497,883✔
426	*ppObj = NULL;	497,883✔
427
428	SHashObj *hash = sdbGetHash(pSdb, type);	497,883✔
429	if (hash == NULL) return NULL;	497,885!
430
431	if (lock) {	497,885✔
432	sdbReadLock(pSdb, type);	489,322✔
433	}
434
435	SSdbRow **ppRow = taosHashIterate(hash, pIter);	497,889✔
436	while (ppRow != NULL) {	497,891✔
437	SSdbRow pRow = ppRow;	392,577✔
438	if (pRow == NULL) {	392,577!
439	ppRow = taosHashIterate(hash, ppRow);	×
440	continue;	×
441	}
442
443	(void)atomic_add_fetch_32(&pRow->refCount, 1);	392,577✔
444	sdbPrintOper(pSdb, pRow, "fetch");	392,578✔
445	*ppObj = pRow->pObj;	392,577✔
446	*status = pRow->status;	392,577✔
447	break;	392,577✔
448	}
449	if (lock) {	497,891✔
450	sdbUnLock(pSdb, type);	489,326✔
451	}
452
453	return ppRow;	497,888✔
454	}
455
456	void sdbCancelFetch(SSdb pSdb, void pIter) {	1,011,314✔
457	if (pIter == NULL) return;	1,011,314✔
458	SSdbRow pRow = (SSdbRow **)pIter;	1,009,055✔
459	mTrace("cancel fetch row:%p", pRow);	1,009,055✔
460	SHashObj *hash = sdbGetHash(pSdb, pRow->type);	1,009,055✔
461	if (hash == NULL) return;	1,009,055!
462
463	int32_t type = pRow->type;	1,009,055✔
464	sdbReadLock(pSdb, type);	1,009,055✔
465	taosHashCancelIterate(hash, pIter);	1,009,047✔
466	sdbUnLock(pSdb, type);	1,009,055✔
467	}
468
469	void sdbCancelFetchByType(SSdb pSdb, void pIter, ESdbType type) {	1✔
470	if (pIter == NULL) return;	1!
471	if (type >= SDB_MAX \|\| type < 0) return;	1!
472	SHashObj *hash = sdbGetHash(pSdb, type);	1✔
473	if (hash == NULL) return;	1!
474
475	sdbReadLock(pSdb, type);	1✔
476	taosHashCancelIterate(hash, pIter);	1✔
477	sdbUnLock(pSdb, type);	1✔
478	}
479
480	void sdbTraverse(SSdb pSdb, ESdbType type, sdbTraverseFp fp, void p1, void p2, void p3) {	41,784✔
481	SHashObj *hash = sdbGetHash(pSdb, type);	41,784✔
482	if (hash == NULL) return;	41,785!
483
484	sdbReadLock(pSdb, type);	41,785✔
485
486	SSdbRow **ppRow = taosHashIterate(hash, NULL);	41,787✔
487	while (ppRow != NULL) {	454,806✔
488	SSdbRow pRow = ppRow;	413,028✔
489	if (pRow->status == SDB_STATUS_READY) {	413,028✔
490	bool isContinue = (*fp)(pSdb->pMnode, pRow->pObj, p1, p2, p3);	402,157✔
491	if (!isContinue) {	401,747✔
492	taosHashCancelIterate(hash, ppRow);	2✔
493	break;	2✔
494	}
495	}
496
497	ppRow = taosHashIterate(hash, ppRow);	412,616✔
498	}
499
500	sdbUnLock(pSdb, type);	41,780✔
501	}
502
503	int32_t sdbGetSize(SSdb *pSdb, ESdbType type) {	568,365✔
504	SHashObj *hash = sdbGetHash(pSdb, type);	568,365✔
505	if (hash == NULL) return 0;	568,363!
506
507	sdbReadLock(pSdb, type);	568,363✔
508	int32_t size = taosHashGetSize(hash);	568,367✔
509	sdbUnLock(pSdb, type);	568,363✔
510
511	return size;	568,366✔
512	}
513
514	int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {	73,587✔
515	SHashObj *hash = sdbGetHash(pSdb, type);	73,587✔
516	if (hash == NULL) return -1;	73,587✔
517
518	if (pSdb->keyTypes[type] != SDB_KEY_INT32) return -1;	73,586✔
519
520	int32_t maxId = 0;	73,580✔
521	sdbReadLock(pSdb, type);	73,580✔
522
523	SSdbRow **ppRow = taosHashIterate(hash, NULL);	73,580✔
524	while (ppRow != NULL) {	146,033✔
525	SSdbRow pRow = ppRow;	72,453✔
526	int32_t id = (int32_t )pRow->pObj;	72,453✔
527	maxId = TMAX(id, maxId);	72,453✔
528	ppRow = taosHashIterate(hash, ppRow);	72,453✔
529	}
530
531	sdbUnLock(pSdb, type);	73,580✔
532	maxId = TMAX(maxId, pSdb->maxId[type]);	73,580✔
533	return maxId + 1;	73,580✔
534	}
535
536	int64_t sdbGetTableVer(SSdb *pSdb, ESdbType type) {	371,585✔
537	if (type >= SDB_MAX \|\| type < 0) {	371,585✔
538	terrno = TSDB_CODE_SDB_INVALID_TABLE_TYPE;	1✔
539	return -1;	1✔
540	}
541
542	return pSdb->tableVer[type];	371,584✔
543	}
544
545	bool countValid(SMnode pMnode, void pObj, void p1, void p2, void *p3) {	×
546	int32_t *pInt = p1;	×
547	(*pInt) += 1;	×
548	return true;	×
549	}
550
551	int32_t sdbGetValidSize(SSdb *pSdb, ESdbType type) {	×
552	int32_t num = 0;	×
553	sdbTraverse(pSdb, type, countValid, &num, 0, 0);	×
554	return num;	×
555	}

taosdata / TDengine / #4452

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