#0 0x00007f89817f4625 in raise () from /lib64/libc.so.6 #1 0x00007f89817f5e05 in abort () from /lib64/libc.so.6 #2 0x00007f898204fa55 in __gnu_cxx::__verbose_terminate_handler () at ../../.././libstdc++-v3/libsupc++/vterminate.cc:95 #3 0x00007f898204dbf6 in __cxxabiv1::__terminate (handler=) at ../../.././libstdc++-v3/libsupc++/eh_terminate.cc:38 #4 0x00007f898204dc23 in std::terminate () at ../../.././libstdc++-v3/libsupc++/eh_terminate.cc:48 #5 0x00007f898204e6df in __cxxabiv1::__cxa_pure_virtual () at ../../.././libstdc++-v3/libsupc++/pure.cc:50 #6 0x0000000000631318 in sofa::pbrpc::RpcByteStream::on_connect(boost::system::error_code const&) () #7 0x000000000062aaed in boost::asio::detail::reactive_socket_connect_op<boost::bi::bind_t<void, boost::mfi::mf1<void, sofa::pbrpc::RpcByteStream, boost::system::error_code const&>, boost::bi::list2boost::bi::value<sofa::pbrpc::shared_ptr<sofa::pbrpc::RpcByteStream >, boost::arg<1> > > >::do_complete(boost::asio::detail::task_io_service, boost::asio::detail::task_io_service_operation, boost::system::error_code const&, unsigned long) () #8 0x000000000062dee9 in boost::asio::detail::epoll_reactor::descriptor_state::do_complete(boost::asio::detail::task_io_service, boost::asio::detail::task_io_service_operation, boost::system::error_code const&, unsigned long) () #9 0x000000000063081e in boost::asio::detail::task_io_service::run(boost::system::error_code&) () #10 0x000000000063118e in sofa::pbrpc::ThreadGroupImpl::thread_run(void*) () #11 0x00007f89827749d1 in start_thread () from /lib64/libpthread.so.0

A new option of rpc_controller connect_timeout is used for connect action.

When rpc_client use rpc_stream to create a connect, it will use rpc_controller::connect_timeout for connect action if it is > 0.

The default of rpc_controller::connect_timeout is -1.

Signed-off-by: Dong Yuan [email protected]

主要改动

• 增加WebService类
• 路由及拼装页面从HttpRpcRequest中移至WebService中
• 提供用户RegisterWebServlet接口以及HttpRequest和HttpResponse的数据结构

功能描述

• 用户使用RegisterWebServlet接口，注册路径及其处理函数，rpc框架收到http请求后准发至用户函数并返回
• 用户现在可以覆盖框架默认的处理函数
• 假如用户只注册了 /dfs 路径，那么 /dfs/* 都会被转发到 /dfs 的处理函数

Hi,

I've added Bazel config files to allow reproductible builds. I tested them on OS X and Linux (Ubuntu).

When building using Bazel, it will download all required dependencies and will compile static binaries for host platform.

To build sofa-pbrpc-client, run:

$ bazel build sofa-pbrpc-client

To build examples:

$ bazel build sample/compress_sample:server
$ bazel build sample/compress_sample:client

$ bazel-bin/sample/compress_sample/server
$ bazel-bin/sample/compress_sample/client

栈状态 #0 0x00007f1834e7b22d in pthread_join () from /lib64/libpthread.so.0 #1 0x00000000006215ab in sofa::pbrpc::ThreadGroupImpl::stop (this=0x228f6b0) at src/sofa/pbrpc/thread_group_impl.h:182 #2 0x00000000006176c7 in sofa::pbrpc::RpcClientImpl::Stop (this=0x22be000) at src/sofa/pbrpc/rpc_client_impl.cc:109

查看io_service的内存信息如下 (gdb) p *(boost::asio::detail::task_io_service * const) 0x22738e0 $26 = {boost::asio::detail::service_baseboost::asio::detail::task_io_service = {boost::asio::io_service::service = {boost::noncopyable_::noncopyable = {}, vptr.service = 0xa24a90 <vtable for boost::asio::detail::task_io_service+16>, key = {type_info_ = 0xa245a0 <typeinfo for boost::asio::detail::typeid_wrapperboost::asio::detail::task_io_service>, id_ = 0x0}, owner_ = @0x228f6d0, next_ = 0x0}, static id = {boost::asio::io_service::id = {boost::noncopyable_::noncopyable = {}, }, }}, one_thread_ = false, mutex_ = {boost::noncopyable_::noncopyable = {}, mutex_ = {__data = {__lock = 0, __count = 0, __owner = 0, _nusers = 7, kind = 0, spins = 0, list = { prev = 0x0, next = 0x0}}, size = '\000' <repeats 12 times>, "\a", '\000' <repeats 26 times>, align = 0}}, task = 0x2266e10, task_operation = {boost::asio::detail::task_io_service_operation = {next = 0x0, func = 0x0, task_result = 0}, }, task_interrupted = false, outstanding_work = {value = 3}, op_queue = {boost::noncopyable::noncopyable = {}, front = 0x0, back = 0x0}, stopped = false, shutdown = false, first_idle_thread = 0x7f182991fce0}

我理解调用后stop函数后task_io_service 的 outstanding_work_变量会被减为0 并退出他的run函数。 从而使得pthread_join函数成功返回。可能的问题点在哪里呢？

sofa-pbrpc，编译对g++的版本，有什么要求呢？我用g++ 4.8.2编译失败了，protubuf，用的是proto 3.0版本

g++ -O2 -pipe -W -Wall -fPIC -D_GNU_SOURCE -D__STDC_LIMIT_MACROS -DHAVE_SNAPPY -Isrc -I/usr/include -I/home/kentpeng/sofa-pbrpc/thirty/protobuf/include -I/home/kentpeng/sofa-pbrpc/thirty/snappy/include -I/home/kentpeng/sofa-pbrpc/thirty/zlib/include -c -o src/sofa/pbrpc/boost_system_error_code.o src/sofa/pbrpc/boost_system_error_code.cc src/sofa/pbrpc/boost_system_error_code.cc:440:1: internal compiler error: in function_and_variable_visibility, at ipa.c:815 } // namespace boost ^ Please submit a full bug report, with preprocessed source if appropriate. See file:///usr/share/doc/gcc-4.8/README.Bugs for instructions. Preprocessed source stored into /tmp/ccpGeFhP.out file, please attach this to your bugreport. make: *** [src/sofa/pbrpc/boost_system_error_code.o] Error 1

Hi all,

我们的bug检测工具报告了两个数据竞争，都在ThreadGroupImpl::start() 和 ThreadGroupImpl::thread_run()这两个函数中。位置分别是 thread_group_impl.c#L255, thread_group_impl.c#L135和thread_group_impl.c#L257, thread_group_impl.c#L133

下面是代码片段

    static void* thread_run(void* param)
    {
        ThreadParam* thread_param = reinterpret_cast<ThreadParam*>(param);
        // init
        if (thread_param->init_func && !thread_param->init_func->Run())
        {
            ...
            thread_param->init_fail = true;
        }
        thread_param->init_done = true;

and

    bool start()
    {
        ...
        for (int i = 0; i < _thread_num; ++i)
        {
            ....
            int ret = pthread_create(&_threads[i], NULL, &ThreadGroupImpl::thread_run, &_thread_params[i]);
            ....
                    if (_thread_params[i].init_done)
                    {
                        if (_thread_params[i].init_fail)
                        {
                            init_fail = true;
                            break;
                        }  
                    ...

这两个函数是通过pthread_create造成并发，并且上下文中没有对这两个变量的保护。

SourceBrella Inc., Yu

描述一个现象：启动了一个server, 启动一个client使用同步方式访问这个服务，客户端看到的时间很多都是200ms，但是通过sofa_pbrpc_client，统计功能，succeed_max_time_us最大不到100us，client-server之间ping延时不到0.1ms,造成这种现象可能的原因是什么？多谢了~ @qinzuoyan

问题描述

当RequestTimeout之后，实际上对应Socket没有关闭，超时后数据还是会被接收，这在传输大数据块时，会占用较大带宽。

原因分析

这个问题存在的原因在于：

• Socket连接是会持续保持的，只要不出现传输上的错误或者keep_alive_time超时，socket是不会被关闭的。这样的好处是socket可以最大限度重用，避免重新建立连接的代价；另外socket是复用的，即不同的request在目标地址相同的情况下会共用一个socket通道，如果因为一个request超时就关闭socket是不合理的。
• 请求超时这不是传输上的错误，因此socket不会关闭，此时关键问题是server端并不知道client端已经超时了，所以依然会正常处理request并把结果发送给client端，造成带宽占用。

解决方案

考虑如下：

• 在client发送request给server端时，携带一个server_timeout，在server端收到request时记录receive_time，在请求处理完成后通过(current_time - receive_time)计算实际处理时间process_time，如果process_time > server_timeout，则可以推断此时client已经超时了，就不用发response消息了
• server_timeout是小于client_timeout的，可以通过client_timeout - (client_send_time - client_begin_time)计算得到
• 可以进一步优化，在request的发送、接收、开始处理、处理完成、发送response的任何时间点，都可以估算剩余时间，如果剩余时间耗尽，在任何时间点都可以提前终止处理，因为client已经超时，再处理也没有意义了。这可以减少不必要的request发送、request在server端的处理、response发送。
• 另外，这个问题其实与cancel是有关联的：google::protobuf::RpcController其实有一个StartCancel()接口，可以cancel掉一个request使其提前终止，其目的有共同之处。这个功能目前还没有实现，在TODO list中。

实现

参见pull request #84 但是还有改进的地方：

• 目前client直接将total_timeout作为server_timeout传给server端，实际上，request在client端的sending buffer中的排队时间没有考虑进来，尤其是网络繁忙的时候排队时间可能会很长，这样server_timeout与实际剩余时间的差别就比较大，仍然会有不必要的response。在RpcClientStream::on_sending()中留了一个TODO就是这个意思。

用户使用http协议发送数据时，性能受制于json与pb互转，在字段较多的情况下尤为明显。

为此，增加了用户post protobuf数据的接口，用户使用http client，在header中声明 Accept: "application/protobuf" 表示client可以支持protobuf数据

用户发送时使用protobuf序列化请求，作为POST方法的body；接收同样使用protobuf反序列化

客户端使用示例及性能对比稍后补充

#0 0x0000000000cb30a4 in async_result_init (orig_handler=..., this=) at thirdparty/boost/asio/async_result.hpp:62 #1 post<boost::_bi::bind_t<void, void ()(google::protobuf::Closure), boost::_bi::list1<boost::_bi::valuegoogle::protobuf::Closure* > >&> (this=, handler=...) at thirdparty/boost/asio/impl/io_service.hpp:106 #2 post<boost::_bi::bind_t<void, void ()(google::protobuf::Closure), boost::_bi::list1<boost::_bi::valuegoogle::protobuf::Closure* > > > (handler=..., this=) at ./sofa/pbrpc/thread_group_impl.h:222 #3 post (handle=0x35375bf40, this=) at ./sofa/pbrpc/thread_group_impl.h:232 #4 sofa::pbrpc::SimpleRpcChannelImpl::DoneCallback (this=0x1b5663900, done=0x35375bf40, cntl=...) at sofa/pbrpc/simple_rpc_channel_impl.cc:180 #5 0x0000000000c9936b in operator() (a0=..., this=) at thirdparty/boost/function/function_template.hpp:767 #6 sofa::pbrpc::RpcControllerImpl::Done (this=0x31d82b520, error_code=, reason=...) at ./sofa/pbrpc/rpc_controller_impl.h:222 #7 0x0000000000c8f27e in sofa::pbrpc::RpcClientImpl::CallMethod (this=, request=request@entry=0x31394c480, response=response@entry=0x22442f4a0, cntl=...) at sofa/pbrpc/rpc_client_impl.cc:227 #8 0x0000000000cb4012 in sofa::pbrpc::SimpleRpcChannelImpl::CallMethod (this=0x1b5663900, method=, controller=0x313135880, request=0x31394c480, response=0x22442f4a0, done=) at sofa/pbrpc/simple_rpc_channel_impl.cc:147 #9 0x0000000000cb9986 in sofa::pbrpc::DynamicRpcChannelImpl::CallMethod (this=0x1770a3a00, method=0x1b3de0120, controller=0x313135880, request=0x31394c480, response=0x22442f4a0, done=0x35375bf40) at sofa/pbrpc/dynamic_rpc_channel_impl.cc:204 #10 0x0000000000b84a45 in libzp::Cluster::DoNodeSyncTask (arg=0x313134900) at jtable-sofa-client/libzp/src/zp_cluster.cc:1148 #11 0x0000000000b84b1f in libzp::Cluster::AddNodeAsyncTask (node=..., context=) at jtable-sofa-client/libzp/src/zp_cluster.cc:1154 #12 0x0000000000b89951 in libzp::Cluster::DoMgetAsyncTask (arg=0x34b4aee00) at jtable-sofa-client/libzp/src/zp_cluster.cc:871 #13 0x0000000000b8a895 in libzp::Cluster::DoAsyncTask_get (arg=arg@entry=0x34b4aee00) at jtable-sofa-client/libzp/src/zp_cluster.cc:934 #14 0x0000000000b8a9cd in libzp::Cluster::AddAsyncTask_get (this=this@entry=0x2f06240, context=context@entry=0x34b4aee00) at jtable-sofa-client/libzp/src/zp_cluster.cc:982 #15 0x0000000000b8abae in libzp::Cluster::Amget (this=0x2f06240, table=..., keys=...,

函数调用栈如图， 求解答，谢谢！

例子中的接收到什么输出什么

询问一下，如何在处理的时候再去调用别的处理 类似有几部需要处理，怎么能自动触发下一步，我这里现在是两步 怎么能再第一个Echo中调用Echo2

class EchoServerImpl : public sofa::pbrpc::test::EchoServer { public: EchoServerImpl() {} virtual ~EchoServerImpl() {}

private: virtual void Echo(google::protobuf::RpcController* controller, const sofa::pbrpc::test::EchoRequest* request, sofa::pbrpc::test::EchoResponse* response, google::protobuf::Closure* done) { sofa::pbrpc::RpcController* cntl = static_castsofa::pbrpc::RpcController*(controller); SLOG(INFO, "Echo(): request message from %s: %s", cntl->RemoteAddress().c_str(), request->message().c_str()); if (cntl->IsHttp()) { SLOG(INFO, "HTTP-PATH="%s"", cntl->HttpPath().c_str()); std::map<std::string, std::string>::const_iterator it; const std::map<std::string, std::string>& query_params = cntl->HttpQueryParams(); for (it = query_params.begin(); it != query_params.end(); ++it) { SLOG(INFO, "QueryParam["%s"]="%s"", it->first.c_str(), it->second.c_str()); } const std::map<std::string, std::string>& headers = cntl->HttpHeaders(); for (it = headers.begin(); it != headers.end(); ++it) { SLOG(INFO, "Header["%s"]="%s"", it->first.c_str(), it->second.c_str()); } } response->set_message("echo message: " + request->message()); done->Run(); } virtual void Echo2(google::protobuf::RpcController* controller, const sofa::pbrpc::test::EchoRequest* request, sofa::pbrpc::test::EchoResponse* response, google::protobuf::Closure* done) { sofa::pbrpc::RpcController* cntl = static_castsofa::pbrpc::RpcController*(controller); SLOG(INFO, "Echo(): request message from %s: %s", cntl->RemoteAddress().c_str(), request->message().c_str()); if (cntl->IsHttp()) { SLOG(INFO, "HTTP-PATH="%s"", cntl->HttpPath().c_str()); std::map<std::string, std::string>::const_iterator it; const std::map<std::string, std::string>& query_params = cntl->HttpQueryParams(); for (it = query_params.begin(); it != query_params.end(); ++it) { SLOG(INFO, "QueryParam["%s"]="%s"", it->first.c_str(), it->second.c_str()); } const std::map<std::string, std::string>& headers = cntl->HttpHeaders(); for (it = headers.begin(); it != headers.end(); ++it) { SLOG(INFO, "Header["%s"]="%s"", it->first.c_str(), it->second.c_str()); } } response->set_message("echo message: " + request->message()); done->Run(); } };

看了代码中有很多volatile修饰的变量，目的是什么，为了statistics更加准确以及为了使用sofa-pbrpc中提供的原子操作吗？ 代码中有这样的注释： // And here we need not lock "_pending_lock" because the "_pending_message_count" is // an volatile value. 请教一下volatile与锁有关系吗？

RpcMessageStream::try_start_send() 1can miss pending messages because it first performs a relaxed load of _send_token which can miss store/store_release of _send_token.

RpcMessageStream::get_from_pending_queue() 2can miss pending messages because it performs a relaxed load of _pending_message_count before entering the critical section guarded by _pending_lock, again, the relaxed load can miss store/store_release of _pending_message_count.

With the above two defects combined, the current message calling try_start_send() and the previous message calling get_from_pending_queue() (on_write_some() ---> try_start_send() ---> get_from_pending_queue() ) both may miss the chance to send the current pending message, which lead to the current pending message siting idle in _pending_callsuntil timeout.

I believe the scenario described above is quite likely to happen under weak memory model such as aarch64 and ppc64.