Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

try to catch issue with subjects #697

Merged
merged 10 commits into from
Jan 11, 2025
Merged

try to catch issue with subjects #697

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
ddd6e9a
try to catch issue
victimsnino Jan 10, 2025
1529116
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jan 10, 2025
a0691f6
Update test_subjects.cpp
victimsnino Jan 10, 2025
ae5cbc7
Update test_subjects.cpp
victimsnino Jan 10, 2025
7b0bc3c
Update test_subjects.cpp
victimsnino Jan 10, 2025
73c834e
fix
victimsnino Jan 10, 2025
b2bd784
fix
victimsnino Jan 10, 2025
5dba2e2
fix
victimsnino Jan 11, 2025
f180ef3
speedup
victimsnino Jan 11, 2025
622be88
fix
victimsnino Jan 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions src/rpp/rpp/subjects/details/subject_state.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,13 +121,16 @@
if (!observers)
return;

const auto begin = observers->cbegin();
const auto end = observers->cend();
auto itr = observers->cbegin();
const auto size = observers->size();

observers_lock.unlock();

std::lock_guard lock{m_serialized_mutex};
std::for_each(begin, end, [&](const observer& obs) { obs->on_next(v); });
for (size_t i = 0; i < size; ++i)
{
(itr++)->on_next(v);

Check failure on line 132 in src/rpp/rpp/subjects/details/subject_state.hpp

View workflow job for this annotation

GitHub Actions / Sanitize - asan 

no member named 'on_next' in 'std::shared_ptr<rpp::details::observers::observer_vtable<int>>'

Check failure on line 132 in src/rpp/rpp/subjects/details/subject_state.hpp

View workflow job for this annotation

GitHub Actions / Sanitize - lsan 

no member named 'on_next' in 'std::shared_ptr<rpp::details::observers::observer_vtable<int>>'

Check failure on line 132 in src/rpp/rpp/subjects/details/subject_state.hpp

View workflow job for this annotation

GitHub Actions / benchmarks ci-macos Release (Optimizations disabled) 

no member named 'on_next' in 'std::shared_ptr<rpp::details::observers::observer_vtable<int>>'

Check failure on line 132 in src/rpp/rpp/subjects/details/subject_state.hpp

View workflow job for this annotation

GitHub Actions / benchmarks ci-ubuntu-clang Release 

no member named 'on_next' in 'std::shared_ptr<rpp::details::observers::observer_vtable<int>>'

Check failure on line 132 in src/rpp/rpp/subjects/details/subject_state.hpp

View workflow job for this annotation

GitHub Actions / Sanitize - ubsan 

no member named 'on_next' in 'std::shared_ptr<rpp::details::observers::observer_vtable<int>>'
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Critical: Fix potential race conditions and iterator invalidation

The current implementation has several thread-safety issues:

  1. Race Condition: Capturing size and releasing the lock creates a time window where the deque could be modified, making the size inconsistent with the actual number of observers.
  2. Iterator Invalidation: The iterator might be invalidated if the deque is modified after unlocking.
  3. Memory Safety: The shared_ptr is released while keeping its iterator, which could lead to undefined behavior.

Consider this safer implementation:

-            auto       itr  = observers->cbegin();
-            const auto size = observers->size();
-
-            observers_lock.unlock();
-
-            std::lock_guard lock{m_serialized_mutex};
-            for (size_t i = 0; i < size; ++i)
-            {
-                (itr++)->on_next(v);
-            }
+            std::lock_guard serialized_lock{m_serialized_mutex};
+            // Keep shared_ptr alive during iteration
+            for (const auto& observer : *observers)
+            {
+                observer->on_next(v);
+            }

This solution:

  • Maintains the shared_ptr throughout iteration
  • Prevents iterator invalidation
  • Eliminates the race condition window
  • Simplifies the code
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
auto itr = observers->cbegin();
const auto size = observers->size();
observers_lock.unlock();
std::lock_guard lock{m_serialized_mutex};
std::for_each(begin, end, [&](const observer& obs) { obs->on_next(v); });
for (size_t i = 0; i < size; ++i)
{
(itr++)->on_next(v);
}
std::lock_guard serialized_lock{m_serialized_mutex};
// Keep shared_ptr alive during iteration
for (const auto& observer : *observers)
{
observer->on_next(v);
}

}

void on_error(const std::exception_ptr& err)
Expand Down
20 changes: 20 additions & 0 deletions src/tests/rpp/test_subjects.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,26 @@ TEST_CASE("subject can be modified from on_next call")
}
}

TEST_CASE("subject handles addition from inside on_next properly")
{
rpp::subjects::publish_subject<int> subject{};

SUBCASE("subscribe inside on_next")
{
int value = {};
subject.get_observable().subscribe([&subject, &value](int v) {
for (int i = 0; i < 100; ++i)
subject.get_observable().subscribe([](int) {});
value = v;
Comment on lines +179 to +182
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Improve subscription management in test

The test creates 100 subscriptions without proper cleanup, which could lead to resource leaks.

Apply this diff to properly manage subscriptions:

-            for (int i = 0; i < 100; ++i)
-                subject.get_observable().subscribe([](int) {});
+            auto d = rpp::composite_disposable_wrapper::make();
+            for (int i = 0; i < 100; ++i)
+                subject.get_observable().subscribe(d, [](int) {});

Committable suggestion skipped: line range outside the PR's diff.

});

for (int i = 0; i < 100; ++i)
subject.get_observer().on_next(i);

REQUIRE(value == 99);
}
}

TEST_CASE("publish subject caches error/completed")
{
auto mock = mock_observer_strategy<int>{};
Expand Down
Loading