NextGraph
/
lmdb
mirror of https://github.com/nextgraph-org/lmdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

More tweaks to copyfd2

Browse Source 
Make sure the writer thread starts and stops when we expect it to.
incre
Howard Chu 11 years ago
parent 03f0ecb001
commit 059b357d1a
1 changed files with 51 additions and 31 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 82
      libraries/liblmdb/mdb.c

82
libraries/liblmdb/mdb.c
Unescape View File

@ -177,6 +177,7 @@
#define THREAD_RET DWORD #define THREAD_RET DWORD
#define pthread_t HANDLE #define pthread_t HANDLE
#define pthread_mutex_t HANDLE #define pthread_mutex_t HANDLE
#define pthread_cond_t HANDLE
#define pthread_key_t DWORD #define pthread_key_t DWORD
#define pthread_self() GetCurrentThreadId() #define pthread_self() GetCurrentThreadId()
#define pthread_key_create(x,y) \ #define pthread_key_create(x,y) \
@ -186,6 +187,8 @@
#define pthread_setspecific(x,y) (TlsSetValue(x,y) ? 0 : ErrCode()) #define pthread_setspecific(x,y) (TlsSetValue(x,y) ? 0 : ErrCode())
#define pthread_mutex_unlock(x) ReleaseMutex(x) #define pthread_mutex_unlock(x) ReleaseMutex(x)
#define pthread_mutex_lock(x) WaitForSingleObject(x, INFINITE) #define pthread_mutex_lock(x) WaitForSingleObject(x, INFINITE)
#define pthread_cond_signal(x) SetEvent(*x)
#define pthread_cond_wait(cond,mutex) SignalObjectAndWait(*mutex, *cond, INFINITE, FALSE); WaitForSingleObject(*mutex, INFINITE)
#define THREAD_CREATE(thr,start,arg) thr=CreateThread(NULL,0,start,arg,0,NULL) #define THREAD_CREATE(thr,start,arg) thr=CreateThread(NULL,0,start,arg,0,NULL)
#define THREAD_FINISH(thr) WaitForSingleObject(thr, INFINITE) #define THREAD_FINISH(thr) WaitForSingleObject(thr, INFINITE)
#define LOCK_MUTEX_R(env) pthread_mutex_lock((env)->me_rmutex) #define LOCK_MUTEX_R(env) pthread_mutex_lock((env)->me_rmutex)
@ -8034,7 +8037,8 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi,
#endif #endif
typedef struct mdb_copy { typedef struct mdb_copy {
pthread_mutex_t mc_mutex[2]; pthread_mutex_t mc_mutex;
pthread_cond_t mc_cond;
char *mc_wbuf[2]; char *mc_wbuf[2];
char *mc_over[2]; char *mc_over[2];
MDB_env *mc_env; MDB_env *mc_env;
@ -8044,6 +8048,7 @@ typedef struct mdb_copy {
pgno_t mc_next_pgno; pgno_t mc_next_pgno;
HANDLE mc_fd; HANDLE mc_fd;
int mc_status; int mc_status;
volatile int mc_new;
int mc_toggle; int mc_toggle;
} mdb_copy; } mdb_copy;
@ -8061,14 +8066,17 @@ mdb_env_copythr(void *arg)
#define DO_WRITE(rc, fd, ptr, w2, len) len = write(fd, ptr, w2); rc = (len >= 0) #define DO_WRITE(rc, fd, ptr, w2, len) len = write(fd, ptr, w2); rc = (len >= 0)
#endif #endif
pthread_mutex_lock(&my->mc_mutex[toggle^1]); pthread_mutex_lock(&my->mc_mutex);
my->mc_new = 0;
pthread_cond_signal(&my->mc_cond);
for(;;) { for(;;) {
pthread_mutex_lock(&my->mc_mutex[toggle]); while (!my->mc_new)
pthread_mutex_unlock(&my->mc_mutex[toggle^1]); pthread_cond_wait(&my->mc_cond, &my->mc_mutex);
if (!my->mc_wlen[toggle]) { if (my->mc_new < 0) {
pthread_mutex_unlock(&my->mc_mutex[toggle]); my->mc_new = 0;
break; break;
} }
my->mc_new = 0;
wsize = my->mc_wlen[toggle]; wsize = my->mc_wlen[toggle];
ptr = my->mc_wbuf[toggle]; ptr = my->mc_wbuf[toggle];
again: again:
@ -8089,7 +8097,6 @@ again:
} }
if (rc) { if (rc) {
my->mc_status = rc; my->mc_status = rc;
pthread_mutex_unlock(&my->mc_mutex[toggle]);
break; break;
} }
/* If there's an overflow page tail, write it too */ /* If there's an overflow page tail, write it too */
@ -8101,23 +8108,29 @@ again:
} }
my->mc_wlen[toggle] = 0; my->mc_wlen[toggle] = 0;
toggle ^= 1; toggle ^= 1;
pthread_cond_signal(&my->mc_cond);
} }
pthread_cond_signal(&my->mc_cond);
pthread_mutex_unlock(&my->mc_mutex);
return (THREAD_RET)0; return (THREAD_RET)0;
#undef DO_WRITE #undef DO_WRITE
} }
static int static int
mdb_env_cthr_toggle(mdb_copy *my) mdb_env_cthr_toggle(mdb_copy *my, int st)
{ {
int toggle = my->mc_toggle ^ 1; int toggle = my->mc_toggle ^ 1;
pthread_mutex_lock(&my->mc_mutex);
pthread_mutex_unlock(&my->mc_mutex[my->mc_toggle]);
pthread_mutex_lock(&my->mc_mutex[toggle]);
if (my->mc_status) { if (my->mc_status) {
pthread_mutex_unlock(&my->mc_mutex[toggle]); pthread_mutex_unlock(&my->mc_mutex);
return my->mc_status; return my->mc_status;
} }
while (my->mc_new == 1)
pthread_cond_wait(&my->mc_cond, &my->mc_mutex);
my->mc_new = st;
my->mc_toggle = toggle; my->mc_toggle = toggle;
pthread_cond_signal(&my->mc_cond);
pthread_mutex_unlock(&my->mc_mutex);
return 0; return 0;
} }
@ -8188,10 +8201,10 @@ mdb_env_cwalk(mdb_copy *my, pgno_t *pg, int flags)
if (rc) if (rc)
goto done; goto done;
if (my->mc_wlen[toggle] >= MDB_WBUF) { if (my->mc_wlen[toggle] >= MDB_WBUF) {
rc = mdb_env_cthr_toggle(my); rc = mdb_env_cthr_toggle(my, 1);
if (rc) if (rc)
goto done; goto done;
toggle ^= 1; toggle = my->mc_toggle;
} }
mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]); mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]);
memcpy(mo, omp, my->mc_env->me_psize); memcpy(mo, omp, my->mc_env->me_psize);
@ -8201,10 +8214,10 @@ mdb_env_cwalk(mdb_copy *my, pgno_t *pg, int flags)
if (omp->mp_pages > 1) { if (omp->mp_pages > 1) {
my->mc_olen[toggle] = my->mc_env->me_psize * (omp->mp_pages - 1); my->mc_olen[toggle] = my->mc_env->me_psize * (omp->mp_pages - 1);
my->mc_over[toggle] = (char *)omp + my->mc_env->me_psize; my->mc_over[toggle] = (char *)omp + my->mc_env->me_psize;
rc = mdb_env_cthr_toggle(my); rc = mdb_env_cthr_toggle(my, 1);
if (rc) if (rc)
goto done; goto done;
toggle ^= 1; toggle = my->mc_toggle;
} }
memcpy(NODEDATA(ni), &mo->mp_pgno, sizeof(pgno_t)); memcpy(NODEDATA(ni), &mo->mp_pgno, sizeof(pgno_t));
} else if (ni->mn_flags & F_SUBDATA) { } else if (ni->mn_flags & F_SUBDATA) {
@ -8250,10 +8263,10 @@ again:
} }
} }
if (my->mc_wlen[toggle] >= MDB_WBUF) { if (my->mc_wlen[toggle] >= MDB_WBUF) {
rc = mdb_env_cthr_toggle(my); rc = mdb_env_cthr_toggle(my, 1);
if (rc) if (rc)
goto done; goto done;
toggle ^= 1; toggle = my->mc_toggle;
} }
mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]); mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]);
mdb_page_copy(mo, mp, my->mc_env->me_psize); mdb_page_copy(mo, mp, my->mc_env->me_psize);
@ -8286,14 +8299,14 @@ mdb_env_copyfd2(MDB_env *env, HANDLE fd)
int rc; int rc;
#ifdef _WIN32 #ifdef _WIN32
my.mc_mutex[0] = CreateMutex(NULL, FALSE, NULL); my.mc_mutex = CreateMutex(NULL, FALSE, NULL);
my.mc_mutex[1] = CreateMutex(NULL, FALSE, NULL); my.mc_cond = CreateEvent(NULL, FALSE, FALSE, NULL);
my.mc_wbuf[0] = _aligned_malloc(MDB_WBUF*2, env->me_psize); my.mc_wbuf[0] = _aligned_malloc(MDB_WBUF*2, env->me_psize);
if (my.mc_wbuf[0] == NULL) if (my.mc_wbuf[0] == NULL)
return errno; return errno;
#else #else
pthread_mutex_init(&my.mc_mutex[0], NULL); pthread_mutex_init(&my.mc_mutex, NULL);
pthread_mutex_init(&my.mc_mutex[1], NULL); pthread_cond_init(&my.mc_cond, NULL);
rc = posix_memalign((void **)&my.mc_wbuf[0], env->me_psize, MDB_WBUF*2); rc = posix_memalign((void **)&my.mc_wbuf[0], env->me_psize, MDB_WBUF*2);
if (rc) if (rc)
return rc; return rc;
@ -8305,11 +8318,10 @@ mdb_env_copyfd2(MDB_env *env, HANDLE fd)
my.mc_olen[1] = 0; my.mc_olen[1] = 0;
my.mc_next_pgno = 2; my.mc_next_pgno = 2;
my.mc_status = 0; my.mc_status = 0;
my.mc_new = 1;
my.mc_toggle = 0; my.mc_toggle = 0;
my.mc_env = env; my.mc_env = env;
my.mc_fd = fd; my.mc_fd = fd;
pthread_mutex_lock(&my.mc_mutex[0]);
THREAD_CREATE(thr, mdb_env_copythr, &my);
/* Do the lock/unlock of the reader mutex before starting the /* Do the lock/unlock of the reader mutex before starting the
* write txn. Otherwise other read txns could block writers. * write txn. Otherwise other read txns could block writers.
@ -8332,6 +8344,7 @@ mdb_env_copyfd2(MDB_env *env, HANDLE fd)
} }
} }
THREAD_CREATE(thr, mdb_env_copythr, &my);
mp = (MDB_page *)my.mc_wbuf[0]; mp = (MDB_page *)my.mc_wbuf[0];
memset(mp, 0, 2*env->me_psize); memset(mp, 0, 2*env->me_psize);
mp->mp_pgno = 0; mp->mp_pgno = 0;
@ -8368,21 +8381,28 @@ mdb_env_copyfd2(MDB_env *env, HANDLE fd)
} }
my.mc_wlen[0] = env->me_psize * 2; my.mc_wlen[0] = env->me_psize * 2;
my.mc_txn = txn; my.mc_txn = txn;
pthread_mutex_lock(&my.mc_mutex);
while(my.mc_new)
pthread_cond_wait(&my.mc_cond, &my.mc_mutex);
pthread_mutex_unlock(&my.mc_mutex);
rc = mdb_env_cwalk(&my, &txn->mt_dbs[1].md_root, 0); rc = mdb_env_cwalk(&my, &txn->mt_dbs[1].md_root, 0);
if (rc == MDB_SUCCESS && my.mc_wlen[my.mc_toggle]) if (rc == MDB_SUCCESS && my.mc_wlen[my.mc_toggle])
rc = mdb_env_cthr_toggle(&my); rc = mdb_env_cthr_toggle(&my, 1);
my.mc_wlen[my.mc_toggle] = 0; mdb_env_cthr_toggle(&my, -1);
pthread_mutex_unlock(&my.mc_mutex[my.mc_toggle]); pthread_mutex_lock(&my.mc_mutex);
while(my.mc_new)
pthread_cond_wait(&my.mc_cond, &my.mc_mutex);
pthread_mutex_unlock(&my.mc_mutex);
THREAD_FINISH(thr); THREAD_FINISH(thr);
leave: leave:
mdb_txn_abort(txn); mdb_txn_abort(txn);
#ifdef _WIN32 #ifdef _WIN32
CloseHandle(my.mc_mutex[1]); CloseHandle(my.mc_cond);
CloseHandle(my.mc_mutex[0]); CloseHandle(my.mc_mutex);
_aligned_free(my.mc_wbuf[0]); _aligned_free(my.mc_wbuf[0]);
#else #else
pthread_mutex_destroy(&my.mc_mutex[1]); pthread_cond_destroy(&my.mc_cond);
pthread_mutex_destroy(&my.mc_mutex[0]); pthread_mutex_destroy(&my.mc_mutex);
free(my.mc_wbuf[0]); free(my.mc_wbuf[0]);
#endif #endif
return rc; return rc;

Write Preview
Loading…
Cancel
Save