在 Linux 上用 C++ 实现线程安全的通用堆栈

Question

在最近的一次采访中，我被要求在 Linux 机器上用 C++ 实现一个线程安全的通用（即基于模板）堆栈。
我很快想出了以下（可能有编译错误）。
我通过了。面试官可能喜欢这个实现中的一些东西。也许是设计部分:)
以下是此实现可能存在的一些问题：-
1. 指示溢出/下溢的错误实现。由于我使用 STL 向量作为底层数据结构，因此没有溢出处理。应该有这样的处理吗？此外，下溢（在 Pop() 中）产生 false 作为返回值。是否应该通过抛出异常来完成？
2. PopElem 例程的实现。下面的实现是否正确？
3.没有真正使用顶部元素。
4. 编写器和读取器线程之间的更好时机。

请提出任何建议/建议/改进。
谢谢。

//实现一个线程安全的泛型栈。

#include<pthread.h>
#include<iostream>
#include<vector>

using namespace std;

template<typename T>
class MyStack
{
public:
//interface
bool Push(T elem);
bool Pop(T& elem);
bool IsEmpty();

//constructor
MyStack() {
pthread_mutex_init(&lock);
top = 0;
}

//destructor
~MyStack() {
pthread_mutex_destroy(&lock);
}

private:
pthread_mutex_t lock;
int top;
vector<T> stack;

bool MyStack::Push(T elem);
bool MyStack::PopElem(T& elem);
}; //end of MyStack

template<typename T>
bool MyStack<T>::Push(T elem)
{
    pthread_mutex_lock(&lock);
    PushElem(elem);
    pthread_mutex_unlock(&lock);
}

template<typename T>
bool MyStack<T>::Pop(T& elem)
{
    pthread_mutex_lock(&lock);
    PopElem(elem);
    pthread_mutex_unlock(&lock);
}

template<typename T>
bool MyStack<T>::PushElem(T elem)
{
    stack.push_back(elem);
     top = stack.size();
}

template<typename T>
bool MyStack<T>::PopElem(T& elem)
{
   if(this.IsEmpty())
   {
        return false;
   }

   elem = stack.back(); //tricky, returns a reference to the last element
   stack.pop_back(); // is elem valid after this ??
   top = stack.size();
   return true;
}      


template<typename T>
bool MyStack<T>::IsEmpty()
{
    return stack.empty();
}


class MyStackTest
{
public:
  void Initialize() {
  pthread_init(&readerT);
  pthread_init(&writerT);
  }

  void Run() {
 pthread_create(writerT,0,writer,0); 
 pthread_create(readerT,0,reader,0);
 pthread_join(&writerT);
 pthread_join(&readerT);
}

private:
pthread_t readerT;
pthread_t writerT;
MyStack<int> stack;

void reader(void);
void writer(void);
};

void MyStackTest::writer() {
  for(int i=0;i<20;i++) {
      stack.Push(i);
      cout<<"\n\t Pushed element: "<<i;
   } //end for
}

void MyStackTest::reader() {
   int elem;
   while(stack.Pop(elem))
   {
     cout<<"\n\t Popped: "<<elem;
   }
}

int main()
{
    MyStackTest Test;

    Test.Run();
}

Answer 1

一些问题：

• 我将实现一个 Locker 类来使用 RAII 声明和释放互斥锁
• 我会使用 std::stack
• 我会让 std::stack 的用户使用 Locker 来实现锁定策略 - 拥有一个锁定自身的堆栈是不好的设计，因为堆栈不知道如何使用它

Answer 2

尼尔，Onebyone：
尝试使用 RAII 进行互斥锁。有cmets吗？

template<typename T> 
class MyStack
{
public:
//interface
bool Push(T elem);
bool Pop(T& elem);
bool IsEmpty();

//constructor
MyStack() {
//top = 0;
}

//destructor
~MyStack() {

}

private:
    class Locker {          //RAII
    public:
        Locker() {
            pthread_mutex_init(&lock);
        }
        ~Locker() {
            pthread_mutex_destroy(&lock);
        }
        void Lock() {
            pthread_mutex_lock(&lock);
        }
        void UnLock() {
            pthread_mutex_unlock(&lock);
        }
    private:
        pthread_mutex_t lock;
    };
Locker MyLock;
//int top;
stack<T> mystack;

bool MyStack::Push(T elem);
bool MyStack::PushElem(T elem);
bool MyStack::Pop(T& elem);
bool MyStack::PopElem(T& elem);
}; //end of MyStack

template<typename T>
bool MyStack<T>::Push(T elem)
{
    MyLock.Lock();
    PushElem(elem);
    MyLock.UnLock();
}

template<typename T>
bool MyStack<T>::Pop(T& elem)
{
    MyLock.Lock();
    PopElem(elem);
    MyLock.UnLock();
}

Answer 3

我会添加一个条件变量，以便“poppers”可以等待而不会消耗 CPU 时间。

Answer 4

// 棘手，返回对最后一个元素的引用

赋值复制了从向量中弹出之前的最后一个元素，所以这很好。

正如您所说，“顶部”毫无意义。您可以随时获取矢量的大小。

您应该只在持有锁的情况下调用 stack.empty()，因为不能保证它会进行原子访问。如果在另一个线程正在更新堆栈的过程中调用它，您可能会得到不一致的答案。所以你的公共 IsEmpty 函数应该使用互斥体，这意味着你不想从其他地方自己调用它。

但无论如何，IsEmpty 在并行代码中并不是很有用。仅仅因为当你调用它时它是错误的，并不意味着当你 Pop 之后它仍然是错误的一行。所以要么你应该从公共接口中去掉它，要么你应该公开锁，以便用户可以编写他们自己的原子操作。在这种情况下，除了调试模式下的断言之外，我根本没有任何下溢检查。但是，我从不相信在没有阅读文档或测试他们的代码的情况下就进入发布模式的人溺爱。

[编辑：如何将 RAII 用于锁

当人们说要使用 RAII 来锁定时，他们不仅仅意味着要确保互斥体被销毁。他们的意思是使用它来确保互斥锁已解锁。关键是，如果您的代码如下所示：

lock();
doSomething();
unlock();

并且 doSomething() 抛出异常，那么您将无法解锁互斥锁。哎哟。

所以，这是一个示例类，以及用法：

class LockSession;
class Lock {
    friend class LockSession;
    public:
    Lock()        { pthread_mutex_init(&lock); }
    ~Lock()       { pthread_mutex_destroy(&lock); }

    private:
    void lock()   { pthread_mutex_lock(&lock); }
    void unlock() { pthread_mutex_unlock(&lock); }

    private:
    Lock(const Lock &);
    const Lock &operator=(const Lock &);

    private:
    pthread_mutex_t lock;
};

class LockSession {
    LockSession(Lock &l): lock(l) { lock.lock(); }
    ~LockSession()                { lock.unlock(); }
    private:
    LockSession(const LockSession &);
    LockSession &operator=(const LockSession &);

    private:
    Lock &lock;
};

然后，您的代码将在某个地方有一个与您要保护的数据相关联的锁，并将使用它，如下所示：

void doSomethingWithLock() {
    LockSession session(lock);
    doSomething();
}

或

void doSeveralThings() {
    int result = bigSlowComputation();  // no lock
    {
        LockSession s(lock);
        result = doSomething(result); // lock is held
    }
    doSomethingElse(result);     // no lock
}

现在doSomething() 抛出异常或正常返回都没有关系（好吧，在第二个示例中，doSomethingElse 不会发生异常，但我假设这是不需要的在错误情况下完成）。无论哪种方式，session 都会被销毁，并且其析构函数会释放互斥锁。特别是，在堆栈上“推送”之类的操作会分配内存，因此可能会抛出异常，因此您需要处理它。

RAII 代表资源获取即初始化。在 doSomethingWithLock() 的情况下，您要获取的资源是您要持有锁。所以你编写了一个类，它允许你通过初始化一个对象（LockSession）来做到这一点。当对象被销毁时，锁被放弃。因此，您对待“锁定/解锁互斥锁”的方式与对待“初始化/取消初始化互斥锁”的方式完全相同，并且您以同样的方式保护自己免受资源泄漏。

一个有点烦人的事实是这段代码完全被破坏和错误，你必须确保不要意外地这样做，即使它看起来像正确的代码一样粗心：

void doSomethingWithLock() {
    LockSession(lock);
    doSomething();
}

这里第一行创建了一个临时对象并立即销毁它，再次释放锁。 doSomething() 不会在持有锁的情况下被调用。

Boost 有一个类模板 scoped_lock，它执行 LockSession 的功能等等。]

Answer 5

一开始我会扔掉上衣。当你不需要它时，它只是浪费！

小而美

另外，如果您想优化对向量的访问：重复处理管理信息（此处为：stacklength）总是容易出错。更好的希望是，这个向量非常快（大多数时候是 STL），所以 empty() 也是。

Answer 6

这不是惯用的 C++，可能没有任何优势，但只是为了新颖，您是否考虑过实现不可变堆栈？这样一来，它就会自动成为线程安全的。

Eric Lippert 完成了C# implementation。诚然，C++ 代码会涉及更多。

Answer 7

您没有解决的一件事是线程取消问题。当在 stl 容器上的操作期间取消线程时，stl 的行为很糟糕。对向量进行操作时需要禁用取消。我发现了这个困难的方法。当您遇到死锁并且线程都在模板化的 stl 代码中并且您正试图准确调试发生的事情时，这并不好玩。使用 pthread_setcancelstate 改变线程的取消状态。