EOS生产区块：解析插件producer_plugin

Posted 2021-01-25 evsward

producer_plugin是控制区块生产的关键插件。

关键字：producer_plugin，同步区块的处理，pending区块，生产区块，最后不可逆区块，生产循环，生产安排，水印轮次，计时器，确认数

producer_plugin生命周期

EOS的所有plugin都有共同的基类，因此每个plugin的研究都可以从生命周期入手。

①set_program_options

向config.ini文件中增加属于producer_plugin的配置项。这些配置项如下表所示。

配置项	解释
enable-stale-production	允许区块生产，即使链是陈腐的，即链生产的区块由于迟到未能被采纳进链。
pause-on-startup	当生产暂停时启动这个节点
max-transaction-time	执行已推送事务代码的最长时间限制，过期则判定为无效，默认30毫秒
max-irreversible-block-age	当前节点生产区块所在链的DPOS不可逆区块的时间限制，按秒计算，默认-1无限制。
producer-name	当前节点的生产者ID，可以被多次指定。
private-key	(已被丢弃，使用以下signature-provider替代)
signature-provider	KV元组使用格式为<public-key>=<provider-spec>，等号前为公钥，等号后为KEY:私钥或KEOSD:私钥。前者为以上公钥对应的私钥，后者为keosd的可用url并且相关钱包要被解锁。
keosd-provider-timeout	发送区块到keosd签名的最大时间，按毫秒计算。
greylist-account	灰名单，记录了无法继承CPU/NET虚拟资源的账户列表。
produce-time-offset-us	非最后一个区块产生时间的偏移量，按微秒计算。负值会导致块更早出去，正值会导致块更晚出去。
last-block-time-offset-us	最后一个区块产生时间的偏移量，按微秒计算。负值会导致块更早出去，正值会导致块更晚出去。
incoming-defer-ratio	当两者都被耗尽时，输入交易和递延交易之间的比率。
snapshots-dir	快照目录的位置(绝对路径或data目录相对路径)

②initialize

插件初始化，第一个阶段是通过现有配置项初始化设置插件。现有配置项来自于配置文件config.ini中producer_plugin相关配置项与命令行参数中producer_plugin相关配置项的交集，同样的配置项以命令行为准。现有配置以boost::program_options::variables_map&类型对象options为参数传入初始化函数。配置过程操作的是producer_plugin的私有成员std::shared_ptr<class producer_plugin_impl> my。my指针拥有producer_plugin_impl对象的成员，这些成员都被设计为与传入配置项对应，逐一设置即可。
第二个阶段是4个远程异步调用的声明：

• 前两个通讯模式是订阅一个channel绑定一个执行函数，一旦嗅到该频道被发布则执行绑定的函数。
  • incoming::channels::block，接收区块的频道，该频道将在bnet_plugin的on_message的on函数中被发布，触发producer_plugin当前的订阅函数on_incoming_block，下面详述。
  • incoming::channels::transaction，接收事务的频道，该频道与上面相同，也将在bnet_plugin的on_message的on函数中被发布，触发producer_plugin当前的订阅函数on_incoming_transaction_async，下面详述。
• 后两个通讯模式是注册一个method，供外部程序调用。
  • incoming::methods::block_sync，接收区块的同步方法。该method将在chain_plugin的read_write::push_block函数中被调用，这部分内容在chain_plugin的文章中有专门的分析。实际上执行的是producer_plugin当下注册的方法on_incoming_block，同上。
  • incoming::methods::transaction_async，接收事务的同步方法。该method将在chain_plugin的read_write::push_transaction函数中被调用，实际上执行的是on_incoming_transaction_async，亦同上。

总结一下会发现，在producer_plugin的初始化阶段：

• 有两个处理对象，
  • 接收的区块，针对该处理对象，均执行函数on_incoming_block
  • 接收的事务，针对该处理对象，均执行函数on_incoming_transaction_async
• 有两个通讯模式，
  • channel的方式，对接的是bnet_plugin
  • method的方式，对接的是chain_plugin

③startup

进入插件的启动阶段，首先设置日志，

const fc::string logger_name("producer_plugin");
fc::logger _log;

const fc::string trx_trace_logger_name("transaction_tracing");
fc::logger _trx_trace_log;

分别创建以producer_plugin插件为主的日志对象，以及事务追踪"transaction_tracing"为主的日志对象。除了这两个在插件内部新建的日志，还有程序自身日志，例如nodeos，日志信息将打印在nodeos的输出位置，输出插件启动日志。接下来的工作列举如下：

• 校验chain的db读取模式以及本地生产者集合是否为空，根据不同情况输出对应日志用于提示用户。
  
• 使用【信号槽技术】分别连接信号accepted_block(绑定本地处理函数on_block和信号irreversible_block(绑定本地处理函数on_irreversible_block，这两个信号将在controller中被发射，从而触发当前信号槽，这两个处理函数将在下面详述。

信号槽的方式，对接的都是controller。

之前讨论过多次，由于解耦的模式，信号发射方和信号槽处理方互不认识，因此一个信号被发射，可以拥有多个信号槽处理方。

• 是针对最后不可逆区块的讨论，下面详述。
  
• 如果本地生产者集合不为空时，输出日志在当前为这些生产者启动区块生产工作。如果本地具备生产能力_production_enabled，如果当前链的头区块号为0，则调用new_chain_banner(chain)，该函数下面详述。
  
• 执行定时生产循环函数schedule_production_loop，下面详述。

④shutdown

释放资源，代码不多如下：

void producer_plugin::plugin_shutdown() {
   try {
      my->_timer.cancel(); // 停止倒计时器
   } catch(fc::exception& e) {
      edump((e.to_detail_string())); // 输出错误日志
   }
   // 重置释放连接槽
   my->_accepted_block_connection.reset();
   my->_irreversible_block_connection.reset();
}

插件关闭阶段，取消计时器，后面会展开对计时器basic_deadline_timer的研究，重置(调用析构函数)清除上面startup阶段启动的两个信号槽。

on_incoming_block 函数

/**
 * 处理incoming接收到的区块。
 * @param block 已签名区块
 */
void on_incoming_block(const signed_block_ptr& block) {
   fc_dlog(_log, "received incoming block ${id}", ("id", block->id()));
   // 判断区块时间是否在当前节点的未来7秒之内，如果不是，则证明这个区块还没到处理的时间。
   EOS_ASSERT( block->timestamp < (fc::time_point::now() + fc::seconds(7)), block_from_the_future, "received a block from the future, ignoring it" );
   // 获取链对象。
   chain::controller& chain = app().get_plugin<chain_plugin>().chain();
   /* 如果本地已经存在接收的区块了，则不必处理，直接返回。*/
   auto id = block->id();
   auto existing = chain.fetch_block_by_id( id );
   if( existing ) { return; }
   // 启动多线程验证区块。这个函数在下面有解释
   auto bsf = chain.create_block_state_future( block );
   // 丢弃pending区块
   chain.abort_block();
   // 抛出异常，保证重启定时生产循环
   auto ensure = fc::make_scoped_exit([this](){
       schedule_production_loop();
   });
   // 向本地链推送新区块
   bool except = false;
   try {
      chain.push_block(block);//推送区块
   } catch ( const guard_exception& e ) {
      // 打印详细错误日志，并跳出循环。
      app().get_plugin<chain_plugin>().handle_guard_exception(e);
      return;
   } catch( const fc::exception& e ) {
      elog((e.to_detail_string()));
      except = true;
   } catch ( boost::interprocess::bad_alloc& ) {
      chain_plugin::handle_db_exhaustion();
      return;
   }
   if( except ) {
      // rejected_block频道发布某区块已被拒绝的消息，该频道已在bnet插件被订阅，当消息发布，bnet插件会调用函数on_bad_block处理被拒区块。
      app().get_channel<channels::rejected_block>().publish( block );
      return;
   }
   // 当链的头块状态中时间戳的下一个点大于等于当前时间时，本地则具备生产能力。
   if( chain.head_block_state()->header.timestamp.next().to_time_point() >= fc::time_point::now() ) {
      _production_enabled = true;
   }
   if( fc::time_point::now() - block->timestamp < fc::minutes(5) || (block->block_num() % 1000 == 0) ) {
      //区块时间点已流逝的时间在5分钟之内的情况，或者区块号是整千时。输出日志模板并替换变量的值。
      ilog("Received block ${id}... #${n} @ ${t} signed by ${p} [trxs: ${count}, lib: ${lib}, conf: ${confs}, latency: ${latency} ms]",
              // p是生产者，
              // id是区块id截取中间的8到16位输出，
              // n是区块号，t是区块时间，
              // count是区块中事务的数量，
              // lib是链最后一个不可逆区块号，
              // confs是区块的确认数
              ("p",block->producer)("id",fc::variant(block->id()).as_string().substr(8,16))
                   ("n",block_header::num_from_id(block->id()))("t",block->timestamp)
                   // confirmed，是生产者在签名一个区块时向前确认的区块数量，默认是1,则只确认前一个区块。
                   // latency，潜伏因素的字面含义。值为当前区块时间点已流逝的时间。
                   // count是时间库中的一个特殊函数，返回某个时间按照某个单位来计数时的字面值，可以用做跨单位的运算。
                   // block_timestamp_type类型定义了区块链的时间戳的默认间隔是500ms，一个周期是2000年。
                   ("count",block->transactions.size())("lib",chain.last_irreversible_block_num())("confs", block->confirmed)("latency", (fc::time_point::now() - block->timestamp).count()/1000 ) );
   }
}

关于block_timestamp_type类型的定义，源码如下：

typedef block_timestamp<config::block_interval_ms,config::block_timestamp_epoch> block_timestamp_type; 

...
const static int      block_interval_ms = 500;
const static uint64_t block_timestamp_epoch = 946684800000ll; // epoch is year 2000.

接着进入函数create_block_state_future，

std::future<block_state_ptr> create_block_state_future( const signed_block_ptr& b ) {
  EOS_ASSERT( b, block_validate_exception, "null block" );//不能为空块
  auto id = b->id();
  // 已存在区块，终止并提示
  auto existing = fork_db.get_block( id );
  EOS_ASSERT( !existing, fork_database_exception, "we already know about this block: ${id}", ("id", id) );

  auto prev = fork_db.get_block( b->previous );// 获得前一个区块，不存在则报错。
  EOS_ASSERT( prev, unlinkable_block_exception, "unlinkable block ${id}", ("id", id)("previous", b->previous) );

  return async_thread_pool( [b, prev]() {// 传入具体task到异步线程池。
     const bool skip_validate_signee = false;
     return std::make_shared<block_state>( *prev, move( b ), skip_validate_signee );
  } );
}

异步线程池async_thread_pool。传入task，由当前同步的待验证区块以及前一个区块组成，返回的是block_state对象。

template<typename F>
auto async_thread_pool( F&& f ) {
  auto task = std::make_shared<std::packaged_task<decltype( f() )()>>( std::forward<F>( f ) );
  boost::asio::post( *thread_pool, [task]() { (*task)(); } );// 将任务上传到线程池，通过boost::asio库异步分配线程并行处理。
  return task->get_future();
}

on_incoming_transaction_async 函数

该函数的工作是处理接收到的事务的本地同步，声明如下：

/**
 * 处理接收到的事务的本地同步工作
 * @param trx 接收的事务，是打包状态的
 * @param persist_until_expired 标志位：事务是否在过期前被持久化了，bool类型
 * @param next 回调函数next方法。
 */
void on_incoming_transaction_async(const packed_transaction_ptr& trx, bool persist_until_expired, next_function<transaction_trace_ptr> next) {}

可以分为三个部分，第一部分是校验工作。

如果链不存在pending区块状态，则在pending接收事务结合中增加接收的事务待start_block中处理，并中止函数返回。

接收到的事务要打包在本地的pending区块中，如果不存在pending区块，说明本地节点未开始生产区块，所以要插入到pending事务集合_pending_incoming_transactions中等待start_block来处理。这部分的校验代码如下：

chain::controller& chain = app().get_plugin<chain_plugin>().chain();
    if (!chain.pending_block_state()) {
      _pending_incoming_transactions.emplace_back(trx, persist_until_expired, next);
      return;   
}

第二部分是该函数定义了一个lambda的内部函数send_response，用于异步发送响应，该内部函数源码如下：

auto send_response = [this, &trx, &chain, &next](const fc::static_variant<fc::exception_ptr, transaction_trace_ptr>& response) {
       next(response);
       if (response.contains<fc::exception_ptr>()) {
          // 如果响应中包含异常指针，则发布异常信息以及事务对象到channels::transaction_ack
          _transaction_ack_channel.publish(std::pair<fc::exception_ptr, packed_transaction_ptr>(response.get<fc::exception_ptr>(), trx));
          if (_pending_block_mode == pending_block_mode::producing) {// 如果pending区块的模式为生产中，则打印出对应的debug日志：区块被拒绝。
             fc_dlog(_trx_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is REJECTING tx: ${txid} : ${why} ",
                     ("block_num", chain.head_block_num() + 1)
                             ("prod", chain.pending_block_state()->header.producer)
                             ("txid", trx->id())
                             ("why",response.get<fc::exception_ptr>()->what()));
          } else {// 如果pending区块的模式为投机中，则打印出对应的debug日志：投机行为被拒绝。
             fc_dlog(_trx_trace_log, "[TRX_TRACE] Speculative execution is REJECTING tx: ${txid} : ${why} ",
                     ("txid", trx->id())
                             ("why",response.get<fc::exception_ptr>()->what()));
          }
       } else {// 响应中无异常。发布空异常信息以及事务对象到channels::transaction_ack
          _transaction_ack_channel.publish(std::pair<fc::exception_ptr, packed_transaction_ptr>(nullptr, trx));
          if (_pending_block_mode == pending_block_mode::producing) {// 仍旧区分pending区块状态生产中与投机行为的不同日志输出。
             fc_dlog(_trx_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is ACCEPTING tx: ${txid}",
                     ("block_num", chain.head_block_num() + 1)
                             ("prod", chain.pending_block_state()->header.producer)
                             ("txid", trx->id()));
          } else {
             fc_dlog(_trx_trace_log, "[TRX_TRACE] Speculative execution is ACCEPTING tx: ${txid}",
                     ("txid", trx->id()));
          }
       }
   };

这部分代码很容易拆解，通过publish/subscribe通讯模式，本地发布频道信息，交由频道订阅者异步处理。
频道：channels::transaction_ack
publisher：on_incoming_transaction_async ->send_response
subscriber：net_plugin::plugin_startup
binding function：net_plugin_impl::transaction_ack
dispatcher：rejected_transaction/bcast_transaction

在这个异步通讯过程中，要加入校验代码。函数体被调用时，send_response已经收到了处理后的事务响应，同时捕获了事务源对象，链对象。链对象在当前程序中应该是单例的，不必在此校验。校验响应事务是否存在异常信息，如果存在则将异常信息附属发布到频道消息，如果不存在则附属空异常。

if (response.contains<fc::exception_ptr>()) {
    _transaction_ack_channel.publish(std::pair<fc::exception_ptr, packed_transaction_ptr>(response.get<fc::exception_ptr>(), trx));
} else {
    _transaction_ack_channel.publish(std::pair<fc::exception_ptr, packed_transaction_ptr>(nullptr, trx));
}

注意，在发布完频道消息以后，要给前台输出事务跟踪日志。

producer_plugin的startup启动阶段分析过，该插件包含三种日志，事务跟踪日志就是其中之一。输出日志要判断pending区块性质是否是正在生产，如果是生产中的区块，则打印区块号，生产者以及事务id，如果不是生产中的区块而是投机区块(可能被生产也可能被丢弃)，则只打印事务id。

publish的消息是trx源事务对象，而不是响应对象response。

前两部分完成以后，本地存在pending区块有打包事务的条件，且发送响应的函数也有了，准备工作已经做好了。接下来进入第三部分，正式开始本地打包接收事务的工作。工作开始之前，仍旧要先校验：

• 接收的事务是否过期，通过比较待打包区块时间和接收事务时间确定事务是否过期，如果过期则发送事务已过期的响应信息并终止程序。
• 接收事的务是否已存在，在本地查找该事务如果查到则发送事务已存在的响应信息并终止程序。

这两个校验的源码如下：

auto block_time = chain.pending_block_state()->header.timestamp.to_time_point();//获得待打包区块时间，即链pending区块头的时间戳转换而来。
auto id = trx->id();
if( fc::time_point(trx->expiration()) < block_time ) {//如果事务的过期时间小于区块时间，说明区块开始打包时事务已过期。报错并中止。
  send_response(std::static_pointer_cast<fc::exception>(std::make_shared<expired_tx_exception>(FC_LOG_MESSAGE(error, "expired transaction ${id}", ("id", id)) )));
  return;
}

if( chain.is_known_unexpired_transaction(id) ) {// 如果在链db中找到了该事务，说明已存在，报错并中止。
  send_response(std::static_pointer_cast<fc::exception>(std::make_shared<tx_duplicate>(FC_LOG_MESSAGE(error, "duplicate transaction ${id}", ("id", id)) )));
  return;
}

两个校验工作结束以后，要确定接收事务的code执行截止时间。初始化的值是当前时间加上本地设置的最大事务执行时间。但如果本地设置未限制最大事务执行时间或者pending区块是本地正在生产且区块时间小于截止时间的，事务截止时间改为区块时间。这段代码如下：

auto block_time = chain.pending_block_state()->header.timestamp.to_time_point();//获得待打包区块时间，即链pending区块头的时间戳转换而来。
auto deadline = fc::time_point::now() + fc::milliseconds(_max_transaction_time_ms);// 算出事务的code执行的截止时间。
bool deadline_is_subjective = false; // 主观截止日期标志位，事务截止时间为区块时间
if (_max_transaction_time_ms < 0 || (_pending_block_mode == pending_block_mode::producing && block_time < deadline) ) {
  deadline_is_subjective = true; // 主观截止日期标志位设置为true。
  deadline = block_time;// 截止时间改为区块时间
}

接下来，确认了事务截止时间以后，执行推送接收的事务到区块链。

// 调用chain推送事务，接收结果储存在trace对象
auto trace = chain.push_transaction(std::make_shared<transaction_metadata>(*trx), deadline);

trace对象接收了chain的推送事务的处理结果。如果判断该结果没有异常则证明处理成功，则要先判断标志位persist_until_expired是否为true，如果为true说明该事务在过期前已被成功持久化，需要在本地持久化事务集合对象中插入事务id，用来保证也能应用在未来的投机区块。最后，将trace对象作为响应信息发送出去。源码如下：

if (persist_until_expired) {// 标志位：事务过期前被持久化
    // 存储事务ID，从而保证它也能应用在未来的投机区块(可逆区块)。
    _persistent_transactions.insert(transaction_id_with_expiry{trx->id(), trx->expiration()});
}
send_response(trace);// 将事务推送结果发送响应。

如果trace结果包含异常，则要判断该异常是否是主观异常。如果是的话，采用上面不存在pending区块的处理方式，将事务插入到pending接收事务集合中，等待start_block处理，同时按照pending区块性质输出日志。如果不是主观失败，则直接丢弃事务，发送异常信息作为响应内容。源码如下：

if (trace->except) {// 异常处理
    if (failure_is_subjective(*trace->except, deadline_is_subjective)) {
        // 主观失败，在pending接收事务结合中增加接收的事务待start_block中处理，并中止函数返回。
        _pending_incoming_transactions.emplace_back(trx, persist_until_expired, next);
        // 仍旧区分pending区块状态生产中与投机行为的不同日志输出。
        if (_pending_block_mode == pending_block_mode::producing) {
           fc_dlog(_trx_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} COULD NOT FIT, tx: ${txid} RETRYING ",
                   ("block_num", chain.head_block_num() + 1)
                           ("prod", chain.pending_block_state()->header.producer)
                           ("txid", trx->id()));
        } else {
           fc_dlog(_trx_trace_log, "[TRX_TRACE] Speculative execution COULD NOT FIT tx: ${txid} RETRYING",
                   ("txid", trx->id()));
        }
    } else {
        // 增加异常信息，发送响应。
        auto e_ptr = trace->except->dynamic_copy_exception();
        send_response(e_ptr);
    }
}

on_block 函数

该函数在controller accepted_block信号处理时有过介绍。下面进行详细分析，首先是三个校验：

if( bsp->header.timestamp <= _last_signed_block_time ) return;
if( bsp->header.timestamp <= _start_time ) return;
if( bsp->block_num <= _last_signed_block_num ) return;

• 如果区块时间小于等于最后签名区块时间，则终止退出当前函数。
• 如果区块时间小于等于开始时间(初始化为当前时间)，则终止退出当前函数。
• 如果区块号小于等于最后签名区块号，则退出当前函数。

通过以上三个校验，可以得知新区块要在最后签名区块之后(生产时间要在它之后，区块号也要在它之后)，另外新区块的生产时间不能是比现在早的时间，必须是之后的时间才可能被当下所处理。校验通过以后，新建活跃生产者账户集合active_producers，插入计划出块的生产者。

const auto& active_producer_to_signing_key = bsp->active_schedule.producers;

flat_set<account_name> active_producers;
active_producers.reserve(bsp->active_schedule.producers.size());
for (const auto& p: bsp->active_schedule.producers) {
  active_producers.insert(p.producer_name);
}

接下来处理接收到的由它节点生产的区块。

// 利用set\\_intersection取本地生产者与集合active\\_producers的交集
std::set_intersection( _producers.begin(), _producers.end(),
                      active_producers.begin(), active_producers.end(),
                      // 将结果存入一个迭代器make_function_output_iterator，迭代执行内部函数
                      boost::make_function_output_iterator( [&]( const chain::account_name& producer )
                      // 如果结果为空，说明本地生产者没有出块权利不属于活跃生产者的一份子
{
    if( producer != bsp->header.producer ) { // 如果交集生产者不等于接收区块的生产者，说明是校验别人生产的区块，如果是相等的不必做特殊处理。
        // 在活跃生产者的key中找到匹配的key(本地生产者账户公钥)
       auto itr = std::find_if( active_producer_to_signing_key.begin(), active_producer_to_signing_key.end(),
                                [&](const producer_key& k){ return k.producer_name == producer; } );
       if( itr != active_producer_to_signing_key.end() ) {// 成功找到，否则说明该区块不是合法生产者签名抛弃不处理。
          auto private_key_itr = _signature_providers.find( itr->block_signing_key );
          // 获取本地生产者私钥
          if( private_key_itr != _signature_providers.end() ) {
             auto d = bsp->sig_digest();
             auto sig = private_key_itr->second( d );
             // 更新producer插件本地标志位
             _last_signed_block_time = bsp->header.timestamp;
             _last_signed_block_num  = bsp->block_num;

             // 组装生产确认数据字段，包括区块id，区块摘要，生产者，签名。发射信号confirmed\\_block。但经过搜索，项目中目前没有对该信号设置槽connection
             _self->confirmed_block( { bsp->id, d, producer, sig } );
          }
       }
    }
} ) );

在区块创建之前要为该区块的生产者设置水印用来标示该区块的生产者是谁。水印就是一个kv结构对象，例如 _producer_watermarks[new_producer] = new_block_num;

chain::controller& chain = app().get_plugin<chain_plugin>().chain();
const auto hbn = bsp->block_num;
// 设置新区块头信息，水印信息，包括时间戳
auto new_block_header = bsp->header; 
new_block_header.timestamp = new_block_header.timestamp.next();
new_block_header.previous = bsp->id;
auto new_bs = bsp->generate_next(new_block_header.timestamp);

接下来，对于新安装的生产者，可以设置他们的水印使他们变为活跃生产者。

if (new_bs.maybe_promote_pending() && bsp->active_schedule.version != new_bs.active_schedule.version) {
    flat_set<account_name> new_producers;
    new_producers.reserve(new_bs.active_schedule.producers.size());
    for( const auto& p: new_bs.active_schedule.producers) {
        if (_producers.count(p.producer_name) > 0)
            new_producers.insert(p.producer_name);
    }
    
    for( const auto& p: bsp->active_schedule.producers) {
        new_producers.erase(p.producer_name);
    }
    
    for (const auto& new_producer: new_producers) {
        _producer_watermarks[new_producer] = hbn;// 水印map，本地变量，用于指挥计划出块的生产者。
    }
}

on_irreversible_block 函数

在producer_plugin中，该函数是用来更新不可逆区块时间的，这个时间在系统中由一个时间变量_irreversible_block_time控制。

void on_irreversible_block( const signed_block_ptr& lib ) {
   _irreversible_block_time = lib->timestamp.to_time_point();
}

这个时间变量将用来计算不可逆区块时间的流逝时间，即当前时间减去该时间变量的结果，如果结果为正数且不小说明很久没有出现不可逆区块了，反之则是刚刚出现不可逆区块。

fc::microseconds get_irreversible_block_age() {
   auto now = fc::time_point::now();
   if (now < _irreversible_block_time) {
      return fc::microseconds(0);
   } else {
      return now - _irreversible_block_time;
   }
}

last_irreversible 的讨论

在producer_plugin的启动阶段，包含一段关于最后不可逆区块的代码：

const auto lib_num = chain.last_irreversible_block_num();// 获取当前最后不可逆区块号
const auto lib = chain.fetch_block_by_number(lib_num); // 获取最后不可逆区块
if (lib) { // 如果最后不可逆区块存在
  my->on_irreversible_block(lib); // 执行函数同步更新本地区块的不可逆时间
} else { // 如果最后不可逆区块不存在
  my->_irreversible_block_time = fc::time_point::maximum();// 区块不可逆时间设置为最大值。
}

通常来讲，最后不可逆区块的存在是被用来定位本地事务被打包至某个区块后是否成功上链变为不可逆状态，只需要这个区块号小于最后不可逆区块即可确定。

以上代码段中if-else语句比较容易理解，是根据最后不可逆区块是否存在对本地区块不可逆时间变量_irreversible_block_time的设置，存在则更新为最后不可逆区块的时间，不存在则将其设置为时间最大值。不存在最后不可逆区块意味着链数据完全是孤立的未经任何确认的，区块链的特性也不再存在，因此本地时间变量设置为了时间的最大值。那么令人费解的是上面两行代码，首先获取最后不可逆区块号，接着通过该区块号获得区块。

controller::last_irreversible_block_num

uint32_t controller::last_irreversible_block_num() const {
   return std::max(std::max(my->head->bft_irreversible_blocknum, my->head->dpos_irreversible_blocknum), my->snapshot_head_block);
}

以上获取最后不可逆区块号函数的源码，可以看出是从当前区块头的bft不可逆区块号、dpos不可逆区块号以及快照头块的区块号三者中选择最大的一个作为结果返回。分别来看这三个区块号的含义：

• bft不可逆区块号，在区块头状态结构中的generate_next函数中有初始化的操作，这个函数主要是用来通过一个给定的时间生成一个模板的区块头状态对象，不包含事务Merkle根、action Merkle根以及新生产者字段数据，因为这些组件是派生自链状态的。总之，在代码中查找，发现bft不可逆区块号只有一个初始化为0的赋值动作，原因可能与EOS计划引入bft而目前还没有bft有关系。因此该值为0。
• dpos不可逆区块号，controller初始化为0。仍旧在generate_next函数中找到该字段的初始化值为calc_dpos_last_irreversible()函数的结果。
• 快照的头块号，初始化是0，如果有快照读入的话，就是快照的头区块号。

calc_dpos_last_irreversible函数

该函数用来计算dpos最后不可逆区块。

uint32_t block_header_state::calc_dpos_last_irreversible()const {
  vector<uint32_t> blocknums; blocknums.reserve( producer_to_last_implied_irb.size() );
  for( auto& i : producer_to_last_implied_irb ) {
     blocknums.push_back(i.second);
  }
  if( blocknums.size() == 0 ) return 0;
  std::sort( blocknums.begin(), blocknums.end() );//默认从小到大排序。less<int>()
  return blocknums[ (blocknums.size()-1) / 3 ];// dpos最后不可逆区块的判断条件是必须在池子里面保持有2/3个区块号是大于自己的。
}

fetch_block_by_number

signed_block_ptr controller::fetch_block_by_number( uint32_t block_num )const  { try {
   auto blk_state = my->fork_db.get_block_in_current_chain_by_num( block_num );// 从分叉库中根据块号获取状态区块。
   if( blk_state && blk_state->block ) {//状态区块存在且其block成员也存在
      return blk_state->block; // 返回其block成员对象。
   }

   return my->blog.read_block_by_num(block_num);// 否则的话从block.log日志中获取区块返回。
} FC_CAPTURE_AND_RETHROW( (block_num) ) }

---

重新回到producer_plugin的启动阶段的last_irreversible 的讨论。首先通过函数last_irreversible_block_num从bft和dpos以及快照三个区块号中获取最大的一个，由于目前未引进bft且有快照进入的概率不高，所以暂定该最后不可逆区块号为dpos的那个号。接着用这个区块号通过fetch_block_by_number中查找，先在fork_db中查找，如果没有则在block.log中查找获得区块对象。不过一般fork_db中不应该存在不可逆区块，如果区块变为不可逆状态应该被立即持久化到block.log，并从fork_db中删除。

new_chain_banner(chain)

该函数翻译过来就是新链的条幅，条幅是显示在日志中的，源码如下：

void new_chain_banner(const eosio::chain::controller& db)
{
   std::cerr << "
"
      "*******************************
"
      "*                             *
"
      "*   ------ NEW CHAIN ------   *
"
      "*   -  Welcome to EOSIO!  -   *
"
      "*   -----------------------   *
"
      "*                             *
"
      "*******************************
"
      "
";

   if( db.head_block_state()->header.timestamp.to_time_point() < (fc::time_point::now() - fc::milliseconds(200 * config::block_interval_ms)))
   {
      std::cerr << "Your genesis seems to have an old timestamp
"
         "Please consider using the --genesis-timestamp option to give your genesis a recent timestamp
"
         "
"
         ;
   }
   return;
}

传入一个链对象(controller实例)，输出一个字符图案在日志中，接着校验genesis的时间戳，如果小于当前时间200个间隔周期，则报错重新设置genesis的时间戳配置为一个就近的时间。

schedule_production_loop

这是一个对于producer_plugin非常重要的函数，是出块节点按计划出块的循环函数。在系统多个功能函数中涉及处理恢复继续按计划出块时，多次被调用到。该函数中大量使用到了_timer对象，下面先研究_timer。

basic_deadline_timer 的研究

对producer_plugin_impl类的共有成员_timer的追踪，可以发现它是basic_deadline_timer类的对象。

该函数提供了可等待的计时器功能。basic_deadline_timer类模板提供了执行阻塞(blocking)或异步等待(asynchronous wait)定时器期满的能力。截止日期计时器总是处于两种状态之一:“过期”或“未过期”。如果在过期计时器上调用wait()或async_wait()函数，则等待操作将立即完成。

使用实例：

①阻塞等待(blocking wait)

为计时器设置一个相对时间。

timer.expires_from_now(boost::posix_time::seconds(5));// 从现在开始计时5秒钟。

等待计时器过期。

timer.wait();

②异步等待(asynchronous wait)

首先要创建一个处理器handler。

void handler(const boost::system::error_code& error)
{
    if (!error)
    {
        // Timer expired.
    }
}

构建一个绝对过期时间的计时器。

boost::asio::deadline_timer timer(io_context, boost::posix_time::time_from_string("2005-12-07 23:59:59.000"));

启动一个异步等待。

timer.async_wait(handler);

③改变过期时间

当存在挂起的异步等待时，更改计时器的过期时间会导致这些等待操作被取消。要确保与计时器关联的操作只执行一次，请使用类似的方法：

// boost::asio::basic\\_deadline\\_timer::expires\\_from\\_now() 函数取消任何挂起的异步等待，并返回已取消的异步等待的数量。如果返回0，则太迟了，且等待处理器已经被执行，或者即将被执行。如果它返回1，那么等待程序会被成功取消。
void on_some_event() // 模拟某事件处理函数
{
    if (my_timer.expires_from_now(seconds(5)) > 0) {
      // 取消计时器，启动一个新的异步等待
      my_timer.async_wait(on_timeout);
    } else {
      // 计时器已过期。
    }
}
// 如果一个等待处理程序被取消，传递给它的boost::system::error\\_code包含值boost::asio::error::operation\\_aborted。
void on_timeout(const boost::system::error_code& e) // 超时事件处理函数
{
    if (e != boost::asio::error::operation_aborted) {
      计时器未取消，继续执行操作。
    }
}

回到producer_plugin的shutdown阶段中_timer的使用。

my->_timer.cancel();

进入basic_deadline_timer::cancel()函数：

std::size_t cancel()
{
    boost::system::error_code ec;
    std::size_t s = this->get_service().cancel(this->get_implementation(), ec);
    boost::asio::detail::throw_error(ec, "cancel");
    return s;
}

该函数将取消所有正在等待计时器的异步操作。

回到schedule_production_loop函数。

这一部分是对计时器的设置。首先重置计时器，获得链对象chain以及弱指针producer_plugin_impl实例。执行start_block，并接收结果，根据结果的不同做不同的处理，该结果为一个枚举类型：

enum class start_block_result {
    succeeded, // 成功
    failed, // 失败
    waiting, // 等待
    exhausted // 耗尽，该状态在producer插件中并没有显式使用，而是其他状态处理完毕剩余的情况。
};

• 如果是failed，启动区块的返回值是失败的，那么要输出提醒日志，同时计时器启动50毫秒倒计时，异步等待到期以后再次尝试重新调用自己schedule_production_loop函数。
• 如果是waiting，等待中。判断生产者如果不是空且启用了生产能力，则调用延时计划生产循环schedule_delayed_production_loop函数。

bool production_disabled_by_policy() { // 确定生产能力是否被禁用的方式。有以下三种判断条件，满足其一即可。
 return !_production_enabled || _pause_production || (_max_irreversible_block_age_us.count() >= 0 && get_irreversible_block_age() >= _max_irreversible_block_age_us);
}

延时计划生产循环schedule_delayed_production_loop函数，主要操作对象是wake_up_time，即该延时操作的唤醒时间。一些列校验判断探测出唤醒时间已到达时，就会调用回schedule_production_loop函数。

• 接下来，start_block结果其他的状态情况，即succeeded或者exhausted。当pending区块模式为生产中时，pending区块的模式分为：

enum class pending_block_mode {
   producing, // 本地生产
   speculating // 外部确认有可能确认失败不一定能成为不可逆，因此是投机性的。
};

到目前这个分支下，换句话讲就是启动区块start_block已经成功succeeded了，但是也有可能耗尽exhausted，这两种情况要通过另外一种判断，即是否pending区块处于生产中的状态来做区分。

pending区块模式为生产中producing

• start_block成功succeeded。这部分代码的工作主要是用来保证区块要在截止时间之前被装运上链。先校验一下是否存在pending区块。接着计算截止时间并按照该时间启动计时器：

// 计算截止时间，epoch默认是从1970/1/1，从epoch开始计算到pending区块的时间加上预设的区块生产时间偏移量。
auto deadline = chain.pending_block_time().time_since_epoch().count() + (last_block ? _last_block_time_offset_us : _produce_time_offset_us);
_timer.expires_at( epoch + boost::posix_time::microseconds( deadline ));// 截止时间加上epoch的时间初始量，按此时间启动计时器。

现行西历即格里历，又译国瑞历、额我略历、格列高利历、格里高利历，称西元。地球每天的自转是有些不规则的，而且正在缓慢减速。所以，格林尼治时间已经不再被作为标准时间使用。现在的标准时间──协调世界时（UTC）──由原子钟提供。

• start_block成功exhausted。仍旧要先检查是否存在pending区块。接着计算预期时间expect_time，是penging区块时间减去一个区块间隔时间0.5秒(现在是设置的0.5秒出一个块，在config.hpp中可以查到)

auto expect_time = chain.pending_block_time() - fc::microseconds(config::block_interval_us);

... config.hpp
const static int      block_interval_ms = 500;
const static int      block_interval_us = block_interval_ms*1000;

下面是判断预期时间和现在时间的对比，如果预期时间已过，则将计时器时间调节为0(立即执行出块)。如果预期时间未到，则设置计时器到预期时间，等待计时完成。

if (fc::time_point::now() >= expect_time) { // 预期时间已过
    _timer.expires_from_now( boost::posix_time::microseconds( 0 )); // 将计时器时间调节为0
    fc_dlog(_log, "Scheduling Block Production on Exhausted Block #${num} immediately", ("num", chain.pending_block_state()->block_num));
} else { // 预期时间未到
    _timer.expires_at(epoch + boost::posix_time::microseconds(expect_time.time_since_epoch().count()));
    fc_dlog(_log, "Scheduling Block Production on Exhausted Block #${num} at ${time}", ("num", chain.pending_block_state()->block_num)("time",expect_time)); // 设置计时器到预期时间
}

分别将succeeded以及exhausted状态的_timer设置完毕以后，下面要处理当计时器到时的事件处理，即_timer.async_wait函数。该函数的参数为匿名内部类组成的异步回调函数。

_timer.async_wait([&chain,weak_this,cid=++_timer_corelation_id](const boost::system::error_code& ec) {
    auto self = weak_this.lock(); // 获得锁。
    if (self && ec != boost::asio::error::operation_aborted && cid == self->_timer_corelation_id) { // 满足生产区块的条件：有锁且操作未被终止且计时器关联id匹配。
        // 内部要校验一遍pending区块是否存在。
        auto block_num = chain.pending_block_state() ? chain.pending_block_state()->block_num : 0; // 区块号的设置，若pending区块存在则设置为pending区块号，若不存在，则设置为0。
        auto res = self->maybe_produce_block(); // 调用maybe_produce_block()函数(下面分析)执行区块的生产，返回生产结果。
        fc_dlog(_log, "Producing Block #${num} returned: ${res}", ("num", block_num)("res", res));
    }
});

计时器关联id匹配。_timer_corelation_id的存在源自一个攻击警报：Boost计时器可能处于一个处理程序尚未执行但不可中止的状态，这个状态给外部攻击提供了可能。关联id的设置可以有效防止，处理程序被改变。在处理程序捕获相关性ID设置时，他们必须执行检查匹配全局变量_timer_corelation_id。如果不匹配，则意味着该方法已被调用，处理程序处于应该取消但无法取消的状态。

pending区块模式为投机中speculating

这个状态下，分两种情况处理：

• 如果生产者存在且具备生产能力(有可能是备用节点)时，校验一番以后最终会调用延时计划出块循环schedule_delayed_production_loop。
• 其他情况则只打印日志，说明创建了投机区块。

maybe_produce_block()函数

前面提到，schedule_production_loop函数是出块者生产区块时，调用start_block函数并根据返回结果设置计时器_timer，并处理计时完成的处理程序，而最终只有start_block结果为succeeded以及exhausted状态，计时完成以后同时满足有锁且操作未被终止且计时器关联id匹配。这全部条件的满足，最后调用区块生产执行函数maybe_produce_block。

简单来讲，schedule_production_loop函数就是通过调用start_block设置timer，计时完成执行maybe_produce_block。所以schedule_production_loop函数的核心是处理_timer。

下面分析函数maybe_produce_block：

bool producer_plugin_impl::maybe_produce_block() {
   // 当前作用域退出时回调schedule_production_loop()继续循环处理出块工作。
   auto reschedule = fc::make_scoped_exit([this]{ 
      schedule_production_loop();
   });

   try {
      try {
         produce_block(); // 实际调用函数produce_block生产区块。
         return true; // 返回true，代表区块生产成功，其他异常状态均返回false，代表出块失败。
      } catch ( const guard_exception& e ) { // 处理守卫异常
         app().get_plugin<chain_plugin>().handle_guard_exception(e);
         return false;
      } FC_LOG_AND_DROP();
   } catch ( boost::interprocess::bad_alloc&) { // 处理内部线程内存错误异常
      raise(SIGUSR1);
      return false;
   }
   // 区块生产出错，丢其区块。
   fc_dlog(_log, "Aborting block due to produce_block error");
   chain::controller& chain = app().get_plugin<chain_plugin>().chain();
   chain.abort_block(); // 丢其区块。
   return false;
}

produce_block函数

区块生产函数用于处理区块生产，前面maybe_produce_block函数的主要功能集中在“maybe”，所以实际出块任务仍旧交由produce_block处理。

void producer_plugin_impl::produce_block() {
   // 区块生产必须是pending区块状态为producing，否则输出错误日志：实际上并没有真正生产区块。
   EOS_ASSERT(_pending_block_mode == pending_block_mode::producing, producer_exception, "called produce_block while not actually producing");
   chain::controller& chain = app().get_plugin<chain_plugin>().chain(); // 获取chain实例
   const auto& pbs = chain.pending_block_state(); // 从chain实例获取当前pending区块，如果获取为空，则输出错误日志：不存在pending区块，可能被其他插件毁坏。
   const auto& hbs = chain.head_block_state(); // 从chain实例获取当前头区块
   EOS_ASSERT(pbs, missing_pending_block_state, "pending_block_state does not exist but it should, another plugin may have corrupted it");
   auto signature_provider_itr = _signature_providers.find( pbs->block_signing_key ); // 通过pending区块的区块签名公钥去内存多索引表_signature_providers中差找signature_provider。
   // 如果未查到有效signature_provider，则输出错误日志：正在尝试生产一个区块，是由一个我们不拥有的私钥所签名。
   EOS_ASSERT(signature_provider_itr != _signature_providers.end(), producer_priv_key_not_found, "Attempting to produce a block for which we don‘t have the private key");
   chain.finalize_block(); // 执行chain的区块完成操作，重置资源(调用的为controller的finalize_block函数)。
   chain.sign_block( [&]( const digest_type& d ) { // 调用controller的sign_block函数进行函数签名，参数为一个回调函数。区块签名最终是由block_header_state来做的实际工作。
      auto debug_logger = maybe_make_debug_time_logger();
      return signature_provider_itr->second(d);
   } );
   chain.commit_block(); // 仍旧是执行controller的commit_block函数进行区块提交。
   block_state_ptr new_bs = chain.head_block_state(); 
   _producer_watermarks[new_bs->header.producer] = chain.head_block_num(); // 设置水印

   // 打印生产结果日志。
   ilog("Produced block ${id}... #${n} @ ${t} signed by ${p} [trxs: ${count}, lib: ${lib}, confirmed: ${confs}]",
        ("p",new_bs->header.producer)("id",fc::variant(new_bs->id).as_string().substr(0,16))
        ("n",new_bs->block_num)("t",new_bs->header.timestamp)
        ("count",new_bs->block->transactions.size())("lib",chain.last_irreversible_block_num())("confs", new_bs->header.confirmed));

}

水印的意义重申一下，是用来给计划出块预先设置出块安排的，即安排下一个区块的生产者，管理出块轮次。

produce_block函数属于producer_plugin，然而其中核心区块处理，例如重置资源准备、区块签名、提交区块都时通过chain_plugin调用了controller的相关函数，而controller只是负责管理与数据层的交互，数据层包括block.log以及及与chainbase的db，区块签名的内容是区块头block_header_state来处理。结构拆分如下图所示：

start_block 函数

该函数是producer插件对出块管理的核心函数。该函数通过对时间的控制管理了出块节奏，管理出块轮次。到这里可以得出producer插件的操作对象是pending区块，所以该函数对pending区块是本地生产还是外部同步进来的做了区分处理。这其中涉及到一个区块同步确认的处理，即生产者生产当前区块时要确认多少个区块：

• 如果区块的生产者不是当前节点的，则假设没有确认(丢弃这个块)。
• 如果区块的生产者是当前节点上从未产生过的生产者，那么保守的方法就是假定没有确认，确保不会在crash之后重复签名。(不过此处有个问题是crash的话，是否要保证水印持久化？否则crash会丢失，答案是肯定的)
• 如果区块的生产者是这个节点上的生产者，这个节点是知道它生成的最后一个块的，则安全地设置它：unless
• 如果区块的生产者在该节点的最后水印中的位置较高，则意味着该区块时在一个不同的分叉上。

本函数大约包含三百多行代码，用于处理pending区块不同情况下的校验以及动作，包括对区块中打包事务的校验和处理，最终返回的时start_block_result状态，前面有介绍过。

start_block的代码不在此详细分析，但总结下来可以得出是对pending区块的区块头校验，包括是否是本地生产抑或是外部同步，然后是对pending区块内事务的处理，包括如何重置打包接收的事务。到这部分相当于将一个区块的头部信息构成以及校验工作和区块体的事务打包内容工作完成了。最后返回一个处理状态，如果通过了层层校验以及无异常的顺利处理，则返回启动区块成功的状态，如果是时间超时，耗尽了规定时间则返回exhausted，其他情况则时failed。

总结

本文分析介绍了producer_plugin的重点功能，研究了其大量内部函数。最初的研究路先是分析该插件的生命周期，然后引申到各个未知或以前未仔细研究过的调用的函数细节。其中，涉及到了出块安排水印、pending区块处理、区块生产循环、区块的生产者校验、是否本地或是同步、计时器的相关知识和应用、最后不可逆块的研究、区块生产、区块签名等，另外还涉及到新版本的多线程校验签名区块的内容。研究过程中，也梳理了producer插件与chain插件的交互以及延伸到controller的内部函数的使用。总之，内容较多篇幅较长，整体研究脉络似乎仍旧不算清晰，但也算是自身知识图谱的“大数据”的一部分，量变引发质变。

更多文章请转到醒者呆的博客园。