Muduo学习：shared_ptr, weak_ptr

 一. shared_ptr线程安全性：

    shared_ptr的线程安全级别和内建类型，标准库容器，std::string是一样的，需要注意的是：这是shared_ptr对象本身的线程安全级别，不是shared_ptr管理的对象的线程安全级别。
    多个线程读写同一个shared_ptr，需要加锁保护，核心想法是生成一个本地的shared_ptr对象供自己使用，而不要直接使用共享的shared_ptr，这里借用书中的代码：
     

MutexLock mutex; // No need for ReaderWriterLock
shared_ptr<Foo> globalPtr;
// 我们的任务是把globalPtr 安全地传给doit()
void doit(const shared_ptr<Foo>& pFoo);

void read()
{
    shared_ptr<Foo> localPtr;
    {
        MutexLockGuard lock(mutex);
        localPtr = globalPtr; // read globalPtr
    }
    // use localPtr since here，读写localPtr 也无须加锁
    doit(localPtr);
}

void write()
{
    shared_ptr<Foo> newPtr(new Foo); // 注意，对象的创建在临界区之外
    {
        MutexLockGuard lock(mutex);
        globalPtr = newPtr; // write to globalPtr
    }
    // use newPtr since here，读写newPtr 无须加锁
    doit(newPtr);
}

二. 利用shared_ptr管理对象池

 (内容来源于书P21以及作者的博客：http://www.cnblogs.com/Solstice/archive/2010/02/10/dtor_meets_threads.html )
 使用一个StockFactory来管理所有的股票，第一版代码如下：

class StockFactory : boost::noncopyable

{ // questionable code
 public:
  shared_ptr<Stock> get(const string& key);

 private:
  std::map<string, shared_ptr<Stock> > stocks_;
  mutable Mutex mutex_;
};

  代码的BUG：StockFactory持有Stock指针的shared_ptr，会导致Stock对象无法销毁，只要stocks_存在，shared_ptr的引用计数就不会为0。

  修改一下，将shared_ptr改为weak_ptr，来一个第二版的代码：

class StockFactory : boost::noncopyable
{
 public:
  shared_ptr<Stock> get(const string& key)
  {
    shared_ptr<Stock> pStock;
    MutexLock lock(mutex_);
    weak_ptr<Stock>& wkStock = stocks_[key];  // 如果 key 不存在，会默认构造一个
    pStock = wkStock.lock();
    if (!pStock) {
      pStock.reset(new Stock(key));
      wkStock = pStock;  // 这里更新了 stocks_[key]，注意 wkStock 是个引用
    }
    return pStock;
  }

 private:
  std::map<string, weak_ptr<Stock> > stocks_;
  mutable Mutex mutex_;
};

  这时不会阻止Stock对象销毁了，不过stocks_只增不减，不会在Stock对象销毁的时候把stocks_里面保存的weak_ptr也销毁掉。

  考虑到shared_ptr有一个参数可以定制销毁行为，我们可以在构造的时候提供一个销毁函数删除weak_ptr，再改一下，得到第三版代码：

class StockFactory : boost::noncopyable
{
 public:
  shared_ptr<Stock> get(const string& key)
  {
    shared_ptr<Stock> pStock;
    MutexLock lock(mutex_);
    weak_ptr<Stock>& wkStock = stocks_[key];  // 如果 key 不存在，会默认构造一个
    pStock = wkStock.lock();
    if (!pStock) {
      // 利用boost::bind绑定一个销毁函数，在销毁Stock对象时调用
      pStock.reset(new Stock(key), boost::bind(&StockFactory::deleteStock, this, _1));
      wkStock = pStock;  // 这里更新了 stocks_[key]，注意 wkStock 是个引用
    }
    return pStock;
  }

 private:
 void deleteStock(Stock* stock)
  {
    if (stock) {
      // 从map中删除weak_ptr
      MutexLock lock(mutex_);
      stocks_.erase(stock->key());
    }
    delete stock;
  }
  std::map<string, weak_ptr<Stock> > stocks_;
  mutable Mutex mutex_;
};

  看起来好像不错，实际上还有一个问题：pStock.reset(new Stock(key), boost::bind(&StockFactory::deleteStock, this, _1));     这里传递给bind的是一个StockFactory的原始指针，如果这个StockFactory在Stock对象销毁之前被销毁了，那再调用deleteStock就挂了。 
 很显然，我们需要在调用deleteStock函数之前知道StockFactory是否还活着，这里需要把this指针包装成shared_ptr对象传递给bind函数，使用enable_shared_from_this来完成这个任务。
 再改改代码，得到第四版：

class StockFactory : public boost::enable_shared_from_this<StockFactory>,
                     boost::noncopyable
{ /* ... */ };

shared_ptr<Stock> StockFactory::get(const string& key)
{
  // change
  // pStock.reset(new Stock(key),
  //               boost::bind(&StockFactory::deleteStock, this, _1));
  // to
  pStock.reset(new Stock(key),
               boost::bind(&StockFactory::deleteStock,
                           shared_from_this(),
                           _1));
  // ...

  这里传递给bind的是shared_ptr对象，不过还有一个问题：因为是shared_ptr，所以只要有一个Stock存在，StockFactory就不会销毁。
  
  因此，解决方式是把shared_ptr转化为weak_ptr，再改改得到第五版：

class StockFactory : public boost::enable_shared_from_this<StockFactory>,
                       boost::noncopyable
{
 public:
  shared_ptr<Stock> get(const string& key)
  {
    shared_ptr<Stock> pStock;
    MutexLock lock(mutex_);
    weak_ptr<Stock>& wkStock = stocks_[key];
    pStock = wkStock.lock();
    if (!pStock) {
      pStock.reset(new Stock(key),
                    boost::bind(&StockFactory::weakDeleteCallback,
                                 boost::weak_ptr<StockFactory>(shared_from_this()),
                                 _1));
      // 上面必须强制把 shared_from_this() 转型为 weak_ptr，才不会延长生命期
      wkStock = pStock;
    }
    return pStock;
  }

 private:
  static void weakDeleteCallback(boost::weak_ptr<StockFactory> wkFactory,
                                    Stock* stock)
  {
    shared_ptr<StockFactory> factory(wkFactory.lock());  // 尝试提升
    if (factory) {  // 如果 factory 还在，那就清理 stocks_
      factory->removeStock(stock);
    }
    delete stock;  // sorry, I lied
  }

  void removeStock(Stock* stock)
  {
    if (stock) {
      MutexLock lock(mutex_);
      stocks_.erase(stock->key());
    }
  }

 private:
  std::map<string, weak_ptr<Stock> > stocks_;
  mutable Mutex mutex_;
};

  OK，成功解决了生存周期的问题，这里还需要注意几个要点：
 1.  shared_from_this不能在构造函数里面调用，因为在构造StockFactory的时候，它还没有交给shared_ptr保管。
 2.  为了使用enable_shared_from_this，StockFactory必须在堆上创建并且保存在shared_ptr中，使用方式是这样的：

    shared_ptr<StockFactory> factory(new StockFactory);
    stock = factory->get("NYSE:IBM");

三. 使用shared_ptr实现COW

对于一个需要共享的容器，我们都能很容易的通过加锁实现线程安全，例如：

std::vector<Foo> foos;
MutexLock mutex;
void post(const Foo& f)
{
  MutexLockGuard lock(mutex);
  foos.push_back(f);
}

void traverse()
{
  MutexLockGuard lock(mutex);
  for (auto it = foos.begin(); it != foos.end(); ++it) { // 用了 0x 新写法
    it->doit();
  }
}

一切看起来没啥问题，不过万一doit函数调用了post怎么办，结果要么vector被破坏要么死锁。
怎么解决这个问题了：使用shared_ptr实现Copy On Write操作，这样就不会死锁了。
借用 代码来讲解

#include "../Mutex.h"
#include "../Thread.h"
#include <vector>
#include <boost/shared_ptr.hpp>
#include <stdio.h>
using namespace muduo;

class Foo
{
 public:
  void doit() const;
};

typedef std::vector<Foo> FooList;
typedef boost::shared_ptr<FooList> FooListPtr;
FooListPtr g_foos;
MutexLock mutex;
void post(const Foo& f)

{
  printf("post\n");
  MutexLockGuard lock(mutex);
 // 1. 检查g_foos是否只有自己在访问
  if (!g_foos.unique())
  {
 // 2. 有读线程正在读数据，根据已有的g_foos构造一个新的FooList，替换g_foos
    g_foos.reset(new FooList(*g_foos));
  }
  assert(g_foos.unique());
 // 3. 对新FooList执行修改操作
  g_foos->push_back(f);
}

void traverse()
{
 // 1. 生成一个本地的FooListPtr，g_foos引用计数加一，这样当写线程执行g_foos.reset时，
 // 旧的FooListPtr不会销毁，而是由本地的foos保存
  FooListPtr foos;
  {
    MutexLockGuard lock(mutex);
    foos = g_foos;
    assert(!g_foos.unique());
  }
  // 2. 对本地的FooListPtr进行操作，执行doit函数，即使doit去执行post也无影响
  for (std::vector<Foo>::const_iterator it = foos->begin();
      it != foos->end(); ++it)
  {
    it->doit();
  }
}


void Foo::doit() const
{
  Foo f;
  post(f);
}

int main()
{
  g_foos.reset(new FooList);
  Foo f;
  post(f);
  traverse();
}

对于上面的post函数，还有几种错误的变体，代码来自书P54

// version 1
void post(const Foo& f) {
 // 直接在原来的数据上修改，错误
    MutexLockGuard lock(mutex);
    g_foos->push(f);
}

// version 2
void post(const Foo& f) {
 // copy旧数据的时候没有加锁保护，如果其他线程正在修改g_foos对象，则获取了不完整的数据，错误
    FooListPtr newFoos(new FoosList(*g_foos));
    newFoos->push_back(f);
    MutexLockGuard lock(mutex);
    g_foos = newFoos;
}



// version 3
void post(const Foo& f) {
 // 把锁拆成两部分，当执行到 FooListPtr newFoos(new FoosList(*oldFoos)) 的时候，还是在无锁的情况copy旧数据，错误。
    FooListPtr oldFoos;
    {
        MutexLockGuard lock(mutex);
        oldFoos = g_foos;
    }
    FooListPtr newFoos(new FoosList(*oldFoos));
    newFoos->push_back(f);
    MutexLockGuard lock(mutex);
    g_foos = newFoos;
}

四. 使用shared_ptr实现对象销毁专有线程

核心想法：利用shared_ptr可以定制销毁函数的特性，在销毁函数中将对象转移到专有的线程中delete

#include <memory>
#include <thread>
#include <string>
#include <mutex>
#include <queue>
#include <condition_variable>
#include <iostream>
#include <cstdlib>
#include <cstdio>
using namespace std;
class TempObject {
    public:

        TempObject(const std::string& str)
            : b(str) {

        }

        ~TempObject(){
            std::cout << "desruct thread " << b <<" : " << std::this_thread::get_id() << std::endl;
        }

        bool isExit() {
            if (b == "quit") {
                return true;
            }
            return false;
        }
    private:
        std::string b;
};


template <typename T>
class BlockingQueue {
    public:
        BlockingQueue()
        {
        }

        ~BlockingQueue(){

        }

        void push(T* a) {
            {
                std::lock_guard<std::mutex> lg(mutexQueue_);
                queue_.push(a);
            }
            condQueue_.notify_one();
        }

        void thread_func() {
            while(true)
            {
                T* obj;
                {
                    std::unique_lock<std::mutex> ul(mutexQueue_);
                    while (queue_.empty()) {
                        condQueue_.wait(ul);
                    }
                    obj = queue_.front();
                    queue_.pop();
                }

                if (obj->isExit()) {
                    delete obj;
                    std::cout << "thread " << std::this_thread::get_id() << " is exit" << std::endl;
                    break;
                }
                else {
                    delete obj;
                }
            }
        }

        void start_thread() {
           thread_ = std::thread(&BlockingQueue::thread_func, this);
        }

        void join() {
            thread_.join();
        }

    private:
        std::mutex mutexQueue_;
        std::condition_variable condQueue_;
        std::queue<T* > queue_;
        std::thread thread_;
};



class TempObjectDeleter {
    public:
        TempObjectDeleter(BlockingQueue<TempObject> * bqp)
            : bq(bqp) {
        }

        void operator () (TempObject* parameter) {
 // 这里不执行销毁，而是将对象压入到专有线程的队列中
            std::cout << "Push thread : " << std::this_thread::get_id() << " \n";
            bq->push(parameter);
        }
    private:
        BlockingQueue<TempObject> * bq;
};

void thread_func_1(BlockingQueue<TempObject> *dQueue) {
        char buffer[10] = {0};
        for (int i = 0; i < 3; i++){
            sprintf(buffer, "b%d", i);
            std::string temp = buffer;
            // 创建时传入对象销毁函数对象
            std::shared_ptr<TempObject> a(new TempObject(temp), TempObjectDeleter(dQueue));
        }
}



int main(void) {
    BlockingQueue<TempObject> dQueue;
    dQueue.start_thread();
    std::weak_ptr<TempObject> b;
    std::thread t(thread_func_1, &dQueue);
    {
        char buffer[10] = {0};
        for (int i = 0; i < 4; i++){
            sprintf(buffer, "a%d", i);
            std::string temp = buffer;
            std::shared_ptr<TempObject> a(new TempObject(temp), TempObjectDeleter(&dQueue));
        }
    }
    t.join();
    {
        std::shared_ptr<TempObject> b2 (new TempObject("quit"), TempObjectDeleter(&dQueue));
    }
    dQueue.join();
    std::cout<< "let's exit" << std::endl;
    return 0;
}