0

我有一个 VS2015 解决方案，由非托管 c++ 代码（用于执行一些 CPU 密集型模拟计算）、围绕此代码的 c++/cli 包装器和通过 DLL 调用 c++/cli 包装器的 ac# 项目组成。下面的例子是完整代码的简化版，抱歉提前写了这么多代码，但为了完整的了解情况，需要它。

非托管 C++ 代码

class diffusion_limited_aggregate {
public:
    diffusion_limited_aggregate() 
        : aggregate_map(), attractor_set(), batch_queue() {}
    std::size_t size() const noexcept { return aggregate_map.size(); }
    std::queue<std::pair<int,int>>& batch_queue_handle() noexcept { return batch_queue; }
    void generate(std::size_t n) {
        initialise_attractor_structure(); // set up initial attractor seed points
        std::size_t count = 0U;
        std::pair<int,int> current = std::make_pair(0,0);
        std::pair<int,int> prev = current;
        bool has_next_spawned = false;
        while (size() < n) {
            if (!has_next_spawned) {
                // => call function to spawn particle setting current 
                has_next_spawned = true;
            }
            prev = current;
            // => call function to update random walking particle position
            // => call function to check for lattice boundary collision
            if (aggregate_collision(current, prev, count)) has_next_spawned = false;
        }
    }
    void initialise_attractor_structure() {
        attractor_set.clear();
        attractor_set.insert(std::make_pair(0,0));
    }
    void push_particle(const std::pair<int,int>& p, std::size_t count) {
        aggregate_map.insert(std::make_pair(p, count));
        batch_queue.push(p);
    }
    bool aggregate_collision(const std::pair<int,int>& current,
        const std::pair<int,int>& prev, std::size_t& count) {
        if (aggregate_map.find(current) != aggregate_map.end() 
            || attractor_set.find(current) != attractor_set.end()) {
            push_particle(previous, ++count);
            return true;
        }
        return false;
    }
private:
    std::unordered_map<std::pair<int,int>, 
        std::size_t,
        utl::tuple_hash> aggregate_map;
    std::unordered_set<std::pair<int,int>, utl::tuple_hash> attractor_set;
    std::queue<std::pair<int,int>> batch_queue; // holds buffer of aggregate points
};

哪里utl::tuple_hash是一个散列函数对象std::pair，更一般地，std::tuple实例，定义为：

namespace utl {
    template<class Tuple, std::size_t N>
    struct tuple_hash_t {
        static std::size_t tuple_hash_compute(const Tuple& t) {
            using type = typename std::tuple_element<N-1, Tuple>::type;
            return tuple_hash_t<Tuple,N-1>::tuple_hash_compute(t)
                + std::hash<type>()(std::get<N-1>(t));
        }
    };
    // base
    template<class Tuple>
    struct tuple_hash_t<Tuple, 1> {
        static std::size_t tuple_hash_compute(const Tuple& t) {
            using type = typename std::tuple_element<0,Tuple>::type;
            return 51U + std::hash<type>()(std::get<0>(t))*51U;
        }
    };
    struct tuple_hash {
        template<class... Args>
        std::size_t operator()(const std::tuple<Args...>& t) const {
            return tuple_hash_t<std::tuple<Args...>,sizeof...(Args)>::tuple_hash_compute(t);
        }
        template<class Ty1, class Ty2>
        std::size_t operator()(const std::pair<Ty1, Ty2>& p) const {
            return tuple_hash_t<std::pair<Ty1,Ty2>,2>::tuple_hash_compute(p);
        }
    };
}

托管 C++/CLI 包装器

以下是 c++/cli 中围绕类的包装器，diffusion_limited_aggregate在这种情况下，重要的方法是ProcessBatchQueue. 此方法是std::deque iterator not dereferencable error必须发生的地方，因为它是唯一batch_queue可以访问和弹出内容的地方。

public ref class ManagedDLA2DContainer {
private:
    diffusion_limited_aggregate* native_dla_2d_ptr;
    System::Object^ lock_obj = gcnew System::Object();
public:
    ManagedDLA2DContainer() : native_dla_2d_ptr(new diffusion_limited_aggregate()) {}
    ~ManagedDLA2DContainer() { delete native_dla_2d_ptr; }
    std::size_t Size() { return native_dla_2d_ptr->size(); }
    void Generate(std::size_t n) { native_dla_2d_ptr->generate(n); }
    System::Collections::Concurrent::BlockingCollection<
        System::Collections::Generic::KeyValuePair<int,int>
    >^ ProcessBatchQueue() {
        // store particles in blocking queue configuration
        System::Collections::Concurrent::BlockingCollection<
            System::Collections::Generic::KeyValuePair<int,int>>^ blocking_queue =
            gcnew System::Collections::Concurrent::BlockingCollection<
                System::Collections::Generic::KeyValuePair<int,int>
            >();
        System::Threading::Monitor::Enter(lock_obj); // define critical section start
        try {
            // get ref to batch_queue
            std::queue<std::pair<int,int>>& bq_ref = native_dla_2d_ptr->batch_queue_handle();
            // loop over bq transferring particles to blocking_queue
            while (!bq_ref.empty()) {
                auto front = std::move(bq_ref.front());
                blocking_queue->Add(System::Collections::Generic::KeyValuePair<int,int>(front.first,front.second));
                bq_ref.pop();
            }
        }
        finally { System::Threading::Monitor::Exit(lock_obj); }
        return blocking_queue;
    }
}

C# 代码

最后，我有以下 c# 代码，用于ManagedDLA2DContainer生成聚合并将它们显示在界面上。

public partial class MainWindow : Window {
    private static readonly System.object locker = new object();
    private readonly ManagedDLA2DContainer dla_2d;
    public MainWindow() {
        InitializeComponent();
        dla_2d = new ManagedDLA2DContainer();
    }
    private void GenerateAggregate(uint n) {
        // start asynchronous task to perform aggregate simulation computations
        Task.Run(() => CallNativeCppAggregateGenerators(n));
        System.Threading.Thread.Sleep(5);
        // start asynchronous task to perform rendering
        Task.Run(() => AggregateUpdateListener(n));
    }
    private void CallNativeCppAggregateGenerators(uint n) {
        dla_2d.Generate(n);
    }
    private void AggregateUpdateListener(uint n) {
        const double interval = 10.0;
        Timer timer = new Timer(interval);
        timer.Elapsed += Update2DAggregateOnTimedEvent;
        timer.AutoReset = true;
        timer.Enabled = true;
    }
    private void Update2DAggregateOnTimedEvent(object source, ElapsedEventArgs e) {
        lock(locker) {
            BlockingCollection<KeyValuePair<int,int>> bq = dla_2d.ProcessBatchQueue();
            while(bq.Count != 0) {
                KeyValuePair<int,int> p = bq.Take();
                Point3D pos = new Point3D(p.Key, p.Value, 0.0);
                // => do stuff with pos, sending to another class method for rendering
                // using Dispatcher.Invoke(() => { ... }); to render in GUI
            }
        }
    }
}

该方法GenerateAggregate仅在每次聚合执行时调用一次，它通过按钮处理程序方法调用，因为我Generate在接口上有一个带有OnGenerateButtonClicked事件处理程序函数的方法，该方法调用GenerateAggreate. 两者CallNativeCppAggregateGenerators和AggregateUpdateListener也不会在代码中的其他任何地方调用。

问题

如托管包装器部分所述，在执行此代码时，我偶尔会遇到运行时断言错误，

std::deque迭代器不可取消引用。

这往往在第一次执行时发生，但它也确实发生在正在进行的聚合生成过程的中间，因此用于生成聚合的启动代码在这里可能不是罪魁祸首。

我该如何解决这个问题？希望这是我的关键部分代码或类似代码中出现一些逻辑错误的简单案例，但我还无法确定确切的问题。

正如评论中所指出的，问题可能是元素不断被添加batch_queue，而 C# 线程调用ProcessBatchQueue正在消耗队列元素，从而可能使batch_queue' 的迭代器无效。是否有可以应用于此用例的典型生产者-消费者设计模式？

编辑：如果反对者可以给出他们的理由，这样我就可以改进这个问题，那就太好了。

4

1 回答 1

1

我找到了解决这个问题的方法，下面将详细介绍。正如问题中所建议的那样，问题是在处理batch_queue它的迭代器时，由于在聚合生成过程中不断地将元素推送到队列中，它偶尔会失效。

此解决方案使用的内存比以前的基于实现稍多batch_queue，但就迭代器有效性而言，它是安全的。我用原生 C++ 代码中的聚合粒子缓冲区替换batch_queue了：std::vector<std::pair<int,int>>

class diffusion_limited_aggregate {
public:
//...
    const std::vector<std::pair<int,int>>& aggregate_buffer() const noexcept { return buffer; }
private:
//...
    std::vector<std::pair<int,int>> buffer;
};

然后ManagedDLA2DContainer::ProcessBatchQueue被替换为ManagedDLA2DContainer::ConsumeBuffer读取到标记的索引并将最近一批聚合粒子推送到 ac# List<KeyValuePair<int,int>>：

System::Collections::Generic::List<System::Collections::Generic::KeyValuePair<int, int>>^ ConsumeBuffer(std::size_t marked_index) {
        System::Collections::Generic::List<System::Collections::Generic::KeyValuePair<int, int>>^ buffer =
            gcnew System::Collections::Generic::List<System::Collections::Generic::KeyValuePair<int, int>>();
        if (native_dla_2d_ptr->aggregate_buffer().empty()) return buffer;
        System::Threading::Monitor::Enter(lock_obj);    // define critical section start
        try {   // execute critical section
            // read from last marked buffer index up to size of buffer and write these data to batch list
            for (int i = marked_index; i < native_dla_2d_ptr->aggregate_buffer().size(); ++i) {
                buffer->Add(System::Collections::Generic::KeyValuePair<int, int>(
                    native_dla_2d_ptr->aggregate_buffer()[i].first,
                    native_dla_2d_ptr->aggregate_buffer()[i].second
                    )
                );
            }
        }
        finally { System::Threading::Monitor::Exit(lock_obj); } // exit critical section by releasing exclusive lock
        return buffer;
}

最后，修改了 c#MainWindow::Update2DAggregateOnTimedEvent方法中的代码以反映 c++/cli 代码中的这些更改：

private void Update2DAggregateOnTimedEvent(object source, ElapsedEventArgs e, uint n) {
    lock (locker) {
        List<KeyValuePair<int,int>> buffer = dla_2d.ConsumeBuffer(
            (current_particles == 0) ? 0 : current_particles-1); // fetch batch list
        foreach (var p in buffer) {
            // => add p co-ords to GUI manager...
            ++current_particles;
            // => render aggregate...
        }
    }
}
于 2016-09-22T19:23:41.293 回答