mirror of
https://github.com/isar/libmdbx.git
synced 2025-01-02 01:34:12 +08:00
mdbx: rework reader_check0() and mutex recovery.
This commit is contained in:
parent
41c51fdac2
commit
55226499a8
@ -829,3 +829,6 @@ static __inline size_t roundup2(size_t value, size_t granularity) {
|
||||
assert(is_power2(granularity));
|
||||
return (value + granularity - 1) & ~(granularity - 1);
|
||||
}
|
||||
|
||||
#define MDBX_IS_ERROR(rc) \
|
||||
((rc) != MDBX_RESULT_TRUE && (rc) != MDBX_RESULT_FALSE)
|
||||
|
@ -121,17 +121,18 @@ int mdbx_rdt_lock(MDB_env *env) {
|
||||
|
||||
void mdbx_rdt_unlock(MDB_env *env) {
|
||||
int rc = mdbx_robust_unlock(env, &env->me_txns->mti_rmutex);
|
||||
if (unlikely(rc != 0))
|
||||
if (unlikely(MDBX_IS_ERROR(rc)))
|
||||
mdbx_panic("%s() failed: errcode %d\n", mdbx_func_, rc);
|
||||
}
|
||||
|
||||
int mdbx_txn_lock(MDB_env *env) {
|
||||
return mdbx_robust_lock(env, &env->me_txns->mti_wmutex);
|
||||
int rc = mdbx_robust_lock(env, &env->me_txns->mti_wmutex);
|
||||
return MDBX_IS_ERROR(rc) ? rc : MDB_SUCCESS;
|
||||
}
|
||||
|
||||
void mdbx_txn_unlock(MDB_env *env) {
|
||||
int rc = mdbx_robust_unlock(env, &env->me_txns->mti_wmutex);
|
||||
if (unlikely(rc != 0))
|
||||
if (unlikely(MDBX_IS_ERROR(rc)))
|
||||
mdbx_panic("%s() failed: errcode %d\n", mdbx_func_, rc);
|
||||
}
|
||||
|
||||
@ -205,42 +206,46 @@ static int mdbx_lck_op(mdbx_filehandle_t fd, int op, int lck, off_t offset) {
|
||||
}
|
||||
}
|
||||
|
||||
#if !__GLIBC_PREREQ(2, 12) && !defined(pthread_mutex_consistent)
|
||||
#define pthread_mutex_consistent(mutex) pthread_mutex_consistent_np(mutex)
|
||||
#endif
|
||||
|
||||
static int __cold mdbx_mutex_failed(MDB_env *env, mdbx_mutex_t *mutex, int rc) {
|
||||
#if MDB_USE_ROBUST
|
||||
if (unlikely(rc == EOWNERDEAD)) {
|
||||
if (rc == EOWNERDEAD) {
|
||||
/* We own the mutex. Clean up after dead previous owner. */
|
||||
rc = MDB_SUCCESS;
|
||||
|
||||
int rlocked = (mutex == &env->me_txns->mti_rmutex);
|
||||
rc = MDB_SUCCESS;
|
||||
if (!rlocked) {
|
||||
/* env is hosed if the dead thread was ours */
|
||||
if (env->me_txn) {
|
||||
if (unlikely(env->me_txn)) {
|
||||
/* env is hosed if the dead thread was ours */
|
||||
env->me_flags |= MDB_FATAL_ERROR;
|
||||
env->me_txn = NULL;
|
||||
rc = MDB_PANIC;
|
||||
}
|
||||
}
|
||||
mdbx_debug("%cmutex owner died, %s", (rlocked ? 'r' : 'w'),
|
||||
(rc ? "this process' env is hosed" : "recovering"));
|
||||
int rc2 = mdbx_reader_check0(env, rlocked, NULL);
|
||||
if (rc2 == 0)
|
||||
#if __GLIBC_PREREQ(2, 12)
|
||||
rc2 = pthread_mutex_consistent(mutex);
|
||||
#else
|
||||
rc2 = pthread_mutex_consistent_np(mutex);
|
||||
#endif
|
||||
if (rc || (rc = rc2)) {
|
||||
mdbx_debug("mutex recovery failed, %s", mdbx_strerror(rc));
|
||||
mdbx_notice("%cmutex owner died, %s", (rlocked ? 'r' : 'w'),
|
||||
(rc ? "this process' env is hosed" : "recovering"));
|
||||
|
||||
int check_rc = mdbx_reader_check0(env, rlocked, NULL);
|
||||
int mreco_rc = pthread_mutex_consistent(mutex);
|
||||
check_rc = (mreco_rc == 0) ? check_rc : mreco_rc;
|
||||
|
||||
if (unlikely(mreco_rc))
|
||||
mdbx_error("mutex recovery failed, %s", mdbx_strerror(mreco_rc));
|
||||
|
||||
rc = (rc == MDB_SUCCESS) ? check_rc : rc;
|
||||
if (MDBX_IS_ERROR(rc))
|
||||
pthread_mutex_unlock(mutex);
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
#endif /* MDB_USE_ROBUST */
|
||||
|
||||
if (unlikely(rc)) {
|
||||
mdbx_debug("lock mutex failed, %s", mdbx_strerror(rc));
|
||||
if (rc != EDEADLK) {
|
||||
env->me_flags |= MDB_FATAL_ERROR;
|
||||
rc = MDB_PANIC;
|
||||
}
|
||||
mdbx_error("lock mutex failed, %s", mdbx_strerror(rc));
|
||||
if (rc != EDEADLK) {
|
||||
env->me_flags |= MDB_FATAL_ERROR;
|
||||
rc = MDB_PANIC;
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
49
src/mdbx.c
49
src/mdbx.c
@ -2129,8 +2129,9 @@ static int mdbx_txn_renew0(MDB_txn *txn, unsigned flags) {
|
||||
mdbx_tid_t tid = mdbx_thread_self();
|
||||
|
||||
rc = mdbx_rdt_lock(env);
|
||||
if (unlikely(rc != MDB_SUCCESS))
|
||||
if (unlikely(MDBX_IS_ERROR(rc)))
|
||||
return rc;
|
||||
rc = MDB_SUCCESS;
|
||||
|
||||
if (unlikely(env->me_live_reader != pid)) {
|
||||
rc = mdbx_rpid_set(env);
|
||||
@ -9129,11 +9130,10 @@ int __cold mdbx_reader_check0(MDB_env *env, int rdt_locked, int *dead) {
|
||||
mdbx_pid_t *pids = alloca((snap_nreaders + 1) * sizeof(mdbx_pid_t));
|
||||
pids[0] = 0;
|
||||
|
||||
unsigned i;
|
||||
int rc = MDBX_RESULT_FALSE, count = 0;
|
||||
MDB_reader *mr = env->me_txns->mti_readers;
|
||||
|
||||
for (i = 0; i < snap_nreaders; i++) {
|
||||
for (unsigned i = 0; i < snap_nreaders; i++) {
|
||||
const mdbx_pid_t pid = mr[i].mr_pid;
|
||||
if (pid == 0)
|
||||
continue;
|
||||
@ -9151,33 +9151,32 @@ int __cold mdbx_reader_check0(MDB_env *env, int rdt_locked, int *dead) {
|
||||
|
||||
/* stale reader found */
|
||||
if (!rdt_locked) {
|
||||
rdt_locked = -1;
|
||||
rc = mdbx_rdt_lock(env);
|
||||
if (rc != MDB_SUCCESS) {
|
||||
if (rc != MDBX_RESULT_TRUE)
|
||||
break; /* lock failed */
|
||||
/* recovered after mutex owner died */
|
||||
snap_nreaders = 0; /* the above checked all readers */
|
||||
} else {
|
||||
/* a other process may have clean and reused slot, recheck */
|
||||
if (mr[i].mr_pid != pid)
|
||||
continue;
|
||||
rc = mdbx_rpid_check(env, pid);
|
||||
if (rc != MDBX_RESULT_FALSE) {
|
||||
if (rc != MDBX_RESULT_TRUE)
|
||||
break; /* mdbx_rpid_check() failed */
|
||||
/* the race with other process, slot reused */
|
||||
rc = MDBX_RESULT_FALSE;
|
||||
continue;
|
||||
}
|
||||
if (MDBX_IS_ERROR(rc))
|
||||
break;
|
||||
|
||||
rdt_locked = -1;
|
||||
if (rc == MDBX_RESULT_TRUE)
|
||||
/* the above checked all readers */
|
||||
break;
|
||||
|
||||
/* a other process may have clean and reused slot, recheck */
|
||||
if (mr[i].mr_pid != pid)
|
||||
continue;
|
||||
|
||||
rc = mdbx_rpid_check(env, pid);
|
||||
if (MDBX_IS_ERROR(rc))
|
||||
break;
|
||||
|
||||
if (rc != MDBX_RESULT_FALSE) {
|
||||
/* the race with other process, slot reused */
|
||||
rc = MDBX_RESULT_FALSE;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
assert(mr[i].mr_pid == pid);
|
||||
|
||||
/* clean it */
|
||||
unsigned j;
|
||||
for (j = i; j < snap_nreaders; j++) {
|
||||
for (unsigned j = i; j < snap_nreaders; j++) {
|
||||
if (mr[j].mr_pid == pid) {
|
||||
mdbx_debug("clear stale reader pid %u txn %zd", (unsigned)pid,
|
||||
mr[j].mr_txnid);
|
||||
|
Loading…
x
Reference in New Issue
Block a user