events.action: opened
events.author: ygm521
events.masked_author: username_0
events.text: 陈哥，请假。我使用的是main ractor+many sub ractor+ thread pool模式处理的，**且结合了bucket剔除空闲连接和限制连接数目**。目的，测试连接数限制是否可用，但是结果挂了，如下：
  exception caught in Thread subEventThPool1
  reason: boost::bad_any_cast: failed conversion using boost::any_cast
  已放弃(吐核)
  
  调试：
  Program terminated with signal 6, Aborted.
  #0 0x00007fcabb03b5f7 in __GI_raise (sig=sig@entry=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:56
  56 return INLINE_SYSCALL (tgkill, 3, pid, selftid, sig);
  (gdb) bt
  #0 0x00007fcabb03b5f7 in __GI_raise (sig=sig@entry=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:56
  #1 0x00007fcabb03cce8 in __GI_abort () at abort.c:90
  #2 0x0000000000447626 in runInThread (this=0x1d83ff0) at /home/wbq/muduo-master/muduo/base/Thread.cpp:103
  #3 muduo::detail::startThread (obj=0x1d83ff0) at /home/wbq/muduo-master/muduo/base/Thread.cpp:126
  #4 0x00007fcabc161dc5 in start_thread (arg=0x7fcab5f91700) at pthread_create.c:308
  #5 0x00007fcabb0fcced in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:113
  **死到onStringMessage函数了， solve中没办法使用非静态成员connectionBuckets_**。
  void SudokuServer::onStringMessage(const TcpConnectionPtr& conn, const string& message, Timestamp)
  {
  processRequest(conn, message);
  assert(!conn->getContext().empty());
  WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));
  EntryPtr entry(weakEntry.lock());
  if (entry) {
  connectionBuckets_.back().insert(entry);
  }
  }
  void SudokuServer::solve(const TcpConnectionPtr& conn,
  const string& request)
  {
  muduo::net::Buffer buf;
  buf.append("hello");
  int32_t len = static_cast<int32_t>(strlen("hello"));
  int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
  buf.prepend(&be32, sizeof be32);
  conn->send(&buf);
  }
  
  void SudokuServer::processRequest(const TcpConnectionPtr& conn, const string& request)
  {
  threadPool_.run(boost::bind(&solve, conn, request));
  }
  
  **加锁也不好使。**
  if (mutex_){
  MutexLockGuard lock(*mutex_);
  connectionBuckets_.back().insert(entry);
  }
events.title: main loop+many sub loop+ threadpoll 限制连接数示例结合踢掉空闲连接
events.type: issue
events.action: created
events.author: chenshuo
events.masked_author: username_1
events.text: this is not an issue of muduo.
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: 如何解决呢？ boost1.60和53均不好使。 你可以试试将踢掉空闲连接和限制连接数目实例总合下测测限制连接数功能， 必定报错。 老哥，麻烦赐教下，谢谢！
events.type: comment
events.action: created
events.author: chenshuo
events.masked_author: username_1
events.text: This issue is not about muduo code.
  
  Thread.cpp is not in muduo's source code.
events.type: comment
events.action: closed
events.author: chenshuo
events.masked_author: username_1
events.type: issue
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: Thread.cpp是我将你的cc重命名了， 代码未做变动。 我使用的是main ractor+many sub ractor+ thread pool模式处理的，且结合了bucket剔除空闲连接和限制连接数目。测试连接数目超过限制，必挂！ 麻烦您看看怎么回事，谢谢！  没问题，我不会说有问题的。
events.type: comment
events.action: created
events.author: chenshuo
events.masked_author: username_1
events.text: 这是你自己写的程序出了问题，不是 muduo 的代码有问题，是你用错了。
  如果你能够用 muduo 的某个 example 不加改动就复现出问题，这才算是 muduo 的 issue 。
  
  你又不把代码完整地亮出来，别人无法复现，自然也帮不到你。
  
  空闲连接那个例子是单线程的，你直接照搬到多线程里，当然会出错。
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: 挂在onStringMessage函数了。 麻烦陈哥帮忙看看，多谢！
  
  #include "server.h"
  
  using namespace boost;
  using namespace muduo;
  using namespace muduo::net;
  using namespace std;
  
  
  
  extern Config config;
  
  
  CutwdServer::CutwdServer(EventLoop* loop, const InetAddress& listenAddr)
  : server_(loop, listenAddr, "subEventThPool"),
    startTime_(Timestamp::now()),
    codec_(boost::bind(&CutwdServer::onStringMessage, this, _1, _2, _3, _4)),
    numConnected_(0),
    connectionBuckets_(config.conOutTime),
    mutex_(new MutexLock)
  {
  	server_.setThreadNum(config.subEventLoop_ThreadNum); 
  
  	server_.setConnectionCallback(
  	    boost::bind(&CutwdServer::onConnection, this, _1));
  
  	server_.setMessageCallback(        
  		boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
  
  	loop->runEvery(1.0, boost::bind(&CutwdServer::onTimer, this));
  	connectionBuckets_.resize(config.conOutTime);
  }
  
  
  
  void CutwdServer::start()
  {
  	LOG_INFO << "starting " << config.worker_ThreadNum << " threads.";
  	threadPool_.start(config.worker_ThreadNum);
  	server_.start();
  }
  
  
  
  void CutwdServer::onConnection(const TcpConnectionPtr& conn)
  {
  	 LOG_INFO <<"CutwdServer::onConnection"<< " [client_ip_port]: " <<conn->peerAddress().toIpPort()
  	 		  << (conn->connected() ? " UP" : " DOWN");
  	 
  	 if (conn->connected()){   
  	 	++numConnected_;    
  		if (numConnected_ > config.maxConnections){   
  			LOG_INFO << "numConnected_>"<<config.maxConnections << " shutdown "<<conn->peerAddress().toIpPort();
  			conn->forceClose();      
  			conn->forceCloseWithDelay(3.0);    
  		}else{
  			EntryPtr entry(new Entry(conn));
  
  			if (mutex_){
  			    MutexLockGuard lock(*mutex_);
  		    	connectionBuckets_.back().insert(entry);
  			}
  			 
  		    WeakEntryPtr weakEntry(entry);
  			//保存在连接的上下文中
  		    conn->setContext(weakEntry);
  		}  
  	 }else  {    
  	 	--numConnected_;  
  		 assert(!conn->getContext().empty());    
  		 WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));
  	 }
  }
  
  
  
  void CutwdServer::onStringMessage(const TcpConnectionPtr& conn, 
  									   const string& message, 
  									   Timestamp, 
  									   Head &PacketHead)
  {
  
  	string miwen = "";
  	string Getbody = "";
  	-------
  	processRequest(conn, Getbody);
  	
  
  	assert(!conn->getContext().empty());  
  	WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));  
  	
  	EntryPtr entry(weakEntry.lock());  
  	if (entry)  {    
  		if (mutex_){
  			MutexLockGuard lock(*mutex_);
  			connectionBuckets_.back().insert(entry);    
  		}
  	}
  }
  
  
  
  void CutwdServer::onTimer()
  {
  	if (mutex_){
  		MutexLockGuard lock(*mutex_);
    		connectionBuckets_.push_back(Bucket());
  	}
  }
  
  
  
  void CutwdServer::solve(const TcpConnectionPtr& conn,
                  const string& request)
  {
  	muduo::net::Buffer buf;
  	buf.append("hello");
  	int32_t len = static_cast<int32_t>(strlen("hello"));
  	int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
  	buf.prepend(&be32, sizeof be32);
  	conn->send(&buf);
  }
  
  
  
  void CutwdServer::processRequest(const TcpConnectionPtr& conn, const string& request)
  {
    	threadPool_.run(boost::bind(&solve, conn, request));
  }
events.type: comment
events.action: created
events.author: chenshuo
events.masked_author: username_1
events.text: 你认为别人有没有可能用你贴出来的东西复现出故障？
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: **主函数文件 server_threadpool.cpp**：
  #include "server.h"
  int main(int argc, char* argv[])
  {
  	EventLoop loop;
  	InetAddress listenAddr(8001);
  	CutwdServer server(&loop, listenAddr);
  	server.start();
  	loop.loop();
  }
  
  **server.h：**
  #include <muduo/base/Mutex.h>
  #include <muduo/base/Types.h>
  #include <boost/bind.hpp>
  #include <boost/atomic.hpp>
  #include <boost/circular_buffer.hpp>
  #include <boost/unordered_set.hpp>
  #include <boost/version.hpp>
  #include <boost/noncopyable.hpp>
  #include <boost/scoped_ptr.hpp>
  #include "headbody.h"
  using namespace boost;
  using namespace muduo;
  using namespace muduo::net;
  using namespace std;
  
  #if BOOST_VERSION < 104700
  	namespace boost
  	{
  		template <typename T>
  		inline size_t hash_value(const boost::shared_ptr<T>& x)
  		{
  		  return boost::hash_value(x.get());
  		}
  	}
  #endif
  
  class CutwdServer
  {
   public:
  	CutwdServer(EventLoop* loop, const InetAddress& listenAddr);
  	void start();
  
   private:
  	typedef boost::weak_ptr<muduo::net::TcpConnection> WeakTcpConnectionPtr;
  	struct Entry : public muduo::copyable{
  		 explicit Entry(const WeakTcpConnectionPtr& weakConn)
  		   : weakConn_(weakConn)
  		 {
  		 }
  	  
  		 ~Entry()
  		 {
  		   muduo::net::TcpConnectionPtr conn = weakConn_.lock();
  		   if (conn)
  		   {
  		     LOG_INFO << conn->peerAddress().toIpPort()<<" timeout!!!";
  			 conn->shutdown();
  		   }
  		 }
  	  
  		 WeakTcpConnectionPtr weakConn_;
  	};
  	typedef boost::shared_ptr<Entry> EntryPtr;
  	typedef boost::weak_ptr<Entry> WeakEntryPtr;
  	typedef boost::unordered_set<EntryPtr> Bucket;
  	typedef boost::circular_buffer<Bucket> WeakConnectionList;
  	void onConnection(const TcpConnectionPtr& conn);
  	void onStringMessage(const TcpConnectionPtr& conn, 
  									   const string& message, 
  									   Timestamp);
  
  	void onTimer();
  	void processRequest(const TcpConnectionPtr& conn, const string& request);
  	static void solve(const TcpConnectionPtr& conn, const string& request);
  
  	TcpServer server_;
  	ThreadPool threadPool_;
  	Timestamp startTime_;
  	LengthHeaderCodec codec_;
  	atomic_int numConnected_;  
  	WeakConnectionList connectionBuckets_;
  	boost::scoped_ptr<MutexLock> mutex_;
  };
  #endif 
  
  **server.cpp：**
  #include "server.h"
  using namespace boost;
  using namespace muduo;
  using namespace muduo::net;
  using namespace std;
  
  CutwdServer::CutwdServer(EventLoop* loop, const InetAddress& listenAddr)
  : server_(loop, listenAddr, "subEventThPool"),
    startTime_(Timestamp::now()),
    codec_(boost::bind(&CutwdServer::onStringMessage, this, _1, _2, _3)),
    numConnected_(0),
    connectionBuckets_(20),
    mutex_(new MutexLock)
  {
  	server_.setThreadNum(2); 
  
  	server_.setConnectionCallback(
  	    boost::bind(&CutwdServer::onConnection, this, _1));
  	server_.setMessageCallback(        
  		boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
  
  	loop->runEvery(1.0, boost::bind(&CutwdServer::onTimer, this));
  	connectionBuckets_.resize(20);
  }
  
  void CutwdServer::start()
  {
  	threadPool_.start(2);
  	server_.start();
  }
  
  void CutwdServer::onConnection(const TcpConnectionPtr& conn)
  {
  	 LOG_INFO <<"CutwdServer::onConnection"<< " [client_ip_port]: " <<conn->peerAddress().toIpPort()
  	 		  << (conn->connected() ? " UP" : " DOWN");
  	 
  	 if (conn->connected()){   
  	 	++numConnected_;    
  		if (numConnected_ > 5){   
  			conn->forceClose();      
  			conn->forceCloseWithDelay(3.0);    
  		}else{
  			EntryPtr entry(new Entry(conn));
  
  			if (mutex_){
  			    MutexLockGuard lock(*mutex_);
  		    	connectionBuckets_.back().insert(entry);
  			}
  			 
  		    WeakEntryPtr weakEntry(entry);
  		    conn->setContext(weakEntry);
  		}  
  	 }else  {    
  	 	--numConnected_;  
  		 assert(!conn->getContext().empty());    
  		 WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));
  	 }
  }
  
  void CutwdServer::onStringMessage(const TcpConnectionPtr& conn, 
  									   const string& message, 
  									   Timestamp)
  {
  	processRequest(conn, message);
  	
  	assert(!conn->getContext().empty());  
  	WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));  
  	EntryPtr entry(weakEntry.lock());  
  	if (entry)  {    
  		if (mutex_){
  			MutexLockGuard lock(*mutex_);
  			connectionBuckets_.back().insert(entry);    
  		}
  	}
  }
  
  void CutwdServer::onTimer()
  {
  	if (mutex_){
  		MutexLockGuard lock(*mutex_);
    		connectionBuckets_.push_back(Bucket());
  	}
  }
  
  void CutwdServer::solve(const TcpConnectionPtr& conn,
                  const string& request)
  {
  	LOG_INFO <<"CutwdServer::solve"<< " server_ip_port_clientId=" << conn->name()<< " recvdata="<<request;
  	muduo::net::Buffer buf;
  	buf.append("hello");
  	int32_t len = static_cast<int32_t>(strlen("hello"));
  	int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
  	buf.prepend(&be32, sizeof be32);
  	conn->send(&buf);
  }
  
  void CutwdServer::processRequest(const TcpConnectionPtr& conn, const string& request)
  {
    	threadPool_.run(boost::bind(&solve, conn, request));
  }
  
  **headbody.h：**
  #ifndef MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  #define MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  
  #include <muduo/base/Logging.h>
  #include <muduo/net/Buffer.h>
  #include <muduo/net/Endian.h>
  #include <muduo/net/TcpConnection.h>
  
  #include <boost/function.hpp>
  #include <boost/noncopyable.hpp>
  
  class LengthHeaderCodec : boost::noncopyable
  {
   public:
    typedef boost::function<void (const muduo::net::TcpConnectionPtr&,
                                  const muduo::string& message,
                                  muduo::Timestamp)> StringMessageCallback;
  
    explicit LengthHeaderCodec(const StringMessageCallback& cb)
      : messageCallback_(cb)
    {
    }
  
    void onMessage(const muduo::net::TcpConnectionPtr& conn,
                   muduo::net::Buffer* buf,
                   muduo::Timestamp receiveTime)
    {
      while (buf->readableBytes() >= kHeaderLen) // kHeaderLen == 4
      {
        // FIXME: use Buffer::peekInt32()
        const void* data = buf->peek();
        int32_t be32 = *static_cast<const int32_t*>(data); // SIGBUS
        const int32_t len = muduo::net::sockets::networkToHost32(be32);
        if (len > 65536 || len < 0)
        {
          LOG_ERROR << "Invalid length " << len;
          conn->shutdown();  // FIXME: disable reading
          break;
        }
        else if (buf->readableBytes() >= len + kHeaderLen)
        {
          buf->retrieve(kHeaderLen);
          muduo::string message(buf->peek(), len);
          messageCallback_(conn, message, receiveTime);
          buf->retrieve(len);
        }
        else
        {
          break;
        }
      }
    }
  
    // FIXME: TcpConnectionPtr
    void send(muduo::net::TcpConnection* conn,
              const muduo::StringPiece& message)
    {
      muduo::net::Buffer buf;
      buf.append(message.data(), message.size());
      int32_t len = static_cast<int32_t>(message.size());
      int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
      buf.prepend(&be32, sizeof be32);
      conn->send(&buf);
    }
  
   private:
    StringMessageCallback messageCallback_;
    const static size_t kHeaderLen = sizeof(int32_t);
  };
  
  #endif  // MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  
  客户端 client.cpp：
  #include "headbody.h"
  
  #include <muduo/base/Logging.h>
  #include <muduo/net/EventLoop.h>
  #include <muduo/net/TcpClient.h>
  
  #include <boost/bind.hpp>
  #include <boost/ptr_container/ptr_vector.hpp>
  #include <muduo/json/dealjson.h>
  #include <muduo/code/endecode.h>
  
  
  #include "headbody.h"
  
  #include <fstream>
  
  #include <stdio.h>
  
  using namespace muduo;
  using namespace muduo::net;
  
  bool verify(const string& result)
  {
    return true;
  }
  
  typedef std::vector<string> Input;
  typedef boost::shared_ptr<Input> InputPtr;
  
  InputPtr readInput(std::istream& in)
  {
    InputPtr input(new Input);
    std::string line;
    while (getline(in, line))
    {
        input->push_back(line.c_str());
    }
    return input;
  }
  
  typedef boost::function<void(const string&, double, int)> DoneCallback;
  
  class SudokuClient : boost::noncopyable
  {
   public:
    SudokuClient(EventLoop* loop,
                 const InetAddress& serverAddr,
                 const InputPtr& input,
                 const string& name,
                 const DoneCallback& cb
                 )
      : name_(name),
        client_(loop, serverAddr, name_),
        input_(input),
        cb_(cb),
        count_(0),
        codec_(boost::bind(&SudokuClient::onStringMessage, this, _1, _2, _3))
    {
      client_.setConnectionCallback(
          boost::bind(&SudokuClient::onConnection, this, _1));
      client_.setMessageCallback(
          boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
    }
  
    void connect()
    {
      client_.connect();
    }
  
   private:
    void onConnection(const TcpConnectionPtr& conn)
    {
      if (conn->connected())
      {
        start_ = Timestamp::now();
  
        for (size_t i = 0; i < input_->size(); ++i)
        {
          LogStream buf;
          buf << (*input_)[i];
  
  		StringPiece message(buf.buffer().data(), buf.buffer().length());
  		LOG_INFO << "[client]: send=" << message;
  		codec_.send(get_pointer(conn),message);
        }
  
      }
      else
      {
        LOG_INFO << name_ << " disconnected";
      }
    }
  
  
     void onStringMessage(const TcpConnectionPtr &conn,
                         const string& message,
                         Timestamp)
    {
  	LOG_INFO << "[client]: recv " << message;
  	//conn->shutdown();
    }
  
    string name_;
    TcpClient client_;
    InputPtr input_;
    DoneCallback cb_;
    int count_;
    LengthHeaderCodec codec_;
    Timestamp start_;
    
  };
  
  Timestamp g_start;
  int g_connections;
  int g_finished;
  EventLoop* g_loop;
  
  void done(const string& name, double elapsed, int count)
  {
    LOG_INFO << name << " " << elapsed << " seconds "
             << Fmt("%.3f", 1000 * 1000 * elapsed / count)
             << " us per request.";
    ++g_finished;
    if (g_finished == g_connections)
    {
      g_loop->runAfter(1.0, boost::bind(&EventLoop::quit, g_loop));
      double total = timeDifference(Timestamp::now(), g_start);
      LOG_INFO << "total " << total << " seconds, "
               << (total/g_connections) << " seconds per client";
    }
  }
  
  
  
  void runClient(std::istream& in, const InetAddress& serverAddr, int conn)
  {
    InputPtr input(readInput(in));
    EventLoop loop;
    g_loop = &loop;
    g_connections = conn;
  
    g_start = Timestamp::now();
    boost::ptr_vector<SudokuClient> clients;
    for (int i = 0; i < conn; ++i)
    {
      Fmt f("client-%03d", i+1);
      string name(f.data(), f.length());
      clients.push_back(new SudokuClient(&loop, serverAddr, input, name, done));
      clients.back().connect();
    }
  
    loop.loop();
  }
  
  
  /*
  ./sudoku_client  test.txt 127.0.0.1  6
  */
  int main(int argc, char* argv[])
  {
    int conn = 1;
    InetAddress serverAddr("127.0.0.1", 8001);
    const char* input = NULL;
  
    switch (argc)
    {
      case 4:
        conn = atoi(argv[3]);
        // FALL THROUGH
      case 3:
        serverAddr = InetAddress(argv[2], 8001);
      case 2:
        input = argv[1];
        break;
      default:
        printf("Usage: %s input server_ip [connections]\n", argv[0]);
        return 0;
    }
  
    std::ifstream in(input);
    if (in)
    {
        runClient(in, serverAddr, conn);
    }
    
  }
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: **主函数文件 server_threadpool.cpp**：
  #include "server.h"
  int main(int argc, char* argv[])
  {
  	EventLoop loop;
  	InetAddress listenAddr(8001);
  	CutwdServer server(&loop, listenAddr);
  	server.start();
  	loop.loop();
  }
  
  **server.h：**
  #include <muduo/base/Mutex.h>
  #include <muduo/base/Types.h>
  #include <boost/bind.hpp>
  #include <boost/atomic.hpp>
  #include <boost/circular_buffer.hpp>
  #include <boost/unordered_set.hpp>
  #include <boost/version.hpp>
  #include <boost/noncopyable.hpp>
  #include <boost/scoped_ptr.hpp>
  #include "headbody.h"
  using namespace boost;
  using namespace muduo;
  using namespace muduo::net;
  using namespace std;
  
  #if BOOST_VERSION < 104700
  	namespace boost
  	{
  		template <typename T>
  		inline size_t hash_value(const boost::shared_ptr<T>& x)
  		{
  		  return boost::hash_value(x.get());
  		}
  	}
  #endif
  
  class CutwdServer
  {
   public:
  	CutwdServer(EventLoop* loop, const InetAddress& listenAddr);
  	void start();
  
   private:
  	typedef boost::weak_ptr<muduo::net::TcpConnection> WeakTcpConnectionPtr;
  	struct Entry : public muduo::copyable{
  		 explicit Entry(const WeakTcpConnectionPtr& weakConn)
  		   : weakConn_(weakConn)
  		 {
  		 }
  	  
  		 ~Entry()
  		 {
  		   muduo::net::TcpConnectionPtr conn = weakConn_.lock();
  		   if (conn)
  		   {
  		     LOG_INFO << conn->peerAddress().toIpPort()<<" timeout!!!";
  			 conn->shutdown();
  		   }
  		 }
  	  
  		 WeakTcpConnectionPtr weakConn_;
  	};
  	typedef boost::shared_ptr<Entry> EntryPtr;
  	typedef boost::weak_ptr<Entry> WeakEntryPtr;
  	typedef boost::unordered_set<EntryPtr> Bucket;
  	typedef boost::circular_buffer<Bucket> WeakConnectionList;
  	void onConnection(const TcpConnectionPtr& conn);
  	void onStringMessage(const TcpConnectionPtr& conn, 
  									   const string& message, 
  									   Timestamp);
  
  	void onTimer();
  	void processRequest(const TcpConnectionPtr& conn, const string& request);
  	static void solve(const TcpConnectionPtr& conn, const string& request);
  
  	TcpServer server_;
  	ThreadPool threadPool_;
  	Timestamp startTime_;
  	LengthHeaderCodec codec_;
  	atomic_int numConnected_;  
  	WeakConnectionList connectionBuckets_;
  	boost::scoped_ptr<MutexLock> mutex_;
  };
  #endif 
  
  **server.cpp：**
  #include "server.h"
  using namespace boost;
  using namespace muduo;
  using namespace muduo::net;
  using namespace std;
  
  CutwdServer::CutwdServer(EventLoop* loop, const InetAddress& listenAddr)
  : server_(loop, listenAddr, "subEventThPool"),
    startTime_(Timestamp::now()),
    codec_(boost::bind(&CutwdServer::onStringMessage, this, _1, _2, _3)),
    numConnected_(0),
    connectionBuckets_(20),
    mutex_(new MutexLock)
  {
  	server_.setThreadNum(2); 
  
  	server_.setConnectionCallback(
  	    boost::bind(&CutwdServer::onConnection, this, _1));
  	server_.setMessageCallback(        
  		boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
  
  	loop->runEvery(1.0, boost::bind(&CutwdServer::onTimer, this));
  	connectionBuckets_.resize(20);
  }
  
  void CutwdServer::start()
  {
  	threadPool_.start(2);
  	server_.start();
  }
  
  void CutwdServer::onConnection(const TcpConnectionPtr& conn)
  {
  	 LOG_INFO <<"CutwdServer::onConnection"<< " [client_ip_port]: " <<conn->peerAddress().toIpPort()
  	 		  << (conn->connected() ? " UP" : " DOWN");
  	 
  	 if (conn->connected()){   
  	 	++numConnected_;    
  		if (numConnected_ > 5){   
  			conn->forceClose();      
  			conn->forceCloseWithDelay(3.0);    
  		}else{
  			EntryPtr entry(new Entry(conn));
  
  			if (mutex_){
  			    MutexLockGuard lock(*mutex_);
  		    	connectionBuckets_.back().insert(entry);
  			}
  			 
  		    WeakEntryPtr weakEntry(entry);
  		    conn->setContext(weakEntry);
  		}  
  	 }else  {    
  	 	--numConnected_;  
  		 assert(!conn->getContext().empty());    
  		 WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));
  	 }
  }
  
  void CutwdServer::onStringMessage(const TcpConnectionPtr& conn, 
  									   const string& message, 
  									   Timestamp)
  {
  	processRequest(conn, message);
  	
  	assert(!conn->getContext().empty());  
  	WeakEntryPtr weakEntry(boost::any_cast<WeakEntryPtr>(conn->getContext()));  
  	EntryPtr entry(weakEntry.lock());  
  	if (entry)  {    
  		if (mutex_){
  			MutexLockGuard lock(*mutex_);
  			connectionBuckets_.back().insert(entry);    
  		}
  	}
  }
  
  void CutwdServer::onTimer()
  {
  	if (mutex_){
  		MutexLockGuard lock(*mutex_);
    		connectionBuckets_.push_back(Bucket());
  	}
  }
  
  void CutwdServer::solve(const TcpConnectionPtr& conn,
                  const string& request)
  {
  	LOG_INFO <<"CutwdServer::solve"<< " server_ip_port_clientId=" << conn->name()<< " recvdata="<<request;
  	muduo::net::Buffer buf;
  	buf.append("hello");
  	int32_t len = static_cast<int32_t>(strlen("hello"));
  	int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
  	buf.prepend(&be32, sizeof be32);
  	conn->send(&buf);
  }
  
  void CutwdServer::processRequest(const TcpConnectionPtr& conn, const string& request)
  {
    	threadPool_.run(boost::bind(&solve, conn, request));
  }
  
  **headbody.h：**
  #ifndef MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  #define MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  
  #include <muduo/base/Logging.h>
  #include <muduo/net/Buffer.h>
  #include <muduo/net/Endian.h>
  #include <muduo/net/TcpConnection.h>
  
  #include <boost/function.hpp>
  #include <boost/noncopyable.hpp>
  
  class LengthHeaderCodec : boost::noncopyable
  {
   public:
    typedef boost::function<void (const muduo::net::TcpConnectionPtr&,
                                  const muduo::string& message,
                                  muduo::Timestamp)> StringMessageCallback;
  
    explicit LengthHeaderCodec(const StringMessageCallback& cb)
      : messageCallback_(cb)
    {
    }
  
    void onMessage(const muduo::net::TcpConnectionPtr& conn,
                   muduo::net::Buffer* buf,
                   muduo::Timestamp receiveTime)
    {
      while (buf->readableBytes() >= kHeaderLen) // kHeaderLen == 4
      {
        // FIXME: use Buffer::peekInt32()
        const void* data = buf->peek();
        int32_t be32 = *static_cast<const int32_t*>(data); // SIGBUS
        const int32_t len = muduo::net::sockets::networkToHost32(be32);
        if (len > 65536 || len < 0)
        {
          LOG_ERROR << "Invalid length " << len;
          conn->shutdown();  // FIXME: disable reading
          break;
        }
        else if (buf->readableBytes() >= len + kHeaderLen)
        {
          buf->retrieve(kHeaderLen);
          muduo::string message(buf->peek(), len);
          messageCallback_(conn, message, receiveTime);
          buf->retrieve(len);
        }
        else
        {
          break;
        }
      }
    }
  
    // FIXME: TcpConnectionPtr
    void send(muduo::net::TcpConnection* conn,
              const muduo::StringPiece& message)
    {
      muduo::net::Buffer buf;
      buf.append(message.data(), message.size());
      int32_t len = static_cast<int32_t>(message.size());
      int32_t be32 = muduo::net::sockets::hostToNetwork32(len);
      buf.prepend(&be32, sizeof be32);
      conn->send(&buf);
    }
  
   private:
    StringMessageCallback messageCallback_;
    const static size_t kHeaderLen = sizeof(int32_t);
  };
  
  #endif  // MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H
  
  客户端 client.cpp：
  #include "headbody.h"
  
  #include <muduo/base/Logging.h>
  #include <muduo/net/EventLoop.h>
  #include <muduo/net/TcpClient.h>
  
  #include <boost/bind.hpp>
  #include <boost/ptr_container/ptr_vector.hpp>
  #include <muduo/json/dealjson.h>
  #include <muduo/code/endecode.h>
  
  
  #include "headbody.h"
  
  #include <fstream>
  
  #include <stdio.h>
  
  using namespace muduo;
  using namespace muduo::net;
  
  bool verify(const string& result)
  {
    return true;
  }
  
  typedef std::vector<string> Input;
  typedef boost::shared_ptr<Input> InputPtr;
  
  InputPtr readInput(std::istream& in)
  {
    InputPtr input(new Input);
    std::string line;
    while (getline(in, line))
    {
        input->push_back(line.c_str());
    }
    return input;
  }
  
  typedef boost::function<void(const string&, double, int)> DoneCallback;
  
  class SudokuClient : boost::noncopyable
  {
   public:
    SudokuClient(EventLoop* loop,
                 const InetAddress& serverAddr,
                 const InputPtr& input,
                 const string& name,
                 const DoneCallback& cb
                 )
      : name_(name),
        client_(loop, serverAddr, name_),
        input_(input),
        cb_(cb),
        count_(0),
        codec_(boost::bind(&SudokuClient::onStringMessage, this, _1, _2, _3))
    {
      client_.setConnectionCallback(
          boost::bind(&SudokuClient::onConnection, this, _1));
      client_.setMessageCallback(
          boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
    }
  
    void connect()
    {
      client_.connect();
    }
  
   private:
    void onConnection(const TcpConnectionPtr& conn)
    {
      if (conn->connected())
      {
        start_ = Timestamp::now();
  
        for (size_t i = 0; i < input_->size(); ++i)
        {
          LogStream buf;
          buf << (*input_)[i];
  
  		StringPiece message(buf.buffer().data(), buf.buffer().length());
  		LOG_INFO << "[client]: send=" << message;
  		codec_.send(get_pointer(conn),message);
        }
  
      }
      else
      {
        LOG_INFO << name_ << " disconnected";
      }
    }
  
  
     void onStringMessage(const TcpConnectionPtr &conn,
                         const string& message,
                         Timestamp)
    {
  	LOG_INFO << "[client]: recv " << message;
  	//conn->shutdown();
    }
  
    string name_;
    TcpClient client_;
    InputPtr input_;
    DoneCallback cb_;
    int count_;
    LengthHeaderCodec codec_;
    Timestamp start_;
    
  };
  
  Timestamp g_start;
  int g_connections;
  int g_finished;
  EventLoop* g_loop;
  
  void done(const string& name, double elapsed, int count)
  {
    LOG_INFO << name << " " << elapsed << " seconds "
             << Fmt("%.3f", 1000 * 1000 * elapsed / count)
             << " us per request.";
    ++g_finished;
    if (g_finished == g_connections)
    {
      g_loop->runAfter(1.0, boost::bind(&EventLoop::quit, g_loop));
      double total = timeDifference(Timestamp::now(), g_start);
      LOG_INFO << "total " << total << " seconds, "
               << (total/g_connections) << " seconds per client";
    }
  }
  
  
  
  void runClient(std::istream& in, const InetAddress& serverAddr, int conn)
  {
    InputPtr input(readInput(in));
    EventLoop loop;
    g_loop = &loop;
    g_connections = conn;
  
    g_start = Timestamp::now();
    boost::ptr_vector<SudokuClient> clients;
    for (int i = 0; i < conn; ++i)
    {
      Fmt f("client-%03d", i+1);
      string name(f.data(), f.length());
      clients.push_back(new SudokuClient(&loop, serverAddr, input, name, done));
      clients.back().connect();
    }
  
    loop.loop();
  }
  
  
  /*
  ./sudoku_client  test.txt 127.0.0.1  6
  */
  int main(int argc, char* argv[])
  {
    int conn = 1;
    InetAddress serverAddr("127.0.0.1", 8001);
    const char* input = NULL;
  
    switch (argc)
    {
      case 4:
        conn = atoi(argv[3]);
        // FALL THROUGH
      case 3:
        serverAddr = InetAddress(argv[2], 8001);
      case 2:
        input = argv[1];
        break;
      default:
        printf("Usage: %s input server_ip [connections]\n", argv[0]);
        return 0;
    }
  
    std::ifstream in(input);
    if (in)
    {
        runClient(in, serverAddr, conn);
    }
    
  }
events.type: comment
events.action: created
events.author: chenshuo
events.masked_author: username_1
events.text: 你有没有试验过这样贴代码别人能不能 copy 下来编译？
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: 算了，自己解决吧。
events.type: comment
events.action: created
events.author: ygm521
events.masked_author: username_0
events.text: 每次提问，都是无结果。 强烈推荐国内openresty!!
events.type: comment