1. 程式人生 > >【文彬】RPC的基礎:調研EOS外掛http_plugin

【文彬】RPC的基礎:調研EOS外掛http_plugin

原文連結:https://www.cnblogs.com/Evsward/p/httpPlugin.html

區塊鏈的應用是基於http服務,這種能力在EOS中是依靠http_plugin外掛賦予的。
關鍵字:通訊模式,add_api,http server,https server,unix server,io_service,socket,connection

通訊模式

EOS中,一個外掛的使用要先獲取其例項,例如http_plugin獲取例項的語句是:

auto& _http_plugin = app().get_plugin<http_plugin>();

其他外掛的獲取方式與此相同。目前為止,包括前文介紹到的method、channel、訊號槽、訊號量,跨模組的互動方式可以總結為五種:

  • method,外掛之間的呼叫,一個外掛A將其函式按key註冊到method池中,其他任意數量的外掛B、C、D均可通過key去method池中找到該函式並呼叫。這種通訊模式是一個由呼叫者主動發起的過程。
  • channel,外掛之間的呼叫,一個外掛A按key找到頻道並向頻道publish一個動作,其他任意數量的外掛B、C、D,甚至在不同節點上的外掛B、C、D,只要是按key訂閱了該channel並綁定了他們各自本地的一個notify function,就會被觸發執行。這種通訊模式是基於釋出訂閱模式,或者說是更高階的觀察者模式,是由釋出者的行為交由channel來觸發所有訂閱者繫結的本地通知函式的過程。
  • 訊號槽,外掛與controller的互動過程。controller下啟基於chainbase的狀態資料庫,上承訊號的管理,通過訊號來與外部進行互動,controller會根據鏈的行為emit一個對應的訊號出來,其他外掛如果有處理該訊號的需求會連線connect該訊號並繫結函式實現。有時候一個訊號會被多個外掛所連線,例如accepted_block_header訊號,是承認區塊頭的訊號,會被net_plugin捕捉並處理,同時該訊號也會被chain_plugin所捕捉,觸發廣播。
  • 訊號量,一般是應用程式與作業系統發生的互動,在EOS中,應用程式的例項是application,它與作業系統發生的互動都是通過訊號量來完成,首先宣告一個訊號,然後通過async_wait觸發訊號完成與作業系統的互動。
  • 例項呼叫,對比以上四種鬆散的方式,這種模式是強關聯,正如我們剛剛學習程式設計時喜歡使用new/create而不考慮物件的垃圾處理以及例項管理,後來會採用解耦的鬆散的統一例項管理框架,或者採用單例而不是每次都要new/create。但這種方式並不是完全不被推薦的,當例項的某個成員直接被需要時,可以直接通過該方式獲取到,而不是通過以上四種方式來使用。

目前總結出來的五種跨模組互動方式,前四種更注重通訊,最後一種更注重其他模組的內容。更注重通訊的前四種是基於同一底層通訊機制(socket),但適用於不同場景的設計實現。

add_api函式

從chain_api_plugin過來,http_plugin的使用方式是:

_http_plugin.add_api({
      CHAIN_RO_CALL(get_info, 200l),
      ...
   });

那麼,就從add_api入手研究http_plugin。add_api函式宣告在http_plugin標頭檔案中,說明該函式的內容很少或很具備通用性。

void add_api(const api_description& api) {
   for (const auto& call : api)
      add_handler(call.first, call.second);
}

從前面的呼叫程式碼可以看出,add_api函式的引數是一個物件集合,它們總體是一個api_description型別的常量引用。

using api_description = std::map<string, url_handler>;

api_description根據原始碼可知是一個map,key為string型別的url路徑地址,值為url_handler是具體實現API功能的處理函式。在add_api的呼叫部分,巨集CHAIN_RO_CALL呼叫了另一個巨集CALL,CALL組裝了map的這兩個數:

#define CALL(api_name, api_handle, api_namespace, call_name) \
{std::string("/v1/" #api_name "/" #call_name), \
   [api_handle](string, string body, url_response_callback cb) mutable { \
          try { \
             if (body.empty()) body = "{}"; \
             auto result = api_handle.call_name(fc::json::from_string(body).as<api_namespace::call_name ## _params>()); \
             cb(200, fc::json::to_string(result)); \
          } catch (...) { \
             http_plugin::handle_exception(#api_name, #call_name, body, cb); \
          } \
       }}

CALL巨集體包含兩個資料,以逗號隔開,前面部分為url路徑地址,後面部分為api_handler,此處實際上是一個匿名內部函式。回到add_api函式的宣告,遍歷整個api,逐一執行add_handler為url和api處理函式新增相互繫結的關係。

add_handler函式

直接進入函式實現的程式碼:

void http_plugin::add_handler(const string& url, const url_handler& handler) {
  ilog( "add api url: ${c}", ("c",url) ); // 輸出日誌
  app().get_io_service().post([=](){
    my->url_handlers.insert(std::make_pair(url,handler));
  });
}

app()前文講到了,是用來獲取application例項的,其包含一個public許可權的成員函式get_io_service:

boost::asio::io_service& get_io_service() { return *io_serv; }

返回的是基於boost::asio::io_service庫的共享指標型別,application的私有成員io_serv的指標。

io_service是asio框架中的排程器,用來排程非同步事件,application例項要儲存一個io_service物件,用於儲存當前例項的所有待排程的非同步事件。

io_service的兩個重要方法:

  • post,用於釋出一個非同步事件,依賴asio庫進行自動排程,不需要顯式呼叫函式。
  • run,顯式呼叫,同步執行回撥函式。

當appbase.exec()執行時,io_service會同步啟動,如果一個外掛需要IO或其他非同步操作,可以通過以下方式進行分發:

app().get_io_service().post( lambda )

那麼,這種分發方式,除了在http_plugin的add_handler函式中使用到,EOSIO/eos中在bnet_plugin外掛中有大量使用到,緣於bnet_plugin對非同步事件釋出的需求。回到add_handler函式,post後面跟隨的是lambda表示式,[=]代表捕獲所有以值訪問的區域性名字。lambda體是將url和handler作為二元組插入到http_plugin_impl物件的唯一指標my的共有成員url_handlers集合中,資料型別與上面的api_description一致。

url_handlers集合

url_handlers集合的資料來源是其他外掛通過add_api函式傳入組裝好的url和handler的物件。該集合作為api的非同步處理器集合,在http_plugin中消費該集合資料的是handle_http_request函式。該函式處理外部請求,根據請求url在url_handlers集合中查詢資料,找到handler以後,傳入外部引數資料並執行handler對應的處理函式。

handle_http_request函式

/**
 * 處理一個http請求(http_plugin)
 * @tparam T socket type
 * @param con 連線物件
 */
template<class T>
void handle_http_request(typename websocketpp::server<T>::connection_ptr con) {
    try {
       auto& req = con->get_request(); // 獲得請求物件req。
       if(!allow_host<T>(req, con))// 檢查host地址是否有效
          return;
       // 根據config.ini中http_plugin相關的連線配置項進行設定。
       if( !access_control_allow_origin.empty()) {
          con->append_header( "Access-Control-Allow-Origin", access_control_allow_origin );
       }
       if( !access_control_allow_headers.empty()) {
          con->append_header( "Access-Control-Allow-Headers", access_control_allow_headers );
       }
       if( !access_control_max_age.empty()) {
          con->append_header( "Access-Control-Max-Age", access_control_max_age );
       }
       if( access_control_allow_credentials ) {
          con->append_header( "Access-Control-Allow-Credentials", "true" );
       }
       if(req.get_method() == "OPTIONS") { // HTTP method包含:`GET` `HEAD` `POST` `OPTIONS` `PUT` `DELETE` `TRACE` `CONNECT`
          con->set_status(websocketpp::http::status_code::ok);
          return;// OPTIONS不能快取,未能獲取到請求的資源。
       }
    
       con->append_header( "Content-type", "application/json" );// 增加請求頭。
       auto body = con->get_request_body(); // 獲得請求體(請求引數)
       auto resource = con->get_uri()->get_resource(); // 獲得請求的路徑(url)
       auto handler_itr = url_handlers.find( resource ); // 在url_handlers集合中找到對應的handler
       if( handler_itr != url_handlers.end()) {
          con->defer_http_response();// 延時響應
          // 呼叫handler,傳入引數、url,回撥函式是lambda表示式,用於將接收到的結果code和響應body賦值給連線。
          handler_itr->second( resource, body, [con]( auto code, auto&& body ) {
             con->set_body( std::move( body )); // 接收到的響應body賦值給連線。
             con->set_status( websocketpp::http::status_code::value( code )); // 接收到的code賦值給連線。
             con->send_http_response();// 傳送http響應
          } );
       } else {
          dlog( "404 - not found: ${ep}", ("ep", resource)); // 未在url_handlers集合中找到
          // 針對失敗的情況,設定http的響應物件資料。
          error_results results{websocketpp::http::status_code::not_found,
                                "Not Found", error_results::error_info(fc::exception( FC_LOG_MESSAGE( error, "Unknown Endpoint" )), verbose_http_errors )};
          con->set_body( fc::json::to_string( results ));
          con->set_status( websocketpp::http::status_code::not_found );
       }
    } catch( ... ) {
       handle_exception<T>( con );
    }
}

下面來看該函式handle_http_request的使用位置。有兩處,均在http_plugin內部:

  • create_server_for_endpoint函式,為websocket物件ws設定http處理函式,是一個lambda表示式,lambda體為handle_http_request函式的呼叫,傳入連線物件con,由hdl轉換而來。另外,create_server_for_endpoint函式在http_plugin::plugin_startup中也有兩處呼叫。
  • http_plugin::plugin_startup,外掛的啟動階段,下面將分析該外掛的生命週期。

http_plugin的生命週期

正如研究其他的外掛一樣,學習路線離不開外掛的生命週期。

外掛一般都是在程式入口(例如nodeos,keosd)進行生命週期的控制的,一般不做區分,由於外掛有共同基類,程式入口做統一控制。

下面依次介紹http_plugin的生命週期。

http_plugin::set_defaults

僅屬於http_plugin外掛的生命週期。設定預設值,預設值僅包含三項:

struct http_plugin_defaults {
  // 如果不為空,該項的值將在被監聽的地址生效。作為不同配置項的字首。
  string address_config_prefix;
  // 如果為空,unix socket支援將被完全禁用。如果不為空,值為data目錄的相對路徑,作為預設路徑啟用unix socket支援。
  string default_unix_socket_path;
  // 如果不是0,HTTP將被啟用於預設給出的埠號。如果是0,HTTP將不被預設啟用。
  uint16_t default_http_port{0};
};

nodeos的set_defaults語句為:

http_plugin::set_defaults({
    .address_config_prefix = "",
    .default_unix_socket_path = "",
    .default_http_port = 8888
});

keosd的set_defaults語句為:

http_plugin::set_defaults({
    .address_config_prefix = "",
    // key_store_executable_name = "keosd";
    .default_unix_socket_path = keosd::config::key_store_executable_name + ".sock", // 預設unix socket路徑為keosd.sock
    .default_http_port = 0
});

http_plugin::set_program_options

設定http_plugin外掛的引數,構建屬於http_plugin的配置選項,將與其他外掛的配置共同組成配置檔案config.ini,在此基礎上新增–help等引數構建程式(例如nodeos)的CLI命令列引數。同時設定引數被設定以後的處理方案。

/**
 * 生命週期 http_plugin::set_program_options
 * @param cfg 命令列和配置檔案的手動配置項的並集,交集以命令列配置為準的配置物件。
 */
void http_plugin::set_program_options(options_description&, options_description& cfg) {
   // 處理預設set_defaults配置項。
  my->mangle_option_names();
  if(current_http_plugin_defaults.default_unix_socket_path.length())// 預設unix socket 路徑
     cfg.add_options()
        (my->unix_socket_path_option_name.c_str(), bpo::value<string>()->default_value(current_http_plugin_defaults.default_unix_socket_path),
         "The filename (relative to data-dir) to create a unix socket for HTTP RPC; set blank to disable.");
  if(current_http_plugin_defaults.default_http_port)// 設定預設http埠
     cfg.add_options()
        (my->http_server_address_option_name.c_str(), bpo::value<string>()->default_value("127.0.0.1:" + std::to_string(current_http_plugin_defaults.default_http_port)),
         "The local IP and port to listen for incoming http connections; set blank to disable.");
  else
     cfg.add_options()
        (my->http_server_address_option_name.c_str(), bpo::value<string>(),
         "The local IP and port to listen for incoming http connections; leave blank to disable.");// 埠配置為空的話禁用http
  // 根據手動配置項來設定
  cfg.add_options()
        (my->https_server_address_option_name.c_str(), bpo::value<string>(),
         "The local IP and port to listen for incoming https connections; leave blank to disable.")// 埠配置為空的話禁用http
        ("https-certificate-chain-file", bpo::value<string>(),// https的配置,證書鏈檔案
         "Filename with the certificate chain to present on https connections. PEM format. Required for https.")
        ("https-private-key-file", bpo::value<string>(),// https的配置,私鑰檔案
         "Filename with https private key in PEM format. Required for https")
        ("access-control-allow-origin", bpo::value<string>()->notifier([this](const string& v) {// 跨域問題,控制訪問源
            my->access_control_allow_origin = v;
            ilog("configured http with Access-Control-Allow-Origin: ${o}", ("o", my->access_control_allow_origin));
         }),
         "Specify the Access-Control-Allow-Origin to be returned on each request.")
        ("access-control-allow-headers", bpo::value<string>()->notifier([this](const string& v) {// 控制允許訪問的http頭
            my->access_control_allow_headers = v;
            ilog("configured http with Access-Control-Allow-Headers : ${o}", ("o", my->access_control_allow_headers));
         }),
         "Specify the Access-Control-Allow-Headers to be returned on each request.")
        ("access-control-max-age", bpo::value<string>()->notifier([this](const string& v) {// 控制訪問的最大快取age
            my->access_control_max_age = v;
            ilog("configured http with Access-Control-Max-Age : ${o}", ("o", my->access_control_max_age));
         }),
         "Specify the Access-Control-Max-Age to be returned on each request.")
        ("access-control-allow-credentials",
         bpo::bool_switch()->notifier([this](bool v) {
            my->access_control_allow_credentials = v;
            if (v) ilog("configured http with Access-Control-Allow-Credentials: true");
         })->default_value(false), // 控制訪問允許的證書
         "Specify if Access-Control-Allow-Credentials: true should be returned on each request.")
         // 最大請求體的大小,預設為1MB。
        ("max-body-size", bpo::value<uint32_t>()->default_value(1024*1024), "The maximum body size in bytes allowed for incoming RPC requests")
        // 列印http詳細的錯誤資訊到日誌,預設為false,不列印。
        ("verbose-http-errors", bpo::bool_switch()->default_value(false), "Append the error log to HTTP responses")
        // 校驗host,如果設定為false,任意host均為有效。預設為true,要校驗host。
        ("http-validate-host", boost::program_options::value<bool>()->default_value(true), "If set to false, then any incoming \"Host\" header is considered valid")
        // 別名。另外可接受的host頭
        ("http-alias", bpo::value<std::vector<string>>()->composing(), "Additionaly acceptable values for the \"Host\" header of incoming HTTP requests, can be specified multiple times.  Includes http/s_server_address by default.");
}

http_plugin::plugin_initialize

外掛初始化的操作。讀取配置並做出處理。

實際上,在set_option_program階段也做了對配置值的讀取及轉儲處理。原因是一些預設引數,即使用者不經常配置的選項,就不需要讀取使用者配置的選項,可以在set_option_program階段做出處理,而那些需要使用者來配置的選項則需要在初始化階段讀入並處理。

初始化階段讀入的配置項包含:

  • validate_host,是否校驗host,bool型別的值。

  • valid_hosts,新增alias別名作為有效host。

  • listen_endpoint,根據在set_option_program階段賦值的my成員http_server_address_option_name,重組處理得到監聽點,同時新增至valid_hosts。

  • unix_endpoint,同樣根據my成員unix_socket_path_option_name處理,得到絕對路徑賦值給unix_endpoint。

  • 對set_option_program階段賦值的my成員https_server_address_option_name的值的處理,https的兩個配置的處理,最終重組處理,分別賦值給my成員https_listen_endpoint,https_cert_chain,https_key,以及valid_hosts。

  • max_body_size,直接賦值。

    當然在初始化階段仍舊可以配置set_option_program階段已做出處理的配置項,以使用者配置為準。

http_plugin::plugin_startup

在外掛中,啟動階段都是非常重要的生命週期。它往往程式碼很簡單甚至簡略,但功能性很強。下面來看http_plugin的啟動階段的內容,g共分為三部分:

  • listen_endpoint,本地節點的http監聽路徑,例如127.0.0.1:8888。
  • unix_endpoint,如果為空,unix socket支援將被完全禁用。如果不為空,值為data目錄的相對路徑,作為預設路徑啟用unix socket支援。
  • https_listen_endpoint,https版本的本地節點http監聽路徑,一般不設定,對應的是配置中的https_server_address選項。

對於以上三種情況,啟動階段分別做了三種對應的處理,首先來看最標準最常見的情況,就是基於http的本地監聽路徑listen_endpoint:

if(my->listen_endpoint) {
    try {
        my->create_server_for_endpoint(*my->listen_endpoint, my->server); // 建立http服務(上面介紹到的函式)。內部呼叫了http請求處理函式。
        ilog("start listening for http requests");
        my->server.listen(*my->listen_endpoint);// 手動監聽設定端點。使用設定繫結內部接收器。
        my->server.start_accept();// 啟動伺服器的非同步連線,開始監聽:無限迴圈接收器。啟動伺服器連線無限迴圈接收器。監聽後必須呼叫。在底層io_service開始執行之前,此方法不會有任何效果。它可以在io_service已經執行之後被呼叫。有關如何停止此驗收迴圈的說明,請參閱傳輸策略的文件。
    } catch ( const fc::exception& e ){
        elog( "http service failed to start: ${e}", ("e",e.to_detail_string()));
        throw;
    } catch ( const std::exception& e ){
        elog( "http service failed to start: ${e}", ("e",e.what()));
        throw;
    } catch (...) {
        elog("error thrown from http io service");
        throw;
    }
}

主要是啟動http服務的流程,包括客戶端和服務端,endpoint和server_endpoint兩個角色的啟動。下面來看基於unix socket的情況unix_endpoint:

if(my->unix_endpoint) {
    try {
        my->unix_server.clear_access_channels(websocketpp::log::alevel::all);// 清除所有登陸的頻道
        my->unix_server.init_asio(&app().get_io_service());// 初始化io_service物件,io_service就是上面分析過的application的io_service物件,傳入asio初始化函式初始化asio傳輸策略。在使用asio transport之前必須要init asio。
        my->unix_server.set_max_http_body_size(my->max_body_size); // 設定HTTP訊息體大小的最大值,該值決定了如果超過這個值的訊息體將導致連線斷開。
        my->unix_server.listen(*my->unix_endpoint); // 手動設定本地socket監聽路徑。
        my->unix_server.set_http_handler([&](connection_hdl hdl) {// 設定http請求處理函式(注意此處不再通過create_server_for_endpoint函式來呼叫,因為不再需要websocket的包裝)。
           my->handle_http_request<detail::asio_local_with_stub_log>( my->unix_server.get_con_from_hdl(hdl));
        });
        my->unix_server.start_accept();// 同上,啟動server端的無限迴圈接收器。
    } catch ( const fc::exception& e ){
        elog( "unix socket service failed to start: ${e}", ("e",e.to_detail_string()));
        throw;
    } catch ( const std::exception& e ){
        elog( "unix socket service failed to start: ${e}", ("e",e.what()));
        throw;
    } catch (...) {
        elog("error thrown from unix socket io service");
        throw;
    }
}

下面來看基於https的本地監聽路徑https_listen_endpointd的處理:

if(my->https_listen_endpoint) {
    try {
        my->create_server_for_endpoint(*my->https_listen_endpoint, my->https_server); // 同上http的原理,只是引數換為https的值。
        // 設定TLS初始化處理器。當請求一個TLS上下文使用時,將呼叫該TLS初始化處理器。該處理器必須返回一個有效TLS上下文,以支援當前端點能夠初始化TLS連線。
        // connection_hdl,一個連線的唯一標識。它是實現了一個弱引用智慧指標weak_ptr指向連線物件。執行緒安全。通過函式endpoint::get_con_from_hdl()可以轉化為一個完整的共享指標。
        my->https_server.set_tls_init_handler([this](websocketpp::connection_hdl hdl) -> ssl_context_ptr{
           return my->on_tls_init(hdl); 
        });
        ilog("start listening for https requests");
        my->https_server.listen(*my->https_listen_endpoint);// 同上http的原理,監聽地址。
        my->https_server.start_accept();// 同上http的原理,啟動服務。
    } catch ( const fc::exception& e ){
        elog( "https service failed to start: ${e}", ("e",e.to_detail_string()));
        throw;
    } catch ( const std::exception& e ){
        elog( "https service failed to start: ${e}", ("e",e.what()));
        throw;
    } catch (...) {
        elog("error thrown from https io service");
        throw;
    }
}

unix server與server的底層實現是一致的,只是外部的包裹處理不同,https_server的型別再加上這個ssl上下文的型別指標ssl_context_ptr。他們的宣告分別是:

using websocket_server_type = websocketpp::server<detail::asio_with_stub_log<websocketpp::transport::asio::basic_socket::endpoint>>; // http server
using websocket_local_server_type = websocketpp::server<detail::asio_local_with_stub_log>; // unix server
using websocket_server_tls_type =  websocketpp::server<detail::asio_with_stub_log<websocketpp::transport::asio::tls_socket::endpoint>>; // https server
using ssl_context_ptr =  websocketpp::lib::shared_ptr<websocketpp::lib::asio::ssl::context>; // https ssl_context_ptr
HTTPS = HTTP over TLS。TLS的前身是SSL。

從上面的宣告可以看出,http和https最大的不同是,前者是basic_socket,後者是tls_socket,socket型別不同,http是基礎socket,https是包裹了tls的socket。

http_plugin::plugin_shutdown

關閉是外掛的最後一個生命週期,程式碼很少,主要執行的是資源釋放工作。

void http_plugin::plugin_shutdown() {
  if(my->server.is_listening())
     my->server.stop_listening();
  if(my->https_server.is_listening())
     my->https_server.stop_listening();
}

此處沒有unix_server的處理[#6393]。http和https都是socket,需要手動停止監聽,啟動無限迴圈接收器。unix server是通過io_service來非同步處理,底層實現邏輯相同,也啟動了無限迴圈接收器。

總結

本文首先以外部使用http_plugin的方式:add_api函式為研究入口,逐層深入分析。接著從整體上研究了http_plugin的生命週期,進一步加深了對http_plugin的http/https/unix三種server的認識。



相關文章和視訊推薦

【劉傑良】使用RPC介面新建EOS賬戶 - 實戰

圓方圓學院彙集大批區塊鏈名師,打造精品的區塊鏈技術課程。 在各大平臺都長期有優質免費公開課,歡迎報名收看。

公開課地址:https://ke.qq.com/course/345101