EOS中plugin之net_plugin

发表于 | Edited on | 分类于 | 次阅读

这部分重点介绍EOS中的服务器端部分nodeos启动之后开启的另外一个重要的插件——net_plugin，这个插件主要负责服务器在网络中的接入、同步区块信息、断开等功能。对于这个插件，首先从其类的定义开始了解。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class net_plugin : public appbase::plugin<net_plugin>
{
    public:
    net_plugin();
    virtual ~net_plugin();

    APPBASE_PLUGIN_REQUIRES((chain_plugin))    // net_plugin这个插件的启动，依赖chain_plugin插件
    virtual void set_program_options(options_description& cli, options_description& cfg) override;

    void plugin_initialize(const variables_map& options);  // 插件初始化
    void plugin_startup();                                 // 插件的启动
    void plugin_shutdown();                                // 插件关闭

    void   broadcast_block(const chain::signed_block &sb); // 广播区块

    string                       connect( const string& endpoint ); // 连接其他端点
    string                       disconnect( const string& endpoint ); // 断开连接
    optional<connection_status>  status( const string& endpoint )const;   // 查看与某个端点的链接状态
    vector<connection_status>    connections()const;                   // 查看连接状态

    size_t num_peers() const;                                          // 查看建立连接的端点个数
    private:
    std::unique_ptr<class net_plugin_impl> my;   // 和producer_plugin_imple一样，这个插件负责网络中的具体操作
};

net_plugin插件的代码还是比较容易理解的，net_plugin的初始化函数非常简单，代码如下：

1
2
3
4
net_plugin::net_plugin()
    :my( new net_plugin_impl ) {
    my_impl = my.get();
}

初始化函数中生成了一阁net_plugin_impl的实例，随后将my的指针赋值给了net_plugin_impl类中的定义的静态指针,这个静态指针的定义如下：

1
static net_plugin_impl *my_impl;

再nodeos的main函数中，net_plugin首先会初始化，随后调用其plugin_initialize函数和EOS中lugin之producer_plugin的介绍的procuder_plugin类初始化方式时一样的，先寻找这个插件依赖的插件，然后初始化依赖的插件、一些启动参数，设置心跳计时器等。

nodeos中初始化完毕，随后调用net_plugin的plugin_startup函数，该函数代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
void net_plugin::plugin_startup() {
    my->producer_plug = app().find_plugin<producer_plugin>();
    if( my->acceptor ) {
        // 常见的网络服务操作,打开监听服务,设置选项,绑定地址,启动监听
        my->acceptor->open(my->listen_endpoint.protocol());
        my->acceptor->set_option(tcp::acceptor::reuse_address(true));
        try {
        // acceptor 来自于boost::asio::ip::tcp，即tcp::acceptor
        my->acceptor->bind(my->listen_endpoint);
        } catch (const std::exception& e) {
        ilog("net_plugin::plugin_startup failed to bind to port ${port}",
            ("port", my->listen_endpoint.port()));
        throw e;
        }
        my->acceptor->listen();
        ilog("starting listener, max clients is ${mc}",("mc",my->max_client_count));
        my->start_listen_loop();   // 循环监听函数
    }
    chain::controller&cc = my->chain_plug->chain();
    {
        cc.accepted_block.connect(  boost::bind(&net_plugin_impl::accepted_block, my.get(), _1));
    }

    my->incoming_transaction_ack_subscription = app().get_channel<channels::transaction_ack>().subscribe(boost::bind(&net_plugin_impl::transaction_ack, my.get(), _1));

    if( cc.get_read_mode() == chain::db_read_mode::READ_ONLY ) {
        my->max_nodes_per_host = 0;
        ilog( "node in read-only mode setting max_nodes_per_host to 0 to prevent connections" );
    }
    // 启动连接和交易到期的监视
    my->start_monitors();

    for( auto seed_node : my->supplied_peers ) {
        connect( seed_node );// 连接种子节点,接入p2p网络
    }

    if(fc::get_logger_map().find(logger_name) != fc::get_logger_map().end())
        logger = fc::get_logger_map()[logger_name];
}

首先打开监听的端口号，设置相关协议，随后绑定端口号，然后开始监听网络中的信息，同时让本端点连接到网络中的种子节点，以此连接EOS中的p2p网络。这里面最重要的2个函数是 my->start_listen_loop()以及my->start_monitors()函数,my->start_listen_loop，通过函数名称可以断定主要用来不断地从网络中监听网络中发送的信息，my->start_monitors()应该是进行监听，但是到底监听什么，我们还不得而知。

这里暂时先不对net_plugin_impl进行具体解析，因为这并不影响我们对这两个函数的分析，另外一方面，net_plugin_impl的介绍对于这两个函数的分析，帮助不大。
因此，我们直接进入start_listen_loop函数一探究竟。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
/// 该函数循环监听信息
void net_plugin_impl::start_listen_loop() {
    auto socket = std::make_shared<tcp::socket>( std::ref( app().get_io_service() ) );
    acceptor->async_accept( *socket, [socket,this]( boost::system::error_code ec ) {
        if( !ec ) {
            uint32_t visitors = 0;
            uint32_t from_addr = 0;
            auto paddr = socket->remote_endpoint(ec).address();
            if (ec) {
                fc_elog(logger,"Error getting remote endpoint: ${m}",("m", ec.message()));
            }
            else {
                for (auto &conn : connections) {
                    if(conn->socket->is_open()) {
                    if (conn->peer_addr.empty()) {
                        visitors++;
                        boost::system::error_code ec;
                        if (paddr == conn->socket->remote_endpoint(ec).address()) {
                            from_addr++;
                        }
                    }
                    }
                }
                if (num_clients != visitors) {
                    ilog("checking max client, visitors = ${v} num clients ${n}",("v",visitors)("n",num_clients));
                    num_clients = visitors;
                }
                if( from_addr < max_nodes_per_host && (max_client_count == 0 || num_clients < max_client_count )) {
                    ++num_clients;
                    connection_ptr c = std::make_shared<connection>( socket );
                    connections.insert( c );
                    start_session( c );

                }
                else {
                    if (from_addr >= max_nodes_per_host) {
                    fc_elog(logger, "Number of connections (${n}) from ${ra} exceeds limit",
                            ("n", from_addr+1)("ra",paddr.to_string()));
                    }
                    else {
                    fc_elog(logger, "Error max_client_count ${m} exceeded",
                            ( "m", max_client_count) );
                    }
                    socket->close();
                }
            }
        } else {
            elog( "Error accepting connection: ${m}",( "m", ec.message() ) );
            // For the listed error codes below, recall start_listen_loop()
            switch (ec.value()) {
                case ECONNABORTED:
                case EMFILE:
                case ENFILE:
                case ENOBUFS:
                case ENOMEM:
                case EPROTO:
                    break;
                default:
                    return;
            }
        }
        start_listen_loop();
        });
}

start_listen_loop()函数中最重要的一个函数是，监听到消息之后的start_session()函数，表示监听到了新的链接，于是开始会话。start_session()函数内容如下所示。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
bool net_plugin_impl::start_session(const connection_ptr& con) {
    boost::asio::ip::tcp::no_delay nodelay( true );
    boost::system::error_code ec;
    con->socket->set_option( nodelay, ec );
    if (ec) {
        // 如果接受数据出错， 直接关闭连接，写日志
        elog( "connection failed to ${peer}: ${error}",
            ( "peer", con->peer_name())("error",ec.message()));
        con->connecting = false;
        close(con);
        return false;
    }
    else {
        // 读取数据，已经开启的session+1
        start_read_message( con );
        ++started_sessions;
        return true;
        // for now, we can just use the application main loop.
        //     con->readloop_complete  = bf::async( [=](){ read_loop( con ); } );
        //     con->writeloop_complete = bf::async( [=](){ write_loop con ); } );
    }
}

其中重要的是start_read_message，即回话过程中读取数据，即start_read_message()函数，其具体内容如下所示。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
void net_plugin_impl::start_read_message(const connection_ptr& conn) {
    try {
        if(!conn->socket) {
        return;
        }
        connection_wptr weak_conn = conn;

        std::size_t minimum_read = conn->outstanding_read_bytes ? *conn->outstanding_read_bytes : message_header_size;

        // 默认为false，在plugin_initialized中使用
        if (use_socket_read_watermark) {
        const size_t max_socket_read_watermark = 4096;
        std::size_t socket_read_watermark = std::min<std::size_t>(minimum_read, max_socket_read_watermark);
        boost::asio::socket_base::receive_low_watermark read_watermark_opt(socket_read_watermark);
        conn->socket->set_option(read_watermark_opt);
        }

        auto completion_handler = [minimum_read](boost::system::error_code ec, std::size_t bytes_transferred) -> std::size_t {
        if (ec || bytes_transferred >= minimum_read ) {
            return 0;
        } else {
            return minimum_read - bytes_transferred;
        }
        };

        // 从stream中异步读取固定大小的数据
        /* async_read(AsyncReadStream &S, const MutableBufferSequence& buffers, ReadHandler&& handler, )
        从*conn->socket中读取data，读到buffers中去，buffers的大小告诉系统读取多少
        handler是读取数据完毕之后调用的函数 */
        boost::asio::async_read(*conn->socket,
        conn->pending_message_buffer.get_buffer_sequence_for_boost_async_read(), 
        completion_handler,
        [this,weak_conn]( boost::system::error_code ec, std::size_t bytes_transferred ) {
            auto conn = weak_conn.lock();
            if (!conn) {
                return;
            }

            conn->outstanding_read_bytes.reset();

            try {
                if( !ec ) {
                    if (bytes_transferred > conn->pending_message_buffer.bytes_to_write()) {
                    elog("async_read_some callback: bytes_transfered = ${bt}, buffer.bytes_to_write = ${btw}",
                            ("bt",bytes_transferred)("btw",conn->pending_message_buffer.bytes_to_write()));
                    }
                    EOS_ASSERT(bytes_transferred <= conn->pending_message_buffer.bytes_to_write(), plugin_exception, "");
                    conn->pending_message_buffer.advance_write_ptr(bytes_transferred);
                    while (conn->pending_message_buffer.bytes_to_read() > 0) {
                    uint32_t bytes_in_buffer = conn->pending_message_buffer.bytes_to_read();

                    if (bytes_in_buffer < message_header_size) {
                        conn->outstanding_read_bytes.emplace(message_header_size - bytes_in_buffer);
                        break;
                    } else {
                        uint32_t message_length;
                        auto index = conn->pending_message_buffer.read_index();
                        conn->pending_message_buffer.peek(&message_length, sizeof(message_length), index);
                        if(message_length > def_send_buffer_size*2 || message_length == 0) {
                            boost::system::error_code ec;
                            elog("incoming message length unexpected (${i}), from ${p}",
                                ("i", message_length)("p",boost::lexical_cast<std::string>(conn->socket->remote_endpoint(ec))));
                            close(conn);
                            return;
                        }

                        auto total_message_bytes = message_length + message_header_size;

                        if (bytes_in_buffer >= total_message_bytes) {
                            conn->pending_message_buffer.advance_read_ptr(message_header_size);
                            // 这一部分是网络通信的内容，对于其中细节不甚了解
                            // 接收的数据传递到pending_message_buffer中，
                            // process_next_message中进行处理从pending_message_buffer中处理数据
                            if (!conn->process_next_message(*this, message_length)) {
                                return;
                            }
                        } else {
                            auto outstanding_message_bytes = total_message_bytes - bytes_in_buffer;
                            auto available_buffer_bytes = conn->pending_message_buffer.bytes_to_write();
                            if (outstanding_message_bytes > available_buffer_bytes) {
                                conn->pending_message_buffer.add_space( outstanding_message_bytes - available_buffer_bytes );
                            }

                            conn->outstanding_read_bytes.emplace(outstanding_message_bytes);
                            break;
                        }
                    }
                    }
                    start_read_message(conn);
                } else {
                    auto pname = conn->peer_name();
                    if (ec.value() != boost::asio::error::eof) {
                    elog( "Error reading message from ${p}: ${m}",("p",pname)( "m", ec.message() ) );
                    } else {
                    ilog( "Peer ${p} closed connection",("p",pname) );
                    }
                    close( conn );
                }
            }
            catch(const std::exception &ex) {
                string pname = conn ? conn->peer_name() : "no connection name";
                elog("Exception in handling read data from ${p} ${s}",("p",pname)("s",ex.what()));
                close( conn );
            }
            catch(const fc::exception &ex) {
                string pname = conn ? conn->peer_name() : "no connection name";
                elog("Exception in handling read data ${s}", ("p",pname)("s",ex.to_string()));
                close( conn );
            }
            catch (...) {
                string pname = conn ? conn->peer_name() : "no connection name";
                elog( "Undefined exception hanlding the read data from connection ${p}",( "p",pname));
                close( conn );
            }
        } );
    } catch (...) {
        string pname = conn ? conn->peer_name() : "no connection name";
        elog( "Undefined exception handling reading ${p}",("p",pname) );
        close( conn );
    }
}

抛去其中try catch语句，我们重点看到conn->process_next_message函数，这个函数正如注释所说，从接受到的pending_message_buffer中处理数据，我们继续追踪这个处理函数，如下所示。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 该函数用于数据同步,使用中心消息处理系统处理数据
bool connection::process_next_message(net_plugin_impl& impl, uint32_t message_length) {
    try {
        auto ds = pending_message_buffer.create_datastream();
        net_message msg;
        fc::raw::unpack(ds, msg);  // 将解压的ds信息放入msg中
        msg_handler m(impl, shared_from_this() );
        // 判断msg的类型，msg可以是带签名的区块或者打包后的交易
        // 如果是区块，获取signed_block后放入m中
        // 如果是trx，则获取packed_trx后放入m中
        if( msg.contains<signed_block>() ) {
        m( std::move( msg.get<signed_block>() ) );
        } else if( msg.contains<packed_transaction>() ) {
        m( std::move( msg.get<packed_transaction>() ) );
        } else {
        msg.visit( m );
        }
    } catch(  const fc::exception& e ) {
        edump((e.to_detail_string() ));
        impl.close( shared_from_this() );
        return false;
    }
    return true;
}

其中重要的是msg_handler类型，这里面分别针对消息不同的类型做不同的处理，首先查看针对如果msg中包含了signed_block，则进入处理block函数；如果msg中包含packed_transaction，则进入处理trx的函数。首先查看msg_handler是如何处理signed_block的，处理区块的函数如下。收到区块之后验证区块，验证通过则接受区块，验证失败则拒绝区块。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
// 如果从网络中收到一个区块，执行相应的处理
void net_plugin_impl::handle_message(const connection_ptr& c, const signed_block_ptr& msg) {
   controller &cc = chain_plug->chain();
   block_id_type blk_id = msg->id();
   uint32_t blk_num = msg->block_num();
   fc_dlog(logger, "canceling wait on ${p}", ("p",c->peer_name()));
   c->cancel_wait();
   try {
      if( cc.fetch_block_by_id(blk_id)) {
         // recv_block函数具体含义为止，同步区块数据？不应该是检查在先？
         sync_master->recv_block(c, blk_id, blk_num);
         return;
      }
   } catch( ...) {
      // should this even be caught?
      elog("Caught an unknown exception trying to recall blockID");
   }

   dispatcher->recv_block(c, blk_id, blk_num);  // 进行报告，我已经从连接c收到一个编号blk_id和blk_num的区块？
   fc::microseconds age( fc::time_point::now() - msg->timestamp);
   peer_ilog(c, "received signed_block : #${n} block age in secs = ${age}",
           ("n",blk_num)("age",age.to_seconds()));

   go_away_reason reason = fatal_other;
   try {
      // 在chain_plug中再检查其具体含义，看catch中的内容，应该是使用chain_plug对区块进行检查
      // 如果检查无错误，reason的值应该是no_reason
      chain_plug->accept_block(msg); //, sync_master->is_active(c));
      reason = no_reason;
   } catch( const unlinkable_block_exception &ex) {
      peer_elog(c, "bad signed_block : ${m}", ("m",ex.what()));
      reason = unlinkable;
   } catch( const block_validate_exception &ex) {
      peer_elog(c, "bad signed_block : ${m}", ("m",ex.what()));
      elog( "block_validate_exception accept block #${n} syncing from ${p}",("n",blk_num)("p",c->peer_name()));
      reason = validation;
   } catch( const assert_exception &ex) {
      peer_elog(c, "bad signed_block : ${m}", ("m",ex.what()));
      elog( "unable to accept block on assert exception ${n} from ${p}",("n",ex.to_string())("p",c->peer_name()));
   } catch( const fc::exception &ex) {
      peer_elog(c, "bad signed_block : ${m}", ("m",ex.what()));
      elog( "accept_block threw a non-assert exception ${x} from ${p}",( "x",ex.to_string())("p",c->peer_name()));
      reason = no_reason;
   } catch( ...) {
      peer_elog(c, "bad signed_block : unknown exception");
      elog( "handle sync block caught something else from ${p}",("num",blk_num)("p",c->peer_name()));
   }

   update_block_num ubn(blk_num);
   if( reason == no_reason ) {
      for (const auto &recpt : msg->transactions) {
         auto id = (recpt.trx.which() == 0) ? recpt.trx.get<transaction_id_type>() : recpt.trx.get<packed_transaction>().id();
         auto ltx = local_txns.get<by_id>().find(id);
         if( ltx != local_txns.end()) {
            local_txns.modify( ltx, ubn );
         }
         auto ctx = c->trx_state.get<by_id>().find(id);
         if( ctx != c->trx_state.end()) {
            c->trx_state.modify( ctx, ubn );
         }
      }
      // 这里再次进行recv_block，不明白其含义
      sync_master->recv_block(c, blk_id, blk_num);
   }
   else {
      // 验证区块过程中出现错误，直接拒绝这个区块
      sync_master->rejected_block(c, blk_num);
   }
}

对交易的验证与对区块的处理方式相同，验证交易，如果本地存在此交易，则丢弃，否则开始验证交易，若验证通过则接受交易并广播交易，否则拒绝交易。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
void net_plugin_impl::handle_message(const connection_ptr& c, const packed_transaction_ptr& trx) {
    fc_dlog(logger, "got a packed transaction, cancel wait");
    peer_ilog(c, "received packed_transaction");
    controller& cc = my_impl->chain_plug->chain();
    if( cc.get_read_mode() == eosio::db_read_mode::READ_ONLY ) {
        fc_dlog(logger, "got a txn in read-only mode - dropping");
        return;
    }
    if( sync_master->is_active(c) ) {
        fc_dlog(logger, "got a txn during sync - dropping");
        return;
    }

    auto ptrx = std::make_shared<transaction_metadata>( trx );
    const auto& tid = ptrx->id;

    c->cancel_wait();
    if(local_txns.get<by_id>().find(tid) != local_txns.end()) {
        fc_dlog(logger, "got a duplicate transaction - dropping");
        return;
    }
    dispatcher->recv_transaction(c, tid);
    // 验证交易
    chain_plug->accept_transaction(ptrx, [c, this, ptrx](const static_variant<fc::exception_ptr, transaction_trace_ptr>& result) {
        if (result.contains<fc::exception_ptr>()) {
        peer_dlog(c, "bad packed_transaction : ${m}", ("m",result.get<fc::exception_ptr>()->what()));
        } else {
        auto trace = result.get<transaction_trace_ptr>();
        if (!trace->except) {
            fc_dlog(logger, "chain accepted transaction");
            // 广播交易
            this->dispatcher->bcast_transaction(ptrx);
            return;
        }

        peer_elog(c, "bad packed_transaction : ${m}", ("m",trace->except->what()));
        }
        // 拒绝交易
        dispatcher->rejected_transaction(ptrx->id);
    });
}

至此，net_plugin以及net_plugin_impl插件中的比较重要的函数已经分析完毕，由于其中细节错综复杂，因此在分析过程中抓住主要脉络进行分析，对于其他细节内容并没有深究。