1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MDEV-28689, MDEV-28690: Incorrect error handling for ctrl_mutex

Browse Source 
comp_thread_ctxt_t: Remove ctrl_mutex, ctrl_cond, started. We do not
actually need them for anything.

destroy_worker_thread(): Split from destroy_worker_threads().

create_worker_threads(): We already initialize
thd->data_avail=FALSE and thd->cancelled=FALSE before
invoking pthread_create(). If any thread creation fails,
clean up by destroy_worker_thread().

compress_worker_thread_func(): Assume that thd->started and
thd->data_avail are already initialized.

Reviewed by: Vladislav Vaintroub
This commit is contained in:
Marko Mäkelä 2022-05-30 15:49:45 +03:00
parent f7137a619f
commit 863c3eda87
1 changed files with 22 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
extra/mariabackup/ds_compress.cc
View File

@ -1,5 +1,6 @@
/******************************************************
Copyright (c) 2011-2013 Percona LLC and/or its affiliates.
Copyright (c) 2022, MariaDB Corporation.
Compressing datasink implementation for XtraBackup.
@ -32,11 +33,8 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA
typedef struct {
pthread_t id;
uint num;
pthread_mutex_t ctrl_mutex;
pthread_cond_t ctrl_cond;
pthread_mutex_t data_mutex;
pthread_cond_t data_cond;
my_bool started;
my_bool data_avail;
my_bool cancelled;
const char *from;
@ -208,14 +206,13 @@ compress_write(ds_file_t *file, const uchar *buf, size_t len)
thd = threads + i;
pthread_mutex_lock(&thd->ctrl_mutex);
pthread_mutex_lock(&thd->data_mutex);
chunk_len = (len > COMPRESS_CHUNK_SIZE) ?
COMPRESS_CHUNK_SIZE : len;
thd->from = ptr;
thd->from_len = chunk_len;
pthread_mutex_lock(&thd->data_mutex);
thd->data_avail = TRUE;
pthread_cond_signal(&thd->data_cond);
pthread_mutex_unlock(&thd->data_mutex);
@ -260,7 +257,6 @@ compress_write(ds_file_t *file, const uchar *buf, size_t len)
}
pthread_mutex_unlock(&threads[i].data_mutex);
pthread_mutex_unlock(&threads[i].ctrl_mutex);
}
}
@ -330,6 +326,23 @@ write_uint64_le(ds_file_t *file, ulonglong n)
return ds_write(file, tmp, sizeof(tmp));
}
static
void
destroy_worker_thread(comp_thread_ctxt_t *thd)
{
pthread_mutex_lock(&thd->data_mutex);
thd->cancelled = TRUE;
pthread_cond_signal(&thd->data_cond);
pthread_mutex_unlock(&thd->data_mutex);
pthread_join(thd->id, NULL);
pthread_cond_destroy(&thd->data_cond);
pthread_mutex_destroy(&thd->data_mutex);
my_free(thd->to);
}
static
comp_thread_ctxt_t *
create_worker_threads(uint n)
@ -344,7 +357,6 @@ create_worker_threads(uint n)
comp_thread_ctxt_t *thd = threads + i;
thd->num = i + 1;
thd->started = FALSE;
thd->cancelled = FALSE;
thd->data_avail = FALSE;
@ -352,46 +364,25 @@ create_worker_threads(uint n)
MY_QLZ_COMPRESS_OVERHEAD,
MYF(MY_FAE));
/* Initialize the control mutex and condition var */
if (pthread_mutex_init(&thd->ctrl_mutex, NULL) ||
pthread_cond_init(&thd->ctrl_cond, NULL)) {
goto err;
}
/* Initialize and data mutex and condition var */
if (pthread_mutex_init(&thd->data_mutex, NULL) ||
pthread_cond_init(&thd->data_cond, NULL)) {
goto err;
}
pthread_mutex_lock(&thd->ctrl_mutex);
if (pthread_create(&thd->id, NULL, compress_worker_thread_func,
thd)) {
msg("compress: pthread_create() failed: "
"errno = %d", errno);
pthread_mutex_unlock(&thd->ctrl_mutex);
goto err;
}
}
/* Wait for the threads to start */
for (i = 0; i < n; i++) {
comp_thread_ctxt_t *thd = threads + i;
while (thd->started == FALSE)
pthread_cond_wait(&thd->ctrl_cond, &thd->ctrl_mutex);
pthread_mutex_unlock(&thd->ctrl_mutex);
}
return threads;
err:
while (i > 0) {
comp_thread_ctxt_t *thd;
i--;
thd = threads + i;
pthread_mutex_unlock(&thd->ctrl_mutex);
for (; i; i--) {
destroy_worker_thread(threads + i);
}
my_free(threads);
@ -405,21 +396,7 @@ destroy_worker_threads(comp_thread_ctxt_t *threads, uint n)
uint i;
for (i = 0; i < n; i++) {
comp_thread_ctxt_t *thd = threads + i;
pthread_mutex_lock(&thd->data_mutex);
threads[i].cancelled = TRUE;
pthread_cond_signal(&thd->data_cond);
pthread_mutex_unlock(&thd->data_mutex);
pthread_join(thd->id, NULL);
pthread_cond_destroy(&thd->data_cond);
pthread_mutex_destroy(&thd->data_mutex);
pthread_cond_destroy(&thd->ctrl_cond);
pthread_mutex_destroy(&thd->ctrl_mutex);
my_free(thd->to);
destroy_worker_thread(threads + i);
}
my_free(threads);
@ -431,19 +408,9 @@ compress_worker_thread_func(void *arg)
{
comp_thread_ctxt_t *thd = (comp_thread_ctxt_t *) arg;
pthread_mutex_lock(&thd->ctrl_mutex);
pthread_mutex_lock(&thd->data_mutex);
thd->started = TRUE;
pthread_cond_signal(&thd->ctrl_cond);
pthread_mutex_unlock(&thd->ctrl_mutex);
while (1) {
thd->data_avail = FALSE;
pthread_cond_signal(&thd->data_cond);
while (!thd->data_avail && !thd->cancelled) {
pthread_cond_wait(&thd->data_cond, &thd->data_mutex);
}