mysql::csa::Log_prefetcher Class Reference

This class starts an asynchronous thread that prefetches consecutive logs recorded in the index file. More...

#include <log_prefetcher.h>

Public Types
using	Mt_key = mysql::concurrency::Mutex_key
using	Cv_key = mysql::concurrency::Cv_key
using	Th_key = mysql::concurrency::Thread_key
using	Mem_key = PSI_memory_key
using	Memory_allocator = mysql::allocators::Allocator< uint8_t >
using	Data_type = Data_source
using	Elem_type = Data_source_sptr

Public Member Functions
	Log_prefetcher (MYSQL_BIN_LOG *log_ptr, Mt_key mt_key_wait=0, Cv_key cv_key_wait=0, Mt_key mt_key_file_move=0, Cv_key cv_key_file_move=0, Th_key key_th_prefetcher=0, Mem_key key_memory=0)
virtual	~Log_prefetcher ()
	Log_prefetcher (const Log_prefetcher &)=delete
Log_prefetcher &	operator= (const Log_prefetcher &)=delete
	Log_prefetcher (Log_prefetcher &&)=delete
Log_prefetcher &	operator= (Log_prefetcher &&)=delete
void	start_prefetcher ()
	Starts asynchronous thread prefetching data. More...
bool	open (const char *file_name)
	Tries to open a new, gifen file. More...
template<typename P >
std::optional< Elem_type >	dequeue (P &&wait_predicate)
	Consumes the next batch of data. More...
bool	is_stopped () const
	Checks whether prefetcher is stopped. More...
void	stop ()
	Gracefully stops the prefetcher. More...
bool	is_error () const
	Returns true if error occurred. More...
bool	is_waiting_for_next_file (const std::string &prev_file, const std::string &next_file)
	Checks whether we reached inactive file after the prev_file This function takes previous file to check if prefetcher finished rotation on this file and consecutive file to check if prefetcher opened it and stopped working until file becomes inactive. More...

Private Types
using	Thread_type = mysql::concurrency::Thread
	Thread type. More...
using	Locking_queue_type = mysql::concurrency::Locking_queue< Elem_type >
using	Sync_bounded_queue = mysql::concurrency::Sync_bounded_queue< Elem_type, tune::prefetcher_queue_max_size >
using	Queue_type = std::conditional< tune::csa_prefetcher_simple_queue, Locking_queue_type, Sync_bounded_queue >::type

Private Member Functions
bool	open_file (bool next_file)
	Opens new or next file for reading. More...
void	ensure_file_done (const std::string &file_name)
	If currently rotated file is file_name, waits until rotate is done. More...
void	set_error (const char *msg)
	Sets internal error (under lock) and internal error message. More...
std::string	get_current_file () const
	Obtains the current file under the lock. More...
void	update_current_file (const char *file_name, bool active)
	Updates the current file and its activity under the lock. More...
void	run_prefetch_thread ()
	Runs a thread that prefetch data from the relay log. More...

Private Attributes
std::atomic< bool >	m_stopped {false}
	Internal stop flag set to true when stop of the thread is requested. More...
MYSQL_BIN_LOG *	m_log
	The MYSQL_BIN_LOG object we use to know from where we need to read data. More...
Queue_type	m_cache
	Queue into which prefetcher puts data batches read from a raw binary file. More...
Thread_type	m_prefetcher
	Thread that runs prefetching. More...
bool	m_log_active {true}
	True in case reading from the active relay log file (used currently by the receiver thread), protected with m_mt_prefetcher. More...
mysql::concurrency::Mutex	m_mt_prefetcher
	Mutex protecting access to m_tasks. More...
mysql::concurrency::Condition_variable	m_cv_prefetcher
	Cv used by the scheduler main thread to wait on, when no task is available or tasks in m_task queue are not read to execute. More...
std::atomic< bool >	m_end {false}
	Notification atomic for end of execution. More...
mysql::concurrency::Mutex	m_mt_move_file
	Mutex protecting access to m_move_file. More...
mysql::concurrency::Condition_variable	m_cv_move_file
	CV to notify the watcher that prefetcher finished moving from m_move_file to the next file (see ensure_file_done) More...
std::string	m_move_file {""}
	Is set to a value if prefetcher initiated file move procedure Watcher may wait until prefetcher moves to the next file by using the m_cv_move_file notification cv and associated m_mt_move_file. More...
std::atomic< std::size_t >	m_bytes_fetched {0}
	This way we track the actual number of bytes that are cached in the m_cache. More...
std::size_t	m_batch_size {tune::prefetcher_batch_size}
	Default batch size ~16MB. More...
std::ifstream	m_istream
	Stream to read from. More...
std::string	m_error_message {""}
	Error message if any. Protected by m_mt_prefetcher. More...
std::atomic< bool >	m_is_error {false}
	Becomes true in case error has been encountered. More...
std::string	m_current_log_name {""}
	Currently processed log file Protected by m_mt_prefetcher. More...
std::size_t	m_current_file_length {0}
	Length of the currently opened file. More...
std::size_t	m_current_offset {0}
	Current file offset. More...
Th_key	m_key_th_prefetcher {0}
	Key for prefetcher thread. More...
Memory_allocator	m_allocator
	Memory_resource to handle all allocations. More...

Static Private Attributes
static constexpr std::size_t	max_bytes_fetched {1073741824}
	The maximum number of bytes we can prefetch. More...

Detailed Description

This class starts an asynchronous thread that prefetches consecutive logs recorded in the index file.

It contains the following main functions:

• dequeue - function to get the next prefetched batch (see Data_source)
• open - open a given file, check if file is hot and notify prefetcher to work on it
• stop - Gracefully stops the prefetcher and blocks until thread is joined
• is_waiting_for_next_file - checks whether prefetcher finished processing a given file and if prefetcher works on the next file or blocked
• is_error - Checks whether prefetcher reported an error

Member Typedef Documentation

Cv_key

using mysql::csa::Log_prefetcher::Cv_key = mysql::concurrency::Cv_key

Data_type

using mysql::csa::Log_prefetcher::Data_type = Data_source

Elem_type

using mysql::csa::Log_prefetcher::Elem_type = Data_source_sptr

Locking_queue_type

using mysql::csa::Log_prefetcher::Locking_queue_type = mysql::concurrency::Locking_queue<Elem_type>

private

Mem_key

using mysql::csa::Log_prefetcher::Mem_key = PSI_memory_key

Memory_allocator

using mysql::csa::Log_prefetcher::Memory_allocator = mysql::allocators::Allocator<uint8_t>

Mt_key

using mysql::csa::Log_prefetcher::Mt_key = mysql::concurrency::Mutex_key

Queue_type

using mysql::csa::Log_prefetcher::Queue_type = std::conditional<tune::csa_prefetcher_simple_queue, Locking_queue_type, Sync_bounded_queue>::type

private

Sync_bounded_queue

using mysql::csa::Log_prefetcher::Sync_bounded_queue = mysql::concurrency::Sync_bounded_queue<Elem_type, tune::prefetcher_queue_max_size>

private

Th_key

using mysql::csa::Log_prefetcher::Th_key = mysql::concurrency::Thread_key

Thread_type

using mysql::csa::Log_prefetcher::Thread_type = mysql::concurrency::Thread

private

Thread type.

Constructor & Destructor Documentation

Log_prefetcher() [1/3]

mysql::csa::Log_prefetcher::Log_prefetcher	(	MYSQL_BIN_LOG *	log_ptr,
		Mt_key	mt_key_wait = 0,
		Cv_key	cv_key_wait = 0,
		Mt_key	mt_key_file_move = 0,
		Cv_key	cv_key_file_move = 0,
		Th_key	key_th_prefetcher = 0,
		Mem_key	key_memory = 0
	)

Parameters

log_ptr	The MYSQL_BIN_LOG object we use to know from where we need to read data
mt_key_wait	Instrumentation key for internal mutex (wait for work)
cv_key_wait	Instrumentation key for internal cv (wait for work)
mt_key_file_move	Instrumentation key for internal mutex (wait for file rotation)
cv_key_file_move	Instrumentation key for internal cv (wait for file rotation)
key_th_prefetcher	Instrumentation key for prefetcher thread
key_memory	Instrumentation key for allocated memory

~Log_prefetcher()

mysql::csa::Log_prefetcher::~Log_prefetcher ( )

virtual

Log_prefetcher() [2/3]

mysql::csa::Log_prefetcher::Log_prefetcher ( const Log_prefetcher &  )

delete

Log_prefetcher() [3/3]

mysql::csa::Log_prefetcher::Log_prefetcher ( Log_prefetcher &&  )

delete

Member Function Documentation

dequeue()

template<typename P >

std::optional< Log_prefetcher::Elem_type > mysql::csa::Log_prefetcher::dequeue

(

P &&

wait_predicate

)

Consumes the next batch of data.

Returns

The next batch of data or empty object in case stopped in the process

ensure_file_done()

void mysql::csa::Log_prefetcher::ensure_file_done ( const std::string &  file_name )

private

If currently rotated file is file_name, waits until rotate is done.

If prefetcher moves a different file, function returns immediately.

Parameters

file_name

File expected to be moved from.

get_current_file()

std::string mysql::csa::Log_prefetcher::get_current_file ( ) const

private

Obtains the current file under the lock.

Returns

File currently being prefetched

is_error()

bool mysql::csa::Log_prefetcher::is_error

(

)

const

Returns true if error occurred.

Returns

True in case error occurred, false otherwise

is_stopped()

bool mysql::csa::Log_prefetcher::is_stopped

(

)

const

Checks whether prefetcher is stopped.

True if stop was requested, false otherwise

is_waiting_for_next_file()

bool mysql::csa::Log_prefetcher::is_waiting_for_next_file	(	const std::string &	prev_file,
		const std::string &	next_file
	)

Checks whether we reached inactive file after the prev_file This function takes previous file to check if prefetcher finished rotation on this file and consecutive file to check if prefetcher opened it and stopped working until file becomes inactive.

Parameters

prev_file	File we know we read all data from (got eof in the last consumed batch) to check if prefetcher finished rotation on this file
next_file	Consecutive file we want to open the stream on.

Return values

true	Prefetcher waits for file_name to become inactive
false	Prefetcher does not wait for 'file_name'

open()

bool mysql::csa::Log_prefetcher::open

(

const char *

file_name

)

Tries to open a new, gifen file.

Prefetcher must stop on cv before calling this function (see is_waiting_for_next_file).

Returns

True in case file is inactive and prefetcher will work on it, false otherwise

Parameters

file_name

Starts reading from this file path

open_file()

bool mysql::csa::Log_prefetcher::open_file ( bool  next_file )

private

Opens new or next file for reading.

Parameters

next_file

True if reading file after the current one

Returns

True in case file is inactive and prefetcher will work on it, false otherwise

operator=() [1/2]

Log_prefetcher & mysql::csa::Log_prefetcher::operator= ( const Log_prefetcher &  )

delete

operator=() [2/2]

Log_prefetcher & mysql::csa::Log_prefetcher::operator= ( Log_prefetcher &&  )

delete

run_prefetch_thread()

void mysql::csa::Log_prefetcher::run_prefetch_thread ( )

private

Runs a thread that prefetch data from the relay log.

get file name after wait (updated by open)

set_error()

void mysql::csa::Log_prefetcher::set_error ( const char *  msg )

private

Sets internal error (under lock) and internal error message.

Later on, stops the prefetcher. This function will report any error to error log

Parameters

msg	Error message

start_prefetcher()

void mysql::csa::Log_prefetcher::start_prefetcher

(

)

Starts asynchronous thread prefetching data.

stop()

void mysql::csa::Log_prefetcher::stop

(

)

Gracefully stops the prefetcher.

Blocks waiting for notification that prefetcher thread is done

update_current_file()

void mysql::csa::Log_prefetcher::update_current_file ( const char *  file_name, bool  active  )

private

Updates the current file and its activity under the lock.

Parameters

file_name	Update the current file name to this file
active	True if file_name is an active log

Member Data Documentation

m_allocator

Memory_allocator mysql::csa::Log_prefetcher::m_allocator

private

Memory_resource to handle all allocations.

m_batch_size

std::size_t mysql::csa::Log_prefetcher::m_batch_size {tune::prefetcher_batch_size}

private

Default batch size ~16MB.

m_bytes_fetched

std::atomic<std::size_t> mysql::csa::Log_prefetcher::m_bytes_fetched {0}

private

This way we track the actual number of bytes that are cached in the m_cache.

m_cache

Queue_type mysql::csa::Log_prefetcher::m_cache

private

Queue into which prefetcher puts data batches read from a raw binary file.

m_current_file_length

std::size_t mysql::csa::Log_prefetcher::m_current_file_length {0}

private

Length of the currently opened file.

m_current_log_name

std::string mysql::csa::Log_prefetcher::m_current_log_name {""}

private

Currently processed log file Protected by m_mt_prefetcher.

m_current_offset

std::size_t mysql::csa::Log_prefetcher::m_current_offset {0}

private

Current file offset.

m_cv_move_file

mysql::concurrency::Condition_variable mysql::csa::Log_prefetcher::m_cv_move_file

private

CV to notify the watcher that prefetcher finished moving from m_move_file to the next file (see ensure_file_done)

m_cv_prefetcher

mysql::concurrency::Condition_variable mysql::csa::Log_prefetcher::m_cv_prefetcher

private

Cv used by the scheduler main thread to wait on, when no task is available or tasks in m_task queue are not read to execute.

m_end

std::atomic<bool> mysql::csa::Log_prefetcher::m_end {false}

private

Notification atomic for end of execution.

m_error_message

std::string mysql::csa::Log_prefetcher::m_error_message {""}

private

Error message if any. Protected by m_mt_prefetcher.

m_is_error

std::atomic<bool> mysql::csa::Log_prefetcher::m_is_error {false}

private

Becomes true in case error has been encountered.

m_istream

std::ifstream mysql::csa::Log_prefetcher::m_istream

private

Stream to read from.

m_key_th_prefetcher

Th_key mysql::csa::Log_prefetcher::m_key_th_prefetcher {0}

private

Key for prefetcher thread.

m_log

MYSQL_BIN_LOG* mysql::csa::Log_prefetcher::m_log

private

The MYSQL_BIN_LOG object we use to know from where we need to read data.

m_log_active

bool mysql::csa::Log_prefetcher::m_log_active {true}

private

True in case reading from the active relay log file (used currently by the receiver thread), protected with m_mt_prefetcher.

Prefetcher can read only from inactive files Protected by m_mt_prefetcher

m_move_file

std::string mysql::csa::Log_prefetcher::m_move_file {""}

private

Is set to a value if prefetcher initiated file move procedure Watcher may wait until prefetcher moves to the next file by using the m_cv_move_file notification cv and associated m_mt_move_file.

m_mt_move_file

mysql::concurrency::Mutex mysql::csa::Log_prefetcher::m_mt_move_file

mutableprivate

Mutex protecting access to m_move_file.

m_mt_prefetcher

mysql::concurrency::Mutex mysql::csa::Log_prefetcher::m_mt_prefetcher

mutableprivate

Mutex protecting access to m_tasks.

m_prefetcher

Thread_type mysql::csa::Log_prefetcher::m_prefetcher

private

Thread that runs prefetching.

m_stopped

std::atomic<bool> mysql::csa::Log_prefetcher::m_stopped {false}

private

Internal stop flag set to true when stop of the thread is requested.

max_bytes_fetched

constexpr std::size_t mysql::csa::Log_prefetcher::max_bytes_fetched {1073741824}

staticconstexprprivate

The maximum number of bytes we can prefetch.

---

The documentation for this class was generated from the following files:

• sql/changestreams/apply/storage/relay_log/log_prefetcher.h
• sql/changestreams/apply/storage/relay_log/log_prefetcher.cpp
• sql/changestreams/apply/storage/relay_log/log_prefetcher_impl.hpp