Skip to content

M
mariadb
Commit 8783b60a authored by unknown's avatar unknown
Browse files
Options

Fix for BUG#37737. 

Fixed prev_last_lsn assignment.

storage/maria/ma_loghandler.c:
  Fix for BUG#37737:
   - buffer flush finish now can be done only in order
   - overlay processing fixed (no back direction bufferslocking now)
  Fixed prev_last_lsn assignment (now only for new buffer and only under buffer lock protection).
parent 9e762653
Hide whitespace changes
Inline Side-by-side
Showing with 101 additions and 45 deletions
storage/maria/ma_loghandler.c
View file @ 8783b60a
......@@ -121,6 +121,8 @@ struct st_translog_buffer
in case of flush by LSN it can be offset + size - TRANSLOG_PAGE_SIZE)
*/
TRANSLOG_ADDRESS next_buffer_offset;
/* Previous buffer offset to detect it flush finish */
TRANSLOG_ADDRESS prev_buffer_offset;
/*
How much is written (or will be written when copy_to_buffer_in_progress
become 0) to this buffer
......@@ -135,12 +137,12 @@ struct st_translog_buffer
/* list of waiting buffer ready threads */
struct st_my_thread_var *waiting_flush;
/*
Pointer on the buffer which overlap with this one (due to flush of
If true then previous buffer overlap with this one (due to flush of
loghandler, the last page of that buffer is the same as the first page
of this buffer) and have to be written first (because contain old
content of page which present in both buffers)
*/
struct st_translog_buffer *overlay;
my_bool overlay;
uint buffer_no;
/*
Lock for the buffer.
......@@ -175,6 +177,14 @@ struct st_translog_buffer
With file and offset it allow detect buffer changes
*/
uint8 ver;
/*
When previous buffer sent to disk it set its address here to allow
to detect when it is done
(we have to keep it in this buffer to lock buffers only in one direction).
*/
TRANSLOG_ADDRESS prev_sent_to_disk;
pthread_cond_t prev_sent_to_disk_cond;
};
......@@ -1421,9 +1431,12 @@ static my_bool translog_buffer_init(struct st_translog_buffer *buffer)
/* list of waiting buffer ready threads */
buffer->waiting_flush= 0;
/* lock for the buffer. Current buffer also lock the handler */
if (pthread_mutex_init(&buffer->mutex, MY_MUTEX_INIT_FAST))
if (pthread_mutex_init(&buffer->mutex, MY_MUTEX_INIT_FAST) ||
pthread_cond_init(&buffer->prev_sent_to_disk_cond, 0))
DBUG_RETURN(1);
buffer->is_closing_buffer= 0;
buffer->prev_sent_to_disk= LSN_IMPOSSIBLE;
buffer->prev_buffer_offset= LSN_IMPOSSIBLE;
buffer->ver= 0;
DBUG_RETURN(0);
}
......@@ -2100,10 +2113,12 @@ static my_bool translog_buffer_next(TRANSLOG_ADDRESS *horizon,
{
translog_lock_assert_owner();
translog_start_buffer(new_buffer, cursor, new_buffer_no);
new_buffer->prev_buffer_offset=
log_descriptor.buffers[old_buffer_no].offset;
new_buffer->prev_last_lsn=
BUFFER_MAX_LSN(log_descriptor.buffers + old_buffer_no);
}
log_descriptor.buffers[old_buffer_no].next_buffer_offset= new_buffer->offset;
new_buffer->prev_last_lsn=
BUFFER_MAX_LSN(log_descriptor.buffers + old_buffer_no);
DBUG_PRINT("info", ("prev_last_lsn set to (%lu,0x%lx) buffer: 0x%lx",
LSN_IN_PARTS(new_buffer->prev_last_lsn),
(ulong) new_buffer));
......@@ -2117,14 +2132,16 @@ static my_bool translog_buffer_next(TRANSLOG_ADDRESS *horizon,
SYNOPSIS
translog_set_sent_to_disk()
lsn LSN to assign
in_buffers to assign to in_buffers_only
buffer buffer which we have sent to disk
TODO: use atomic operations if possible (64bit architectures?)
*/
static void translog_set_sent_to_disk(LSN lsn, TRANSLOG_ADDRESS in_buffers)
static void translog_set_sent_to_disk(struct st_translog_buffer *buffer)
{
LSN lsn= buffer->last_lsn;
TRANSLOG_ADDRESS in_buffers= buffer->next_buffer_offset;
DBUG_ENTER("translog_set_sent_to_disk");
pthread_mutex_lock(&log_descriptor.sent_to_disk_lock);
DBUG_PRINT("enter", ("lsn: (%lu,0x%lx) in_buffers: (%lu,0x%lx) "
......@@ -2415,6 +2432,51 @@ static uint16 translog_get_total_chunk_length(uchar *page, uint16 offset)
}
}
/*
@brief Waits previous buffer flush finish
@param buffer buffer for check
@retval 0 previous buffer flushed and this thread have to flush this one
@retval 1 previous buffer flushed and this buffer flushed by other thread too
*/
my_bool translog_prev_buffer_flush_wait(struct st_translog_buffer *buffer)
{
TRANSLOG_ADDRESS offset= buffer->offset;
TRANSLOG_FILE *file= buffer->file;
uint8 ver= buffer->ver;
DBUG_ENTER("translog_prev_buffer_flush_wait");
DBUG_PRINT("enter", ("buffer: 0x%lx #%u offset: (%lu,0x%lx) "
"prev sent: (%lu,0x%lx) prev offset: (%lu,0x%lx)",
(ulong) buffer, (uint) buffer->buffer_no,
LSN_IN_PARTS(buffer->offset),
LSN_IN_PARTS(buffer->prev_sent_to_disk),
LSN_IN_PARTS(buffer->prev_buffer_offset)));
translog_buffer_lock_assert_owner(buffer);
/*
if prev_sent_to_disk == LSN_IMPOSSIBLE then
prev_buffer_offset should be LSN_IMPOSSIBLE
because it means that this buffer was never used
*/
DBUG_ASSERT((buffer->prev_sent_to_disk == LSN_IMPOSSIBLE &&
buffer->prev_buffer_offset == LSN_IMPOSSIBLE) ||
buffer->prev_sent_to_disk != LSN_IMPOSSIBLE);
if (buffer->prev_buffer_offset != buffer->prev_sent_to_disk)
{
do {
pthread_cond_wait(&buffer->prev_sent_to_disk_cond, &buffer->mutex);
if (buffer->file != file || buffer->offset != offset ||
buffer->ver != ver)
{
translog_buffer_unlock(buffer);
DBUG_RETURN(1); /* some the thread flushed the buffer already */
}
} while(buffer->prev_buffer_offset != buffer->prev_sent_to_disk);
}
DBUG_RETURN(0);
}
/*
Flush given buffer
......@@ -2460,39 +2522,8 @@ static my_bool translog_buffer_flush(struct st_translog_buffer *buffer)
if (buffer->file != file || buffer->offset != offset || buffer->ver != ver)
DBUG_RETURN(0); /* some the thread flushed the buffer already */
if (buffer->overlay && buffer->overlay->file == buffer->file &&
cmp_translog_addr(buffer->overlay->offset + buffer->overlay->size,
buffer->offset) > 0)
{
/*
This can't happen for normal translog_flush,
only during destroying the loghandler
*/
struct st_translog_buffer *overlay= buffer->overlay;
TRANSLOG_ADDRESS buffer_offset= buffer->offset;
TRANSLOG_FILE *fl= buffer->file;
uint8 ver= buffer->ver;
translog_buffer_unlock(buffer);
translog_buffer_lock(overlay);
/* rechecks under mutex protection that overlay is still our overlay */
if (buffer->overlay->file == fl &&
cmp_translog_addr(buffer->overlay->offset + buffer->overlay->size,
buffer_offset) > 0)
{
translog_wait_for_buffer_free(overlay);
}
translog_buffer_unlock(overlay);
translog_buffer_lock(buffer);
if (buffer->file != fl || buffer_offset != buffer->offset ||
ver != buffer->ver)
{
/*
This means that somebody else flushed the buffer while we was
waiting for overlay then for locking buffer again.
*/
DBUG_RETURN(0);
}
}
if (buffer->overlay && translog_prev_buffer_flush_wait(buffer))
DBUG_RETURN(0); /* some the thread flushed the buffer already */
/*
Send page by page in the pagecache what we are going to write on the
......@@ -2553,10 +2584,34 @@ static my_bool translog_buffer_flush(struct st_translog_buffer *buffer)
file->is_sync= 0;
if (LSN_OFFSET(buffer->last_lsn) != 0) /* if buffer->last_lsn is set */
translog_set_sent_to_disk(buffer->last_lsn,
buffer->next_buffer_offset);
{
if (translog_prev_buffer_flush_wait(buffer))
DBUG_RETURN(0); /* some the thread flushed the buffer already */
translog_set_sent_to_disk(buffer);
}
else
translog_set_only_in_buffers(buffer->next_buffer_offset);
/* say to next buffer that we are finished */
{
struct st_translog_buffer *next_buffer=
log_descriptor.buffers + ((buffer->buffer_no + 1) % TRANSLOG_BUFFERS_NO);
if (likely(translog_status == TRANSLOG_OK)){
translog_buffer_lock(next_buffer);
next_buffer->prev_sent_to_disk= buffer->offset;
translog_buffer_unlock(next_buffer);
pthread_cond_broadcast(&next_buffer->prev_sent_to_disk_cond);
}
else
{
/*
It is shutdown =>
1) there is only one thread
2) mutexes of other buffers can be destroyed => we can't use them
*/
next_buffer->prev_sent_to_disk= buffer->offset;
}
}
/* Free buffer */
buffer->file= NULL;
buffer->overlay= 0;
......@@ -4640,6 +4695,7 @@ static my_bool translog_advance_pointer(int pages, uint16 last_page_data)
}
translog_start_buffer(new_buffer, &log_descriptor.bc, new_buffer_no);
old_buffer->next_buffer_offset= new_buffer->offset;
new_buffer->prev_buffer_offset= old_buffer->offset;
translog_buffer_unlock(old_buffer);
offset-= min_offset;
}
......@@ -7355,7 +7411,7 @@ static void translog_force_current_buffer_to_finish()
log_descriptor.bc.ptr+= current_page_fill;
log_descriptor.bc.buffer->size= log_descriptor.bc.current_page_fill=
current_page_fill;
new_buffer->overlay= old_buffer;
new_buffer->overlay= 1;
}
else
translog_new_page_header(&log_descriptor.horizon, &log_descriptor.bc);
......@@ -7428,8 +7484,8 @@ static void translog_force_current_buffer_to_finish()
memcpy(new_buffer->buffer, data, current_page_fill);
}
old_buffer->next_buffer_offset= new_buffer->offset;
translog_buffer_lock(new_buffer);
new_buffer->prev_buffer_offset= old_buffer->offset;
translog_buffer_decrease_writers(new_buffer);
translog_buffer_unlock(new_buffer);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7