diff --git a/contrib/test/craft/clear.sh b/contrib/test/craft/clear.sh index 6412656d77..398b3088f2 100644 --- a/contrib/test/craft/clear.sh +++ b/contrib/test/craft/clear.sh @@ -1,3 +1,4 @@ #!/bin/bash rm -rf 127.0.0.1* +rm -rf ./data diff --git a/contrib/test/craft/common.h b/contrib/test/craft/common.h new file mode 100644 index 0000000000..1e94ee8bca --- /dev/null +++ b/contrib/test/craft/common.h @@ -0,0 +1,34 @@ +#ifndef TDENGINE_COMMON_H +#define TDENGINE_COMMON_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#define MAX_PEERS 10 +#define COMMAND_LEN 512 +#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 diff --git a/contrib/test/craft/raftMain.c b/contrib/test/craft/raftMain.c new file mode 100644 index 0000000000..52e0b694dc --- /dev/null +++ b/contrib/test/craft/raftMain.c @@ -0,0 +1,367 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "raftServer.h" +#include "common.h" + +const char *exe_name; + +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 parseCommand(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); +} + +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 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); + } +} + +void printRaftState(struct raft *r) { + printf("----Raft State: -----------\n"); + 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)); + + parseCommand(cmd_buf, cmd, param1, param2, 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, "put") == 0) { + char buf[256]; + snprintf(buf, sizeof(buf), "%s--%s", param1, param2); + putValue(&pRaftServer->raft, buf); + + } else if (strcmp(cmd, "get") == 0) { + getValue(param1); + + } else if (strcmp(cmd, "state") == 0) { + printRaftState(&pRaftServer->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); + + 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; +} diff --git a/contrib/test/craft/raftServer.c b/contrib/test/craft/raftServer.c new file mode 100644 index 0000000000..6f4dbc1997 --- /dev/null +++ b/contrib/test/craft/raftServer.c @@ -0,0 +1,156 @@ +#include +#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; +} + +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)); + + pRaftServer->raftId = raftId(pRaftServer->host, pRaftServer->port); + pRaftServer->fsm = pFsm; + + ret = uv_loop_init(&pRaftServer->loop); + if (!ret) { + fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft)); + } + + ret = raft_uv_tcp_init(&pRaftServer->transport, &pRaftServer->loop); + if (!ret) { + fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft)); + } + + ret = raft_uv_init(&pRaftServer->io, &pRaftServer->loop, pRaftServer->dir, &pRaftServer->transport); + if (!ret) { + fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft)); + } + + ret = raft_init(&pRaftServer->raft, &pRaftServer->io, pRaftServer->fsm, pRaftServer->raftId, pRaftServer->address); + if (!ret) { + fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft)); + } + + struct raft_configuration conf; + raft_configuration_init(&conf); + 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); + 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->raft, &conf); + + return 0; +} + +int32_t raftServerStart(SRaftServer *pRaftServer) { + int ret; + ret = raft_start(&pRaftServer->raft); + if (!ret) { + fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->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; +} diff --git a/contrib/test/craft/raftServer.h b/contrib/test/craft/raftServer.h new file mode 100644 index 0000000000..5fccde6bf2 --- /dev/null +++ b/contrib/test/craft/raftServer.h @@ -0,0 +1,61 @@ +#ifndef TDENGINE_RAFT_SERVER_H +#define TDENGINE_RAFT_SERVER_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include +#include +#include +#include +#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 * 2]; /* Data dir of UV I/O backend */ + char host[HOST_LEN]; + uint32_t port; + char address[ADDRESS_LEN]; /* Raft instance address */ + 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 */ + struct uv_loop_s loop; /* UV loop */ + struct raft_uv_transport transport; /* UV I/O backend transport */ +} SRaftServer; + +#define MAX_TOKEN_LEN 32 +int splitString(const char* str, char* separator, char (*arr)[MAX_TOKEN_LEN], int n_arr); + +uint64_t raftId(const char *host, uint32_t port); +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