diff --git a/source/libs/sync/inc/syncIO.h b/source/libs/sync/inc/syncIO.h index ebe9fb32d1dc342e467ec548fac05d8dbd654444..bf3b3d34c1ae3a6cbe0c0d330e49ca8b7bf8c3e5 100644 --- a/source/libs/sync/inc/syncIO.h +++ b/source/libs/sync/inc/syncIO.h @@ -20,11 +20,38 @@ extern "C" { #endif +#include #include #include -#include +#include "os.h" +#include "syncInt.h" #include "taosdef.h" +#include "tqueue.h" +#include "trpc.h" + +typedef struct SSyncIO { + void * serverRpc; + void * clientRpc; + STaosQueue *pMsgQ; + STaosQset * pQset; + pthread_t tid; + int8_t isStart; + + int32_t (*start)(struct SSyncIO *ths); + int32_t (*stop)(struct SSyncIO *ths); + int32_t (*ping)(struct SSyncIO *ths); + int32_t (*onMessage)(struct SSyncIO *ths, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); + int32_t (*destroy)(struct SSyncIO *ths); + +} SSyncIO; + +SSyncIO * syncIOCreate(); +static int32_t syncIOStart(SSyncIO *io); +static int32_t syncIOStop(SSyncIO *io); +static int32_t syncIOPing(SSyncIO *io); +static int32_t syncIOOnMessage(struct SSyncIO *io, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); +static int32_t syncIODestroy(SSyncIO *io); #ifdef __cplusplus } diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index c1aba0518c36e31e233a7e3ac16ac2a9a14d377c..df8412ab237a03d0e41f4ceaa624ca7f30853111 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -23,16 +23,14 @@ extern "C" { #include #include #include -#include "taosdef.h" #include "sync.h" +#include "taosdef.h" #include "tlog.h" extern int32_t sDebugFlag; -#define sLog(...) \ - { \ - taosPrintLog("SYN FATAL ", sDebugFlag, __VA_ARGS__); \ - } +#define sLog(...) \ + { taosPrintLog("SYN FATAL ", sDebugFlag, __VA_ARGS__); } #define sFatal(...) \ { \ diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index 1286108664270b81a4b5ae70f6e592dc67a6a5ad..65654564ab9d7d14249dd7ee4d3e4e4f43e4829d 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -17,101 +17,96 @@ #include "sync.h" void appendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) { - -// TLA+ Spec -//AppendEntries(i, j) == -// /\ i /= j -// /\ state[i] = Leader -// /\ LET prevLogIndex == nextIndex[i][j] - 1 -// prevLogTerm == IF prevLogIndex > 0 THEN -// log[i][prevLogIndex].term -// ELSE -// 0 -// \* Send up to 1 entry, constrained by the end of the log. -// lastEntry == Min({Len(log[i]), nextIndex[i][j]}) -// entries == SubSeq(log[i], nextIndex[i][j], lastEntry) -// IN Send([mtype |-> AppendEntriesRequest, -// mterm |-> currentTerm[i], -// mprevLogIndex |-> prevLogIndex, -// mprevLogTerm |-> prevLogTerm, -// mentries |-> entries, -// \* mlog is used as a history variable for the proof. -// \* It would not exist in a real implementation. -// mlog |-> log[i], -// mcommitIndex |-> Min({commitIndex[i], lastEntry}), -// msource |-> i, -// mdest |-> j]) -// /\ UNCHANGED <> - + // TLA+ Spec + // AppendEntries(i, j) == + // /\ i /= j + // /\ state[i] = Leader + // /\ LET prevLogIndex == nextIndex[i][j] - 1 + // prevLogTerm == IF prevLogIndex > 0 THEN + // log[i][prevLogIndex].term + // ELSE + // 0 + // \* Send up to 1 entry, constrained by the end of the log. + // lastEntry == Min({Len(log[i]), nextIndex[i][j]}) + // entries == SubSeq(log[i], nextIndex[i][j], lastEntry) + // IN Send([mtype |-> AppendEntriesRequest, + // mterm |-> currentTerm[i], + // mprevLogIndex |-> prevLogIndex, + // mprevLogTerm |-> prevLogTerm, + // mentries |-> entries, + // \* mlog is used as a history variable for the proof. + // \* It would not exist in a real implementation. + // mlog |-> log[i], + // mcommitIndex |-> Min({commitIndex[i], lastEntry}), + // msource |-> i, + // mdest |-> j]) + // /\ UNCHANGED <> } void onAppendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) { - -// TLA+ Spec -//HandleAppendEntriesRequest(i, j, m) == -// LET logOk == \/ m.mprevLogIndex = 0 -// \/ /\ m.mprevLogIndex > 0 -// /\ m.mprevLogIndex <= Len(log[i]) -// /\ m.mprevLogTerm = log[i][m.mprevLogIndex].term -// IN /\ m.mterm <= currentTerm[i] -// /\ \/ /\ \* reject request -// \/ m.mterm < currentTerm[i] -// \/ /\ m.mterm = currentTerm[i] -// /\ state[i] = Follower -// /\ \lnot logOk -// /\ Reply([mtype |-> AppendEntriesResponse, -// mterm |-> currentTerm[i], -// msuccess |-> FALSE, -// mmatchIndex |-> 0, -// msource |-> i, -// mdest |-> j], -// m) -// /\ UNCHANGED <> -// \/ \* return to follower state -// /\ m.mterm = currentTerm[i] -// /\ state[i] = Candidate -// /\ state' = [state EXCEPT ![i] = Follower] -// /\ UNCHANGED <> -// \/ \* accept request -// /\ m.mterm = currentTerm[i] -// /\ state[i] = Follower -// /\ logOk -// /\ LET index == m.mprevLogIndex + 1 -// IN \/ \* already done with request -// /\ \/ m.mentries = << >> -// \/ /\ m.mentries /= << >> -// /\ Len(log[i]) >= index -// /\ log[i][index].term = m.mentries[1].term -// \* This could make our commitIndex decrease (for -// \* example if we process an old, duplicated request), -// \* but that doesn't really affect anything. -// /\ commitIndex' = [commitIndex EXCEPT ![i] = -// m.mcommitIndex] -// /\ Reply([mtype |-> AppendEntriesResponse, -// mterm |-> currentTerm[i], -// msuccess |-> TRUE, -// mmatchIndex |-> m.mprevLogIndex + -// Len(m.mentries), -// msource |-> i, -// mdest |-> j], -// m) -// /\ UNCHANGED <> -// \/ \* conflict: remove 1 entry -// /\ m.mentries /= << >> -// /\ Len(log[i]) >= index -// /\ log[i][index].term /= m.mentries[1].term -// /\ LET new == [index2 \in 1..(Len(log[i]) - 1) |-> -// log[i][index2]] -// IN log' = [log EXCEPT ![i] = new] -// /\ UNCHANGED <> -// \/ \* no conflict: append entry -// /\ m.mentries /= << >> -// /\ Len(log[i]) = m.mprevLogIndex -// /\ log' = [log EXCEPT ![i] = -// Append(log[i], m.mentries[1])] -// /\ UNCHANGED <> -// /\ UNCHANGED <> -// - - + // TLA+ Spec + // HandleAppendEntriesRequest(i, j, m) == + // LET logOk == \/ m.mprevLogIndex = 0 + // \/ /\ m.mprevLogIndex > 0 + // /\ m.mprevLogIndex <= Len(log[i]) + // /\ m.mprevLogTerm = log[i][m.mprevLogIndex].term + // IN /\ m.mterm <= currentTerm[i] + // /\ \/ /\ \* reject request + // \/ m.mterm < currentTerm[i] + // \/ /\ m.mterm = currentTerm[i] + // /\ state[i] = Follower + // /\ \lnot logOk + // /\ Reply([mtype |-> AppendEntriesResponse, + // mterm |-> currentTerm[i], + // msuccess |-> FALSE, + // mmatchIndex |-> 0, + // msource |-> i, + // mdest |-> j], + // m) + // /\ UNCHANGED <> + // \/ \* return to follower state + // /\ m.mterm = currentTerm[i] + // /\ state[i] = Candidate + // /\ state' = [state EXCEPT ![i] = Follower] + // /\ UNCHANGED <> + // \/ \* accept request + // /\ m.mterm = currentTerm[i] + // /\ state[i] = Follower + // /\ logOk + // /\ LET index == m.mprevLogIndex + 1 + // IN \/ \* already done with request + // /\ \/ m.mentries = << >> + // \/ /\ m.mentries /= << >> + // /\ Len(log[i]) >= index + // /\ log[i][index].term = m.mentries[1].term + // \* This could make our commitIndex decrease (for + // \* example if we process an old, duplicated request), + // \* but that doesn't really affect anything. + // /\ commitIndex' = [commitIndex EXCEPT ![i] = + // m.mcommitIndex] + // /\ Reply([mtype |-> AppendEntriesResponse, + // mterm |-> currentTerm[i], + // msuccess |-> TRUE, + // mmatchIndex |-> m.mprevLogIndex + + // Len(m.mentries), + // msource |-> i, + // mdest |-> j], + // m) + // /\ UNCHANGED <> + // \/ \* conflict: remove 1 entry + // /\ m.mentries /= << >> + // /\ Len(log[i]) >= index + // /\ log[i][index].term /= m.mentries[1].term + // /\ LET new == [index2 \in 1..(Len(log[i]) - 1) |-> + // log[i][index2]] + // IN log' = [log EXCEPT ![i] = new] + // /\ UNCHANGED <> + // \/ \* no conflict: append entry + // /\ m.mentries /= << >> + // /\ Len(log[i]) = m.mprevLogIndex + // /\ log' = [log EXCEPT ![i] = + // Append(log[i], m.mentries[1])] + // /\ UNCHANGED <> + // /\ UNCHANGED <> + // } diff --git a/source/libs/sync/src/syncAppendEntriesReply.c b/source/libs/sync/src/syncAppendEntriesReply.c index 4a9055e172fce1aad4753410301c34f4c8823275..20235ef720e06437afde80d21d90d613ae46b3e5 100644 --- a/source/libs/sync/src/syncAppendEntriesReply.c +++ b/source/libs/sync/src/syncAppendEntriesReply.c @@ -17,18 +17,16 @@ #include "sync.h" void onAppendEntriesReply(SRaft *pRaft, const SyncAppendEntriesReply *pMsg) { - -// TLA+ Spec -//HandleAppendEntriesResponse(i, j, m) == -// /\ m.mterm = currentTerm[i] -// /\ \/ /\ m.msuccess \* successful -// /\ nextIndex' = [nextIndex EXCEPT ![i][j] = m.mmatchIndex + 1] -// /\ matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex] -// \/ /\ \lnot m.msuccess \* not successful -// /\ nextIndex' = [nextIndex EXCEPT ![i][j] = -// Max({nextIndex[i][j] - 1, 1})] -// /\ UNCHANGED <> -// /\ Discard(m) -// /\ UNCHANGED <> - + // TLA+ Spec + // HandleAppendEntriesResponse(i, j, m) == + // /\ m.mterm = currentTerm[i] + // /\ \/ /\ m.msuccess \* successful + // /\ nextIndex' = [nextIndex EXCEPT ![i][j] = m.mmatchIndex + 1] + // /\ matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex] + // \/ /\ \lnot m.msuccess \* not successful + // /\ nextIndex' = [nextIndex EXCEPT ![i][j] = + // Max({nextIndex[i][j] - 1, 1})] + // /\ UNCHANGED <> + // /\ Discard(m) + // /\ UNCHANGED <> } diff --git a/source/libs/sync/src/syncIO.c b/source/libs/sync/src/syncIO.c index 738fc4c5e129562d368451c3f9d1eb3315060bb4..b20367cf568ccaffcecb47879d35bc094686a126 100644 --- a/source/libs/sync/src/syncIO.c +++ b/source/libs/sync/src/syncIO.c @@ -13,4 +13,191 @@ * along with this program. If not, see . */ -#include "sync.h" +#include "syncIO.h" +#include "syncOnMessage.h" + +void *syncConsumer(void *param) { + SSyncIO *io = param; + + STaosQall *qall; + SRpcMsg *pRpcMsg, rpcMsg; + int type; + + qall = taosAllocateQall(); + + while (1) { + int numOfMsgs = taosReadAllQitemsFromQset(io->pQset, qall, NULL, NULL); + sDebug("%d sync-io msgs are received", numOfMsgs); + if (numOfMsgs <= 0) break; + + for (int i = 0; i < numOfMsgs; ++i) { + taosGetQitem(qall, (void **)&pRpcMsg); + sDebug("sync-io recv msg: %s", (char *)(pRpcMsg->pCont)); + } + + taosResetQitems(qall); + for (int i = 0; i < numOfMsgs; ++i) { + taosGetQitem(qall, (void **)&pRpcMsg); + rpcFreeCont(pRpcMsg->pCont); + + /* + int msgSize = 128; + memset(&rpcMsg, 0, sizeof(rpcMsg)); + rpcMsg.pCont = rpcMallocCont(msgSize); + rpcMsg.contLen = msgSize; + rpcMsg.handle = pRpcMsg->handle; + rpcMsg.code = 0; + rpcSendResponse(&rpcMsg); + */ + + taosFreeQitem(pRpcMsg); + } + } + + taosFreeQall(qall); +} + +static int retrieveAuthInfo(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey) { + // app shall retrieve the auth info based on meterID from DB or a data file + // demo code here only for simple demo + int ret = 0; + return ret; +} + +static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { + /* +// SInfo *pInfo = (SInfo *)pMsg->ahandle; +sDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, + pMsg->code); + +if (pEpSet) pInfo->epSet = *pEpSet; + +rpcFreeCont(pMsg->pCont); +// tsem_post(&pInfo->rspSem); +tsem_post(&pInfo->rspSem); +*/ +} + +static void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { + SSyncIO *io = pParent; + SRpcMsg *pTemp; + + pTemp = taosAllocateQitem(sizeof(SRpcMsg)); + memcpy(pTemp, pMsg, sizeof(SRpcMsg)); + + sDebug("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp); + taosWriteQitem(io->pMsgQ, pTemp); +} + +SSyncIO *syncIOCreate() { + SSyncIO *io = (SSyncIO *)malloc(sizeof(SSyncIO)); + memset(io, 0, sizeof(*io)); + + io->pMsgQ = taosOpenQueue(); + io->pQset = taosOpenQset(); + taosAddIntoQset(io->pQset, io->pMsgQ, NULL); + + io->start = syncIOStart; + io->stop = syncIOStop; + io->ping = syncIOPing; + io->onMessage = syncIOOnMessage; + io->destroy = syncIODestroy; + + return io; +} + +static int32_t syncIOStart(SSyncIO *io) { + taosBlockSIGPIPE(); + + // cient rpc init + { + SRpcInit rpcInit; + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localPort = 0; + rpcInit.label = "SYNC-IO-CLIENT"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = processResponse; + rpcInit.sessions = 100; + rpcInit.idleTime = 100; + rpcInit.user = "sync-io"; + rpcInit.secret = "sync-io"; + rpcInit.ckey = "key"; + rpcInit.spi = 1; + rpcInit.connType = TAOS_CONN_CLIENT; + + io->clientRpc = rpcOpen(&rpcInit); + if (io->clientRpc == NULL) { + sError("failed to initialize RPC"); + return -1; + } + } + + // server rpc init + { + SRpcInit rpcInit; + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localPort = 38000; + rpcInit.label = "SYNC-IO-SERVER"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = processRequestMsg; + rpcInit.sessions = 1000; + rpcInit.idleTime = 2 * 1500; + rpcInit.afp = retrieveAuthInfo; + rpcInit.parent = io; + rpcInit.connType = TAOS_CONN_SERVER; + + void *pRpc = rpcOpen(&rpcInit); + if (pRpc == NULL) { + sError("failed to start RPC server"); + return -1; + } + } + + // start consumer thread + { + if (pthread_create(&io->tid, NULL, syncConsumer, NULL) != 0) { + sError("failed to create sync consumer thread since %s", strerror(errno)); + terrno = TAOS_SYSTEM_ERROR(errno); + return -1; + } + } + + return 0; +} + +static int32_t syncIOStop(SSyncIO *io) { + atomic_store_8(&io->isStart, 0); + pthread_join(io->tid, NULL); + return 0; +} + +static int32_t syncIOPing(SSyncIO *io) { return 0; } + +static int32_t syncIOOnMessage(struct SSyncIO *io, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { return 0; } + +static int32_t syncIODestroy(SSyncIO *io) { + int8_t start = atomic_load_8(&io->isStart); + assert(start == 0); + + if (io->serverRpc != NULL) { + free(io->serverRpc); + io->serverRpc = NULL; + } + + if (io->clientRpc != NULL) { + free(io->clientRpc); + io->clientRpc = NULL; + } + + if (io->pMsgQ != NULL) { + free(io->pMsgQ); + io->pMsgQ = NULL; + } + + if (io->pQset != NULL) { + free(io->pQset); + io->pQset = NULL; + } + + return 0; +} \ No newline at end of file diff --git a/source/libs/sync/src/syncRequestVote.c b/source/libs/sync/src/syncRequestVote.c index c31ec0f34d521b214b7d45a3ce32c3890ebcabd5..88056c95ff1fc93a7e67967d4c09244cb961ed5c 100644 --- a/source/libs/sync/src/syncRequestVote.c +++ b/source/libs/sync/src/syncRequestVote.c @@ -17,43 +17,39 @@ #include "sync.h" void requestVote(SRaft *pRaft, const SyncRequestVote *pMsg) { - -// TLA+ Spec -//RequestVote(i, j) == -// /\ state[i] = Candidate -// /\ j \notin votesResponded[i] -// /\ Send([mtype |-> RequestVoteRequest, -// mterm |-> currentTerm[i], -// mlastLogTerm |-> LastTerm(log[i]), -// mlastLogIndex |-> Len(log[i]), -// msource |-> i, -// mdest |-> j]) -// /\ UNCHANGED <> - + // TLA+ Spec + // RequestVote(i, j) == + // /\ state[i] = Candidate + // /\ j \notin votesResponded[i] + // /\ Send([mtype |-> RequestVoteRequest, + // mterm |-> currentTerm[i], + // mlastLogTerm |-> LastTerm(log[i]), + // mlastLogIndex |-> Len(log[i]), + // msource |-> i, + // mdest |-> j]) + // /\ UNCHANGED <> } void onRequestVote(SRaft *pRaft, const SyncRequestVote *pMsg) { - -// TLA+ Spec -//HandleRequestVoteRequest(i, j, m) == -// LET logOk == \/ m.mlastLogTerm > LastTerm(log[i]) -// \/ /\ m.mlastLogTerm = LastTerm(log[i]) -// /\ m.mlastLogIndex >= Len(log[i]) -// grant == /\ m.mterm = currentTerm[i] -// /\ logOk -// /\ votedFor[i] \in {Nil, j} -// IN /\ m.mterm <= currentTerm[i] -// /\ \/ grant /\ votedFor' = [votedFor EXCEPT ![i] = j] -// \/ ~grant /\ UNCHANGED votedFor -// /\ Reply([mtype |-> RequestVoteResponse, -// mterm |-> currentTerm[i], -// mvoteGranted |-> grant, -// \* mlog is used just for the `elections' history variable for -// \* the proof. It would not exist in a real implementation. -// mlog |-> log[i], -// msource |-> i, -// mdest |-> j], -// m) -// /\ UNCHANGED <> - + // TLA+ Spec + // HandleRequestVoteRequest(i, j, m) == + // LET logOk == \/ m.mlastLogTerm > LastTerm(log[i]) + // \/ /\ m.mlastLogTerm = LastTerm(log[i]) + // /\ m.mlastLogIndex >= Len(log[i]) + // grant == /\ m.mterm = currentTerm[i] + // /\ logOk + // /\ votedFor[i] \in {Nil, j} + // IN /\ m.mterm <= currentTerm[i] + // /\ \/ grant /\ votedFor' = [votedFor EXCEPT ![i] = j] + // \/ ~grant /\ UNCHANGED votedFor + // /\ Reply([mtype |-> RequestVoteResponse, + // mterm |-> currentTerm[i], + // mvoteGranted |-> grant, + // \* mlog is used just for the `elections' history variable for + // \* the proof. It would not exist in a real implementation. + // mlog |-> log[i], + // msource |-> i, + // mdest |-> j], + // m) + // /\ UNCHANGED <> } diff --git a/source/libs/sync/src/syncRequestVoteReply.c b/source/libs/sync/src/syncRequestVoteReply.c index ba9787f00c156a7b205a87108dd5670ff0f921b3..4ca1b1343f524d1cb8abbd499373b7636609c862 100644 --- a/source/libs/sync/src/syncRequestVoteReply.c +++ b/source/libs/sync/src/syncRequestVoteReply.c @@ -17,22 +17,20 @@ #include "sync.h" void onRequestVoteReply(SRaft *pRaft, const SyncRequestVoteReply *pMsg) { - -// TLA+ Spec -//HandleRequestVoteResponse(i, j, m) == -// \* This tallies votes even when the current state is not Candidate, but -// \* they won't be looked at, so it doesn't matter. -// /\ m.mterm = currentTerm[i] -// /\ votesResponded' = [votesResponded EXCEPT ![i] = -// votesResponded[i] \cup {j}] -// /\ \/ /\ m.mvoteGranted -// /\ votesGranted' = [votesGranted EXCEPT ![i] = -// votesGranted[i] \cup {j}] -// /\ voterLog' = [voterLog EXCEPT ![i] = -// voterLog[i] @@ (j :> m.mlog)] -// \/ /\ ~m.mvoteGranted -// /\ UNCHANGED <> -// /\ Discard(m) -// /\ UNCHANGED <> - + // TLA+ Spec + // HandleRequestVoteResponse(i, j, m) == + // \* This tallies votes even when the current state is not Candidate, but + // \* they won't be looked at, so it doesn't matter. + // /\ m.mterm = currentTerm[i] + // /\ votesResponded' = [votesResponded EXCEPT ![i] = + // votesResponded[i] \cup {j}] + // /\ \/ /\ m.mvoteGranted + // /\ votesGranted' = [votesGranted EXCEPT ![i] = + // votesGranted[i] \cup {j}] + // /\ voterLog' = [voterLog EXCEPT ![i] = + // voterLog[i] @@ (j :> m.mlog)] + // \/ /\ ~m.mvoteGranted + // /\ UNCHANGED <> + // /\ Discard(m) + // /\ UNCHANGED <> } diff --git a/source/libs/sync/test/syncTest.cpp b/source/libs/sync/test/syncTest.cpp index e069091ad88584bfca2162eb7fa379f0e1e3f7f1..cba196db277718055205353669e5c88bb3c63deb 100644 --- a/source/libs/sync/test/syncTest.cpp +++ b/source/libs/sync/test/syncTest.cpp @@ -1,26 +1,19 @@ #include +#include "syncIO.h" #include "syncInt.h" int main() { - printf("test \n"); + tsAsyncLog = 0; + taosInitLog((char*)"syncTest.log", 100000, 10); - syncStartEnv(); + sDebug("sync test"); + syncStartEnv(); - char temp[100]; - snprintf(temp, 100, "./debug.log"); - taosInitLog(temp, 10000, 1); - tsAsyncLog = 0; + SSyncIO *syncIO = syncIOCreate(); + assert(syncIO != NULL); - for (int i = 0; i < 100; i++) { - sDebug("log:%d -------- \n", i); - } - - fflush(NULL); - //taosCloseLog(); - - while(1) { - sleep(3); - } - return 0; + while (1) { + sleep(3); + } + return 0; } -