Commit 9d87f3ca authored by Dominik Charousset's avatar Dominik Charousset

Update docs and remove dead code

parent 1d8ee28a
......@@ -112,13 +112,13 @@ struct downstream_msg : tag::boxing_type {
content_type content;
};
/// Allows the testing DSL to unbox `stream_msg` automagically.
/// Allows the testing DSL to unbox `downstream_msg` automagically.
template <class T>
const T& get(const downstream_msg& x) {
return get<T>(x.content);
}
/// Allows the testing DSL to check whether `stream_msg` holds a `T`.
/// Allows the testing DSL to check whether `downstream_msg` holds a `T`.
template <class T>
bool is(const downstream_msg& x) {
return holds_alternative<T>(x.content);
......@@ -155,7 +155,7 @@ typename Inspector::result_type inspect(Inspector& f,
/// @relates downstream_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg& x) {
return f(meta::type_name("stream_msg"), x.slots, x.sender, x.content);
return f(meta::type_name("downstream_msg"), x.slots, x.sender, x.content);
}
} // namespace caf
......
......@@ -120,30 +120,28 @@ public:
~inbound_path();
/// Updates `last_batch_id` and `assigned_credit` before calling
/// `mgr->handle(this, x)`.
/// Updates `last_batch_id` and `assigned_credit` before dispatching to the
/// manager.
void handle(downstream_msg::batch& x);
/// Calls `mgr->handle(this, x)`.
inline void handle(downstream_msg::close& x) {
/// Dispatches any `downstream_msg` other than `batch` directly to the
/// manager.
template <class T>
void handle(T& x) {
mgr->handle(this, x);
}
/// Calls `mgr->handle(this, x)`.
inline void handle(downstream_msg::forced_close& x) {
mgr->handle(this, x);
}
/// Emits a `stream_msg::ack_batch` on this path and sets `assigned_credit`
/// to `initial_demand`.
/// Emits an `upstream_msg::ack_batch`.
void emit_ack_open(local_actor* self, actor_addr rebind_from);
/// Sends a `stream_msg::ack_batch` with credit for the next round. Credit is
/// calculated as `max_queue_size - (assigned_credit - queued_items)`, whereas
/// `max_queue_size` is `2 * ...`.
/// @param self Points to the parent actor.
/// @param queued_items Counts the accumulated size of all batches that are
/// currently waiting in the mailbox.
/// Sends an `upstream_msg::ack_batch` for granting new credit. Credit is
/// calculated from sampled batch durations, the cycle duration and the
/// desired batch complexity.
/// @param self Points to the parent actor, i.e., sender of the message.
/// @param queued_items Accumulated size of all batches that are currently
/// waiting in the mailbox.
/// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, long queued_items, timespan cycle,
timespan desired_batch_complexity);
......@@ -151,13 +149,13 @@ public:
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
bool up_to_date();
/// Sends a `stream_msg::close` on this path.
/// Sends an `upstream_msg::drop` on this path.
void emit_regular_shutdown(local_actor* self);
/// Sends a `stream_msg::forced_close` on this path.
/// Sends an `upstream_msg::forced_drop` on this path.
void emit_irregular_shutdown(local_actor* self, error reason);
/// Sends a `stream_msg::forced_close` on this path.
/// Sends an `upstream_msg::forced_drop`.
static void emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl,
error reason);
......
......@@ -69,12 +69,12 @@ public:
/// Identifies one-to-one messages with normal priority.
static constexpr uint64_t default_message_category = 0; // 0b00
/// Identifies stream messages received from upstream actors, e.g.,
/// `stream_msg::batch`.
/// Identifies stream messages that flow upstream, e.g.,
/// `upstream_msg::ack_batch`.
static constexpr uint64_t upstream_message_category = 1; // 0b01
/// Identifies stream messages received from downstream actors, e.g.,
/// `stream_msg::ack_batch`.
/// Identifies stream messages that flow downstream, e.g.,
/// `downstream_msg::batch`.
static constexpr uint64_t downstream_message_category = 2; // 0b10
/// Identifies one-to-one messages with high priority.
......
......@@ -50,14 +50,6 @@ public:
/// Stores batches until receiving corresponding ACKs.
using cache_type = std::deque<std::pair<int64_t, downstream_msg::batch>>;
// -- nested classes ---------------------------------------------------------
/// Stores information about the initiator of the steam.
struct client_data {
strong_actor_ptr hdl;
message_id mid;
};
// -- constants --------------------------------------------------------------
/// Stream aborter flag to monitor a path.
......@@ -72,15 +64,12 @@ public:
// -- downstream communication -----------------------------------------------
/// Sends a stream handshake.
/// Sends an `open_stream_msg` handshake.
static void emit_open(local_actor* self, stream_slot slot,
strong_actor_ptr to, message handshake_data,
stream_priority prio);
/// Sends a `stream_msg::batch` on this path, decrements `open_credit` by
/// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void handle_ack_open(long initial_credit);
/// Sends a `downstream_msg::batch` on this path. Decrements `open_credit` by
/// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(local_actor* self, long xs_size, message xs);
......@@ -114,13 +103,13 @@ public:
emit_batch(self, cache.size(), make_message(std::move(cache)));
}
/// Sends a `stream_msg::drop` on this path.
/// Sends a `downstream_msg::close` on this path.
void emit_regular_shutdown(local_actor* self);
/// Sends a `stream_msg::forced_drop` on this path.
/// Sends a `downstream_msg::forced_close` on this path.
void emit_irregular_shutdown(local_actor* self, error reason);
/// Sends a `stream_msg::forced_drop` on this path.
/// Sends a `downstream_msg::forced_close`.
static void emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl,
error reason);
......@@ -144,39 +133,22 @@ public:
/// Next expected batch ID.
int64_t next_batch_id;
/// Currently available credit for this path.
/// Currently available credit on this path.
long open_credit;
/// Batch size configured by the downstream actor.
/// Ideal batch size. Configured by the sink.
uint64_t desired_batch_size;
/// Next expected batch ID to be acknowledged. Actors can receive a more
/// advanced batch ID in an ACK message, since CAF uses accumulative ACKs.
/// ID of the first unacknowledged batch. Note that CAF uses accumulative
/// ACKs, i.e., receiving an ACK with a higher ID is not an error.
int64_t next_ack_id;
/// Caches the initiator of the stream (client) with the original request ID
/// until the stream handshake is either confirmed or aborted. Once
/// confirmed, the next stage takes responsibility for answering to the
/// client.
client_data cd;
/// Stores whether an error occurred during stream processing.
error shutdown_reason;
};
/// @relates outbound_path::client_data
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
outbound_path::client_data& x) {
return f(x.hdl, x.mid);
}
/// @relates outbound_path
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, outbound_path& x) {
return f(meta::type_name("outbound_path"), x.slots, x.hdl, x.next_batch_id,
x.open_credit, x.desired_batch_size, x.next_ack_id, x.cd,
x.shutdown_reason);
x.open_credit, x.desired_batch_size, x.next_ack_id);
}
} // namespace caf
......
......@@ -38,8 +38,7 @@
namespace caf {
/// Manages a single stream with any number of down- and upstream actors.
/// @relates stream_msg
/// Manages a single stream with any number of in- and outbound paths.
class stream_manager : public ref_counted {
public:
// -- member types -----------------------------------------------------------
......
......@@ -111,7 +111,7 @@ struct open_stream_msg {
stream_priority priority;
};
/// @relates stream_handshake_msg
/// @relates open_stream_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, open_stream_msg& x) {
return f(meta::type_name("open_stream_msg"), x.slot, x.msg, x.prev_stage,
......
......@@ -37,7 +37,7 @@
namespace caf {
/// Stream messages that travel upstream, i.e., acks and drop messages.
/// Stream messages that flow upstream, i.e., acks and drop messages.
struct upstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
......@@ -135,13 +135,13 @@ struct upstream_msg : tag::boxing_type {
content_type content;
};
/// Allows the testing DSL to unbox `stream_msg` automagically.
/// Allows the testing DSL to unbox `upstream_msg` automagically.
template <class T>
const T& get(const upstream_msg& x) {
return get<T>(x.content);
}
/// Allows the testing DSL to check whether `stream_msg` holds a `T`.
/// Allows the testing DSL to check whether `upstream_msg` holds a `T`.
template <class T>
bool is(const upstream_msg& x) {
return holds_alternative<T>(x.content);
......@@ -187,7 +187,7 @@ typename Inspector::result_type inspect(Inspector& f,
/// @relates upstream_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, upstream_msg& x) {
return f(meta::type_name("stream_msg"), x.slots, x.sender, x.content);
return f(meta::type_name("upstream_msg"), x.slots, x.sender, x.content);
}
} // namespace caf
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment