AutoResetEvent 和多个 Set

Question

我正在尝试围绕堆栈设计一种数据结构，该数据结构会阻塞，直到堆栈有可用的项目为止。我尝试使用 AutoResetEvent，但我认为我误解了同步过程的工作原理。基本上，看看下面的代码，我试图在没有可用的情况下从堆栈中弹出。

看起来 AutoResetEvent 的行为就像信号量。这是正确的吗？我可以摆脱 BlockingStack.Get() 中的 Set() 并完成它吗？或者这会导致我只使用我的堆栈项目之一的情况。

public class BlockingStack
{
    private Stack<MyType> _internalStack;
    private AutoResetEvent _blockUntilAvailable;

    public BlockingStack()
    {
        _internalStack = new Stack<MyType>(5);
        _blockUntilAvailable = new AutoResetEvent(false);

        for (int i = 0; i < 5; ++i)
        {
            var obj = new MyType();
            Add(obj);
        }
    }

    public MyType Get()
    {
        _blockUntilAvailable.WatiOne();

        lock (_internalStack)
        {
            var obj = _internalStack.Pop();
            if (_internalStack.Count > 0)
            {
                _blockUntilAvailable.Set(); // do I need to do this?
            }

            return obj;
        }
    }

    public void Add(MyType obj)
    {
        lock (_internalStack)
        {
            _internalStack.Push(obj);
            _blockUntilAvailable.Set();
        }
    }
}

我的假设是，当一个线程完成 WaitOne() 函数调用时，AutoResetEvent 会重置所有等待线程。然而，似乎有多个线程正在进入。除非我在某个地方搞乱了我的逻辑。

编辑：这是针对 Silverlight 的。

Answer 1

除非您只是想了解线程的工作原理，否则最好使用阻塞集合。这将为您提供一个由堆栈支持的阻塞集合：

ConcurrentStack<SomeType> MyStack = new ConcurrentStack<SomeType>();
BlockingCollection<SomeType> SharedStack = new BlockingCollection<SomeType>(MyStack)

然后您可以以线程安全的方式访问它，并为您正确完成所有阻塞。请参阅此处

您可以通过调用sharedStack.Take来使用sharedStack () 然后会阻塞获取，直到有东西可以从堆栈中获取。

---

编辑：
我花了一段时间（和两次尝试），但我想我已经解决了你的问题。

考虑一个空堆栈，其中有 3 个线程正在等待该事件。

调用 Add 时，堆栈有一个对象，并且允许一个线程通过该事件。

立即再次调用 Add。

第一个线程现在等待从 Add 获取锁。

Add 将第二个对象添加到堆栈中，并让另一个线程通过该事件。

现在堆栈上有两个对象，事件中有两个线程，都在等待锁。

第一个 Get 线程现在获取锁定并弹出。看到堆栈上仍然有一个对象并调用 SET。

第三个线程允许该事件发生。

第二个 Get 线程现在获取锁定并弹出。在堆栈中看不到任何内容，并且不调用 set。

但。太晚了。第三个线程已经被允许通过，因此当第二个线程放弃锁时，第三个线程尝试从空堆栈中弹出并抛出异常。

Answer 2

不，你当前的代码没有任何意义。目前，每次调用 Get 方法（.WaitOne 调用）时，您都会阻塞线程。

您可能需要类似以下内容：< /del>

public class BlockingStack<T>
{
    private Stack<T> _internalStack;
    private AutoResetEvent _blockUntilAvailable;

    public BlockingStack()
    {
        _internalStack = new Stack<T>(5);
        _blockUntilAvailable = new AutoResetEvent(false);
    }

    public T Pop()
    {
        lock (_internalStack)
        {
            if (_internalStack.Count == 0)
                _blockUntilAvailable.WaitOne();

            return _internalStack.Pop();
        }
    }

    public void Push(T obj)
    {
        lock (_internalStack)
        {
            _internalStack.Push(obj);

            if(_internalStack.Count == 0)
                _blockUntilAvailable.Set();
        }
    }
}

这个想法是，如果 _internalStack 中当前的项目数为 0，那么它应该等待来自 Push 方法的信号。一旦收到信号，它就会继续移动并从堆栈中弹出一个项目。

---

编辑：
上面的代码有两个问题：

1. 每当 Pop 阻塞 .WaitOne 时，它不会释放对
   _internalStack，因此 Push 永远无法获得锁。

2. 当在同一线程上多次调用 Pop 时，它们共享
   AutoResetEvent 的初始状态相同 - 例如。推送信号
   添加项目时的AutoResetEvent。现在当我弹出一个项目时
   第一次工作正常，因为实际上有一个项目
   堆栈。然而第二次，Stack 中没有任何值，所以
   它通过在 AutoResetEvent 上调用 .WaitOne 来等待 - 但自从
   对 Push 的调用发出了此事件的信号，它只会返回 true，并且
   未按预期等待。

一个（有效的）替代方案：

public class BlockingStack<T>
{
    private Stack<T> _internalStack;

    public BlockingStack()
    {
        _internalStack = new Stack<T>(5);
    }

    public T Pop()
    {
        lock (_internalStack)
        {
            if (_internalStack.Count == 0)
                Monitor.Wait(_internalStack);

            return _internalStack.Pop();
        }
    }

    public void Push(T obj)
    {
        lock (_internalStack)
        {
            _internalStack.Push(obj);
            Monitor.Pulse(_internalStack);
        }
    }
}

Answer 3

我没有验证基于 Monitor 的解决方案，但我确实编写了一个基于信号量的解决方案，该解决方案似乎有效：

public class Semaphore
{
    private int _count;
    private int _maximum;
    private object _countGuard;

    public Semaphore(int maximum)
    {
        _count = 0;
        _maximum = maximum;
        _countGuard = new object();
    }

    public void WaitOne()
    {
        while (true)
        {
            lock (_countGuard)
            {
                if (_count < _maximum)
                {
                    _count++;
                    return;
                }
            }
            Thread.Sleep(50);
        }
    }

    public void ReleaseOne()
    {
        lock (_countGuard)
        {
            if (_count > 0)
            {
                _count--;
            }
        }
    }
}

public class BlockingStack
{
    private Stack<MyType> _internalStack;
    private Semaphore _blockUntilAvailable;

    public BlockingStack()
    {
        _internalStack = new Stack<MyType>(5);
        _blockUntilAvailable = new Semaphore(5);

        for (int i = 0; i < 5; ++i)
        {
            var obj = new MyType();
            Add(obj);
        }
    }

    public MyType Get()
    {
        _blockUntilAvailable.WaitOne();

        lock (_internalStack)
        {
            var obj = _internalStack.Pop();
            return obj;
        }
    }

    public void Add(MyType obj)
    {
        lock (_internalStack)
        {
            _internalStack.Push(obj);
            _blockUntilAvailable.ReleaseOne();
        }
    }
}