iox-#2177 Update SPSCSoFi #2178
Conversation
- add comments to clarify concurrent behavior - fix memory orders Signed-off-by: albtam <[email protected]>
|
@elBoberido Same as the FiFo, could you please review the changes? |
iceoryx_hoofs/concurrent/buffer/include/iox/detail/spsc_sofi.hpp
Outdated
Show resolved
Hide resolved
|
|
||
| /// @brief This is the resulting internal size on creation | ||
| static constexpr uint32_t INTERNAL_SPSC_SOFI_SIZE = CapacityValue + INTERNAL_SIZE_ADD_ON; | ||
| // A limitation of the current implementation (first write and then read in the push() method) is |
There was a problem hiding this comment.
I would phrase this a bit different. Without the additional memory there would be these four steps
- advance the read position (this effectively reduces the capacity and is the reason the internal capacity needs to be larger; the consumer cannot pop out CAPACITY amount of samples even though the queue is full if the push thread is suspended right after this operation)
- copy the read data to the out variable
- copy the write data to the free slot
- advance the write position
Therefore a better phrasing would be: To ensure a consumer gets at least the amount of capacity of data when a queue is full, an additional free slot is required.
Followed by the explanation why this is not possible without the additional free slot and then the explanation with the free slot.
| // ======================================================================== | ||
| // Consider the following scenario when there is no "capacity add-on": | ||
| // 1. CapacityValue = 2 | ||
| // |--1--|--2--| |
There was a problem hiding this comment.
Can you please use letters like A, B, C to make a clear distictionbetween data and indices
| // 2. We want to push a new element: | ||
| // - we first write at index 2 % size | ||
| // - we detect that the queue if full so we want to read at r=0 | ||
| // - we cannot read anymore the value since we've just overwritten it |
There was a problem hiding this comment.
This is a bit confusion and I'm a bit lost. I think I know what you want to say because I know the underlying problem but without that knowledge I would be totally lost 😅
I think an ASCII art will tell more than a thousand words
// initial state
+-----+-----+
| A | B |
+-----+-----+
w | r
---
// advance next read position
+-----+-----+
| A | B |
+-----+-----+
w r
---
// take overflow data
+-----+-----+
| - | B |
+-----+-----+
w r
---
// write new data
+-----+-----+
| C | B |
+-----+-----+
w r
---
// advance next write position
+-----+-----+
| C | B |
+-----+-----+
w | r
With this it is clear that there is a small Window where the queue holds less data than CAPACITY although the queue should be full
| uint64_t currentReadPos = m_readPosition.load(std::memory_order_relaxed); | ||
|
|
||
| bool popWasSuccessful{true}; | ||
| do | ||
| { | ||
| if (currentReadPosition == m_writePosition.load(std::memory_order_acquire)) | ||
| // SYNC POINT READ: m_data | ||
| // See explanation of the corresponding synchronization point in push() | ||
| if (currentReadPos == m_writePosition.load(std::memory_order_acquire)) | ||
| { | ||
| nextReadPosition = currentReadPosition; | ||
| popWasSuccessful = false; | ||
| return false; | ||
| // We don't need to check if read has changed, as it is enough to know that the empty state | ||
| // was valid in the past. The same race can also happen after the while loop and before the | ||
| // return operation | ||
| } |
There was a problem hiding this comment.
You cannot do this. The code you removed earned me a crate of beer after someone claimed cough @budrus cough that the SOFI would be bug free. I'm still waiting for the beer though. Yes, I'm looking at you @budrus 🤣
The CPU can reorder the load of m_readPosition after the load of m_writePosition resulting in the hallucination that the SOFI is empty although it is full.
Unfortunately this cannot be triggered on x86 and our stress tests currently also do not cover this scenario. I found the issue with the RacerPeppi, which does something similar to https://www.1024cores.net/home/relacy-race-detector but in a more primitive manner. I would guess that MIRI will also find these issues when the code is ported to Rust.
There was a problem hiding this comment.
Just for clarification, here some ASCII art
Start condition
- SOFI with an internal capacity of 3
- read position = 301
- read index = 1
- write position = 303
- write index = 0
| |
+-----+-----+-----+-----+-----+-----+-----+
| - | B | C | - | - | - | - | data
+-----+-----+-----+-----+-----+-----+-----+
| 0 | 1 | 2 | 0 | 1 | 2 | 0 | index
+-----+-----+-----+-----+-----+-----+-----+
| 300 | 301 | 302 | 303 | 304 | 305 | 306 | position
+-----+-----+-----+-----+-----+-----+-----+
r | w |
Now the following happens
- consumer: m_writePosition.load(std::memory_order_acquire) -> 303
- producer: push two times -> read position move to 303
| |
+-----+-----+-----+-----+-----+-----+-----+
| - | - | - | D | E | - | - | data
+-----+-----+-----+-----+-----+-----+-----+
| 0 | 1 | 2 | 0 | 1 | 2 | 0 | index
+-----+-----+-----+-----+-----+-----+-----+
| 300 | 301 | 302 | 303 | 304 | 305 | 306 | position
+-----+-----+-----+-----+-----+-----+-----+
| r w |
- consumer: m_readPosition.load(std::memory_order_relaxed) -> 303 is stored in currentReadPos
-> if (303 == 303)
-> return false
There was a problem hiding this comment.
You have a point that this comment is probably wrong:
operation reordering: cannot compile if moves down
About:
"resulting in the hallucination that the SOFI is empty although it is full"
isn't something that is inherent to a concurrent queue, i.e. you call push and then pop, pop doesn't return anything. You call again pop and this time, it doesn't return a value.
There was a problem hiding this comment.
isn't something that is inherent to a concurrent queue, i.e. you call push and then pop, pop doesn't return anything. You call again pop and this time, it doesn't return a value.
I guess you refer to this comment #2168 (comment)
It is different in this case. Just assume what would happen on the consumer side if there wasn't a concurrent push on the producer side. In the linked comment, the result would be the same, an empty pop. The situation here is different. In this scenario the pop would not be empty if there were no concurrent push, which means the queue was not empty an therefore it is a bug and not one of these inherent races. These inherent races affect only future events, not past ones.
You can also imagine it as a queue with a capacity of 1 (internally it will be two due to the addon). Let's further assume the producer threads pushes ten times as often as the consumer thread pops. I think we can agree that there should always be data in the queue when the pop occurs. Which this hallucination it could theoretically look like the queue is empty every time the consumer thread tries to pop. Granted, it is not likely to happen all the time (it's a race after all) but is shows that the algorithm is broken.
| } while (!m_readPosition.compare_exchange_weak( | ||
| currentReadPosition, nextReadPosition, std::memory_order_acq_rel, std::memory_order_acquire)); | ||
| currentReadPos, currentReadPos + 1U, std::memory_order_relaxed, std::memory_order_relaxed)); |
There was a problem hiding this comment.
I have to think about this change a bit more. We might still need to synchronize with a corresponding m_readPos release operation.
| if (m_readPosition.compare_exchange_strong( | ||
| currentReadPosition, nextReadPosition, std::memory_order_acq_rel, std::memory_order_relaxed)) | ||
| currentReadPos, currentReadPos + 1U, std::memory_order_relaxed, std::memory_order_relaxed)) |
There was a problem hiding this comment.
Please revert all the changes from this method. I would bet the crate of beer @budrus still owes me that test_stress_spsc_sofi.cpp will fail when run on a CPU with a weakly ordered CPU like on an armv8.
Keep in mind, the CPU is allowed to do quite a lot of reordering, e.g.
uint64_t currentWritePos = m_writePosition.load(std::memory_order_acquire);
uint64_t nextWritePos = currentWritePos + 1U;
uint64_t currentReadPos = m_readPosition.load(std::memory_order_relaxed);
if (nextWritePos < currentReadPos + m_size) {} // woohoo, there is an overflow and I do not need execute 'm_data[currentWritePos % m_size] = valueIn;' and 'm_writePosition.store(nextWritePos, std::memory_order_release);' and return; lets continue the ride
if (m_readPosition.compare_exchange_strong(
currentReadPos, currentReadPos + 1U, std::memory_order_relaxed, std::memory_order_relaxed))
{
// hehe, take this INTERNAL_CAPACITY_ADDON, I'm still able to break the contract to have at least CapacityValue of data in a full queue ;)
std::memcpy(&valueOut, &m_data[currentReadPos % m_size], sizeof(ValueType));
// oh no, I have to return; guess it is time to write the value
m_data[currentWritePos % m_size] = valueIn;
m_writePosition.store(nextWritePos, std::memory_order_release);
return SOFI_OVERFLOW;Signed-off-by: albtam <[email protected]>
|
@elBoberido I tried to port the SpscSoFi to Rust. Somehow miri always reports a data race. Any idea what could be wrong? |
Signed-off-by: albtam <[email protected]>
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## master #2178 +/- ##
==========================================
- Coverage 80.17% 79.96% -0.21%
==========================================
Files 421 420 -1
Lines 16346 16349 +3
Branches 2265 2273 +8
==========================================
- Hits 13105 13074 -31
- Misses 2431 2439 +8
- Partials 810 836 +26
Flags with carried forward coverage won't be shown. Click here to find out more.
|
|
@albtam up until now I did not run miri with a compare and swap loop. It might be that miri does not understand that the value is not used but discarded or there is a bug in the implementation. Can you try to port the current algorithm from upstream? I know that miri got a feature to fail on 80% on compare_exchange_weak to simulate spurious wakeups. So it might be that it works in general with compare and swap loops but not with the specifics in SOFI |
|
@albtam I could make your example pass miri by making both compare and swap loops use either But when increasing the number of pushes a lot it still detects (false positive?) races. The tests also need to be done in a different way, e.g. the cancelation points of the loop become more complex due to the overflow behavior. |
Pre-Review Checklist for the PR Author
iox-123-this-is-a-branch)iox-#123 commit text)task-list-completed)iceoryx_hoofsare added to./clang-tidy-diff-scans.txtNotes for Reviewer
Checklist for the PR Reviewer
iceoryx_hoofshave been added to./clang-tidy-diff-scans.txtPost-review Checklist for the PR Author
References