This flow is listening to data from a TCP connection, segments the incoming data into TcpMessages based on the incoming size, and forwards packets containing the TCP message as string payload. More...

#include <IfTcpImpl.hxx>

Inheritance diagram for openlcb::FdToTcpParser:

Public Member Functions
	FdToTcpParser (FdHubPortService s, HubPortInterface dst, HubPortInterface *skipMember)
	Constructor.

void	shutdown ()
	Stops listening and terminates the flow.

Public Member Functions inherited from StateFlowBase
void	run () override
	Callback from the executor.

void	notify () override
	Wakeup call arrived.

Service *	service ()
	Return a pointer to the service I am bound to.

Public Member Functions inherited from Executable
void	test_deletion ()

Public Member Functions inherited from QMember
void	init ()
	Initiailize a QMember, in place of a public placement construction.

Private Member Functions
FdHubPortService *	device ()

Action	start_msg ()
	In this state the internal buffer (bufOfs_) points at the beginning of a message.

Action	parse_bytes ()
	Intermediate state where we are.

Action	read_more_bytes ()
	Helper state to call the stateflow kernel to read bytes from the fd into the internal buffer.

Action	read_done ()
	Callback state when the kernel read is completed.

Action	send_entry ()
	Sends an assembled message (1) to the destination flow.

Action	have_alloc_msg ()
	Sends an assembled message (2) to the destination flow.

void	notify_barrier ()
	Calls into the parent flow's barrier notify, but makes sure to only do this once in the lifetime of *this.

Private Attributes
uint8_t	buffer_ [READ_BUFFER_SIZE]
	Temporary buffer to read into from the FD.

uint8_t	barrierOwned_ {1}
	true iff pending parent->barrier_.notify()

uint16_t	bufEnd_
	Offset of the end in the read buffer.

uint16_t	bufOfs_
	First active byte (offset of the beginning) in the read buffer.

int	expectedLen_
	How many bytes we think the current message will be.

string	msg_
	Assembly buffer. Holds the captured message so far.

HubPortInterface *	dst_
	Where to send parsed messages to.

HubPortInterface *	skipMember_
	Parsed messages will be initialized to this skipMember_.

StateFlowSelectHelper	helper_ {this}

Static Private Attributes
static constexpr unsigned	READ_BUFFER_SIZE = 300
	We attempt to read this many bytes in one go from the FD.

static constexpr unsigned	DEFAULT_PACKET_SIZE
	If we have to guess at the size of the packet, we start by allocating this many bytes.

static constexpr unsigned	MIN_SIZE_GUESS = TcpDefs::HDR_SIZE_END

static constexpr unsigned	READ_PRIO = Selectable::MAX_PRIO
	What priority to use for reads from fds.

Additional Inherited Members
Static Public Member Functions inherited from StateFlowBase
template<class T , typename... Args>
static void	invoke_subflow_and_ignore_result (FlowInterface< Buffer< T > > *target_flow, Args &&... args)
	Calls a helper flow to perform some actions.

Protected Types inherited from StateFlowBase
typedef Action(StateFlowBase::*	Callback) ()
	State Flow callback prototype.

Protected Member Functions inherited from StateFlowBase
	StateFlowBase (Service *service)
	Constructor.

	~StateFlowBase ()
	Destructor.

void	reset_flow (Callback c)
	Resets the flow to the specified state.

bool	is_state (Callback c)

bool	is_terminated ()

void	start_flow (Callback c)
	Resets the flow to the specified state and starts it.

Action	again ()
	Call the current state again via call_immediately.

Action	exit ()
	Terminate current StateFlow activity.

Action	delete_this ()
	Terminates the flow and deletes *this.

Action	set_terminated ()
	Sets the flow to terminated state.

Action	call_immediately (Callback c)
	Imediately call the next state upon return.

Action	wait ()
	Wait for an asynchronous call.

Action	wait_and_call (Callback c)
	Wait for resource to become available before proceeding to next state.

template<class T >
Action	allocate_and_call (FlowInterface< Buffer< T > > target_flow, Callback c, Pool pool=nullptr)
	Allocates a buffer from a pool and proceed to the next state when allocation is successful.

Action	allocate_and_call (Callback c, QAsync *queue)
	Allocates an entry from an asynchronous queue, and transitions to a state once the allocation is complete.

template<class T >
Buffer< T > *	full_allocation_result (FlowInterface< Buffer< T > > *target_flow)
	Takes the result of the asynchronous allocation without resetting the object.

template<class T >
T *	full_allocation_result (TypedQAsync< T > *queue)
	Takes the result of the asynchronous allocation without resetting the object.

template<class T >
void	cast_allocation_result (T **member)
	Takes the result of the asynchronous allocation without resetting the object.

template<class T >
Buffer< T > *	get_allocation_result (FlowInterface< Buffer< T > > *target_flow)
	Takes the result of the asynchronous allocation.

Action	yield_and_call (Callback c)
	Place the current flow to the back of the executor, and transition to a new state after we get the CPU again.

Action	yield ()
	Place the current flow to the back of the executor, and re-try the current state after we get the CPU again.

Action	sleep_and_call (::Timer *timer, long long timeout_nsec, Callback c)
	Suspends execution of this control flow for a specified time.

template<class T , typename... Args>
Action	invoke_subflow_and_wait (FlowInterface< Buffer< T > > *target_flow, Callback c, Args &&... args)
	Calls a helper flow to perform some actions.

Action	read_repeated (StateFlowSelectHelper helper, int fd, void buf, size_t size, Callback c, unsigned priority=Selectable::MAX_PRIO)
	Blocks until size bytes are read and then invokes the next state.

Action	read_single (StateFlowSelectHelper helper, int fd, void buf, size_t size, Callback c, unsigned priority=Selectable::MAX_PRIO)
	Attempts to read at most size_t bytes, and blocks the caller until at least one byte is read.

Action	read_nonblocking (StateFlowSelectHelper helper, int fd, void buf, size_t size, Callback c, unsigned priority=Selectable::MAX_PRIO)
	Attempts to read at most size bytes, and then invokes the next state, even if only zero bytes are available right now.

Action	read_repeated_with_timeout (StateFlowTimedSelectHelper helper, long long timeout_nsec, int fd, void buf, size_t size, Callback c, unsigned priority=Selectable::MAX_PRIO)
	Blocks until size bytes are read, or a timeout expires.

Action	internal_try_read ()
	Implementation state that gets repeatedly called upon every wakeup and tries to make progress on reading.

Action	write_repeated (StateFlowSelectHelper helper, int fd, const void buf, size_t size, Callback c, unsigned priority=Selectable::MAX_PRIO)
	Writes some data into a file descriptor, repeating the operation as necessary until all bytes are written.

Action	internal_try_write ()
	Implementation state that gets repeatedly called upon every wakeup and tries to make progress on writing.

Protected Member Functions inherited from QMember
	QMember ()
	Constructor.

	~QMember ()
	Destructor.

Protected Attributes inherited from QMember
QMember *	next
	pointer to the next member in the queue

Detailed Description

This flow is listening to data from a TCP connection, segments the incoming data into TcpMessages based on the incoming size, and forwards packets containing the TCP message as string payload.

Definition at line 380 of file IfTcpImpl.hxx.

Constructor & Destructor Documentation

◆ FdToTcpParser()

openlcb::FdToTcpParser::FdToTcpParser	(	FdHubPortService *	s,
		HubPortInterface *	dst,
		HubPortInterface *	skipMember
	)

inline

Constructor.

Parameters

s	the parent (owning) service that holds the different flows together.
dst	where to forward the segmented packets.
skipMember	all forwarded messages will have their skipMember set to this value. Usually the output port.

Definition at line 389 of file IfTcpImpl.hxx.

Member Function Documentation

◆ device()

FdHubPortService * openlcb::FdToTcpParser::device ( )

inlineprivate

Returns: the typed service

Definition at line 416 of file IfTcpImpl.hxx.

◆ have_alloc_msg()

Action openlcb::FdToTcpParser::have_alloc_msg ( )

inlineprivate

Sends an assembled message (2) to the destination flow.

Returns: next state.

Todo:: (balazs.racz): there should be some form of throttling here, ot not read more bytes from the tcp socket than how much RAM we have available.

Definition at line 528 of file IfTcpImpl.hxx.

◆ notify_barrier()

void openlcb::FdToTcpParser::notify_barrier ( )

inlineprivate

Calls into the parent flow's barrier notify, but makes sure to only do this once in the lifetime of *this.

Definition at line 543 of file IfTcpImpl.hxx.

◆ parse_bytes()

Action openlcb::FdToTcpParser::parse_bytes ( )

inlineprivate

Intermediate state where we are.

figuring out what is the length of the message we are parsing
allocating the output buffer
copying bytes of the message from the internal buffer into the output buffer
Returns
next state

Definition at line 437 of file IfTcpImpl.hxx.

◆ read_done()

Action openlcb::FdToTcpParser::read_done ( )

inlineprivate

Callback state when the kernel read is completed.

Returns: next state

Definition at line 506 of file IfTcpImpl.hxx.

◆ read_more_bytes()

Action openlcb::FdToTcpParser::read_more_bytes ( )

inlineprivate

Helper state to call the stateflow kernel to read bytes from the fd into the internal buffer.

Returns: next state

Definition at line 493 of file IfTcpImpl.hxx.

◆ send_entry()

Action openlcb::FdToTcpParser::send_entry ( )

inlineprivate

Sends an assembled message (1) to the destination flow.

Returns: next state.

Definition at line 521 of file IfTcpImpl.hxx.

◆ shutdown()

void openlcb::FdToTcpParser::shutdown ( )

inline

Stops listening and terminates the flow.

Definition at line 402 of file IfTcpImpl.hxx.

◆ start_msg()

Action openlcb::FdToTcpParser::start_msg ( )

inlineprivate

In this state the internal buffer (bufOfs_) points at the beginning of a message.

Returns: next state.

Definition at line 424 of file IfTcpImpl.hxx.

Member Data Documentation

◆ barrierOwned_

uint8_t openlcb::FdToTcpParser::barrierOwned_ {1}

private

true iff pending parent->barrier_.notify()

Definition at line 565 of file IfTcpImpl.hxx.

◆ bufEnd_

uint16_t openlcb::FdToTcpParser::bufEnd_

private

Offset of the end in the read buffer.

Definition at line 567 of file IfTcpImpl.hxx.

◆ buffer_

uint8_t openlcb::FdToTcpParser::buffer_[READ_BUFFER_SIZE]

private

Temporary buffer to read into from the FD.

Definition at line 563 of file IfTcpImpl.hxx.

◆ bufOfs_

uint16_t openlcb::FdToTcpParser::bufOfs_

private

First active byte (offset of the beginning) in the read buffer.

Definition at line 569 of file IfTcpImpl.hxx.

◆ DEFAULT_PACKET_SIZE

constexpr unsigned openlcb::FdToTcpParser::DEFAULT_PACKET_SIZE

staticconstexprprivate

Initial value:

=

TcpDefs::MIN_ADR_MESSAGE_SIZE + 16

openlcb::TcpDefs::MIN_ADR_MESSAGE_SIZE

@ MIN_ADR_MESSAGE_SIZE

Minimum length of a valid message that has an addressed MTI.

Definition IfTcpImpl.hxx:233

If we have to guess at the size of the packet, we start by allocating this many bytes.

Definition at line 556 of file IfTcpImpl.hxx.

◆ dst_

HubPortInterface* openlcb::FdToTcpParser::dst_

private

Where to send parsed messages to.

Definition at line 576 of file IfTcpImpl.hxx.

◆ expectedLen_

int openlcb::FdToTcpParser::expectedLen_

private

How many bytes we think the current message will be.

-1 if we don't know yet.

Definition at line 572 of file IfTcpImpl.hxx.

◆ helper_

StateFlowSelectHelper openlcb::FdToTcpParser::helper_ {this}

private

Definition at line 579 of file IfTcpImpl.hxx.

◆ MIN_SIZE_GUESS

constexpr unsigned openlcb::FdToTcpParser::MIN_SIZE_GUESS = TcpDefs::HDR_SIZE_END

staticconstexprprivate

Definition at line 558 of file IfTcpImpl.hxx.

◆ msg_

string openlcb::FdToTcpParser::msg_

private

Assembly buffer. Holds the captured message so far.

Definition at line 574 of file IfTcpImpl.hxx.

◆ READ_BUFFER_SIZE

constexpr unsigned openlcb::FdToTcpParser::READ_BUFFER_SIZE = 300

staticconstexprprivate

We attempt to read this many bytes in one go from the FD.

Definition at line 553 of file IfTcpImpl.hxx.

◆ READ_PRIO

constexpr unsigned openlcb::FdToTcpParser::READ_PRIO = Selectable::MAX_PRIO

staticconstexprprivate

What priority to use for reads from fds.

Definition at line 560 of file IfTcpImpl.hxx.

◆ skipMember_

HubPortInterface* openlcb::FdToTcpParser::skipMember_

private

Parsed messages will be initialized to this skipMember_.

Definition at line 578 of file IfTcpImpl.hxx.

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

IfTcpImpl.hxx

Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ FdToTcpParser()

Member Function Documentation

◆ device()

◆ have_alloc_msg()

◆ notify_barrier()

◆ parse_bytes()

◆ read_done()

◆ read_more_bytes()

◆ send_entry()

◆ shutdown()

◆ start_msg()

Member Data Documentation

◆ barrierOwned_

◆ bufEnd_

◆ buffer_

◆ bufOfs_

◆ DEFAULT_PACKET_SIZE

◆ dst_

◆ expectedLen_

◆ helper_

◆ MIN_SIZE_GUESS

◆ msg_

◆ READ_BUFFER_SIZE

◆ READ_PRIO

◆ skipMember_