Thrill  0.1
stream_sink.cpp
Go to the documentation of this file.
1 /*******************************************************************************
2  * thrill/data/stream_sink.cpp
3  *
4  * Part of Project Thrill - http://project-thrill.org
5  *
6  * Copyright (C) 2015 Timo Bingmann <[email protected]>
7  *
8  * All rights reserved. Published under the BSD-2 license in the LICENSE file.
9  ******************************************************************************/
10 
11 #include <thrill/data/stream_sink.hpp>
12 
13 #include <thrill/data/cat_stream.hpp>
14 #include <thrill/data/mix_stream.hpp>
15 #include <thrill/data/multiplexer_header.hpp>
16 #include <thrill/data/stream.hpp>
17 
18 #include <tlx/string/hexdump.hpp>
19 
20 namespace thrill {
21 namespace data {
22 
23 StreamSink::StreamSink()
24  : BlockSink(nullptr, -1), closed_(true) { }
25 
26 StreamSink::StreamSink(StreamDataPtr stream, BlockPool& block_pool,
27  net::Connection* connection,
28  MagicByte magic, StreamId stream_id,
29  size_t host_rank, size_t host_local_worker,
30  size_t peer_rank, size_t peer_local_worker)
31  : BlockSink(block_pool, host_local_worker),
32  stream_(std::move(stream)),
33  connection_(connection),
34  magic_(magic),
35  id_(stream_id),
36  host_rank_(host_rank),
37  peer_rank_(peer_rank),
38  peer_local_worker_(peer_local_worker) {
39  logger()
40  << "class" << "StreamSink"
41  << "event" << "open"
42  << "id" << id_
43  << "peer_host" << peer_rank_
44  << "src_worker" << my_worker_rank()
45  << "tgt_worker" << peer_worker_rank();
46 }
47 
48 StreamSink::StreamSink(StreamDataPtr stream, BlockPool& block_pool,
49  BlockQueue* block_queue,
50  StreamId stream_id,
51  size_t host_rank, size_t host_local_worker,
52  size_t peer_rank, size_t peer_local_worker)
53  : BlockSink(block_pool, host_local_worker),
54  stream_(std::move(stream)),
55  block_queue_(block_queue),
56  id_(stream_id),
57  host_rank_(host_rank),
58  peer_rank_(peer_rank),
59  peer_local_worker_(peer_local_worker) {
60  logger()
61  << "class" << "StreamSink"
62  << "event" << "open"
63  << "id" << id_
64  << "peer_host" << peer_rank_
65  << "src_worker" << my_worker_rank()
66  << "tgt_worker" << peer_worker_rank();
67 }
68 
69 StreamSink::StreamSink(StreamDataPtr stream, BlockPool& block_pool,
70  MixStreamDataPtr target,
71  StreamId stream_id,
72  size_t host_rank, size_t host_local_worker,
73  size_t peer_rank, size_t peer_local_worker)
74  : BlockSink(block_pool, host_local_worker),
75  stream_(std::move(stream)),
76  target_mix_stream_(target),
77  id_(stream_id),
78  host_rank_(host_rank),
79  peer_rank_(peer_rank),
80  peer_local_worker_(peer_local_worker) {
81  logger()
82  << "class" << "StreamSink"
83  << "event" << "open"
84  << "id" << id_
85  << "peer_host" << peer_rank_
86  << "src_worker" << my_worker_rank()
87  << "tgt_worker" << peer_worker_rank();
88 }
89 
90 size_t StreamSink::my_worker_rank() const {
91  return host_rank_ * workers_per_host() + local_worker_id_;
92 }
93 
94 size_t StreamSink::peer_worker_rank() const {
95  return peer_rank_ * workers_per_host() + peer_local_worker_;
96 }
97 
98 void StreamSink::AppendBlock(const Block& block, bool is_last_block) {
99  if (block.size() == 0) return;
100 
101  if (block_queue_) {
102  LOG << "StreamSink::AppendBlock()"
103  << " block=" << block
104  << " is_last_block=" << is_last_block
105  << " id_= " << id_
106  << " host_rank_=" << host_rank_
107  << " local_worker_id_=" << local_worker_id_
108  << " peer_rank_=" << peer_rank_
109  << " peer_local_worker_=" << peer_local_worker_
110  << " item_counter_=" << item_counter_
111  << " byte_counter_=" << byte_counter_
112  << " block_counter_=" << block_counter_;
113 
114  // StreamSink statistics
115  item_counter_ += block.num_items();
116  byte_counter_ += block.size();
117  block_counter_++;
118 
119  // StreamData statistics for internal transfer
120  stream_->tx_int_items_ += block.num_items();
121  stream_->tx_int_bytes_ += block.size();
122  stream_->tx_int_blocks_++;
123 
124  return block_queue_->AppendBlock(block, is_last_block);
125  }
126  if (target_mix_stream_) {
127  LOG << "StreamSink::AppendBlock()"
128  << " block=" << block
129  << " is_last_block=" << is_last_block
130  << " id_= " << id_
131  << " host_rank_=" << host_rank_
132  << " local_worker_id_=" << local_worker_id_
133  << " peer_rank_=" << peer_rank_
134  << " peer_local_worker_=" << peer_local_worker_
135  << " item_counter_=" << item_counter_
136  << " byte_counter_=" << byte_counter_
137  << " block_counter_=" << block_counter_;
138 
139  // StreamSink statistics
140  item_counter_ += block.num_items();
141  byte_counter_ += block.size();
142  block_counter_++;
143 
144  // StreamData statistics for internal transfer
145  stream_->tx_int_items_ += block.num_items();
146  stream_->tx_int_bytes_ += block.size();
147  stream_->tx_int_blocks_++;
148 
149  return target_mix_stream_->OnStreamBlock(
150  my_worker_rank(), block_counter_ - 1, Block(block));
151  }
152 
153  // otherwise: pin for network transfer
154  return AppendPinnedBlock(block.PinWait(local_worker_id()), is_last_block);
155 }
156 
157 void StreamSink::AppendBlock(Block&& block, bool is_last_block) {
158  die("FIXME: this should never be used?");
159  return AppendBlock(block, is_last_block);
160 }
161 
162 void StreamSink::AppendPinnedBlock(PinnedBlock&& block, bool is_last_block) {
163  if (block.size() == 0) return;
164 
165  LOG << "StreamSink::AppendPinnedBlock()"
166  << " block=" << block
167  << " is_last_block=" << is_last_block
168  << " id_= " << id_
169  << " host_rank_=" << host_rank_
170  << " local_worker_id_=" << local_worker_id_
171  << " peer_rank_=" << peer_rank_
172  << " peer_local_worker_=" << peer_local_worker_
173  << " item_counter_=" << item_counter_
174  << " byte_counter_=" << byte_counter_
175  << " block_counter_=" << block_counter_;
176 
177  // StreamSink statistics
178  item_counter_ += block.num_items();
179  byte_counter_ += block.size();
180  block_counter_++;
181 
182  if (block_queue_) {
183  // StreamData statistics for internal transfer
184  stream_->tx_int_items_ += block.num_items();
185  stream_->tx_int_bytes_ += block.size();
186  stream_->tx_int_blocks_++;
187 
188  return block_queue_->AppendPinnedBlock(std::move(block), is_last_block);
189  }
190  if (target_mix_stream_) {
191  // StreamData statistics for internal transfer
192  stream_->tx_int_items_ += block.num_items();
193  stream_->tx_int_bytes_ += block.size();
194  stream_->tx_int_blocks_++;
195 
196  return target_mix_stream_->OnStreamBlock(
197  my_worker_rank(), block_counter_ - 1,
198  std::move(block).MoveToBlock());
199  }
200 
201  LOG0 << "StreamSink::AppendPinnedBlock()"
202  << " data=" << tlx::hexdump(block.ToString());
203 
204  StreamMultiplexerHeader header(magic_, block);
205  header.stream_id = id_;
206  header.sender_worker = my_worker_rank();
207  header.receiver_local_worker = peer_local_worker_;
208  header.seq = block_counter_ - 1;
209  header.is_last_block = is_last_block;
210 
211  net::BufferBuilder bb;
212  header.Serialize(bb);
213 
214  net::Buffer buffer = bb.ToBuffer();
215  assert(buffer.size() == MultiplexerHeader::total_size);
216 
217  size_t send_size = buffer.size() + block.size();
218  // stream_->sem_queue_.wait(send_size);
219 
220  // StreamData statistics for network transfer
221  stream_->tx_net_items_ += block.num_items();
222  stream_->tx_net_bytes_ += send_size;
223  stream_->tx_net_blocks_++;
224  byte_counter_ += buffer.size();
225 
226  stream_->multiplexer_.dispatcher_.AsyncWrite(
227  *connection_, 42 + (connection_->tx_seq_.fetch_add(2) & 0xFFFF),
228  // send out Buffer and Block, guaranteed to be successive
229  std::move(buffer), std::move(block),
230  [s = stream_, send_size](net::Connection&) {
231  s->sem_queue_.signal(send_size);
232  });
233 
234  if (is_last_block) {
235  assert(!closed_);
236  closed_ = true;
237 
238  LOG << "StreamSink::AppendPinnedBlock()"
239  << " sent 'piggy-backed close stream' id=" << id_
240  << " from=" << my_worker_rank()
241  << " (host=" << host_rank_ << ")"
242  << " to=" << peer_worker_rank()
243  << " (host=" << peer_rank_ << ")";
244 
245  stream_->OnWriterClosed(peer_worker_rank(), /* sent */ true);
246 
247  Finalize();
248  }
249 }
250 
251 void StreamSink::Close() {
252  if (closed_) return;
253  closed_ = true;
254 
255  LOG << "StreamSink::Close() sending 'close stream' id=" << id_
256  << " from=" << my_worker_rank()
257  << " (host=" << host_rank_ << ")"
258  << " to=" << peer_worker_rank()
259  << " (host=" << peer_rank_ << ")";
260 
261  block_counter_++;
262 
263  if (block_queue_) {
264  // StreamData statistics for internal transfer
265  stream_->tx_int_blocks_++;
266  stream_->OnWriterClosed(peer_worker_rank(), /* sent */ true);
267  return block_queue_->Close();
268  }
269  if (target_mix_stream_) {
270  // StreamData statistics for internal transfer
271  stream_->tx_int_blocks_++;
272  stream_->OnWriterClosed(peer_worker_rank(), /* sent */ true);
273  return target_mix_stream_->OnStreamBlock(
274  my_worker_rank(), block_counter_ - 1, Block());
275  }
276 
277  stream_->OnWriterClosed(peer_worker_rank(), /* sent */ false);
278 
279  Finalize();
280 }
281 
282 void StreamSink::Finalize() {
283  logger()
284  << "class" << "StreamSink"
285  << "event" << "close"
286  << "id" << id_
287  << "peer_host" << peer_rank_
288  << "src_worker" << my_worker_rank()
289  << "tgt_worker" << peer_worker_rank()
290  << "items" << item_counter_
291  << "bytes" << byte_counter_
292  << "blocks" << block_counter_
293  << "timespan" << timespan_;
294 }
295 
296 } // namespace data
297 } // namespace thrill
298 
299 /******************************************************************************/
thrill::data::Block::PinWait
PinnedBlock PinWait(size_t local_worker_id) const
Convenience function to call Pin() and wait for the future.
Definition: block.cpp:35
thrill::data::Block
Block combines a reference to a read-only ByteBlock and book-keeping information. ...
Definition: block.hpp:52
thrill::data::BlockSink::AppendPinnedBlock
virtual void AppendPinnedBlock(PinnedBlock &&b, bool is_last_block)
Appends the PinnedBlock.
Definition: block_sink.hpp:89
thrill::data::StreamSink::my_worker_rank
size_t my_worker_rank() const
return local worker rank
Definition: stream_sink.cpp:90
LOG0
#define LOG0
Override default output: never or always output log.
Definition: logger.hpp:27
thrill::data::StreamSink::AppendBlock
void AppendBlock(const Block &block, bool is_last_block) final
Appends data to the StreamSink. Data may be sent but may be delayed.
Definition: stream_sink.cpp:98
std
STL namespace.
thrill::net::BufferBuilder::ToBuffer
Buffer ToBuffer()
Explicit conversion to Buffer MOVING the memory ownership.
Definition: buffer_builder.hpp:230
thrill::data::MagicByte
MagicByte
Definition: stream_data.hpp:34
die
#define die(msg)
Instead of std::terminate(), throw the output the message via an exception.
Definition: die.hpp:22
thrill::data::Block::size
size_t size() const
return length of valid data in bytes.
Definition: block.hpp:100
thrill::data::StreamSink::Close
void Close() final
Closes the connection.
Definition: stream_sink.cpp:251
stream_sink.hpp
cat_stream.hpp
thrill::data::MultiplexerHeader::total_size
static constexpr size_t total_size
Definition: multiplexer_header.hpp:66
thrill::data::BlockSink::local_worker_id_
size_t local_worker_id_
local worker id to associate pinned block with
Definition: block_sink.hpp:103
thrill::data::BlockPool
Pool to allocate, keep, swap out/in, and free all ByteBlocks on the host.
Definition: block_pool.hpp:42
thrill::data::StreamMultiplexerHeader
Block header is sent before a sequence of blocks it indicates the number of elements and their bounda...
Definition: multiplexer_header.hpp:80
thrill::data::BlockSink::local_worker_id
size_t local_worker_id() const
local worker id to associate pinned block with
Definition: block_sink.hpp:94
thrill::net::Connection
A Connection represents a link to another peer in a network group.
Definition: connection.hpp:49
thrill::data::BlockSink::block_pool
BlockPool * block_pool() const
Returns block_pool_.
Definition: block_sink.hpp:69
thrill::data::StreamSink::AppendPinnedBlock
void AppendPinnedBlock(PinnedBlock &&block, bool is_last_block) final
Appends data to the StreamSink. Data may be sent but may be delayed.
Definition: stream_sink.cpp:162
thrill::data::BlockQueue::AppendBlock
void AppendBlock(const Block &b, bool) final
Appends the (unpinned) Block.
Definition: block_queue.hpp:72
thrill::data::Block::num_items
size_t num_items() const
return number of items beginning in this block
Definition: block.hpp:85
thrill::net::BufferBuilder
BufferBuilder represents a dynamically growable area of memory, which can be modified by appending in...
Definition: buffer_builder.hpp:42
thrill::data::BlockSink
Pure virtual base class for all things that can receive Blocks from a BlockWriter.
Definition: block_sink.hpp:28
thrill::data::PinnedBlock
A pinned / pin-counted derivative of a Block.
Definition: block.hpp:157
thrill::data::StreamSink::target_mix_stream_
MixStreamDataPtr target_mix_stream_
destination mix stream
Definition: stream_sink.hpp:127
thrill::net::Connection::tx_seq_
std::atomic< uint32_t > tx_seq_
send sequence
Definition: connection.hpp:598
thrill::net::Buffer
Simple buffer of characters without initialization or growing functionality.
Definition: buffer.hpp:40
tlx::hexdump
std::string hexdump(const void *const data, size_t size)
Dump a (binary) string as a sequence of uppercase hexadecimal pairs.
Definition: hexdump.cpp:21
thrill::data::StreamSink::Finalize
void Finalize()
Finalize structure after sending the piggybacked or explicit close.
Definition: stream_sink.cpp:282
multiplexer_header.hpp
thrill::data::StreamSink::connection_
net::Connection * connection_
Definition: stream_sink.hpp:111
thrill::data::StreamSink::peer_worker_rank
size_t peer_worker_rank() const
return remote worker rank
Definition: stream_sink.cpp:94
mix_stream.hpp
thrill::data::BlockSink::workers_per_host
size_t workers_per_host() const
return number of workers per host
Definition: block_sink.hpp:72
LOG
#define LOG
Default logging method: output if the local debug variable is true.
Definition: logger.hpp:24
thrill::data::BlockSink::logger
common::JsonLogger & logger()
Returns BlockPool.logger_.
Definition: block_sink.hpp:66
thrill::data::BlockQueue
A BlockQueue is a thread-safe queue used to hand-over Block objects between threads.
Definition: block_queue.hpp:47
thrill::net::Buffer::size
size_type size() const noexcept
return number of items in Buffer
Definition: buffer.hpp:141
thrill::data::BlockQueue::Close
void Close() final
Close called by BlockWriter.
Definition: block_queue.cpp:28
thrill::data::StreamSink::timespan_
common::StatsTimerStart timespan_
Definition: stream_sink.hpp:141