Commit a642ccd6 authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of continue_reading

parent d07c530b
...@@ -410,9 +410,15 @@ public: ...@@ -410,9 +410,15 @@ public:
if (max_read_size_ == 0) { if (max_read_size_ == 0) {
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent); auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
upper_layer_.continue_reading(this_layer_ptr); upper_layer_.continue_reading(this_layer_ptr);
if (max_read_size_ != 0) {
return handle_buffered_data(parent);
} else {
return read_result::stop;
} }
} else {
return handle_buffered_data(parent); return handle_buffered_data(parent);
} }
}
template <class ParentPtr> template <class ParentPtr>
write_result handle_continue_writing(ParentPtr) { write_result handle_continue_writing(ParentPtr) {
......
...@@ -78,7 +78,7 @@ operation to_operation(const socket_manager_ptr& mgr, ...@@ -78,7 +78,7 @@ operation to_operation(const socket_manager_ptr& mgr,
// -- static utility functions ------------------------------------------------- // -- static utility functions -------------------------------------------------
#ifndef CAF_WINDOWS #ifdef CAF_LINUX
void multiplexer::block_sigpipe() { void multiplexer::block_sigpipe() {
sigset_t sigpipe_mask; sigset_t sigpipe_mask;
...@@ -186,11 +186,7 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -186,11 +186,7 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
void multiplexer::continue_reading(const socket_manager_ptr& mgr) { void multiplexer::continue_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_continue_reading(mgr);
} else {
write_to_pipe(pollset_updater::code::continue_reading, mgr.get()); write_to_pipe(pollset_updater::code::continue_reading, mgr.get());
}
} }
void multiplexer::continue_writing(const socket_manager_ptr& mgr) { void multiplexer::continue_writing(const socket_manager_ptr& mgr) {
...@@ -509,11 +505,11 @@ short multiplexer::active_mask_of(const socket_manager_ptr& mgr) { ...@@ -509,11 +505,11 @@ short multiplexer::active_mask_of(const socket_manager_ptr& mgr) {
} }
bool multiplexer::is_reading(const socket_manager_ptr& mgr) { bool multiplexer::is_reading(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & input_mask); return (active_mask_of(mgr) & input_mask) != 0;
} }
bool multiplexer::is_writing(const socket_manager_ptr& mgr) { bool multiplexer::is_writing(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & output_mask); return (active_mask_of(mgr) & output_mask) != 0;
} }
// -- internal callbacks the pollset updater ----------------------------------- // -- internal callbacks the pollset updater -----------------------------------
...@@ -556,7 +552,7 @@ void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) { ...@@ -556,7 +552,7 @@ void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) {
if (!is_reading(mgr)) { if (!is_reading(mgr)) {
switch (mgr->handle_continue_reading()) { switch (mgr->handle_continue_reading()) {
default: // socket_manager::read_result::stop default: // socket_manager::read_result::stop
// Nothing to do, bitmask may remain unchanged (i.e., not reading). update_for(mgr).events &= ~input_mask;
break; break;
case socket_manager::read_result::again: case socket_manager::read_result::again:
update_for(mgr).events |= input_mask; update_for(mgr).events |= input_mask;
...@@ -566,7 +562,6 @@ void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) { ...@@ -566,7 +562,6 @@ void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) {
break; break;
case socket_manager::read_result::handover: { case socket_manager::read_result::handover: {
do_handover(mgr); do_handover(mgr);
return;
} }
} }
} }
...@@ -576,7 +571,7 @@ void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) { ...@@ -576,7 +571,7 @@ void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) {
if (!is_writing(mgr)) { if (!is_writing(mgr)) {
switch (mgr->handle_continue_writing()) { switch (mgr->handle_continue_writing()) {
default: // socket_manager::read_result::stop default: // socket_manager::read_result::stop
// Nothing to do, bitmask may remain unchanged (i.e., not writing). update_for(mgr).events &= ~output_mask;
break; break;
case socket_manager::write_result::again: case socket_manager::write_result::again:
update_for(mgr).events |= output_mask; update_for(mgr).events |= output_mask;
...@@ -586,7 +581,6 @@ void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) { ...@@ -586,7 +581,6 @@ void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) {
break; break;
case socket_manager::write_result::handover: { case socket_manager::write_result::handover: {
do_handover(mgr); do_handover(mgr);
return;
} }
} }
} }
......
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