提交 8938c902 编写于 作者: L Liu Jicong

merge from 3.0

......@@ -18,7 +18,8 @@
"ms-vscode.cpptools",
"ms-vscode.cmake-tools",
"austin.code-gnu-global",
"visualstudioexptteam.vscodeintel"
"visualstudioexptteam.vscodeintel",
"eamodio.gitlens"
],
// Use 'forwardPorts' to make a list of ports inside the container available locally.
......
......@@ -56,6 +56,12 @@ option(
OFF
)
option(
BUILD_WITH_TRAFT
"If build with traft"
OFF
)
option(
BUILD_DEPENDENCY_TESTS
"If build dependency tests"
......
# traft
ExternalProject_Add(traft
GIT_REPOSITORY https://github.com/taosdata/traft.git
GIT_TAG for_3.0
SOURCE_DIR "${CMAKE_CONTRIB_DIR}/traft"
BINARY_DIR "${CMAKE_CONTRIB_DIR}/traft"
#BUILD_IN_SOURCE TRUE
# https://answers.ros.org/question/333125/how-to-include-external-automakeautoconf-projects-into-ament_cmake/
CONFIGURE_COMMAND COMMAND autoreconf -i COMMAND ./configure --enable-example
BUILD_COMMAND "$(MAKE)"
INSTALL_COMMAND ""
TEST_COMMAND ""
)
......@@ -41,6 +41,12 @@ if(${BUILD_WITH_CRAFT})
SET(BUILD_WITH_UV ON CACHE BOOL "craft need libuv" FORCE)
endif(${BUILD_WITH_CRAFT})
# traft
if(${BUILD_WITH_TRAFT})
cat("${CMAKE_SUPPORT_DIR}/traft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
SET(BUILD_WITH_UV ON CACHE BOOL "traft need libuv" FORCE)
endif(${BUILD_WITH_TRAFT})
#libuv
if(${BUILD_WITH_UV})
cat("${CMAKE_SUPPORT_DIR}/libuv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
......@@ -173,6 +179,18 @@ if(${BUILD_WITH_CRAFT})
# )
endif(${BUILD_WITH_CRAFT})
# TRAFT
if(${BUILD_WITH_TRAFT})
add_library(traft STATIC IMPORTED GLOBAL)
set_target_properties(traft PROPERTIES
IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/traft/.libs/libraft.a"
INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/traft/include"
)
# target_link_libraries(craft
# INTERFACE pthread
# )
endif(${BUILD_WITH_TRAFT})
# LIBUV
if(${BUILD_WITH_UV})
add_subdirectory(libuv)
......
......@@ -19,4 +19,8 @@ if(${BUILD_WITH_CRAFT})
add_subdirectory(craft)
endif(${BUILD_WITH_CRAFT})
if(${BUILD_WITH_TRAFT})
add_subdirectory(traft)
endif(${BUILD_WITH_TRAFT})
add_subdirectory(tdev)
......@@ -20,6 +20,7 @@ typedef struct {
} Addr;
typedef struct {
int voter;
Addr me;
Addr peers[MAX_PEERS];
int peersCount;
......
......@@ -104,7 +104,7 @@ const char* state2String(unsigned short state) {
void printRaftConfiguration(struct raft_configuration *c) {
printf("configuration: \n");
for (int i = 0; i < c->n; ++i) {
printf("%llu -- %d -- %s\n", c->servers->id, c->servers->role, c->servers->address);
printf("%llu -- %d -- %s\n", c->servers[i].id, c->servers[i].role, c->servers[i].address);
}
}
......@@ -119,11 +119,9 @@ void printRaftState(struct raft *r) {
printf("last_applied: %llu \n", r->last_applied);
printf("last_stored: %llu \n", r->last_stored);
/*
printf("configuration_index: %llu \n", r->configuration_index);
printf("configuration_uncommitted_index: %llu \n", r->configuration_uncommitted_index);
printRaftConfiguration(&r->configuration);
*/
printf("----------------------------\n");
}
......@@ -164,6 +162,18 @@ void getValue(const char *key) {
}
}
void raft_change_cb_add(struct raft_change *req, int status) {
printf("raft_change_cb_add status:%d ... \n", status);
}
void raft_change_cb_assign(struct raft_change *req, int status) {
printf("raft_change_cb_assign status:%d ... \n", status);
}
void raft_change_cb_remove(struct raft_change *req, int status) {
printf("raft_change_cb_remove status:%d ... \n", status);
}
void console(SRaftServer *pRaftServer) {
while (1) {
char cmd_buf[COMMAND_LEN];
......@@ -193,30 +203,59 @@ void console(SRaftServer *pRaftServer) {
parseCommand(cmd_buf, cmd, param1, param2, TOKEN_LEN);
if (strcmp(cmd, "addnode") == 0) {
printf("not support \n");
//printf("not support \n");
/*
char host[HOST_LEN];
uint32_t port;
parseAddr(param1, host, HOST_LEN, &port);
uint64_t rid = raftId(host, port);
struct raft_change *req = raft_malloc(sizeof(*req));
int r = raft_add(&pRaftServer->raft, req, rid, param1, NULL);
int r = raft_add(&pRaftServer->raft, req, rid, param1, raft_change_cb_add);
if (r != 0) {
printf("raft_add: %s \n", raft_errmsg(&pRaftServer->raft));
printf("raft_add error: %s \n", raft_errmsg(&pRaftServer->raft));
}
printf("add node: %lu %s \n", rid, param1);
struct raft_change *req2 = raft_malloc(sizeof(*req2));
r = raft_assign(&pRaftServer->raft, req2, rid, RAFT_VOTER, NULL);
r = raft_assign(&pRaftServer->raft, req2, rid, RAFT_VOTER, raft_change_cb_assign);
if (r != 0) {
printf("raft_assign error: %s \n", raft_errmsg(&pRaftServer->raft));
}
printf("raft_assign: %s %d \n", param1, RAFT_VOTER);
} else if (strcmp(cmd, "activate") == 0) {
char host[HOST_LEN];
uint32_t port;
parseAddr(param1, host, HOST_LEN, &port);
uint64_t rid = raftId(host, port);
struct raft_change *req2 = raft_malloc(sizeof(*req2));
int r = raft_assign(&pRaftServer->raft, req2, rid, RAFT_VOTER, raft_change_cb_assign);
if (r != 0) {
printf("raft_assign: %s \n", raft_errmsg(&pRaftServer->raft));
printf("raft_assign error: %s \n", raft_errmsg(&pRaftServer->raft));
}
*/
printf("raft_assign: %s %d \n", param1, RAFT_VOTER);
} else if (strcmp(cmd, "dropnode") == 0) {
printf("not support \n");
char host[HOST_LEN];
uint32_t port;
parseAddr(param1, host, HOST_LEN, &port);
uint64_t rid = raftId(host, port);
struct raft_change *req = raft_malloc(sizeof(*req));
int r = raft_remove(&pRaftServer->raft, req, rid, raft_change_cb_remove);
if (r != 0) {
printf("raft_remove: %s \n", raft_errmsg(&pRaftServer->raft));
}
printf("drop node: %lu %s \n", rid, param1);
} else if (strcmp(cmd, "put") == 0) {
char buf[256];
......@@ -234,6 +273,7 @@ void console(SRaftServer *pRaftServer) {
} else if (strcmp(cmd, "help") == 0) {
printf("addnode \"127.0.0.1:8888\" \n");
printf("activate \"127.0.0.1:8888\" \n");
printf("dropnode \"127.0.0.1:8888\" \n");
printf("put key value \n");
printf("get key \n");
......@@ -256,7 +296,9 @@ void *startConsoleFunc(void *param) {
// Config ---------------------------------
void usage() {
printf("\nusage: \n");
printf("%s --me=127.0.0.1:10000 --dir=./data \n", exe_name);
printf("%s --me=127.0.0.1:10000 --dir=./data --voter \n", exe_name);
printf("%s --me=127.0.0.1:10001 --dir=./data \n", exe_name);
printf("%s --me=127.0.0.1:10002 --dir=./data \n", exe_name);
printf("\n");
printf("%s --me=127.0.0.1:10000 --peers=127.0.0.1:10001,127.0.0.1:10002 --dir=./data \n", exe_name);
printf("%s --me=127.0.0.1:10001 --peers=127.0.0.1:10000,127.0.0.1:10002 --dir=./data \n", exe_name);
......@@ -271,13 +313,15 @@ void parseConf(int argc, char **argv, SRaftServerConfig *pConf) {
option_index = 0;
static struct option long_options[] = {
{"help", no_argument, NULL, 'h'},
{"voter", no_argument, NULL, 'v'},
{"peers", required_argument, NULL, 'p'},
{"me", required_argument, NULL, 'm'},
{"dir", required_argument, NULL, 'd'},
{NULL, 0, NULL, 0}
};
while ((option_value = getopt_long(argc, argv, "hp:m:d:", long_options, &option_index)) != -1) {
pConf->voter = 0;
while ((option_value = getopt_long(argc, argv, "hvp:m:d:", long_options, &option_index)) != -1) {
switch (option_value) {
case 'm': {
parseAddr(optarg, pConf->me.host, sizeof(pConf->me.host), &pConf->me.port);
......@@ -295,6 +339,10 @@ void parseConf(int argc, char **argv, SRaftServerConfig *pConf) {
break;
}
case 'v': {
pConf->voter = 1;
break;
}
case 'd': {
snprintf(pConf->dir, sizeof(pConf->dir), "%s", optarg);
......@@ -338,6 +386,8 @@ int main(int argc, char **argv) {
exit(-1);
}
signal(SIGPIPE, SIG_IGN);
SRaftServerConfig conf;
parseConf(argc, argv, &conf);
printConf(&conf);
......
......@@ -85,29 +85,45 @@ int32_t raftServerInit(SRaftServer *pRaftServer, const SRaftServerConfig *pConf,
pRaftServer->fsm = pFsm;
ret = uv_loop_init(&pRaftServer->loop);
if (!ret) {
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
assert(0);
}
ret = raft_uv_tcp_init(&pRaftServer->transport, &pRaftServer->loop);
if (!ret) {
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
assert(0);
}
ret = raft_uv_init(&pRaftServer->io, &pRaftServer->loop, pRaftServer->dir, &pRaftServer->transport);
if (!ret) {
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
assert(0);
}
ret = raft_init(&pRaftServer->raft, &pRaftServer->io, pRaftServer->fsm, pRaftServer->raftId, pRaftServer->address);
if (!ret) {
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
assert(0);
}
struct raft_configuration conf;
raft_configuration_init(&conf);
raft_configuration_add(&conf, pRaftServer->raftId, pRaftServer->address, RAFT_VOTER);
if (pConf->voter == 0) {
raft_configuration_add(&conf, pRaftServer->raftId, pRaftServer->address, RAFT_SPARE);
} else {
raft_configuration_add(&conf, pRaftServer->raftId, pRaftServer->address, RAFT_VOTER);
}
printf("add myself: %llu - %s \n", pRaftServer->raftId, pRaftServer->address);
for (int i = 0; i < pConf->peersCount; ++i) {
const Addr *pAddr = &pConf->peers[i];
raft_id rid = raftId(pAddr->host, pAddr->port);
......
add_executable(raftMain "")
target_sources(raftMain
PRIVATE
"raftMain.c"
"raftServer.c"
)
target_link_libraries(raftMain PUBLIC traft lz4 uv_a)
#!/bin/bash
rm -rf 127.0.0.1*
rm -rf ./data
#ifndef TDENGINE_COMMON_H
#define TDENGINE_COMMON_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
#define MAX_INSTANCE_NUM 100
#define MAX_PEERS 10
#define COMMAND_LEN 1024
#define TOKEN_LEN 128
#define DIR_LEN 256
#define HOST_LEN 64
#define ADDRESS_LEN (HOST_LEN + 16)
typedef struct {
char host[HOST_LEN];
uint32_t port;
} Addr;
typedef struct {
Addr me;
Addr peers[MAX_PEERS];
int peersCount;
char dir[DIR_LEN];
char dataDir[DIR_LEN + HOST_LEN * 2];
} SRaftServerConfig;
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_COMMON_H
make raftServer
all:
gcc raftMain.c raftServer.c -I ../../traft/include/ ../../traft/.libs/libraft.a -o raftMain -luv -llz4 -lpthread -g
clean:
rm -f raftMain
sh clear.sh
make traft:
sudo apt-get install libuv1-dev liblz4-dev
autoreconf -i
./configure --enable-example
make
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include <assert.h>
#include <getopt.h>
#include <time.h>
#include <stdlib.h>
#include <getopt.h>
#include <raft.h>
#include <raft/uv.h>
#include "raftServer.h"
#include "common.h"
const char *exe_name;
typedef struct LeaderState {
char address[48];
int leaderCount;
} LeaderState;
#define NODE_COUNT 3
LeaderState leaderStates[NODE_COUNT];
void printLeaderCount() {
for (int i = 0; i < NODE_COUNT; ++i) {
printf("%s: leaderCount:%d \n", leaderStates[i].address, leaderStates[i].leaderCount);
}
}
void updateLeaderStates(SRaftServer *pRaftServer) {
for (int i = 0; i < pRaftServer->instance[0].raft.configuration.n; ++i) {
snprintf(leaderStates[i].address, sizeof(leaderStates[i].address), "%s", pRaftServer->instance[0].raft.configuration.servers[i].address);
leaderStates[i].leaderCount = 0;
}
for (int i = 0; i < pRaftServer->instanceCount; ++i) {
struct raft *r = &pRaftServer->instance[i].raft;
char leaderAddress[128];
memset(leaderAddress, 0, sizeof(leaderAddress));
if (r->state == RAFT_LEADER) {
snprintf(leaderAddress, sizeof(leaderAddress), "%s", r->address);
} else if (r->state == RAFT_FOLLOWER) {
snprintf(leaderAddress, sizeof(leaderAddress), "%s", r->follower_state.current_leader.address);
}
for (int j = 0; j < NODE_COUNT; j++) {
if (strcmp(leaderAddress, leaderStates[j].address) == 0) {
leaderStates[j].leaderCount++;
}
}
}
}
void raftTransferCb(struct raft_transfer *req) {
SRaftServer *pRaftServer = req->data;
raft_free(req);
printf("raftTransferCb: \n");
updateLeaderStates(pRaftServer);
printLeaderCount();
int myLeaderCount;
for (int i = 0; i < NODE_COUNT; ++i) {
if (strcmp(pRaftServer->address, leaderStates[i].address) == 0) {
myLeaderCount = leaderStates[i].leaderCount;
}
}
printf("myLeaderCount:%d waterLevel:%d \n", myLeaderCount, pRaftServer->instanceCount / NODE_COUNT);
if (myLeaderCount > pRaftServer->instanceCount / NODE_COUNT) {
struct raft *r;
for (int j = 0; j < pRaftServer->instanceCount; ++j) {
if (pRaftServer->instance[j].raft.state == RAFT_LEADER) {
r = &pRaftServer->instance[j].raft;
}
}
struct raft_transfer *transfer = raft_malloc(sizeof(*transfer));
transfer->data = pRaftServer;
uint64_t destRaftId;
int minIndex = -1;
int minLeaderCount = myLeaderCount;
for (int j = 0; j < NODE_COUNT; ++j) {
if (strcmp(leaderStates[j].address, pRaftServer->address) == 0) continue;
if (leaderStates[j].leaderCount <= minLeaderCount) {
minIndex = j;
}
}
char myHost[48];
uint16_t myPort;
uint16_t myVid;
decodeRaftId(r->id, myHost, sizeof(myHost), &myPort, &myVid);
char *destAddress = leaderStates[minIndex].address;
char tokens[MAX_PEERS][MAX_TOKEN_LEN];
splitString(destAddress, ":", tokens, 2);
char *destHost = tokens[0];
uint16_t destPort = atoi(tokens[1]);
destRaftId = encodeRaftId(destHost, destPort, myVid);
raft_transfer(r, transfer, destRaftId, raftTransferCb);
}
}
void parseAddr(const char *addr, char *host, int len, uint32_t *port) {
char* tmp = (char*)malloc(strlen(addr) + 1);
strcpy(tmp, addr);
char* context;
char* separator = ":";
char* token = strtok_r(tmp, separator, &context);
if (token) {
snprintf(host, len, "%s", token);
}
token = strtok_r(NULL, separator, &context);
if (token) {
sscanf(token, "%u", port);
}
free(tmp);
}
// only parse 3 tokens
int parseCommand3(const char* str, char* token1, char* token2, char* token3, int len)
{
char* tmp = (char*)malloc(strlen(str) + 1);
strcpy(tmp, str);
char* context;
char* separator = " ";
int n = 0;
char* token = strtok_r(tmp, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token1, token, len);
n++;
}
token = strtok_r(NULL, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token2, token, len);
n++;
}
token = strtok_r(NULL, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token3, token, len);
n++;
}
ret:
return n;
free(tmp);
}
// only parse 4 tokens
int parseCommand4(const char* str, char* token1, char* token2, char* token3, char *token4, int len)
{
char* tmp = (char*)malloc(strlen(str) + 1);
strcpy(tmp, str);
char* context;
char* separator = " ";
int n = 0;
char* token = strtok_r(tmp, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token1, token, len);
n++;
}
token = strtok_r(NULL, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token2, token, len);
n++;
}
token = strtok_r(NULL, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token3, token, len);
n++;
}
token = strtok_r(NULL, separator, &context);
if (!token) {
goto ret;
}
if (strcmp(token, "") != 0) {
strncpy(token4, token, len);
n++;
}
ret:
return n;
free(tmp);
}
void *startServerFunc(void *param) {
SRaftServer *pServer = (SRaftServer*)param;
int32_t r = raftServerStart(pServer);
assert(r == 0);
return NULL;
}
// Console ---------------------------------
const char* state2String(unsigned short state) {
if (state == RAFT_UNAVAILABLE) {
return "RAFT_UNAVAILABLE";
} else if (state == RAFT_FOLLOWER) {
return "RAFT_FOLLOWER";
} else if (state == RAFT_CANDIDATE) {
return "RAFT_CANDIDATE";
} else if (state == RAFT_LEADER) {
return "RAFT_LEADER";
}
return "UNKNOWN_RAFT_STATE";
}
void printRaftState2(struct raft *r) {
char leaderAddress[128];
memset(leaderAddress, 0, sizeof(leaderAddress));
if (r->state == RAFT_LEADER) {
snprintf(leaderAddress, sizeof(leaderAddress), "%s", r->address);
} else if (r->state == RAFT_FOLLOWER) {
snprintf(leaderAddress, sizeof(leaderAddress), "%s", r->follower_state.current_leader.address);
}
for (int i = 0; i < r->configuration.n; ++i) {
char tmpAddress[128];
snprintf(tmpAddress, sizeof(tmpAddress), "%s", r->configuration.servers[i].address);
uint64_t raftId = r->configuration.servers[i].id;
char host[128];
uint16_t port;
uint16_t vid;
decodeRaftId(raftId, host, 128, &port, &vid);
char buf[512];
memset(buf, 0, sizeof(buf));
if (strcmp(tmpAddress, leaderAddress) == 0) {
snprintf(buf, sizeof(buf), "<%s:%u-%u-LEADER>\t", host, port, vid);
} else {
snprintf(buf, sizeof(buf), "<%s:%u-%u-FOLLOWER>\t", host, port, vid);
}
printf("%s", buf);
}
printf("\n");
}
void printRaftConfiguration(struct raft_configuration *c) {
printf("configuration: \n");
for (int i = 0; i < c->n; ++i) {
printf("%llu -- %d -- %s\n", c->servers[i].id, c->servers[i].role, c->servers[i].address);
}
}
void printRaftState(struct raft *r) {
printf("----Raft State: -----------\n");
printf("mem_addr: %p \n", r);
printf("my_id: %llu \n", r->id);
printf("address: %s \n", r->address);
printf("current_term: %llu \n", r->current_term);
printf("voted_for: %llu \n", r->voted_for);
printf("role: %s \n", state2String(r->state));
printf("commit_index: %llu \n", r->commit_index);
printf("last_applied: %llu \n", r->last_applied);
printf("last_stored: %llu \n", r->last_stored);
printf("configuration_index: %llu \n", r->configuration_index);
printf("configuration_uncommitted_index: %llu \n", r->configuration_uncommitted_index);
printRaftConfiguration(&r->configuration);
printf("----------------------------\n");
}
void putValueCb(struct raft_apply *req, int status, void *result) {
raft_free(req);
struct raft *r = req->data;
if (status != 0) {
printf("putValueCb: %s \n", raft_errmsg(r));
} else {
printf("putValueCb: %s \n", "ok");
}
}
void putValue(struct raft *r, const char *value) {
struct raft_buffer buf;
buf.len = TOKEN_LEN;;
buf.base = raft_malloc(buf.len);
snprintf(buf.base, buf.len, "%s", value);
struct raft_apply *req = raft_malloc(sizeof(struct raft_apply));
req->data = r;
int ret = raft_apply(r, req, &buf, 1, putValueCb);
if (ret == 0) {
printf("put %s \n", (char*)buf.base);
} else {
printf("put error: %s \n", raft_errmsg(r));
}
}
void getValue(const char *key) {
char *ptr = getKV(key);
if (ptr) {
printf("get value: [%s] \n", ptr);
} else {
printf("value not found for key: [%s] \n", key);
}
}
void console(SRaftServer *pRaftServer) {
while (1) {
char cmd_buf[COMMAND_LEN];
memset(cmd_buf, 0, sizeof(cmd_buf));
char *ret = fgets(cmd_buf, COMMAND_LEN, stdin);
if (!ret) {
exit(-1);
}
int pos = strlen(cmd_buf);
if(cmd_buf[pos - 1] == '\n') {
cmd_buf[pos - 1] = '\0';
}
if (strncmp(cmd_buf, "", COMMAND_LEN) == 0) {
continue;
}
char cmd[TOKEN_LEN];
memset(cmd, 0, sizeof(cmd));
char param1[TOKEN_LEN];
memset(param1, 0, sizeof(param1));
char param2[TOKEN_LEN];
memset(param2, 0, sizeof(param2));
char param3[TOKEN_LEN];
memset(param2, 0, sizeof(param2));
parseCommand4(cmd_buf, cmd, param1, param2, param3, TOKEN_LEN);
if (strcmp(cmd, "addnode") == 0) {
printf("not support \n");
/*
char host[HOST_LEN];
uint32_t port;
parseAddr(param1, host, HOST_LEN, &port);
uint64_t rid = raftId(host, port);
struct raft_change *req = raft_malloc(sizeof(*req));
int r = raft_add(&pRaftServer->raft, req, rid, param1, NULL);
if (r != 0) {
printf("raft_add: %s \n", raft_errmsg(&pRaftServer->raft));
}
printf("add node: %lu %s \n", rid, param1);
struct raft_change *req2 = raft_malloc(sizeof(*req2));
r = raft_assign(&pRaftServer->raft, req2, rid, RAFT_VOTER, NULL);
if (r != 0) {
printf("raft_assign: %s \n", raft_errmsg(&pRaftServer->raft));
}
*/
} else if (strcmp(cmd, "dropnode") == 0) {
printf("not support \n");
} else if (strcmp(cmd, "rebalance") == 0) {
/*
updateLeaderStates(pRaftServer);
int myLeaderCount;
for (int i = 0; i < NODE_COUNT; ++i) {
if (strcmp(pRaftServer->address, leaderStates[i].address) == 0) {
myLeaderCount = leaderStates[i].leaderCount;
}
}
while (myLeaderCount > pRaftServer->instanceCount / NODE_COUNT) {
printf("myLeaderCount:%d waterLevel:%d \n", myLeaderCount, pRaftServer->instanceCount / NODE_COUNT);
struct raft *r;
for (int j = 0; j < pRaftServer->instanceCount; ++j) {
if (pRaftServer->instance[j].raft.state == RAFT_LEADER) {
r = &pRaftServer->instance[j].raft;
}
}
struct raft_transfer *transfer = raft_malloc(sizeof(*transfer));
transfer->data = pRaftServer;
uint64_t destRaftId;
int minIndex = -1;
int minLeaderCount = myLeaderCount;
for (int j = 0; j < NODE_COUNT; ++j) {
if (strcmp(leaderStates[j].address, pRaftServer->address) == 0) continue;
printf("-----leaderStates[%d].leaderCount:%d \n", j, leaderStates[j].leaderCount);
if (leaderStates[j].leaderCount <= minLeaderCount) {
minIndex = j;
printf("++++ assign minIndex : %d \n", minIndex);
}
}
printf("minIndex:%d minLeaderCount:%d \n", minIndex, minLeaderCount);
char myHost[48];
uint16_t myPort;
uint16_t myVid;
decodeRaftId(r->id, myHost, sizeof(myHost), &myPort, &myVid);
char *destAddress = leaderStates[minIndex].address;
char tokens[MAX_PEERS][MAX_TOKEN_LEN];
splitString(destAddress, ":", tokens, 2);
char *destHost = tokens[0];
uint16_t destPort = atoi(tokens[1]);
destRaftId = encodeRaftId(destHost, destPort, myVid);
printf("destHost:%s destPort:%u myVid:%u", destHost, destPort, myVid);
raft_transfer(r, transfer, destRaftId, raftTransferCb);
sleep(1);
for (int i = 0; i < NODE_COUNT; ++i) {
if (strcmp(pRaftServer->address, leaderStates[i].address) == 0) {
myLeaderCount = leaderStates[i].leaderCount;
}
}
}
*/
int leaderCount = 0;
struct raft *firstR;
for (int i = 0; i < pRaftServer->instanceCount; ++i) {
struct raft *r = &pRaftServer->instance[i].raft;
if (r->state == RAFT_LEADER) {
leaderCount++;
firstR = r;
}
}
if (leaderCount > pRaftServer->instanceCount / NODE_COUNT) {
struct raft_transfer *transfer = raft_malloc(sizeof(*transfer));
transfer->data = pRaftServer;
raft_transfer(firstR, transfer, 0, raftTransferCb);
}
} else if (strcmp(cmd, "put") == 0) {
char buf[256];
uint16_t vid;
sscanf(param1, "%hu", &vid);
snprintf(buf, sizeof(buf), "%s--%s", param2, param3);
putValue(&pRaftServer->instance[vid].raft, buf);
} else if (strcmp(cmd, "get") == 0) {
getValue(param1);
} else if (strcmp(cmd, "transfer") == 0) {
uint16_t vid;
sscanf(param1, "%hu", &vid);
struct raft_transfer transfer;
raft_transfer(&pRaftServer->instance[vid].raft, &transfer, 0, NULL);
} else if (strcmp(cmd, "state") == 0) {
for (int i = 0; i < pRaftServer->instanceCount; ++i) {
printf("instance %d: ", i);
printRaftState(&pRaftServer->instance[i].raft);
}
} else if (strcmp(cmd, "state2") == 0) {
for (int i = 0; i < pRaftServer->instanceCount; ++i) {
printRaftState2(&pRaftServer->instance[i].raft);
}
} else if (strcmp(cmd, "snapshot") == 0) {
printf("not support \n");
} else if (strcmp(cmd, "help") == 0) {
printf("addnode \"127.0.0.1:8888\" \n");
printf("dropnode \"127.0.0.1:8888\" \n");
printf("put key value \n");
printf("get key \n");
printf("state \n");
} else {
printf("unknown command: [%s], type \"help\" to see help \n", cmd);
}
//printf("cmd_buf: [%s] \n", cmd_buf);
}
}
void *startConsoleFunc(void *param) {
SRaftServer *pServer = (SRaftServer*)param;
console(pServer);
return NULL;
}
// Config ---------------------------------
void usage() {
printf("\nusage: \n");
printf("%s --me=127.0.0.1:10000 --dir=./data \n", exe_name);
printf("\n");
printf("%s --me=127.0.0.1:10000 --peers=127.0.0.1:10001,127.0.0.1:10002 --dir=./data \n", exe_name);
printf("%s --me=127.0.0.1:10001 --peers=127.0.0.1:10000,127.0.0.1:10002 --dir=./data \n", exe_name);
printf("%s --me=127.0.0.1:10002 --peers=127.0.0.1:10000,127.0.0.1:10001 --dir=./data \n", exe_name);
printf("\n");
}
void parseConf(int argc, char **argv, SRaftServerConfig *pConf) {
memset(pConf, 0, sizeof(*pConf));
int option_index, option_value;
option_index = 0;
static struct option long_options[] = {
{"help", no_argument, NULL, 'h'},
{"peers", required_argument, NULL, 'p'},
{"me", required_argument, NULL, 'm'},
{"dir", required_argument, NULL, 'd'},
{NULL, 0, NULL, 0}
};
while ((option_value = getopt_long(argc, argv, "hp:m:d:", long_options, &option_index)) != -1) {
switch (option_value) {
case 'm': {
parseAddr(optarg, pConf->me.host, sizeof(pConf->me.host), &pConf->me.port);
break;
}
case 'p': {
char tokens[MAX_PEERS][MAX_TOKEN_LEN];
int peerCount = splitString(optarg, ",", tokens, MAX_PEERS);
pConf->peersCount = peerCount;
for (int i = 0; i < peerCount; ++i) {
Addr *pAddr = &pConf->peers[i];
parseAddr(tokens[i], pAddr->host, sizeof(pAddr->host), &pAddr->port);
}
break;
}
case 'd': {
snprintf(pConf->dir, sizeof(pConf->dir), "%s", optarg);
break;
}
case 'h': {
usage();
exit(-1);
}
default: {
usage();
exit(-1);
}
}
}
snprintf(pConf->dataDir, sizeof(pConf->dataDir), "%s/%s_%u", pConf->dir, pConf->me.host, pConf->me.port);
}
void printConf(SRaftServerConfig *pConf) {
printf("\nconf: \n");
printf("me: %s:%u \n", pConf->me.host, pConf->me.port);
printf("peersCount: %d \n", pConf->peersCount);
for (int i = 0; i < pConf->peersCount; ++i) {
Addr *pAddr = &pConf->peers[i];
printf("peer%d: %s:%u \n", i, pAddr->host, pAddr->port);
}
printf("dataDir: %s \n\n", pConf->dataDir);
}
int main(int argc, char **argv) {
srand(time(NULL));
int32_t ret;
exe_name = argv[0];
if (argc < 3) {
usage();
exit(-1);
}
SRaftServerConfig conf;
parseConf(argc, argv, &conf);
printConf(&conf);
signal(SIGPIPE, SIG_IGN);
/*
char cmd_buf[COMMAND_LEN];
snprintf(cmd_buf, sizeof(cmd_buf), "mkdir -p %s", conf.dataDir);
system(cmd_buf);
*/
struct raft_fsm fsm;
initFsm(&fsm);
SRaftServer raftServer;
ret = raftServerInit(&raftServer, &conf, &fsm);
assert(ret == 0);
pthread_t tidRaftServer;
pthread_create(&tidRaftServer, NULL, startServerFunc, &raftServer);
pthread_t tidConsole;
pthread_create(&tidConsole, NULL, startConsoleFunc, &raftServer);
while (1) {
sleep(10);
}
return 0;
}
#include <unistd.h>
#include <stdlib.h>
#include "common.h"
#include "raftServer.h"
char *keys;
char *values;
void initStore() {
keys = malloc(MAX_RECORD_COUNT * MAX_KV_LEN);
values = malloc(MAX_RECORD_COUNT * MAX_KV_LEN);
writeIndex = 0;
}
void destroyStore() {
free(keys);
free(values);
}
void putKV(const char *key, const char *value) {
if (writeIndex < MAX_RECORD_COUNT) {
strncpy(&keys[writeIndex], key, MAX_KV_LEN);
strncpy(&values[writeIndex], value, MAX_KV_LEN);
writeIndex++;
}
}
char *getKV(const char *key) {
for (int i = 0; i < MAX_RECORD_COUNT; ++i) {
if (strcmp(&keys[i], key) == 0) {
return &values[i];
}
}
return NULL;
}
int splitString(const char* str, char* separator, char (*arr)[MAX_TOKEN_LEN], int n_arr)
{
if (n_arr <= 0) {
return -1;
}
char* tmp = (char*)malloc(strlen(str) + 1);
strcpy(tmp, str);
char* context;
int n = 0;
char* token = strtok_r(tmp, separator, &context);
if (!token) {
goto ret;
}
strncpy(arr[n], token, MAX_TOKEN_LEN);
n++;
while (1) {
token = strtok_r(NULL, separator, &context);
if (!token || n >= n_arr) {
goto ret;
}
strncpy(arr[n], token, MAX_TOKEN_LEN);
n++;
}
ret:
free(tmp);
return n;
}
/*
uint64_t raftId(const char *host, uint32_t port) {
uint32_t host_uint32 = (uint32_t)inet_addr(host);
assert(host_uint32 != (uint32_t)-1);
uint64_t code = ((uint64_t)host_uint32) << 32 | port;
return code;
}
*/
/*
uint64_t encodeRaftId(const char *host, uint16_t port, uint16_t vid) {
uint64_t raftId;
uint32_t host_uint32 = (uint32_t)inet_addr(host);
assert(host_uint32 != (uint32_t)-1);
raftId = (((uint64_t)host_uint32) << 32) | (((uint32_t)port) << 16) | vid;
return raftId;
}
void decodeRaftId(uint64_t raftId, char *host, int32_t len, uint16_t *port, uint16_t *vid) {
uint32_t host32 = (uint32_t)((raftId >> 32) & 0x00000000FFFFFFFF);
struct in_addr addr;
addr.s_addr = host32;
snprintf(host, len, "%s", inet_ntoa(addr));
*port = (uint16_t)((raftId & 0x00000000FFFF0000) >> 16);
*vid = (uint16_t)(raftId & 0x000000000000FFFF);
}
*/
int32_t raftServerInit(SRaftServer *pRaftServer, const SRaftServerConfig *pConf, struct raft_fsm *pFsm) {
int ret;
snprintf(pRaftServer->host, sizeof(pRaftServer->host), "%s", pConf->me.host);
pRaftServer->port = pConf->me.port;
snprintf(pRaftServer->address, sizeof(pRaftServer->address), "%s:%u", pRaftServer->host, pRaftServer->port);
//strncpy(pRaftServer->dir, pConf->dataDir, sizeof(pRaftServer->dir));
ret = uv_loop_init(&pRaftServer->loop);
if (ret != 0) {
fprintf(stderr, "uv_loop_init error: %s \n", uv_strerror(ret));
assert(0);
}
ret = raft_uv_tcp_init(&pRaftServer->transport, &pRaftServer->loop);
if (ret != 0) {
fprintf(stderr, "raft_uv_tcp_init: error %d \n", ret);
assert(0);
}
uint16_t vid;
pRaftServer->instanceCount = 20;
for (int i = 0; i < pRaftServer->instanceCount; ++i)
{
//vid = 0;
vid = i;
pRaftServer->instance[vid].raftId = encodeRaftId(pRaftServer->host, pRaftServer->port, vid);
snprintf(pRaftServer->instance[vid].dir, sizeof(pRaftServer->instance[vid].dir), "%s_%llu", pConf->dataDir, pRaftServer->instance[vid].raftId);
char cmd_buf[COMMAND_LEN];
snprintf(cmd_buf, sizeof(cmd_buf), "mkdir -p %s", pRaftServer->instance[vid].dir);
system(cmd_buf);
sleep(1);
pRaftServer->instance[vid].fsm = pFsm;
ret = raft_uv_init(&pRaftServer->instance[vid].io, &pRaftServer->loop, pRaftServer->instance[vid].dir, &pRaftServer->transport);
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->instance[vid].raft));
assert(0);
}
ret = raft_init(&pRaftServer->instance[vid].raft, &pRaftServer->instance[vid].io, pRaftServer->instance[vid].fsm, pRaftServer->instance[vid].raftId, pRaftServer->address);
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->instance[vid].raft));
assert(0);
}
struct raft_configuration conf;
raft_configuration_init(&conf);
raft_configuration_add(&conf, pRaftServer->instance[vid].raftId, pRaftServer->address, RAFT_VOTER);
printf("add myself: %llu - %s \n", pRaftServer->instance[vid].raftId, pRaftServer->address);
for (int i = 0; i < pConf->peersCount; ++i) {
const Addr *pAddr = &pConf->peers[i];
raft_id rid = encodeRaftId(pAddr->host, pAddr->port, vid);
char addrBuf[ADDRESS_LEN];
snprintf(addrBuf, sizeof(addrBuf), "%s:%u", pAddr->host, pAddr->port);
raft_configuration_add(&conf, rid, addrBuf, RAFT_VOTER);
printf("add peers: %llu - %s \n", rid, addrBuf);
}
raft_bootstrap(&pRaftServer->instance[vid].raft, &conf);
}
return 0;
}
int32_t raftServerStart(SRaftServer *pRaftServer) {
int ret;
for (int i = 0; i < pRaftServer->instanceCount; ++i) {
ret = raft_start(&pRaftServer->instance[i].raft);
if (ret != 0) {
fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->instance[i].raft));
}
}
uv_run(&pRaftServer->loop, UV_RUN_DEFAULT);
}
void raftServerClose(SRaftServer *pRaftServer) {
}
int fsmApplyCb(struct raft_fsm *pFsm, const struct raft_buffer *buf, void **result) {
char *msg = (char*)buf->base;
printf("fsm apply: %s \n", msg);
char arr[2][MAX_TOKEN_LEN];
splitString(msg, "--", arr, 2);
putKV(arr[0], arr[1]);
return 0;
}
int32_t initFsm(struct raft_fsm *fsm) {
initStore();
fsm->apply = fsmApplyCb;
return 0;
}
#ifndef TDENGINE_RAFT_SERVER_H
#define TDENGINE_RAFT_SERVER_H
#ifdef __cplusplus
extern "C" {
#endif
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <string.h>
#include "raft.h"
#include "raft/uv.h"
#include "common.h"
// simulate a db store, just for test
#define MAX_KV_LEN 100
#define MAX_RECORD_COUNT 500
char *keys;
char *values;
int writeIndex;
void initStore();
void destroyStore();
void putKV(const char *key, const char *value);
char *getKV(const char *key);
typedef struct {
char dir[DIR_LEN + HOST_LEN * 4]; /* Data dir of UV I/O backend */
raft_id raftId; /* For vote */
struct raft_fsm *fsm; /* Sample application FSM */
struct raft raft; /* Raft instance */
struct raft_io io; /* UV I/O backend */
} SInstance;
typedef struct {
char host[HOST_LEN];
uint32_t port;
char address[ADDRESS_LEN]; /* Raft instance address */
struct uv_loop_s loop; /* UV loop */
struct raft_uv_transport transport; /* UV I/O backend transport */
SInstance instance[MAX_INSTANCE_NUM];
int32_t instanceCount;
} SRaftServer;
#define MAX_TOKEN_LEN 32
int splitString(const char* str, char* separator, char (*arr)[MAX_TOKEN_LEN], int n_arr);
int32_t raftServerInit(SRaftServer *pRaftServer, const SRaftServerConfig *pConf, struct raft_fsm *pFsm);
int32_t raftServerStart(SRaftServer *pRaftServer);
void raftServerClose(SRaftServer *pRaftServer);
int initFsm(struct raft_fsm *fsm);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_RAFT_SERVER_H
......@@ -30,11 +30,12 @@ extern char tsLocalEp[];
extern uint16_t tsServerPort;
extern int32_t tsStatusInterval;
extern int8_t tsEnableTelemetryReporting;
extern int32_t tsNumOfSupportVnodes;
// common
extern int tsRpcTimer;
extern int tsRpcMaxTime;
extern int tsRpcForceTcp; // all commands go to tcp protocol if this is enabled
extern int tsRpcForceTcp; // all commands go to tcp protocol if this is enabled
extern int32_t tsMaxConnections;
extern int32_t tsMaxShellConns;
extern int32_t tsShellActivityTimer;
......@@ -48,14 +49,18 @@ extern int32_t tsCompressMsgSize;
extern int32_t tsCompressColData;
extern int32_t tsMaxNumOfDistinctResults;
extern char tsTempDir[];
extern int64_t tsMaxVnodeQueuedBytes;
extern int tsCompatibleModel; // 2.0 compatible model
//query buffer management
extern int32_t tsQueryBufferSize; // maximum allowed usage buffer size in MB for each data node during query processing
extern int64_t tsQueryBufferSizeBytes; // maximum allowed usage buffer size in byte for each data node during query processing
extern int32_t tsRetrieveBlockingModel;// retrieve threads will be blocked
extern int8_t tsKeepOriginalColumnName;
extern int tsCompatibleModel; // 2.0 compatible model
extern int8_t tsEnableSlaveQuery;
extern int8_t tsEnableAdjustMaster;
extern int8_t tsPrintAuth;
extern int64_t tsTickPerDay[3];
// query buffer management
extern int32_t tsQueryBufferSize; // maximum allowed usage buffer size in MB for each data node during query processing
extern int64_t tsQueryBufferSizeBytes; // maximum allowed usage buffer size in byte for each data node
extern int32_t tsRetrieveBlockingModel; // retrieve threads will be blocked
extern int8_t tsKeepOriginalColumnName;
extern int8_t tsDeadLockKillQuery;
// client
extern int32_t tsMaxSQLStringLen;
......@@ -72,27 +77,17 @@ extern float tsStreamComputDelayRatio; // the delayed computing ration of the
extern int32_t tsProjectExecInterval;
extern int64_t tsMaxRetentWindow;
// balance
extern int8_t tsEnableSlaveQuery;
// interna
extern int8_t tsPrintAuth;
extern char tsVnodeDir[];
extern char tsMnodeDir[];
extern int64_t tsTickPerDay[3];
// system info
extern float tsTotalLogDirGB;
extern float tsTotalTmpDirGB;
extern float tsTotalDataDirGB;
extern float tsAvailLogDirGB;
extern float tsAvailTmpDirectorySpace;
extern float tsAvailDataDirGB;
extern float tsUsedDataDirGB;
extern float tsMinimalLogDirGB;
extern float tsReservedTmpDirectorySpace;
extern float tsMinimalDataDirGB;
extern float tsTotalLogDirGB;
extern float tsTotalTmpDirGB;
extern float tsTotalDataDirGB;
extern float tsAvailLogDirGB;
extern float tsAvailTmpDirectorySpace;
extern float tsAvailDataDirGB;
extern float tsUsedDataDirGB;
extern float tsMinimalLogDirGB;
extern float tsReservedTmpDirectorySpace;
extern float tsMinimalDataDirGB;
extern uint32_t tsVersion;
// build info
......@@ -102,17 +97,13 @@ extern char gitinfo[];
extern char gitinfoOfInternal[];
extern char buildinfo[];
#ifdef TD_TSZ
// lossy
extern char lossyColumns[];
extern double fPrecision;
extern double dPrecision;
extern uint32_t maxRange;
extern uint32_t curRange;
extern char Compressor[];
#endif
// long query
extern int8_t tsDeadLockKillQuery;
// lossy
extern char tsLossyColumns[];
extern double tsFPrecision;
extern double tsDPrecision;
extern uint32_t tsMaxRange;
extern uint32_t tsCurRange;
extern char tsCompressor[];
typedef struct {
char dir[TSDB_FILENAME_LEN];
......
......@@ -23,9 +23,12 @@ extern "C" {
#include "encode.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tarray.h"
#include "tcoding.h"
#include "tdataformat.h"
#include "tlist.h"
/* ------------------------ MESSAGE DEFINITIONS ------------------------ */
#define TD_MSG_NUMBER_
#undef TD_MSG_DICT_
#undef TD_MSG_INFO_
......@@ -54,6 +57,7 @@ extern int tMsgDict[];
typedef uint16_t tmsg_t;
/* ------------------------ OTHER DEFINITIONS ------------------------ */
// IE type
#define TSDB_IE_TYPE_SEC 1
#define TSDB_IE_TYPE_META 2
......@@ -71,6 +75,9 @@ typedef enum _mgmt_table {
TSDB_MGMT_TABLE_TABLE,
TSDB_MGMT_TABLE_DNODE,
TSDB_MGMT_TABLE_MNODE,
TSDB_MGMT_TABLE_QNODE,
TSDB_MGMT_TABLE_SNODE,
TSDB_MGMT_TABLE_BNODE,
TSDB_MGMT_TABLE_VGROUP,
TSDB_MGMT_TABLE_STB,
TSDB_MGMT_TABLE_MODULE,
......@@ -127,6 +134,7 @@ typedef enum _mgmt_table {
typedef struct SBuildTableMetaInput {
int32_t vgId;
char* dbName;
char* tableFullName;
} SBuildTableMetaInput;
......@@ -262,19 +270,7 @@ typedef struct {
SMsgHead head;
char name[TSDB_TABLE_FNAME_LEN];
uint64_t suid;
int32_t sverson;
uint32_t ttl;
uint32_t keep;
int32_t numOfTags;
int32_t numOfColumns;
SSchema pSchema[];
} SCreateStbInternalMsg;
typedef struct {
SMsgHead head;
char name[TSDB_TABLE_FNAME_LEN];
uint64_t suid;
} SDropStbInternalMsg;
} SVDropStbReq;
typedef struct {
SMsgHead head;
......@@ -521,7 +517,7 @@ typedef struct {
typedef struct {
SMsgHead header;
union {
int32_t showId;
int64_t showId;
int64_t qhandle;
int64_t qId;
}; // query handle
......@@ -669,8 +665,8 @@ typedef struct {
int64_t clusterId;
int64_t rebootTime;
int64_t updateTime;
int16_t numOfCores;
int16_t numOfSupportVnodes;
int32_t numOfCores;
int32_t numOfSupportVnodes;
char dnodeEp[TSDB_EP_LEN];
SClusterCfg clusterCfg;
SVnodeLoads vnodeLoads;
......@@ -749,8 +745,9 @@ typedef struct {
} SAuthVnodeMsg;
typedef struct {
int32_t vgId;
char tableFname[TSDB_TABLE_FNAME_LEN];
SMsgHead header;
char dbFname[TSDB_DB_FNAME_LEN];
char tableFname[TSDB_TABLE_FNAME_LEN];
} STableInfoMsg;
typedef struct {
......@@ -784,6 +781,7 @@ typedef struct {
typedef struct {
char tbFname[TSDB_TABLE_FNAME_LEN]; // table full name
char stbFname[TSDB_TABLE_FNAME_LEN];
char dbFname[TSDB_DB_FNAME_LEN];
int32_t numOfTags;
int32_t numOfColumns;
int8_t precision;
......@@ -841,7 +839,7 @@ typedef struct {
} SCompactMsg;
typedef struct SShowRsp {
int32_t showId;
int64_t showId;
STableMetaMsg tableMeta;
} SShowRsp;
......@@ -861,29 +859,25 @@ typedef struct {
typedef struct {
int32_t dnodeId;
} SCreateMnodeMsg, SDropMnodeMsg;
} SMCreateMnodeMsg, SMDropMnodeMsg, SDDropMnodeMsg;
typedef struct {
int32_t dnodeId;
int8_t replica;
SReplica replicas[TSDB_MAX_REPLICA];
} SCreateMnodeInMsg, SAlterMnodeInMsg;
typedef struct {
int32_t dnodeId;
} SDropMnodeInMsg;
} SDCreateMnodeMsg, SDAlterMnodeMsg;
typedef struct {
int32_t dnodeId;
} SCreateQnodeInMsg, SDropQnodeInMsg;
} SMCreateQnodeMsg, SMDropQnodeMsg, SDCreateQnodeMsg, SDDropQnodeMsg;
typedef struct {
int32_t dnodeId;
} SCreateSnodeInMsg, SDropSnodeInMsg;
} SMCreateSnodeMsg, SMDropSnodeMsg, SDCreateSnodeMsg, SDDropSnodeMsg;
typedef struct {
int32_t dnodeId;
} SCreateBnodeInMsg, SDropBnodeInMsg;
} SMCreateBnodeMsg, SMDropBnodeMsg, SDCreateBnodeMsg, SDDropBnodeMsg;
typedef struct {
int32_t dnodeId;
......@@ -1038,6 +1032,7 @@ typedef struct {
} SUpdateTagValRsp;
typedef struct SSubQueryMsg {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
......@@ -1046,6 +1041,7 @@ typedef struct SSubQueryMsg {
} SSubQueryMsg;
typedef struct SResReadyMsg {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
......@@ -1056,6 +1052,7 @@ typedef struct SResReadyRsp {
} SResReadyRsp;
typedef struct SResFetchMsg {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
......@@ -1261,22 +1258,22 @@ typedef struct {
char* executor;
int32_t sqlLen;
char* sql;
} SCreateTopicInternalMsg;
} SDCreateTopicMsg;
typedef struct {
SMsgHead head;
char name[TSDB_TABLE_FNAME_LEN];
uint64_t tuid;
} SDropTopicInternalMsg;
} SDDropTopicMsg;
typedef struct SVCreateTbReq {
uint64_t ver; // use a general definition
char* name;
uint32_t ttl;
uint32_t keep;
#define TD_SUPER_TABLE 0
#define TD_CHILD_TABLE 1
#define TD_NORMAL_TABLE 2
#define TD_SUPER_TABLE TSDB_SUPER_TABLE
#define TD_CHILD_TABLE TSDB_CHILD_TABLE
#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE
uint8_t type;
union {
struct {
......@@ -1297,100 +1294,16 @@ typedef struct SVCreateTbReq {
};
} SVCreateTbReq;
static FORCE_INLINE int tSerializeSVCreateTbReq(void** buf, const SVCreateTbReq* pReq) {
int tlen = 0;
tlen += taosEncodeFixedU64(buf, pReq->ver);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep);
tlen += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case TD_SUPER_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
}
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return tlen;
}
typedef struct {
uint64_t ver; // use a general definition
SArray* pArray;
} SVCreateTbBatchReq;
static FORCE_INLINE void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case TD_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
pReq->stbCfg.pSchema = (SSchema*)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
}
buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
pReq->stbCfg.pTagSchema = (SSchema*)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
pReq->ntbCfg.pSchema = (SSchema*)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
int tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
int tSVCreateTbBatchReqSerialize(void** buf, SVCreateTbBatchReq* pReq);
void* tSVCreateTbBatchReqDeserialize(void* buf, SVCreateTbBatchReq* pReq);
return buf;
}
typedef struct SVCreateTbRsp {
} SVCreateTbRsp;
......@@ -1405,7 +1318,7 @@ typedef struct SVShowTablesRsp {
typedef struct SVShowTablesFetchReq {
SMsgHead head;
int64_t id;
int32_t id;
} SVShowTablesFetchReq;
typedef struct SVShowTablesFetchRsp {
......
......@@ -70,6 +70,15 @@ enum {
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_MNODE, "dnode-create-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_MNODE, "dnode-alter-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_MNODE, "dnode-drop-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_QNODE, "dnode-create-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_QNODE, "dnode-alter-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_QNODE, "dnode-drop-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_SNODE, "dnode-create-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_SNODE, "dnode-alter-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_SNODE, "dnode-drop-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_BNODE, "dnode-create-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_BNODE, "dnode-alter-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_BNODE, "dnode-drop-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "dnode-create-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_ALTER_VNODE, "dnode-alter-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL)
......@@ -90,9 +99,20 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_DROP_USER, "mnode-drop-user", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DNODE, "mnode-create-dnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CONFIG_DNODE, "mnode-config-dnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DNODE, "mnode-alter-dnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_DNODE, "mnode-drop-dnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "mnode-create-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "mnode-alter-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "mnode-drop-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "mnode-create-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "mnode-alter-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "mnode-drop-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SNODE, "mnode-create-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_SNODE, "mnode-alter-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_SNODE, "mnode-drop-snode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_BNODE, "mnode-create-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_BNODE, "mnode-alter-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_BNODE, "mnode-drop-bnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DB, "mnode-create-db", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_DB, "mnode-drop-db", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_USE_DB, "mnode-use-db", NULL, NULL)
......
......@@ -40,8 +40,9 @@ enum {
// the SQL below is for mgmt node
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_MGMT, "mgmt" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_DB, "create-db" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_STABLE, "create-stable" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_TABLE, "create-table" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_FUNCTION, "create-function" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_FUNCTION, "create-function" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_DROP_DB, "drop-db" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_DROP_TABLE, "drop-table" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_DROP_FUNCTION, "drop-function" )
......
......@@ -17,6 +17,7 @@
#define TDENGINE_TNAME_H
#include "tdef.h"
#include "tmsg.h"
#define TSDB_DB_NAME_T 1
#define TSDB_TABLE_NAME_T 2
......@@ -58,4 +59,6 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type);
int32_t tNameSetAcctId(SName* dst, int32_t acctId);
SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name);
#endif // TDENGINE_TNAME_H
......@@ -27,8 +27,8 @@ typedef struct SDnode SDnode;
typedef struct {
int32_t sver;
int16_t numOfCores;
int16_t numOfSupportVnodes;
int32_t numOfCores;
int32_t numOfSupportVnodes;
int16_t numOfCommitThreads;
int8_t enableTelem;
int32_t statusInterval;
......
......@@ -157,18 +157,21 @@ typedef enum {
SDB_TRANS = 1,
SDB_CLUSTER = 2,
SDB_MNODE = 3,
SDB_DNODE = 4,
SDB_USER = 5,
SDB_AUTH = 6,
SDB_ACCT = 7,
SDB_CONSUMER = 8,
SDB_CGROUP = 9,
SDB_TOPIC = 10,
SDB_VGROUP = 11,
SDB_STB = 12,
SDB_DB = 13,
SDB_FUNC = 14,
SDB_MAX = 15
SDB_QNODE = 4,
SDB_SNODE = 5,
SDB_BNODE = 6,
SDB_DNODE = 7,
SDB_USER = 8,
SDB_AUTH = 9,
SDB_ACCT = 10,
SDB_CONSUMER = 11,
SDB_CGROUP = 12,
SDB_TOPIC = 13,
SDB_VGROUP = 14,
SDB_STB = 15,
SDB_DB = 16,
SDB_FUNC = 17,
SDB_MAX = 18
} ESdbType;
typedef struct SSdb SSdb;
......@@ -178,6 +181,7 @@ typedef int32_t (*SdbDeleteFp)(SSdb *pSdb, void *pObj);
typedef int32_t (*SdbDeployFp)(SMnode *pMnode);
typedef SSdbRow *(*SdbDecodeFp)(SSdbRaw *pRaw);
typedef SSdbRaw *(*SdbEncodeFp)(void *pObj);
typedef bool (*sdbTraverseFp)(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3);
typedef struct {
ESdbType sdbType;
......@@ -276,7 +280,7 @@ void sdbRelease(SSdb *pSdb, void *pObj);
*
* @param pSdb The sdb object.
* @param type The type of the table.
* @param type The initial iterator of the table.
* @param pIter The initial iterator of the table.
* @param pObj The object of the row just fetched.
* @return void* The next iterator of the table.
*/
......@@ -286,11 +290,22 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj);
* @brief Cancel a traversal
*
* @param pSdb The sdb object.
* @param pIter The iterator of the table.
* @param type The initial iterator of table.
*/
void sdbCancelFetch(SSdb *pSdb, void *pIter);
/**
* @brief Traverse a sdb
*
* @param pSdb The sdb object.
* @param type The initial iterator of table.
* @param fp The function pointer.
* @param p1 The callback param.
* @param p2 The callback param.
* @param p3 The callback param.
*/
void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2, void *p3);
/**
* @brief Get the number of rows in the table
*
......
......@@ -25,18 +25,25 @@
extern "C" {
#endif
#define META_SUPER_TABLE TD_SUPER_TABLE
#define META_CHILD_TABLE TD_CHILD_TABLE
#define META_NORMAL_TABLE TD_NORMAL_TABLE
// Types exported
typedef struct SMeta SMeta;
#define META_SUPER_TABLE 0
#define META_CHILD_TABLE 1
#define META_NORMAL_TABLE 2
typedef struct SMetaCfg {
/// LRU cache size
uint64_t lruSize;
} SMetaCfg;
typedef struct {
uint32_t nCols;
SSchema *pSchema;
} SSchemaWrapper;
typedef struct SMTbCursor SMTbCursor;
typedef SVCreateTbReq STbCfg;
// SMeta operations
......@@ -48,7 +55,13 @@ int metaDropTable(SMeta *pMeta, tb_uid_t uid);
int metaCommit(SMeta *pMeta);
// For Query
int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg);
STbCfg * metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid);
STbCfg * metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline);
SMTbCursor *metaOpenTbCursor(SMeta *pMeta);
void metaCloseTbCursor(SMTbCursor *pTbCur);
char * metaTbCursorNext(SMTbCursor *pTbCur);
// Options
void metaOptionsInit(SMetaCfg *pMetaCfg);
......
......@@ -54,14 +54,26 @@ int32_t catalogInit(SCatalogCfg *cfg);
/**
* Get a cluster's catalog handle for all later operations.
* @param clusterId (input, end with \0)
* @param clusterId
* @param catalogHandle (output, NO need to free it)
* @return error code
*/
int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle);
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
/**
* Get a DB's all vgroup info.
* @param pCatalog (input, got with catalogGetHandle)
* @param pRpc (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pDBName (input, full db name)
* @param forceUpdate (input, force update db vgroup info from mnode)
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, int32_t forceUpdate, SArray** pVgroupList);
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
/**
......
......@@ -27,7 +27,7 @@ extern "C" {
#include "tname.h"
#include "tvariant.h"
/*
/**
* The first field of a node of any type is guaranteed to be the int16_t.
* Hence the type of any node can be gotten by casting it to SQueryNode.
*/
......@@ -157,7 +157,7 @@ typedef struct SVgDataBlocks {
typedef struct SInsertStmtInfo {
int16_t nodeType;
SArray* pDataBlocks; // data block for each vgroup, SArray<SVgDataBlocks*>.
int8_t schemaAttache; // denote if submit block is built with table schema or not
int8_t schemaAttache; // denote if submit block is built with table schema or not
uint8_t payloadType; // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
uint32_t insertType; // insert data from [file|sql statement| bound statement]
const char* sql; // current sql statement position
......@@ -169,6 +169,7 @@ typedef struct SDclStmtInfo {
SEpSet epSet;
char* pMsg;
int32_t msgLen;
void* pExtension; // todo remove it soon
} SDclStmtInfo;
#ifdef __cplusplus
......
......@@ -74,7 +74,6 @@ int32_t getExprFunctionLevel(const SQueryStmtInfo* pQueryInfo);
STableMetaInfo* getMetaInfo(const SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name);
int32_t getNewResColId();
void addIntoSourceParam(SSourceParam* pSourceParam, tExprNode* pNode, SColumn* pColumn);
......
......@@ -75,6 +75,7 @@ typedef struct STableMeta {
} STableMeta;
typedef struct SDBVgroupInfo {
SRWLatch lock;
int32_t vgVersion;
int8_t hashMethod;
SHashObj *vgInfo; //key:vgId, value:SVgroupInfo
......
......@@ -54,8 +54,11 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
void qWorkerDestroy(void **qWorkerMgmt);
int32_t qWorkerProcessShowMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessShowFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
void qWorkerDestroy(void **qWorkerMgmt);
#ifdef __cplusplus
}
......
......@@ -50,23 +50,37 @@ typedef struct SQueryProfileSummary {
uint64_t resultSize; // generated result size in Kb.
} SQueryProfileSummary;
typedef struct SQueryNodeAddr{
int32_t nodeId; //vgId or qnodeId
int8_t inUse;
int8_t numOfEps;
SEpAddrMsg epAddr[TSDB_MAX_REPLICA];
} SQueryNodeAddr;
typedef struct SQueryResult {
int32_t code;
uint64_t numOfRows;
int32_t msgSize;
char *msg;
} SQueryResult;
int32_t schedulerInit(SSchedulerCfg *cfg);
/**
* Process the query job, generated according to the query physical plan.
* This is a synchronized API, and is also thread-safety.
* @param qnodeList Qnode address list, element is SEpAddr
* @param nodeList Qnode/Vnode address list, element is SQueryNodeAddr
* @return
*/
int32_t scheduleExecJob(void *transport, SArray *qnodeList, SQueryDag* pDag, void** pJob, uint64_t *numOfRows);
int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob, SQueryResult *pRes);
/**
* Process the query job, generated according to the query physical plan.
* This is a asynchronized API, and is also thread-safety.
* @param qnodeList Qnode address list, element is SEpAddr
* @param nodeList Qnode/Vnode address list, element is SQueryNodeAddr
* @return
*/
int32_t scheduleAsyncExecJob(void *transport, SArray *qnodeList, SQueryDag* pDag, void** pJob);
int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob);
int32_t scheduleFetchRows(void *pJob, void **data);
......
......@@ -21,7 +21,6 @@ extern "C" {
#endif
extern char tsOsName[];
extern char tsDnodeDir[];
extern char tsDataDir[];
extern char tsLogDir[];
extern char tsScriptDir[];
......
此差异已折叠。
/*
* 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/>.
*/
#ifndef _TD_UTIL_FREELIST_H_
#define _TD_UTIL_FREELIST_H_
#include "os.h"
#include "tlist.h"
#ifdef __cplusplus
extern "C" {
#endif
struct SFreeListNode {
TD_SLIST_NODE(SFreeListNode);
char payload[];
};
typedef TD_SLIST(SFreeListNode) SFreeList;
#define TFL_MALLOC(SIZE, LIST) \
({ \
void *ptr = malloc((SIZE) + sizeof(struct SFreeListNode)); \
if (ptr) { \
TD_SLIST_PUSH((LIST), (struct SFreeListNode *)ptr); \
ptr = ((struct SFreeListNode *)ptr)->payload; \
} \
ptr; \
})
#define tFreeListInit(pFL) TD_SLIST_INIT(pFL)
static FORCE_INLINE void tFreeListClear(SFreeList *pFL) {
struct SFreeListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(pFL);
if (pNode == NULL) break;
TD_SLIST_POP(pFL);
free(pNode);
}
}
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_FREELIST_H_*/
\ No newline at end of file
......@@ -172,10 +172,15 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_INVALID_DNODE_EP TAOS_DEF_ERROR_CODE(0, 0x0347)
#define TSDB_CODE_MND_INVALID_DNODE_ID TAOS_DEF_ERROR_CODE(0, 0x0348)
// mnode-mnode
// mnode-node
#define TSDB_CODE_MND_MNODE_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0350)
#define TSDB_CODE_MND_MNODE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0351)
#define TSDB_CODE_MND_TOO_MANY_MNODES TAOS_DEF_ERROR_CODE(0, 0x0352)
#define TSDB_CODE_MND_QNODE_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0352)
#define TSDB_CODE_MND_QNODE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0353)
#define TSDB_CODE_MND_SNODE_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0354)
#define TSDB_CODE_MND_SNODE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0355)
#define TSDB_CODE_MND_BNODE_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0356)
#define TSDB_CODE_MND_BNODE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0357)
// mnode-acct
#define TSDB_CODE_MND_ACCT_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0360)
......
......@@ -38,6 +38,7 @@ int64_t tfOpenCreateWriteAppend(const char *pathname);
int64_t tfClose(int64_t tfd);
int64_t tfWrite(int64_t tfd, void *buf, int64_t count);
int64_t tfRead(int64_t tfd, void *buf, int64_t count);
int64_t tfPread(int64_t tfd, void *buf, int64_t count, int64_t offset);
int32_t tfFsync(int64_t tfd);
bool tfValid(int64_t tfd);
int64_t tfLseek(int64_t tfd, int64_t offset, int32_t whence);
......@@ -47,4 +48,4 @@ int32_t tfFtruncate(int64_t tfd, int64_t length);
}
#endif
#endif /*_TD_UTIL_FILE_H*/
#endif /*_TD_UTIL_FILE_H*/
......@@ -144,6 +144,16 @@ void *taosHashGet(SHashObj *pHashObj, const void *key, size_t keyLen);
*/
void *taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void* destBuf);
/**
* Clone the result to interval allocated buffer
* @param pHashObj
* @param key
* @param keyLen
* @param destBuf
* @return
*/
void* taosHashGetCloneExt(SHashObj *pHashObj, const void *key, size_t keyLen, void (*fp)(void *), void** d, size_t *sz);
/**
* remove item with the specified key
* @param pHashObj
......@@ -200,6 +210,26 @@ void taosHashCancelIterate(SHashObj *pHashObj, void *p);
*/
int32_t taosHashGetKey(void *data, void** key, size_t* keyLen);
/**
* return the payload data with the specified key(reference number added)
*
* @param pHashObj
* @param key
* @param keyLen
* @return
*/
void* taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen);
/**
* release the prevous acquired obj
*
* @param pHashObj
* @param data
* @return
*/
void taosHashRelease(SHashObj *pHashObj, void *p);
#ifdef __cplusplus
}
#endif
......
......@@ -37,7 +37,6 @@ extern int32_t mqttDebugFlag;
extern int32_t monDebugFlag;
extern int32_t uDebugFlag;
extern int32_t rpcDebugFlag;
extern int32_t odbcDebugFlag;
extern int32_t qDebugFlag;
extern int32_t wDebugFlag;
extern int32_t sDebugFlag;
......
......@@ -22,59 +22,57 @@ extern "C" {
/*
This set of API for queue is designed specially for vnode/mnode. The main purpose is to
consume all the items instead of one item from a queue by one single read. Also, it can
combine multiple queues into a queue set, a consumer thread can consume a queue set via
This set of API for queue is designed specially for vnode/mnode. The main purpose is to
consume all the items instead of one item from a queue by one single read. Also, it can
combine multiple queues into a queue set, a consumer thread can consume a queue set via
a single API instead of looping every queue by itself.
Notes:
1: taosOpenQueue/taosCloseQueue, taosOpenQset/taosCloseQset is NOT multi-thread safe
1: taosOpenQueue/taosCloseQueue, taosOpenQset/taosCloseQset is NOT multi-thread safe
2: after taosCloseQueue/taosCloseQset is called, read/write operation APIs are not safe.
3: read/write operation APIs are multi-thread safe
To remove the limitation and make this set of queue APIs multi-thread safe, REF(tref.c)
shall be used to set up the protection.
shall be used to set up the protection.
*/
typedef void *taos_queue;
typedef void *taos_qset;
typedef void *taos_qall;
typedef struct STaosQueue STaosQueue;
typedef struct STaosQset STaosQset;
typedef struct STaosQall STaosQall;
typedef void (*FProcessItem)(void *ahandle, void *pItem);
typedef void (*FProcessItems)(void *ahandle, taos_qall qall, int numOfItems);
taos_queue taosOpenQueue();
void taosCloseQueue(taos_queue);
void taosSetQueueFp(taos_queue, FProcessItem, FProcessItems);
void *taosAllocateQitem(int size);
void taosFreeQitem(void *pItem);
int taosWriteQitem(taos_queue, void *pItem);
int taosReadQitem(taos_queue, void **pItem);
bool taosQueueEmpty(taos_queue);
taos_qall taosAllocateQall();
void taosFreeQall(taos_qall);
int taosReadAllQitems(taos_queue, taos_qall);
int taosGetQitem(taos_qall, void **pItem);
void taosResetQitems(taos_qall);
taos_qset taosOpenQset();
void taosCloseQset();
void taosQsetThreadResume(taos_qset param);
int taosAddIntoQset(taos_qset, taos_queue, void *ahandle);
void taosRemoveFromQset(taos_qset, taos_queue);
int taosGetQueueNumber(taos_qset);
int taosReadQitemFromQset(taos_qset, void **pItem, void **ahandle, FProcessItem *);
int taosReadAllQitemsFromQset(taos_qset, taos_qall, void **ahandle, FProcessItems *);
int taosGetQueueItemsNumber(taos_queue param);
int taosGetQsetItemsNumber(taos_qset param);
typedef void (*FProcessItems)(void *ahandle, STaosQall *qall, int32_t numOfItems);
STaosQueue *taosOpenQueue();
void taosCloseQueue(STaosQueue *queue);
void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsFp);
void *taosAllocateQitem(int32_t size);
void taosFreeQitem(void *pItem);
int32_t taosWriteQitem(STaosQueue *queue, void *pItem);
int32_t taosReadQitem(STaosQueue *queue, void **ppItem);
bool taosQueueEmpty(STaosQueue *queue);
STaosQall *taosAllocateQall();
void taosFreeQall(STaosQall *qall);
int32_t taosReadAllQitems(STaosQueue *queue, STaosQall *qall);
int32_t taosGetQitem(STaosQall *qall, void **ppItem);
void taosResetQitems(STaosQall *qall);
STaosQset *taosOpenQset();
void taosCloseQset(STaosQset *qset);
void taosQsetThreadResume(STaosQset *qset);
int32_t taosAddIntoQset(STaosQset *qset, STaosQueue *queue, void *ahandle);
void taosRemoveFromQset(STaosQset *qset, STaosQueue *queue);
int32_t taosGetQueueNumber(STaosQset *qset);
int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FProcessItem *itemFp);
int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FProcessItems *itemsFp);
int32_t taosGetQueueItemsNumber(STaosQueue *queue);
int32_t taosGetQsetItemsNumber(STaosQset *qset);
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_QUEUE_H*/
......@@ -35,7 +35,7 @@ typedef struct SWorkerPool {
int32_t max; // max number of workers
int32_t min; // min number of workers
int32_t num; // current number of workers
taos_qset qset;
STaosQset *qset;
const char *name;
SWorker *workers;
pthread_mutex_t mutex;
......@@ -44,8 +44,8 @@ typedef struct SWorkerPool {
typedef struct SMWorker {
int32_t id; // worker id
pthread_t thread; // thread
taos_qall qall;
taos_qset qset; // queue set
STaosQall *qall;
STaosQset *qset; // queue set
SMWorkerPool *pool;
} SMWorker;
......@@ -57,15 +57,15 @@ typedef struct SMWorkerPool {
pthread_mutex_t mutex;
} SMWorkerPool;
int32_t tWorkerInit(SWorkerPool *pool);
void tWorkerCleanup(SWorkerPool *pool);
taos_queue tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp);
void tWorkerFreeQueue(SWorkerPool *pool, taos_queue queue);
int32_t tWorkerInit(SWorkerPool *pool);
void tWorkerCleanup(SWorkerPool *pool);
STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp);
void tWorkerFreeQueue(SWorkerPool *pool, STaosQueue *queue);
int32_t tMWorkerInit(SMWorkerPool *pool);
void tMWorkerCleanup(SMWorkerPool *pool);
taos_queue tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp);
void tMWorkerFreeQueue(SMWorkerPool *pool, taos_queue queue);
int32_t tMWorkerInit(SMWorkerPool *pool);
void tMWorkerCleanup(SMWorkerPool *pool);
STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp);
void tMWorkerFreeQueue(SMWorkerPool *pool, STaosQueue *queue);
#ifdef __cplusplus
}
......
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util common transport parser planner catalog scheduler function qcom
)
ADD_SUBDIRECTORY(test)
\ No newline at end of file
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
\ No newline at end of file
......@@ -101,10 +101,17 @@ typedef struct SReqResultInfo {
uint32_t current;
} SReqResultInfo;
typedef struct SShowReqInfo {
int64_t execId; // showId/queryId
int32_t vgId;
SArray *pArray; // SArray<SVgroupInfo>
int32_t currentIndex; // current accessed vgroup index.
} SShowReqInfo;
typedef struct SRequestSendRecvBody {
tsem_t rspSem; // not used now
void* fp;
int64_t execId; // showId/queryId
SShowReqInfo showInfo; // todo this attribute will be removed after the query framework being completed.
SDataBuf requestMsg;
SReqResultInfo resInfo;
} SRequestSendRecvBody;
......@@ -121,6 +128,7 @@ typedef struct SRequestObj {
char *msgBuf;
void *pInfo; // sql parse info, generated by parser module
int32_t code;
uint64_t affectedRows;
SQueryExecMetric metric;
SRequestSendRecvBody body;
} SRequestObj;
......@@ -131,7 +139,7 @@ extern int32_t clientConnRefPool;
extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code);
SMsgSendInfo* buildSendMsgInfoImpl(SRequestObj*);
SMsgSendInfo* buildMsgInfoImpl(SRequestObj*);
int taos_init();
......
......@@ -18,6 +18,7 @@
#include "clientInt.h"
#include "clientLog.h"
#include "query.h"
#include "scheduler.h"
#include "tmsg.h"
#include "tcache.h"
#include "tconfig.h"
......@@ -230,6 +231,8 @@ void taos_init_imp(void) {
SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
catalogInit(&cfg);
SSchedulerCfg scfg = {.maxJobNum = 100};
schedulerInit(&scfg);
tscDebug("starting to initialize TAOS driver, local ep: %s", tsLocalEp);
taosSetCoreDump(true);
......
......@@ -156,19 +156,12 @@ int32_t parseSql(SRequestObj* pRequest, SQueryNode** pQuery) {
};
cxt.ctx.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
// todo OPT performance
char buf[12] = {0};
sprintf(buf, "%"PRId64, pTscObj->pAppInfo->clusterId);
struct SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(buf, &pCatalog);
int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.ctx.pCatalog);
if (code != TSDB_CODE_SUCCESS) {
tfree(cxt.ctx.db);
return code;
}
cxt.ctx.pCatalog = pCatalog;
code = qParseQuerySql(&cxt, pQuery);
tfree(cxt.ctx.db);
......@@ -181,10 +174,17 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
pRequest->body.requestMsg = (SDataBuf){.pData = pDcl->pMsg, .len = pDcl->msgLen};
STscObj* pTscObj = pRequest->pTscObj;
SMsgSendInfo* pSendMsg = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* pSendMsg = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
if (pDcl->msgType == TDMT_VND_CREATE_TABLE) {
if (pDcl->msgType == TDMT_VND_CREATE_TABLE || pDcl->msgType == TDMT_VND_SHOW_TABLES) {
if (pDcl->msgType == TDMT_VND_SHOW_TABLES) {
SShowReqInfo* pShowReqInfo = &pRequest->body.showInfo;
if (pShowReqInfo->pArray == NULL) {
pShowReqInfo->currentIndex = 0;
pShowReqInfo->pArray = pDcl->pExtension;
}
}
asyncSendMsgToServer(pTscObj->pTransporter, &pDcl->epSet, &transporterId, pSendMsg);
} else {
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
......@@ -196,7 +196,15 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
return TSDB_CODE_SUCCESS;
}
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQuery, SQueryDag** pDag) {
pRequest->type = pQuery->type;
return qCreateQueryDag(pQuery, pDag);
}
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, void** pJob) {
if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
return scheduleExecJob(pRequest->pTscObj->pTransporter, NULL/*todo appInfo.xxx*/, pDag, pJob, &pRequest->affectedRows);
}
return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL/*todo appInfo.xxx*/, pDag, pJob);
}
......@@ -243,7 +251,7 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
SMsgSendInfo* body = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
int64_t transporterId = 0;
......@@ -283,7 +291,7 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
if (qIsDdlQuery(pQuery)) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQuery), _return);
} else {
CHECK_CODE_GOTO(qCreateQueryDag(pQuery, &pDag), _return);
CHECK_CODE_GOTO(getPlan(pRequest, pQuery, &pDag), _return);
CHECK_CODE_GOTO(scheduleQuery(pRequest, pDag, &pJob), _return);
}
......@@ -490,9 +498,38 @@ void* doFetchRow(SRequestObj* pRequest) {
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
pRequest->type = TDMT_MND_SHOW_RETRIEVE;
if (pRequest->type == TDMT_MND_SHOW) {
pRequest->type = TDMT_MND_SHOW_RETRIEVE;
} else if (pRequest->type == TDMT_VND_SHOW_TABLES) {
pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
} else if (pRequest->type == TDMT_VND_SHOW_TABLES_FETCH) {
pRequest->type = TDMT_VND_SHOW_TABLES;
SShowReqInfo* pShowReqInfo = &pRequest->body.showInfo;
pShowReqInfo->currentIndex += 1;
if (pShowReqInfo->currentIndex >= taosArrayGetSize(pShowReqInfo->pArray)) {
return NULL;
}
SVgroupInfo* pVgroupInfo = taosArrayGet(pShowReqInfo->pArray, pShowReqInfo->currentIndex);
SVShowTablesReq* pShowReq = calloc(1, sizeof(SVShowTablesReq));
pShowReq->head.vgId = htonl(pVgroupInfo->vgId);
pRequest->body.requestMsg.len = sizeof(SVShowTablesReq);
pRequest->body.requestMsg.pData = pShowReq;
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
destroySendMsgInfo(body);
pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
}
SMsgSendInfo* body = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
......
......@@ -18,23 +18,26 @@
#include "tname.h"
#include "clientInt.h"
#include "clientLog.h"
#include "trpc.h"
int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
static void setErrno(SRequestObj* pRequest, int32_t code) {
pRequest->code = code;
terrno = code;
}
int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
pRequest->code = code;
setErrno(pRequest, code);
sem_post(&pRequest->body.rspSem);
return 0;
return code;
}
int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
terrno = code;
setErrno(pRequest, code);
sem_post(&pRequest->body.rspSem);
return code;
}
......@@ -74,52 +77,55 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
return 0;
}
static int32_t buildRetrieveMnodeMsg(SRequestObj *pRequest, SMsgSendInfo* pMsgSendInfo) {
pMsgSendInfo->msgType = TDMT_MND_SHOW_RETRIEVE;
pMsgSendInfo->msgInfo.len = sizeof(SRetrieveTableMsg);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->fp = handleRequestRspFp[TMSG_INDEX(pMsgSendInfo->msgType)];
SRetrieveTableMsg *pRetrieveMsg = calloc(1, sizeof(SRetrieveTableMsg));
if (pRetrieveMsg == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
pRetrieveMsg->showId = htonl(pRequest->body.execId);
pMsgSendInfo->msgInfo.pData = pRetrieveMsg;
return TSDB_CODE_SUCCESS;
}
SMsgSendInfo* buildSendMsgInfoImpl(SRequestObj *pRequest) {
SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) {
SMsgSendInfo* pMsgSendInfo = calloc(1, sizeof(SMsgSendInfo));
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE) {
buildRetrieveMnodeMsg(pRequest, pMsgSendInfo);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->requestId = pRequest->requestId;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->msgType = pRequest->type;
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE || pRequest->type == TDMT_VND_SHOW_TABLES_FETCH) {
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE) {
SRetrieveTableMsg* pRetrieveMsg = calloc(1, sizeof(SRetrieveTableMsg));
if (pRetrieveMsg == NULL) {
return NULL;
}
pRetrieveMsg->showId = htobe64(pRequest->body.showInfo.execId);
pMsgSendInfo->msgInfo.pData = pRetrieveMsg;
pMsgSendInfo->msgInfo.len = sizeof(SRetrieveTableMsg);
} else {
SVShowTablesFetchReq* pFetchMsg = calloc(1, sizeof(SVShowTablesFetchReq));
if (pFetchMsg == NULL) {
return NULL;
}
pFetchMsg->id = htobe64(pRequest->body.showInfo.execId);
pFetchMsg->head.vgId = htonl(pRequest->body.showInfo.vgId);
pMsgSendInfo->msgInfo.pData = pFetchMsg;
pMsgSendInfo->msgInfo.len = sizeof(SVShowTablesFetchReq);
}
} else {
assert(pRequest != NULL);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->msgInfo = pRequest->body.requestMsg;
pMsgSendInfo->msgType = pRequest->type;
pMsgSendInfo->requestId = pRequest->requestId;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
}
pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
return pMsgSendInfo;
}
int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
SShowRsp* pShow = (SShowRsp *)pMsg->pData;
pShow->showId = htonl(pShow->showId);
pShow->showId = htobe64(pShow->showId);
STableMetaMsg *pMetaMsg = &(pShow->tableMeta);
pMetaMsg->numOfColumns = htonl(pMetaMsg->numOfColumns);
......@@ -128,6 +134,7 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pMetaMsg->tuid = htobe64(pMetaMsg->tuid);
for (int i = 0; i < pMetaMsg->numOfColumns; ++i) {
pSchema->bytes = htonl(pSchema->bytes);
pSchema->colId = htonl(pSchema->colId);
pSchema++;
}
......@@ -148,34 +155,81 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pResInfo->pCol = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->length = calloc(pResInfo->numOfCols, sizeof(int32_t));
pRequest->body.execId = pShow->showId;
pRequest->body.showInfo.execId = pShow->showId;
// todo
if (pRequest->type == TDMT_VND_SHOW_TABLES) {
SShowReqInfo* pShowInfo = &pRequest->body.showInfo;
int32_t index = pShowInfo->currentIndex;
SVgroupInfo* pInfo = taosArrayGet(pShowInfo->pArray, index);
pShowInfo->vgId = pInfo->vgId;
}
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processRetrieveMnodeRsp(void* param, const SDataBuf* pMsg, int32_t code) {
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
SRequestObj *pRequest = param;
SReqResultInfo *pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
SRequestObj* pRequest = param;
// tfree(pRequest->body.resInfo.pRspMsg);
// pRequest->body.resInfo.pRspMsg = pMsg->pData;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
SRetrieveTableRsp *pRetrieve = (SRetrieveTableRsp *) pMsg->pData;
pRetrieve->numOfRows = htonl(pRetrieve->numOfRows);
pRetrieve->precision = htons(pRetrieve->precision);
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
pResInfo->pRspMsg = pMsg->pData;
pResInfo->numOfRows = pRetrieve->numOfRows;
pResInfo->pData = pRetrieve->data; // todo fix this in async model
pResInfo->pData = pRetrieve->data;
pResInfo->current = 0;
setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows);
tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pRetrieve->numOfRows,
pRetrieve->completed, pRequest->body.execId);
pRetrieve->completed, pRequest->body.showInfo.execId);
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processRetrieveVndRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
pResInfo->pRspMsg = pMsg->pData;
SVShowTablesFetchRsp *pFetchRsp = (SVShowTablesFetchRsp *) pMsg->pData;
pFetchRsp->numOfRows = htonl(pFetchRsp->numOfRows);
pFetchRsp->precision = htons(pFetchRsp->precision);
pResInfo->pRspMsg = pMsg->pData;
pResInfo->numOfRows = pFetchRsp->numOfRows;
pResInfo->pData = pFetchRsp->data;
pResInfo->current = 0;
setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows);
tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pFetchRsp->numOfRows,
pFetchRsp->completed, pRequest->body.showInfo.execId);
tsem_post(&pRequest->body.rspSem);
return 0;
......@@ -191,34 +245,48 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
SUseDbRsp* pUseDbRsp = (SUseDbRsp*)pMsg->pData;
SName name = {0};
tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT | T_NAME_DB);
SUseDbRsp* pUseDbRsp = (SUseDbRsp*) pMsg->pData;
SName name = {0};
tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT|T_NAME_DB);
char db[TSDB_DB_NAME_LEN] = {0};
tNameGetDbName(&name, db);
setConnectionDB(pRequest->pTscObj, db);
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processCreateTableRsp(void* param, const SDataBuf* pMsg, int32_t code) {
assert(pMsg != NULL);
assert(pMsg != NULL && param != NULL);
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
tsem_post(&pRequest->body.rspSem);
return code;
}
int32_t processDropDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
// todo: Remove cache in catalog cache.
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
tsem_post(&pRequest->body.rspSem);
return code;
}
void initMsgHandleFp() {
......@@ -304,4 +372,7 @@ void initMsgHandleFp() {
handleRequestRspFp[TMSG_INDEX(TDMT_MND_USE_DB)] = processUseDbRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = processCreateTableRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = processDropDbRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = processShowRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = processRetrieveVndRsp;
}
\ No newline at end of file
此差异已折叠。
......@@ -12,4 +12,6 @@ target_link_libraries(
INTERFACE api
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
......@@ -15,17 +15,18 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taosdef.h"
#include "taoserror.h"
#include "ulog.h"
#include "tlog.h"
#include "tcompare.h"
#include "tconfig.h"
#include "tep.h"
#include "tglobal.h"
#include "tcompare.h"
#include "tutil.h"
#include "ttimezone.h"
#include "tlocale.h"
#include "tep.h"
#include "tlog.h"
#include "ttimezone.h"
#include "tutil.h"
#include "ulog.h"
// cluster
char tsFirst[TSDB_EP_LEN] = {0};
......@@ -36,22 +37,24 @@ uint16_t tsServerPort = 6030;
int32_t tsStatusInterval = 1; // second
int8_t tsEnableTelemetryReporting = 0;
char tsEmail[TSDB_FQDN_LEN] = {0};
int32_t tsNumOfSupportVnodes = 16;
// common
int32_t tsRpcTimer = 300;
int32_t tsRpcMaxTime = 600; // seconds;
int32_t tsRpcForceTcp = 1; //disable this, means query, show command use udp protocol as default
int32_t tsMaxShellConns = 50000;
int32_t tsRpcTimer = 300;
int32_t tsRpcMaxTime = 600; // seconds;
int32_t tsRpcForceTcp = 1; // disable this, means query, show command use udp protocol as default
int32_t tsMaxShellConns = 50000;
int32_t tsMaxConnections = 5000;
int32_t tsShellActivityTimer = 3; // second
int32_t tsShellActivityTimer = 3; // second
float tsNumOfThreadsPerCore = 1.0f;
int32_t tsNumOfCommitThreads = 4;
float tsRatioOfQueryCores = 1.0f;
int8_t tsDaylight = 0;
int8_t tsDaylight = 0;
int8_t tsEnableCoreFile = 0;
int32_t tsMaxBinaryDisplayWidth = 30;
int64_t tsMaxVnodeQueuedBytes = 1024*1024*1024; //1GB
int8_t tsEnableSlaveQuery = 1;
int8_t tsEnableAdjustMaster = 1;
int8_t tsPrintAuth = 0;
/*
* denote if the server needs to compress response message at the application layer to client, including query rsp,
* metricmeta rsp, and multi-meter query rsp message body. The client compress the submit message to server.
......@@ -79,7 +82,7 @@ int32_t tsMaxSQLStringLen = TSDB_MAX_ALLOWED_SQL_LEN;
int32_t tsMaxWildCardsLen = TSDB_PATTERN_STRING_DEFAULT_LEN;
int32_t tsMaxRegexStringLen = TSDB_REGEX_STRING_DEFAULT_LEN;
int8_t tsTscEnableRecordSql = 0;
int8_t tsTscEnableRecordSql = 0;
// the maximum number of results for projection query on super table that are returned from
// one virtual node, to order according to timestamp
......@@ -89,7 +92,7 @@ int32_t tsMaxNumOfOrderedResults = 100000;
int32_t tsMinSlidingTime = 10;
// the maxinum number of distict query result
int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
// 1 us for interval time range, changed accordingly
int32_t tsMinIntervalTime = 1;
......@@ -101,7 +104,7 @@ int32_t tsMaxStreamComputDelay = 20000;
int32_t tsStreamCompStartDelay = 10000;
// the stream computing delay time after executing failed, change accordingly
int32_t tsRetryStreamCompDelay = 10*1000;
int32_t tsRetryStreamCompDelay = 10 * 1000;
// The delayed computing ration. 10% of the whole computing time window by default.
float tsStreamComputDelayRatio = 0.1f;
......@@ -120,30 +123,16 @@ int64_t tsQueryBufferSizeBytes = -1;
int32_t tsRetrieveBlockingModel = 0;
// last_row(*), first(*), last_row(ts, col1, col2) query, the result fields will be the original column name
int8_t tsKeepOriginalColumnName = 0;
int8_t tsKeepOriginalColumnName = 0;
// long query death-lock
int8_t tsDeadLockKillQuery = 0;
// tsdb config
// tsdb config
// For backward compatibility
bool tsdbForceKeepFile = false;
// balance
int8_t tsEnableFlowCtrl = 1;
int8_t tsEnableSlaveQuery = 1;
int8_t tsEnableAdjustMaster = 1;
// monitor
char tsMonitorDbName[TSDB_DB_NAME_LEN] = "log";
char tsInternalPass[] = "secretkey";
// internal
int8_t tsCompactMnodeWal = 0;
int8_t tsPrintAuth = 0;
char tsVnodeDir[PATH_MAX] = {0};
char tsDnodeDir[PATH_MAX] = {0};
char tsMnodeDir[PATH_MAX] = {0};
int32_t tsDiskCfgNum = 0;
int32_t tsDiskCfgNum = 0;
#ifndef _STORAGE
SDiskCfg tsDiskCfg[1];
......@@ -160,31 +149,28 @@ SDiskCfg tsDiskCfg[TSDB_MAX_DISKS];
int64_t tsTickPerDay[] = {86400000L, 86400000000L, 86400000000000L};
// system info
float tsTotalTmpDirGB = 0;
float tsTotalDataDirGB = 0;
float tsAvailTmpDirectorySpace = 0;
float tsAvailDataDirGB = 0;
float tsUsedDataDirGB = 0;
float tsReservedTmpDirectorySpace = 1.0f;
float tsMinimalDataDirGB = 2.0f;
int32_t tsTotalMemoryMB = 0;
float tsTotalTmpDirGB = 0;
float tsTotalDataDirGB = 0;
float tsAvailTmpDirectorySpace = 0;
float tsAvailDataDirGB = 0;
float tsUsedDataDirGB = 0;
float tsReservedTmpDirectorySpace = 1.0f;
float tsMinimalDataDirGB = 2.0f;
int32_t tsTotalMemoryMB = 0;
uint32_t tsVersion = 0;
#ifdef TD_TSZ
//
// lossy compress 6
//
char lossyColumns[32] = ""; // "float|double" means all float and double columns can be lossy compressed. set empty can close lossy compress.
// below option can take effect when tsLossyColumns not empty
double fPrecision = 1E-8; // float column precision
double dPrecision = 1E-16; // double column precision
uint32_t maxRange = 500; // max range
uint32_t curRange = 100; // range
char Compressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
#endif
char tsLossyColumns[32] = ""; // "float|double" means all float and double columns can be lossy compressed. set empty
// can close lossy compress.
// below option can take effect when tsLossyColumns not empty
double tsFPrecision = 1E-8; // float column precision
double tsDPrecision = 1E-16; // double column precision
uint32_t tsMaxRange = 500; // max range
uint32_t tsCurRange = 100; // range
char tsCompressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
// long query death-lock
int8_t tsDeadLockKillQuery = 0;
int32_t (*monStartSystemFp)() = NULL;
void (*monStopSystemFp)() = NULL;
......@@ -195,13 +181,12 @@ char *qtypeStr[] = {"rpc", "fwd", "wal", "cq", "query"};
static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT;
void taosSetAllDebugFlag() {
if (debugFlag != 0) {
if (debugFlag != 0) {
mDebugFlag = debugFlag;
dDebugFlag = debugFlag;
vDebugFlag = debugFlag;
jniDebugFlag = debugFlag;
odbcDebugFlag = debugFlag;
qDebugFlag = debugFlag;
qDebugFlag = debugFlag;
rpcDebugFlag = debugFlag;
uDebugFlag = debugFlag;
sDebugFlag = debugFlag;
......@@ -213,12 +198,12 @@ void taosSetAllDebugFlag() {
}
int32_t taosCfgDynamicOptions(char *msg) {
char *option, *value;
int32_t olen, vlen;
int32_t vint = 0;
char *option, *value;
int32_t olen, vlen;
int32_t vint = 0;
paGetToken(msg, &option, &olen);
if (olen == 0) return -1;;
if (olen == 0) return -1;
paGetToken(option + olen + 1, &value, &vlen);
if (vlen == 0)
......@@ -231,9 +216,9 @@ int32_t taosCfgDynamicOptions(char *msg) {
for (int32_t i = 0; i < tsGlobalConfigNum; ++i) {
SGlobalCfg *cfg = tsGlobalConfig + i;
//if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
// if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
if (cfg->valType != TAOS_CFG_VTYPE_INT32 && cfg->valType != TAOS_CFG_VTYPE_INT8) continue;
int32_t cfgLen = (int32_t)strlen(cfg->option);
if (cfgLen != olen) continue;
if (strncasecmp(option, cfg->option, olen) != 0) continue;
......@@ -262,7 +247,7 @@ int32_t taosCfgDynamicOptions(char *msg) {
return 0;
}
if (strncasecmp(cfg->option, "debugFlag", olen) == 0) {
taosSetAllDebugFlag();
taosSetAllDebugFlag();
}
return 0;
}
......@@ -323,7 +308,7 @@ static void doInitGlobalConfig(void) {
srand(taosSafeRand());
SGlobalCfg cfg = {0};
// ip address
cfg.option = "firstEp";
cfg.ptr = tsFirst;
......@@ -366,6 +351,16 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
cfg.option = "supportVnodes";
cfg.ptr = &tsNumOfSupportVnodes;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
cfg.maxValue = 65536;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
// directory
cfg.option = "configDir";
cfg.ptr = configDir;
......@@ -442,8 +437,8 @@ static void doInitGlobalConfig(void) {
cfg.ptr = &tsMaxNumOfDistinctResults;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 10*10000;
cfg.maxValue = 10000*10000;
cfg.minValue = 10 * 10000;
cfg.maxValue = 10000 * 10000;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
......@@ -749,17 +744,6 @@ static void doInitGlobalConfig(void) {
cfg.maxValue = 10000000;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_GB;
taosAddConfigOption(cfg);
// module configs
cfg.option = "flowctrl";
cfg.ptr = &tsEnableFlowCtrl;
cfg.valType = TAOS_CFG_VTYPE_INT8;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
cfg.minValue = 0;
cfg.maxValue = 1;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
cfg.option = "slaveQuery";
......@@ -893,16 +877,6 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
cfg.option = "odbcDebugFlag";
cfg.ptr = &odbcDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
cfg.maxValue = 255;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
cfg.option = "uDebugFlag";
cfg.ptr = &uDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
......@@ -1034,7 +1008,7 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
// enable kill long query
// enable kill long query
cfg.option = "deadLockKillQuery";
cfg.ptr = &tsDeadLockKillQuery;
cfg.valType = TAOS_CFG_VTYPE_INT8;
......@@ -1066,7 +1040,6 @@ static void doInitGlobalConfig(void) {
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosAddConfigOption(cfg);
cfg.option = "dPrecision";
......@@ -1100,23 +1073,20 @@ static void doInitGlobalConfig(void) {
taosAddConfigOption(cfg);
assert(tsGlobalConfigNum == TSDB_CFG_MAX_NUM);
#else
//assert(tsGlobalConfigNum == TSDB_CFG_MAX_NUM - 5);
// assert(tsGlobalConfigNum == TSDB_CFG_MAX_NUM - 5);
#endif
}
void taosInitGlobalCfg() {
pthread_once(&tsInitGlobalCfgOnce, doInitGlobalConfig);
}
void taosInitGlobalCfg() { pthread_once(&tsInitGlobalCfgOnce, doInitGlobalConfig); }
int32_t taosCheckAndPrintCfg() {
char fqdn[TSDB_FQDN_LEN];
char fqdn[TSDB_FQDN_LEN];
uint16_t port;
if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG || debugFlag & DEBUG_DUMP) {
taosSetAllDebugFlag();
}
if (tsLocalFqdn[0] == 0) {
taosGetFqdn(tsLocalFqdn);
}
......@@ -1143,7 +1113,7 @@ int32_t taosCheckAndPrintCfg() {
if (taosDirExist(tsTempDir) != 0) {
return -1;
}
taosGetSystemInfo();
tsSetLocale();
......
......@@ -25,4 +25,126 @@
#undef TD_MSG_INFO_
#define TD_MSG_DICT_
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"
\ No newline at end of file
#include "tmsgdef.h"
int tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
int tlen = 0;
tlen += taosEncodeFixedU64(buf, pReq->ver);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep);
tlen += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case TD_SUPER_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
}
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return tlen;
}
void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case TD_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
pReq->stbCfg.pSchema = (SSchema *)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
}
buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
pReq->stbCfg.pTagSchema = (SSchema *)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
pReq->ntbCfg.pSchema = (SSchema *)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return buf;
}
int tSVCreateTbBatchReqSerialize(void **buf, SVCreateTbBatchReq *pReq) {
int tlen = 0;
tlen += taosEncodeFixedU64(buf, pReq->ver);
tlen += taosEncodeFixedU32(buf, taosArrayGetSize(pReq->pArray));
for (size_t i = 0; i < taosArrayGetSize(pReq->pArray); i++) {
SVCreateTbReq *pCreateTbReq = taosArrayGet(pReq->pArray, i);
tlen += tSerializeSVCreateTbReq(buf, pCreateTbReq);
}
return tlen;
}
void *tSVCreateTbBatchReqDeserialize(void *buf, SVCreateTbBatchReq *pReq) {
uint32_t nsize = 0;
buf = taosDecodeFixedU64(buf, &pReq->ver);
buf = taosDecodeFixedU32(buf, &nsize);
for (size_t i = 0; i < nsize; i++) {
SVCreateTbReq req;
buf = tDeserializeSVCreateTbReq(buf, &req);
taosArrayPush(pReq->pArray, &req);
}
return buf;
}
\ No newline at end of file
......@@ -259,3 +259,13 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
return 0;
}
SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name) {
SSchema s = {0};
s.type = type;
s.bytes = bytes;
s.colId = colId;
tstrncpy(s.name, name, tListLen(s.name));
return s;
}
\ No newline at end of file
......@@ -18,11 +18,11 @@ TARGET_INCLUDE_DIRECTORIES(
)
# tmsg test
add_executable(tmsgTest "")
target_sources(tmsgTest
PRIVATE
"tmsgTest.cpp"
"../src/tmsg.c"
)
target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file
# add_executable(tmsgTest "")
# target_sources(tmsgTest
# PRIVATE
# "tmsgTest.cpp"
# "../src/tmsg.c"
# )
# target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
# target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file
......@@ -139,8 +139,8 @@ void dmnWaitSignal() {
void dmnInitOption(SDnodeOpt *pOption) {
pOption->sver = 30000000; //3.0.0.0
pOption->numOfCores = tsNumOfCores;
pOption->numOfSupportVnodes = 16;
pOption->numOfCommitThreads = 1;
pOption->numOfSupportVnodes = tsNumOfSupportVnodes;
pOption->numOfCommitThreads = tsNumOfCommitThreads;
pOption->statusInterval = tsStatusInterval;
pOption->numOfThreadsPerCore = tsNumOfThreadsPerCore;
pOption->ratioOfQueryCores = tsRatioOfQueryCores;
......
......@@ -64,7 +64,7 @@ typedef struct {
int32_t maxNum;
void *queueFp;
SDnode *pDnode;
taos_queue queue;
STaosQueue *queue;
union {
SWorkerPool pool;
SMWorkerPool mpool;
......@@ -92,7 +92,7 @@ typedef struct {
SDnodeEps *dnodeEps;
pthread_t *threadId;
SRWLatch latch;
taos_queue pMgmtQ;
STaosQueue *pMgmtQ;
SWorkerPool mgmtPool;
} SDnodeMgmt;
......
......@@ -19,7 +19,7 @@
#include "dndTransport.h"
#include "dndWorker.h"
static void dndProcessBnodeQueue(SDnode *pDnode, taos_qall qall, int32_t numOfMsgs);
static void dndProcessBnodeQueue(SDnode *pDnode, STaosQall *qall, int32_t numOfMsgs);
static SBnode *dndAcquireBnode(SDnode *pDnode) {
SBnodeMgmt *pMgmt = &pDnode->bmgmt;
......@@ -256,7 +256,7 @@ static int32_t dndDropBnode(SDnode *pDnode) {
}
int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SCreateBnodeInMsg *pMsg = pRpcMsg->pCont;
SDCreateBnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......@@ -268,7 +268,7 @@ int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
int32_t dndProcessDropBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDropBnodeInMsg *pMsg = pRpcMsg->pCont;
SDDropBnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......@@ -286,7 +286,7 @@ static void dndSendBnodeErrorRsp(SRpcMsg *pMsg, int32_t code) {
taosFreeQitem(pMsg);
}
static void dndSendBnodeErrorRsps(taos_qall qall, int32_t numOfMsgs, int32_t code) {
static void dndSendBnodeErrorRsps(STaosQall *qall, int32_t numOfMsgs, int32_t code) {
for (int32_t i = 0; i < numOfMsgs; ++i) {
SRpcMsg *pMsg = NULL;
taosGetQitem(qall, (void **)&pMsg);
......@@ -294,7 +294,7 @@ static void dndSendBnodeErrorRsps(taos_qall qall, int32_t numOfMsgs, int32_t cod
}
}
static void dndProcessBnodeQueue(SDnode *pDnode, taos_qall qall, int32_t numOfMsgs) {
static void dndProcessBnodeQueue(SDnode *pDnode, STaosQall *qall, int32_t numOfMsgs) {
SBnode *pBnode = dndAcquireBnode(pDnode);
if (pBnode == NULL) {
dndSendBnodeErrorRsps(qall, numOfMsgs, TSDB_CODE_OUT_OF_MEMORY);
......
......@@ -15,9 +15,12 @@
#define _DEFAULT_SOURCE
#include "dndDnode.h"
#include "dndBnode.h"
#include "dndMnode.h"
#include "dndQnode.h"
#include "dndSnode.h"
#include "dndTransport.h"
#include "dndVnodes.h"
#include "dndMnode.h"
static int32_t dndInitMgmtWorker(SDnode *pDnode);
static void dndCleanupMgmtWorker(SDnode *pDnode);
......@@ -367,8 +370,8 @@ void dndSendStatusMsg(SDnode *pDnode) {
pStatus->clusterId = htobe64(pMgmt->clusterId);
pStatus->rebootTime = htobe64(pMgmt->rebootTime);
pStatus->updateTime = htobe64(pMgmt->updateTime);
pStatus->numOfCores = htons(pDnode->opt.numOfCores);
pStatus->numOfSupportVnodes = htons(pDnode->opt.numOfSupportVnodes);
pStatus->numOfCores = htonl(pDnode->opt.numOfCores);
pStatus->numOfSupportVnodes = htonl(pDnode->opt.numOfSupportVnodes);
tstrncpy(pStatus->dnodeEp, pDnode->opt.localEp, TSDB_EP_LEN);
pStatus->clusterCfg.statusInterval = htonl(pDnode->opt.statusInterval);
......@@ -394,7 +397,7 @@ void dndSendStatusMsg(SDnode *pDnode) {
static void dndUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
if (pMgmt->dnodeId == 0) {
dInfo("set dnodeId:%d clusterId:% " PRId64, pCfg->dnodeId, pCfg->clusterId);
dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
taosWLockLatch(&pMgmt->latch);
pMgmt->dnodeId = pCfg->dnodeId;
pMgmt->clusterId = pCfg->clusterId;
......@@ -437,19 +440,21 @@ static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pMsg) {
}
SStatusRsp *pRsp = pMsg->pCont;
SDnodeCfg *pCfg = &pRsp->dnodeCfg;
pCfg->dnodeId = htonl(pCfg->dnodeId);
pCfg->clusterId = htobe64(pCfg->clusterId);
dndUpdateDnodeCfg(pDnode, pCfg);
if (pMsg->pCont != NULL && pMsg->contLen != 0) {
SDnodeCfg *pCfg = &pRsp->dnodeCfg;
pCfg->dnodeId = htonl(pCfg->dnodeId);
pCfg->clusterId = htobe64(pCfg->clusterId);
dndUpdateDnodeCfg(pDnode, pCfg);
SDnodeEps *pDnodeEps = &pRsp->dnodeEps;
pDnodeEps->num = htonl(pDnodeEps->num);
for (int32_t i = 0; i < pDnodeEps->num; ++i) {
pDnodeEps->eps[i].id = htonl(pDnodeEps->eps[i].id);
pDnodeEps->eps[i].port = htons(pDnodeEps->eps[i].port);
}
SDnodeEps *pDnodeEps = &pRsp->dnodeEps;
pDnodeEps->num = htonl(pDnodeEps->num);
for (int32_t i = 0; i < pDnodeEps->num; ++i) {
pDnodeEps->eps[i].id = htonl(pDnodeEps->eps[i].id);
pDnodeEps->eps[i].port = htons(pDnodeEps->eps[i].port);
dndUpdateDnodeEps(pDnode, pDnodeEps);
}
dndUpdateDnodeEps(pDnode, pDnodeEps);
pMgmt->statusSent = 0;
}
......@@ -648,6 +653,24 @@ static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg) {
case TDMT_DND_DROP_MNODE:
code = dndProcessDropMnodeReq(pDnode, pMsg);
break;
case TDMT_DND_CREATE_QNODE:
code = dndProcessCreateQnodeReq(pDnode, pMsg);
break;
case TDMT_DND_DROP_QNODE:
code = dndProcessDropQnodeReq(pDnode, pMsg);
break;
case TDMT_DND_CREATE_SNODE:
code = dndProcessCreateSnodeReq(pDnode, pMsg);
break;
case TDMT_DND_DROP_SNODE:
code = dndProcessDropSnodeReq(pDnode, pMsg);
break;
case TDMT_DND_CREATE_BNODE:
code = dndProcessCreateBnodeReq(pDnode, pMsg);
break;
case TDMT_DND_DROP_BNODE:
code = dndProcessDropBnodeReq(pDnode, pMsg);
break;
case TDMT_DND_CONFIG_DNODE:
code = dndProcessConfigDnodeReq(pDnode, pMsg);
break;
......
......@@ -299,7 +299,7 @@ static void dndBuildMnodeOpenOption(SDnode *pDnode, SMnodeOpt *pOption) {
memcpy(&pOption->replicas, pMgmt->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
}
static int32_t dndBuildMnodeOptionFromMsg(SDnode *pDnode, SMnodeOpt *pOption, SCreateMnodeInMsg *pMsg) {
static int32_t dndBuildMnodeOptionFromMsg(SDnode *pDnode, SMnodeOpt *pOption, SDCreateMnodeMsg *pMsg) {
dndInitMnodeOption(pDnode, pOption);
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
......@@ -417,8 +417,8 @@ static int32_t dndDropMnode(SDnode *pDnode) {
return 0;
}
static SCreateMnodeInMsg *dndParseCreateMnodeMsg(SRpcMsg *pRpcMsg) {
SCreateMnodeInMsg *pMsg = pRpcMsg->pCont;
static SDCreateMnodeMsg *dndParseCreateMnodeMsg(SRpcMsg *pRpcMsg) {
SDCreateMnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
for (int32_t i = 0; i < pMsg->replica; ++i) {
pMsg->replicas[i].id = htonl(pMsg->replicas[i].id);
......@@ -429,7 +429,7 @@ static SCreateMnodeInMsg *dndParseCreateMnodeMsg(SRpcMsg *pRpcMsg) {
}
int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SCreateMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
SDCreateMnodeMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
......@@ -445,7 +445,7 @@ int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SAlterMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
SDAlterMnodeMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
......@@ -465,7 +465,7 @@ int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDropMnodeInMsg *pMsg = pRpcMsg->pCont;
SDDropMnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......
......@@ -261,7 +261,7 @@ static int32_t dndDropQnode(SDnode *pDnode) {
}
int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SCreateQnodeInMsg *pMsg = pRpcMsg->pCont;
SDCreateQnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......@@ -273,7 +273,7 @@ int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
int32_t dndProcessDropQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDropQnodeInMsg *pMsg = pRpcMsg->pCont;
SDDropQnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......
......@@ -256,7 +256,7 @@ static int32_t dndDropSnode(SDnode *pDnode) {
}
int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SCreateSnodeInMsg *pMsg = pRpcMsg->pCont;
SDCreateSnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......@@ -268,7 +268,7 @@ int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
SDropSnodeInMsg *pMsg = pRpcMsg->pCont;
SDDropSnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
......
......@@ -37,6 +37,18 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_MNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_MNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_MNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_QNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_QNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_QNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_QNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_SNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_SNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_SNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_SNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_BNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_BNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_BNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_BNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_VNODE)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_VNODE_RSP)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_VNODE)] = dndProcessMgmtMsg;
......@@ -66,6 +78,12 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_DNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_MNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_MNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_QNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_QNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_SNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_SNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_BNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_BNODE)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_DB)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_USE_DB)] = dndProcessMnodeWriteMsg;
......@@ -85,7 +103,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_HEARTBEAT)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW_RETRIEVE)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS_RSP)] = dndProcessMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_TRANS)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_TRANS_RSP)] = dndProcessMnodeWriteMsg;
......@@ -103,8 +121,8 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLES_META)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLES_META)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_CONSUME)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_QUERY)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_CONNECT)] = dndProcessVnodeWriteMsg;
......@@ -123,6 +141,8 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CREATE_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = dndProcessVnodeFetchMsg;
}
static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
......@@ -365,6 +385,7 @@ void dndCleanupTrans(SDnode *pDnode) {
void dndSendMsgToDnode(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pMsg) {
STransMgmt *pMgmt = &pDnode->tmgmt;
if (pMgmt->clientRpc == NULL) return;
rpcSendRequest(pMgmt->clientRpc, pEpSet, pMsg, NULL);
}
......
......@@ -8,6 +8,9 @@ add_subdirectory(db)
add_subdirectory(dnode)
# add_subdirectory(func)
add_subdirectory(mnode)
add_subdirectory(qnode)
add_subdirectory(snode)
add_subdirectory(bnode)
add_subdirectory(profile)
add_subdirectory(show)
add_subdirectory(stb)
......
aux_source_directory(. STEST_SRC)
add_executable(dnode_test_bnode ${STEST_SRC})
target_link_libraries(
dnode_test_bnode
PUBLIC sut
)
add_test(
NAME dnode_test_bnode
COMMAND dnode_test_bnode
)
此差异已折叠。
aux_source_directory(. QTEST_SRC)
add_executable(dnode_test_qnode ${QTEST_SRC})
target_link_libraries(
dnode_test_qnode
PUBLIC sut
)
add_test(
NAME dnode_test_qnode
COMMAND dnode_test_qnode
)
此差异已折叠。
aux_source_directory(. STEST_SRC)
add_executable(dnode_test_snode ${STEST_SRC})
target_link_libraries(
dnode_test_snode
PUBLIC sut
)
add_test(
NAME dnode_test_snode
COMMAND dnode_test_snode
)
此差异已折叠。
......@@ -146,8 +146,8 @@ TEST_F(DndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
pSchema->bytes = htonl(pSchema->bytes);
}
EXPECT_STREQ(pRsp->tbFname, "");
EXPECT_STREQ(pRsp->stbFname, "1.d1.stb");
EXPECT_STREQ(pRsp->tbFname, "1.d1.stb");
EXPECT_STREQ(pRsp->stbFname, "");
EXPECT_EQ(pRsp->numOfColumns, 2);
EXPECT_EQ(pRsp->numOfTags, 3);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册