From 6cff4dcfd83be7d9f5181be72ebfafa6fb80b48a Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 15 Aug 2022 14:42:40 +0800 Subject: [PATCH 1/4] refactor(sync): add syncIsReadyForRead --- include/libs/sync/sync.h | 4 ++ source/dnode/vnode/src/inc/vnd.h | 3 +- source/dnode/vnode/src/vnd/vnodeSvr.c | 4 +- source/dnode/vnode/src/vnd/vnodeSync.c | 14 +++++++ source/libs/sync/src/syncMain.c | 56 ++++++++++++++++++++++++++ 5 files changed, 78 insertions(+), 3 deletions(-) diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index aa563343f8..6d8895eb96 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -30,6 +30,7 @@ extern bool gRaftDetailLog; #define SYNC_SPEED_UP_HB_TIMER 400 #define SYNC_SPEED_UP_AFTER_MS (1000 * 20) #define SYNC_SLOW_DOWN_RANGE 100 +#define SYNC_MAX_READ_RANGE 10 #define SYNC_MAX_BATCH_SIZE 1 #define SYNC_INDEX_BEGIN 0 @@ -210,9 +211,12 @@ void syncStop(int64_t rid); int32_t syncSetStandby(int64_t rid); ESyncState syncGetMyRole(int64_t rid); bool syncIsReady(int64_t rid); +bool syncIsReadyForRead(int64_t rid); const char* syncGetMyRoleStr(int64_t rid); bool syncRestoreFinish(int64_t rid); SyncTerm syncGetMyTerm(int64_t rid); +SyncIndex syncGetLastIndex(int64_t rid); +SyncIndex syncGetCommitIndex(int64_t rid); SyncGroupId syncGetVgId(int64_t rid); void syncGetEpSet(int64_t rid, SEpSet* pEpSet); void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet); diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index dd1facb462..5608efa7af 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -80,7 +80,7 @@ int32_t vnodeQueryOpen(SVnode* pVnode); void vnodeQueryClose(SVnode* pVnode); int32_t vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg, bool direct); int vnodeGetTableCfg(SVnode* pVnode, SRpcMsg* pMsg, bool direct); -int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg); +int32_t vnodeGetBatchMeta(SVnode* pVnode, SRpcMsg* pMsg); // vnodeCommit.c int32_t vnodeBegin(SVnode* pVnode); @@ -98,6 +98,7 @@ void vnodeSyncStart(SVnode* pVnode); void vnodeSyncClose(SVnode* pVnode); void vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg); bool vnodeIsLeader(SVnode* pVnode); +bool vnodeIsReadyForRead(SVnode* pVnode); #ifdef __cplusplus } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index d5c5e18668..e17d4a0b6d 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -281,7 +281,7 @@ int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { vTrace("message in vnode query queue is processing"); - if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) { + if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsReadyForRead(pVnode)) { vnodeRedirectRpcMsg(pVnode, pMsg); return 0; } @@ -305,7 +305,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) { vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg); if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG || pMsg->msgType == TDMT_VND_BATCH_META) && - !vnodeIsLeader(pVnode)) { + !vnodeIsReadyForRead(pVnode)) { vnodeRedirectRpcMsg(pVnode, pMsg); return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index 50d32f5f5e..2779ab724f 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -771,3 +771,17 @@ bool vnodeIsLeader(SVnode *pVnode) { return true; } + +bool vnodeIsReadyForRead(SVnode *pVnode) { + if (syncIsReady(pVnode->sync)) { + return true; + } + + if (syncIsReadyForRead(pVnode->sync)) { + return true; + } + + vDebug("vgId:%d, vnode not ready for read, state:%s, last:%ld, cmt:%ld", pVnode->config.vgId, + syncGetMyRoleStr(pVnode->sync), syncGetLastIndex(pVnode->sync), syncGetCommitIndex(pVnode->sync)); + return false; +} diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index c7784cd62e..5656f19890 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -392,6 +392,29 @@ bool syncIsReady(int64_t rid) { return b; } +bool syncIsReadyForRead(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return false; + } + ASSERT(rid == pSyncNode->rid); + + // TODO: last not noop? + SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); + bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && (pSyncNode->commitIndex >= lastIndex - SYNC_MAX_READ_RANGE); + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + + // if false, set error code + if (false == b) { + if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) { + terrno = TSDB_CODE_SYN_NOT_LEADER; + } else { + terrno = TSDB_CODE_APP_NOT_READY; + } + } + return b; +} + bool syncIsRestoreFinish(int64_t rid) { SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { @@ -519,6 +542,30 @@ SyncTerm syncGetMyTerm(int64_t rid) { return term; } +SyncIndex syncGetLastIndex(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return SYNC_INDEX_INVALID; + } + ASSERT(rid == pSyncNode->rid); + SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return lastIndex; +} + +SyncIndex syncGetCommitIndex(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return SYNC_INDEX_INVALID; + } + ASSERT(rid == pSyncNode->rid); + SyncIndex cmtIndex = pSyncNode->commitIndex; + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return cmtIndex; +} + SyncGroupId syncGetVgId(int64_t rid) { SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { @@ -828,6 +875,15 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) { pSyncNode->changing = true; } + // not restored + if (!pSyncNode->restoreFinish) { + ret = -1; + terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY; + sError("vgId:%d, failed to sync propose since not ready, type:%s, last:%ld, cmt:%ld", pSyncNode->vgId, + TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex); + goto _END; + } + SRespStub stub; stub.createTime = taosGetTimestampMs(); stub.rpcMsg = *pMsg; From 947d90e196a69b176814c0110ee31052f0b30072 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 15 Aug 2022 17:36:46 +0800 Subject: [PATCH 2/4] refactor(sync): add syncIsReadyForRead --- source/dnode/vnode/src/inc/vnd.h | 1 + 1 file changed, 1 insertion(+) diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index 5608efa7af..5164e22474 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -99,6 +99,7 @@ void vnodeSyncClose(SVnode* pVnode); void vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg); bool vnodeIsLeader(SVnode* pVnode); bool vnodeIsReadyForRead(SVnode* pVnode); +bool vnodeIsRoleLeader(SVnode* pVnode); #ifdef __cplusplus } From 4164d1463754e4f086f1695f4794299ad551eec1 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 15 Aug 2022 18:50:14 +0800 Subject: [PATCH 3/4] refactor(sync): add syncIsReadyForRead --- source/libs/sync/src/syncMain.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 5656f19890..1991560d42 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -875,8 +875,8 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) { pSyncNode->changing = true; } - // not restored - if (!pSyncNode->restoreFinish) { + // not restored, vnode enable + if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) { ret = -1; terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY; sError("vgId:%d, failed to sync propose since not ready, type:%s, last:%ld, cmt:%ld", pSyncNode->vgId, From 4b85d81d373ad91d497127b65292049175d79b10 Mon Sep 17 00:00:00 2001 From: t_max <1172915550@qq.com> Date: Tue, 16 Aug 2022 14:31:42 +0800 Subject: [PATCH 4/4] docs(TMQ): improve tmq document --- docs/zh/07-develop/07-tmq.mdx | 105 ++++++++++++++++++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index f36f76fd85..aa599c2173 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -87,6 +87,25 @@ void commitSync() throws SQLException; void close() throws SQLException; ``` + + + +```go +func NewConsumer(conf *Config) (*Consumer, error) + +func (c *Consumer) Close() error + +func (c *Consumer) Commit(ctx context.Context, message unsafe.Pointer) error + +func (c *Consumer) FreeMessage(message unsafe.Pointer) + +func (c *Consumer) Poll(timeout time.Duration) (*Result, error) + +func (c *Consumer) Subscribe(topics []string) error + +func (c *Consumer) Unsubscribe() error +``` + @@ -229,6 +248,56 @@ public class MetersDeserializer extends ReferenceDeserializer { } ``` + + + +```go +config := tmq.NewConfig() +defer config.Destroy() +err = config.SetGroupID("test") +if err != nil { + panic(err) +} +err = config.SetAutoOffsetReset("earliest") +if err != nil { + panic(err) +} +err = config.SetConnectIP("127.0.0.1") +if err != nil { + panic(err) +} +err = config.SetConnectUser("root") +if err != nil { + panic(err) +} +err = config.SetConnectPass("taosdata") +if err != nil { + panic(err) +} +err = config.SetConnectPort("6030") +if err != nil { + panic(err) +} +err = config.SetMsgWithTableName(true) +if err != nil { + panic(err) +} +err = config.EnableHeartBeat() +if err != nil { + panic(err) +} +err = config.EnableAutoCommit(func(result *wrapper.TMQCommitCallbackResult) { + if result.ErrCode != 0 { + errStr := wrapper.TMQErr2Str(result.ErrCode) + err := errors.NewError(int(result.ErrCode), errStr) + panic(err) + } +}) +if err != nil { + panic(err) +} +``` + @@ -260,6 +329,20 @@ topics.add("tmq_topic"); consumer.subscribe(topics); ``` + + + +```go +consumer, err := tmq.NewConsumer(config) +if err != nil { + panic(err) +} +err = consumer.Subscribe([]string{"example_tmq_topic"}) +if err != nil { + panic(err) +} +``` + @@ -293,6 +376,21 @@ while(running){ } ``` + + + +```go +for { + result, err := consumer.Poll(time.Second) + if err != nil { + panic(err) + } + fmt.Println(result) + consumer.Commit(context.Background(), result.Message) + consumer.FreeMessage(result.Message) +} +``` + @@ -322,6 +420,13 @@ consumer.unsubscribe(); consumer.close(); ``` + + + +```go +consumer.Close() +``` +