Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 c7d0bee0 编写于 作者: D dapan1121
浏览文件
操作

Merge branch '3.0' into catalog_dev

上级 924e3ad6 33a13ff5
展开全部 隐藏空白更改
内联 并排
Showing with 988 addition and 1338 deletion
include/dnode/mnode/sdb/sdb.h
浏览文件 @ c7d0bee0
...@@ -144,9 +144,10 @@ typedef struct SSdbRow SSdbRow; ...@@ -144,9 +144,10 @@ typedef struct SSdbRow SSdbRow;
typedef enum { SDB_KEY_BINARY = 1, SDB_KEY_INT32 = 2, SDB_KEY_INT64 = 3 } EKeyType; typedef enum { SDB_KEY_BINARY = 1, SDB_KEY_INT32 = 2, SDB_KEY_INT64 = 3 } EKeyType;
typedef enum { typedef enum {
SDB_STATUS_CREATING = 1, SDB_STATUS_CREATING = 1,
SDB_STATUS_READY = 2, SDB_STATUS_UPDATING = 2,
SDB_STATUS_DROPPING = 3, SDB_STATUS_DROPPING = 3,
SDB_STATUS_DROPPED = 4 SDB_STATUS_READY = 4,
SDB_STATUS_DROPPED = 5
} ESdbStatus; } ESdbStatus;
typedef enum { typedef enum {
...@@ -174,67 +175,19 @@ typedef SSdbRow *(*SdbDecodeFp)(SSdbRaw *pRaw); ...@@ -174,67 +175,19 @@ typedef SSdbRow *(*SdbDecodeFp)(SSdbRaw *pRaw);
typedef SSdbRaw *(*SdbEncodeFp)(void *pObj); typedef SSdbRaw *(*SdbEncodeFp)(void *pObj);
typedef struct { typedef struct {
/** ESdbType sdbType;
* @brief The sdb type of the table. EKeyType keyType;
*
*/
ESdbType sdbType;
/**
* @brief The key type of the table.
*
*/
EKeyType keyType;
/**
* @brief The callback function when the table is first deployed.
*
*/
SdbDeployFp deployFp; SdbDeployFp deployFp;
/**
* @brief Encode one row of the table into rawdata.
*
*/
SdbEncodeFp encodeFp; SdbEncodeFp encodeFp;
/**
* @brief Decode one row of the table from rawdata.
*
*/
SdbDecodeFp decodeFp; SdbDecodeFp decodeFp;
/**
* @brief The callback function when insert a row to sdb.
*
*/
SdbInsertFp insertFp; SdbInsertFp insertFp;
/**
* @brief The callback function when undate a row in sdb.
*
*/
SdbUpdateFp updateFp; SdbUpdateFp updateFp;
/**
* @brief The callback function when delete a row from sdb.
*
*/
SdbDeleteFp deleteFp; SdbDeleteFp deleteFp;
} SSdbTable; } SSdbTable;
typedef struct SSdbOpt { typedef struct SSdbOpt {
/**
* @brief The path of the sdb file.
*
*/
const char *path; const char *path;
SMnode *pMnode;
/**
* @brief The mnode object.
*
*/
SMnode *pMnode;
} SSdbOpt; } SSdbOpt;
/** /**
......
include/dnode/vnode/vnode.h
浏览文件 @ c7d0bee0
...@@ -68,9 +68,11 @@ typedef struct SVnodeCfg { ...@@ -68,9 +68,11 @@ typedef struct SVnodeCfg {
/** /**
* @brief Initialize the vnode module * @brief Initialize the vnode module
* *
* @param nthreads number of commit threads. 0 for no threads and
* a schedule queue should be given (TODO)
* @return int 0 for success and -1 for failure * @return int 0 for success and -1 for failure
*/ */
int vnodeInit(); int vnodeInit(uint16_t nthreads);
/** /**
* @brief clear a vnode * @brief clear a vnode
......
include/libs/planner/planner.h
浏览文件 @ c7d0bee0
...@@ -22,6 +22,7 @@ extern "C" { ...@@ -22,6 +22,7 @@ extern "C" {
#define QUERY_TYPE_MERGE 1 #define QUERY_TYPE_MERGE 1
#define QUERY_TYPE_PARTIAL 2 #define QUERY_TYPE_PARTIAL 2
#define QUERY_TYPE_SCAN 3
enum OPERATOR_TYPE_E { enum OPERATOR_TYPE_E {
OP_TableScan = 1, OP_TableScan = 1,
...@@ -54,90 +55,37 @@ enum OPERATOR_TYPE_E { ...@@ -54,90 +55,37 @@ enum OPERATOR_TYPE_E {
struct SEpSet; struct SEpSet;
struct SQueryPlanNode; struct SQueryPlanNode;
struct SQueryDistPlanNode; struct SPhyNode;
struct SQueryStmtInfo; struct SQueryStmtInfo;
typedef struct SSubquery { typedef struct SSubplan {
int64_t queryId; // the subquery id created by qnode int32_t type; // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL|QUERY_TYPE_SCAN
int32_t type; // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL SArray *pDatasource; // the datasource subplan,from which to fetch the result
int32_t level; // the execution level of current subquery, starting from 0. struct SPhyNode *pNode; // physical plan of current subplan
SArray *pUpstream; // the upstream,from which to fetch the result } SSubplan;
struct SQueryDistPlanNode *pNode; // physical plan of current subquery
} SSubquery;
typedef struct SQueryJob { typedef struct SQueryDag {
SArray **pSubqueries; SArray **pSubplans;
int32_t numOfLevels; } SQueryDag;
int32_t currentLevel;
} SQueryJob;
/**
* Optimize the query execution plan, currently not implement yet.
* @param pQueryNode
* @return
*/
int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
/**
* Create the query plan according to the bound AST, which is in the form of pQueryInfo
* @param pQueryInfo
* @param pQueryNode
* @return
*/
int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
/**
* Convert the query plan to string, in order to display it in the shell.
* @param pQueryNode
* @return
*/
int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
/** /**
* Restore the SQL statement according to the logic query plan. * Create the physical plan for the query, according to the AST.
* @param pQueryNode
* @param sql
* @return
*/ */
int32_t qQueryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql); int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag);
/** int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str);
* Create the physical plan for the query, according to the logic plan.
* @param pQueryNode
* @param pPhyNode
* @return
*/
int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDistPlanNode *pPhyNode);
/** /**
* Convert to physical plan to string to enable to print it out in the shell. * Convert to subplan to string for the scheduler to send to the executor
* @param pPhyNode
* @param str
* @return
*/ */
int32_t qPhyPlanToString(struct SQueryDistPlanNode *pPhyNode, char** str); int32_t qSubPlanToString(struct SSubplan *pPhyNode, char** str);
/**
* Destroy the query plan object.
* @return
*/
void* qDestroyQueryPlan(struct SQueryPlanNode* pQueryNode);
/** /**
* Destroy the physical plan. * Destroy the physical plan.
* @param pQueryPhyNode * @param pQueryPhyNode
* @return * @return
*/ */
void* qDestroyQueryPhyPlan(struct SQueryDistPlanNode* pQueryPhyNode); void* qDestroyQueryDag(struct SQueryDag* pDag);
/**
* Create the query job from the physical execution plan
* @param pPhyNode
* @param pJob
* @return
*/
int32_t qCreateQueryJob(const struct SQueryDistPlanNode* pPhyNode, struct SQueryJob** pJob);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
include/libs/scheduler/scheduler.h
浏览文件 @ c7d0bee0
...@@ -20,7 +20,42 @@ ...@@ -20,7 +20,42 @@
extern "C" { extern "C" {
#endif #endif
struct SQueryJob; typedef struct SQueryProfileSummary {
int64_t startTs; // Object created and added into the message queue
int64_t endTs; // the timestamp when the task is completed
int64_t cputime; // total cpu cost, not execute elapsed time
int64_t loadRemoteDataDuration; // remote io time
int64_t loadNativeDataDuration; // native disk io time
uint64_t loadNativeData; // blocks + SMA + header files
uint64_t loadRemoteData; // remote data acquired by exchange operator.
uint64_t waitDuration; // the time to waiting to be scheduled in queue does matter, so we need to record it
int64_t addQTs; // the time to be added into the message queue, used to calculate the waiting duration in queue.
uint64_t totalRows;
uint64_t loadRows;
uint32_t totalBlocks;
uint32_t loadBlocks;
uint32_t loadBlockAgg;
uint32_t skipBlocks;
uint64_t resultSize; // generated result size in Kb.
} SQueryProfileSummary;
typedef struct SQueryTask {
uint64_t queryId; // query id
uint64_t taskId; // task id
char *pSubplan; // operator tree
uint64_t status; // task status
SQueryProfileSummary summary; // task execution summary
void *pOutputHandle; // result buffer handle, to temporarily keep the output result for next stage
} SQueryTask;
typedef struct SQueryJob {
SArray **pSubtasks;
// todo
} SQueryJob;
/** /**
* Process the query job, generated according to the query physical plan. * Process the query job, generated according to the query physical plan.
......
include/libs/wal/wal.h
浏览文件 @ c7d0bee0
...@@ -32,23 +32,19 @@ extern int32_t wDebugFlag; ...@@ -32,23 +32,19 @@ extern int32_t wDebugFlag;
#define wDebug(...) { if (wDebugFlag & DEBUG_DEBUG) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }} #define wDebug(...) { if (wDebugFlag & DEBUG_DEBUG) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }}
#define wTrace(...) { if (wDebugFlag & DEBUG_TRACE) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }} #define wTrace(...) { if (wDebugFlag & DEBUG_TRACE) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }}
#define WAL_PREFIX "wal" #define WAL_HEAD_VER 0
#define WAL_PREFIX_LEN 3
#define WAL_NOSUFFIX_LEN 20 #define WAL_NOSUFFIX_LEN 20
#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN+1) #define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN+1)
#define WAL_LOG_SUFFIX "log" #define WAL_LOG_SUFFIX "log"
#define WAL_INDEX_SUFFIX "idx" #define WAL_INDEX_SUFFIX "idx"
#define WAL_REFRESH_MS 1000 #define WAL_REFRESH_MS 1000
#define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead) + 16) #define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead))
#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) #define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12)
#define WAL_FILE_LEN (WAL_PATH_LEN + 32) #define WAL_FILE_LEN (WAL_PATH_LEN + 32)
#define WAL_IDX_ENTRY_SIZE (sizeof(int64_t)*2) #define WAL_CUR_FAILED 1
#define WAL_CUR_POS_WRITABLE 1
#define WAL_CUR_FILE_WRITABLE 2
#define WAL_CUR_FAILED 4
#pragma pack(push,1) #pragma pack(push, 1)
typedef enum { typedef enum {
TAOS_WAL_NOLOG = 0, TAOS_WAL_NOLOG = 0,
TAOS_WAL_WRITE = 1, TAOS_WAL_WRITE = 1,
...@@ -56,11 +52,11 @@ typedef enum { ...@@ -56,11 +52,11 @@ typedef enum {
} EWalType; } EWalType;
typedef struct SWalReadHead { typedef struct SWalReadHead {
int8_t sver; int8_t headVer;
uint8_t msgType; uint8_t msgType;
int8_t reserved[2]; int8_t reserved[2];
int32_t len; int32_t len;
//int64_t ingestTs; //not implemented int64_t ingestTs; //not implemented
int64_t version; int64_t version;
char body[]; char body[];
} SWalReadHead; } SWalReadHead;
...@@ -72,18 +68,10 @@ typedef struct { ...@@ -72,18 +68,10 @@ typedef struct {
int32_t rollPeriod; // secs int32_t rollPeriod; // secs
int64_t retentionSize; int64_t retentionSize;
int64_t segSize; int64_t segSize;
EWalType level; // wal level EWalType level; // wal level
} SWalCfg; } SWalCfg;
typedef struct { typedef struct {
//union {
//uint32_t info;
//struct {
//uint32_t sver:3;
//uint32_t msgtype: 5;
//uint32_t reserved : 24;
//};
//};
uint32_t cksumHead; uint32_t cksumHead;
uint32_t cksumBody; uint32_t cksumBody;
SWalReadHead head; SWalReadHead head;
...@@ -102,16 +90,16 @@ typedef struct SWal { ...@@ -102,16 +90,16 @@ typedef struct SWal {
SWalCfg cfg; SWalCfg cfg;
SWalVer vers; SWalVer vers;
//file set //file set
int32_t writeCur;
int64_t writeLogTfd; int64_t writeLogTfd;
int64_t writeIdxTfd; int64_t writeIdxTfd;
int32_t writeCur;
SArray* fileInfoSet; SArray* fileInfoSet;
//statistics
int64_t totSize;
int64_t lastRollSeq;
//ctl //ctl
int32_t curStatus;
int32_t fsyncSeq; int32_t fsyncSeq;
int64_t totSize;
int64_t refId; int64_t refId;
int64_t lastRollSeq;
pthread_mutex_t mutex; pthread_mutex_t mutex;
//path //path
char path[WAL_PATH_LEN]; char path[WAL_PATH_LEN];
...@@ -131,7 +119,7 @@ typedef struct SWalReadHandle { ...@@ -131,7 +119,7 @@ typedef struct SWalReadHandle {
} SWalReadHandle; } SWalReadHandle;
#pragma pack(pop) #pragma pack(pop)
typedef int32_t (*FWalWrite)(void *ahandle, void *pHead); //typedef int32_t (*FWalWrite)(void *ahandle, void *pHead);
// module initialization // module initialization
int32_t walInit(); int32_t walInit();
...@@ -151,8 +139,8 @@ int32_t walCommit(SWal *, int64_t ver); ...@@ -151,8 +139,8 @@ int32_t walCommit(SWal *, int64_t ver);
// truncate after // truncate after
int32_t walRollback(SWal *, int64_t ver); int32_t walRollback(SWal *, int64_t ver);
// notify that previous logs can be pruned safely // notify that previous logs can be pruned safely
int32_t walBeginTakeSnapshot(SWal *, int64_t ver); int32_t walBeginSnapshot(SWal *, int64_t ver);
int32_t walEndTakeSnapshot(SWal *); int32_t walEndSnapshot(SWal *);
//int32_t walDataCorrupted(SWal*); //int32_t walDataCorrupted(SWal*);
// read // read
...@@ -161,7 +149,7 @@ void walCloseReadHandle(SWalReadHandle *); ...@@ -161,7 +149,7 @@ void walCloseReadHandle(SWalReadHandle *);
int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver); int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver);
int32_t walRead(SWal *, SWalHead **, int64_t ver); int32_t walRead(SWal *, SWalHead **, int64_t ver);
int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum); //int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum);
// lifecycle check // lifecycle check
int64_t walGetFirstVer(SWal *); int64_t walGetFirstVer(SWal *);
......
include/util/tdlist.h
浏览文件 @ c7d0bee0
...@@ -58,8 +58,8 @@ extern "C" { ...@@ -58,8 +58,8 @@ extern "C" {
// Double linked list // Double linked list
#define TD_DLIST_NODE(TYPE) \ #define TD_DLIST_NODE(TYPE) \
struct { \ struct { \
TYPE *dl_prev_; \ struct TYPE *dl_prev_; \
TYPE *dl_next_; \ struct TYPE *dl_next_; \
} }
#define TD_DLIST(TYPE) \ #define TD_DLIST(TYPE) \
......
include/util/tmacro.h
浏览文件 @ c7d0bee0
...@@ -29,13 +29,11 @@ extern "C" { ...@@ -29,13 +29,11 @@ extern "C" {
#define TD_MOD_UNCLEARD 0 #define TD_MOD_UNCLEARD 0
#define TD_MOD_CLEARD 1 #define TD_MOD_CLEARD 1
#define TD_DEF_MOD_INIT_FLAG(MOD) static int8_t MOD##InitFlag = TD_MOD_UNINITIALIZED typedef int8_t td_mode_flag_t;
#define TD_DEF_MOD_CLEAR_FLAG(MOD) static int8_t MOD##ClearFlag = TD_MOD_UNCLEARD
#define TD_CHECK_AND_SET_MODE_INIT(MOD) \ #define TD_CHECK_AND_SET_MODE_INIT(FLAG) atomic_val_compare_exchange_8((FLAG), TD_MOD_UNINITIALIZED, TD_MOD_INITIALIZED)
atomic_val_compare_exchange_8(&(MOD##InitFlag), TD_MOD_UNINITIALIZED, TD_MOD_INITIALIZED)
#define TD_CHECK_AND_SET_MOD_CLEAR(MOD) atomic_val_compare_exchange_8(&(MOD##ClearFlag), TD_MOD_UNCLEARD, TD_MOD_CLEARD) #define TD_CHECK_AND_SET_MOD_CLEAR(FLAG) atomic_val_compare_exchange_8((FLAG), TD_MOD_UNCLEARD, TD_MOD_CLEARD)
#ifdef __cplusplus #ifdef __cplusplus
} }
......
include/util/tref.h
浏览文件 @ c7d0bee0
...@@ -17,6 +17,8 @@ ...@@ -17,6 +17,8 @@
#ifndef _TD_UTIL_REF_H #ifndef _TD_UTIL_REF_H
#define _TD_UTIL_REF_H #define _TD_UTIL_REF_H
#include "os.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
......
source/dnode/mgmt/impl/test/dnode/dnode.cpp
浏览文件 @ c7d0bee0
...@@ -379,41 +379,41 @@ TEST_F(DndTestDnode, RestartDnode_01) { ...@@ -379,41 +379,41 @@ TEST_F(DndTestDnode, RestartDnode_01) {
const char* fqdn = "localhost"; const char* fqdn = "localhost";
const char* firstEp = "localhost:9521"; const char* firstEp = "localhost:9521";
pServer1 = startServer("/tmp/dndTestDnode1", fqdn, 9521, firstEp); pServer1 = startServer("/tmp/dndTestDnode1", fqdn, 9521, firstEp);
// pServer1 = startServer("/tmp/dndTestDnode3", fqdn, 9523, firstEp); pServer3 = startServer("/tmp/dndTestDnode3", fqdn, 9523, firstEp);
// pServer1 = startServer("/tmp/dndTestDnode4", fqdn, 9524, firstEp); pServer4 = startServer("/tmp/dndTestDnode4", fqdn, 9524, firstEp);
// pServer1 = startServer("/tmp/dndTestDnode5", fqdn, 9525, firstEp); pServer5 = startServer("/tmp/dndTestDnode5", fqdn, 9525, firstEp);
uInfo("all server is running"); uInfo("all server is running");
// taosMsleep(1300); taosMsleep(1300);
// SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "show dnodes", 7); SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "show dnodes", 7);
// SendThenCheckShowRetrieveMsg(4); SendThenCheckShowRetrieveMsg(4);
// CheckInt16(1); CheckInt16(1);
// CheckInt16(3); CheckInt16(3);
// CheckInt16(4); CheckInt16(4);
// CheckInt16(5); CheckInt16(5);
// CheckBinary("localhost:9521", TSDB_EP_LEN); CheckBinary("localhost:9521", TSDB_EP_LEN);
// CheckBinary("localhost:9523", TSDB_EP_LEN); CheckBinary("localhost:9523", TSDB_EP_LEN);
// CheckBinary("localhost:9524", TSDB_EP_LEN); CheckBinary("localhost:9524", TSDB_EP_LEN);
// CheckBinary("localhost:9525", TSDB_EP_LEN); CheckBinary("localhost:9525", TSDB_EP_LEN);
// CheckInt16(0); CheckInt16(0);
// CheckInt16(0); CheckInt16(0);
// CheckInt16(0); CheckInt16(0);
// CheckInt16(0); CheckInt16(0);
// CheckInt16(1); CheckInt16(1);
// CheckInt16(1); CheckInt16(1);
// CheckInt16(1); CheckInt16(1);
// CheckInt16(1); CheckInt16(1);
// CheckBinary("ready", 10); CheckBinary("ready", 10);
// CheckBinary("ready", 10); CheckBinary("ready", 10);
// CheckBinary("ready", 10); CheckBinary("ready", 10);
// CheckBinary("ready", 10); CheckBinary("ready", 10);
// CheckTimestamp(); CheckTimestamp();
// CheckTimestamp(); CheckTimestamp();
// CheckTimestamp(); CheckTimestamp();
// CheckTimestamp(); CheckTimestamp();
// CheckBinary("", 24); CheckBinary("", 24);
// CheckBinary("", 24); CheckBinary("", 24);
// CheckBinary("", 24); CheckBinary("", 24);
// CheckBinary("", 24); CheckBinary("", 24);
} }
source/dnode/mgmt/impl/test/sut/deploy.cpp
浏览文件 @ c7d0bee0
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
void initLog(const char* path) { void initLog(const char* path) {
dDebugFlag = 143; dDebugFlag = 143;
vDebugFlag = 0; vDebugFlag = 0;
mDebugFlag = 143; mDebugFlag = 207;
cDebugFlag = 0; cDebugFlag = 0;
jniDebugFlag = 0; jniDebugFlag = 0;
tmrDebugFlag = 0; tmrDebugFlag = 0;
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ c7d0bee0
...@@ -73,7 +73,8 @@ typedef enum { ...@@ -73,7 +73,8 @@ typedef enum {
TRN_STAGE_EXECUTE = 2, TRN_STAGE_EXECUTE = 2,
TRN_STAGE_COMMIT = 3, TRN_STAGE_COMMIT = 3,
TRN_STAGE_ROLLBACK = 4, TRN_STAGE_ROLLBACK = 4,
TRN_STAGE_RETRY = 5 TRN_STAGE_RETRY = 5,
TRN_STAGE_OVER = 6,
} ETrnStage; } ETrnStage;
typedef enum { TRN_POLICY_ROLLBACK = 1, TRN_POLICY_RETRY = 2 } ETrnPolicy; typedef enum { TRN_POLICY_ROLLBACK = 1, TRN_POLICY_RETRY = 2 } ETrnPolicy;
...@@ -103,7 +104,6 @@ typedef struct STrans { ...@@ -103,7 +104,6 @@ typedef struct STrans {
int32_t id; int32_t id;
ETrnStage stage; ETrnStage stage;
ETrnPolicy policy; ETrnPolicy policy;
SMnode *pMnode;
void *rpcHandle; void *rpcHandle;
SArray *redoLogs; SArray *redoLogs;
SArray *undoLogs; SArray *undoLogs;
...@@ -304,6 +304,7 @@ typedef struct SMnodeMsg { ...@@ -304,6 +304,7 @@ typedef struct SMnodeMsg {
typedef struct { typedef struct {
int32_t id; int32_t id;
int32_t code;
void *rpcHandle; void *rpcHandle;
} STransMsg; } STransMsg;
......
source/dnode/mnode/impl/inc/mndSync.h
浏览文件 @ c7d0bee0
...@@ -25,7 +25,7 @@ extern "C" { ...@@ -25,7 +25,7 @@ extern "C" {
int32_t mndInitSync(SMnode *pMnode); int32_t mndInitSync(SMnode *pMnode);
void mndCleanupSync(SMnode *pMnode); void mndCleanupSync(SMnode *pMnode);
bool mndIsMaster(SMnode *pMnode); bool mndIsMaster(SMnode *pMnode);
int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg); int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mnode/impl/inc/mndTrans.h
浏览文件 @ c7d0bee0
...@@ -32,10 +32,10 @@ int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw); ...@@ -32,10 +32,10 @@ int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *, void *pMsg); int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *, void *pMsg);
int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *, void *pMsg); int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *, void *pMsg);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
int32_t mndTransPrepare(STrans *pTrans);
void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code); void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code);
int32_t mndTransExecute(SSdb *pSdb, int32_t tranId); char *mndTransStageStr(ETrnStage stage);
char *mndTransPolicyStr(ETrnPolicy policy);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ c7d0bee0
...@@ -357,7 +357,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDbMsg *pCreat ...@@ -357,7 +357,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDbMsg *pCreat
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -491,7 +491,7 @@ static int32_t mndUpdateDb(SMnode *pMnode, SMnodeMsg *pMsg, SDbObj *pOldDb, SDbO ...@@ -491,7 +491,7 @@ static int32_t mndUpdateDb(SMnode *pMnode, SMnodeMsg *pMsg, SDbObj *pOldDb, SDbO
} }
sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY); sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -571,7 +571,7 @@ static int32_t mndDropDb(SMnode *pMnode, SMnodeMsg *pMsg, SDbObj *pDb) { ...@@ -571,7 +571,7 @@ static int32_t mndDropDb(SMnode *pMnode, SMnodeMsg *pMsg, SDbObj *pDb) {
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @ c7d0bee0
...@@ -413,25 +413,9 @@ static int32_t mndCreateDnode(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDnodeMsg * ...@@ -413,25 +413,9 @@ static int32_t mndCreateDnode(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDnodeMsg *
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING); sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
SSdbRaw *pUndoRaw = mndDnodeActionEncode(&dnodeObj); if (mndTransPrepare(pMnode, pTrans) != 0) {
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) {
mError("trans:%d, failed to append undo log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED);
SSdbRaw *pCommitRaw = mndDnodeActionEncode(&dnodeObj);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -485,25 +469,9 @@ static int32_t mndDropDnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode) ...@@ -485,25 +469,9 @@ static int32_t mndDropDnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode)
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING); sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED);
SSdbRaw *pUndoRaw = mndDnodeActionEncode(pDnode);
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) {
mError("trans:%d, failed to append undo log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
SSdbRaw *pCommitRaw = mndDnodeActionEncode(pDnode);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/impl/src/mndFunc.c
浏览文件 @ c7d0bee0
...@@ -183,7 +183,7 @@ static int32_t mndCreateFunc(SMnode *pMnode, SMnodeMsg *pMsg, SCreateFuncMsg *pC ...@@ -183,7 +183,7 @@ static int32_t mndCreateFunc(SMnode *pMnode, SMnodeMsg *pMsg, SCreateFuncMsg *pC
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -226,7 +226,7 @@ static int32_t mndDropFunc(SMnode *pMnode, SMnodeMsg *pMsg, SFuncObj *pFunc) { ...@@ -226,7 +226,7 @@ static int32_t mndDropFunc(SMnode *pMnode, SMnodeMsg *pMsg, SFuncObj *pFunc) {
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ c7d0bee0
...@@ -238,7 +238,7 @@ static int32_t mndCreateMnode(SMnode *pMnode, SMnodeMsg *pMsg, SCreateMnodeMsg * ...@@ -238,7 +238,7 @@ static int32_t mndCreateMnode(SMnode *pMnode, SMnodeMsg *pMsg, SCreateMnodeMsg *
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -313,7 +313,7 @@ static int32_t mndDropMnode(SMnode *pMnode, SMnodeMsg *pMsg, SMnodeObj *pMnodeOb ...@@ -313,7 +313,7 @@ static int32_t mndDropMnode(SMnode *pMnode, SMnodeMsg *pMsg, SMnodeObj *pMnodeOb
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ c7d0bee0
...@@ -285,7 +285,7 @@ static int32_t mndCreateStb(SMnode *pMnode, SMnodeMsg *pMsg, SCreateStbMsg *pCre ...@@ -285,7 +285,7 @@ static int32_t mndCreateStb(SMnode *pMnode, SMnodeMsg *pMsg, SCreateStbMsg *pCre
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -433,7 +433,7 @@ static int32_t mndDropStb(SMnode *pMnode, SMnodeMsg *pMsg, SStbObj *pStb) { ...@@ -433,7 +433,7 @@ static int32_t mndDropStb(SMnode *pMnode, SMnodeMsg *pMsg, SStbObj *pStb) {
} }
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/impl/src/mndSync.c
浏览文件 @ c7d0bee0
...@@ -21,16 +21,16 @@ ...@@ -21,16 +21,16 @@
int32_t mndInitSync(SMnode *pMnode) { return 0; } int32_t mndInitSync(SMnode *pMnode) { return 0; }
void mndCleanupSync(SMnode *pMnode) {} void mndCleanupSync(SMnode *pMnode) {}
int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg) { int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw) {
int32_t code = 0; int32_t code = 0;
int32_t len = sdbGetRawTotalSize(pRaw); // int32_t len = sdbGetRawTotalSize(pRaw);
SSdbRaw *pReceived = calloc(1, len); // SSdbRaw *pReceived = calloc(1, len);
memcpy(pReceived, pRaw, len); // memcpy(pReceived, pRaw, len);
mDebug("trans:%d, data:%p recv from sync, code:0x%x pMsg:%p", pMsg->id, pReceived, code & 0xFFFF, pMsg); // mDebug("trans:%d, data:%p recv from sync, code:0x%x pMsg:%p", pMsg->id, pReceived, code & 0xFFFF, pMsg);
mndTransApply(pMnode, pReceived, pMsg, code); // mndTransApply(pMnode, pReceived, code);
return 0; return code;
} }
bool mndIsMaster(SMnode *pMnode) { return true; } bool mndIsMaster(SMnode *pMnode) { return true; }
\ No newline at end of file
source/dnode/mnode/impl/src/mndUser.c
浏览文件 @ c7d0bee0
...@@ -207,25 +207,9 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, char *user, char *pass, ...@@ -207,25 +207,9 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, char *user, char *pass,
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING); sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
SSdbRaw *pUndoRaw = mndUserActionEncode(&userObj);
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) {
mError("trans:%d, failed to append undo log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED);
SSdbRaw *pCommitRaw = mndUserActionEncode(&userObj);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -251,15 +235,7 @@ static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOldUser, SUserObj *pNewU ...@@ -251,15 +235,7 @@ static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOldUser, SUserObj *pNewU
} }
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY); sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
SSdbRaw *pUndoRaw = mndUserActionEncode(pOldUser); if (mndTransPrepare(pMnode, pTrans) != 0) {
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) {
mError("trans:%d, failed to append undo log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
if (mndTransPrepare(pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
...@@ -283,25 +259,9 @@ static int32_t mndDropUser(SMnode *pMnode, SMnodeMsg *pMsg, SUserObj *pUser) { ...@@ -283,25 +259,9 @@ static int32_t mndDropUser(SMnode *pMnode, SMnodeMsg *pMsg, SUserObj *pUser) {
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING); sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED);
SSdbRaw *pUndoRaw = mndUserActionEncode(pUser);
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) {
mError("trans:%d, failed to append undo log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
SSdbRaw *pCommitRaw = mndUserActionEncode(pUser);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
......
source/dnode/mnode/sdb/src/sdbHash.c
浏览文件 @ c7d0bee0
...@@ -132,11 +132,6 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * ...@@ -132,11 +132,6 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
taosHashRemove(hash, pOldRow->pObj, keySize); taosHashRemove(hash, pOldRow->pObj, keySize);
taosWUnLockLatch(pLock); taosWUnLockLatch(pLock);
SdbDeleteFp deleteFp = pSdb->deleteFps[pOldRow->type];
if (deleteFp != NULL) {
code = (*deleteFp)(pSdb, pOldRow->pObj);
}
sdbRelease(pSdb, pOldRow->pObj); sdbRelease(pSdb, pOldRow->pObj);
sdbFreeRow(pRow); sdbFreeRow(pRow);
return code; return code;
...@@ -161,6 +156,7 @@ int32_t sdbWriteNotFree(SSdb *pSdb, SSdbRaw *pRaw) { ...@@ -161,6 +156,7 @@ int32_t sdbWriteNotFree(SSdb *pSdb, SSdbRaw *pRaw) {
case SDB_STATUS_CREATING: case SDB_STATUS_CREATING:
code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize); code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize);
break; break;
case SDB_STATUS_UPDATING:
case SDB_STATUS_READY: case SDB_STATUS_READY:
case SDB_STATUS_DROPPING: case SDB_STATUS_DROPPING:
code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize); code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize);
...@@ -228,6 +224,11 @@ void sdbRelease(SSdb *pSdb, void *pObj) { ...@@ -228,6 +224,11 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1); int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) { if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) {
SdbDeleteFp deleteFp = pSdb->deleteFps[pRow->type];
if (deleteFp != NULL) {
(*deleteFp)(pSdb, pRow->pObj);
}
sdbFreeRow(pRow); sdbFreeRow(pRow);
} }
......
source/dnode/vnode/impl/inc/vnodeBufferPool.h
浏览文件 @ c7d0bee0
...@@ -27,6 +27,8 @@ typedef struct SVBufPool SVBufPool; ...@@ -27,6 +27,8 @@ typedef struct SVBufPool SVBufPool;
int vnodeOpenBufPool(SVnode *pVnode); int vnodeOpenBufPool(SVnode *pVnode);
void vnodeCloseBufPool(SVnode *pVnode); void vnodeCloseBufPool(SVnode *pVnode);
int vnodeBufPoolSwitch(SVnode *pVnode);
int vnodeBufPoolRecycle(SVnode *pVnode);
void *vnodeMalloc(SVnode *pVnode, uint64_t size); void *vnodeMalloc(SVnode *pVnode, uint64_t size);
bool vnodeBufPoolIsFull(SVnode *pVnode); bool vnodeBufPoolIsFull(SVnode *pVnode);
......
source/dnode/vnode/impl/inc/vnodeCommit.h
浏览文件 @ c7d0bee0
...@@ -24,6 +24,7 @@ extern "C" { ...@@ -24,6 +24,7 @@ extern "C" {
#define vnodeShouldCommit vnodeBufPoolIsFull #define vnodeShouldCommit vnodeBufPoolIsFull
int vnodeAsyncCommit(SVnode *pVnode); int vnodeAsyncCommit(SVnode *pVnode);
int vnodeCommit(void *arg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/vnode/impl/inc/vnodeDef.h
浏览文件 @ c7d0bee0
...@@ -21,9 +21,11 @@ ...@@ -21,9 +21,11 @@
#include "tcoding.h" #include "tcoding.h"
#include "tdlist.h" #include "tdlist.h"
#include "tlockfree.h" #include "tlockfree.h"
#include "tmacro.h"
#include "wal.h" #include "wal.h"
#include "vnode.h" #include "vnode.h"
#include "vnodeBufferPool.h" #include "vnodeBufferPool.h"
#include "vnodeCfg.h" #include "vnodeCfg.h"
#include "vnodeCommit.h" #include "vnodeCommit.h"
...@@ -37,6 +39,27 @@ ...@@ -37,6 +39,27 @@
extern "C" { extern "C" {
#endif #endif
typedef struct SVnodeTask {
TD_DLIST_NODE(SVnodeTask);
void* arg;
int (*execute)(void*);
} SVnodeTask;
typedef struct SVnodeMgr {
td_mode_flag_t vnodeInitFlag;
td_mode_flag_t vnodeClearFlag;
// For commit
bool stop;
uint16_t nthreads;
pthread_t* threads;
pthread_mutex_t mutex;
pthread_cond_t hasTask;
TD_DLIST(SVnodeTask) queue;
// For vnode Mgmt
} SVnodeMgr;
extern SVnodeMgr vnodeMgr;
struct SVnode { struct SVnode {
char* path; char* path;
SVnodeCfg config; SVnodeCfg config;
...@@ -50,6 +73,8 @@ struct SVnode { ...@@ -50,6 +73,8 @@ struct SVnode {
SVnodeFS* pFs; SVnodeFS* pFs;
}; };
int vnodeScheduleTask(SVnodeTask* task);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
source/dnode/vnode/impl/inc/vnodeMemAllocator.h
浏览文件 @ c7d0bee0
...@@ -33,6 +33,7 @@ struct SVArenaNode { ...@@ -33,6 +33,7 @@ struct SVArenaNode {
}; };
struct SVMemAllocator { struct SVMemAllocator {
T_REF_DECLARE()
TD_DLIST_NODE(SVMemAllocator); TD_DLIST_NODE(SVMemAllocator);
uint64_t capacity; uint64_t capacity;
uint64_t ssize; uint64_t ssize;
......
source/dnode/vnode/impl/src/vnodeBufferPool.c
浏览文件 @ c7d0bee0
...@@ -19,6 +19,8 @@ ...@@ -19,6 +19,8 @@
#define VNODE_BUF_POOL_SHARDS 3 #define VNODE_BUF_POOL_SHARDS 3
struct SVBufPool { struct SVBufPool {
pthread_mutex_t mutex;
pthread_cond_t hasFree;
TD_DLIST(SVMemAllocator) free; TD_DLIST(SVMemAllocator) free;
TD_DLIST(SVMemAllocator) incycle; TD_DLIST(SVMemAllocator) incycle;
SVMemAllocator *inuse; SVMemAllocator *inuse;
...@@ -28,7 +30,6 @@ struct SVBufPool { ...@@ -28,7 +30,6 @@ struct SVBufPool {
int vnodeOpenBufPool(SVnode *pVnode) { int vnodeOpenBufPool(SVnode *pVnode) {
uint64_t capacity; uint64_t capacity;
// EVMemAllocatorT type = E_V_ARENA_ALLOCATOR;
if ((pVnode->pBufPool = (SVBufPool *)calloc(1, sizeof(SVBufPool))) == NULL) { if ((pVnode->pBufPool = (SVBufPool *)calloc(1, sizeof(SVBufPool))) == NULL) {
/* TODO */ /* TODO */
...@@ -79,6 +80,27 @@ void vnodeCloseBufPool(SVnode *pVnode) { ...@@ -79,6 +80,27 @@ void vnodeCloseBufPool(SVnode *pVnode) {
} }
} }
int vnodeBufPoolSwitch(SVnode *pVnode) {
SVMemAllocator *pvma = pVnode->pBufPool->inuse;
pVnode->pBufPool->inuse = NULL;
tDListAppend(&(pVnode->pBufPool->incycle), pvma);
return 0;
}
int vnodeBufPoolRecycle(SVnode *pVnode) {
SVBufPool * pBufPool = pVnode->pBufPool;
SVMemAllocator *pvma = TD_DLIST_HEAD(&(pBufPool->incycle));
ASSERT(pvma != NULL);
tDListPop(&(pBufPool->incycle), pvma);
vmaReset(pvma);
tDListAppend(&(pBufPool->free), pvma);
return 0;
}
void *vnodeMalloc(SVnode *pVnode, uint64_t size) { void *vnodeMalloc(SVnode *pVnode, uint64_t size) {
SVBufPool *pBufPool = pVnode->pBufPool; SVBufPool *pBufPool = pVnode->pBufPool;
...@@ -89,6 +111,8 @@ void *vnodeMalloc(SVnode *pVnode, uint64_t size) { ...@@ -89,6 +111,8 @@ void *vnodeMalloc(SVnode *pVnode, uint64_t size) {
if (pBufPool->inuse) { if (pBufPool->inuse) {
tDListPop(&(pBufPool->free), pBufPool->inuse); tDListPop(&(pBufPool->free), pBufPool->inuse);
break; break;
} else {
// tsem_wait(&(pBufPool->hasFree));
} }
} }
} }
......
source/dnode/vnode/impl/src/vnodeCfg.c
浏览文件 @ c7d0bee0
...@@ -15,7 +15,8 @@ ...@@ -15,7 +15,8 @@
#include "vnodeDef.h" #include "vnodeDef.h"
const SVnodeCfg defaultVnodeOptions = {.wsize = 16 * 1024 * 1024, .walCfg = {.level = TAOS_WAL_WRITE}}; /* TODO */ const SVnodeCfg defaultVnodeOptions = {
.wsize = 96 * 1024 * 1024, .ssize = 1 * 1024 * 1024, .lsize = 1024, .walCfg = {.level = TAOS_WAL_WRITE}}; /* TODO */
void vnodeOptionsInit(SVnodeCfg *pVnodeOptions) { /* TODO */ void vnodeOptionsInit(SVnodeCfg *pVnodeOptions) { /* TODO */
vnodeOptionsCopy(pVnodeOptions, &defaultVnodeOptions); vnodeOptionsCopy(pVnodeOptions, &defaultVnodeOptions);
......
source/dnode/vnode/impl/src/vnodeCommit.c
浏览文件 @ c7d0bee0
...@@ -19,28 +19,21 @@ static int vnodeStartCommit(SVnode *pVnode); ...@@ -19,28 +19,21 @@ static int vnodeStartCommit(SVnode *pVnode);
static int vnodeEndCommit(SVnode *pVnode); static int vnodeEndCommit(SVnode *pVnode);
int vnodeAsyncCommit(SVnode *pVnode) { int vnodeAsyncCommit(SVnode *pVnode) {
#if 0 vnodeBufPoolSwitch(pVnode);
if (vnodeStartCommit(pVnode) < 0) { SVnodeTask *pTask = (SVnodeTask *)malloc(sizeof(*pTask));
// TODO
}
if (tqCommit(pVnode->pTQ) < 0) { pTask->execute = vnodeCommit; // TODO
// TODO pTask->arg = pVnode; // TODO
}
if (metaCommit(pVnode->pMeta) < 0) { vnodeScheduleTask(pTask);
// TODO return 0;
} }
if (tsdbCommit(pVnode->pTsdb) < 0) {
// TODO
}
if (vnodeEndCommit(pVnode) < 0) { int vnodeCommit(void *arg) {
// TODO SVnode *pVnode = (SVnode *)arg;
}
#endif vnodeBufPoolRecycle(pVnode);
// TODO
return 0; return 0;
} }
......
source/dnode/vnode/impl/src/vnodeMain.c
浏览文件 @ c7d0bee0
...@@ -13,7 +13,6 @@ ...@@ -13,7 +13,6 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "tmacro.h"
#include "vnodeDef.h" #include "vnodeDef.h"
static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg); static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg);
...@@ -21,29 +20,6 @@ static void vnodeFree(SVnode *pVnode); ...@@ -21,29 +20,6 @@ static void vnodeFree(SVnode *pVnode);
static int vnodeOpenImpl(SVnode *pVnode); static int vnodeOpenImpl(SVnode *pVnode);
static void vnodeCloseImpl(SVnode *pVnode); static void vnodeCloseImpl(SVnode *pVnode);
TD_DEF_MOD_INIT_FLAG(vnode);
TD_DEF_MOD_CLEAR_FLAG(vnode);
int vnodeInit() {
if (TD_CHECK_AND_SET_MODE_INIT(vnode) == TD_MOD_INITIALIZED) {
return 0;
}
if (walInit() < 0) {
return -1;
}
return 0;
}
void vnodeClear() {
if (TD_CHECK_AND_SET_MOD_CLEAR(vnode) == TD_MOD_CLEARD) {
return;
}
walCleanUp();
}
SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) { SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
SVnode *pVnode = NULL; SVnode *pVnode = NULL;
......
source/dnode/vnode/impl/src/vnodeMgr.c 0 → 100644
浏览文件 @ c7d0bee0
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "vnodeDef.h"
SVnodeMgr vnodeMgr = {.vnodeInitFlag = TD_MOD_UNINITIALIZED, .vnodeClearFlag = TD_MOD_UNCLEARD, .stop = false};
static void* loop(void* arg);
int vnodeInit(uint16_t nthreads) {
if (TD_CHECK_AND_SET_MODE_INIT(&(vnodeMgr.vnodeInitFlag)) == TD_MOD_INITIALIZED) {
return 0;
}
// Start commit handers
if (nthreads > 0) {
vnodeMgr.nthreads = nthreads;
vnodeMgr.threads = (pthread_t*)calloc(nthreads, sizeof(pthread_t));
if (vnodeMgr.threads == NULL) {
return -1;
}
pthread_mutex_init(&(vnodeMgr.mutex), NULL);
pthread_cond_init(&(vnodeMgr.hasTask), NULL);
tDListInit(&(vnodeMgr.queue));
for (uint16_t i = 0; i < nthreads; i++) {
pthread_create(&(vnodeMgr.threads[i]), NULL, loop, NULL);
}
} else {
// TODO: if no commit thread is set, then another mechanism should be
// given. Otherwise, it is a false.
ASSERT(0);
}
if (walInit() < 0) {
return -1;
}
return 0;
}
void vnodeClear() {
if (TD_CHECK_AND_SET_MOD_CLEAR(&(vnodeMgr.vnodeClearFlag)) == TD_MOD_CLEARD) {
return;
}
walCleanUp();
// Stop commit handler
pthread_mutex_lock(&(vnodeMgr.mutex));
vnodeMgr.stop = true;
pthread_cond_broadcast(&(vnodeMgr.hasTask));
pthread_mutex_unlock(&(vnodeMgr.mutex));
for (uint16_t i = 0; i < vnodeMgr.nthreads; i++) {
pthread_join(vnodeMgr.threads[i], NULL);
}
tfree(vnodeMgr.threads);
pthread_cond_destroy(&(vnodeMgr.hasTask));
pthread_mutex_destroy(&(vnodeMgr.mutex));
}
int vnodeScheduleTask(SVnodeTask* pTask) {
pthread_mutex_lock(&(vnodeMgr.mutex));
tDListAppend(&(vnodeMgr.queue), pTask);
pthread_cond_signal(&(vnodeMgr.hasTask));
pthread_mutex_unlock(&(vnodeMgr.mutex));
return 0;
}
/* ------------------------ STATIC METHODS ------------------------ */
static void* loop(void* arg) {
SVnodeTask* pTask;
for (;;) {
pthread_mutex_lock(&(vnodeMgr.mutex));
for (;;) {
pTask = TD_DLIST_HEAD(&(vnodeMgr.queue));
if (pTask == NULL) {
if (vnodeMgr.stop) {
pthread_mutex_unlock(&(vnodeMgr.mutex));
return NULL;
} else {
pthread_cond_wait(&(vnodeMgr.hasTask), &(vnodeMgr.mutex));
}
} else {
tDListPop(&(vnodeMgr.queue), pTask);
break;
}
}
pthread_mutex_unlock(&(vnodeMgr.mutex));
(*(pTask->execute))(pTask->arg);
}
return NULL;
}
\ No newline at end of file
source/dnode/vnode/impl/test/vnodeApiTests.cpp
浏览文件 @ c7d0bee0
...@@ -92,7 +92,7 @@ TEST(vnodeApiTest, test_create_table_encode_and_decode_function) { ...@@ -92,7 +92,7 @@ TEST(vnodeApiTest, test_create_table_encode_and_decode_function) {
#endif #endif
TEST(vnodeApiTest, vnodeOpen_vnodeClose_test) { TEST(vnodeApiTest, vnodeOpen_vnodeClose_test) {
GTEST_ASSERT_GE(vnodeInit(), 0); GTEST_ASSERT_GE(vnodeInit(2), 0);
// Create and open a vnode // Create and open a vnode
SVnode *pVnode = vnodeOpen("vnode1", NULL); SVnode *pVnode = vnodeOpen("vnode1", NULL);
......
source/libs/index/inc/index_fst_automation.h
浏览文件 @ c7d0bee0
...@@ -20,6 +20,8 @@ extern "C" { ...@@ -20,6 +20,8 @@ extern "C" {
#endif #endif
#include "index_fst_util.h" #include "index_fst_util.h"
typedef struct AutomationCtx AutomationCtx; typedef struct AutomationCtx AutomationCtx;
typedef enum AutomationType { typedef enum AutomationType {
...@@ -38,18 +40,30 @@ typedef struct Complement { ...@@ -38,18 +40,30 @@ typedef struct Complement {
// automation // automation
typedef struct AutomationCtx { typedef struct AutomationCtx {
AutomationType type; AutomationType type;
void *data; void *stdata;
char *data;
} AutomationCtx; } AutomationCtx;
typedef enum ValueType { FST_INT, FST_CHAR, FST_ARRAY} ValueType;
typedef enum StartWithStateKind { Done, Running } StartWithStateKind; typedef enum StartWithStateKind { Done, Running } StartWithStateKind;
typedef struct StartWithStateValue { typedef struct StartWithStateValue {
StartWithStateKind kind; StartWithStateKind kind;
void *value; ValueType type;
union {
int val;
char *ptr;
SArray *arr;
// add more type
} ;
} StartWithStateValue; } StartWithStateValue;
StartWithStateValue *startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void *val);
StartWithStateValue *startWithStateValueDump(StartWithStateValue *sv);
void startWithStateValueDestroy(void *sv);
typedef struct AutomationFunc { typedef struct AutomationFunc {
void* (*start)(AutomationCtx *ctx) ; void* (*start)(AutomationCtx *ctx) ;
bool (*isMatch)(AutomationCtx *ctx, void *); bool (*isMatch)(AutomationCtx *ctx, void *);
...@@ -59,7 +73,7 @@ typedef struct AutomationFunc { ...@@ -59,7 +73,7 @@ typedef struct AutomationFunc {
void* (*acceptEof)(AutomationCtx *ct, void *state); void* (*acceptEof)(AutomationCtx *ct, void *state);
} AutomationFunc; } AutomationFunc;
AutomationCtx *automCtxCreate(void *data, AutomationType type); AutomationCtx *automCtxCreate(void *data, AutomationType atype);
void automCtxDestroy(AutomationCtx *ctx); void automCtxDestroy(AutomationCtx *ctx);
extern AutomationFunc automFuncs[]; extern AutomationFunc automFuncs[];
......
source/libs/index/src/index_fst.c
浏览文件 @ c7d0bee0
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
#include "tcoding.h" #include "tcoding.h"
#include "tchecksum.h" #include "tchecksum.h"
#include "indexInt.h" #include "indexInt.h"
#include "index_fst_automation.h"
static void fstPackDeltaIn(FstCountingWriter *wrt, CompiledAddr nodeAddr, CompiledAddr transAddr, uint8_t nBytes) { static void fstPackDeltaIn(FstCountingWriter *wrt, CompiledAddr nodeAddr, CompiledAddr transAddr, uint8_t nBytes) {
...@@ -1322,6 +1323,7 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb ...@@ -1322,6 +1323,7 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
return swsResultCreate(&s, output, callback(start)); return swsResultCreate(&s, output, callback(start));
} }
} }
SArray *nodes = taosArrayInit(8, sizeof(FstNode *));
while (taosArrayGetSize(sws->stack) > 0) { while (taosArrayGetSize(sws->stack) > 0) {
StreamState *p = (StreamState *)taosArrayPop(sws->stack); StreamState *p = (StreamState *)taosArrayPop(sws->stack);
if (p->trans >= FST_NODE_LEN(p->node) || automFuncs[aut->type].canMatch(aut, p->autState)) { if (p->trans >= FST_NODE_LEN(p->node) || automFuncs[aut->type].canMatch(aut, p->autState)) {
...@@ -1337,8 +1339,8 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb ...@@ -1337,8 +1339,8 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
void* nextState = automFuncs[aut->type].accept(aut, p->autState, trn.inp); void* nextState = automFuncs[aut->type].accept(aut, p->autState, trn.inp);
void* tState = callback(nextState); void* tState = callback(nextState);
bool isMatch = automFuncs[aut->type].isMatch(aut, nextState); bool isMatch = automFuncs[aut->type].isMatch(aut, nextState);
//bool isMatch = sws->aut->isMatch(nextState);
FstNode *nextNode = fstGetNode(sws->fst, trn.addr); FstNode *nextNode = fstGetNode(sws->fst, trn.addr);
taosArrayPush(nodes, &nextNode);
taosArrayPush(sws->inp, &(trn.inp)); taosArrayPush(sws->inp, &(trn.inp));
if (FST_NODE_IS_FINAL(nextNode)) { if (FST_NODE_IS_FINAL(nextNode)) {
...@@ -1354,26 +1356,35 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb ...@@ -1354,26 +1356,35 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState}; StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState};
taosArrayPush(sws->stack, &s2); taosArrayPush(sws->stack, &s2);
uint8_t *buf = (uint8_t *)malloc(taosArrayGetSize(sws->inp) * sizeof(uint8_t));
for (uint32_t i = 0; i < taosArrayGetSize(sws->inp); i++) { size_t isz = taosArrayGetSize(sws->inp);
uint8_t *t = (uint8_t *)taosArrayGet(sws->inp, i); uint8_t *buf = (uint8_t *)malloc(isz * sizeof(uint8_t));
buf[i] = *t; for (uint32_t i = 0; i < isz; i++) {
buf[i] = *(uint8_t *)taosArrayGet(sws->inp, i);
} }
FstSlice slice = fstSliceCreate(buf, taosArrayGetSize(sws->inp)); FstSlice slice = fstSliceCreate(buf, taosArrayGetSize(sws->inp));
if (fstBoundWithDataExceededBy(sws->endAt, &slice)) { if (fstBoundWithDataExceededBy(sws->endAt, &slice)) {
taosArrayDestroyEx(sws->stack, streamStateDestroy); taosArrayDestroyEx(sws->stack, streamStateDestroy);
sws->stack = (SArray *)taosArrayInit(256, sizeof(StreamState)); sws->stack = (SArray *)taosArrayInit(256, sizeof(StreamState));
free(buf);
fstSliceDestroy(&slice); fstSliceDestroy(&slice);
return NULL; return NULL;
} }
if (FST_NODE_IS_FINAL(nextNode) && isMatch) { if (FST_NODE_IS_FINAL(nextNode) && isMatch) {
FstOutput fOutput = {.null = false, .out = out + FST_NODE_FINAL_OUTPUT(nextNode)}; FstOutput fOutput = {.null = false, .out = out + FST_NODE_FINAL_OUTPUT(nextNode)};
StreamWithStateResult *result = swsResultCreate(&slice, fOutput , tState); StreamWithStateResult *result = swsResultCreate(&slice, fOutput, tState);
free(buf);
fstSliceDestroy(&slice); fstSliceDestroy(&slice);
return result; return result;
} }
free(buf);
fstSliceDestroy(&slice); fstSliceDestroy(&slice);
} }
for (size_t i = 0; i < taosArrayGetSize(nodes); i++) {
FstNode** node = (FstNode **)taosArrayGet(nodes, i);
fstNodeDestroy(*node);
}
taosArrayDestroy(nodes);
return NULL; return NULL;
} }
...@@ -1392,6 +1403,7 @@ void swsResultDestroy(StreamWithStateResult *result) { ...@@ -1392,6 +1403,7 @@ void swsResultDestroy(StreamWithStateResult *result) {
if (NULL == result) { return; } if (NULL == result) { return; }
fstSliceDestroy(&result->data); fstSliceDestroy(&result->data);
startWithStateValueDestroy(result->state);
free(result); free(result);
} }
......
source/libs/index/src/index_fst_automation.c
浏览文件 @ c7d0bee0
...@@ -16,21 +16,81 @@ ...@@ -16,21 +16,81 @@
#include "index_fst_automation.h" #include "index_fst_automation.h"
StartWithStateValue *startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void *val) {
StartWithStateValue *nsv = calloc(1, sizeof(StartWithStateValue));
if (nsv == NULL) { return NULL; }
nsv->kind = kind;
nsv->type = ty;
if (ty == FST_INT) {
nsv->val = *(int *)val;
} else if (ty == FST_CHAR) {
size_t len = strlen((char *)val);
nsv->ptr = (char *)calloc(1, len + 1);
memcpy(nsv->ptr, val, len);
} else if (ty == FST_ARRAY) {
//TODO,
//nsv->arr = taosArrayFromList()
}
return nsv;
}
void startWithStateValueDestroy(void *val) {
StartWithStateValue *sv = (StartWithStateValue *)val;
if (sv == NULL) { return; }
if (sv->type == FST_INT) {
//
} else if (sv->type == FST_CHAR) {
free(sv->ptr);
} else if (sv->type == FST_ARRAY) {
taosArrayDestroy(sv->arr);
}
free(sv);
}
StartWithStateValue *startWithStateValueDump(StartWithStateValue *sv) {
StartWithStateValue *nsv = calloc(1, sizeof(StartWithStateValue));
if (nsv == NULL) { return NULL; }
nsv->kind = sv->kind;
nsv->type= sv->type;
if (nsv->type == FST_INT) {
nsv->val = sv->val;
} else if (nsv->type == FST_CHAR) {
size_t len = strlen(sv->ptr);
nsv->ptr = (char *)calloc(1, len + 1);
memcpy(nsv->ptr, sv->ptr, len);
} else if (nsv->type == FST_ARRAY) {
}
return nsv;
}
// prefix query, impl later // prefix query, impl later
static void* prefixStart(AutomationCtx *ctx) { static void* prefixStart(AutomationCtx *ctx) {
StartWithStateValue *data = (StartWithStateValue *)(ctx->data); StartWithStateValue *data = (StartWithStateValue *)(ctx->data);
return data; return startWithStateValueDump(data);
}; };
static bool prefixIsMatch(AutomationCtx *ctx, void *data) { static bool prefixIsMatch(AutomationCtx *ctx, void *sv) {
return true; StartWithStateValue* ssv = (StartWithStateValue *)sv;
return ssv->val == strlen(ctx->data);
} }
static bool prefixCanMatch(AutomationCtx *ctx, void *data) { static bool prefixCanMatch(AutomationCtx *ctx, void *sv) {
return true; StartWithStateValue* ssv = (StartWithStateValue *)sv;
return ssv->val >= 0;
} }
static bool prefixWillAlwaysMatch(AutomationCtx *ctx, void *state) { static bool prefixWillAlwaysMatch(AutomationCtx *ctx, void *state) {
return true; return true;
} }
static void* prefixAccept(AutomationCtx *ctx, void *state, uint8_t byte) { static void* prefixAccept(AutomationCtx *ctx, void *state, uint8_t byte) {
StartWithStateValue* ssv = (StartWithStateValue *)state;
if (ssv == NULL || ctx == NULL) {return NULL;}
char *data = ctx->data;
if ((strlen(data) > ssv->val) && data[ssv->val] == byte) {
int val = ssv->val + 1;
return startWithStateValueCreate(Running, FST_INT, &val);
}
return NULL; return NULL;
} }
static void* prefixAcceptEof(AutomationCtx *ctx, void *state) { static void* prefixAcceptEof(AutomationCtx *ctx, void *state) {
...@@ -79,28 +139,33 @@ AutomationFunc automFuncs[] = {{ ...@@ -79,28 +139,33 @@ AutomationFunc automFuncs[] = {{
// add more search type // add more search type
}; };
AutomationCtx* automCtxCreate(void *data, AutomationType type) { AutomationCtx* automCtxCreate(void *data,AutomationType atype) {
AutomationCtx *ctx = calloc(1, sizeof(AutomationCtx)); AutomationCtx *ctx = calloc(1, sizeof(AutomationCtx));
if (ctx == NULL) { return NULL; } if (ctx == NULL) { return NULL; }
if (type == AUTOMATION_PREFIX) { StartWithStateValue *sv = NULL;
StartWithStateValue *swsv = (StartWithStateValue *)calloc(1, sizeof(StartWithStateValue)); if (atype == AUTOMATION_PREFIX) {
swsv->kind = Done; sv = startWithStateValueCreate(Running, FST_INT, 0);
swsv->value = NULL; ctx->stdata = (void *)sv;
ctx->data = (void *)swsv; } else if (atype == AUTMMATION_MATCH) {
} else if (type == AUTMMATION_MATCH) {
} else { } else {
// add more search type // add more search type
} }
ctx->type = type; char* src = (char *)data;
size_t len = strlen(src);
char* dst = (char *)malloc(len * sizeof(char) + 1);
memcpy(dst, src, len);
dst[len] = 0;
ctx->data = dst;
ctx->type = atype;
ctx->stdata = (void *)sv;
return ctx; return ctx;
} }
void automCtxDestroy(AutomationCtx *ctx) { void automCtxDestroy(AutomationCtx *ctx) {
if (ctx->type == AUTOMATION_PREFIX) { startWithStateValueDestroy(ctx->stdata);
free(ctx->data); free(ctx->data);
} else if (ctx->type == AUTMMATION_MATCH) {
}
free(ctx); free(ctx);
} }
source/libs/parser/test/plannerTest.cpp
浏览文件 @ c7d0bee0
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "tdef.h" #include "tdef.h"
#include "tvariant.h" #include "tvariant.h"
#include "planner.h" #include "planner.h"
#include "../../planner/inc/plannerInt.h"
namespace { namespace {
void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_t colId) { void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_t colId) {
...@@ -92,10 +93,10 @@ void generateLogicplan(const char* sql) { ...@@ -92,10 +93,10 @@ void generateLogicplan(const char* sql) {
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
struct SQueryPlanNode* n = nullptr; struct SQueryPlanNode* n = nullptr;
code = qCreateQueryPlan(pQueryInfo, &n); code = createQueryPlan(pQueryInfo, &n);
char* str = NULL; char* str = NULL;
qQueryPlanToString(n, &str); queryPlanToString(n, &str);
printf("--------SQL:%s\n", sql); printf("--------SQL:%s\n", sql);
printf("%s\n", str); printf("%s\n", str);
...@@ -155,10 +156,10 @@ TEST(testCase, planner_test) { ...@@ -155,10 +156,10 @@ TEST(testCase, planner_test) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2); ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
struct SQueryPlanNode* n = nullptr; struct SQueryPlanNode* n = nullptr;
code = qCreateQueryPlan(pQueryInfo, &n); code = createQueryPlan(pQueryInfo, &n);
char* str = NULL; char* str = NULL;
qQueryPlanToString(n, &str); queryPlanToString(n, &str);
printf("%s\n", str); printf("%s\n", str);
destroyQueryInfo(pQueryInfo); destroyQueryInfo(pQueryInfo);
......
source/libs/planner/inc/plannerInt.h
浏览文件 @ c7d0bee0
...@@ -25,6 +25,19 @@ extern "C" { ...@@ -25,6 +25,19 @@ extern "C" {
#include "planner.h" #include "planner.h"
#include "taosmsg.h" #include "taosmsg.h"
enum LOGIC_PLAN_E {
LP_SCAN = 1,
LP_SESSION = 2,
LP_STATE = 3,
LP_INTERVAL = 4,
LP_FILL = 5,
LP_AGG = 6,
LP_JOIN = 7,
LP_PROJECT = 8,
LP_DISTINCT = 9,
LP_ORDER = 10
};
typedef struct SQueryNodeBasicInfo { typedef struct SQueryNodeBasicInfo {
int32_t type; // operator type int32_t type; // operator type
char *name; // operator name char *name; // operator name
...@@ -57,50 +70,94 @@ typedef struct SQueryPlanNode { ...@@ -57,50 +70,94 @@ typedef struct SQueryPlanNode {
struct SQueryPlanNode *nextNode; struct SQueryPlanNode *nextNode;
} SQueryPlanNode; } SQueryPlanNode;
typedef struct SQueryDistPlanNode { typedef SSchema SSlotSchema;
typedef struct SDataBlockSchema {
int32_t index;
SSlotSchema *pSchema;
int32_t numOfCols; // number of columns
} SDataBlockSchema;
typedef struct SPhyNode {
SQueryNodeBasicInfo info; SQueryNodeBasicInfo info;
SSchema *pSchema; // the schema of the input SSDatablock SArray *pTargets; // target list to be computed or scanned at this node
int32_t numOfCols; // number of input columns SArray *pConditions; // implicitly-ANDed qual conditions
SArray *pExpr; // the query functions or sql aggregations SDataBlockSchema targetSchema;
int32_t numOfExpr; // number of result columns, which is also the number of pExprs // children plan to generated result for current node to process
void *pExtInfo; // additional information // in case of join, multiple plan nodes exist.
SArray *pChildren;
} SPhyNode;
typedef struct SScanPhyNode {
SPhyNode node;
uint64_t uid; // unique id of the table
} SScanPhyNode;
typedef SScanPhyNode STagScanPhyNode;
typedef SScanPhyNode SSystemTableScanPhyNode;
typedef struct SMultiTableScanPhyNode {
SScanPhyNode scan;
SArray *pTagsConditions; // implicitly-ANDed tag qual conditions
} SMultiTableScanPhyNode;
typedef SMultiTableScanPhyNode SMultiTableSeqScanPhyNode;
typedef struct SProjectPhyNode {
SPhyNode node;
} SProjectPhyNode;
/**
* Optimize the query execution plan, currently not implement yet.
* @param pQueryNode
* @return
*/
int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
// previous operator to generated result for current node to process /**
// in case of join, multiple prev nodes exist. * Create the query plan according to the bound AST, which is in the form of pQueryInfo
SArray *pPrevNodes; // upstream nodes, or exchange operator to load data from multiple sources. * @param pQueryInfo
} SQueryDistPlanNode; * @param pQueryNode
* @return
typedef struct SQueryCostSummary { */
int64_t startTs; // Object created and added into the message queue int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
int64_t endTs; // the timestamp when the task is completed
int64_t cputime; // total cpu cost, not execute elapsed time /**
* Convert the query plan to string, in order to display it in the shell.
int64_t loadRemoteDataDuration; // remote io time * @param pQueryNode
int64_t loadNativeDataDuration; // native disk io time * @return
*/
uint64_t loadNativeData; // blocks + SMA + header files int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
uint64_t loadRemoteData; // remote data acquired by exchange operator.
/**
uint64_t waitDuration; // the time to waiting to be scheduled in queue does matter, so we need to record it * Restore the SQL statement according to the logic query plan.
int64_t addQTs; // the time to be added into the message queue, used to calculate the waiting duration in queue. * @param pQueryNode
* @param sql
uint64_t totalRows; * @return
uint64_t loadRows; */
uint32_t totalBlocks; int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
uint32_t loadBlocks;
uint32_t loadBlockAgg; /**
uint32_t skipBlocks; * Convert to physical plan to string to enable to print it out in the shell.
uint64_t resultSize; // generated result size in Kb. * @param pPhyNode
} SQueryCostSummary; * @param str
* @return
typedef struct SQueryTask { */
uint64_t queryId; // query id int32_t phyPlanToString(struct SPhyNode *pPhyNode, char** str);
uint64_t taskId; // task id
SQueryDistPlanNode *pNode; // operator tree /**
uint64_t status; // task status * Destroy the query plan object.
SQueryCostSummary summary; // task execution summary * @return
void *pOutputHandle; // result buffer handle, to temporarily keep the output result for next stage */
} SQueryTask; void* destroyQueryPlan(struct SQueryPlanNode* pQueryNode);
/**
* Destroy the physical plan.
* @param pQueryPhyNode
* @return
*/
void* destroyQueryPhyPlan(struct SPhyNode* pQueryPhyNode);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
src/tsdb/inc/tsdbCommitQueue.h → source/libs/planner/src/physicalPlan.c
浏览文件 @ c7d0bee0
...@@ -13,11 +13,24 @@ ...@@ -13,11 +13,24 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#ifndef _TD_TSDB_COMMIT_QUEUE_H_ #include "plannerInt.h"
#define _TD_TSDB_COMMIT_QUEUE_H_
typedef enum { COMMIT_REQ, COMPACT_REQ,COMMIT_CONFIG_REQ } TSDB_REQ_T; SPhyNode* createScanNode(SQueryPlanNode* pPlanNode) {
return NULL;
}
int tsdbScheduleCommit(STsdbRepo *pRepo, TSDB_REQ_T req); SPhyNode* createPhyNode(SQueryPlanNode* node) {
switch (node->info.type) {
case LP_SCAN:
return createScanNode(node);
}
return NULL;
}
#endif /* _TD_TSDB_COMMIT_QUEUE_H_ */ SPhyNode* createSubplan(SQueryPlanNode* pSubquery) {
\ No newline at end of file return NULL;
}
int32_t createDag(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDag** pDag) {
return 0;
}
source/libs/planner/src/planner.c
浏览文件 @ c7d0bee0
...@@ -48,11 +48,11 @@ static SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo); ...@@ -48,11 +48,11 @@ static SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo);
static void doDestroyQueryNode(SQueryPlanNode* pQueryNode); static void doDestroyQueryNode(SQueryPlanNode* pQueryNode);
int32_t printExprInfo(const char* buf, const SQueryPlanNode* pQueryNode, int32_t len); int32_t printExprInfo(const char* buf, const SQueryPlanNode* pQueryNode, int32_t len);
int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode) { int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
return 0; return 0;
} }
int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) { int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) {
SArray* upstream = createQueryPlanImpl((struct SQueryStmtInfo*) pQueryInfo); SArray* upstream = createQueryPlanImpl((struct SQueryStmtInfo*) pQueryInfo);
assert(taosArrayGetSize(upstream) == 1); assert(taosArrayGetSize(upstream) == 1);
...@@ -62,19 +62,20 @@ int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryP ...@@ -62,19 +62,20 @@ int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryP
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qQueryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) { int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) {
return 0; return 0;
} }
int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDistPlanNode *pPhyNode) { int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDag** pDag) {
return 0; return 0;
} }
int32_t qPhyPlanToString(struct SQueryDistPlanNode *pPhyNode, char** str) { int32_t phyPlanToString(struct SPhyNode *pPhyNode, char** str) {
return 0; return 0;
} }
void* qDestroyQueryPlan(SQueryPlanNode* pQueryNode) { void* destroyQueryPlan(SQueryPlanNode* pQueryNode) {
if (pQueryNode == NULL) { if (pQueryNode == NULL) {
return NULL; return NULL;
} }
...@@ -83,14 +84,10 @@ void* qDestroyQueryPlan(SQueryPlanNode* pQueryNode) { ...@@ -83,14 +84,10 @@ void* qDestroyQueryPlan(SQueryPlanNode* pQueryNode) {
return NULL; return NULL;
} }
void* qDestroyQueryPhyPlan(struct SQueryDistPlanNode* pQueryPhyNode) { void* destroyQueryPhyPlan(struct SPhyNode* pQueryPhyNode) {
return NULL; return NULL;
} }
int32_t qCreateQueryJob(const struct SQueryDistPlanNode* pPhyNode, struct SQueryJob** pJob) {
return 0;
}
//====================================================================================================================== //======================================================================================================================
static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPlanNode** prev, int32_t numOfPrev, static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPlanNode** prev, int32_t numOfPrev,
...@@ -619,7 +616,7 @@ int32_t queryPlanToStringImpl(char* buf, SQueryPlanNode* pQueryNode, int32_t lev ...@@ -619,7 +616,7 @@ int32_t queryPlanToStringImpl(char* buf, SQueryPlanNode* pQueryNode, int32_t lev
return len; return len;
} }
int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) { int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) {
assert(pQueryNode); assert(pQueryNode);
*str = calloc(1, 4096); *str = calloc(1, 4096);
......
source/libs/wal/inc/walInt.h
浏览文件 @ c7d0bee0
...@@ -33,12 +33,10 @@ typedef struct WalFileInfo { ...@@ -33,12 +33,10 @@ typedef struct WalFileInfo {
int64_t fileSize; int64_t fileSize;
} WalFileInfo; } WalFileInfo;
#pragma pack(push,1)
typedef struct WalIdxEntry { typedef struct WalIdxEntry {
int64_t ver; int64_t ver;
int64_t offset; int64_t offset;
} WalIdxEntry; } WalIdxEntry;
#pragma pack(pop)
static inline int32_t compareWalFileInfo(const void* pLeft, const void* pRight) { static inline int32_t compareWalFileInfo(const void* pLeft, const void* pRight) {
WalFileInfo* pInfoLeft = (WalFileInfo*)pLeft; WalFileInfo* pInfoLeft = (WalFileInfo*)pLeft;
...@@ -107,8 +105,16 @@ static inline uint32_t walCalcBodyCksum(const void* body, uint32_t len) { ...@@ -107,8 +105,16 @@ static inline uint32_t walCalcBodyCksum(const void* body, uint32_t len) {
return taosCalcChecksum(0, (uint8_t*)body, len); return taosCalcChecksum(0, (uint8_t*)body, len);
} }
int walReadMeta(SWal* pWal); static inline void walResetVer(SWalVer* pVer) {
int walWriteMeta(SWal* pWal); pVer->firstVer = -1;
pVer->verInSnapshotting = -1;
pVer->snapshotVer = -1;
pVer->commitVer = -1;
pVer->lastVer = -1;
}
int walLoadMeta(SWal* pWal);
int walSaveMeta(SWal* pWal);
int walRollFileInfo(SWal* pWal); int walRollFileInfo(SWal* pWal);
char* walMetaSerialize(SWal* pWal); char* walMetaSerialize(SWal* pWal);
......
source/libs/wal/src/walMeta.c
浏览文件 @ c7d0bee0
...@@ -24,18 +24,22 @@ ...@@ -24,18 +24,22 @@
#include <libgen.h> #include <libgen.h>
#include <regex.h> #include <regex.h>
int64_t walGetFirstVer(SWal *pWal) { int64_t inline walGetFirstVer(SWal *pWal) {
return pWal->vers.firstVer; return pWal->vers.firstVer;
} }
int64_t walGetSnaphostVer(SWal *pWal) { int64_t inline walGetSnaphostVer(SWal *pWal) {
return pWal->vers.snapshotVer; return pWal->vers.snapshotVer;
} }
int64_t walGetLastVer(SWal *pWal) { int64_t inline walGetLastVer(SWal *pWal) {
return pWal->vers.lastVer; return pWal->vers.lastVer;
} }
static inline int walBuildMetaName(SWal* pWal, int metaVer, char* buf) {
return sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer);
}
int walRollFileInfo(SWal* pWal) { int walRollFileInfo(SWal* pWal) {
int64_t ts = taosGetTimestampSec(); int64_t ts = taosGetTimestampSec();
...@@ -150,10 +154,6 @@ int walMetaDeserialize(SWal* pWal, const char* bytes) { ...@@ -150,10 +154,6 @@ int walMetaDeserialize(SWal* pWal, const char* bytes) {
return 0; return 0;
} }
static inline int walBuildMetaName(SWal* pWal, int metaVer, char* buf) {
return sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer);
}
static int walFindCurMetaVer(SWal* pWal) { static int walFindCurMetaVer(SWal* pWal) {
const char * pattern = "^meta-ver[0-9]+$"; const char * pattern = "^meta-ver[0-9]+$";
regex_t walMetaRegexPattern; regex_t walMetaRegexPattern;
...@@ -182,7 +182,7 @@ static int walFindCurMetaVer(SWal* pWal) { ...@@ -182,7 +182,7 @@ static int walFindCurMetaVer(SWal* pWal) {
return metaVer; return metaVer;
} }
int walWriteMeta(SWal* pWal) { int walSaveMeta(SWal* pWal) {
int metaVer = walFindCurMetaVer(pWal); int metaVer = walFindCurMetaVer(pWal);
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
walBuildMetaName(pWal, metaVer+1, fnameStr); walBuildMetaName(pWal, metaVer+1, fnameStr);
...@@ -207,7 +207,7 @@ int walWriteMeta(SWal* pWal) { ...@@ -207,7 +207,7 @@ int walWriteMeta(SWal* pWal) {
return 0; return 0;
} }
int walReadMeta(SWal* pWal) { int walLoadMeta(SWal* pWal) {
ASSERT(pWal->fileInfoSet->size == 0); ASSERT(pWal->fileInfoSet->size == 0);
//find existing meta file //find existing meta file
int metaVer = walFindCurMetaVer(pWal); int metaVer = walFindCurMetaVer(pWal);
......
source/libs/wal/src/walMgmt.c
浏览文件 @ c7d0bee0
...@@ -21,23 +21,17 @@ ...@@ -21,23 +21,17 @@
#include "compare.h" #include "compare.h"
#include "walInt.h" #include "walInt.h"
//internal
int32_t walGetNextFile(SWal *pWal, int64_t *nextFileId);
int32_t walGetOldFile(SWal *pWal, int64_t curFileId, int32_t minDiff, int64_t *oldFileId);
int32_t walGetNewFile(SWal *pWal, int64_t *newFileId);
typedef struct { typedef struct {
int32_t refSetId;
uint32_t seq;
int8_t stop; int8_t stop;
int8_t inited; int8_t inited;
uint32_t seq;
int32_t refSetId;
pthread_t thread; pthread_t thread;
} SWalMgmt; } SWalMgmt;
static SWalMgmt tsWal = {0, .seq = 1}; static SWalMgmt tsWal = {0, .seq = 1};
static int32_t walCreateThread(); static int32_t walCreateThread();
static void walStopThread(); static void walStopThread();
static int32_t walInitObj(SWal *pWal);
static void walFreeObj(void *pWal); static void walFreeObj(void *pWal);
int64_t walGetSeq() { int64_t walGetSeq() {
...@@ -68,7 +62,7 @@ int32_t walInit() { ...@@ -68,7 +62,7 @@ int32_t walInit() {
} }
void walCleanUp() { void walCleanUp() {
int old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 0); int8_t old = atomic_val_compare_exchange_8(&tsWal.inited, 1, 0);
if(old == 0) { if(old == 0) {
return; return;
} }
...@@ -83,48 +77,59 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) { ...@@ -83,48 +77,59 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return NULL; return NULL;
} }
memset(pWal, 0, sizeof(SWal));
pWal->writeLogTfd = -1;
pWal->writeIdxTfd = -1;
pWal->writeCur = -1;
//set config //set config
memcpy(&pWal->cfg, pCfg, sizeof(SWalCfg)); memcpy(&pWal->cfg, pCfg, sizeof(SWalCfg));
pWal->fsyncSeq = pCfg->fsyncPeriod / 1000;
if(pWal->fsyncSeq <= 0) pWal->fsyncSeq = 1;
//init version info tstrncpy(pWal->path, path, sizeof(pWal->path));
pWal->vers.firstVer = -1; if(taosMkDir(pWal->path) != 0) {
pWal->vers.commitVer = -1; wError("vgId:%d, path:%s, failed to create directory since %s", pWal->cfg.vgId, pWal->path, strerror(errno));
pWal->vers.snapshotVer = -1; return NULL;
pWal->vers.lastVer = -1; }
pWal->vers.verInSnapshotting = -1;
pWal->totSize = 0; //open meta
pWal->writeLogTfd = -1;
pWal->writeIdxTfd = -1;
pWal->writeCur = -1;
pWal->fileInfoSet = taosArrayInit(8, sizeof(WalFileInfo));
if(pWal->fileInfoSet == NULL) {
wError("vgId:%d, path:%s, failed to init taosArray %s", pWal->cfg.vgId, pWal->path, strerror(errno));
free(pWal);
return NULL;
}
//init status //init status
walResetVer(&pWal->vers);
pWal->totSize = 0;
pWal->lastRollSeq = -1; pWal->lastRollSeq = -1;
//init write buffer //init write buffer
memset(&pWal->writeHead, 0, sizeof(SWalHead)); memset(&pWal->writeHead, 0, sizeof(SWalHead));
pWal->writeHead.head.sver = 0; pWal->writeHead.head.headVer = WAL_HEAD_VER;
tstrncpy(pWal->path, path, sizeof(pWal->path)); if(pthread_mutex_init(&pWal->mutex, NULL) < 0) {
pthread_mutex_init(&pWal->mutex, NULL); taosArrayDestroy(pWal->fileInfoSet);
free(pWal);
pWal->fsyncSeq = pCfg->fsyncPeriod / 1000; return NULL;
if (pWal->fsyncSeq <= 0) pWal->fsyncSeq = 1; }
if (walInitObj(pWal) != 0) { pWal->refId = taosAddRef(tsWal.refSetId, pWal);
walFreeObj(pWal); if(pWal->refId < 0) {
pthread_mutex_destroy(&pWal->mutex);
taosArrayDestroy(pWal->fileInfoSet);
free(pWal);
return NULL; return NULL;
} }
pWal->refId = taosAddRef(tsWal.refSetId, pWal); if(walLoadMeta(pWal) < 0) {
if (pWal->refId < 0) { taosRemoveRef(tsWal.refSetId, pWal->refId);
walFreeObj(pWal); pthread_mutex_destroy(&pWal->mutex);
taosArrayDestroy(pWal->fileInfoSet);
free(pWal);
return NULL; return NULL;
} }
walReadMeta(pWal);
wDebug("vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d", pWal->cfg.vgId, pWal, pWal->cfg.level, pWal->cfg.fsyncPeriod); wDebug("vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d", pWal->cfg.vgId, pWal, pWal->cfg.level, pWal->cfg.fsyncPeriod);
...@@ -152,43 +157,23 @@ int32_t walAlter(SWal *pWal, SWalCfg *pCfg) { ...@@ -152,43 +157,23 @@ int32_t walAlter(SWal *pWal, SWalCfg *pCfg) {
} }
void walClose(SWal *pWal) { void walClose(SWal *pWal) {
if (pWal == NULL) return;
pthread_mutex_lock(&pWal->mutex); pthread_mutex_lock(&pWal->mutex);
tfClose(pWal->writeLogTfd); tfClose(pWal->writeLogTfd);
pWal->writeLogTfd = -1; pWal->writeLogTfd = -1;
tfClose(pWal->writeIdxTfd); tfClose(pWal->writeIdxTfd);
pWal->writeIdxTfd = -1; pWal->writeIdxTfd = -1;
walWriteMeta(pWal); walSaveMeta(pWal);
taosArrayDestroy(pWal->fileInfoSet); taosArrayDestroy(pWal->fileInfoSet);
pWal->fileInfoSet = NULL; pWal->fileInfoSet = NULL;
pthread_mutex_unlock(&pWal->mutex); pthread_mutex_unlock(&pWal->mutex);
taosRemoveRef(tsWal.refSetId, pWal->refId);
}
static int32_t walInitObj(SWal *pWal) {
if (taosMkDir(pWal->path) != 0) {
wError("vgId:%d, path:%s, failed to create directory since %s", pWal->cfg.vgId, pWal->path, strerror(errno));
return TAOS_SYSTEM_ERROR(errno);
}
pWal->fileInfoSet = taosArrayInit(8, sizeof(WalFileInfo));
if(pWal->fileInfoSet == NULL) {
wError("vgId:%d, path:%s, failed to init taosArray %s", pWal->cfg.vgId, pWal->path, strerror(errno));
return TAOS_SYSTEM_ERROR(errno);
}
wDebug("vgId:%d, object is initialized", pWal->cfg.vgId); taosRemoveRef(tsWal.refSetId, pWal->refId);
return 0;
} }
static void walFreeObj(void *wal) { static void walFreeObj(void *wal) {
SWal *pWal = wal; SWal *pWal = wal;
wDebug("vgId:%d, wal:%p is freed", pWal->cfg.vgId, pWal); wDebug("vgId:%d, wal:%p is freed", pWal->cfg.vgId, pWal);
tfClose(pWal->writeLogTfd);
tfClose(pWal->writeIdxTfd);
taosArrayDestroy(pWal->fileInfoSet);
pWal->fileInfoSet = NULL;
pthread_mutex_destroy(&pWal->mutex); pthread_mutex_destroy(&pWal->mutex);
tfree(pWal); tfree(pWal);
} }
......
source/libs/wal/src/walRead.c
浏览文件 @ c7d0bee0
...@@ -54,7 +54,7 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i ...@@ -54,7 +54,7 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i
int64_t logTfd = pRead->readLogTfd; int64_t logTfd = pRead->readLogTfd;
//seek position //seek position
int64_t offset = (ver - fileFirstVer) * WAL_IDX_ENTRY_SIZE; int64_t offset = (ver - fileFirstVer) * sizeof(WalIdxEntry);
code = tfLseek(idxTfd, offset, SEEK_SET); code = tfLseek(idxTfd, offset, SEEK_SET);
if(code < 0) { if(code < 0) {
return -1; return -1;
...@@ -210,6 +210,6 @@ int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) { ...@@ -210,6 +210,6 @@ int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) {
return 0; return 0;
} }
int32_t walReadWithFp(SWal *pWal, FWalWrite writeFp, int64_t verStart, int32_t readNum) { /*int32_t walReadWithFp(SWal *pWal, FWalWrite writeFp, int64_t verStart, int32_t readNum) {*/
return 0; /*return 0;*/
} /*}*/
source/libs/wal/src/walSeek.c
浏览文件 @ c7d0bee0
...@@ -27,7 +27,7 @@ static int walSeekFilePos(SWal* pWal, int64_t ver) { ...@@ -27,7 +27,7 @@ static int walSeekFilePos(SWal* pWal, int64_t ver) {
int64_t logTfd = pWal->writeLogTfd; int64_t logTfd = pWal->writeLogTfd;
//seek position //seek position
int64_t offset = (ver - walGetCurFileFirstVer(pWal)) * WAL_IDX_ENTRY_SIZE; int64_t offset = (ver - walGetCurFileFirstVer(pWal)) * sizeof(WalIdxEntry);
code = tfLseek(idxTfd, offset, SEEK_SET); code = tfLseek(idxTfd, offset, SEEK_SET);
if(code != 0) { if(code != 0) {
return -1; return -1;
...@@ -66,8 +66,6 @@ int walChangeFileToLast(SWal *pWal) { ...@@ -66,8 +66,6 @@ int walChangeFileToLast(SWal *pWal) {
//switch file //switch file
pWal->writeIdxTfd = idxTfd; pWal->writeIdxTfd = idxTfd;
pWal->writeLogTfd = logTfd; pWal->writeLogTfd = logTfd;
//change status
pWal->curStatus = WAL_CUR_FILE_WRITABLE;
return 0; return 0;
} }
...@@ -93,13 +91,11 @@ int walChangeFile(SWal *pWal, int64_t ver) { ...@@ -93,13 +91,11 @@ int walChangeFile(SWal *pWal, int64_t ver) {
int64_t fileFirstVer = pRet->firstVer; int64_t fileFirstVer = pRet->firstVer;
//closed //closed
if(taosArrayGetLast(pWal->fileInfoSet) != pRet) { if(taosArrayGetLast(pWal->fileInfoSet) != pRet) {
pWal->curStatus &= ~WAL_CUR_FILE_WRITABLE;
walBuildIdxName(pWal, fileFirstVer, fnameStr); walBuildIdxName(pWal, fileFirstVer, fnameStr);
idxTfd = tfOpenRead(fnameStr); idxTfd = tfOpenRead(fnameStr);
walBuildLogName(pWal, fileFirstVer, fnameStr); walBuildLogName(pWal, fileFirstVer, fnameStr);
logTfd = tfOpenRead(fnameStr); logTfd = tfOpenRead(fnameStr);
} else { } else {
pWal->curStatus |= WAL_CUR_FILE_WRITABLE;
walBuildIdxName(pWal, fileFirstVer, fnameStr); walBuildIdxName(pWal, fileFirstVer, fnameStr);
idxTfd = tfOpenReadWrite(fnameStr); idxTfd = tfOpenReadWrite(fnameStr);
walBuildLogName(pWal, fileFirstVer, fnameStr); walBuildLogName(pWal, fileFirstVer, fnameStr);
......
source/libs/wal/src/walUtil.c 已删除 100644 → 0
浏览文件 @ 924e3ad6
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "walInt.h"
#if 0
int32_t walGetNextFile(SWal *pWal, int64_t *nextFileId) {
int64_t curFileId = *nextFileId;
int64_t minFileId = INT64_MAX;
DIR *dir = opendir(pWal->path);
if (dir == NULL) {
wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
return -1;
}
struct dirent *ent;
while ((ent = readdir(dir)) != NULL) {
char *name = ent->d_name;
if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
int64_t id = atoll(name + WAL_PREFIX_LEN);
if (id <= curFileId) continue;
if (id < minFileId) {
minFileId = id;
}
}
}
closedir(dir);
if (minFileId == INT64_MAX) return -1;
*nextFileId = minFileId;
wTrace("vgId:%d, path:%s, curFileId:%" PRId64 " nextFileId:%" PRId64, pWal->vgId, pWal->path, curFileId, *nextFileId);
return 0;
}
int32_t walGetOldFile(SWal *pWal, int64_t curFileId, int32_t minDiff, int64_t *oldFileId) {
int64_t minFileId = INT64_MAX;
DIR *dir = opendir(pWal->path);
if (dir == NULL) {
wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
return -1;
}
struct dirent *ent;
while ((ent = readdir(dir)) != NULL) {
char *name = ent->d_name;
if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
int64_t id = atoll(name + WAL_PREFIX_LEN);
if (id >= curFileId) continue;
minDiff--;
if (id < minFileId) {
minFileId = id;
}
}
}
closedir(dir);
if (minFileId == INT64_MAX) return -1;
if (minDiff > 0) return -1;
*oldFileId = minFileId;
wTrace("vgId:%d, path:%s, curFileId:%" PRId64 " oldFildId:%" PRId64, pWal->vgId, pWal->path, curFileId, *oldFileId);
return 0;
}
int32_t walGetNewFile(SWal *pWal, int64_t *newFileId) {
int64_t maxFileId = INT64_MIN;
DIR *dir = opendir(pWal->path);
if (dir == NULL) {
wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
return -1;
}
struct dirent *ent;
while ((ent = readdir(dir)) != NULL) {
char *name = ent->d_name;
if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
int64_t id = atoll(name + WAL_PREFIX_LEN);
if (id > maxFileId) {
maxFileId = id;
}
}
}
closedir(dir);
if (maxFileId == INT64_MIN) {
*newFileId = 0;
} else {
*newFileId = maxFileId;
}
wTrace("vgId:%d, path:%s, newFileId:%" PRId64, pWal->vgId, pWal->path, *newFileId);
return 0;
}
#endif
source/libs/wal/src/walWrite.c
浏览文件 @ c7d0bee0
...@@ -21,98 +21,6 @@ ...@@ -21,98 +21,6 @@
#include "tfile.h" #include "tfile.h"
#include "walInt.h" #include "walInt.h"
#if 0
static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name, int64_t fileId);
int32_t walRenew(void *handle) {
if (handle == NULL) return 0;
SWal * pWal = handle;
int32_t code = 0;
/*if (pWal->stop) {*/
/*wDebug("vgId:%d, do not create a new wal file", pWal->vgId);*/
/*return 0;*/
/*}*/
pthread_mutex_lock(&pWal->mutex);
if (tfValid(pWal->logTfd)) {
tfClose(pWal->logTfd);
wDebug("vgId:%d, file:%s, it is closed while renew", pWal->vgId, pWal->logName);
}
/*if (pWal->keep == TAOS_WAL_KEEP) {*/
/*pWal->fileId = 0;*/
/*} else {*/
/*if (walGetNewFile(pWal, &pWal->fileId) != 0) pWal->fileId = 0;*/
/*pWal->fileId++;*/
/*}*/
snprintf(pWal->logName, sizeof(pWal->logName), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, pWal->curFileId);
pWal->logTfd = tfOpenCreateWrite(pWal->logName);
if (!tfValid(pWal->logTfd)) {
code = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, file:%s, failed to open since %s", pWal->vgId, pWal->logName, strerror(errno));
} else {
wDebug("vgId:%d, file:%s, it is created and open while renew", pWal->vgId, pWal->logName);
}
pthread_mutex_unlock(&pWal->mutex);
return code;
}
void walRemoveOneOldFile(void *handle) {
SWal *pWal = handle;
if (pWal == NULL) return;
/*if (pWal->keep == TAOS_WAL_KEEP) return;*/
if (!tfValid(pWal->logTfd)) return;
pthread_mutex_lock(&pWal->mutex);
// remove the oldest wal file
int64_t oldFileId = -1;
if (walGetOldFile(pWal, pWal->curFileId, WAL_FILE_NUM, &oldFileId) == 0) {
char walName[WAL_FILE_LEN] = {0};
snprintf(walName, sizeof(walName), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, oldFileId);
if (remove(walName) < 0) {
wError("vgId:%d, file:%s, failed to remove since %s", pWal->vgId, walName, strerror(errno));
} else {
wInfo("vgId:%d, file:%s, it is removed", pWal->vgId, walName);
}
}
pthread_mutex_unlock(&pWal->mutex);
}
void walRemoveAllOldFiles(void *handle) {
if (handle == NULL) return;
SWal * pWal = handle;
int64_t fileId = -1;
pthread_mutex_lock(&pWal->mutex);
tfClose(pWal->logTfd);
wDebug("vgId:%d, file:%s, it is closed before remove all wals", pWal->vgId, pWal->logName);
while (walGetNextFile(pWal, &fileId) >= 0) {
snprintf(pWal->logName, sizeof(pWal->logName), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
if (remove(pWal->logName) < 0) {
wError("vgId:%d, wal:%p file:%s, failed to remove since %s", pWal->vgId, pWal, pWal->logName, strerror(errno));
} else {
wInfo("vgId:%d, wal:%p file:%s, it is removed", pWal->vgId, pWal, pWal->logName);
}
}
pthread_mutex_unlock(&pWal->mutex);
}
#endif
int32_t walCommit(SWal *pWal, int64_t ver) { int32_t walCommit(SWal *pWal, int64_t ver) {
ASSERT(pWal->vers.commitVer >= pWal->vers.snapshotVer); ASSERT(pWal->vers.commitVer >= pWal->vers.snapshotVer);
ASSERT(pWal->vers.commitVer <= pWal->vers.lastVer); ASSERT(pWal->vers.commitVer <= pWal->vers.lastVer);
...@@ -166,7 +74,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) { ...@@ -166,7 +74,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
pthread_mutex_unlock(&pWal->mutex); pthread_mutex_unlock(&pWal->mutex);
return -1; return -1;
} }
int idxOff = (ver - walGetCurFileFirstVer(pWal)) * WAL_IDX_ENTRY_SIZE; int idxOff = (ver - walGetCurFileFirstVer(pWal)) * sizeof(WalIdxEntry);
code = tfLseek(idxTfd, idxOff, SEEK_SET); code = tfLseek(idxTfd, idxOff, SEEK_SET);
if(code < 0) { if(code < 0) {
pthread_mutex_unlock(&pWal->mutex); pthread_mutex_unlock(&pWal->mutex);
...@@ -229,7 +137,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) { ...@@ -229,7 +137,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
return 0; return 0;
} }
int32_t walBeginTakeSnapshot(SWal* pWal, int64_t ver) { int32_t walBeginSnapshot(SWal* pWal, int64_t ver) {
pWal->vers.verInSnapshotting = ver; pWal->vers.verInSnapshotting = ver;
//check file rolling //check file rolling
if(pWal->cfg.retentionPeriod == 0) { if(pWal->cfg.retentionPeriod == 0) {
...@@ -239,7 +147,7 @@ int32_t walBeginTakeSnapshot(SWal* pWal, int64_t ver) { ...@@ -239,7 +147,7 @@ int32_t walBeginTakeSnapshot(SWal* pWal, int64_t ver) {
return 0; return 0;
} }
int32_t walEndTakeSnapshot(SWal *pWal) { int32_t walEndSnapshot(SWal *pWal) {
int64_t ver = pWal->vers.verInSnapshotting; int64_t ver = pWal->vers.verInSnapshotting;
if(ver == -1) return -1; if(ver == -1) return -1;
...@@ -287,7 +195,7 @@ int32_t walEndTakeSnapshot(SWal *pWal) { ...@@ -287,7 +195,7 @@ int32_t walEndTakeSnapshot(SWal *pWal) {
pWal->vers.verInSnapshotting = -1; pWal->vers.verInSnapshotting = -1;
//save snapshot ver, commit ver //save snapshot ver, commit ver
int code = walWriteMeta(pWal); int code = walSaveMeta(pWal);
if(code != 0) { if(code != 0) {
return -1; return -1;
} }
...@@ -314,13 +222,13 @@ int walRoll(SWal *pWal) { ...@@ -314,13 +222,13 @@ int walRoll(SWal *pWal) {
int64_t newFileFirstVersion = pWal->vers.lastVer + 1; int64_t newFileFirstVersion = pWal->vers.lastVer + 1;
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
walBuildIdxName(pWal, newFileFirstVersion, fnameStr); walBuildIdxName(pWal, newFileFirstVersion, fnameStr);
idxTfd = tfOpenCreateWrite(fnameStr); idxTfd = tfOpenCreateWriteAppend(fnameStr);
if(idxTfd < 0) { if(idxTfd < 0) {
ASSERT(0); ASSERT(0);
return -1; return -1;
} }
walBuildLogName(pWal, newFileFirstVersion, fnameStr); walBuildLogName(pWal, newFileFirstVersion, fnameStr);
logTfd = tfOpenCreateWrite(fnameStr); logTfd = tfOpenCreateWriteAppend(fnameStr);
if(logTfd < 0) { if(logTfd < 0) {
ASSERT(0); ASSERT(0);
return -1; return -1;
...@@ -335,8 +243,6 @@ int walRoll(SWal *pWal) { ...@@ -335,8 +243,6 @@ int walRoll(SWal *pWal) {
pWal->writeIdxTfd = idxTfd; pWal->writeIdxTfd = idxTfd;
pWal->writeLogTfd = logTfd; pWal->writeLogTfd = logTfd;
pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1; pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
//change status
pWal->curStatus = WAL_CUR_FILE_WRITABLE & WAL_CUR_POS_WRITABLE;
pWal->lastRollSeq = walGetSeq(); pWal->lastRollSeq = walGetSeq();
return 0; return 0;
...@@ -425,74 +331,6 @@ void walFsync(SWal *pWal, bool forceFsync) { ...@@ -425,74 +331,6 @@ void walFsync(SWal *pWal, bool forceFsync) {
} }
} }
#if 0
int32_t walRestore(void *handle, void *pVnode, FWalWrite writeFp) {
if (handle == NULL) return -1;
SWal * pWal = handle;
int32_t count = 0;
int32_t code = 0;
int64_t fileId = -1;
while ((code = walGetNextFile(pWal, &fileId)) >= 0) {
/*if (fileId == pWal->curFileId) continue;*/
char walName[WAL_FILE_LEN];
snprintf(walName, sizeof(pWal->logName), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
wInfo("vgId:%d, file:%s, will be restored", pWal->vgId, walName);
code = walRestoreWalFile(pWal, pVnode, writeFp, walName, fileId);
if (code != TSDB_CODE_SUCCESS) {
wError("vgId:%d, file:%s, failed to restore since %s", pWal->vgId, walName, tstrerror(code));
continue;
}
wInfo("vgId:%d, file:%s, restore success, wver:%" PRIu64, pWal->vgId, walName, pWal->curVersion);
count++;
}
/*if (pWal->keep != TAOS_WAL_KEEP) return TSDB_CODE_SUCCESS;*/
if (count == 0) {
wDebug("vgId:%d, wal file not exist, renew it", pWal->vgId);
return walRenew(pWal);
} else {
// open the existing WAL file in append mode
/*pWal->curFileId = 0;*/
snprintf(pWal->logName, sizeof(pWal->logName), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, pWal->curFileId);
pWal->logTfd = tfOpenCreateWriteAppend(pWal->logName);
if (!tfValid(pWal->logTfd)) {
wError("vgId:%d, file:%s, failed to open since %s", pWal->vgId, pWal->logName, strerror(errno));
return TAOS_SYSTEM_ERROR(errno);
}
wDebug("vgId:%d, file:%s, it is created and open while restore", pWal->vgId, pWal->logName);
}
return TSDB_CODE_SUCCESS;
}
int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) {
if (handle == NULL) return -1;
SWal *pWal = handle;
if (*fileId == 0) *fileId = -1;
pthread_mutex_lock(&(pWal->mutex));
int32_t code = walGetNextFile(pWal, fileId);
if (code >= 0) {
sprintf(fileName, "wal/%s%" PRId64, WAL_PREFIX, *fileId);
/*code = (*fileId == pWal->curFileId) ? 0 : 1;*/
}
wDebug("vgId:%d, get wal file, code:%d curId:%" PRId64 " outId:%" PRId64, pWal->vgId, code, pWal->curFileId, *fileId);
pthread_mutex_unlock(&(pWal->mutex));
return code;
}
#endif
/*static int walValidateOffset(SWal* pWal, int64_t ver) {*/ /*static int walValidateOffset(SWal* pWal, int64_t ver) {*/
/*int code = 0;*/ /*int code = 0;*/
/*SWalHead *pHead = NULL;*/ /*SWalHead *pHead = NULL;*/
...@@ -516,139 +354,3 @@ int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) { ...@@ -516,139 +354,3 @@ int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) {
/*return 0;*/ /*return 0;*/
/*}*/ /*}*/
#if 0
static int32_t walSkipCorruptedRecord(SWal *pWal, SWalHead *pHead, int64_t tfd, int64_t *offset) {
int64_t pos = *offset;
while (1) {
pos++;
if (tfLseek(tfd, pos, SEEK_SET) < 0) {
wError("vgId:%d, failed to seek from corrupted wal file since %s", pWal->vgId, strerror(errno));
return TSDB_CODE_WAL_FILE_CORRUPTED;
}
if (tfRead(tfd, pHead, sizeof(SWalHead)) <= 0) {
wError("vgId:%d, read to end of corrupted wal file, offset:%" PRId64, pWal->vgId, pos);
return TSDB_CODE_WAL_FILE_CORRUPTED;
}
if (pHead->signature != WAL_SIGNATURE) {
continue;
}
if (pHead->sver >= 1) {
if (tfRead(tfd, pHead->cont, pHead->len) < pHead->len) {
wError("vgId:%d, read to end of corrupted wal file, offset:%" PRId64, pWal->vgId, pos);
return TSDB_CODE_WAL_FILE_CORRUPTED;
}
if (walValidateChecksum(pHead)) {
wInfo("vgId:%d, wal whole cksum check passed, offset:%" PRId64, pWal->vgId, pos);
*offset = pos;
return TSDB_CODE_SUCCESS;
}
}
}
return TSDB_CODE_WAL_FILE_CORRUPTED;
}
static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name, int64_t fileId) {
int32_t size = WAL_MAX_SIZE;
void * buffer = malloc(size);
if (buffer == NULL) {
wError("vgId:%d, file:%s, failed to open for restore since %s", pWal->vgId, name, strerror(errno));
return TAOS_SYSTEM_ERROR(errno);
}
int64_t tfd = tfOpenReadWrite(name);
if (!tfValid(tfd)) {
wError("vgId:%d, file:%s, failed to open for restore since %s", pWal->vgId, name, strerror(errno));
tfree(buffer);
return TAOS_SYSTEM_ERROR(errno);
} else {
wDebug("vgId:%d, file:%s, open for restore", pWal->vgId, name);
}
int32_t code = TSDB_CODE_SUCCESS;
int64_t offset = 0;
SWalHead *pHead = buffer;
while (1) {
int32_t ret = (int32_t)tfRead(tfd, pHead, sizeof(SWalHead));
if (ret == 0) break;
if (ret < 0) {
wError("vgId:%d, file:%s, failed to read wal head since %s", pWal->vgId, name, strerror(errno));
code = TAOS_SYSTEM_ERROR(errno);
break;
}
if (ret < sizeof(SWalHead)) {
wError("vgId:%d, file:%s, failed to read wal head, ret is %d", pWal->vgId, name, ret);
walFtruncate(pWal, tfd, offset);
break;
}
if ((pHead->sver == 0 && !walValidateChecksum(pHead)) || pHead->sver < 0 || pHead->sver > 2) {
wError("vgId:%d, file:%s, wal head cksum is messed up, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
pHead->version, pHead->len, offset);
code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);
if (code != TSDB_CODE_SUCCESS) {
walFtruncate(pWal, tfd, offset);
break;
}
}
if (pHead->len < 0 || pHead->len > size - sizeof(SWalHead)) {
wError("vgId:%d, file:%s, wal head len out of range, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
pHead->version, pHead->len, offset);
code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);
if (code != TSDB_CODE_SUCCESS) {
walFtruncate(pWal, tfd, offset);
break;
}
}
ret = (int32_t)tfRead(tfd, pHead->cont, pHead->len);
if (ret < 0) {
wError("vgId:%d, file:%s, failed to read wal body since %s", pWal->vgId, name, strerror(errno));
code = TAOS_SYSTEM_ERROR(errno);
break;
}
if (ret < pHead->len) {
wError("vgId:%d, file:%s, failed to read wal body, ret:%d len:%d", pWal->vgId, name, ret, pHead->len);
offset += sizeof(SWalHead);
continue;
}
if ((pHead->sver >= 1) && !walValidateChecksum(pHead)) {
wError("vgId:%d, file:%s, wal whole cksum is messed up, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
pHead->version, pHead->len, offset);
code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);
if (code != TSDB_CODE_SUCCESS) {
walFtruncate(pWal, tfd, offset);
break;
}
}
offset = offset + sizeof(SWalHead) + pHead->len;
wTrace("vgId:%d, restore wal, fileId:%" PRId64 " hver:%" PRIu64 " wver:%" PRIu64 " len:%d offset:%" PRId64,
pWal->vgId, fileId, pHead->version, pWal->curVersion, pHead->len, offset);
pWal->curVersion = pHead->version;
// wInfo("writeFp: %ld", offset);
(*writeFp)(pVnode, pHead);
}
tfClose(tfd);
tfree(buffer);
wDebug("vgId:%d, file:%s, it is closed after restore", pWal->vgId, name);
return code;
}
#endif
source/libs/wal/test/walMetaTest.cpp
浏览文件 @ c7d0bee0
...@@ -142,7 +142,7 @@ TEST_F(WalCleanEnv, serialize) { ...@@ -142,7 +142,7 @@ TEST_F(WalCleanEnv, serialize) {
char*ss = walMetaSerialize(pWal); char*ss = walMetaSerialize(pWal);
printf("%s\n", ss); printf("%s\n", ss);
free(ss); free(ss);
code = walWriteMeta(pWal); code = walSaveMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
} }
...@@ -150,11 +150,11 @@ TEST_F(WalCleanEnv, removeOldMeta) { ...@@ -150,11 +150,11 @@ TEST_F(WalCleanEnv, removeOldMeta) {
int code = walRollFileInfo(pWal); int code = walRollFileInfo(pWal);
ASSERT(code == 0); ASSERT(code == 0);
ASSERT(pWal->fileInfoSet != NULL); ASSERT(pWal->fileInfoSet != NULL);
code = walWriteMeta(pWal); code = walSaveMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
code = walRollFileInfo(pWal); code = walRollFileInfo(pWal);
ASSERT(code == 0); ASSERT(code == 0);
code = walWriteMeta(pWal); code = walSaveMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
} }
...@@ -199,7 +199,7 @@ TEST_F(WalCleanEnv, write) { ...@@ -199,7 +199,7 @@ TEST_F(WalCleanEnv, write) {
ASSERT_EQ(code, -1); ASSERT_EQ(code, -1);
ASSERT_EQ(pWal->vers.lastVer, i); ASSERT_EQ(pWal->vers.lastVer, i);
} }
code = walWriteMeta(pWal); code = walSaveMeta(pWal);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
} }
...@@ -216,7 +216,7 @@ TEST_F(WalCleanEnv, rollback) { ...@@ -216,7 +216,7 @@ TEST_F(WalCleanEnv, rollback) {
code = walRollback(pWal, 3); code = walRollback(pWal, 3);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 2); ASSERT_EQ(pWal->vers.lastVer, 2);
code = walWriteMeta(pWal); code = walSaveMeta(pWal);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
} }
...@@ -231,9 +231,9 @@ TEST_F(WalCleanDeleteEnv, roll) { ...@@ -231,9 +231,9 @@ TEST_F(WalCleanDeleteEnv, roll) {
ASSERT_EQ(pWal->vers.commitVer, i); ASSERT_EQ(pWal->vers.commitVer, i);
} }
walBeginTakeSnapshot(pWal, i-1); walBeginSnapshot(pWal, i-1);
ASSERT_EQ(pWal->vers.verInSnapshotting, i-1); ASSERT_EQ(pWal->vers.verInSnapshotting, i-1);
walEndTakeSnapshot(pWal); walEndSnapshot(pWal);
ASSERT_EQ(pWal->vers.snapshotVer, i-1); ASSERT_EQ(pWal->vers.snapshotVer, i-1);
ASSERT_EQ(pWal->vers.verInSnapshotting, -1); ASSERT_EQ(pWal->vers.verInSnapshotting, -1);
...@@ -247,9 +247,9 @@ TEST_F(WalCleanDeleteEnv, roll) { ...@@ -247,9 +247,9 @@ TEST_F(WalCleanDeleteEnv, roll) {
ASSERT_EQ(pWal->vers.commitVer, i); ASSERT_EQ(pWal->vers.commitVer, i);
} }
code = walBeginTakeSnapshot(pWal, i - 1); code = walBeginSnapshot(pWal, i - 1);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
code = walEndTakeSnapshot(pWal); code = walEndSnapshot(pWal);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
} }
......
src/tsdb/src/tsdbCommitQueue.c 已删除 100644 → 0
浏览文件 @ 924e3ad6
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tsdbint.h"
typedef struct {
bool stop;
pthread_mutex_t lock;
pthread_cond_t queueNotEmpty;
int nthreads;
int refCount;
SList * queue;
pthread_t * threads;
} SCommitQueue;
typedef struct {
TSDB_REQ_T req;
STsdbRepo *pRepo;
} SReq;
static void *tsdbLoopCommit(void *arg);
static SCommitQueue tsCommitQueue = {0};
int tsdbInitCommitQueue() {
int nthreads = tsNumOfCommitThreads;
SCommitQueue *pQueue = &tsCommitQueue;
if (nthreads < 1) nthreads = 1;
pQueue->stop = false;
pQueue->nthreads = nthreads;
pQueue->queue = tdListNew(0);
if (pQueue->queue == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
pQueue->threads = (pthread_t *)calloc(nthreads, sizeof(pthread_t));
if (pQueue->threads == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tdListFree(pQueue->queue);
return -1;
}
pthread_mutex_init(&(pQueue->lock), NULL);
pthread_cond_init(&(pQueue->queueNotEmpty), NULL);
for (int i = 0; i < nthreads; i++) {
pthread_create(pQueue->threads + i, NULL, tsdbLoopCommit, NULL);
}
return 0;
}
void tsdbDestroyCommitQueue() {
SCommitQueue *pQueue = &tsCommitQueue;
pthread_mutex_lock(&(pQueue->lock));
if (pQueue->stop) {
pthread_mutex_unlock(&(pQueue->lock));
return;
}
pQueue->stop = true;
pthread_cond_broadcast(&(pQueue->queueNotEmpty));
pthread_mutex_unlock(&(pQueue->lock));
for (size_t i = 0; i < pQueue->nthreads; i++) {
pthread_join(pQueue->threads[i], NULL);
}
free(pQueue->threads);
tdListFree(pQueue->queue);
pthread_cond_destroy(&(pQueue->queueNotEmpty));
pthread_mutex_destroy(&(pQueue->lock));
}
int tsdbScheduleCommit(STsdbRepo *pRepo, TSDB_REQ_T req) {
SCommitQueue *pQueue = &tsCommitQueue;
SListNode *pNode = (SListNode *)calloc(1, sizeof(SListNode) + sizeof(SReq));
if (pNode == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
((SReq *)pNode->data)->req = req;
((SReq *)pNode->data)->pRepo = pRepo;
pthread_mutex_lock(&(pQueue->lock));
// ASSERT(pQueue->stop);
tdListAppendNode(pQueue->queue, pNode);
pthread_cond_signal(&(pQueue->queueNotEmpty));
pthread_mutex_unlock(&(pQueue->lock));
return 0;
}
static void tsdbApplyRepoConfig(STsdbRepo *pRepo) {
pthread_mutex_lock(&pRepo->save_mutex);
pRepo->config_changed = false;
STsdbCfg * pSaveCfg = &pRepo->save_config;
STsdbCfg oldCfg;
int32_t oldTotalBlocks = pRepo->config.totalBlocks;
memcpy(&oldCfg, &(pRepo->config), sizeof(STsdbCfg));
pRepo->config.compression = pRepo->save_config.compression;
pRepo->config.keep = pRepo->save_config.keep;
pRepo->config.keep1 = pRepo->save_config.keep1;
pRepo->config.keep2 = pRepo->save_config.keep2;
pRepo->config.cacheLastRow = pRepo->save_config.cacheLastRow;
pRepo->config.totalBlocks = pRepo->save_config.totalBlocks;
pthread_mutex_unlock(&pRepo->save_mutex);
tsdbInfo("vgId:%d apply new config: compression(%d), keep(%d,%d,%d), totalBlocks(%d), cacheLastRow(%d->%d),totalBlocks(%d->%d)",
REPO_ID(pRepo),
pSaveCfg->compression, pSaveCfg->keep,pSaveCfg->keep1, pSaveCfg->keep2,
pSaveCfg->totalBlocks, oldCfg.cacheLastRow, pSaveCfg->cacheLastRow, oldTotalBlocks, pSaveCfg->totalBlocks);
int err = tsdbExpandPool(pRepo, oldTotalBlocks);
if (!TAOS_SUCCEEDED(err)) {
tsdbError("vgId:%d expand pool from %d to %d fail,reason:%s",
REPO_ID(pRepo), oldTotalBlocks, pSaveCfg->totalBlocks, tstrerror(err));
}
if (oldCfg.cacheLastRow != pRepo->config.cacheLastRow) {
if (tsdbLockRepo(pRepo) < 0) return;
tsdbCacheLastData(pRepo, &oldCfg);
tsdbUnlockRepo(pRepo);
}
}
static void *tsdbLoopCommit(void *arg) {
SCommitQueue *pQueue = &tsCommitQueue;
SListNode * pNode = NULL;
STsdbRepo * pRepo = NULL;
TSDB_REQ_T req;
setThreadName("tsdbCommit");
while (true) {
pthread_mutex_lock(&(pQueue->lock));
while (true) {
pNode = tdListPopHead(pQueue->queue);
if (pNode == NULL) {
if (pQueue->stop && pQueue->refCount <= 0) {
pthread_mutex_unlock(&(pQueue->lock));
goto _exit;
} else {
pthread_cond_wait(&(pQueue->queueNotEmpty), &(pQueue->lock));
}
} else {
break;
}
}
pthread_mutex_unlock(&(pQueue->lock));
req = ((SReq *)pNode->data)->req;
pRepo = ((SReq *)pNode->data)->pRepo;
if (req == COMMIT_REQ) {
tsdbCommitData(pRepo);
} else if (req == COMPACT_REQ) {
tsdbCompactImpl(pRepo);
} else if (req == COMMIT_CONFIG_REQ) {
ASSERT(pRepo->config_changed);
tsdbApplyRepoConfig(pRepo);
tsem_post(&(pRepo->readyToCommit));
} else {
ASSERT(0);
}
listNodeFree(pNode);
}
_exit:
return NULL;
}
void tsdbIncCommitRef(int vgId) {
int refCount = atomic_add_fetch_32(&tsCommitQueue.refCount, 1);
tsdbDebug("vgId:%d, inc commit queue ref to %d", vgId, refCount);
}
void tsdbDecCommitRef(int vgId) {
int refCount = atomic_sub_fetch_32(&tsCommitQueue.refCount, 1);
pthread_cond_broadcast(&(tsCommitQueue.queueNotEmpty));
tsdbDebug("vgId:%d, dec commit queue ref to %d", vgId, refCount);
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或