opened
ygm521
username_0
陈哥，请假。我使用的是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);
}
main loop+many sub loop+ threadpoll 限制连接数示例结合踢掉空闲连接
issue
created
chenshuo
username_1
this is not an issue of muduo.
comment
created
ygm521
username_0
如何解决呢？ boost1.60和53均不好使。 你可以试试将踢掉空闲连接和限制连接数目实例总合下测测限制连接数功能， 必定报错。 老哥，麻烦赐教下，谢谢！
comment
created
chenshuo
username_1
This issue is not about muduo code.

Thread.cpp is not in muduo's source code.
comment
closed
chenshuo
username_1
issue
created
ygm521
username_0
Thread.cpp是我将你的cc重命名了， 代码未做变动。 我使用的是main ractor+many sub ractor+ thread pool模式处理的，且结合了bucket剔除空闲连接和限制连接数目。测试连接数目超过限制，必挂！ 麻烦您看看怎么回事，谢谢！  没问题，我不会说有问题的。
comment
created
chenshuo
username_1
这是你自己写的程序出了问题，不是 muduo 的代码有问题，是你用错了。
如果你能够用 muduo 的某个 example 不加改动就复现出问题，这才算是 muduo 的 issue 。

你又不把代码完整地亮出来，别人无法复现，自然也帮不到你。

空闲连接那个例子是单线程的，你直接照搬到多线程里，当然会出错。
comment
created
ygm521
username_0
挂在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));
}
comment
created
chenshuo
username_1
你认为别人有没有可能用你贴出来的东西复现出故障？
comment
created
ygm521
username_0
**主函数文件 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);
  }
  
}
comment
created
ygm521
username_0
**主函数文件 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);
  }
  
}
comment
created
chenshuo
username_1
你有没有试验过这样贴代码别人能不能 copy 下来编译？
comment
created
ygm521
username_0
算了，自己解决吧。
comment
created
ygm521
username_0
每次提问，都是无结果。 强烈推荐国内openresty!!
comment