xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #28791 from taosdata/fix/TS-4937-2 

enh(tsdb/cache): update lru when committing
This commit is contained in:
Shengliang Guan 2024-12-16 15:46:48 +08:00 committed by GitHub
parent db9da46307 3d5e54259f
commit 84b7e9b312
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 1032 additions and 344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/common/tglobal.h
View File

@ -85,7 +85,7 @@ extern int32_t tsQueryConcurrentTaskNum;
extern int32_t tsSingleQueryMaxMemorySize;
extern int8_t tsQueryUseMemoryPool;
extern int8_t tsMemPoolFullFunc;
//extern int32_t tsQueryBufferPoolSize;
// extern int32_t tsQueryBufferPoolSize;
extern int32_t tsMinReservedMemorySize;
extern int64_t tsCurrentAvailMemorySize;
extern int8_t tsNeedTrim;
@ -283,7 +283,7 @@ extern int32_t tsS3MigrateIntervalSec;
extern bool tsS3MigrateEnabled;
extern int32_t tsGrantHBInterval;
extern int32_t tsUptimeInterval;
extern bool tsUpdateCacheBatch;
extern bool tsDisableStream;
extern int64_t tsStreamBufferSize;
extern int tsStreamAggCnt;

2
source/common/src/tglobal.c
View File

@ -329,6 +329,8 @@ bool tsFilterScalarMode = false;
int tsResolveFQDNRetryTime = 100; // seconds
int tsStreamAggCnt = 100000;
bool tsUpdateCacheBatch = true;
int8_t tsS3EpNum = 0;
char tsS3Endpoint[TSDB_MAX_EP_NUM][TSDB_FQDN_LEN] = {"<endpoint>"};
char tsS3AccessKey[TSDB_MAX_EP_NUM][TSDB_FQDN_LEN] = {"<accesskey>"};

3
source/dnode/vnode/src/inc/tsdb.h
View File

@ -250,6 +250,7 @@ void *tsdbTbDataIterDestroy(STbDataIter *pIter);
void tsdbTbDataIterOpen(STbData *pTbData, STsdbRowKey *pFrom, int8_t backward, STbDataIter *pIter);
bool tsdbTbDataIterNext(STbDataIter *pIter);
void tsdbMemTableCountRows(SMemTable *pMemTable, SSHashObj *pTableMap, int64_t *rowsNum);
int32_t tsdbMemTableSaveToCache(SMemTable *pMemTable, void *func);
// STbData
int32_t tsdbGetNRowsInTbData(STbData *pTbData);
@ -335,7 +336,6 @@ struct STsdbFS {
typedef struct {
rocksdb_t *db;
rocksdb_comparator_t *my_comparator;
rocksdb_cache_t *blockcache;
rocksdb_block_based_table_options_t *tableoptions;
rocksdb_options_t *options;
rocksdb_flushoptions_t *flushoptions;
@ -347,6 +347,7 @@ typedef struct {
tb_uid_t suid;
tb_uid_t uid;
STSchema *pTSchema;
SArray *ctxArray;
} SRocksCache;
typedef struct {

1039
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

File diff suppressed because it is too large Load Diff

25
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -325,6 +325,27 @@ void tsdbMemTableCountRows(SMemTable *pMemTable, SSHashObj *pTableMap, int64_t *
taosRUnLockLatch(&pMemTable->latch);
}
typedef int32_t (*__tsdb_cache_update)(SMemTable *imem, int64_t suid, int64_t uid);
int32_t tsdbMemTableSaveToCache(SMemTable *pMemTable, void *func) {
int32_t code = 0;
__tsdb_cache_update cb = (__tsdb_cache_update)func;
for (int32_t i = 0; i < pMemTable->nBucket; ++i) {
STbData *pTbData = pMemTable->aBucket[i];
while (pTbData) {
code = (*cb)(pMemTable, pTbData->suid, pTbData->uid);
if (code) {
TAOS_RETURN(code);
}
pTbData = pTbData->next;
}
}
return code;
}
static int32_t tsdbMemTableRehash(SMemTable *pMemTable) {
int32_t code = 0;
@ -659,7 +680,7 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData,
pTbData->maxKey = key.key.ts;
}
if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config)) {
if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config) && !tsUpdateCacheBatch) {
if (tsdbCacheColFormatUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, pBlockData) != 0) {
tsdbError("vgId:%d, failed to update cache data from table suid:%" PRId64 " uid:%" PRId64 " at version %" PRId64,
TD_VID(pMemTable->pTsdb->pVnode), pTbData->suid, pTbData->uid, version);
@ -721,7 +742,7 @@ static int32_t tsdbInsertRowDataToTable(SMemTable *pMemTable, STbData *pTbData,
if (key.key.ts >= pTbData->maxKey) {
pTbData->maxKey = key.key.ts;
}
if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config)) {
if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config) && !tsUpdateCacheBatch) {
TAOS_UNUSED(tsdbCacheRowFormatUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, version, nRow, aRow));
}

91
tests/army/output.txt
View File

@ -1,91 +0,0 @@
[10/28 19:12:21.666563] SUCC: created database (db_sub)
[10/28 19:12:21.694603] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 19:12:21.823202] SUCC: Spent 0.1290 seconds to create 1000 table(s) with 8 thread(s) speed: 7752 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 19:12:22.127442] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 441047.79 records/second
[10/28 19:12:22.128649] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 440895.33 records/second
[10/28 19:12:22.129478] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 440151.69 records/second
[10/28 19:12:22.133756] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 433268.05 records/second
[10/28 19:12:22.135211] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 430329.63 records/second
[10/28 19:12:22.137335] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 425800.08 records/second
[10/28 19:12:22.138252] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 426330.15 records/second
[10/28 19:12:22.141351] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 422778.64 records/second
[10/28 19:12:22.141585] SUCC: Spent 0.311648 (real 0.289041) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 3208748.33 (real 3459716.79) records/second
[10/28 19:12:22.141590] SUCC: insert delay, min: 0.9600ms, avg: 2.3123ms, p90: 3.1790ms, p95: 3.5080ms, p99: 4.2230ms, max: 4.9040ms
[10/28 19:28:50.798427] SUCC: created database (db_sub)
[10/28 19:28:50.828326] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 19:28:50.936429] SUCC: Spent 0.1080 seconds to create 1000 table(s) with 8 thread(s) speed: 9259 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 19:28:51.187235] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 539204.48 records/second
[10/28 19:28:51.189941] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 532329.43 records/second
[10/28 19:28:51.191551] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 530954.66 records/second
[10/28 19:28:51.191858] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 529259.59 records/second
[10/28 19:28:51.192459] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 530229.44 records/second
[10/28 19:28:51.195372] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 522099.42 records/second
[10/28 19:28:51.197727] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 516620.72 records/second
[10/28 19:28:51.197883] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 517125.12 records/second
[10/28 19:28:51.198123] SUCC: Spent 0.255536 (real 0.237135) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 3913342.93 (real 4217007.19) records/second
[10/28 19:28:51.198130] SUCC: insert delay, min: 0.9200ms, avg: 1.8971ms, p90: 2.6870ms, p95: 2.9520ms, p99: 3.5880ms, max: 4.0710ms
[10/28 19:31:44.377691] SUCC: created database (db_sub)
[10/28 19:31:44.392998] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 19:31:44.696768] SUCC: Spent 0.3040 seconds to create 1000 table(s) with 8 thread(s) speed: 3289 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 19:31:45.126910] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 304775.47 records/second
[10/28 19:31:45.131979] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 301117.75 records/second
[10/28 19:31:45.135106] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 299854.39 records/second
[10/28 19:31:45.135675] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 298322.24 records/second
[10/28 19:31:45.137069] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 297733.89 records/second
[10/28 19:31:45.137952] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 296900.13 records/second
[10/28 19:31:45.138834] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 295170.54 records/second
[10/28 19:31:45.145048] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 291966.71 records/second
[10/28 19:31:45.145369] SUCC: Spent 0.442506 (real 0.419200) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 2259856.36 (real 2385496.18) records/second
[10/28 19:31:45.145377] SUCC: insert delay, min: 1.0400ms, avg: 3.3536ms, p90: 5.3120ms, p95: 7.9660ms, p99: 13.1570ms, max: 19.1410ms
[10/28 19:44:19.873056] SUCC: created database (db_sub)
[10/28 19:44:19.904701] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 19:44:20.053846] SUCC: Spent 0.1490 seconds to create 1000 table(s) with 8 thread(s) speed: 6711 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 19:44:20.328698] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 485742.49 records/second
[10/28 19:44:20.330777] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 481686.29 records/second
[10/28 19:44:20.331290] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 480911.65 records/second
[10/28 19:44:20.331665] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 481043.06 records/second
[10/28 19:44:20.333451] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 477172.09 records/second
[10/28 19:44:20.334745] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 475675.84 records/second
[10/28 19:44:20.335056] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 474158.37 records/second
[10/28 19:44:20.337919] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 470816.89 records/second
[10/28 19:44:20.338144] SUCC: Spent 0.277921 (real 0.261310) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 3598144.80 (real 3826872.30) records/second
[10/28 19:44:20.338153] SUCC: insert delay, min: 0.9180ms, avg: 2.0905ms, p90: 2.6490ms, p95: 3.0620ms, p99: 4.1480ms, max: 4.7840ms
[10/28 19:58:27.100989] SUCC: created database (db_sub)
[10/28 19:58:27.115572] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 19:58:27.362948] SUCC: Spent 0.2470 seconds to create 1000 table(s) with 8 thread(s) speed: 4049 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 19:58:27.807669] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 291891.03 records/second
[10/28 19:58:27.818785] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 285413.54 records/second
[10/28 19:58:27.819649] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 284193.61 records/second
[10/28 19:58:27.819844] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 284352.64 records/second
[10/28 19:58:27.820170] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 284576.63 records/second
[10/28 19:58:27.821489] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 283781.33 records/second
[10/28 19:58:27.822061] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 283112.24 records/second
[10/28 19:58:27.823513] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 282730.59 records/second
[10/28 19:58:27.823779] SUCC: Spent 0.455783 (real 0.438625) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 2194026.54 (real 2279851.81) records/second
[10/28 19:58:27.823786] SUCC: insert delay, min: 0.9780ms, avg: 3.5090ms, p90: 5.5650ms, p95: 6.8600ms, p99: 10.6010ms, max: 13.4400ms
[10/28 20:00:06.417182] SUCC: created database (db_sub)
[10/28 20:00:06.448202] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 20:00:06.596961] SUCC: Spent 0.1480 seconds to create 1000 table(s) with 8 thread(s) speed: 6757 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 20:00:06.895455] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 443978.76 records/second
[10/28 20:00:06.896986] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 442549.94 records/second
[10/28 20:00:06.897536] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 440927.99 records/second
[10/28 20:00:06.898905] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 439131.15 records/second
[10/28 20:00:06.899024] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 439628.46 records/second
[10/28 20:00:06.901861] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 435197.37 records/second
[10/28 20:00:06.902305] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 434812.86 records/second
[10/28 20:00:06.904698] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 433406.26 records/second
[10/28 20:00:06.904905] SUCC: Spent 0.301788 (real 0.284949) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 3313584.37 (real 3509399.93) records/second
[10/28 20:00:06.904912] SUCC: insert delay, min: 0.8770ms, avg: 2.2796ms, p90: 3.1340ms, p95: 3.6480ms, p99: 4.8280ms, max: 6.0880ms
[10/28 20:05:34.756207] SUCC: created database (db_sub)
[10/28 20:05:34.784793] INFO: start creating 1000 table(s) with 8 thread(s)
[10/28 20:05:34.927068] SUCC: Spent 0.1430 seconds to create 1000 table(s) with 8 thread(s) speed: 6993 tables/s, already exist 0 table(s), actual 1000 table(s) pre created, 0 table(s) will be auto created
[10/28 20:05:35.213741] SUCC: thread[4] progressive mode, completed total inserted rows: 125000, 466952.82 records/second
[10/28 20:05:35.215403] SUCC: thread[3] progressive mode, completed total inserted rows: 125000, 463804.68 records/second
[10/28 20:05:35.221132] SUCC: thread[2] progressive mode, completed total inserted rows: 125000, 453322.31 records/second
[10/28 20:05:35.221224] SUCC: thread[1] progressive mode, completed total inserted rows: 125000, 453671.11 records/second
[10/28 20:05:35.222003] SUCC: thread[0] progressive mode, completed total inserted rows: 125000, 452641.07 records/second
[10/28 20:05:35.222536] SUCC: thread[5] progressive mode, completed total inserted rows: 125000, 451796.89 records/second
[10/28 20:05:35.223663] SUCC: thread[7] progressive mode, completed total inserted rows: 125000, 449643.52 records/second
[10/28 20:05:35.225246] SUCC: thread[6] progressive mode, completed total inserted rows: 125000, 447768.68 records/second
[10/28 20:05:35.225659] SUCC: Spent 0.290871 (real 0.274808) seconds to insert rows: 1000000 with 8 thread(s) into db_sub 3437950.16 (real 3638904.25) records/second
[10/28 20:05:35.225666] SUCC: insert delay, min: 0.9360ms, avg: 2.1985ms, p90: 2.9290ms, p95: 3.4580ms, p99: 4.6030ms, max: 6.2660ms

212
tests/script/api/last-query-ws.cpp Normal file
View File

@ -0,0 +1,212 @@
// g++ --std=c++17 -o multiQueryLastrow multiQueryLastrow.cpp -ltaos -lpthread -ltaosws
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
#include <unordered_map>
#include "taos.h"
#include "taosws.h"
int numThreads = 5;
int numQuerys = 100;
int queryType = 0;
int numConnections = 1;
bool useWebSocket = 0;
using namespace std;
const std::string dbName = "iot";
const std::string sTableName = "m";
int maxTableIndex = 50000;
std::mutex mtx;
std::condition_variable cv;
vector<TAOS*> taosArray;
vector<WS_TAOS*> wtaosArray;
std::atomic<int> finishCounter;
std::chrono::system_clock::time_point startTime;
std::chrono::system_clock::time_point stopTime;
unordered_map<int, chrono::nanoseconds> consumeHash;
static void query(int numQuerys, int id, int type);
void threadFunction(int id) {
// std::unique_lock<std::mutex> lock(mtx);
// cv.wait(lock);
// lock.unlock();
//auto startQueryTime = std::chrono::system_clock::now();
query(numQuerys, id, queryType);
//consumeHash[id] = std::chrono::system_clock::now() - startQueryTime;
// int counter = finishCounter.fetch_add(1);
// if (counter == numThreads - 1) {
// stopTime = std::chrono::system_clock::now();
// }
}
void createThreads(const int numThreads, std::vector<std::thread>* pThreads) {
for (int i = 0; i < numThreads; ++i) {
pThreads->emplace_back(threadFunction, i);
}
std::cout << "2. Threads created\n";
}
void connect() {
void* res = NULL;
for (auto i = 0; i < numConnections; i++) {
if (useWebSocket) {
const char* dsn = "taos+ws://localhost:6041";
WS_TAOS* wtaos = ws_connect(dsn);
int32_t code = 0;
if (wtaos == NULL) {
code = ws_errno(NULL);
const char* errstr = ws_errstr(NULL);
std::cout << "Connection failed[" << code << "]: " << errstr << "\n";
return;
}
code = ws_select_db(wtaos, dbName.c_str());
const char* errstr = ws_errstr(wtaos);
if (code) {
std::cout << "Connection failed on select db[" << code << "]: " << errstr << "\n";
return;
}
wtaosArray.push_back(wtaos);
} else {
TAOS* taos = taos_connect("127.0.0.1", "root", "taosdata", dbName.c_str(), 0);
if (!taos) {
std::cerr << "Failed to connect to TDengine\n";
return;
}
taosArray.push_back(taos);
}
}
std::cout << "1. Success to connect to TDengine\n";
}
void query(int numQuerys, int id, int type) {
int connIdx = id % numConnections;
for (int i = 0; i < numQuerys; i++) {
std::string sql;
if (type == 0) {
sql = "select last_row(ts) from " + sTableName + std::to_string((i * numThreads + id) % maxTableIndex);
} else {
sql = "select first(ts) from " + sTableName + std::to_string((i * numThreads + id) % maxTableIndex);
}
if (!useWebSocket) {
TAOS* taos = taosArray[connIdx];
TAOS_RES* res = taos_query(taos, sql.c_str());
if (!res) {
std::cerr << "Failed to query TDengine\n";
return;
}
if (taos_errno(res) != 0) {
std::cerr << "Failed to query TDengine since: " << taos_errstr(res) << "\n";
return;
}
taos_free_result(res);
} else {
WS_TAOS* wtaos = wtaosArray[connIdx];
WS_RES* wres = ws_query(wtaos, sql.c_str());
if (!wres) {
std::cerr << "Failed to query TDengine\n";
return;
}
int32_t code = ws_errno(wres);
if (code != 0) {
std::cerr << "Failed to query TDengine since: " << ws_errstr(wres) << "\n";
return;
}
ws_free_result(wres);
}
}
}
void printHelp() {
std::cout << "./multiQueryLastrow {numThreads} {numQuerys} {queryType} {numConnections} {useWebSocket}\n";
exit(-1);
}
int main(int argc, char* argv[]) {
if (argc != 6) {
printHelp();
}
numThreads = atoi(argv[1]);
numQuerys = atoi(argv[2]);
queryType = atoi(argv[3]);
numConnections = atoi(argv[4]);
useWebSocket = atoi(argv[5]);
std::string queryTypeStr = (queryType == 0) ? "last_row(ts)" : "first(ts)";
std::cout << "numThreads:" << numThreads << ", queryTimes:" << numQuerys << ", queryType:" << queryTypeStr
<< ", numConnections:" << numConnections << ", useWebSocket:" << useWebSocket << "\n";
finishCounter.store(0);
connect();
//startTime = std::chrono::system_clock::now();
std::vector<std::thread> threads;
createThreads(numThreads, &threads);
//std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "3. Start quering\n";
startTime = std::chrono::system_clock::now();
//cv.notify_all();
for (auto& t : threads) {
t.join();
}
stopTime = std::chrono::system_clock::now();
for (auto& taos : taosArray) {
taos_close(taos);
}
for (auto& wtaos : wtaosArray) {
ws_close(wtaos);
}
std::cout << "4. All job done\n";
int64_t totalQueryConsumeMs = 0;
for (auto& res : consumeHash) {
totalQueryConsumeMs += res.second.count() /1000000;
}
std::chrono::nanoseconds elp = stopTime - startTime;
int64_t elpMs = elp.count() / 1000000;
int64_t totalQueryCount = numThreads * numQuerys;
std::cout << totalQueryCount << " queries finished in " << elpMs << " ms\n";
std::cout << (float)totalQueryCount * 1000 / elpMs << "q/s\n";
std::cout << "avg cost:" << totalQueryConsumeMs / totalQueryCount << " ms/q\n";
return 0;
}