refact

include/util/amalloc.h

@@ -0,0 +1,47 @@
+/*
+ * 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_AMALLOC_H_
+#define _TD_AMALLOC_H_
+
+#include "os.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// Interfaces to implement
+typedef struct {
+  void *(*malloc)(void *, size_t size);
+  void *(*calloc)(void *, size_t nmemb, size_t size);
+  void (*free)(void *ptr, size_t size); // Do we need to set size in the allocated memory?
+  void *(*realloc)(void *ptr, size_t size);
+} SMemAllocatorIf;
+
+typedef struct {
+  void *          impl;
+  SMemAllocatorIf interface;
+} SMemAllocator;
+
+#define amalloc(allocator, size) (*((allocator)->interface.malloc))((allocator)->impl, size)
+#define acalloc(allocator, nmemb, size) (*((allocator)->interface.calloc))((allocator)->impl, nmemb, size)
+#define arealloc(allocator, ptr, size) (*((allocator)->interface.realloc))((allocator)->impl, ptr, size)
+#define afree(allocator, ptr, size) (*((allocator)->interface.free))((allocator)->impl, ptr, size)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_AMALLOC_H_*/

include/util/tlockfree.h

@@ -71,7 +71,7 @@ typedef void (*_ref_fn_t)(const void* pObj);
 
 
 // single writer multiple reader lock
-typedef int32_t SRWLatch;
+typedef volatile int32_t SRWLatch;
 
 void taosInitRWLatch(SRWLatch *pLatch);
 void taosWLockLatch(SRWLatch *pLatch);

source/server/vnode/CMakeLists.txt

@@ -14,6 +14,7 @@ target_link_libraries(
     PUBLIC meta
     PUBLIC tq
     PUBLIC tsdb
-    PRIVATE os
-    PRIVATE common
+    PUBLIC os
+    PUBLIC common
+    PUBLIC util
 )

source/server/vnode/inc/vnodeInt.h

@@ -16,6 +16,7 @@
 #ifndef _TD_VNODE_INT_H_
 #define _TD_VNODE_INT_H_
 
+#include "amalloc.h"
 #include "tq.h"
 #include "tsdb.h"
 #include "meta.h"
@@ -25,10 +26,10 @@ extern "C" {
 #endif
 
 typedef struct SVnode {
-  SMeta *pMeta;
-  STsdb *pTsdb;
-  STQ *  pTQ;
-  void * allocator; // TODO
+  SMeta *        pMeta;
+  STsdb *        pTsdb;
+  STQ *          pTQ;
+  SMemAllocator *allocator;
 } SVnode;
 
 #ifdef __cplusplus

source/server/vnode/inc/vnodeMemAllocator.h

@@ -16,6 +16,8 @@
 #ifndef _TD_VNODE_MEM_ALLOCATOR_H_
 #define _TD_VNODE_MEM_ALLOCATOR_H_
 
+#include "amalloc.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif

source/server/vnode/src/vnodeWrite.c

@@ -20,14 +20,14 @@ int vnodeProcessSubmitReq(SVnode *pVnode, SSubmitReq *pReq, SSubmitRsp *pRsp) {
 
 #if 1
   void *pMem = NULL;
-  if ((pMem = aMalloc(pVnode->allocator, REQ_SIZE(pReq))) == NULL) {
+  if ((pMem = amalloc(pVnode->allocator, REQ_SIZE(pReq))) == NULL) {
     // No more memory to allocate, schedule an async commit
     // and continue
     vnodeAsyncCommit(pVnode);
 
     // Reset allocator and allocat more
     vnodeResetAllocator(pVnode);
-    pMem = aMalloc(pVnode->allocator, REQ_SIZE(pReq));
+    pMem = amalloc(pVnode->allocator, REQ_SIZE(pReq));
     if (pMem == NULL) {
       // TODO: handle the error
     }
@@ -43,7 +43,7 @@ int vnodeProcessSubmitReq(SVnode *pVnode, SSubmitReq *pReq, SSubmitRsp *pRsp) {
   SSubmitReqReader reader;
   taosInitSubmitReqReader(&reader, (SSubmitReq *)pMem);
 
-  if (tsdbInsertData((SSubmitReq *)pMem) < 0) {
+  if (tsdbInsert(pVnode->pTsdb, (SSubmitReq *)pMem) < 0) {
     // TODO: handler error
   }