碼雲分佈式之 Brzo 服務器

時間 2019-11-24

原文原文鏈接

前言

碼雲是國內最大的代碼託管平臺。碼雲基於 Gitlab 5.5 開發，通過幾年的開發已經和官方的 Gitlab 有了很大的不一樣。爲了支撐更大的用戶規模，碼雲也在不斷的改進，而本文也主要分享碼雲分佈式 Brzo GIT HTTP 服務器的開發經驗。html

碼雲分佈式概述

自碼雲研發分佈式以來，其分佈式方案也發生了幾回演。在 2014 年，碼雲(當時的 GIT@OSC ) 出現了高速的增加，用戶和項目愈來愈多，在舊的方案中，多個機器經過 NFS 掛載在前端服務器上，用戶對倉庫的讀寫和網頁瀏覽最終都是在前端服務器上被處理，這樣的機制容易帶來嚴重的性能問題，第一是 IO 與計算過於集中，第二是 NFS 帶來了巨大的內網流量。前端

團隊決定使用 Ceph FS，當服務最終遷移到 Ceph 上時，遷移成功一天以後就出現了嚴重的宕機事故，通過研究發現， Git 存儲庫具備海量的小文件，海量小文件一直是分佈式文件系統的難題，而且當時 Ceph 也並未完善，咱們不得不回退到舊的 NFS 方案。碼雲的分佈式由此被提出到議程。ios

碼雲分佈式研發之初，最早被提出的方案是也就是直接使用分佈式 RPC，然而開發團隊並未對 Git 版本控制軟件的基礎特性有深刻的研究，而且團隊也缺少基礎服務開發人員，基於 RPC 的分佈式方案也就只限於個人 demo 之中。nginx

後來有團隊成員提出了使用 NGINX 動態代理，經過解析請求的 URL，將與存儲相關的請求代理到存儲服務器上，而後，存儲服務器上的 Gitlab 對請求進行解析。這樣一來，瀏覽器訪問，以及 Git 的 HTTP 協議 clone 都可以分發到各個存儲服務器上。這個時候剩下的就是如何實現 NGINX 動態代理了，因爲我實現 git 的 svn 接入時有過 NGINX 模塊開發經驗，因此就被安排到 NGINX 動態代理模塊的開發，以及路由模塊的開發。路由策略一開始直接使用Gitlab 的 Magic Path 策略，即取用戶的前兩個字符 (A~Z|a~z|0~9|-_) 等，不一樣的 Magic Path 對應不一樣的內網 IP。後來改成在 MySQL 中存儲用戶的倉庫所在的機器內網 IP，並將 IP 緩存到 Redis 中，獨立存儲。路由模塊自主的向 Redis~MySQL 讀取路由，當從 MySQL 中也找不到存儲機器時，才返回錯誤。對於存儲庫無關的 URL 請求，隨機分發到不一樣的存儲服務器上。前後有 zouqilin 和 lowkey2046 參與開發。c++

NGINX 的動態代理方案中，其模塊開發也經歷了基於 NDK 舊版路由，NDK 新版路由，以及 Upstream 新版路由的演進。目前已經穩定運行，其中不乏企業用戶的私有化部署。git

對於 SSH 協議方案的分佈式支持，最初採用的是 zouqilin 的意見: 使用端口轉發。 gitlab-shell 只須要少許修改就能支持了 SSH 端口轉發。這個時候，惟一沒有分佈式支持的就是 svn 協議，做爲 svn 兼容實現的開發者，我在接受 svn 分佈式任務後，使用 Boost.Asio 開發實現了 svnsrv 動態代理服務器，通過一些波折，svnsrv 服務器也逐漸穩定下來。github

在今年初，我研究 Git 協議後，開發了 git-srv 服務器，這個服務器接受一些參數，而後啓動 git 傳輸命令 (這些命令有 git-upload-pack git-receive-pack git-upload-archive) ，將接收的網絡數據寫到命令的標準輸入，將命令的標準輸出，標準錯誤經過網絡發送給客戶端。基於 git-srv ，實現了 hook 的 git-upload-pack，git-receive-pack，git-upload-archive。在這些命令啓動時，會加載 CratosMini 路由庫，自動鏈接到對應的存儲服務器上的 git-srv， Git 的 Git 協議和 HTTP 協議以及 SSH 協議操做均可以經過這些命令支持分佈式。然而這個方案須要頻繁的啓動進程，並非很是高效。後來便開始開發 Miracle(SSHD)，Mixture，Hover 這些項目。固然 SSH 方案也有新的 SSHD 取代。web

Sshd (ssh://) 基於 libssh 開發，減小了 ssh 鏈接過程的進程建立次數，直接與 git-srv 通訊。而 Github 實際上也是使用 libssh 開發的服務器。 Mixture (git-daemon git://) 基於 Boost.Asio 開發，是 git 協議分佈式動態服務器，直接與 git-srv 通訊。 Hover (Brzo http://) 基於 Boost.Asio 開發，是 HTTP 協議服務器，直接與 git-srv 通訊。 Aton 基於 Crow 開發，是監聽服務器，將機器上的服務信息以及機器信息輸出成 JSON 格式，返回給管理員。算法

這些服務的實現，使得碼雲整個架構變得清晰起來。也可以支撐更大的用戶規模，更好的橫向擴展。shell

Brzo 架構與實現

Brzo 是碼雲分佈式架構的重要組件，它實現了 Git HTTP 協議的分佈式，在 Git 的網絡協議中，HTTPS 流量佔據了很大一部分。在碼雲團隊實現了 SSH， GIT 協議的分佈式，以及存儲機器上的分佈式基礎服務 (git-srv) 後， Git HTTP 分佈式的改造也提上日程。

Git 的 HTTP 協議能夠分爲啞協議和智能協議，啞協議就是經過 GET 獲取到存儲庫中的引用和包文件。這個不須要在服務器上安裝 git 就能夠訪問，目前，包括碼雲在內的代碼託管平臺基本上都不支持啞協議。另外一類協議是智能協議，使用 HTTP 請求，方式描述以下:

Git clone 或者 fetch 操做:

GET /pathto/repo.git/info/refs?service=git-upload-pack
POST /pathto/repo.git/git-upload-pack

Git push 操做:

GET /pathto/repo.git/info/refs/service=git-receive-pack
POST /pathto/repo.git/git-receive-pack

GET 拿到的是服務器的引用列表和支持的操做， POST 在 clone 時推送須要的引用，返回遠程庫打包的 pack 文件，POST 在 push 時，推送本地倉庫與服務器引用差別的打包，返回服務器解包的結果，這些都是動態生成的。瞭解到 GIT 的 HTTP 協議原理，才能更好的實現 GIT 的 HTTP 協議分佈式服務器。

在項目初期，我曾經使用 .Net Core 實現過 Brzo 同等功能的服務器，在 Linux 上正常運行，因爲團隊沒有 C# 使用經驗，項目可能沒法維護，因而 C# 版也就沒有繼續開發了，僅僅是個實驗性項目。

碼雲的基礎服務主要是使用 C++開發，在使用 C++ 的過程當中，雖然 C++ 標準沒有添加網絡庫，可是有許多第三網絡庫能夠被開發者使用，操做系統提供的 API 也能直接被 C++ 項目使用( 好比在 Windows 系統，若是開發 HTTP 服務器，能夠直接使用HTTP.sys 提供的 API，這個是通過內核優化，再使用 RIO 優化，效率一騎絕塵)。

在選擇第三方庫時，卻苦於這些第三方 HTTP 庫並不必定適合服務場景，好比 Microsoft 開源的 cpprestsdk，基於 HTTP.sys ( Linux 是 Boost.Asio )，支持 Linux，還專門實現了 Parallel Patterns Library，使用體驗和 C# await 相似，而 Brzo 須要動態代理到存儲服務器，而且須要針對 Git 的特殊場景進行優化。 Brzo 須要支持 Git 的智能協議，而且與 git-srv 通信，cpprestsdk 在實現這些功能時顯得麻煩而且低效，後來我還使用過 Boost。HTTP 庫，發現並非很合適，鑑於 HTTP 1.1協議比較簡單，我在使用 CURL(WinHTTP) 實現 HttpRequest 時，曾經作過一些簡單解析，因而我乾脆直接基於 Boost.Asio 實現 HTTP 協議服務器 Brzo。 Brzo 被設計爲一個針對 GIT HTTP 協議優化的服務器，須要支持 HTTP 1.1，支持 Chunked Encoding，支持 GZip 解析(能夠不支持 GZip 響應); 因爲 Brzo 可能與 NGINX 一同運行在同一前端機器，支持 Unix domain socket 可以優化反向代理效率，故而 Brzo 添加了 Unix domain socket 支持。

HTTP 協議解析

要獲取 HTTP 協議全文，能夠訪問: RFC7230， RFC7231 ，RFC7232 RFC7233， RFC7234， RFC7235， RFC7236， RFC7237。

除此以外，還能夠閱讀《HTTP 權威指南》。

Git 的 HTTP 協議是 HTTP 協議的真子集，當 GIT 使用啞協議訪問遠程倉庫時，就是純粹的 GET 請求，請求的資源都是靜態的存在在遠程服務器上，面對這種請求， NGINX 開啓 sendfile 就能很好的支持。當 GIT 使用智能協議訪問遠程倉庫時，狀況變得稍微複雜，請求分爲 GET 和 POST，而後頭部的一些字段的屬性須要符合 GIT 的規範，好比 Content-Type。而且請求體也多是動態生成的，這個時候就是 chunked 編碼了。

瞭解了 GIT 的 HTTP 協議，若是要針對 GIT 實現 HTTP 服務器，首先要解析頭部，而後請求體解析須要支持解析 gzip，以及 chunked 編碼，因爲 git 不會同時使用chunked+gzip 編碼，因此這一點能夠忽略。而後就是生成 chunked 編碼。 HTTP 1.1協議要支持 KeepAlive，因此 Brzo 還要支持 KeepAlive。

HTTP KeepAlive 策略

KeepAlive 的實現簡單來講就是打開 socket 後，處理完流程後，服務端 socket 並不主動斷開，而是設置超時，超時時間內，有新的鏈接就繼續處理，重設定時器。若是超時時間事後仍然沒有新的鏈接，就關閉 socket。

若是使用 Session 來描述整個 HTTP 處理流程，處理完成後重置 Session，繼續等待請求便可。如圖:

若是是客戶端以下圖:

圖片來自於 HTTP Keepalive Connections and Web Performance

Chunked 編碼

在網絡的世界裏，有些資源是靜態的，大小可期的，使用 HTTP 請求獲取文件時，Content-Length 就可以拿到大小，從而按照大小將數據所有讀取，然而，還有不少資源是動態生成的，而 GIT 的 HTTP 協議，Push 的 POST 操做的請求體是 send-pack 的標準輸出，這個大小隻能邊讀取邊計算。因此這個時候的請求體就是 chunked 編碼。在服務器上，不管是 fetch 仍是 push 操做，都是 git-upload-pack (git-receive-pack) 的標準輸出，這個時候響應體也是 chunked 編碼。

Chunked 編碼的 BNF 格式描述以下:

Chunked-Body   = *chunk
                        last-chunk
                        trailer
                        CRLF
       chunk          = chunk-size [ chunk-extension ] CRLF
                        chunk-data CRLF
       chunk-size     = 1*HEX
       last-chunk     = 1*("0") [ chunk-extension ] CRLF
       chunk-extension= *( ";" chunk-ext-name [ "=" chunk-ext-val ] )
       chunk-ext-name = token
       chunk-ext-val  = token | quoted-string
       chunk-data     = chunk-size(OCTET)
       trailer        = *(entity-header CRLF)

在使用 Boost.Asio 實現 HTTP 協議時，遇到 Chunked 編碼的第一選擇是使用 boost::asio::streambuf 配合 boost::asio::async_read_until 先讀取 chunk-size 而後讀取 chunk-data，cpprestsdk 正是使用 streambuf 解析 chunked-encoding， async_read_until 先讀取必定長度的數據，若是存在 CRLF 就返回，不存在就繼續度， async_read_until 內部使用 boost::regex 實現。出於內存分配和讀取效率上的考量， Brzo 使用固定長度緩衝區，而且封裝了一個 chunked 解析狀態機:

static const int8_t unhex[256] = {
    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 0,  1,  2,  3,  4,  5,  6,  7,  8,
    9,  -1, -1, -1, -1, -1, -1, -1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1,
    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
    -1, -1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1};
class ChunkedsizeImpl {
public:
  enum State {
    StatusClear, ////
    RequireInput,
    ChunkedLengthOK
  };
  void Reset() {
    state_ = StatusClear;
    offset_ = 0;
    chklen_ = 0;
  }
  int ChunkedsizeEx(const char *data, size_t datalen){
        switch (state_) {
  case StatusClear:
    break;
  case RequireInput:
    offset_ = 0;
    break;
  case ChunkedLengthOK:
    offset_ = 0;
    chklen_ = 0;
    break;
  }
  state_ = RequireInput;
  for (size_t i = 0; i < datalen; i++) {
    uint8_t c = static_cast<uint8_t>(data[i]);
    switch (c) {
    case '\r':
      break;
    case '\n': {
      offset_ = offset_ + i + 1;
      state_ = ChunkedLengthOK;
      return 1;
    }
    default:
      int8_t c2 = unhex[c];
      if (c2 == -1) {
        return -1;
      }
      chklen_ *= 16;
      chklen_ += c2;
      break;
    }
  }
  return 0;
  }
  std::size_t Chklen() const { return chklen_; }
  std::size_t Offset() const { return offset_; }

private:
  State state_{StatusClear};
  std::size_t offset_{0};
  std::size_t chklen_{0}; //// FM
};

ChunkedsizeEx 負責解析，根據返回值的不一樣，Session 將決定寫 chunk 到後端仍是繼續讀取。這樣雖然使編碼變得複雜，當減小內存分配，下降拷貝。這對於長期運行的服務器來講，是很是重要的。

生成 chunked 編碼時，使用 async_read_some 讀取必定大小的數據，準備的緩衝區預留前8個字節，後2字節，共計10字節，讀取到數據後，將數據長度按照 chunk 規定格式寫入前8個字節，8字節與緩衝區大小有關，尾部寫入 "\r\n"。與存儲服務器鏈接斷開時(或者進程關閉輸出)，寫入 chunk-end 即 "0\r\n\r\n"

這種 chunk 策略，可以減小內存拷貝，下降服務器資源佔用。Brzo 的 clone 速度和 GIT 協議服務器 Mixture (git-daemon) 接近。

GZip 編碼

Git 客戶端在 fetch 的 POST 請求時，發送給客戶端的須要的引用列表通常使用 gzip 編碼，此時 Content-Encoding 對應的編碼是 gzip。

gzip 使用的是 Deflate 算法，要解析 gzip，一般的辦法是使用 zlib 來實現。在我開發 Exile 時，就使用過 zlib 解析 gzip GZipStream.cpp 固然， GZipStream.cpp 解析的是已知 gzip 編碼大小，一次性解析完畢就行，Brzo 解析則是讀取多少解析多少，因此這裏使用了相似與 ChunkedsizeImpl 的策略。

class GZipExchange {
public:
  GZipExchange();
  ~GZipExchange();
  bool Initialize(char *refbuf, size_t bufsize); ////
  bool IsEmpty() const { return stream_.avail_out != 0; }
  bool AvailableJoin(const char *buf, size_t len);
  ssize_t Decompress();

private:
  z_stream stream_;
  uint8_t *out_;
  std::size_t chunksize_;
  bool initialized_{false};
};

其中 Initialize 將緩衝區綁定到 GitZipExchange，而後 AvailableJoin 就是待解析的 gzip 緩存區， Decompress 就是不斷解析，直至 IsEmpty 爲真。在 Session 中， GitZipExchange 變量使用 shared_ptr 包裝， KeepAlive 重置時， reset 便可。

Unix domain socket 支持

Brzo 核心版運行在 Linux 上，用戶在使用 NGINX， Apache 之類的 HTTP 服務器做爲負載均衡服務器時，能夠經過 Unix domain socket 與 Brzo 通訊， Brzo 基於 Boost.Asio 開發，在實現對 Unix domain socket 的支持時也就考慮到使用 Boost.Asio 的方案。

Boost.Asio 支持 Unix domain socket 可使用: boost::asio::local::stream_protocol::socket

而普通的 TCP 使用 boost::asio::ip::tcp::socket，兩個類都擁有相同的讀寫函數，能夠直接使用模板包裝一下，就能夠支持兩種 socket 了。更多的細節能夠查看開源版本源碼。

Brzo 開源版本

Brzo 開源版本移除了驗證和分佈式功能，而 ProcessAsync 取代了後端的 socket。

進程的包裝

ProcessAsync 就是進程的包裝，將輸入輸出與對於平臺的流綁定到一塊兒。而後實現 async_read_some async_write_some 這樣的函數。

#ifndef PROCESS_HPP
#define PROCESS_HPP
#include <boost/asio.hpp>
#ifdef _WIN32
#include <boost/asio/windows/stream_handle.hpp>
typedef boost::asio::windows::stream_handle stdiostream;
typedef DWORD ProcessId;
#else
#include <boost/asio/posix/stream_descriptor.hpp>
typedef boost::asio::posix::stream_descriptor stdiostream;
typedef pid_t ProcessId;
#endif

class ProcessAsync {
public:
  ProcessAsync(boost::asio::io_service &ios) : input_(ios), output_(ios) {}
  ~ProcessAsync() { ProcessClean(); }
  int Execute(int Argc, char **Argv);
  void ProcessClean();
  boost::asio::io_service &get_io_service() { return input_。get_io_service(); }
  /// Write
  template <typename ConstBufferSequence, typename WriteHandler>
  void async_write_some(const ConstBufferSequence &buffers,
                        WriteHandler &&handler) {
    input_.async_write_some(buffers, handler);
  }
  //// Read
  template <typename MutableBufferSequence, typename ReadHandler>
  void async_read_some(const MutableBufferSequence &buffers,
                       ReadHandler &&handler) {
    output_.async_read_some(buffers, handler);
  }
  ////
  template <typename ConstBufferSequence, typename WriteHandler>
  void async_write(const ConstBufferSequence &buffers, WriteHandler &&handler) {
    boost::asio::async_write(input_, buffers, handler);
  }
  /////
  template <typename MutableBufferSequence, typename ReadHandler>
  void async_read(const MutableBufferSequence &buffers, ReadHandler &&handler) {
    boost::asio::async_read(output_, buffers, handler);
  }
  void cancel() {
    input_.cancel();
    output_.cancel();
  }
  void cancel(boost::system::error_code &ec) {
    input_.cancel(ec);
    output_.cancel(ec);
  }

private:
  stdiostream input_;
  stdiostream output_;
  ProcessId id_{0};
};

#endif

對於不一樣平臺，Execute 函數是重中之重，啓動進程，修改輸入輸出等等。

Windows 命名管道

在 Windows 中，匿名管道不支持端口完成，而對於 boost stream_handle 而言，不支持端口完成意味着不能異步讀寫，因此咱們須要改造新的管道，實際上匿名管道是命名管道的一種特殊實現，一樣的，咱們也可使用命名管道實現支持端口完成的等價匿名管道。

BOOL WINAPI MzCreatePipeEx(OUT LPHANDLE lpReadPipe, OUT LPHANDLE lpWritePipe,
                           IN LPSECURITY_ATTRIBUTES lpPipeAttributes) {
  HANDLE ReadPipeHandle, WritePipeHandle;
  DWORD dwError;
  WCHAR PipeNameBuffer[MAX_PATH];

  //
  // Only one valid OpenMode flag - FILE_FLAG_OVERLAPPED
  //

  auto PipeId = InterlockedIncrement(&ProcessPipeId_);
  StringCchPrintfW(PipeNameBuffer, MAX_PATH, L"\\\\.\\Pipe\\Brzo.%08x.%08x",
                   GetCurrentProcessId(), PipeId);

  ReadPipeHandle = CreateNamedPipeW(
      PipeNameBuffer, PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED,
      PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
      1,     // Number of pipes
      65536, // Out buffer size
      65536, // In buffer size
      0,     //
      lpPipeAttributes);

  if (!ReadPipeHandle) {
    return FALSE;
  }

  WritePipeHandle = CreateFileW(
      PipeNameBuffer, GENERIC_WRITE,
      0, // No sharing
      lpPipeAttributes, OPEN_EXISTING,
      FILE_ATTRIBUTE_NORMAL |
          FILE_FLAG_OVERLAPPED, /// child process input output is wait
      NULL                      // Template file
      );

  if (INVALID_HANDLE_VALUE == WritePipeHandle) {
    dwError = GetLastError();
    CloseHandle(ReadPipeHandle);
    SetLastError(dwError);
    return FALSE;
  }

  *lpReadPipe = ReadPipeHandle;
  *lpWritePipe = WritePipeHandle;
  return TRUE;
}