Это реализация для нескольких производителей и одного потребителя, переведенная с Rust, для вопросов языкового юриста, переписанная на C++
template
struct Node{
std::atomic next;
T value;
Node(T v):value(v),next(){}
};
template
struct Queue {
std::atomic head;
Node* tail;
Queue(){
auto h = new Node{T{}};
head.store(h);
tail = h;
}
void push(T t){
auto node = new Node{t};
auto pre = this->head.exchange(node,std::memory_order::acq_rel);
pre->next.store(node,std::memory_order::release);
}
T pop(){
auto tail = this->tail;
auto next = tail->next.load(std::memory_order::acquire);
if(next){
this->tail = next;
auto ret = next->value;
delete tail;
return ret;
}
if (this->head.load(std::memory_order::acquire) == tail){ // #1
throw "empty";
}else{
throw "Inconsistent";
}
}
};
Интересно, почему порядок памяти в #1 должен быть Acquire? Цель состоит только в том, чтобы сравнить, равны ли хвост значения указателя и значение указателя, хранящееся в this->head, что не позволяет получить доступ к референту, на который указывает значение указателя. В отличие от вышеописанного, где требуется доступ к референту после загрузки, для этого требуется, чтобы инициализация узла произошла до доступа к узлу. Итак, почему порядок памяти в #1 не может быть Memory_order::relaxed?
Обновление:
Исходный код — mpsc_queue.rs. Вот как библиотека комментирует, когда возвращаемое значение является «непоследовательным».
PS: процитируйте исходный фрагмент кода здесь
//! A mostly lock-free multi-producer, single consumer queue.
//!
//! This module contains an implementation of a concurrent MPSC queue. This
//! queue can be used to share data between threads, and is also used as the
//! building block of channels in rust.
//!
//! Note that the current implementation of this queue has a caveat of the `pop`
//! method, and see the method for more information about it. Due to this
//! caveat, this queue might not be appropriate for all use-cases.
// https://www.1024cores.net/home/lock-free-algorithms
// /queues/non-intrusive-mpsc-node-based-queue
#[cfg(all(test, not(target_os = "emscripten")))]
mod tests;
pub use self::PopResult::*;
use core::cell::UnsafeCell;
use core::ptr;
use crate::boxed::Box;
use crate::sync::atomic::{AtomicPtr, Ordering};
/// A result of the `pop` function.
pub enum PopResult {
/// Some data has been popped
Data(T),
/// The queue is empty
Empty,
/// The queue is in an inconsistent state. Popping data should succeed, but
/// some pushers have yet to make enough progress in order allow a pop to
/// succeed. It is recommended that a pop() occur "in the near future" in
/// order to see if the sender has made progress or not
Inconsistent,
}
struct Node {
next: AtomicPtr,
value: Option,
}
/// The multi-producer single-consumer structure. This is not cloneable, but it
/// may be safely shared so long as it is guaranteed that there is only one
/// popper at a time (many pushers are allowed).
pub struct Queue {
head: AtomicPtr,
tail: UnsafeCell,
}
unsafe impl Send for Queue {}
unsafe impl Sync for Queue {}
impl Node {
unsafe fn new(v: Option) -> *mut Node {
Box::into_raw(box Node { next: AtomicPtr::new(ptr::null_mut()), value: v })
}
}
impl Queue {
/// Creates a new queue that is safe to share among multiple producers and
/// one consumer.
pub fn new() -> Queue {
let stub = unsafe { Node::new(None) };
Queue { head: AtomicPtr::new(stub), tail: UnsafeCell::new(stub) }
}
/// Pushes a new value onto this queue.
pub fn push(&self, t: T) {
unsafe {
let n = Node::new(Some(t)); // #1
let prev = self.head.swap(n, Ordering::AcqRel); // #2
(*prev).next.store(n, Ordering::Release); // #3
}
}
/// Pops some data from this queue.
///
/// Note that the current implementation means that this function cannot
/// return `Option`. It is possible for this queue to be in an
/// inconsistent state where many pushes have succeeded and completely
/// finished, but pops cannot return `Some(t)`. This inconsistent state
/// happens when a pusher is pre-empted at an inopportune moment.
///
/// This inconsistent state means that this queue does indeed have data, but
/// it does not currently have access to it at this time.
pub fn pop(&self) -> PopResult {
unsafe {
let tail = *self.tail.get();
let next = (*tail).next.load(Ordering::Acquire); // #4
if !next.is_null() { // #5
*self.tail.get() = next;
assert!((*tail).value.is_none());
assert!((*next).value.is_some());
let ret = (*next).value.take().unwrap();
let _: Box = Box::from_raw(tail);
return Data(ret);
}
if self.head.load(Ordering::Acquire) == tail { Empty } else { Inconsistent }
}
}
}
impl Drop for Queue {
fn drop(&mut self) {
unsafe {
let mut cur = *self.tail.get();
while !cur.is_null() {
let next = (*cur).next.load(Ordering::Relaxed);
let _: Box = Box::from_raw(cur);
cur = next;
}
}
}
}
Подробнее здесь: https://stackoverflow.com/questions/797 ... ed-list-qu
