码云分布式之 Brzo 服务器

前言

码云是国内最大的代码托管平台。码云基于 Gitlab 5.5 开发，通过几年的开发已经和官方的 Gitlab 有了很大的不一样。 为了支撑更大的用户规模，码云也在不断的改进，而本文也主要分享码云分布式 Brzo GIT HTTP 服务器的开发经验。

码云分布式概述

自码云研发分布式以来，其分布式方案也发生了几回演。在 2014 年，码云(当时的 GIT@OSC ) 出现了高速的增加， 用户和项目愈来愈多，在旧的方案中，多个机器经过 NFS 挂载在前端服务器上，用户对仓库的读写和网页浏览最终都是在前端服务器上被处理， 这样的机制容易带来严重的性能问题，第一是 IO 与计算 过于集中，第二是 NFS 带来了巨大的内网流量。

团队决定使用 Ceph FS，当服务最终迁移到 Ceph 上时，迁移成功一天以后就出现了严重的宕机事故，通过研究发现， Git 存储库具备海量的小文件，海量小文件一直是分布式文件系统的难题，而且当时 Ceph 也并未完善，咱们不得不回退到旧的 NFS 方案。 码云的分布式由此被提出到议程。

码云分布式研发之初，最早被提出的方案是也就是直接使用分布式 RPC， 然而开发团队并未对 Git 版本控制软件的基础特性有深刻的研究， 而且团队也缺少基础服务开发人员，基于 RPC 的分布式方案也就只限于个人 demo 之中。

后来有团队成员提出了使用 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 参与开发。

NGINX 的动态代理方案中，其模块开发也经历了基于 NDK 旧版路由，NDK 新版路由，以及 Upstream 新版路由的演进。 目前已经稳定运行，其中不乏企业用户的私有化部署。

对于 SSH 协议方案的分布式支持，最初采用的是 zouqilin 的意见: 使用端口转发。 gitlab-shell 只须要少许修改就能支持了 SSH 端口转发。 这个时候，惟一没有分布式支持的就是 svn 协议，做为 svn 兼容实现的开发者，我在接受 svn 分布式任务后，使用 Boost.Asio 开发实现了 svnsrv 动态代理服务器，通过一些波折，svnsrv 服务器也逐渐稳定下来。

在今年初，我研究 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 取代。

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 格式，返回给管理员。

这些服务的实现，使得码云整个架构变得清晰起来。也可以支撑更大的用户规模，更好的横向扩展。

Brzo 架构与实现

Brzo 是码云分布式架构的重要组件，它实现了 Git HTTP 协议的分布式，在 Git 的网络协议中，HTTPS 流量占据了很大一部分。 在码云团队实现了 SSH， GIT 协议的分布式，以及存储机器上的分布式基础服务 (git-srv) 后， Git HTTP 分布式的改造也提上日程。

Git 的 HTTP 协议能够分为哑协议和智能协议，哑协议就是经过 GET 获取到存储库中的引用和包文件。这个不须要在服务器上安装 git 就能够访问， 目前，包括码云在内的代码托管平台基本上都不支持哑协议。 另外一类协议是智能协议，使用 HTTP 请求，方式描述以下:

Git clone 或者 fetch 操做:

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

Git push 操做:

1. GET /pathto/repo.git/info/refs/service=git-receive-pack
2. 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;
}

最后

在实现 Brzo 的过程当中，遇到诸多难题，在解决这些难题的过程当中也收获不少经验。关于 Brzo 的开源版，咱们也将适时发布。

备注

1. 此处 NDK 是 ngx_devel_kit， nginx development kit
2. RIO - Registered Input/Output (RIO) API Extensions