The Event Reader / Controller class. More...

#include <event_reader_controller.h>

Inheritance diagram for mysql::csa::Event_reader_controller:

Public Member Functions
	Event_reader_controller (Relay_log_info *rli, Log_prefetcher_sptr prefetcher)

bool	open ()
	Opens the first relay log. More...

void	close ()
	Closes readers, stops prefetcher, clears internal state including error state. More...

std::optional< Event_file_metadata >	read_next (unsigned int return_timeout_ms, Reader_controller_read_type read_type)
	Fetches next: event data, event metadata, event metadata plus raw payload, depending on read type. More...

bool	concurrent_purge (const std::string &log_filename) override
	Register log_filename as a log ready to be purged. More...

const std::string &	get_file_name () const
	Obtain currently processed file. More...

bool	is_error () const
	Check if reader has an error. More...

void	stop ()
	Stop the reader. More...

bool	is_stopped () const
	Check if reader is stopped. More...

Public Member Functions inherited from mysql::csa::Log_purge_controller
virtual	~Log_purge_controller ()

Private Member Functions
bool	move_to_log (bool next_log=true, my_off_t offset=0)
	When next_log is true, opens the next file. More...

bool	move_to_next_log ()
	Moves to the next log file. More...

bool	purge_applied_logs (const char *to_log)
	Purge relay log files up to to_log. More...

bool	wait_for_new_event (unsigned int return_timeout_ms)
	In case we read from active file, we use this function to passively wait for new event. More...

bool	check_cache_truncated ()
	If cache is truncated, reopen the reader to avoid reading trash data. More...

bool	set_error (const char *msg)
	Sets internal error to msg. More...

bool	is_data_available ()
	Checks whether there is data in the current file. More...

bool	wait_data_ready (unsigned int return_timeout_ms)
	Waits until data is available, stopped or return_timeout_ms is reached. More...

void	choose_reader (bool next_log, const std::string &prev_file)
	Implements internal logic to choose between active and inactive file reading. More...

Private Attributes
bool	m_is_error {false}
	Flag which is true in case any error occurred. More...

std::string	m_error_msg {""}
	Error message if any. More...

Relay_log_info *	m_rli {nullptr}
	non-owning RLI object pointer More...

Log_prefetcher_sptr	m_prefetcher
	Relay log prefetcher. More...

Relaylog_file_reader	m_active_reader
	Reader for active files. More...

Prefetched_relaylog_reader	m_inactive_reader
	Reader for inactive files. More...

IBasic_binlog_file_reader *	m_current_reader {nullptr}
	Non-owning pointer to currently used reader (m_inactive_reader or m_active_reader) More...

std::string	m_file_name
	Stores the current file as obtaining from stream. More...

std::unordered_set< std::string >	m_logs_to_purge
	Here we keep the list of logs to be purged (in case later logs are applied before), protected with m_rli->data_lock. More...

bool	m_using_prefetcher {false}
	Flag indicating whether currently opened log file is active. More...

bool	m_active_file_reading {false}
	This flag is true in case we are using active file reader and reading from the active relay log file. More...

bool	m_enable_prefetcher {tune::prefetcher_enable}
	Variable to decide on whether we want to run prefetcher. More...

std::atomic< bool >	m_is_stopped {false}
	Stop flag. More...

Detailed Description

The Event Reader / Controller class.

This class uses the low level reader to read consecutive events from the relay log. Also, it exposed functions to remove consumed relay log files. Class provides methods to:

initialize internal data (open)
deinitialize internal data (close)
fetch the next event from the relay log (read_next)
fetch the next event metadata from the relay log (read_next)
register consumed relay log for purge and purge registered consecutive logs (concurrent_purge), which implements the Log_purge_controller interface

This class may be seen as 'Rpl_applier_reader' created for the CSA. Relay_log_decoder works with prefetched relay log files or stream build on top of the IO_CACHE. The first one is used when reading from inactive files. When reading from active files, Event_reader_controller utilizes the IO_CACHE implementation, since it allows the applier to fetch data from cache instead of fetching data from disk. Data will be consumed from cache provided that the applier keeps up with Receiver thread and cache is 'large' enough to keep the recent data. When reading from inactive files, the 'Event_reader_controller' utilizes the prefetcher class utility to fetch data. When reading from active files, the Event Reader / Controller needs to rely on the 'MYSQL_BIN_LOG' synchronization primitives and supply the implementation of passive waiting for data (is_data_available, wait_data_ready). Following the legacy design, streams implemented on top of prefetcher are allowed to move to the next file upon the caller request. Therefore, the 'Event_reader_controller' is responsible for checking file boundaries and reopening the streams on top of new files when needed.

Constructor & Destructor Documentation

◆ Event_reader_controller()

mysql::csa::Event_reader_controller::Event_reader_controller	(	Relay_log_info *	rli,
		Log_prefetcher_sptr	prefetcher
	)

Member Function Documentation

◆ check_cache_truncated()

bool mysql::csa::Event_reader_controller::check_cache_truncated ( )

private

If cache is truncated, reopen the reader to avoid reading trash data.

Hack function that solves problem of relay log IO_CACHE truncation on active relay log files

Returns: true if failure when reopening the file, false on success

See also: Rpl_applier_reader::reopen_log_reader_if_needed

◆ choose_reader()

void mysql::csa::Event_reader_controller::choose_reader	(	bool	next_log,
		const std::string &	prev_file
	)

private

Implements internal logic to choose between active and inactive file reading.

Parameters

next_log	True if open was called on the next log
prev_file	Previous relay log file processed

◆ close()

void mysql::csa::Event_reader_controller::close ( )

Closes readers, stops prefetcher, clears internal state including error state.

◆ concurrent_purge()

bool mysql::csa::Event_reader_controller::concurrent_purge ( const std::string & log_filename )

overridevirtual

Register log_filename as a log ready to be purged.

This function will purge registered logs if they are in order according to the index file content.

Parameters

[in] log_filename Registeres this log as a log ready to be purged. If registered logs for purging are in order, purges up to log_filename, included.

Implements mysql::csa::Log_purge_controller.

◆ get_file_name()

const std::string & mysql::csa::Event_reader_controller::get_file_name ( ) const

inline

Obtain currently processed file.

Returns: Current file name

◆ is_data_available()

bool mysql::csa::Event_reader_controller::is_data_available ( )

private

Checks whether there is data in the current file.

Return values

true	Data is available
false	No data

For prefetcher, we don't wait for data here, because passive waiting for data is implemented in the prefetcher itself. For active relay log file reading, we need implement passive waiting within the Event_reader_controller and using the MYSQL_BIN_LOG synchronization primitives.

◆ is_error()

bool mysql::csa::Event_reader_controller::is_error ( ) const

inline

Check if reader has an error.

Returns: True in case an error occurred, false otherwise

◆ is_stopped()

bool mysql::csa::Event_reader_controller::is_stopped ( ) const

Check if reader is stopped.

Returns: True when stop was requested; false otherwise

◆ move_to_log()

bool mysql::csa::Event_reader_controller::move_to_log	(	bool	next_log = `true`,
		my_off_t	offset = `0`
	)

private

When next_log is true, opens the next file.

Otherwise, opens the current file

Parameters

next_log	When true, moves to next log after the current
offset	Requested file offset

Return values

false	Success
true	failure

◆ move_to_next_log()

bool mysql::csa::Event_reader_controller::move_to_next_log ( )

private

Moves to the next log file.

Return values

false	Success
true	failure

◆ open()

bool mysql::csa::Event_reader_controller::open ( )

Opens the first relay log.

Return values

false	Success
true	failure

◆ purge_applied_logs()

bool mysql::csa::Event_reader_controller::purge_applied_logs ( const char * to_log )

private

Purge relay log files up to to_log.

Parameters

to_log Purge logs up to this log, exclusively

Return values

false	Success
true	Error

◆ read_next()

std::optional< Event_file_metadata > mysql::csa::Event_reader_controller::read_next	(	unsigned int	return_timeout_ms,
		Reader_controller_read_type	read_type
	)

Fetches next: event data, event metadata, event metadata plus raw payload, depending on read type.

Parameters

return_timeout_ms	If specified, will wait up to `return_timeout_ms` miliseconds. If timeout is reached, returns true.
read_type	Type of read: full event, event metadata or event metadata and raw payload

Returns: Next: event data, event metadata, event metadata + raw payload depending on read type. Empty object if an error occurred

◆ set_error()

bool mysql::csa::Event_reader_controller::set_error ( const char * msg )

private

Sets internal error to msg.

Parameters

msg	Error message

Returns: Error state : true

◆ stop()

void mysql::csa::Event_reader_controller::stop ( )

Stop the reader.

◆ wait_data_ready()

bool mysql::csa::Event_reader_controller::wait_data_ready ( unsigned int return_timeout_ms )

private

Waits until data is available, stopped or return_timeout_ms is reached.

Parameters

return_timeout_ms Timeout after which we will exit from waiting

Return values

true	Data is available
false	Data is not ready / stopped / error

◆ wait_for_new_event()

bool mysql::csa::Event_reader_controller::wait_for_new_event ( unsigned int return_timeout_ms )

private

In case we read from active file, we use this function to passively wait for new event.

Parameters

return_timeout_ms Timeout after which we will exit from waiting

Return values

false	New event is available
true	Timeout occurred

Member Data Documentation

◆ m_active_file_reading

bool mysql::csa::Event_reader_controller::m_active_file_reading {false}

private

This flag is true in case we are using active file reader and reading from the active relay log file.

◆ m_active_reader

Relaylog_file_reader mysql::csa::Event_reader_controller::m_active_reader

private

Reader for active files.

◆ m_current_reader

IBasic_binlog_file_reader* mysql::csa::Event_reader_controller::m_current_reader {nullptr}

private

Non-owning pointer to currently used reader (m_inactive_reader or m_active_reader)

◆ m_enable_prefetcher

bool mysql::csa::Event_reader_controller::m_enable_prefetcher {tune::prefetcher_enable}

private

Variable to decide on whether we want to run prefetcher.

◆ m_error_msg

std::string mysql::csa::Event_reader_controller::m_error_msg {""}

private

Error message if any.

◆ m_file_name

std::string mysql::csa::Event_reader_controller::m_file_name

private

Stores the current file as obtaining from stream.

◆ m_inactive_reader

Prefetched_relaylog_reader mysql::csa::Event_reader_controller::m_inactive_reader

private

Reader for inactive files.

◆ m_is_error

bool mysql::csa::Event_reader_controller::m_is_error {false}

private

Flag which is true in case any error occurred.

Otherwise, it is set to false.

◆ m_is_stopped

std::atomic<bool> mysql::csa::Event_reader_controller::m_is_stopped {false}

private

Stop flag.

◆ m_logs_to_purge

std::unordered_set<std::string> mysql::csa::Event_reader_controller::m_logs_to_purge

private

Here we keep the list of logs to be purged (in case later logs are applied before), protected with m_rli->data_lock.

◆ m_prefetcher

Log_prefetcher_sptr mysql::csa::Event_reader_controller::m_prefetcher

private

Relay log prefetcher.

◆ m_rli

Relay_log_info* mysql::csa::Event_reader_controller::m_rli {nullptr}

private

non-owning RLI object pointer

◆ m_using_prefetcher

bool mysql::csa::Event_reader_controller::m_using_prefetcher {false}

private

Flag indicating whether currently opened log file is active.

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

sql/changestreams/apply/storage/relay_log/event_reader_controller.h
sql/changestreams/apply/storage/relay_log/event_reader_controller.cpp

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ Event_reader_controller()

Member Function Documentation

◆ check_cache_truncated()

◆ choose_reader()

◆ close()

◆ concurrent_purge()

◆ get_file_name()

◆ is_data_available()

◆ is_error()

◆ is_stopped()

◆ move_to_log()

◆ move_to_next_log()

◆ open()

◆ purge_applied_logs()

◆ read_next()

◆ set_error()

◆ stop()

◆ wait_data_ready()

◆ wait_for_new_event()

Member Data Documentation

◆ m_active_file_reading

◆ m_active_reader

◆ m_current_reader

◆ m_enable_prefetcher

◆ m_error_msg

◆ m_file_name

◆ m_inactive_reader

◆ m_is_error

◆ m_is_stopped

◆ m_logs_to_purge

◆ m_prefetcher

◆ m_rli

◆ m_using_prefetcher