Many hyperlinks are disabled.
Use anonymous login
to enable hyperlinks.
Overview
| Comment: | Fix a deadlock in multi-threaded test code. |
|---|---|
| Downloads: | Tarball | ZIP archive | SQL archive |
| Timelines: | family | ancestors | descendants | both | trunk |
| Files: | files | file ages | folders |
| SHA1: |
e16b04ca69fa131b6a7a370e2afc7879 |
| User & Date: | dan 2012-09-27 04:59:26 |
Context
|
2012-09-27
| ||
| 16:09 | Fix to ensure that the log file is always deleted following a successful system shutdown. check-in: 3d1dacff87 user: dan tags: trunk | |
| 04:59 | Fix a deadlock in multi-threaded test code. check-in: e16b04ca69 user: dan tags: trunk | |
|
2012-09-26
| ||
| 15:38 | Merge rework-flow-control branch with trunk. check-in: cf2ef747ad user: dan tags: trunk | |
Changes
Changes to lsm-test/lsmtest_tdb3.c.
928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 .... 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 .... 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 .... 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 |
static void *worker_main(void *pArg){
LsmWorker *p = (LsmWorker *)pArg;
lsm_db *pWorker; /* Connection to access db through */
pthread_mutex_lock(&p->worker_mutex);
while( (pWorker = p->pWorker) ){
int nWrite = 0;
int rc = LSM_OK;
/* Do some work. If an error occurs, exit. */
pthread_mutex_unlock(&p->worker_mutex);
if( p->bCkpt==0 ){
static const int nLimit = 16*1024*1024;
static const int nIncr = 4*1024*1024;
int nMax = 100;
int nByte = 0;
lsm_ckpt_size(pWorker, &nByte);
if( nByte>nLimit ){
int nSleep = 0;
while( nByte>nLimit ){
nMax = nMax<<1;
nByte -= nIncr;
}
while( nSleep<nMax ){
lsm_ckpt_size(pWorker, &nByte);
if( nByte<nLimit ) break;
mt_signal_worker(p->pDb, 1);
usleep(1000);
nSleep++;
}
#if 0
if( nSleep ) printf("nSleep=%d/%d (worker)\n", nSleep, nMax);
#endif
}
}
if( p->lsm_work_npage ){
rc = lsm_work(pWorker, p->lsm_work_flags, p->lsm_work_npage, &nWrite);
/* printf("# worked %d units\n", nWrite); */
}
if( rc==LSM_OK && p->bCkpt ){
rc = lsm_checkpoint(pWorker, 0);
}
pthread_mutex_lock(&p->worker_mutex);
if( rc!=LSM_OK && rc!=LSM_BUSY ){
p->worker_rc = rc;
break;
}
if( nWrite && p->bCkpt==0 ){
mt_signal_worker(p->pDb, 1);
}
/* If the call to lsm_work() indicates that there is nothing more
** to do at this point, wait on the condition variable. The thread will
** wake up when it is signaled either because the client thread has
** flushed an in-memory tree into the db file or when the connection
** is being closed. */
if( nWrite==0 ){
if( p->pWorker && p->bDoWork==0 ){
pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
}
p->bDoWork = 0;
}
}
pthread_mutex_unlock(&p->worker_mutex);
return 0;
}
................................................................................
static void mt_client_work_hook(lsm_db *db, void *pArg){
LsmDb *pDb = (LsmDb *)pArg; /* LsmDb database handle */
int i; /* Iterator variable */
/* Invoke the user level work-hook, if any. */
if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
/* printf("# signalling worker threads\n"); */
/* Signal each worker thread */
for(i=0; i<pDb->nWorker; i++){
mt_signal_worker(pDb, i);
}
}
static void mt_worker_work_hook(lsm_db *db, void *pArg){
LsmDb *pDb = (LsmDb *)pArg; /* LsmDb database handle */
/* Invoke the user level work-hook, if any. */
if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
................................................................................
/* Open the worker connection */
if( rc==0 ) rc = lsm_new(&pDb->env, &p->pWorker);
if( zCfg ){
test_lsm_config_str(pDb, p->pWorker, 1, zCfg, 0);
}
if( rc==0 ) rc = lsm_open(p->pWorker, zFilename);
#if 0
lsm_config_log(p->pWorker, xLog, (void *)"worker");
#endif
/* Configure the work-hook */
if( rc==0 ){
lsm_config_work_hook(p->pWorker, mt_worker_work_hook, (void *)pDb);
}
/* Kick off the worker thread. */
................................................................................
pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * nWorker);
memset(pDb->aWorker, 0, sizeof(LsmWorker) * nWorker);
pDb->nWorker = nWorker;
if( nWorker==1 ){
int flags = LSM_WORK_FLUSH;
rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 2048, 1);
}else{
int flags = LSM_WORK_FLUSH;
rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 1024, 0);
if( rc==LSM_OK ){
rc = mt_start_worker(pDb, 1, zFilename, zCfg, 0, 0, 1);
}
}
return rc;
}
|
< > > | < < < < > | < < < < < > > | > > > > > | | > | < > | | < < | < < < < > > < < < < < | | | | < | < < | < < | < | | |
928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 .... 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 .... 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 .... 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 |
static void *worker_main(void *pArg){
LsmWorker *p = (LsmWorker *)pArg;
lsm_db *pWorker; /* Connection to access db through */
pthread_mutex_lock(&p->worker_mutex);
while( (pWorker = p->pWorker) ){
int rc = LSM_OK;
int nLimit = -1;
int nCkpt = -1;
/* Do some work. If an error occurs, exit. */
pthread_mutex_unlock(&p->worker_mutex);
lsm_config(pWorker, LSM_CONFIG_WRITE_BUFFER, &nLimit);
if( p->bCkpt ){
lsm_ckpt_size(pWorker, &nCkpt);
if( nCkpt>(nLimit*2) ){
rc = lsm_checkpoint(pWorker, 0);
}
}else{
int nWrite = 0;
int nAuto = -1;
lsm_config(pWorker, LSM_CONFIG_AUTOCHECKPOINT, &nAuto);
do {
int nSleep = 0;
lsm_ckpt_size(pWorker, &nCkpt);
while( nAuto==0 && nCkpt>(nLimit*4) ){
usleep(1000);
mt_signal_worker(p->pDb, 1);
nSleep++;
lsm_ckpt_size(pWorker, &nCkpt);
}
#if 0
if( nSleep ) printf("nLimit=%d nSleep=%d (worker)\n", nLimit, nSleep);
#endif
rc = lsm_work(pWorker, p->lsm_work_flags, p->lsm_work_npage, &nWrite);
if( nAuto==0 && nWrite && rc==LSM_OK ) mt_signal_worker(p->pDb, 1);
}while( nWrite );
}
pthread_mutex_lock(&p->worker_mutex);
if( rc!=LSM_OK && rc!=LSM_BUSY ){
p->worker_rc = rc;
break;
}
/* If the call to lsm_work() indicates that there is nothing more
** to do at this point, wait on the condition variable. The thread will
** wake up when it is signaled either because the client thread has
** flushed an in-memory tree into the db file or when the connection
** is being closed. */
if( p->pWorker && p->bDoWork==0 ){
pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
}
p->bDoWork = 0;
}
pthread_mutex_unlock(&p->worker_mutex);
return 0;
}
................................................................................
static void mt_client_work_hook(lsm_db *db, void *pArg){
LsmDb *pDb = (LsmDb *)pArg; /* LsmDb database handle */
int i; /* Iterator variable */
/* Invoke the user level work-hook, if any. */
if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
/* Signal the lsm_work() thread */
mt_signal_worker(pDb, 0);
}
static void mt_worker_work_hook(lsm_db *db, void *pArg){
LsmDb *pDb = (LsmDb *)pArg; /* LsmDb database handle */
/* Invoke the user level work-hook, if any. */
if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
................................................................................
/* Open the worker connection */
if( rc==0 ) rc = lsm_new(&pDb->env, &p->pWorker);
if( zCfg ){
test_lsm_config_str(pDb, p->pWorker, 1, zCfg, 0);
}
if( rc==0 ) rc = lsm_open(p->pWorker, zFilename);
lsm_config_log(p->pWorker, xLog, (void *)"worker");
/* Configure the work-hook */
if( rc==0 ){
lsm_config_work_hook(p->pWorker, mt_worker_work_hook, (void *)pDb);
}
/* Kick off the worker thread. */
................................................................................
pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * nWorker);
memset(pDb->aWorker, 0, sizeof(LsmWorker) * nWorker);
pDb->nWorker = nWorker;
if( nWorker==1 ){
int flags = LSM_WORK_FLUSH;
rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 256, 0);
}else{
int flags = LSM_WORK_FLUSH;
rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 256, 0);
if( rc==LSM_OK ){
rc = mt_start_worker(pDb, 1, zFilename, zCfg, 0, 0, 1);
}
}
return rc;
}
|
Changes to src/lsm_shared.c.
468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 |
}
if( rc==LSM_OK && bInUse==0 ){
iRet = pFree->aEntry[0].iBlk;
flRemoveEntry0(pFree);
assert( iRet!=0 );
}
#if 0
lsmLogMessage(
pDb, 0, "%d reusing block %d%s", (iRet==0 ? "not " : "")
pFree->aEntry[0].iBlk,
bInUse==0 ? "" : bInUse==1 ? " (client)" : " (unsynced)"
);
#endif
}
/* If no block was allocated from the free-list, allocate one at the
** end of the file. */
if( rc==LSM_OK && iRet==0 ){
iRet = ++pDb->pWorker->nBlock;
}
*piBlk = iRet;
return LSM_OK;
}
/*
|
| | > > > |
468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 |
}
if( rc==LSM_OK && bInUse==0 ){
iRet = pFree->aEntry[0].iBlk;
flRemoveEntry0(pFree);
assert( iRet!=0 );
}
#ifdef LSM_LOG_BLOCKS
lsmLogMessage(
pDb, 0, "%s reusing block %d%s", (iRet==0 ? "not " : ""),
pFree->aEntry[0].iBlk,
bInUse==0 ? "" : bInUse==1 ? " (client)" : " (unsynced)"
);
#endif
}
/* If no block was allocated from the free-list, allocate one at the
** end of the file. */
if( rc==LSM_OK && iRet==0 ){
iRet = ++pDb->pWorker->nBlock;
#ifdef LSM_LOG_BLOCKS
lsmLogMessage(pDb, 0, "extending file to %d blocks", iRet);
#endif
}
*piBlk = iRet;
return LSM_OK;
}
/*
|
Changes to tool/lsmperf.tcl.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
...
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
#!/bin/sh
# \
exec tclsh "$0" "$@"
proc exec_lsmtest_speed {nSec spec} {
set fd [open [list |lsmtest speed2 {*}$spec]]
set res [list]
puts "lsmtest speed2 $spec"
set initial [clock seconds]
while {![eof $fd] && ($nSec==0 || ([clock second]-$initial)<$nSec)} {
set line [gets $fd]
puts $line
if {[string range $line 0 0]=="#"} continue
if {[llength $line]==0} continue
lappend res $line
}
catch { close $fd }
set res
}
................................................................................
append script $data2
append script $data3
append script "pause -1\n"
exec_gnuplot_script $script $zPng
}
do_write_test x.png 60 25000 0 40 {
lsm-mt "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
LevelDB leveldb
}
# lsm-mt "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# lsm-st "mmap=1 multi_proc=0 safety=1 threads=1 autowork=1"
# LevelDB leveldb
# SQLite sqlite3
|
|
|
>
>
>
>
>
>
>
>
|
>
>
|
|
|
>
>
>
>
>
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
...
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
#!/bin/sh
# \
exec tclsh "$0" "$@"
proc exec_lsmtest_speed {nSec spec} {
set fd [open [list |./lsmtest speed2 {*}$spec]]
set res [list]
puts -nonewline "./lsmtest speed2"
foreach s $spec {
if {[llength $s]==1} {
puts -nonewline " $s"
} else {
puts -nonewline " \"$s\""
}
}
puts ""
set initial [clock seconds]
while {![eof $fd] && ($nSec==0 || ([clock second]-$initial)<$nSec)} {
set line [gets $fd]
set sz 0
catch { set sz [file size testdb.lsm] }
puts "$line ([expr $sz/1024])"
if {[string range $line 0 0]=="#"} continue
if {[llength $line]==0} continue
lappend res $line
}
catch { close $fd }
set res
}
................................................................................
append script $data2
append script $data3
append script "pause -1\n"
exec_gnuplot_script $script $zPng
}
do_write_test x.png 120 50000 50000 30 {
lsm-mt "mmap=1 multi_proc=0 threads=3 autowork=0 autocheckpoint=0"
leveldb leveldb
}
# lsm-mt "mmap=1 multi_proc=0 threads=2 autowork=0 autocheckpoint=8192000"
# lsm-mt "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# lsm-st "mmap=1 multi_proc=0 safety=1 threads=1 autowork=1"
# lsm-mt "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# lsm-mt "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# LevelDB leveldb
# lsm-st "mmap=1 multi_proc=0 safety=1 threads=1 autowork=1"
# LevelDB leveldb
# SQLite sqlite3
|