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

fix invalid read/write error

This commit is contained in:
yihaoDeng 2023-04-03 03:32:01 +00:00
parent 7f4d0a17e6
commit 6ee0180217
1 changed files with 59 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
source/libs/stream/src/streamStateRocksdb.c
View File

@ -296,8 +296,13 @@ const char* comparePartagKey(void* name) { return cfName[5]; }
void destroyFunc(void* stata) { return; } void destroyFunc(void* stata) { return; }
int streamInitBackend(SStreamState* pState, char* path) { int streamInitBackend(SStreamState* pState, char* path) {
rocksdb_env_t* env = rocksdb_create_default_env(); // rocksdb_envoptions_create();
rocksdb_env_set_low_priority_background_threads(env, 15);
rocksdb_env_set_high_priority_background_threads(env, 5);
rocksdb_options_t* opts = rocksdb_options_create(); rocksdb_options_t* opts = rocksdb_options_create();
rocksdb_options_increase_parallelism(opts, 8); rocksdb_options_set_env(opts, env);
// rocksdb_options_increase_parallelism(opts, 8);
// rocksdb_options_optimize_level_style_compaction(opts, 0); // rocksdb_options_optimize_level_style_compaction(opts, 0);
// create the DB if it's not already present // create the DB if it's not already present
rocksdb_options_set_create_if_missing(opts, 1); rocksdb_options_set_create_if_missing(opts, 1);
@ -576,66 +581,66 @@ int32_t streamStateFillDel_rocksdb(SStreamState* pState, const SWinKey* key) {
int32_t streamStateClear_rocksdb(SStreamState* pState) { int32_t streamStateClear_rocksdb(SStreamState* pState) {
qDebug("streamStateClear_rocksdb"); qDebug("streamStateClear_rocksdb");
// SStateKey sKey = {.key = {.ts = 0, .groupId = 0}, .opNum = pState->number}; SStateKey sKey = {.key = {.ts = 0, .groupId = 0}, .opNum = pState->number};
// SStateKey eKey = {.key = {.ts = INT64_MAX, .groupId = UINT64_MAX}, .opNum = pState->number}; SStateKey eKey = {.key = {.ts = INT64_MAX, .groupId = UINT64_MAX}, .opNum = pState->number};
// char sKeyStr[128] = {0}; char sKeyStr[128] = {0};
// char eKeyStr[128] = {0}; char eKeyStr[128] = {0};
// int sLen = stateKeyEncode(&sKey, sKeyStr); int sLen = stateKeyEncode(&sKey, sKeyStr);
// int eLen = stateKeyEncode(&eKey, eKeyStr); int eLen = stateKeyEncode(&eKey, eKeyStr);
// char toStringStart[128] = {0}; char toStringStart[128] = {0};
// char toStringEnd[128] = {0}; char toStringEnd[128] = {0};
// if (qDebugFlag & DEBUG_TRACE) { if (qDebugFlag & DEBUG_TRACE) {
// stateKeyToString(&sKey, toStringStart); stateKeyToString(&sKey, toStringStart);
// stateKeyToString(&eKey, toStringEnd); stateKeyToString(&eKey, toStringEnd);
// } }
// char* err = NULL; char* err = NULL;
// rocksdb_delete_range_cf(pState->pTdbState->rocksdb, pState->pTdbState->writeOpts, pState->pTdbState->pHandle[0], rocksdb_delete_range_cf(pState->pTdbState->rocksdb, pState->pTdbState->writeOpts, pState->pTdbState->pHandle[0],
// sKeyStr, sLen, eKeyStr, eLen, &err); sKeyStr, sLen, eKeyStr, eLen, &err);
// // rocksdb_compact_range_cf(pState->pTdbState->rocksdb, pState->pTdbState->pHandle[0], sKeyStr, sLen, eKeyStr, // rocksdb_compact_range_cf(pState->pTdbState->rocksdb, pState->pTdbState->pHandle[0], sKeyStr, sLen, eKeyStr, eLen);
// eLen); if (err != NULL) { if (err != NULL) {
// qWarn("failed to delete range cf(default) err: %s, start: %s, end:%s", err, toStringStart, toStringEnd); qWarn("failed to delete range cf(default) err: %s, start: %s, end:%s", err, toStringStart, toStringEnd);
// taosMemoryFree(err); taosMemoryFree(err);
// } }
// del one by one // del one by one
char buf[128] = {0}; // char buf[128] = {0};
char* s = "null"; // char* s = "null";
SWinKey key = {.ts = 0, .groupId = 0}; // SWinKey key = {.ts = 0, .groupId = 0};
SStateKey skey = {.key = key, .opNum = pState->number}; // SStateKey skey = {.key = key, .opNum = pState->number};
int sLen = stateKeyEncode(&skey, buf); // int sLen = stateKeyEncode(&skey, buf);
streamStatePut_rocksdb(pState, &key, s, strlen(s)); // streamStatePut_rocksdb(pState, &key, s, strlen(s));
rocksdb_readoptions_t* opt = NULL; // rocksdb_readoptions_t* opt = NULL;
rocksdb_iterator_t* iter = streamStateIterCreate(pState, "default", NULL, &opt); // rocksdb_iterator_t* iter = streamStateIterCreate(pState, "default", NULL, &opt);
rocksdb_iter_seek(iter, buf, sLen); // rocksdb_iter_seek(iter, buf, sLen);
char* err = NULL; // char* err = NULL;
while (rocksdb_iter_valid(iter)) { // while (rocksdb_iter_valid(iter)) {
int kLen = 0; // int32_t kLen = 0;
char* key = (char*)rocksdb_iter_key(iter, (size_t*)&kLen); // char* key = (char*)rocksdb_iter_key(iter, (size_t*)&kLen);
SStateKey ckey = {0}; // SStateKey ckey = {0};
stateKeyDecode((void*)&ckey, key); // stateKeyDecode((void*)&ckey, key);
if (ckey.opNum != pState->number) { // if (ckey.opNum != pState->number) {
break; // break;
} // }
if (stateKeyCmpr(&skey, sizeof(skey), &ckey, sizeof(ckey)) > 0) { // if (stateKeyCmpr(&skey, sizeof(skey), &ckey, sizeof(ckey)) > 0) {
break; // break;
} // }
rocksdb_delete_cf(pState->pTdbState->rocksdb, pState->pTdbState->writeOpts, pState->pTdbState->pHandle[0], key, // rocksdb_delete_cf(pState->pTdbState->rocksdb, pState->pTdbState->writeOpts, pState->pTdbState->pHandle[0], key,
kLen, &err); // kLen, &err);
if (err != NULL) { // if (err != NULL) {
taosMemoryFree(err); // taosMemoryFree(err);
} // }
if (rocksdb_iter_valid(iter)) rocksdb_iter_next(iter); // if (rocksdb_iter_valid(iter)) rocksdb_iter_next(iter);
} // }
rocksdb_iter_destroy(iter); // rocksdb_iter_destroy(iter);
return 0; return 0;
} }
@ -898,7 +903,7 @@ int32_t streamStateSessionGetKVByCur_rocksdb(SStreamStateCur* pCur, SSessionKey*
return -1; return -1;
} }
SStateSessionKey ktmp = {0}; SStateSessionKey ktmp = {0};
int32_t kLen, vLen; size_t kLen = 0, vLen = 0;
if (!rocksdb_iter_valid(pCur->iter)) { if (!rocksdb_iter_valid(pCur->iter)) {
return -1; return -1;
@ -1050,7 +1055,7 @@ int32_t streamStateSessionGetKeyByRange_rocksdb(SStreamState* pState, const SSes
return -1; return -1;
} }
int32_t kLen; size_t kLen;
const char* iKeyStr = rocksdb_iter_key(pCur->iter, (size_t*)&kLen); const char* iKeyStr = rocksdb_iter_key(pCur->iter, (size_t*)&kLen);
SStateSessionKey iKey = {0}; SStateSessionKey iKey = {0};
stateSessionKeyDecode(&iKey, (char*)iKeyStr); stateSessionKeyDecode(&iKey, (char*)iKeyStr);