Fix classic race with atomics + mutex

[victimsnino]

https://cohost.org/zeux/post/520125-condition-variables

[github-actions]

BENCHMARK RESULTS (AUTOGENERATED)

ci-ubuntu-gcc

General

name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	305.52 ns	2.16 ns	2.16 ns	1.00
Subscribe empty callbacks to empty observable via pipe operator	302.20 ns	2.16 ns	2.16 ns	1.00

Sources

name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	699.17 ns	0.62 ns	0.31 ns	2.00
from array of 1 - create + subscribe + current_thread	1042.98 ns	4.63 ns	5.25 ns	0.88
concat_as_source of just(1 immediate) create + subscribe	2254.94 ns	114.82 ns	118.81 ns	0.97
defer from array of 1 - defer + create + subscribe + immediate	729.89 ns	0.31 ns	0.31 ns	1.00
interval - interval + take(3) + subscribe + immediate	2126.46 ns	59.19 ns	58.04 ns	1.02
interval - interval + take(3) + subscribe + current_thread	3082.02 ns	32.73 ns	32.12 ns	1.02

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take(1)+subscribe	1079.01 ns	0.31 ns	0.31 ns	1.00
immediate_just+filter(true)+subscribe	846.21 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,2)+skip(1)+subscribe	982.52 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe	843.20 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,2)+first()+subscribe	1232.21 ns	0.62 ns	0.62 ns	1.00
immediate_just(1,2)+last()+subscribe	921.79 ns	0.31 ns	0.31 ns	1.00
immediate_just+take_last(1)+subscribe	1106.33 ns	18.53 ns	18.53 ns	1.00

Schedulers

name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	267.06 ns	2.16 ns	2.16 ns	1.00
current_thread scheduler create worker + schedule	374.87 ns	6.81 ns	7.41 ns	0.92
current_thread scheduler create worker + schedule + recursive schedule	849.96 ns	63.85 ns	64.24 ns	0.99

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+map(v*2)+subscribe	848.46 ns	0.31 ns	0.31 ns	1.00
immediate_just+scan(10, std::plus)+subscribe	894.17 ns	0.31 ns	0.31 ns	1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe	2323.59 ns	160.09 ns	176.23 ns	0.91
immediate_just+buffer(2)+subscribe	1513.80 ns	14.19 ns	14.52 ns	0.98
immediate_just+window(2)+subscribe + subscsribe inner	2393.38 ns	1041.11 ns	1079.43 ns	0.96

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take_while(false)+subscribe	833.43 ns	-	-	0.00
immediate_just+take_while(true)+subscribe	864.45 ns	0.31 ns	0.31 ns	1.00

Utility Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(1)+subscribe_on(immediate)+subscribe	1978.83 ns	0.31 ns	0.31 ns	1.00

Combining Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	3400.52 ns	175.54 ns	193.09 ns	0.91
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	3598.02 ns	164.18 ns	181.09 ns	0.91
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	138.99 ns	148.62 ns	0.94
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	3537.57 ns	955.05 ns	968.66 ns	0.99
immediate_just(1) + zip(immediate_just(2)) + subscribe	2109.06 ns	211.77 ns	226.45 ns	0.94

Subjects

name	rxcpp	rpp	prev rpp	ratio
publish_subject with 1 observer - on_next	34.59 ns	12.02 ns	12.03 ns	1.00

Scenarios

name	rxcpp	rpp	prev rpp	ratio
basic sample	1357.92 ns	15.74 ns	16.37 ns	0.96
basic sample with immediate scheduler	1403.21 ns	5.55 ns	5.55 ns	1.00

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+reduce(10, std::plus)+subscribe	910.03 ns	0.31 ns	0.31 ns	1.00

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	1070.69 ns	121.25 ns	121.73 ns	1.00

ci-macos

General

name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	989.63 ns	3.86 ns	3.84 ns	1.01
Subscribe empty callbacks to empty observable via pipe operator	1024.53 ns	4.14 ns	3.83 ns	1.08

Sources

name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	1937.27 ns	0.24 ns	0.24 ns	1.00
from array of 1 - create + subscribe + current_thread	2472.19 ns	25.73 ns	25.32 ns	1.02
concat_as_source of just(1 immediate) create + subscribe	5470.54 ns	328.00 ns	324.84 ns	1.01
defer from array of 1 - defer + create + subscribe + immediate	1993.95 ns	0.24 ns	0.23 ns	1.03
interval - interval + take(3) + subscribe + immediate	4998.80 ns	115.67 ns	118.79 ns	0.97
interval - interval + take(3) + subscribe + current_thread	6181.26 ns	108.63 ns	118.43 ns	0.92

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take(1)+subscribe	2873.76 ns	0.23 ns	0.30 ns	0.79
immediate_just+filter(true)+subscribe	2139.52 ns	0.24 ns	0.28 ns	0.85
immediate_just(1,2)+skip(1)+subscribe	3093.11 ns	0.24 ns	0.27 ns	0.86
immediate_just(1,1,2)+distinct_until_changed()+subscribe	2089.26 ns	0.47 ns	0.56 ns	0.85
immediate_just(1,2)+first()+subscribe	3202.56 ns	0.23 ns	0.27 ns	0.86
immediate_just(1,2)+last()+subscribe	2370.78 ns	0.24 ns	0.27 ns	0.87
immediate_just+take_last(1)+subscribe	3031.14 ns	73.66 ns	88.93 ns	0.83

Schedulers

name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	857.47 ns	4.18 ns	4.70 ns	0.89
current_thread scheduler create worker + schedule	1189.02 ns	38.13 ns	46.30 ns	0.82
current_thread scheduler create worker + schedule + recursive schedule	2077.34 ns	224.41 ns	269.42 ns	0.83

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+map(v*2)+subscribe	2123.39 ns	0.23 ns	0.23 ns	1.00
immediate_just+scan(10, std::plus)+subscribe	2342.84 ns	0.47 ns	0.54 ns	0.87
immediate_just+flat_map(immediate_just(v*2))+subscribe	5340.79 ns	414.68 ns	496.39 ns	0.84
immediate_just+buffer(2)+subscribe	2475.28 ns	71.17 ns	78.22 ns	0.91
immediate_just+window(2)+subscribe + subscsribe inner	5347.96 ns	2328.74 ns	2998.44 ns	0.78

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take_while(false)+subscribe	2274.30 ns	-	-	0.00
immediate_just+take_while(true)+subscribe	2124.87 ns	0.24 ns	0.26 ns	0.94

Utility Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(1)+subscribe_on(immediate)+subscribe	4984.64 ns	0.25 ns	0.28 ns	0.88

Combining Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	7408.56 ns	452.69 ns	550.93 ns	0.82
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	8564.46 ns	456.67 ns	552.95 ns	0.83
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	476.03 ns	563.03 ns	0.85
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	8106.53 ns	1892.22 ns	2362.13 ns	0.80
immediate_just(1) + zip(immediate_just(2)) + subscribe	5157.81 ns	861.63 ns	1012.54 ns	0.85

Subjects

name	rxcpp	rpp	prev rpp	ratio
publish_subject with 1 observer - on_next	75.67 ns	49.35 ns	60.51 ns	0.82

Scenarios

name	rxcpp	rpp	prev rpp	ratio
basic sample	2773.18 ns	102.68 ns	120.80 ns	0.85
basic sample with immediate scheduler	2957.28 ns	15.88 ns	17.06 ns	0.93

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+reduce(10, std::plus)+subscribe	2408.39 ns	0.24 ns	0.26 ns	0.91

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	6508.66 ns	4057.81 ns	5380.62 ns	0.75

ci-ubuntu-clang

General

name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	271.52 ns	0.88 ns	1.54 ns	0.57
Subscribe empty callbacks to empty observable via pipe operator	269.18 ns	0.88 ns	1.54 ns	0.57

Sources

name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	575.29 ns	0.42 ns	0.31 ns	1.35
from array of 1 - create + subscribe + current_thread	801.24 ns	5.55 ns	5.56 ns	1.00
concat_as_source of just(1 immediate) create + subscribe	1935.52 ns	114.23 ns	112.38 ns	1.02
defer from array of 1 - defer + create + subscribe + immediate	589.25 ns	0.31 ns	0.31 ns	1.00
interval - interval + take(3) + subscribe + immediate	1513.07 ns	57.09 ns	57.09 ns	1.00
interval - interval + take(3) + subscribe + current_thread	2072.91 ns	30.85 ns	30.92 ns	1.00

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take(1)+subscribe	934.82 ns	0.31 ns	0.31 ns	1.00
immediate_just+filter(true)+subscribe	659.35 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,2)+skip(1)+subscribe	856.01 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe	689.83 ns	0.62 ns	0.62 ns	1.00
immediate_just(1,2)+first()+subscribe	1048.10 ns	0.31 ns	0.31 ns	1.00
immediate_just(1,2)+last()+subscribe	728.78 ns	0.31 ns	0.31 ns	1.00
immediate_just+take_last(1)+subscribe	967.82 ns	0.31 ns	0.31 ns	1.00

Schedulers

name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	198.26 ns	0.88 ns	1.54 ns	0.57
current_thread scheduler create worker + schedule	307.44 ns	5.88 ns	5.57 ns	1.06
current_thread scheduler create worker + schedule + recursive schedule	618.90 ns	58.85 ns	58.74 ns	1.00

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+map(v*2)+subscribe	658.10 ns	0.31 ns	0.31 ns	1.00
immediate_just+scan(10, std::plus)+subscribe	709.40 ns	0.31 ns	0.31 ns	1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe	1791.86 ns	121.15 ns	121.61 ns	1.00
immediate_just+buffer(2)+subscribe	1345.91 ns	14.49 ns	14.20 ns	1.02
immediate_just+window(2)+subscribe + subscsribe inner	2177.05 ns	803.88 ns	809.17 ns	0.99

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take_while(false)+subscribe	654.98 ns	-	-	0.00
immediate_just+take_while(true)+subscribe	644.80 ns	0.31 ns	0.31 ns	1.00

Utility Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(1)+subscribe_on(immediate)+subscribe	1584.45 ns	0.31 ns	0.31 ns	1.00

Combining Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	2510.73 ns	127.05 ns	125.68 ns	1.01
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	2990.60 ns	121.01 ns	121.29 ns	1.00
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	112.53 ns	112.35 ns	1.00
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	2635.64 ns	727.66 ns	726.02 ns	1.00
immediate_just(1) + zip(immediate_just(2)) + subscribe	1768.03 ns	175.18 ns	172.89 ns	1.01

Subjects

name	rxcpp	rpp	prev rpp	ratio
publish_subject with 1 observer - on_next	26.32 ns	14.80 ns	14.19 ns	1.04

Scenarios

name	rxcpp	rpp	prev rpp	ratio
basic sample	1027.15 ns	13.57 ns	13.58 ns	1.00
basic sample with immediate scheduler	1047.07 ns	5.86 ns	5.86 ns	1.00

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+reduce(10, std::plus)+subscribe	737.67 ns	0.31 ns	0.31 ns	1.00

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	849.42 ns	126.96 ns	125.71 ns	1.01

ci-windows

General

name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	583.71 ns	4.94 ns	4.94 ns	1.00
Subscribe empty callbacks to empty observable via pipe operator	600.52 ns	4.94 ns	4.93 ns	1.00

Sources

name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	1180.18 ns	5.55 ns	5.56 ns	1.00
from array of 1 - create + subscribe + current_thread	1463.35 ns	18.51 ns	18.82 ns	0.98
concat_as_source of just(1 immediate) create + subscribe	4632.26 ns	172.36 ns	172.47 ns	1.00
defer from array of 1 - defer + create + subscribe + immediate	1210.24 ns	5.86 ns	5.86 ns	1.00
interval - interval + take(3) + subscribe + immediate	2987.99 ns	133.64 ns	133.69 ns	1.00
interval - interval + take(3) + subscribe + current_thread	3478.83 ns	58.72 ns	58.72 ns	1.00

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take(1)+subscribe	1863.59 ns	12.87 ns	12.87 ns	1.00
immediate_just+filter(true)+subscribe	1389.67 ns	11.72 ns	11.72 ns	1.00
immediate_just(1,2)+skip(1)+subscribe	1825.47 ns	13.22 ns	13.20 ns	1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe	1407.98 ns	15.77 ns	15.77 ns	1.00
immediate_just(1,2)+first()+subscribe	2123.92 ns	12.95 ns	12.96 ns	1.00
immediate_just(1,2)+last()+subscribe	1868.66 ns	14.12 ns	14.17 ns	1.00
immediate_just+take_last(1)+subscribe	2059.63 ns	58.82 ns	59.25 ns	0.99

Schedulers

name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	502.54 ns	6.17 ns	6.17 ns	1.00
current_thread scheduler create worker + schedule	676.79 ns	22.21 ns	18.20 ns	1.22
current_thread scheduler create worker + schedule + recursive schedule	1116.28 ns	112.32 ns	107.78 ns	1.04

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+map(v*2)+subscribe	1328.22 ns	10.94 ns	11.39 ns	0.96
immediate_just+scan(10, std::plus)+subscribe	1437.54 ns	21.58 ns	21.58 ns	1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe	3571.43 ns	226.29 ns	231.21 ns	0.98
immediate_just+buffer(2)+subscribe	2335.56 ns	58.63 ns	60.13 ns	0.98
immediate_just+window(2)+subscribe + subscsribe inner	4123.17 ns	1555.16 ns	1561.20 ns	1.00

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+take_while(false)+subscribe	1327.55 ns	11.46 ns	11.44 ns	1.00
immediate_just+take_while(true)+subscribe	1339.45 ns	11.73 ns	11.72 ns	1.00

Utility Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(1)+subscribe_on(immediate)+subscribe	4140.89 ns	8.02 ns	8.02 ns	1.00

Combining Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	5167.88 ns	244.74 ns	244.80 ns	1.00
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	6503.41 ns	235.02 ns	235.78 ns	1.00
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	238.39 ns	242.04 ns	0.98
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	5885.71 ns	966.10 ns	951.77 ns	1.02
immediate_just(1) + zip(immediate_just(2)) + subscribe	3631.21 ns	535.56 ns	536.42 ns	1.00

Subjects

name	rxcpp	rpp	prev rpp	ratio
publish_subject with 1 observer - on_next	37.13 ns	25.90 ns	25.90 ns	1.00

Scenarios

name	rxcpp	rpp	prev rpp	ratio
basic sample	1888.91 ns	59.80 ns	59.12 ns	1.01
basic sample with immediate scheduler	2230.87 ns	37.03 ns	37.03 ns	1.00

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio
immediate_just+reduce(10, std::plus)+subscribe	1480.16 ns	19.96 ns	19.95 ns	1.00

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	1978.79 ns	336.63 ns	342.14 ns	0.98

[sonarqubecloud]

Quality Gate failed

Failed conditions
B Maintainability Rating on New Code (required ≥ A)

See analysis details on SonarCloud

Catch issues before they fail your Quality Gate with our IDE extension SonarLint