Commit 984dc60e authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #1178

parents f7d03ca3 b8be8483
...@@ -49,6 +49,12 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -49,6 +49,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Fix memory leaks when deserializing URIs and when detaching the content of - Fix memory leaks when deserializing URIs and when detaching the content of
messages (#1160). messages (#1160).
- Fix undefined behavior in `string_view::compare` (#1164). - Fix undefined behavior in `string_view::compare` (#1164).
- Fix undefined behavior when passing `--config-file=` (i.e., without actual
argument) to CAF applications (#1167).
- Protect against self-assignment in a couple of CAF classes (#1169).
- Skipping high-priority messages resulted in CAF lowering the priority to
normal. This unintentional demotion has been fixed (#1171).
- Fix undefined behavior in the experimental datagram brokers (#1174).
## [0.18.0-rc.1] - 2020-09-09 ## [0.18.0-rc.1] - 2020-09-09
......
...@@ -51,7 +51,7 @@ public: ...@@ -51,7 +51,7 @@ public:
} }
explicit limited_vector(size_t initial_size) : size_(initial_size) { explicit limited_vector(size_t initial_size) : size_(initial_size) {
T tmp; auto tmp = T{};
std::fill_n(begin(), initial_size, tmp); std::fill_n(begin(), initial_size, tmp);
} }
......
...@@ -129,11 +129,16 @@ public: ...@@ -129,11 +129,16 @@ public:
return *this; return *this;
} }
intrusive_ptr& operator=(intrusive_ptr other) noexcept { intrusive_ptr& operator=(intrusive_ptr&& other) noexcept {
swap(other); swap(other);
return *this; return *this;
} }
intrusive_ptr& operator=(const intrusive_ptr& other) noexcept {
reset(other.ptr_);
return *this;
}
pointer get() const noexcept { pointer get() const noexcept {
return ptr_; return ptr_;
} }
......
...@@ -67,7 +67,8 @@ protected: ...@@ -67,7 +67,8 @@ protected:
workers_.reserve(num); workers_.reserve(num);
// Create worker instanes. // Create worker instanes.
for (size_t i = 0; i < num; ++i) for (size_t i = 0; i < num; ++i)
workers_.emplace_back(new worker_type(i, this, init, max_throughput_)); workers_.emplace_back(
std::make_unique<worker_type>(i, this, init, max_throughput_));
// Start all workers. // Start all workers.
for (auto& w : workers_) for (auto& w : workers_)
w->start(); w->start();
......
...@@ -105,11 +105,16 @@ public: ...@@ -105,11 +105,16 @@ public:
return *this; return *this;
} }
weak_intrusive_ptr& operator=(weak_intrusive_ptr other) noexcept { weak_intrusive_ptr& operator=(weak_intrusive_ptr&& other) noexcept {
swap(other); swap(other);
return *this; return *this;
} }
weak_intrusive_ptr& operator=(const weak_intrusive_ptr& other) noexcept {
reset(other.ptr_);
return *this;
}
pointer get() const noexcept { pointer get() const noexcept {
return ptr_; return ptr_;
} }
......
...@@ -409,8 +409,9 @@ error actor_system_config::extract_config_file_path(string_list& args) { ...@@ -409,8 +409,9 @@ error actor_system_config::extract_config_file_path(string_list& args) {
if (i == args.end()) if (i == args.end())
return none; return none;
if (path.empty()) { if (path.empty()) {
auto str = std::move(*i);
args.erase(i); args.erase(i);
return make_error(pec::missing_argument, std::string{*i}); return make_error(pec::missing_argument, std::move(str));
} }
auto evalue = ptr->parse(path); auto evalue = ptr->parse(path);
if (!evalue) if (!evalue)
......
...@@ -45,6 +45,8 @@ struct dyn_type_id_list { ...@@ -45,6 +45,8 @@ struct dyn_type_id_list {
other.hash = 0; other.hash = 0;
} }
dyn_type_id_list& operator=(dyn_type_id_list&&) noexcept = delete;
~dyn_type_id_list() { ~dyn_type_id_list() {
free(storage); free(storage);
} }
......
...@@ -39,7 +39,9 @@ error::error(const error& x) : data_(x ? new data(*x.data_) : nullptr) { ...@@ -39,7 +39,9 @@ error::error(const error& x) : data_(x ? new data(*x.data_) : nullptr) {
} }
error& error::operator=(const error& x) { error& error::operator=(const error& x) {
if (x) { if (this == &x) {
// nop
} else if (x) {
if (data_ == nullptr) if (data_ == nullptr)
data_.reset(new data(*x.data_)); data_.reset(new data(*x.data_));
else else
......
...@@ -131,15 +131,17 @@ bool load_data(Deserializer& source, message::data_ptr& data) { ...@@ -131,15 +131,17 @@ bool load_data(Deserializer& source, message::data_ptr& data) {
swap(meta, other.meta); swap(meta, other.meta);
} }
object_ptr& operator=(object_ptr&& other) noexcept { object_ptr& operator=(object_ptr&& other) noexcept {
if (obj) { if (this != &other) {
meta->destroy(obj); if (obj) {
free(obj); meta->destroy(obj);
obj = nullptr; free(obj);
meta = nullptr; obj = nullptr;
meta = nullptr;
}
using std::swap;
swap(obj, other.obj);
swap(meta, other.meta);
} }
using std::swap;
swap(obj, other.obj);
swap(meta, other.meta);
return *this; return *this;
} }
~object_ptr() noexcept { ~object_ptr() noexcept {
......
...@@ -89,7 +89,8 @@ message_builder& message_builder::append_from(const caf::message& msg, ...@@ -89,7 +89,8 @@ message_builder& message_builder::append_from(const caf::message& msg,
auto* meta = detail::global_meta_object(tid); auto* meta = detail::global_meta_object(tid);
storage_size_ += meta->padded_size; storage_size_ += meta->padded_size;
types_.push_back(tid); types_.push_back(tid);
elements_.emplace_back(new message_builder_element_adapter(msg, index)); elements_.emplace_back(
std::make_unique<message_builder_element_adapter>(msg, index));
} }
return *this; return *this;
} }
......
...@@ -932,16 +932,14 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) { ...@@ -932,16 +932,14 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
using namespace intrusive; using namespace intrusive;
auto& p = mailbox_.queue().policy(); auto& p = mailbox_.queue().policy();
auto& qs = mailbox_.queue().queues(); auto& qs = mailbox_.queue().queues();
// TODO: use generic lambda to avoid code duplication when switching to C++14 auto push = [&ptr](auto& q) {
if (p.id_of(*ptr) == normal_queue_index) {
auto& q = std::get<normal_queue_index>(qs);
q.inc_total_task_size(q.policy().task_size(*ptr)); q.inc_total_task_size(q.policy().task_size(*ptr));
q.cache().push_back(ptr.release()); q.cache().push_back(ptr.release());
} else { };
auto& q = std::get<normal_queue_index>(qs); if (p.id_of(*ptr) == normal_queue_index)
q.inc_total_task_size(q.policy().task_size(*ptr)); push(std::get<normal_queue_index>(qs));
q.cache().push_back(ptr.release()); else
} push(std::get<urgent_queue_index>(qs));
} }
scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() { scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
......
...@@ -78,7 +78,6 @@ CAF_TEST(move_assignment) { ...@@ -78,7 +78,6 @@ CAF_TEST(move_assignment) {
x = std::move(y); x = std::move(y);
CAF_CHECK_EQUAL(x, make_tuple(3, 4)); CAF_CHECK_EQUAL(x, make_tuple(3, 4));
CAF_CHECK_EQUAL(x.unique(), true); CAF_CHECK_EQUAL(x.unique(), true);
CAF_CHECK_EQUAL(y.ptr(), nullptr);
} }
CAF_TEST(make_cow_tuple) { CAF_TEST(make_cow_tuple) {
......
...@@ -28,7 +28,7 @@ bool inspect(Inspector& f, weekday& x) { ...@@ -28,7 +28,7 @@ bool inspect(Inspector& f, weekday& x) {
if (f.has_human_readable_format()) { if (f.has_human_readable_format()) {
auto get = [&x] { return to_string(x); }; auto get = [&x] { return to_string(x); };
auto set = [&x](std::string str) { return parse(str, x); }; auto set = [&x](std::string str) { return parse(str, x); };
f.object(x).fields(f.field("value", get, set)); return f.object(x).fields(f.field("value", get, set));
} else { } else {
using default_impl = caf::default_inspector_access<weekday>; using default_impl = caf::default_inspector_access<weekday>;
return default_impl::apply_object(f, x); return default_impl::apply_object(f, x);
......
...@@ -56,14 +56,22 @@ CAF_TEST(string conversion) { ...@@ -56,14 +56,22 @@ CAF_TEST(string conversion) {
} }
CAF_TEST(substrings) { CAF_TEST(substrings) {
auto without_prefix = [](string_view str, size_t n) {
str.remove_prefix(n);
return str;
};
auto without_suffix = [](string_view str, size_t n) {
str.remove_suffix(n);
return str;
};
auto x = "abcdefghi"_sv; auto x = "abcdefghi"_sv;
CAF_CHECK(x.remove_prefix(3), "defghi"); CAF_CHECK_EQUAL(without_prefix(x, 3), "defghi");
CAF_CHECK(x.remove_suffix(3), "abcdef"); CAF_CHECK_EQUAL(without_suffix(x, 3), "abcdef");
CAF_CHECK(x.substr(3, 3), "def"); CAF_CHECK_EQUAL(x.substr(3, 3), "def");
CAF_CHECK(x.remove_prefix(9), ""); CAF_CHECK_EQUAL(without_prefix(x, 9), "");
CAF_CHECK(x.remove_suffix(9), ""); CAF_CHECK_EQUAL(without_suffix(x, 9), "");
CAF_CHECK(x.substr(9), ""); CAF_CHECK_EQUAL(x.substr(9), "");
CAF_CHECK(x.substr(0, 0), ""); CAF_CHECK_EQUAL(x.substr(0, 0), "");
} }
CAF_TEST(compare) { CAF_TEST(compare) {
......
...@@ -75,27 +75,25 @@ abstract_broker::~abstract_broker() { ...@@ -75,27 +75,25 @@ abstract_broker::~abstract_broker() {
void abstract_broker::configure_read(connection_handle hdl, void abstract_broker::configure_read(connection_handle hdl,
receive_policy::config cfg) { receive_policy::config cfg) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(cfg)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(cfg));
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (x)
x->configure_read(cfg); x->configure_read(cfg);
} }
void abstract_broker::ack_writes(connection_handle hdl, bool enable) { void abstract_broker::ack_writes(connection_handle hdl, bool enable) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable));
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (x)
x->ack_writes(enable); x->ack_writes(enable);
} }
byte_buffer& abstract_broker::wr_buf(connection_handle hdl) { byte_buffer& abstract_broker::wr_buf(connection_handle hdl) {
CAF_ASSERT(hdl != invalid_connection_handle); CAF_ASSERT(hdl != invalid_connection_handle);
auto x = by_id(hdl); if (auto x = by_id(hdl)) {
if (!x) { return x->wr_buf();
} else {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown connection_handle:" CAF_LOG_ERROR("tried to access wr_buf() of an unknown connection_handle:"
<< CAF_ARG(hdl)); << CAF_ARG(hdl));
return dummy_wr_buf_; return dummy_wr_buf_;
} }
return x->wr_buf();
} }
void abstract_broker::write(connection_handle hdl, size_t bs, const void* buf) { void abstract_broker::write(connection_handle hdl, size_t bs, const void* buf) {
...@@ -110,34 +108,32 @@ void abstract_broker::write(connection_handle hdl, span<const byte> buf) { ...@@ -110,34 +108,32 @@ void abstract_broker::write(connection_handle hdl, span<const byte> buf) {
} }
void abstract_broker::flush(connection_handle hdl) { void abstract_broker::flush(connection_handle hdl) {
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (x)
x->flush(); x->flush();
} }
void abstract_broker::ack_writes(datagram_handle hdl, bool enable) { void abstract_broker::ack_writes(datagram_handle hdl, bool enable) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable));
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (x)
x->ack_writes(enable); x->ack_writes(enable);
} }
byte_buffer& abstract_broker::wr_buf(datagram_handle hdl) { byte_buffer& abstract_broker::wr_buf(datagram_handle hdl) {
auto x = by_id(hdl); if (auto x = by_id(hdl)) {
if (!x) { return x->wr_buf(hdl);
} else {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown" CAF_LOG_ERROR("tried to access wr_buf() of an unknown"
"datagram_handle"); "datagram_handle");
return dummy_wr_buf_; return dummy_wr_buf_;
} }
return x->wr_buf(hdl);
} }
void abstract_broker::enqueue_datagram(datagram_handle hdl, byte_buffer buf) { void abstract_broker::enqueue_datagram(datagram_handle hdl, byte_buffer buf) {
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (!x) x->enqueue_datagram(hdl, std::move(buf));
else
CAF_LOG_ERROR("tried to access datagram_buffer() of an unknown" CAF_LOG_ERROR("tried to access datagram_buffer() of an unknown"
"datagram_handle"); "datagram_handle");
x->enqueue_datagram(hdl, std::move(buf));
} }
void abstract_broker::write(datagram_handle hdl, size_t bs, const void* buf) { void abstract_broker::write(datagram_handle hdl, size_t bs, const void* buf) {
...@@ -148,8 +144,7 @@ void abstract_broker::write(datagram_handle hdl, size_t bs, const void* buf) { ...@@ -148,8 +144,7 @@ void abstract_broker::write(datagram_handle hdl, size_t bs, const void* buf) {
} }
void abstract_broker::flush(datagram_handle hdl) { void abstract_broker::flush(datagram_handle hdl) {
auto x = by_id(hdl); if (auto x = by_id(hdl))
if (x)
x->flush(); x->flush();
} }
...@@ -337,11 +332,12 @@ uint16_t abstract_broker::local_port(datagram_handle hdl) { ...@@ -337,11 +332,12 @@ uint16_t abstract_broker::local_port(datagram_handle hdl) {
} }
bool abstract_broker::remove_endpoint(datagram_handle hdl) { bool abstract_broker::remove_endpoint(datagram_handle hdl) {
auto x = by_id(hdl); if (auto x = by_id(hdl)) {
if (!x) x->remove_endpoint(hdl);
return true;
} else {
return false; return false;
x->remove_endpoint(hdl); }
return true;
} }
void abstract_broker::close_all() { void abstract_broker::close_all() {
......
...@@ -62,9 +62,11 @@ ip_endpoint::ip_endpoint(const ip_endpoint& other) { ...@@ -62,9 +62,11 @@ ip_endpoint::ip_endpoint(const ip_endpoint& other) {
} }
ip_endpoint& ip_endpoint::operator=(const ip_endpoint& other) { ip_endpoint& ip_endpoint::operator=(const ip_endpoint& other) {
ptr_.reset(new ip_endpoint::impl); if (this != &other) {
memcpy(address(), other.caddress(), sizeof(sockaddr_storage)); ptr_.reset(new ip_endpoint::impl);
*length() = *other.clength(); memcpy(address(), other.caddress(), sizeof(sockaddr_storage));
*length() = *other.clength();
}
return *this; return *this;
} }
......
...@@ -120,7 +120,11 @@ void close_socket(native_socket fd) { ...@@ -120,7 +120,11 @@ void close_socket(native_socket fd) {
} }
bool would_block_or_temporarily_unavailable(int errcode) { bool would_block_or_temporarily_unavailable(int errcode) {
# if EAGAIN != EWOULDBLOCK
return errcode == EAGAIN || errcode == EWOULDBLOCK; return errcode == EAGAIN || errcode == EWOULDBLOCK;
# else
return errcode == EAGAIN;
# endif
} }
string last_socket_error_as_string() { string last_socket_error_as_string() {
......
...@@ -78,8 +78,8 @@ behavior pong(event_based_actor* self, suite_state_ptr ssp) { ...@@ -78,8 +78,8 @@ behavior pong(event_based_actor* self, suite_state_ptr ssp) {
behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) { behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("peer_fun called"); CAF_MESSAGE("peer_fun called");
CAF_REQUIRE(self != nullptr);
CAF_REQUIRE(self->subtype() == resumable::io_actor); CAF_REQUIRE(self->subtype() == resumable::io_actor);
CAF_CHECK(self != nullptr);
self->monitor(buddy); self->monitor(buddy);
self->set_down_handler([self](down_msg& dm) { self->set_down_handler([self](down_msg& dm) {
// Stop if buddy is done. // Stop if buddy is done.
......
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