Skip to content

Commit

Permalink
lmdb: Sync to upstream.
Browse files Browse the repository at this point in the history
  • Loading branch information
Marco van Wieringen committed Jun 30, 2016
1 parent a8727a6 commit 1dedc34
Show file tree
Hide file tree
Showing 2 changed files with 100 additions and 83 deletions.
1 change: 1 addition & 0 deletions src/lmdb/lmdb.h
View file
Expand Up @@ -688,6 +688,7 @@ int mdb_env_copyfd(MDB_env *env, mdb_filehandle_t fd);
* <li>#MDB_CP_COMPACT - Perform compaction while copying: omit free
* pages and sequentially renumber all pages in output. This option
* consumes more CPU and runs more slowly than the default.
* Currently it fails if the environment has suffered a page leak.
* </ul>
* @return A non-zero error value on failure and 0 on success.
*/
Expand Down
182 changes: 99 additions & 83 deletions src/lmdb/mdb.c
View file
Expand Up @@ -340,8 +340,10 @@ typedef HANDLE mdb_mutex_t, mdb_mutexref_t;
#define pthread_mutex_lock(x) WaitForSingleObject(*x, INFINITE)
#define pthread_cond_signal(x) SetEvent(*x)
#define pthread_cond_wait(cond,mutex) do{SignalObjectAndWait(*mutex, *cond, INFINITE, FALSE); WaitForSingleObject(*mutex, INFINITE);}while(0)
#define THREAD_CREATE(thr,start,arg) thr=CreateThread(NULL,0,start,arg,0,NULL)
#define THREAD_FINISH(thr) WaitForSingleObject(thr, INFINITE)
#define THREAD_CREATE(thr,start,arg) \
(((thr) = CreateThread(NULL, 0, start, arg, 0, NULL)) ? 0 : ErrCode())
#define THREAD_FINISH(thr) \
(WaitForSingleObject(thr, INFINITE) ? ErrCode() : 0)
#define LOCK_MUTEX0(mutex) WaitForSingleObject(mutex, INFINITE)
#define UNLOCK_MUTEX(mutex) ReleaseMutex(mutex)
#define mdb_mutex_consistent(mutex) 0
Expand Down Expand Up @@ -9645,11 +9647,12 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi,
#ifndef MDB_WBUF
#define MDB_WBUF (1024*1024)
#endif
#define MDB_EOF 0x10 /**< #mdb_env_copyfd1() is done reading */

/** State needed for a compacting copy. */
/** State needed for a double-buffering compacting copy. */
typedef struct mdb_copy {
pthread_mutex_t mc_mutex;
pthread_cond_t mc_cond;
pthread_cond_t mc_cond; /**< Condition variable for #mc_new */
char *mc_wbuf[2];
char *mc_over[2];
MDB_env *mc_env;
Expand All @@ -9658,10 +9661,9 @@ typedef struct mdb_copy {
int mc_olen[2];
pgno_t mc_next_pgno;
HANDLE mc_fd;
int mc_status;
volatile int mc_new;
int mc_toggle;

int mc_toggle; /**< Buffer number in provider */
int mc_new; /**< (0-2 buffers to write) | (#MDB_EOF at end) */
volatile int mc_error; /**< Error code, never cleared if set */
} mdb_copy;

/** Dedicated writer thread for compacting copy. */
Expand All @@ -9680,20 +9682,16 @@ mdb_env_copythr(void *arg)
#endif

pthread_mutex_lock(&my->mc_mutex);
my->mc_new = 0;
pthread_cond_signal(&my->mc_cond);
for(;;) {
while (!my->mc_new)
pthread_cond_wait(&my->mc_cond, &my->mc_mutex);
if (my->mc_new < 0) {
my->mc_new = 0;
if (my->mc_new == 0 + MDB_EOF) /* 0 buffers, just EOF */
break;
}
my->mc_new = 0;
wsize = my->mc_wlen[toggle];
ptr = my->mc_wbuf[toggle];
again:
while (wsize > 0) {
rc = MDB_SUCCESS;
while (wsize > 0 && !my->mc_error) {
DO_WRITE(rc, my->mc_fd, ptr, wsize, len);
if (!rc) {
rc = ErrCode();
Expand All @@ -9709,8 +9707,7 @@ mdb_env_copythr(void *arg)
}
}
if (rc) {
my->mc_status = rc;
break;
my->mc_error = rc;
}
/* If there's an overflow page tail, write it too */
if (my->mc_olen[toggle]) {
Expand All @@ -9721,31 +9718,34 @@ mdb_env_copythr(void *arg)
}
my->mc_wlen[toggle] = 0;
toggle ^= 1;
/* Return the empty buffer to provider */
my->mc_new--;
pthread_cond_signal(&my->mc_cond);
}
pthread_cond_signal(&my->mc_cond);
pthread_mutex_unlock(&my->mc_mutex);
return (THREAD_RET)0;
#undef DO_WRITE
}

/** Tell the writer thread there's a buffer ready to write */
/** Give buffer and/or #MDB_EOF to writer thread, await unused buffer.
*
* @param[in] my control structure.
* @param[in] adjust (1 to hand off 1 buffer) | (MDB_EOF when ending).
*/
static int ESECT
mdb_env_cthr_toggle(mdb_copy *my, int st)
mdb_env_cthr_toggle(mdb_copy *my, int adjust)
{
int toggle = my->mc_toggle ^ 1;
pthread_mutex_lock(&my->mc_mutex);
if (my->mc_status) {
pthread_mutex_unlock(&my->mc_mutex);
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_new += adjust;
pthread_cond_signal(&my->mc_cond);
while (my->mc_new & 2) /* both buffers in use */
pthread_cond_wait(&my->mc_cond, &my->mc_mutex);
pthread_mutex_unlock(&my->mc_mutex);
return 0;

my->mc_toggle ^= (adjust & 1);
/* Both threads reset mc_wlen, to be safe from threading errors */
my->mc_wlen[my->mc_toggle] = 0;
return my->mc_error;
}

/** Depth-first tree traversal for compacting copy. */
Expand Down Expand Up @@ -9811,6 +9811,7 @@ mdb_env_cwalk(mdb_copy *my, pgno_t *pg, int flags)
}

memcpy(&pg, NODEDATA(ni), sizeof(pg));
memcpy(NODEDATA(ni), &my->mc_next_pgno, sizeof(pgno_t));
rc = mdb_page_get(&mc, pg, &omp, NULL);
if (rc)
goto done;
Expand All @@ -9833,7 +9834,6 @@ mdb_env_cwalk(mdb_copy *my, pgno_t *pg, int flags)
goto done;
toggle = my->mc_toggle;
}
memcpy(NODEDATA(ni), &mo->mp_pgno, sizeof(pgno_t));
} else if (ni->mn_flags & F_SUBDATA) {
MDB_db db;

Expand Down Expand Up @@ -9911,47 +9911,56 @@ mdb_env_copyfd1(MDB_env *env, HANDLE fd)
{
MDB_meta *mm;
MDB_page *mp;
mdb_copy my;
mdb_copy my = {0};
MDB_txn *txn = NULL;
pthread_t thr;
int rc;
pgno_t root, new_root;
int rc = MDB_SUCCESS;

#ifdef _WIN32
my.mc_mutex = CreateMutex(NULL, FALSE, NULL);
my.mc_cond = CreateEvent(NULL, FALSE, FALSE, NULL);
if (!(my.mc_mutex = CreateMutex(NULL, FALSE, NULL)) ||
!(my.mc_cond = CreateEvent(NULL, FALSE, FALSE, NULL))) {
rc = ErrCode();
goto done;
}
my.mc_wbuf[0] = _aligned_malloc(MDB_WBUF*2, env->me_os_psize);
if (my.mc_wbuf[0] == NULL)
return errno;
if (my.mc_wbuf[0] == NULL) {
/* _aligned_malloc() sets errno, but we use Windows error codes */
rc = ERROR_NOT_ENOUGH_MEMORY;
goto done;
}
#else
pthread_mutex_init(&my.mc_mutex, NULL);
pthread_cond_init(&my.mc_cond, NULL);
if ((rc = pthread_mutex_init(&my.mc_mutex, NULL)) != 0)
return rc;
if ((rc = pthread_cond_init(&my.mc_cond, NULL)) != 0)
goto done2;
#ifdef HAVE_MEMALIGN
my.mc_wbuf[0] = memalign(env->me_os_psize, MDB_WBUF*2);
if (my.mc_wbuf[0] == NULL)
return errno;
if (my.mc_wbuf[0] == NULL) {
rc = errno;
goto done;
}
#else
rc = posix_memalign((void **)&my.mc_wbuf[0], env->me_os_psize, MDB_WBUF*2);
if (rc)
return rc;
{
void *p;
if ((rc = posix_memalign(&p, env->me_os_psize, MDB_WBUF*2)) != 0)
goto done;
my.mc_wbuf[0] = p;
}
#endif
#endif
memset(my.mc_wbuf[0], 0, MDB_WBUF*2);
my.mc_wbuf[1] = my.mc_wbuf[0] + MDB_WBUF;
my.mc_wlen[0] = 0;
my.mc_wlen[1] = 0;
my.mc_olen[0] = 0;
my.mc_olen[1] = 0;
my.mc_next_pgno = NUM_METAS;
my.mc_status = 0;
my.mc_new = 1;
my.mc_toggle = 0;
my.mc_env = env;
my.mc_fd = fd;
THREAD_CREATE(thr, mdb_env_copythr, &my);
rc = THREAD_CREATE(thr, mdb_env_copythr, &my);
if (rc)
goto done;

rc = mdb_txn_begin(env, NULL, MDB_RDONLY, &txn);
if (rc)
return rc;
goto finish;

mp = (MDB_page *)my.mc_wbuf[0];
memset(mp, 0, NUM_METAS * env->me_psize);
Expand All @@ -9967,57 +9976,64 @@ mdb_env_copyfd1(MDB_env *env, HANDLE fd)
*(MDB_meta *)METADATA(mp) = *mm;
mm = (MDB_meta *)METADATA(mp);

/* Count the number of free pages, subtract from lastpg to find
* number of active pages
*/
{
/* Set metapage 1 with current main DB */
root = new_root = txn->mt_dbs[MAIN_DBI].md_root;
if (root != P_INVALID) {
/* Count free pages + freeDB pages. Subtract from last_pg
* to find the new last_pg, which also becomes the new root.
*/
MDB_ID freecount = 0;
MDB_cursor mc;
MDB_val key, data;
mdb_cursor_init(&mc, txn, FREE_DBI, NULL);
while ((rc = mdb_cursor_get(&mc, &key, &data, MDB_NEXT)) == 0)
freecount += *(MDB_ID *)data.mv_data;
if (rc != MDB_NOTFOUND)
goto finish;
freecount += txn->mt_dbs[FREE_DBI].md_branch_pages +
txn->mt_dbs[FREE_DBI].md_leaf_pages +
txn->mt_dbs[FREE_DBI].md_overflow_pages;

/* Set metapage 1 */
mm->mm_last_pg = txn->mt_next_pgno - freecount - 1;
new_root = txn->mt_next_pgno - 1 - freecount;
mm->mm_last_pg = new_root;
mm->mm_dbs[MAIN_DBI] = txn->mt_dbs[MAIN_DBI];
if (mm->mm_last_pg > NUM_METAS-1) {
mm->mm_dbs[MAIN_DBI].md_root = mm->mm_last_pg;
mm->mm_txnid = 1;
} else {
mm->mm_dbs[MAIN_DBI].md_root = P_INVALID;
}
mm->mm_dbs[MAIN_DBI].md_root = new_root;
} else {
/* When the DB is empty, handle it specially to
* fix any breakage like page leaks from ITS#8174.
*/
mm->mm_dbs[MAIN_DBI].md_flags = txn->mt_dbs[MAIN_DBI].md_flags;
}
if (root != P_INVALID || mm->mm_dbs[MAIN_DBI].md_flags) {
mm->mm_txnid = 1; /* use metapage 1 */
}

my.mc_wlen[0] = env->me_psize * NUM_METAS;
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[MAIN_DBI].md_root, 0);
if (rc == MDB_SUCCESS && my.mc_wlen[my.mc_toggle])
rc = mdb_env_cthr_toggle(&my, 1);
mdb_env_cthr_toggle(&my, -1);
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);
rc = mdb_env_cwalk(&my, &root, 0);
if (rc == MDB_SUCCESS && root != new_root) {
rc = MDB_INCOMPATIBLE; /* page leak or corrupt DB */
}

finish:
if (rc)
my.mc_error = rc;
mdb_env_cthr_toggle(&my, 1 | MDB_EOF);
rc = THREAD_FINISH(thr);
mdb_txn_abort(txn);

done:
#ifdef _WIN32
CloseHandle(my.mc_cond);
CloseHandle(my.mc_mutex);
_aligned_free(my.mc_wbuf[0]);
if (my.mc_wbuf[0]) _aligned_free(my.mc_wbuf[0]);
if (my.mc_cond) CloseHandle(my.mc_cond);
if (my.mc_mutex) CloseHandle(my.mc_mutex);
#else
free(my.mc_wbuf[0]);
pthread_cond_destroy(&my.mc_cond);
done2:
pthread_mutex_destroy(&my.mc_mutex);
free(my.mc_wbuf[0]);
#endif
return rc;
return rc ? rc : my.mc_error;
}

/** Copy environment as-is. */
Expand Down

0 comments on commit 1dedc34

Please sign in to comment.