提交 1549819b 编写于 作者: H Hongze Cheng

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

......@@ -2,7 +2,7 @@
# taosws-rs
ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG e771403
GIT_TAG 76bc64d
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
bin
obj
.vs
*.sln
\ No newline at end of file
*.sln
wsConnect/obj
wsInsert/obj
wsQuery/obj
wsStmt/obj
wsConnect/bin
wsInsert/bin
wsQuery/bin
wsStmt/bin
\ No newline at end of file
using System;
using TDengineWS.Impl;
namespace Examples
{
public class WSConnExample
{
static void Main(string[] args)
{
string DSN = "ws://root:taosdata@127.0.0.1:6041/test";
IntPtr wsConn = LibTaosWS.WSConnectWithDSN(DSN);
if (wsConn == IntPtr.Zero)
{
throw new Exception($"get WS connection failed,reason:{LibTaosWS.WSErrorStr(IntPtr.Zero)} code:{LibTaosWS.WSErrorNo(IntPtr.Zero)}");
}
else
{
Console.WriteLine("Establish connect success.");
}
// close connection.
LibTaosWS.WSClose(wsConn);
}
}
}
\ No newline at end of file
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="TDengine.Connector" Version="3.0.1" />
</ItemGroup>
</Project>
using System;
using TDengineWS.Impl;
namespace Examples
{
public class WSInsertExample
{
static void Main(string[] args)
{
string DSN = "ws://root:taosdata@127.0.0.1:6041/test";
IntPtr wsConn = LibTaosWS.WSConnectWithDSN(DSN);
// Assert if connection is validate
if (wsConn == IntPtr.Zero)
{
throw new Exception($"get WS connection failed,reason:{LibTaosWS.WSErrorStr(IntPtr.Zero)} code:{LibTaosWS.WSErrorNo(IntPtr.Zero)}");
}
else
{
Console.WriteLine("Establish connect success.");
}
string createTable = "CREATE STABLE test.meters (ts timestamp, current float, voltage int, phase float) TAGS (location binary(64), groupId int);";
string insert = "INSERT INTO test.d1001 USING test.meters TAGS('California.SanFrancisco', 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000)" +
"test.d1002 USING test.meters TAGS('California.SanFrancisco', 3) VALUES('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000)" +
"test.d1003 USING test.meters TAGS('California.LosAngeles', 2) VALUES('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000)('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000) " +
"test.d1004 USING test.meters TAGS('California.LosAngeles', 3) VALUES('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000)('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)";
IntPtr wsRes = LibTaosWS.WSQuery(wsConn, createTable);
ValidInsert("create table", wsRes);
LibTaosWS.WSFreeResult(wsRes);
wsRes = LibTaosWS.WSQuery(wsConn, insert);
ValidInsert("insert data", wsRes);
LibTaosWS.WSFreeResult(wsRes);
// close connection.
LibTaosWS.WSClose(wsConn);
}
static void ValidInsert(string desc, IntPtr wsRes)
{
int code = LibTaosWS.WSErrorNo(wsRes);
if (code != 0)
{
throw new Exception($"execute SQL failed: reason: {LibTaosWS.WSErrorStr(wsRes)}, code:{code}");
}
else
{
Console.WriteLine("{0} success affect {2} rows, cost {1} nanoseconds", desc, LibTaosWS.WSTakeTiming(wsRes),LibTaosWS.WSAffectRows(wsRes));
}
}
}
}
// Establish connect success.
// create table success affect 0 rows, cost 3717542 nanoseconds
// insert data success affect 8 rows, cost 2613637 nanoseconds
\ No newline at end of file
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="TDengine.Connector" Version="3.0.1" />
</ItemGroup>
</Project>
using System;
using TDengineWS.Impl;
using System.Collections.Generic;
using TDengineDriver;
namespace Examples
{
public class WSQueryExample
{
static void Main(string[] args)
{
string DSN = "ws://root:taosdata@127.0.0.1:6041/test";
IntPtr wsConn = LibTaosWS.WSConnectWithDSN(DSN);
if (wsConn == IntPtr.Zero)
{
throw new Exception($"get WS connection failed,reason:{LibTaosWS.WSErrorStr(IntPtr.Zero)} code:{LibTaosWS.WSErrorNo(IntPtr.Zero)}");
}
else
{
Console.WriteLine("Establish connect success.");
}
string select = "select * from test.meters";
// optional:wsRes = LibTaosWS.WSQuery(wsConn, select);
IntPtr wsRes = LibTaosWS.WSQueryTimeout(wsConn, select, 1);
// Assert if query execute success.
int code = LibTaosWS.WSErrorNo(wsRes);
if (code != 0)
{
throw new Exception($"execute SQL failed: reason: {LibTaosWS.WSErrorStr(wsRes)}, code:{code}");
}
// get meta data
List<TDengineMeta> metas = LibTaosWS.WSGetFields(wsRes);
// get retrieved data
List<object> dataSet = LibTaosWS.WSGetData(wsRes);
// do something with result.
foreach (var meta in metas)
{
Console.Write("{0} {1}({2}) \t|\t", meta.name, meta.TypeName(), meta.size);
}
Console.WriteLine("");
for (int i = 0; i < dataSet.Count;)
{
for (int j = 0; j < metas.Count; j++)
{
Console.Write("{0}\t|\t", dataSet[i]);
i++;
}
Console.WriteLine("");
}
// Free result after use.
LibTaosWS.WSFreeResult(wsRes);
// close connection.
LibTaosWS.WSClose(wsConn);
}
}
}
// Establish connect success.
// ts TIMESTAMP(8) | current FLOAT(4) | voltage INT(4) | phase FLOAT(4) | location BINARY(64) | groupid INT(4) |
// 1538548685000 | 10.8 | 223 | 0.29 | California.LosAngeles | 3 |
// 1538548686500 | 11.5 | 221 | 0.35 | California.LosAngeles | 3 |
// 1538548685500 | 11.8 | 221 | 0.28 | California.LosAngeles | 2 |
// 1538548696600 | 13.4 | 223 | 0.29 | California.LosAngeles | 2 |
// 1538548685000 | 10.3 | 219 | 0.31 | California.SanFrancisco | 2 |
// 1538548695000 | 12.6 | 218 | 0.33 | California.SanFrancisco | 2 |
// 1538548696800 | 12.3 | 221 | 0.31 | California.SanFrancisco | 2 |
// 1538548696650 | 10.3 | 218 | 0.25 | California.SanFrancisco | 3 |
\ No newline at end of file
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="TDengine.Connector" Version="3.0.1" />
</ItemGroup>
</Project>
using System;
using TDengineWS.Impl;
using TDengineDriver;
using System.Runtime.InteropServices;
namespace Examples
{
public class WSStmtExample
{
static void Main(string[] args)
{
const string DSN = "ws://root:taosdata@127.0.0.1:6041/test";
const string table = "meters";
const string database = "test";
const string childTable = "d1005";
string insert = $"insert into ? using {database}.{table} tags(?,?) values(?,?,?,?)";
const int numOfTags = 2;
const int numOfColumns = 4;
// Establish connection
IntPtr wsConn = LibTaosWS.WSConnectWithDSN(DSN);
if (wsConn == IntPtr.Zero)
{
throw new Exception($"get WS connection failed,reason:{LibTaosWS.WSErrorStr(IntPtr.Zero)} code:{LibTaosWS.WSErrorNo(IntPtr.Zero)}");
}
else
{
Console.WriteLine("Establish connect success...");
}
// init stmt
IntPtr wsStmt = LibTaosWS.WSStmtInit(wsConn);
if (wsStmt != IntPtr.Zero)
{
int code = LibTaosWS.WSStmtPrepare(wsStmt, insert);
ValidStmtStep(code, wsStmt, "WSStmtPrepare");
TAOS_MULTI_BIND[] wsTags = new TAOS_MULTI_BIND[] { WSMultiBind.WSBindNchar(new string[] { "California.SanDiego" }), WSMultiBind.WSBindInt(new int?[] { 4 }) };
code = LibTaosWS.WSStmtSetTbnameTags(wsStmt, $"{database}.{childTable}", wsTags, numOfTags);
ValidStmtStep(code, wsStmt, "WSStmtSetTbnameTags");
TAOS_MULTI_BIND[] data = new TAOS_MULTI_BIND[4];
data[0] = WSMultiBind.WSBindTimestamp(new long[] { 1538548687000, 1538548688000, 1538548689000, 1538548690000, 1538548691000 });
data[1] = WSMultiBind.WSBindFloat(new float?[] { 10.30F, 10.40F, 10.50F, 10.60F, 10.70F });
data[2] = WSMultiBind.WSBindInt(new int?[] { 223, 221, 222, 220, 219 });
data[3] = WSMultiBind.WSBindFloat(new float?[] { 0.31F, 0.32F, 0.33F, 0.35F, 0.28F });
code = LibTaosWS.WSStmtBindParamBatch(wsStmt, data, numOfColumns);
ValidStmtStep(code, wsStmt, "WSStmtBindParamBatch");
code = LibTaosWS.WSStmtAddBatch(wsStmt);
ValidStmtStep(code, wsStmt, "WSStmtAddBatch");
IntPtr stmtAffectRowPtr = Marshal.AllocHGlobal(Marshal.SizeOf(typeof(Int32)));
code = LibTaosWS.WSStmtExecute(wsStmt, stmtAffectRowPtr);
ValidStmtStep(code, wsStmt, "WSStmtExecute");
Console.WriteLine("WS STMT insert {0} rows...", Marshal.ReadInt32(stmtAffectRowPtr));
Marshal.FreeHGlobal(stmtAffectRowPtr);
LibTaosWS.WSStmtClose(wsStmt);
// Free unmanaged memory
WSMultiBind.WSFreeTaosBind(wsTags);
WSMultiBind.WSFreeTaosBind(data);
//check result with SQL "SELECT * FROM test.d1005;"
}
else
{
throw new Exception("Init STMT failed...");
}
// close connection.
LibTaosWS.WSClose(wsConn);
}
static void ValidStmtStep(int code, IntPtr wsStmt, string desc)
{
if (code != 0)
{
throw new Exception($"{desc} failed,reason: {LibTaosWS.WSErrorStr(wsStmt)}, code: {code}");
}
else
{
Console.WriteLine("{0} success...", desc);
}
}
}
}
// WSStmtPrepare success...
// WSStmtSetTbnameTags success...
// WSStmtBindParamBatch success...
// WSStmtAddBatch success...
// WSStmtExecute success...
// WS STMT insert 5 rows...
\ No newline at end of file
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="TDengine.Connector" Version="3.0.1" />
</ItemGroup>
</Project>
......@@ -2,7 +2,7 @@
{{#include docs/examples/csharp/ConnectExample.cs}}
```
:::info
C# 连接器目前只支持原生连接。
```csharp title="WebSocket 连接"
{{#include docs/examples/csharp/wsConnect/Program.cs}}
```
:::
......@@ -17,7 +17,7 @@ import CSAsyncQuery from "../07-develop/04-query-data/_cs_async.mdx"
`TDengine.Connector` 是 TDengine 提供的 C# 语言连接器。C# 开发人员可以通过它开发存取 TDengine 集群数据的 C# 应用软件。
`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动(taosc)建立与 TDengine 运行实例的连接,提供数据写入、查询、订阅、schemaless 数据写入、参数绑定接口数据写入等功能 `TDengine.Connector` 目前暂未提供 REST 连接方式,用户可以参考 [REST API](../rest-api/) 文档自行编写
`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动(taosc)建立与 TDengine 运行实例的连接,提供数据写入、查询、数据订阅、schemaless 数据写入、参数绑定接口数据写入等功能。 `TDengine.Connector` 还支持 WebSocket,通过 DSN 建立 WebSocket 连接,提供数据写入、查询、参数绑定接口数据写入等功能
本文介绍如何在 Linux 或 Windows 环境中安装 `TDengine.Connector`,并通过 `TDengine.Connector` 连接 TDengine 集群,进行数据写入、查询等基本操作。
......@@ -35,12 +35,29 @@ import CSAsyncQuery from "../07-develop/04-query-data/_cs_async.mdx"
## 支持的功能特性
<Tabs defaultValue="native">
<TabItem value="native" label="原生连接">
1. 连接管理
2. 普通查询
3. 连续查询
4. 参数绑定
5. 订阅功能
5. 数据订阅(TMQ)
6. Schemaless
</TabItem>
<TabItem value="rest" label="WebSocket 连接">
1. 连接管理
2. 普通查询
3. 连续查询
4. 参数绑定
</TabItem>
</Tabs>
## 安装步骤
......@@ -79,7 +96,13 @@ dotnet add exmaple.csproj reference src/TDengine.csproj
## 建立连接
``` C#
<Tabs defaultValue="native">
<TabItem value="native" label="原生连接">
使用 host、username、password、port 等信息建立连接。
``` csharp
using TDengineDriver;
namespace TDengineExample
......@@ -109,17 +132,63 @@ namespace TDengineExample
}
}
}
```
</TabItem>
<TabItem value="rest" label="WebSocket 连接">
使用 DSN 建立 WebSocket 连接 DSN 连接。 描述字符串基本结构如下:
```text
[<protocol>]://[[<username>:<password>@]<host>:<port>][/<database>][?<p1>=<v1>[&<p2>=<v2>]]
|------------|---|-----------|-----------|------|------|------------|-----------------------|
| protocol | | username | password | host | port | database | params |
```
各部分意义见下表:
* **protocol**: 显示指定以何种方式建立连接,例如:`ws://localhost:6041` 指定以 Websocket 方式建立连接(支持http/ws)。
* **username/password**: 用于创建连接的用户名及密码(默认`root/taosdata`)。
* **host/port**: 指定创建连接的服务器及端口,WebSocket 连接默认为 `localhost:6041` 。
* **database**: 指定默认连接的数据库名。
* **params**:其他可选参数。
``` csharp
{{#include docs/examples/csharp/wsConnect/Program.cs}}
```
</TabItem>
</Tabs>
## 使用示例
### 写入数据
#### SQL 写入
<Tabs defaultValue="native">
<TabItem value="native" label="原生连接">
<CSInsert />
</TabItem>
<TabItem value="rest" label="WebSocket 连接">
```csharp
{{#include docs/examples/csharp/wsInsert/Program.cs}}
```
</TabItem>
</Tabs>
#### InfluxDB 行协议写入
<CSInfluxLine />
......@@ -132,12 +201,50 @@ namespace TDengineExample
<CSOpenTSDBJson />
#### 参数绑定
<Tabs defaultValue="native">
<TabItem value="native" label="原生连接">
``` csharp
{{#include docs/examples/csharp/StmtInsertExample.cs}}
```
</TabItem>
<TabItem value="rest" label="WebSocket 连接">
```csharp
{{#include docs/examples/csharp/wsStmt/Program.cs}}
```
</TabItem>
</Tabs>
### 查询数据
#### 同步查询
<Tabs defaultValue="native">
<TabItem value="native" label="原生连接">
<CSQuery />
</TabItem>
<TabItem value="rest" label="WebSocket 连接">
```csharp
{{#include docs/examples/csharp/wsQuery/Program.cs}}
```
</TabItem>
</Tabs>
#### 异步查询
<CSAsyncQuery />
......@@ -151,12 +258,15 @@ namespace TDengineExample
| [stmt](https://github.com/taosdata/taos-connector-dotnet/tree/3.0/examples/Stmt) | 使用 TDengine.Connector 实现的参数绑定插入和查询的示例 |
| [schemaless](https://github.com/taosdata/taos-connector-dotnet/blob/3.0/examples/schemaless) | 使用 TDengine.Connector 实现的使用 schemaless 写入的示例 |
| [async query](https://github.com/taosdata/taos-connector-dotnet/blob/3.0/examples/AsyncQuery/QueryAsync.cs) | 使用 TDengine.Connector 实现的异步查询的示例 |
| [TMQ](https://github.com/taosdata/taos-connector-dotnet/blob/3.0/examples/TMQ/TMQ.cs) | 使用 TDengine.Connector 实现的订阅数据的示例 |
| [数据订阅(TMQ)](https://github.com/taosdata/taos-connector-dotnet/blob/3.0/examples/TMQ/TMQ.cs) | 使用 TDengine.Connector 实现的订阅数据的示例 |
| [Basic WebSocket Usage](https://github.com/taosdata/taos-connector-dotnet/blob/5a4a7cd0dbcda114447cdc6d0c6dedd8e84a52da/examples/WS/WebSocketSample.cs) | 使用 TDengine.Connector 的 WebSocket 基本的示例 |
| [Basic WebSocket STMT](https://github.com/taosdata/taos-connector-dotnet/blob/5a4a7cd0dbcda114447cdc6d0c6dedd8e84a52da/examples/WS/WebSocketSTMT.cs) | 使用 TDengine.Connector 的 WebSocket STMT 基本的示例 |
## 重要更新记录
| TDengine.Connector | 说明 |
|--------------------|--------------------------------|
| 3.0.1 | 支持 WebSocket 和 Cloud,查询,插入,参数绑定。 |
| 3.0.0 | 支持 TDengine 3.0.0.0,不兼容 2.x。新增接口TDengine.Impl.GetData(),解析查询结果。 |
| 1.0.7 | 修复 TDengine.Query()内存泄露。 |
| 1.0.6 | 修复 schemaless 在 1.0.4 和 1.0.5 中失效 bug。 |
......
......@@ -1233,7 +1233,7 @@ SELECT SERVER_VERSION();
### SERVER_STATUS
```sql
SELECT SERVER_VERSION();
SELECT SERVER_STATUS();
```
**说明**:返回服务端当前的状态。
......@@ -22,6 +22,7 @@ extern "C" {
#include "cJSON.h"
#include "tdef.h"
#include "tlrucache.h"
#include "tmsgcb.h"
extern bool gRaftDetailLog;
......@@ -153,7 +154,8 @@ typedef struct SSyncFSM {
// abstract definition of log store in raft
// SWal implements it
typedef struct SSyncLogStore {
void* data;
SLRUCache* pCache;
void* data;
// append one log entry
int32_t (*appendEntry)(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
......
......@@ -362,8 +362,9 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
// update uid index
metaUpdateUidIdx(pMeta, &nStbEntry);
if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);
metaULock(pMeta);
if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);
tDecoderClear(&dc);
tdbTbcClose(pTbDbc);
tdbTbcClose(pUidIdxc);
......@@ -922,6 +923,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
taosArrayDestroy(pTagArray);
}
metaWLock(pMeta);
// save to table.db
metaSaveToTbDb(pMeta, &ctbEntry);
......@@ -936,6 +939,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
tdbTbUpsert(pMeta->pCtbIdx, &ctbIdxKey, sizeof(ctbIdxKey), ctbEntry.ctbEntry.pTags,
((STag *)(ctbEntry.ctbEntry.pTags))->len, &pMeta->txn);
metaULock(pMeta);
tDecoderClear(&dc1);
tDecoderClear(&dc2);
if (ctbEntry.ctbEntry.pTags) taosMemoryFree((void *)ctbEntry.ctbEntry.pTags);
......
......@@ -69,15 +69,26 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
if (agree) {
// term
SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index);
ASSERT(pEntry != NULL);
SSyncRaftEntry* pEntry = NULL;
SLRUCache* pCache = pSyncNode->pLogStore->pCache;
LRUHandle* h = taosLRUCacheLookup(pCache, &index, sizeof(index));
if (h) {
pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
} else {
pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index);
ASSERT(pEntry != NULL);
}
// cannot commit, even if quorum agree. need check term!
if (pEntry->term <= pSyncNode->pRaftStore->currentTerm) {
// update commit index
newCommitIndex = index;
syncEntryDestory(pEntry);
if (h) {
taosLRUCacheRelease(pCache, h, false);
} else {
syncEntryDestory(pEntry);
}
break;
} else {
do {
......@@ -88,7 +99,11 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
} while (0);
}
syncEntryDestory(pEntry);
if (h) {
taosLRUCacheRelease(pCache, h, false);
} else {
syncEntryDestory(pEntry);
}
}
}
......
......@@ -2581,6 +2581,20 @@ static int32_t syncNodeEqNoop(SSyncNode* ths) {
return ret;
}
static void deleteCacheEntry(const void* key, size_t keyLen, void* value) { taosMemoryFree(value); }
static int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h) {
int code = 0;
int entryLen = sizeof(*pEntry) + pEntry->dataLen;
LRUStatus status = taosLRUCacheInsert(pLogStore->pCache, &pEntry->index, sizeof(pEntry->index), pEntry, entryLen,
deleteCacheEntry, h, TAOS_LRU_PRIORITY_LOW);
if (status != TAOS_LRU_STATUS_OK) {
code = -1;
}
return code;
}
static int32_t syncNodeAppendNoop(SSyncNode* ths) {
int32_t ret = 0;
......@@ -2589,13 +2603,21 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) {
SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
ASSERT(pEntry != NULL);
LRUHandle* h = NULL;
syncCacheEntry(ths->pLogStore, pEntry, &h);
if (ths->state == TAOS_SYNC_STATE_LEADER) {
int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
ASSERT(code == 0);
syncNodeReplicate(ths, false);
}
syncEntryDestory(pEntry);
if (h) {
taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
} else {
syncEntryDestory(pEntry);
}
return ret;
}
......@@ -2654,6 +2676,9 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg, SyncI
SSyncRaftEntry* pEntry = syncEntryBuild2((SyncClientRequest*)pMsg, term, index);
ASSERT(pEntry != NULL);
LRUHandle* h = NULL;
syncCacheEntry(ths->pLogStore, pEntry, &h);
if (ths->state == TAOS_SYNC_STATE_LEADER) {
// append entry
code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
......@@ -2685,7 +2710,12 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg, SyncI
}
}
syncEntryDestory(pEntry);
if (h) {
taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
} else {
syncEntryDestory(pEntry);
}
return ret;
}
......@@ -2973,9 +3003,15 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
if (i != SYNC_INDEX_INVALID) {
SSyncRaftEntry* pEntry;
code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry);
ASSERT(code == 0);
ASSERT(pEntry != NULL);
SLRUCache* pCache = ths->pLogStore->pCache;
LRUHandle* h = taosLRUCacheLookup(pCache, &i, sizeof(i));
if (h) {
pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
} else {
code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry);
ASSERT(code == 0);
ASSERT(pEntry != NULL);
}
SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pEntry, &rpcMsg);
......@@ -3058,7 +3094,11 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
}
rpcFreeCont(rpcMsg.pCont);
syncEntryDestory(pEntry);
if (h) {
taosLRUCacheRelease(pCache, h, false);
} else {
syncEntryDestory(pEntry);
}
}
}
}
......
......@@ -53,6 +53,15 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
ASSERT(pLogStore != NULL);
pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5);
if (pLogStore->pCache == NULL) {
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
taosMemoryFree(pLogStore);
return NULL;
}
taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);
pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
ASSERT(pLogStore->data != NULL);
......@@ -102,6 +111,10 @@ void logStoreDestory(SSyncLogStore* pLogStore) {
taosThreadMutexDestroy(&(pData->mutex));
taosMemoryFree(pLogStore->data);
taosLRUCacheEraseUnrefEntries(pLogStore->pCache);
taosLRUCacheCleanup(pLogStore->pCache);
taosMemoryFree(pLogStore);
}
}
......@@ -243,7 +256,7 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
int32_t code;
int32_t code = 0;
*ppEntry = NULL;
......@@ -257,6 +270,7 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
taosThreadMutexLock(&(pData->mutex));
code = walReadVer(pWalHandle, index);
// code = walReadVerCached(pWalHandle, index);
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
......@@ -412,6 +426,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
ASSERT(pWalHandle != NULL);
int32_t code = walReadVer(pWalHandle, index);
// int32_t code = walReadVerCached(pWalHandle, index);
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
......
......@@ -284,7 +284,6 @@ static char* evacOneDataPage(SDiskbasedBuf* pBuf) {
// all pages are referenced by user, try to allocate new space
if (pn == NULL) {
assert(0);
int32_t prev = pBuf->inMemPages;
// increase by 50% of previous mem pages
......@@ -304,7 +303,6 @@ static char* evacOneDataPage(SDiskbasedBuf* pBuf) {
bufPage = flushPageToDisk(pBuf, d);
}
ASSERT((bufPage != NULL) || terrno != TSDB_CODE_SUCCESS);
return bufPage;
}
......@@ -377,12 +375,6 @@ void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
char* availablePage = NULL;
if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
availablePage = evacOneDataPage(pBuf);
// Failed to allocate a new buffer page, and there is an error occurs.
if (availablePage == NULL) {
assert(0);
return NULL;
}
}
SPageInfo* pi = NULL;
......@@ -652,4 +644,4 @@ void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
pBuf->totalBufSize = 0;
pBuf->allocateId = -1;
pBuf->fileSize = 0;
}
\ No newline at end of file
}
......@@ -6,23 +6,24 @@ pgrep taosd || taosd >> /dev/null 2>&1 &
pgrep taosadapter || taosadapter >> /dev/null 2>&1 &
cd ../../docs/examples/csharp
dotnet run --project connect.csproj
#dotnet run --project connect.csproj
taos -s "drop database if exists power"
dotnet run --project sqlinsert.csproj
dotnet run --project query.csproj
dotnet run --project asyncquery.csproj
dotnet run --project subscribe.csproj
#taos -s "drop database if exists power"
#dotnet run --project sqlinsert.csproj
#dotnet run --project query.csproj
#dotnet run --project asyncquery.csproj
#dotnet run --project subscribe.csproj
taos -s "drop topic if exists topic_example"
taos -s "drop database if exists power"
dotnet run --project stmtinsert.csproj
#taos -s "drop topic if exists topic_example"
#taos -s "drop database if exists power"
#dotnet run --project stmtinsert.csproj
taos -s "drop database if exists test"
dotnet run --project influxdbline.csproj
#taos -s "drop database if exists test"
#dotnet run --project influxdbline.csproj
taos -s "drop database if exists test"
dotnet run --project optstelnet.csproj
#taos -s "drop database if exists test"
#dotnet run --project optstelnet.csproj
taos -s "drop database if exists test"
dotnet run --project optsjson.csproj
\ No newline at end of file
#taos -s "drop database if exists test"
#dotnet run --project optsjson.csproji
echo "uncomment temporily"
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册