perf improvements

4 files changed, 11 insertions, 10 deletions

diff --git a/Rx/v2/test/operators/concat.cpp b/Rx/v2/test/operators/concat.cpp
index ddd6145..864166c 100644
--- a/Rx/v2/test/operators/concat.cpp
+++ b/Rx/v2/test/operators/concat.cpp
@@ -29,7 +29,7 @@ SCENARIO("synchronize concat ranges", "[hide][range][synchronize][concat][perf]"
             //auto sc = rxsc::make_new_thread();
             auto so = rxsub::synchronize_observable(sc);
 
-            int c = 0;
+            std::atomic<int> c(0);
             int n = 1;
             auto sectionCount = onnextcalls / 3;
             auto start = clock::now();
@@ -46,7 +46,7 @@ SCENARIO("synchronize concat ranges", "[hide][range][synchronize][concat][perf]"
                         wake.notify_one();});
 
             std::unique_lock<std::mutex> guard(lock);
-            wake.wait(guard);
+            wake.wait(guard, [&](){return c == onnextcalls;});
 
             auto finish = clock::now();
             auto msElapsed = duration_cast<milliseconds>(finish.time_since_epoch()) -
diff --git a/Rx/v2/test/operators/concat_map.cpp b/Rx/v2/test/operators/concat_map.cpp
index 322d92d..058a645 100644
--- a/Rx/v2/test/operators/concat_map.cpp
+++ b/Rx/v2/test/operators/concat_map.cpp
@@ -73,7 +73,7 @@ SCENARIO("synchronize concat_map pythagorian ranges", "[hide][range][concat_map]
             auto so = rxsub::synchronize_observable(sc);
 
             int c = 0;
-            int ct = 0;
+            std::atomic<int> ct(0);
             int n = 1;
             auto start = clock::now();
             auto triples =
@@ -109,7 +109,7 @@ SCENARIO("synchronize concat_map pythagorian ranges", "[hide][range][concat_map]
                         wake.notify_one();});
 
             std::unique_lock<std::mutex> guard(lock);
-            wake.wait(guard);
+            wake.wait(guard, [&](){return ct == tripletCount;});
 
             auto finish = clock::now();
             auto msElapsed = duration_cast<milliseconds>(finish.time_since_epoch()) -
diff --git a/Rx/v2/test/operators/flat_map.cpp b/Rx/v2/test/operators/flat_map.cpp
index 7b63664..cb9b2a0 100644
--- a/Rx/v2/test/operators/flat_map.cpp
+++ b/Rx/v2/test/operators/flat_map.cpp
@@ -110,7 +110,7 @@ SCENARIO("synchronize flat_map pythagorian ranges", "[hide][range][flat_map][syn
             auto so = rxsub::synchronize_observable(sc);
 
             int c = 0;
-            int ct = 0;
+            std::atomic<int> ct(0);
             int n = 1;
             auto start = clock::now();
             auto triples =
@@ -146,7 +146,7 @@ SCENARIO("synchronize flat_map pythagorian ranges", "[hide][range][flat_map][syn
                         wake.notify_one();});
 
             std::unique_lock<std::mutex> guard(lock);
-            wake.wait(guard);
+            wake.wait(guard, [&](){return ct == tripletCount;});
 
             auto finish = clock::now();
             auto msElapsed = duration_cast<milliseconds>(finish.time_since_epoch()) -
diff --git a/Rx/v2/test/operators/merge.cpp b/Rx/v2/test/operators/merge.cpp
index 77f2033..931d26f 100644
--- a/Rx/v2/test/operators/merge.cpp
+++ b/Rx/v2/test/operators/merge.cpp
@@ -29,24 +29,25 @@ SCENARIO("synchronize merge ranges", "[hide][range][synchronize][merge][perf]"){
             //auto sc = rxsc::make_new_thread();
             auto so = rxsub::synchronize_observable(sc);
 
-            int c = 0;
+            std::atomic<int> c(0);
             int n = 1;
             auto sectionCount = onnextcalls / 3;
             auto start = clock::now();
             rxs::range(0, sectionCount - 1, 1, sc)
                 .merge(
                     so,
-                    rxs::range(sectionCount, sectionCount * 2 - 1, 1, sc),
+                    rxs::range(sectionCount, (sectionCount * 2) - 1, 1, sc),
                     rxs::range(sectionCount * 2, onnextcalls - 1, 1, sc))
                 .subscribe(
                     [&c](int x){
                         ++c;},
                     [](std::exception_ptr){abort();},
                     [&](){
-                        wake.notify_one();});
+                        wake.notify_one();
+                    });
 
             std::unique_lock<std::mutex> guard(lock);
-            wake.wait(guard);
+            wake.wait(guard, [&](){return c == onnextcalls;});
 
             auto finish = clock::now();
             auto msElapsed = duration_cast<milliseconds>(finish.time_since_epoch()) -