Boost互斥抛出(奇数?)异常

我使用的是一个阻塞队列的例子，我从这个网站得到的，认为这是相当不错的。这个阻塞队列正在使用boost::互斥锁。有时会抛出一个异常:

terminate called after throwing an instance of 'boost::exception_detail::clone_impl<boost::exception_detail::error_info_injector<boost::system::system_error> >'

what():错误的文件描述符

阻塞队列代码:

#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/condition_variable.hpp>
#include <exception>
#include <list>
#include <stdio.h>
struct BlockingQueueTerminate
  : std::exception
{};
namespace tools {
  template<class T>
  class BlockingQueue
  {
  private:
    boost::mutex mtx_;
    boost::condition_variable cnd_;
    std::list<T> q_;
    unsigned blocked_;
    bool stop_;
  public:
    BlockingQueue()
      : blocked_()
      , stop_()
    {}
    ~BlockingQueue()
    {
      this->stop(true);
    }
    void stop(bool wait)
    {
      // tell threads blocked on BlockingQueue::pull() to leave
      boost::mutex::scoped_lock lock(mtx_);
      stop_ = true;
      cnd_.notify_all();
      if(wait) // wait till all threads blocked on the queue leave BlockingQueue::pull()
    while(blocked_)
      cnd_.wait(lock);
    }
    void put(T t)
    {
      boost::mutex::scoped_lock lock(mtx_);  // The exception is thrown here !
      q_.push_back(t);
      cnd_.notify_one();
    }
  T pull()
    {
      boost::mutex::scoped_lock lock(mtx_);
      ++blocked_;
      while(!stop_ && q_.empty())
    cnd_.wait(lock);
      --blocked_;
      if(stop_) {
    cnd_.notify_all(); // tell stop() this thread has left
    throw BlockingQueueTerminate();
      }
      T front = q_.front();
      q_.pop_front();
      return front;
    }
  };
}

有人能看出这里出了什么问题吗?因为我花了一整天的时间都在想这件事，却一无所获。我想我需要一个局外人来看待它。查找注释'//这里抛出异常!'，看看问题到底发生在哪里。

编辑1:

上下文:我使用这个阻塞队列是为了创建一个MySQL异步包装器。

这是我的MySQL.hh

#ifndef MYSQL_HH_
# define MYSQL_HH_
# include <boost/asio.hpp>
# include <boost/thread.hpp>
# include <boost/function.hpp>
# include <mysql++/mysql++.h>
# include <queue>
# include "async_executor.hh"
# include "BlockingQueue.hh"
class t_mysql_event {
public:
  t_mysql_event(std::string query, boost::function<void(mysqlpp::StoreQueryResult)> cb) :
    m_query(query), m_store_cb(cb), m_store_bool(true) {}
  t_mysql_event(std::string query, boost::function<void()> cb) :
    m_query(query), m_exec_cb(cb),  m_store_bool(false) {}
  bool is_store_query() {
    return m_store_bool;
  }
  std::string toString() {
    return m_query;
  }
  std::string                       m_query;
  boost::function<void(mysqlpp::StoreQueryResult)>  m_store_cb;
  boost::function<void()>               m_exec_cb;
private:
  bool                          m_store_bool;
};
namespace pools {
  class MySQL {
  public:
    ~MySQL() {}
    static MySQL* create_instance(boost::asio::io_service& io);
    static MySQL* get_instance();
    void exec(std::string query, boost::function<void()> cb);
    void store(std::string query, boost::function<void(mysqlpp::StoreQueryResult)> cb);
  private:
    MySQL(boost::asio::io_service& io) : executor(io, 100), parent_io(io), m_strand(io)
    {
      for (int i=0; i < 100; ++i) {
    boost::thread(boost::bind(&MySQL::retreive, this));
      }
    }
    void async_exec(std::string query, boost::function<void()> cb, mysqlpp::Connection& conn);
    void async_store(std::string query, boost::function<void(mysqlpp::StoreQueryResult)> cb, mysqlpp::Connection& conn);
    void retreive();
  private:
    tools::async_executor           executor;
    boost::asio::io_service&        parent_io;
    boost::asio::strand         m_strand;
    tools::BlockingQueue<t_mysql_event*>    m_events;
    std::queue<mysqlpp::Connection*>    m_stack;
  };
}
#endif //MYSQL_HH_

这是MySQL。Cc:

#include "MySQL.hh"
static pools::MySQL* _instance = 0;
namespace pools {

  MySQL* MySQL::create_instance(boost::asio::io_service& io) {
    if (!_instance)
      _instance = new MySQL(io);
    return _instance;
  }
  MySQL* MySQL::get_instance() {
    if (!_instance) {
      exit(1);
    }
    return _instance;
  }
  void MySQL::exec(std::string query, boost::function<void()> cb) {
    m_events.put(new t_mysql_event(query, cb));
  }
  void MySQL::store(std::string query, boost::function<void(mysqlpp::StoreQueryResult)> cb) {
    m_events.put(new t_mysql_event(query, cb));
  }
  void MySQL::retreive() {
    mysqlpp::Connection conn("***", "***", "***", "***");
    for(;;) {
      t_mysql_event *event = m_events.pull();
      if (event->is_store_query())
    async_store(event->m_query, event->m_store_cb, conn);
      else
    async_exec(event->m_query, event->m_exec_cb, conn);
      delete event;
    }
  }
  void MySQL::async_exec(std::string query, boost::function<void()> cb, mysqlpp::Connection& conn) {
    mysqlpp::Query db_q = conn.query(query.c_str());
    db_q.exec();
    parent_io.post(cb);
  }
  void MySQL::async_store(std::string query, boost::function<void(mysqlpp::StoreQueryResult)> cb, mysqlpp::Connection& conn) {
    mysqlpp::Query db_q = conn.query(query.c_str());
    mysqlpp::StoreQueryResult res = db_q.store();
    parent_io.post(boost::bind(cb, res));
   }
}

后记:

class MyClass {
public:
   MyClass() : _mysql(pools::MySQL::get_instance()) {}
   startQueries();
private:
   void Query1() {
      std::stringstream query("");
      query << "INSERT INTO Table1 ***";
      _mysql->exec(query.str(),
                   boost::bind(&MyClass::Query2, this, _1));
   }
   void Query2() {
      std::stringstream query("");
      query << "INSERT INTO Table2 ***";
      _mysql->exec(query.str(),
                   boost::bind(&MyClass::Query3, this, _1));
   }
   void Query3() {
      std::stringstream query("");
      query << "INSERT INTO Table3 ***";
      _mysql->exec(query.str(),
                   boost::bind(&MyClass::done, this, _1));
   }
   void done() {}
   pools::MySQL *_mysql;
};

希望能回答一些需要更多信息的请求…

有趣的事:

如果我用pool::MySQL::get_instance()代替每个_mysql，我似乎不会崩溃。