Commit 07734fb0 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Keep track of inbound paths in stream manager

parent eb049066
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_GCD_HPP
#define CAF_DETAIL_GCD_HPP
namespace caf {
namespace detail {
template <class T>
static T gcd(T a, T b) {
T r;
while (b != 0) {
r = a % b;
a = b;
b = r;
}
return a;
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_GCD_HPP
...@@ -50,8 +50,7 @@ public: ...@@ -50,8 +50,7 @@ public:
: stream_manager(self), : stream_manager(self),
fun_(std::move(fun)), fun_(std::move(fun)),
fin_(std::move(fin)), fin_(std::move(fin)),
out_(self), out_(self) {
open_inputs_(0) {
// nop // nop
} }
...@@ -64,7 +63,7 @@ public: ...@@ -64,7 +63,7 @@ public:
} }
bool done() const override { bool done() const override {
return open_inputs_ == 0; return this->inbound_paths_.empty();
} }
error handle(inbound_path*, downstream_msg::batch& x) override { error handle(inbound_path*, downstream_msg::batch& x) override {
...@@ -80,18 +79,6 @@ public: ...@@ -80,18 +79,6 @@ public:
return sec::unexpected_message; return sec::unexpected_message;
} }
void register_input_path(inbound_path* x) override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
++open_inputs_;
}
void deregister_input_path(inbound_path* x) noexcept override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
--open_inputs_;
}
protected: protected:
message make_final_result() override { message make_final_result() override {
return trait::make_result(state_, fin_); return trait::make_result(state_, fin_);
...@@ -102,7 +89,6 @@ private: ...@@ -102,7 +89,6 @@ private:
Fun fun_; Fun fun_;
Finalize fin_; Finalize fin_;
terminal_stream_scatterer out_; terminal_stream_scatterer out_;
long open_inputs_;
}; };
template <class Init, class Fun, class Finalize> template <class Init, class Fun, class Finalize>
......
...@@ -47,8 +47,7 @@ public: ...@@ -47,8 +47,7 @@ public:
: stream_manager(self), : stream_manager(self),
fun_(std::move(fun)), fun_(std::move(fun)),
cleanup_(std::move(cleanup)), cleanup_(std::move(cleanup)),
out_(self), out_(self) {
open_inputs_(0) {
// nop // nop
} }
...@@ -61,19 +60,7 @@ public: ...@@ -61,19 +60,7 @@ public:
} }
bool done() const override { bool done() const override {
return open_inputs_ == 0 && out_.clean(); return this->inbound_paths_.empty() && out_.clean();
}
void register_input_path(inbound_path* x) override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
++open_inputs_;
}
void deregister_input_path(inbound_path* x) noexcept override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
--open_inputs_;
} }
error handle(inbound_path*, downstream_msg::batch& x) override { error handle(inbound_path*, downstream_msg::batch& x) override {
...@@ -104,7 +91,6 @@ private: ...@@ -104,7 +91,6 @@ private:
Fun fun_; Fun fun_;
Cleanup cleanup_; Cleanup cleanup_;
DownstreamPolicy out_; DownstreamPolicy out_;
long open_inputs_;
}; };
template <class Init, class Fun, class Cleanup, class Scatterer> template <class Init, class Fun, class Cleanup, class Scatterer>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_TICK_EMITTER_HPP
#define CAF_DETAIL_TICK_EMITTER_HPP
#include <chrono>
#include "caf/config.hpp"
namespace caf {
namespace detail {
/// Converts realtime into a series of ticks, whereas each tick represents a
/// preconfigured timespan. For example, a tick emitter configured with a
/// timespan of 25ms generates a tick every 25ms after starting it.
class tick_emitter {
public:
// -- member types -----------------------------------------------------------
/// Discrete point in time.
using clock_type = std::chrono::steady_clock;
/// Discrete point in time.
using time_point = typename clock_type::time_point;
/// Difference between two points in time.
using duration_type = typename time_point::duration;
// -- constructors, destructors, and assignment operators --------------------
tick_emitter(time_point now)
: start_(std::move(now)),
interval_(0),
last_tick_id_(0) {
// nop
}
void interval(duration_type x) {
interval_ = x;
}
template <class F>
void update(time_point now, F& consumer) {
CAF_ASSERT(interval_.count() != 0);
auto diff = now - start_;
auto current_tick_id = diff.count() / interval_.count();
while (last_tick_id_ < current_tick_id)
consumer(++last_tick_id_);
}
private:
time_point start_;
duration_type interval_;
long last_tick_id_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_TICK_EMITTER_HPP
...@@ -75,21 +75,6 @@ public: ...@@ -75,21 +75,6 @@ public:
virtual error handle(outbound_path* from, upstream_msg::forced_drop& x); virtual error handle(outbound_path* from, upstream_msg::forced_drop& x);
/*
/// Adds a new sink to the stream.
virtual bool add_sink(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable);
/// Adds the source `hdl` to a stream. Convenience function for calling
/// `in().add_path(sid, hdl).second`.
virtual bool add_source(stream_slot slot, strong_actor_ptr source_ptr,
strong_actor_ptr original_stage, stream_priority prio,
bool redeployable, response_promise result_cb);
*/
/// Closes the stream when the parent terminates with default exit reason or /// Closes the stream when the parent terminates with default exit reason or
/// the stream reached its end. /// the stream reached its end.
virtual void close(); virtual void close();
...@@ -167,11 +152,13 @@ protected: ...@@ -167,11 +152,13 @@ protected:
/// implementation does nothing. /// implementation does nothing.
virtual void output_closed(error reason); virtual void output_closed(error reason);
/// Pointer to the parent actor. /// Points to the parent actor.
local_actor* self_; local_actor* self_;
/// Keeps track of pending handshakes. /// Stores non-owning pointers to all input paths.
std::vector<inbound_path*> inbound_paths_;
/// Keeps track of pending handshakes.
}; };
/// A reference counting pointer to a `stream_manager`. /// A reference counting pointer to a `stream_manager`.
......
...@@ -84,7 +84,9 @@ void stream_manager::abort(error reason) { ...@@ -84,7 +84,9 @@ void stream_manager::abort(error reason) {
void stream_manager::push() { void stream_manager::push() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
out().emit_batches(); do {
out().emit_batches();
} while (generate_messages());
} }
bool stream_manager::congested() const { bool stream_manager::congested() const {
...@@ -95,12 +97,18 @@ bool stream_manager::generate_messages() { ...@@ -95,12 +97,18 @@ bool stream_manager::generate_messages() {
return false; return false;
} }
void stream_manager::register_input_path(inbound_path*) { void stream_manager::register_input_path(inbound_path* ptr) {
// nop inbound_paths_.emplace_back(ptr);
} }
void stream_manager::deregister_input_path(inbound_path*) noexcept { void stream_manager::deregister_input_path(inbound_path* ptr) noexcept {
// nop using std::swap;
if (ptr != inbound_paths_.back()) {
auto i = std::find(inbound_paths_.begin(), inbound_paths_.end(), ptr);
CAF_ASSERT(i != inbound_paths_.end());
swap(*i, inbound_paths_.back());
}
inbound_paths_.pop_back();
} }
message stream_manager::make_final_result() { message stream_manager::make_final_result() {
......
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