StreamSink Class Reference

Detailed Description

StreamSink is an BlockSink that sends data via a network socket to the StreamData object on a different worker.

Definition at line 38 of file stream_sink.hpp.

Inheritance diagram for StreamSink:

Collaboration diagram for StreamSink:

#include <stream_sink.hpp>

Public Types
using	StreamId = size_t

Public Member Functions
	StreamSink ()
	StreamSink (StreamDataPtr stream, BlockPool &block_pool, net::Connection *connection, MagicByte magic, StreamId stream_id, size_t host_rank, size_t host_local_worker, size_t peer_rank, size_t peer_local_worker)
	StreamSink sending out to network. More...
	StreamSink (StreamDataPtr stream, BlockPool &block_pool, BlockQueue *block_queue, StreamId stream_id, size_t host_rank, size_t host_local_worker, size_t peer_rank, size_t peer_local_worker)
	StreamSink sending out local BlockQueue. More...
	StreamSink (StreamDataPtr stream, BlockPool &block_pool, MixStreamDataPtr target, StreamId stream_id, size_t host_rank, size_t host_local_worker, size_t peer_rank, size_t peer_local_worker)
	StreamSink sending out local MixBlockQueue. More...
	StreamSink (StreamSink &&)=default
void	AppendBlock (const Block &block, bool is_last_block) final
	Appends data to the StreamSink. Data may be sent but may be delayed. More...
void	AppendBlock (Block &&block, bool is_last_block) final
	Appends data to the StreamSink. Data may be sent but may be delayed. More...
void	AppendPinnedBlock (PinnedBlock &&block, bool is_last_block) final
	Appends data to the StreamSink. Data may be sent but may be delayed. More...
void	Close () final
	Closes the connection. More...
bool	closed () const
	return close flag More...
void	Finalize ()
	Finalize structure after sending the piggybacked or explicit close. More...
bool	IsValid () const
	is valid? More...
size_t	my_worker_rank () const
	return local worker rank More...
StreamSink &	operator= (StreamSink &&)=default
size_t	peer_worker_rank () const
	return remote worker rank More...
Public Member Functions inherited from BlockSink
	BlockSink (BlockPool &block_pool, size_t local_worker_id)
	constructor with reference to BlockPool More...
	BlockSink (BlockPool *block_pool, size_t local_worker_id)
	constructor with reference to BlockPool More...
	BlockSink (const BlockSink &)=default
	default copy-constructor More...
	BlockSink (BlockSink &&)=default
	move-constructor: default More...
virtual	~BlockSink ()
	required virtual destructor More...
virtual PinnedByteBlockPtr	AllocateByteBlock (size_t block_size)
BlockPool *	block_pool () const
	Returns block_pool_. More...
size_t	local_worker_id () const
	local worker id to associate pinned block with More...
common::JsonLogger &	logger ()
	Returns BlockPool.logger_. More...
BlockSink &	operator= (const BlockSink &)=default
	default assignment operator More...
BlockSink &	operator= (BlockSink &&)=default
	move-assignment operator: default More...
virtual void	ReleaseByteBlock (ByteBlockPtr &block)
	Release an unused ByteBlock with n bytes backing memory. More...
size_t	workers_per_host () const
	return number of workers per host More...

Static Public Attributes
static constexpr bool	allocate_can_fail_ = false
Static Public Attributes inherited from BlockSink
static constexpr bool	allocate_can_fail_ = false

Private Attributes
size_t	block_counter_ = 0
size_t	byte_counter_ = 0
bool	closed_ = false
size_t	host_rank_ = size_t(-1)
StreamId	id_ = size_t(-1)
size_t	item_counter_ = 0
size_t	peer_local_worker_ = size_t(-1)
size_t	peer_rank_ = size_t(-1)
StreamDataPtr	stream_
common::StatsTimerStart	timespan_
StreamSink To Network
net::Connection *	connection_ = nullptr
MagicByte	magic_ = MagicByte::Invalid
StreamSink To BlockQueue (CatStream Loopback)
BlockQueue *	block_queue_ = nullptr
StreamSink To MixBlockQueue (MixStream Loopback)
MixStreamDataPtr	target_mix_stream_
	destination mix stream More...

Static Private Attributes
static constexpr bool	debug = false

Additional Inherited Members
Protected Attributes inherited from BlockSink
size_t	local_worker_id_
	local worker id to associate pinned block with More...

Member Typedef Documentation

StreamId

using StreamId = size_t

Definition at line 43 of file stream_sink.hpp.

Constructor & Destructor Documentation

StreamSink() [1/5]

StreamSink

(

)

Construct invalid StreamSink, needed for placeholders in sinks arrays where Blocks are directly sent to local workers.

Definition at line 23 of file stream_sink.cpp.

StreamSink() [2/5]

StreamSink	(	StreamDataPtr	stream,
		BlockPool &	block_pool,
		net::Connection *	connection,
		MagicByte	magic,
		StreamId	stream_id,
		size_t	host_rank,
		size_t	host_local_worker,
		size_t	peer_rank,
		size_t	peer_local_worker
	)

StreamSink sending out to network.

Definition at line 26 of file stream_sink.cpp.

References StreamSink::id_, BlockSink::logger(), StreamSink::my_worker_rank(), StreamSink::peer_rank_, and StreamSink::peer_worker_rank().

StreamSink() [3/5]

StreamSink	(	StreamDataPtr	stream,
		BlockPool &	block_pool,
		BlockQueue *	block_queue,
		StreamId	stream_id,
		size_t	host_rank,
		size_t	host_local_worker,
		size_t	peer_rank,
		size_t	peer_local_worker
	)

StreamSink sending out local BlockQueue.

Definition at line 48 of file stream_sink.cpp.

References StreamSink::id_, BlockSink::logger(), StreamSink::my_worker_rank(), StreamSink::peer_rank_, and StreamSink::peer_worker_rank().

StreamSink() [4/5]

StreamSink	(	StreamDataPtr	stream,
		BlockPool &	block_pool,
		MixStreamDataPtr	target,
		StreamId	stream_id,
		size_t	host_rank,
		size_t	host_local_worker,
		size_t	peer_rank,
		size_t	peer_local_worker
	)

StreamSink sending out local MixBlockQueue.

Definition at line 69 of file stream_sink.cpp.

References StreamSink::id_, BlockSink::logger(), StreamSink::my_worker_rank(), StreamSink::peer_rank_, and StreamSink::peer_worker_rank().

StreamSink() [5/5]

StreamSink ( StreamSink &&  )

default

Member Function Documentation

AppendBlock() [1/2]

void AppendBlock ( const Block &  block, bool  is_last_block  )

finalvirtual

Appends data to the StreamSink. Data may be sent but may be delayed.

Implements BlockSink.

Definition at line 98 of file stream_sink.cpp.

References BlockQueue::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::block_counter_, StreamSink::block_queue_, StreamSink::byte_counter_, StreamSink::host_rank_, StreamSink::id_, StreamSink::item_counter_, BlockSink::local_worker_id(), BlockSink::local_worker_id_, LOG, StreamSink::my_worker_rank(), Block::num_items(), StreamSink::peer_local_worker_, StreamSink::peer_rank_, Block::PinWait(), Block::size(), StreamSink::stream_, and StreamSink::target_mix_stream_.

Referenced by StreamSink::AppendBlock().

AppendBlock() [2/2]

void AppendBlock ( Block &&  block, bool  is_last_block  )

finalvirtual

Appends data to the StreamSink. Data may be sent but may be delayed.

Implements BlockSink.

Definition at line 157 of file stream_sink.cpp.

References StreamSink::AppendBlock(), and die.

AppendPinnedBlock()

void AppendPinnedBlock ( PinnedBlock &&  block, bool  is_last_block  )

finalvirtual

Appends data to the StreamSink. Data may be sent but may be delayed.

Reimplemented from BlockSink.

Definition at line 162 of file stream_sink.cpp.

References BlockSink::AppendPinnedBlock(), StreamSink::block_counter_, StreamSink::block_queue_, StreamSink::byte_counter_, StreamSink::closed_, StreamSink::connection_, StreamSink::Finalize(), tlx::hexdump(), StreamSink::host_rank_, StreamSink::id_, StreamSink::item_counter_, BlockSink::local_worker_id_, LOG, LOG0, StreamSink::magic_, StreamSink::my_worker_rank(), StreamSink::peer_local_worker_, StreamSink::peer_rank_, StreamSink::peer_worker_rank(), Buffer::size(), StreamSink::stream_, StreamSink::target_mix_stream_, BufferBuilder::ToBuffer(), MultiplexerHeader::total_size, and Connection::tx_seq_.

Referenced by StreamSink::AppendBlock().

Close()

void Close ( )

finalvirtual

Closes the connection.

Implements BlockSink.

Definition at line 251 of file stream_sink.cpp.

References StreamSink::block_counter_, StreamSink::block_queue_, BlockQueue::Close(), StreamSink::closed_, StreamSink::Finalize(), StreamSink::host_rank_, StreamSink::id_, LOG, StreamSink::my_worker_rank(), StreamSink::peer_rank_, StreamSink::peer_worker_rank(), StreamSink::stream_, and StreamSink::target_mix_stream_.

closed()

bool closed ( ) const

inline

return close flag

Definition at line 89 of file stream_sink.hpp.

References StreamSink::closed_.

Finalize()

void Finalize

(

)

Finalize structure after sending the piggybacked or explicit close.

Definition at line 282 of file stream_sink.cpp.

References StreamSink::block_counter_, StreamSink::byte_counter_, StreamSink::id_, StreamSink::item_counter_, BlockSink::logger(), StreamSink::my_worker_rank(), StreamSink::peer_rank_, StreamSink::peer_worker_rank(), and StreamSink::timespan_.

Referenced by StreamSink::AppendPinnedBlock(), and StreamSink::Close().

IsValid()

bool IsValid ( ) const

inline

is valid?

Definition at line 92 of file stream_sink.hpp.

References StreamSink::stream_.

my_worker_rank()

size_t my_worker_rank

(

)

const

return local worker rank

Definition at line 90 of file stream_sink.cpp.

References StreamSink::host_rank_, BlockSink::local_worker_id_, and BlockSink::workers_per_host().

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), StreamSink::Finalize(), and StreamSink::StreamSink().

operator=()

StreamSink& operator= ( StreamSink &&  )

default

peer_worker_rank()

size_t peer_worker_rank

(

)

const

return remote worker rank

Definition at line 94 of file stream_sink.cpp.

References StreamSink::peer_local_worker_, StreamSink::peer_rank_, and BlockSink::workers_per_host().

Referenced by StreamSink::AppendPinnedBlock(), StreamSink::Close(), StreamSink::Finalize(), and StreamSink::StreamSink().

Member Data Documentation

allocate_can_fail_

constexpr bool allocate_can_fail_ = false

static

boolean flag whether to check if AllocateByteBlock can fail in any subclass (if false: accelerate BlockWriter to not be able to cope with nullptr).

Definition at line 97 of file stream_sink.hpp.

block_counter_

size_t block_counter_ = 0

private

Definition at line 140 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), and StreamSink::Finalize().

block_queue_

BlockQueue* block_queue_ = nullptr

private

Definition at line 119 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), and StreamSink::Close().

byte_counter_

size_t byte_counter_ = 0

private

Definition at line 139 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), and StreamSink::Finalize().

closed_

bool closed_ = false

private

Definition at line 136 of file stream_sink.hpp.

Referenced by StreamSink::AppendPinnedBlock(), StreamSink::Close(), and StreamSink::closed().

connection_

net::Connection* connection_ = nullptr

private

Definition at line 111 of file stream_sink.hpp.

Referenced by StreamSink::AppendPinnedBlock().

debug

constexpr bool debug = false

staticprivate

Definition at line 40 of file stream_sink.hpp.

host_rank_

size_t host_rank_ = size_t(-1)

private

Definition at line 132 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), and StreamSink::my_worker_rank().

id_

StreamId id_ = size_t(-1)

private

Definition at line 131 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), StreamSink::Finalize(), and StreamSink::StreamSink().

item_counter_

size_t item_counter_ = 0

private

Definition at line 138 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), and StreamSink::Finalize().

magic_

MagicByte magic_ = MagicByte::Invalid

private

Definition at line 112 of file stream_sink.hpp.

Referenced by StreamSink::AppendPinnedBlock().

peer_local_worker_

size_t peer_local_worker_ = size_t(-1)

private

Definition at line 135 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), and StreamSink::peer_worker_rank().

peer_rank_

size_t peer_rank_ = size_t(-1)

private

Definition at line 134 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), StreamSink::Finalize(), StreamSink::peer_worker_rank(), and StreamSink::StreamSink().

stream_

StreamDataPtr stream_

private

Definition at line 106 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), StreamSink::Close(), and StreamSink::IsValid().

target_mix_stream_

MixStreamDataPtr target_mix_stream_

private

destination mix stream

Definition at line 127 of file stream_sink.hpp.

Referenced by StreamSink::AppendBlock(), StreamSink::AppendPinnedBlock(), and StreamSink::Close().

timespan_

common::StatsTimerStart timespan_

private

Definition at line 141 of file stream_sink.hpp.

Referenced by StreamSink::Finalize().

---

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

• thrill/data/stream_sink.hpp
• thrill/data/stream_sink.cpp