Skip to content

[To dev/1.3] Fix SessionDataSet sync-over-async deadlock and migrate to 4-arg RowRecord constructor #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CritasWang wants to merge 1 commit into
base: dev/1.3
Choose a base branch
from
Open

[To dev/1.3] Fix SessionDataSet sync-over-async deadlock and migrate to 4-arg RowRecord constructor #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions Apache-IoTDB-Client-CSharp-UserCase/Apache-IoTDB-Client-CSharp-UserCase.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk">

<PropertyGroup>
<OutputType>Exe</OutputType>
Expand All @@ -8,7 +8,7 @@
</PropertyGroup>

<ItemGroup>
<PackageReference Include="Apache.IoTDB" Version="1.0.0.3" />
<ProjectReference Include="..\src\Apache.IoTDB\Apache.IoTDB.csproj" />
</ItemGroup>

</Project>
</Project>
25 changes: 17 additions & 8 deletions Apache-IoTDB-Client-CSharp-UserCase/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Apache.IoTDB;
using Apache.IoTDB.DataStructure;
Expand All @@ -32,10 +31,19 @@ class Program
static int port = 6667;
static int pool_size = 2;

static SessionPool CreateSessionPool()
{
return new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(pool_size)
.Build();
}


static async Task OpenAndCloseSessionPool()
{
var session_pool = new SessionPool(host, port, pool_size);
var session_pool = CreateSessionPool();
await session_pool.Open(false);
if (session_pool.IsOpen())
{
Expand All @@ -50,7 +58,7 @@ static async Task OpenAndCloseSessionPool()

static async Task CreateTimeseries()
{
var session_pool = new SessionPool(host, port, pool_size);
var session_pool = CreateSessionPool();
await session_pool.Open(false);

await session_pool.DeleteDatabaseAsync("root.ln.wf01.wt01");
Expand All @@ -63,20 +71,21 @@ static async Task CreateTimeseries()

static async Task InsertRecord()
{
var session_pool = new SessionPool(host, port, pool_size);
var session_pool = CreateSessionPool();
await session_pool.Open(false);
long timestamp = 1;
var values = new List<object> { true, (double)1.1, "test" };
var measures = new List<string> { "status", "temperature", "hardware" };
var rowRecord = new RowRecord(timestamp, values, measures);
var dataTypes = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.DOUBLE, TSDataType.TEXT };
var rowRecord = new RowRecord(timestamp, values, measures, dataTypes);
var status = await session_pool.InsertRecordAsync("root.ln.wf01.wt01", rowRecord);

await session_pool.Close();
}

static async Task InsertTablet()
{
var session_pool = new SessionPool(host, port, pool_size);
var session_pool = CreateSessionPool();
await session_pool.Open(false);
var device_id = "root.ln.wf01.wt01";
var measurement_lst = new List<string> { "status", "temperature", "hardware" };
Expand All @@ -95,11 +104,11 @@ static async Task InsertTablet()

static async Task ExecuteQueryStatement()
{
var session_pool = new SessionPool(host, port, pool_size);
var session_pool = CreateSessionPool();
await session_pool.Open(false);
var res = await session_pool.ExecuteQueryStatementAsync("select * from root.ln.wf01.wt01");
res.ShowTableNames();
while (res.HasNext())
while (await res.HasNextAsync())
{
Console.WriteLine(res.Next());
}
Expand Down
783 changes: 397 additions & 386 deletions samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs
View file
Open in desktop

Large diffs are not rendered by default.

242 changes: 121 additions & 121 deletions samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,100 +25,100 @@

namespace Apache.IoTDB.Samples
{
public partial class SessionPoolTest
{
public async Task TestInsertAlignedTablet()
public partial class SessionPoolTest
{
var session_pool = new SessionPool(host, port, poolSize);
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
public async Task TestInsertAlignedTablet()
{
var session_pool = new SessionPool(host, port, poolSize);
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

System.Diagnostics.Debug.Assert(session_pool.IsOpen());
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
var measurement_lst = new List<string>
System.Diagnostics.Debug.Assert(session_pool.IsOpen());
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
var measurement_lst = new List<string>
{ testMeasurements[1],
testMeasurements[2],
testMeasurements[3]
};
var value_lst = new List<List<object>>
var value_lst = new List<List<object>>
{
new() {"iotdb", true, (int) 12}, new() {"c#", false, (int) 13},
new() {"client", true, (int) 14}
};
var timestamp_lst = new List<long> { 1, 2, 3 };
var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 };
var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst);
status = await session_pool.InsertAlignedTabletAsync(tablet);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15");
res.ShowTableNames();
while (res.HasNext()) Console.WriteLine(res.Next());
var timestamp_lst = new List<long> { 1, 2, 3 };
var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 };
var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst);
status = await session_pool.InsertAlignedTabletAsync(tablet);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15");
res.ShowTableNames();
while (await res.HasNextAsync()) Console.WriteLine(res.Next());

await res.Close();
// large data test
value_lst = new List<List<object>>() { };
timestamp_lst = new List<long>() { };
var tasks = new List<Task<int>>();
var start_ms = DateTime.Now.Ticks / 10000;
for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++)
{
timestamp_lst.Add(timestamp);
value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp });
if (timestamp % fetchSize == 0)
{
tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst);
tasks.Add(session_pool.InsertAlignedTabletAsync(tablet));
value_lst = new List<List<object>>() { };
timestamp_lst = new List<long>() { };
}
}
Console.WriteLine(tasks.Count);
await res.Close();
// large data test
value_lst = new List<List<object>>() { };
timestamp_lst = new List<long>() { };
var tasks = new List<Task<int>>();
var start_ms = DateTime.Now.Ticks / 10000;
for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++)
{
timestamp_lst.Add(timestamp);
value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp });
if (timestamp % fetchSize == 0)
{
tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst);
tasks.Add(session_pool.InsertAlignedTabletAsync(tablet));
value_lst = new List<List<object>>() { };
timestamp_lst = new List<long>() { };
}
}
Console.WriteLine(tasks.Count);

Task.WaitAll(tasks.ToArray());
var end_ms = DateTime.Now.Ticks / 10000;
Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms));
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
{
res.Next();
res_count += 1;
}
Task.WaitAll(tasks.ToArray());
var end_ms = DateTime.Now.Ticks / 10000;
Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms));
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
res.ShowTableNames();
var res_count = 0;
while (await res.HasNextAsync())
{
res.Next();
res_count += 1;
}

await res.Close();
Console.WriteLine(res_count + " " + fetchSize * processedSize);
System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize);
status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
System.Diagnostics.Debug.Assert(status == 0);
await session_pool.Close();
Console.WriteLine("TestInsertAlignedTablet Passed!");
}
await res.Close();
Console.WriteLine(res_count + " " + fetchSize * processedSize);
System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize);
status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
System.Diagnostics.Debug.Assert(status == 0);
await session_pool.Close();
Console.WriteLine("TestInsertAlignedTablet Passed!");
}

public async Task TestInsertAlignedTablets()
{
var session_pool = new SessionPool(host, port, poolSize);
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
public async Task TestInsertAlignedTablets()
{
var session_pool = new SessionPool(host, port, poolSize);
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

System.Diagnostics.Debug.Assert(session_pool.IsOpen());
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = new List<string>()
System.Diagnostics.Debug.Assert(session_pool.IsOpen());
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = new List<string>()
{
string.Format("{0}.{1}", testDatabaseName, testDevices[1]),
string.Format("{0}.{1}", testDatabaseName, testDevices[2])
};
var measurements_lst = new List<List<string>>()
var measurements_lst = new List<List<string>>()
{
new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] },
new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] }
};
var values_lst = new List<List<List<object>>>()
var values_lst = new List<List<List<object>>>()
{
new()
{
Expand All @@ -131,65 +131,65 @@ public async Task TestInsertAlignedTablets()
new List<object>() {"client_2", true, (int) 3}
}
};
var datatype_lst = new List<List<TSDataType>>()
var datatype_lst = new List<List<TSDataType>>()
{
new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32},
new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32}
};
var timestamp_lst = new List<List<long>>()
var timestamp_lst = new List<List<long>>()
{new() {2, 1, 3}, new() {3, 1, 2}};
var tablets = new List<Tablet>() { };
for (var i = 0; i < device_id.Count; i++)
{
var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]);
tablets.Add(tablet);
}
var tablets = new List<Tablet>() { };
for (var i = 0; i < device_id.Count; i++)
{
var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]);
tablets.Add(tablet);
}

status = await session_pool.InsertAlignedTabletsAsync(tablets);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15");
res.ShowTableNames();
while (res.HasNext()) Console.WriteLine(res.Next());
status = await session_pool.InsertAlignedTabletsAsync(tablets);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15");
res.ShowTableNames();
while (await res.HasNextAsync()) Console.WriteLine(res.Next());

// large data test
var tasks = new List<Task<int>>();
// tablets = new List<Tablet>() { };
for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++)
{
var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]);
var local_measurements = new List<string>()
// large data test
var tasks = new List<Task<int>>();
// tablets = new List<Tablet>() { };
for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++)
{
var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]);
var local_measurements = new List<string>()
{testMeasurements[1], testMeasurements[2], testMeasurements[3]};
var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } };
var local_data_type = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 };
var local_timestamp = new List<long> { timestamp };
var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp);
tablets.Add(tablet);
if (timestamp % fetchSize == 0)
{
tasks.Add(session_pool.InsertAlignedTabletsAsync(tablets));
tablets = new List<Tablet>() { };
}
}
var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } };
var local_data_type = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 };
var local_timestamp = new List<long> { timestamp };
var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp);
tablets.Add(tablet);
if (timestamp % fetchSize == 0)
{
tasks.Add(session_pool.InsertAlignedTabletsAsync(tablets));
tablets = new List<Tablet>() { };
}
}

Task.WaitAll(tasks.ToArray());
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
{
res.Next();
res_count += 1;
}
Task.WaitAll(tasks.ToArray());
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]));
res.ShowTableNames();
var res_count = 0;
while (await res.HasNextAsync())
{
res.Next();
res_count += 1;
}

await res.Close();
Console.WriteLine(res_count + " " + fetchSize * processedSize);
System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize);
status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
System.Diagnostics.Debug.Assert(status == 0);
await session_pool.Close();
Console.WriteLine("TestInsertAlignedTablets Passed!");
await res.Close();
Console.WriteLine(res_count + " " + fetchSize * processedSize);
System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize);
status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
System.Diagnostics.Debug.Assert(status == 0);
await session_pool.Close();
Console.WriteLine("TestInsertAlignedTablets Passed!");
}
}
}
}
Loading
Loading