State flow that reads the FD and sends the read data to the direct hub. More...

Inheritance diagram for DirectHubPortSelect::DirectHubReadFlow:

Public Member Functions
	DirectHubReadFlow (DirectHubPortSelect *parent, std::unique_ptr< MessageSegmenter > segmenter)

void	start ()
	Starts the current flow.

void	read_shutdown ()
	Requests the read port to shut down.

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
Action	alloc_for_read ()
	Root of the read flow.

Action	barrier_allocated ()
	Intermediate step if asynchronous allocation was necessary for the read barrier.

Action	get_read_buffer ()
	Invoked when we have a bufferNotifiable_ from the barrier pool.

Action	do_some_read ()

Action	read_done ()

Action	eval_segment ()
	Checks the segmenter output; if it indicates a complete message, clears the segmenter and sends off the message.

Action	call_head_segmenter ()
	Clears the segmenter and starts segmenting from the beginning of the buf_.

Action	incomplete_message ()
	Called when the segmenter says that we need to read more bytes to complete the current message.

Action	send_prefix ()
	Called to send a given prefix segment to the hub.

Action	send_callback ()
	This is the callback state that is invoked inline by the hub.

Action	send_done ()

Private Attributes
LinkedDataBufferPtr	buf_
	Current buffer that we are filling.

BarrierNotifiable *	bufferNotifiable_
	Barrier notifiable to keep track of the buffer's contents.

ssize_t	segmentSize_
	Output of the last segmenter call.

uint16_t	inlineCall_: 1
	1 if we got the send callback inline from the read_done.

uint16_t	sendComplete_: 1
	1 if the run callback actually happened inline.

AsyncNotifiableBlock	pendingLimiterPool_
	Pool of BarrierNotifiables that limit the amount of inflight bytes we have.

StateFlowSelectHelper	helper_ {this}
	Helper object for Select.

DirectHubPortSelect *	parent_
	Pointer to the owninng port.

std::unique_ptr< MessageSegmenter >	segmenter_
	Implementation (and state) of the business logic that segments incoming bytes into messages that shall be given to the hub.

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

State flow that reads the FD and sends the read data to the direct hub.

Definition at line 267 of file DirectHub.cxx.

Constructor & Destructor Documentation

◆ DirectHubReadFlow()

DirectHubPortSelect::DirectHubReadFlow::DirectHubReadFlow	(	DirectHubPortSelect *	parent,
		std::unique_ptr< MessageSegmenter >	segmenter
	)

inline

Definition at line 270 of file DirectHub.cxx.

Member Function Documentation

◆ alloc_for_read()

Action DirectHubPortSelect::DirectHubReadFlow::alloc_for_read ( )

inlineprivate

Root of the read flow.

Starts with getting the barrier notifiable, either synchronously if one is available, or asynchronously.

Definition at line 308 of file DirectHub.cxx.

◆ barrier_allocated()

Action DirectHubPortSelect::DirectHubReadFlow::barrier_allocated ( )

inlineprivate

Intermediate step if asynchronous allocation was necessary for the read barrier.

Definition at line 325 of file DirectHub.cxx.

◆ call_head_segmenter()

Action DirectHubPortSelect::DirectHubReadFlow::call_head_segmenter ( )

inlineprivate

Clears the segmenter and starts segmenting from the beginning of the buf_.

Definition at line 411 of file DirectHub.cxx.

◆ do_some_read()

Action DirectHubPortSelect::DirectHubReadFlow::do_some_read ( )

inlineprivate

Definition at line 361 of file DirectHub.cxx.

◆ eval_segment()

Action DirectHubPortSelect::DirectHubReadFlow::eval_segment ( )

inlineprivate

Checks the segmenter output; if it indicates a complete message, clears the segmenter and sends off the message.

Definition at line 395 of file DirectHub.cxx.

◆ get_read_buffer()

Action DirectHubPortSelect::DirectHubReadFlow::get_read_buffer ( )

inlineprivate

Invoked when we have a bufferNotifiable_ from the barrier pool.

Definition at line 335 of file DirectHub.cxx.

◆ incomplete_message()

Action DirectHubPortSelect::DirectHubReadFlow::incomplete_message ( )

inlineprivate

Called when the segmenter says that we need to read more bytes to complete the current message.

Definition at line 424 of file DirectHub.cxx.

◆ read_done()

Action DirectHubPortSelect::DirectHubReadFlow::read_done ( )

inlineprivate

Definition at line 375 of file DirectHub.cxx.

◆ read_shutdown()

void DirectHubPortSelect::DirectHubReadFlow::read_shutdown ( )

inline

Requests the read port to shut down.

Must be called on the main executor. Causes the flow to notify the parent via the read_flow_exit() function then terminate, either inline or not.

Todo:: We should first clean up the async notifiable block and only signal the exit afterwards.

Definition at line 288 of file DirectHub.cxx.

◆ send_callback()

Action DirectHubPortSelect::DirectHubReadFlow::send_callback ( )

inlineprivate

This is the callback state that is invoked inline by the hub.

Since the hub invokes this->run(), a standard StateFlow will execute whatever state is current. We have set STATE(send_callback) as the current state above, hence the code continues in this function.

Definition at line 456 of file DirectHub.cxx.

◆ send_done()

Action DirectHubPortSelect::DirectHubReadFlow::send_done ( )

inlineprivate

Todo:: consider not resetting here, but allowing an empty but linked DataBuffer* start the next chain.

Definition at line 481 of file DirectHub.cxx.

◆ send_prefix()

Action DirectHubPortSelect::DirectHubReadFlow::send_prefix ( )

inlineprivate

Called to send a given prefix segment to the hub.

segmentSize_ is filled in before.

Definition at line 435 of file DirectHub.cxx.

◆ start()

void DirectHubPortSelect::DirectHubReadFlow::start ( )

inline

Starts the current flow.

Definition at line 280 of file DirectHub.cxx.

Member Data Documentation

◆ buf_

LinkedDataBufferPtr DirectHubPortSelect::DirectHubReadFlow::buf_

private

Current buffer that we are filling.

Definition at line 505 of file DirectHub.cxx.

◆ bufferNotifiable_

BarrierNotifiable* DirectHubPortSelect::DirectHubReadFlow::bufferNotifiable_

private

Barrier notifiable to keep track of the buffer's contents.

Definition at line 507 of file DirectHub.cxx.

◆ helper_

StateFlowSelectHelper DirectHubPortSelect::DirectHubReadFlow::helper_ {this}

private

Helper object for Select.

Definition at line 519 of file DirectHub.cxx.

◆ inlineCall_

uint16_t DirectHubPortSelect::DirectHubReadFlow::inlineCall_

private

1 if we got the send callback inline from the read_done.

Definition at line 511 of file DirectHub.cxx.

◆ parent_

DirectHubPortSelect* DirectHubPortSelect::DirectHubReadFlow::parent_

private

Pointer to the owninng port.

Definition at line 521 of file DirectHub.cxx.

◆ pendingLimiterPool_

AsyncNotifiableBlock DirectHubPortSelect::DirectHubReadFlow::pendingLimiterPool_

private

Initial value:

{

(unsigned)config_directhub_port_max_incoming_packets()}

Pool of BarrierNotifiables that limit the amount of inflight bytes we have.

Definition at line 516 of file DirectHub.cxx.

◆ segmenter_

std::unique_ptr<MessageSegmenter> DirectHubPortSelect::DirectHubReadFlow::segmenter_

private

Implementation (and state) of the business logic that segments incoming bytes into messages that shall be given to the hub.

Definition at line 524 of file DirectHub.cxx.

◆ segmentSize_

ssize_t DirectHubPortSelect::DirectHubReadFlow::segmentSize_

private

Output of the last segmenter call.

Definition at line 509 of file DirectHub.cxx.

◆ sendComplete_

uint16_t DirectHubPortSelect::DirectHubReadFlow::sendComplete_

private

1 if the run callback actually happened inline.

Definition at line 513 of file DirectHub.cxx.

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

DirectHub.cxx

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ DirectHubReadFlow()

Member Function Documentation

◆ alloc_for_read()

◆ barrier_allocated()

◆ call_head_segmenter()

◆ do_some_read()

◆ eval_segment()

◆ get_read_buffer()

◆ incomplete_message()

◆ read_done()

◆ read_shutdown()

◆ send_callback()

◆ send_done()

◆ send_prefix()

◆ start()

Member Data Documentation

◆ buf_

◆ bufferNotifiable_

◆ helper_

◆ inlineCall_

◆ parent_

◆ pendingLimiterPool_

◆ segmenter_

◆ segmentSize_

◆ sendComplete_