Мобильная версияКод: Выделить всё
template
struct Packet{
Queue queue;
atomic cnt;
void send(t: T) {
if self.cnt.load(memory_order::seq_cst) < DISCONNECTED + FUDGE {
throw "Err"
}
this->queue.push(t);
if(this->cnt.fetch_add(1, memory_order:seq_cst) == -1) {
this->take_to_wake().signal();
}
}
T recv(Option deadline) {
try{
auto t = this->try_recv();
return t;
}catch(e){
}
auto [wait_token, signal_token] = blocking::tokens();
if this->decrement(signal_token) == Installed {
wait_token.wait();
}
try{
auto t = this->try_recv();
this->steals.get() -= 1;
return t;
}catch(...){
}
}
fn decrement(SignalToken token) -> StartResult {
unsafe {
assert_eq(this->to_wake.load(memory_order:seq_cst), 0);
auto ptr = token.cast_to_usize();
this->to_wake.store(ptr, memory_order:seq_cst);
auto steals = ptr::replace(self.steals.get(), 0);
auto r = self.cnt.fetch_sub(1 + steals, memory_order:seq_cst);
if(r==DISCONNECTED ) {
this->cnt.store(DISCONNECTED, memory_order:seq_cst);
}else{
assert(n >= 0);
if n - steals to_wake.store(0, memory_order:seq_cst);
drop(SignalToken::cast_from_usize(ptr));
return Abort;
}
}
T try_recv() {
while(true){
try{
auto t = this->queue.pop();
return t;
}catch(e){
if(e=="Empty"){
throw e;
}
if(e=="Inconsistent"){
thread::yield_now();
}
}
}
}
};
Код: Выделить всё
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";
}
}
};
Предполагая, что данные отправляются тремя потоками. Согласно [intro.races] p14
Если побочный эффект X на атомарный объект M происходит до вычисления значения B для M, то оценка B берет свое значение из X или из побочного эффекта Y, который следует за X в порядке модификации M.
Если нет других дополнительных синхронизаций, поток pop не гарантирует, что увидит последующие узлы, установленные в нажимать потоки, потому что принцип «прежде чем ничего» не обеспечивает видимость; то есть увидеть эти изменения после первоначального побочного эффекта в порядке модификаций. Однако заголовок начального значения и его поле, следующее за ним со значением null, гарантированно будут видны.
Согласно [atomic.order] p10
Атомарные операции чтения-изменения-записи всегда должны считывать последнее значение (в порядке модификации), записанное перед записью, связанной с операцией чтения-изменения-записи.
Если одна из них RMW считывает начальное значение 0, другие операции RMW должны считывать более позднюю модификацию в порядке модификации, и никакие две операции RMW не могут считывать одну и ту же модификацию. Все операции над cnt являются операциями RMW; это гарантирует, что загрузочная часть операции RMW всегда может двигаться вперед и не может наступать на другую (т. е. все операции сериализуются и не могут перекрываться). При синхронизации, установленной сериализованными операциями RMW (т. е. cnt.fetch_xxx), при загрузке в next и head гарантированно будут отображаться более поздние значения в их порядках модификации, поскольку между побочными эффектами next или head и загрузками существуют отношения «до того, как произошло».
Каждый обмен в push может установить синхронизацию с другим. Однако, поскольку fetch_add(1,...) является упорядоченным после вызова push, не происходит событие-before, установленное синхронизацией в push для любых двух fetch_add, следовательно, нет никаких ограничений на эти fetch_add(1,...) для их порядка в порядке модификации; может быть случай, когда fetch_add, соответствующий первому узлу, отличному от head, может быть заказан позже в порядке его модификации. В частности, я попытался проиллюстрировать этот случай на следующем изображении:
Первое всплывающее окно начинается с этой головы и head->next имеет значение null, cnt.fetch_sub(1) синхронизируется с M1, это может гарантировать только то, что node2->next = node3 произойдет до try_recv(), который располагается после cnt.fetch_sub(1). Если я не ошибся в своем изображении, head->next = node1 по-прежнему не упорядочен по head->next.load(...) в try_recv (т. е. в очереди.pop()), однако self.head.load гарантированно увидит node3 из-за синхронизации, установленной операциями RMW на cnt. Таким образом, try_recv должен перейти в ветку Inconsistent, чтобы выполнить цикл и дождаться обновления, даже если это все еще не может гарантировать появление нового значения, как указано в [atomics].order] p12.
Реализация должна сделать атомарные хранилища видимыми для атомарных загрузок, а атомарные загрузки должны наблюдать за атомарными хранилищами в течение разумного периода времени.
То есть цикл технически может быть бесконечным (хотя в реальной реализации это невозможно).
Не только в этом случае, если мы поменяем местами порядок M1 и cnt.fetch_sub(1) на картинке, в этом случае мы все равно можем перейти в противоречивое состояние? Верны ли мои рассуждения?
Подробнее здесь: https://stackoverflow.com/questions/798 ... for-this-m
