AskOverflow.Dev

AskOverflow.Dev Logo AskOverflow.Dev Logo

AskOverflow.Dev Navigation

  • Início
  • system&network
  • Ubuntu
  • Unix
  • DBA
  • Computer
  • Coding
  • LangChain

Mobile menu

Close
  • Início
  • system&network
    • Recentes
    • Highest score
    • tags
  • Ubuntu
    • Recentes
    • Highest score
    • tags
  • Unix
    • Recentes
    • tags
  • DBA
    • Recentes
    • tags
  • Computer
    • Recentes
    • tags
  • Coding
    • Recentes
    • tags
Início / coding / Perguntas / 78950148
Accepted
ayaan098
ayaan098
Asked: 2024-09-05 02:45:19 +0800 CST2024-09-05 02:45:19 +0800 CST 2024-09-05 02:45:19 +0800 CST

Dados não ordenados corretamente na fila spsc sem bloqueio

  • 772

Estou tentando escrever uma fila limitada de produtor único e consumidor único sem bloqueio, no entanto, a ordem dos elementos inseridos na fila não é a mesma que a ordem dos elementos removidos da fila. O código de teste em main.cpp adiciona 100.000.000 de inteiros sequenciais à fila em um thread e exibe os valores em outro, verificando se os valores são sequenciais.

Atualmente, ao executar o programa, ele imprime "Oh não! o último valor foi: 219718, mas este valor é 220231" (os valores numéricos diferem entre as execuções do programa). O intervalo dos números parece depender de yield_frequency, writer_threadvalores menores de yield_frequencyfazem com que o teste (frequentemente, mas nem sempre) falhe em um número na casa das centenas.

(O código é um wip, então provavelmente há algum código que não corresponde às melhores práticas do C++)

spsc_queue.hpp

#pragma once

#include <memory>
#include <utility>
#include <bit>
#include <atomic>
#include <optional>

template <typename T> class ChannelReader;
template <typename T> class ChannelWriter;

// instances of this class cannot be created directly, make_queue returns a pair of ChannelReader and ChannelWriter instances that wrap the queue to prevent reads (or writes) from multiple threads
template <typename T>
class SpscQueue {
    friend ChannelReader<T>;
    friend ChannelWriter<T>;
    public:
    static std::pair<ChannelReader<T>, ChannelWriter<T>> make_queue(size_t size) {
        if (!std::__has_single_bit(size)) throw "";
        std::shared_ptr<SpscQueue<T>> queue_ptr(new SpscQueue<T>(size));
        return std::make_pair(ChannelReader<T>(queue_ptr), ChannelWriter<T>(queue_ptr));
    }
    ~SpscQueue() {
        delete[] data;
    }
    private:
    SpscQueue (size_t _size): data(new T[_size]()), size(_size) {}
    alignas(64) T* data;
    const size_t size;
    alignas(64) std::atomic<size_t> read_idx{0};
    alignas(64) std::atomic<size_t> write_idx{0};
};

// wrapper class that allows ONE thread to read from a SPSC queue
template <typename T>
class ChannelReader {
    friend SpscQueue<T>;
    public:
    // attempt to read from the queue, returning the value if successful, and std::nullopt otherwise
    std::optional<T> try_get_next() {
        size_t read_idx = queue->read_idx;
        size_t write_idx = cached_write_idx;
        if (write_idx <= read_idx) {
            write_idx = queue->write_idx;
            cached_write_idx = write_idx;
        }
        if (write_idx <= read_idx) {
            return std::nullopt;
        } else {
            queue->read_idx++;
            return std::move(queue->data[read_idx % queue->size]);
        }
    }
    ChannelReader(const ChannelReader&) = delete;
    ChannelReader& operator=(const ChannelReader&) = delete;
    ChannelReader(ChannelReader&&) = default;
    ChannelReader& operator=(ChannelReader&&) = default;
    private:
    ChannelReader(std::shared_ptr<SpscQueue<T>> ptr): queue(ptr) {}
    std::shared_ptr<SpscQueue<T>> queue;
    alignas(64) size_t cached_write_idx{0};
};

// wrapper class that allows ONE thread to write to a SPSC queue
template <typename T>
class ChannelWriter {
    friend SpscQueue<T>;
    public:
    // attempt to write to the queue, returns true if the write was successful, and false otherwise
    bool try_write_next(const T& obj) {
        size_t read_idx = cached_read_idx;
        size_t write_idx = queue->write_idx;
        if (write_idx >= read_idx + queue->size) {
            read_idx = queue->read_idx;
            cached_read_idx = read_idx;          
        }
        if (write_idx >= read_idx + queue->size) {
            return false;
        } else {
            queue->data[write_idx % queue->size] = obj;
            ++queue->write_idx;
            return true;
        }
    }
    ChannelWriter(const ChannelWriter&) = delete;
    ChannelWriter& operator=(const ChannelWriter&) = delete;
    ChannelWriter(ChannelWriter&&) = default;
    ChannelWriter& operator=(ChannelWriter&&) = default;
    private:
    ChannelWriter(std::shared_ptr<SpscQueue<T>> ptr): queue(ptr) {}
    std::shared_ptr<SpscQueue<T>> queue;
    alignas(64) size_t cached_read_idx{0};
};

principal.cpp

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

#include "../include/spsc_queue.hpp"

namespace chrono = std::chrono;

int main() {
  std::atomic<bool> latch{false};
  auto [reader, writer] = SpscQueue<int>::make_queue(512);
  std::thread reader_thread([_reader = std::move(reader), &latch]() mutable {
    latch.wait(false);
    std::cerr << "Starting reader...\n";
    int last_val = -1;
    while (last_val != 100'000'000 - 1) {
      if (auto data = _reader.try_get_next()) {
        if (*data != last_val + 1) {
          std::cerr << "Oh no! last value was: " << last_val
                    << " but this value is " << *data << '\n';
          std::exit(1);
        }
        last_val = *data;
        // std::cerr << last_val << " Read\n";
      }
    }
  });
  std::thread writer_thread([_writer = std::move(writer), &latch]() mutable {
    latch.wait(false);
    std::cerr << "Starting writer...\n";
    for (int i = 0; i < 100'000'000; ++i) {
      for (int j = 0; !_writer.try_write_next(i); ++j) {
        constexpr int yield_frequency = 1 << 0;
        if (j % yield_frequency)
          std::this_thread::yield();
      }
      // std::cerr << "writer wrote value" << i << '\n';
      // if (i == 10'000'000) std::exit(1);
    }
  });
  std::cout << "Start" << std::endl;
  {
    auto start = chrono::steady_clock::now();
    latch = true;
    latch.notify_all();
    reader_thread.join();
    writer_thread.join();
    auto finish = chrono::steady_clock::now();
    auto elapsed_seconds =
        chrono::duration_cast<chrono::duration<double>>(finish - start).count();
    std::cout << elapsed_seconds << std::endl;
  }
  std::cout << "End" << std::endl;
}
c++
  • 1 1 respostas
  • 53 Views

1 respostas

  • Voted
  1. Best Answer
    Ahmed AEK
    2024-09-05T05:10:14+08:002024-09-05T05:10:14+08:00

    220231 - 219718 = 513e o tamanho do buffer circular é 512, então o escritor e o leitor estão correndo sobre esse último valor... você está incrementando o índice de leitura antes de ler o valor, o escritor pode sobrescrever o valor que o leitor vai ler.

    queue->read_idx++; // tell writer data is already read
    // writer modifies data here
    return std::move(queue->data[read_idx % queue->size]); // read new data
    

    a solução é ler o valor antes de fazer o incremento, da mesma forma que você faz ao escrever, primeiro escreva o valor e depois incremente o índice de gravação.

    auto result = std::move(queue->data[read_idx % queue->size]);
    ++queue->read_idx;
    return result;
    
    • 2

relate perguntas

  • Por que os compiladores perdem a vetorização aqui?

  • Erro de compilação usando CMake com biblioteca [fechada]

  • Erro lançado toda vez que tento executar o premake

  • Como criar um tipo de octeto semelhante a std::byte em C++?

  • Somente operações bit a bit para std::byte em C++ 17?

Sidebar

Stats

  • Perguntas 205573
  • respostas 270741
  • best respostas 135370
  • utilizador 68524
  • Highest score
  • respostas
  • Marko Smith

    Vue 3: Erro na criação "Identificador esperado, mas encontrado 'import'" [duplicado]

    • 1 respostas
  • Marko Smith

    Por que esse código Java simples e pequeno roda 30x mais rápido em todas as JVMs Graal, mas não em nenhuma JVM Oracle?

    • 1 respostas
  • Marko Smith

    Qual é o propósito de `enum class` com um tipo subjacente especificado, mas sem enumeradores?

    • 1 respostas
  • Marko Smith

    Como faço para corrigir um erro MODULE_NOT_FOUND para um módulo que não importei manualmente?

    • 6 respostas
  • Marko Smith

    `(expression, lvalue) = rvalue` é uma atribuição válida em C ou C++? Por que alguns compiladores aceitam/rejeitam isso?

    • 3 respostas
  • Marko Smith

    Quando devo usar um std::inplace_vector em vez de um std::vector?

    • 3 respostas
  • Marko Smith

    Um programa vazio que não faz nada em C++ precisa de um heap de 204 KB, mas não em C

    • 1 respostas
  • Marko Smith

    PowerBI atualmente quebrado com BigQuery: problema de driver Simba com atualização do Windows

    • 2 respostas
  • Marko Smith

    AdMob: MobileAds.initialize() - "java.lang.Integer não pode ser convertido em java.lang.String" para alguns dispositivos

    • 1 respostas
  • Marko Smith

    Estou tentando fazer o jogo pacman usando apenas o módulo Turtle Random e Math

    • 1 respostas
  • Martin Hope
    Aleksandr Dubinsky Por que a correspondência de padrões com o switch no InetAddress falha com 'não cobre todos os valores de entrada possíveis'? 2024-12-23 06:56:21 +0800 CST
  • Martin Hope
    Phillip Borge Por que esse código Java simples e pequeno roda 30x mais rápido em todas as JVMs Graal, mas não em nenhuma JVM Oracle? 2024-12-12 20:46:46 +0800 CST
  • Martin Hope
    Oodini Qual é o propósito de `enum class` com um tipo subjacente especificado, mas sem enumeradores? 2024-12-12 06:27:11 +0800 CST
  • Martin Hope
    sleeptightAnsiC `(expression, lvalue) = rvalue` é uma atribuição válida em C ou C++? Por que alguns compiladores aceitam/rejeitam isso? 2024-11-09 07:18:53 +0800 CST
  • Martin Hope
    The Mad Gamer Quando devo usar um std::inplace_vector em vez de um std::vector? 2024-10-29 23:01:00 +0800 CST
  • Martin Hope
    Chad Feller O ponto e vírgula agora é opcional em condicionais bash com [[ .. ]] na versão 5.2? 2024-10-21 05:50:33 +0800 CST
  • Martin Hope
    Wrench Por que um traço duplo (--) faz com que esta cláusula MariaDB seja avaliada como verdadeira? 2024-05-05 13:37:20 +0800 CST
  • Martin Hope
    Waket Zheng Por que `dict(id=1, **{'id': 2})` às vezes gera `KeyError: 'id'` em vez de um TypeError? 2024-05-04 14:19:19 +0800 CST
  • Martin Hope
    user924 AdMob: MobileAds.initialize() - "java.lang.Integer não pode ser convertido em java.lang.String" para alguns dispositivos 2024-03-20 03:12:31 +0800 CST
  • Martin Hope
    MarkB Por que o GCC gera código que executa condicionalmente uma implementação SIMD? 2024-02-17 06:17:14 +0800 CST

Hot tag

python javascript c++ c# java typescript sql reactjs html

Explore

  • Início
  • Perguntas
    • Recentes
    • Highest score
  • tag
  • help

Footer

AskOverflow.Dev

About Us

  • About Us
  • Contact Us

Legal Stuff

  • Privacy Policy

Language

  • Pt
  • Server
  • Unix

© 2023 AskOverflow.DEV All Rights Reserve