字节跳动百万级Metrics Agent性能优化的探索与实践

610次阅读  |  发布于10月以前

背景


metricserver2 (以下简称Agent)是与字节内场时序数据库 ByteTSD 配套使用的用户指标打点 Agent,用于在物理机粒度收集用户的指标打点数据,在字节内几乎所有的服务节点上均有部署集成,装机量达到百万以上。此外Agent需要负责打点数据的解析、聚合、压缩、协议转换和发送,属于CPU和Mem密集的服务。两者结合,使得Agent在监控全链路服务成本中占比达到70%以上,对Agent进行性能优化,降本增效是刻不容缓的命题。本文将介绍我们在Agent性能优化上的探索和实践。

基本架构

我们将按照数据接收、数据处理、数据发送三个部分来分析Agent优化的性能热点。

数据接收

Case 1

Agent与用户SDK通信的时候,使用 msgpack 对数据进行序列化。它的数据格式与json类似,但在存储时对数字、多字节字符、数组等都做了优化,减少了无用的字符,下图是其与json的简单对比:

Agent在获得数据后,需要通过msgpack.unpack进行反序列化,然后把数据重新组织成 std::vector。这个过程中,有两步复制的操作,分别是:从上游数据反序列为 msgpack::object 和 msgpack::object 转换 std::vector。

{ // Process Function
    msgpack::unpacked msg;
    msgpack::unpack(&msg, buffer.data(), buffer.size());
    msgpack::object obj = msg.get();

    std::vector<std::vector<std::string>> vecs;
    if (obj.via.array.ptr[0].type == 5) {
        std::vector<std::string> vec;
        obj.convert(&vec);
        vecs.push_back(vec);
    } else if (obj.via.array.ptr[0].type == 6) {
        obj.convert(&vecs);
    } else {
        ++fail_count;
        return result;
    }

    // Some more process steps
}

但实际上,整个数据的处理都在处理函数中。这意味着传过来的数据在整个处理周期都是存在的,因此这两步复制可以视为额外的开销。

msgpack协议在对数据进行反序列化解析的时候,其内存管理的基本逻辑如下:

为了避免复制 string,bin 这些类型的数据,msgpack 支持在解析的时候传入一个函数,用来决定这些类型的数据是否需要进行复制:

因此在第二步,对 msgpack::object 进行转换的时候,我们不再转换为 string,而是使用 string_view,可以优化掉 string 的复制和内存分配等:

// Define string_view convert struct.
template <>
struct msgpack::adaptor::convert<std::string_view> {
    msgpack::object const& operator()(msgpack::object const& o, std::string_view& v) const {
        switch (o.type) {
        case msgpack::type::BIN:
            v = std::string_view(o.via.bin.ptr, o.via.bin.size);
            break;
        case msgpack::type::STR:
            v = std::string_view(o.via.str.ptr, o.via.str.size);
            break;
        default:
            throw msgpack::type_error();
            break;
        }
        return o;
    }
};

static bool string_reference(msgpack::type::object_type type, std::size_t, void*) {
    return type == msgpack::type::STR;
}

{ 
    msgpack::unpacked msg;
    msgpack::unpack(msg, buffer.data(), buffer.size(), string_reference);
    msgpack::object obj = msg.get();

    std::vector<std::vector<std::string_view>> vecs;
    if (obj.via.array.ptr[0].type == msgpack::type::STR) {
        std::vector<std::string_view> vec;
        obj.convert(&vec);
        vecs.push_back(vec);
    } else if (obj.via.array.ptr[0].type == msgpack::type::ARRAY) {
        obj.convert(&vecs);
    } else {
        ++fail_count;
        return result;
    }
}

经过验证可以看到:零拷贝的时候,转换完的所有数据的内存地址都在原来的的 buffer 的内存地址范围内。而使用 string 进行复制的时候,内存地址和 buffer 的内存地址明显不同。

Case 2

Agent在接收端通过系统调用完成数据接收后,会立刻将数据投递到异步的线程池内,进行数据的解析工作,以达到不阻塞接收端的效果。但我们在对线上数据进行分析时发现,用户产生的数据包大小是不固定的,并且存在大量的小包(比如一条打点数据)。这会导致异步线程池内的任务数量较多,平均每个任务的体积较小,线程池需要频繁的从队列获取新的任务,带来了处理性能的下降。

因此我们充分理解了msgpack的协议格式(https://github.com/msgpack/msgpack/blob/master/spec.md)后,在接收端将多个数据小包(一条打点数据)聚合成一个数据大包(多条打点数据),进行一次任务提交,提高了接收端的处理性能,降低了线程切换的开销。

static inline bool tryMerge(std::string& merge_buf, std::string& recv_buf, int msg_size, int merge_buf_cap) {
    uint16_t big_endian_len, host_endian_len, cur_msg_len;

    memcpy(&big_endian_len, (void*)&merge_buf[1], sizeof(big_endian_len));
    host_endian_len = ntohs(big_endian_len);
    cur_msg_len = recv_buf[0] & 0x0f;

    if((recv_buf[0] & 0xf0) != 0x90 || merge_buf.size() + msg_size > merge_buf_cap || host_endian_len + cur_msg_len > 0xffff) {
        // upper 4 digits are not 1001
        // or merge_buf cannot hold anymore data
        // or array 16 in the merge_buf cannot hold more objs (although not possible right now, but have to check)
        return false;
    }

    // start merging
    host_endian_len += cur_msg_len;
    merge_buf.append(++recv_buf.begin(), recv_buf.begin() + msg_size);

    // update elem cnt in array 16
    big_endian_len = htons(host_endian_len);

    memcpy((void*)&merge_buf[1], &big_endian_len, sizeof(big_endian_len));
    return true;
}

{ // receiver function 
    // array 16 with 0 member
    std::string merge_buf({(char)0xdc, (char)0x00, (char)0x00});

    for(int i = 0 ; i < 1024; ++i) {
        int r = recv(fd, const_cast<char *>(tmp_buffer_.data()), tmp_buffer_size_, 0);
        if (r > 0) {
            if(!tryMerge(merge_buf, tmp_buffer_, r, tmp_buffer_size_)) {
                // Submit Task
            }
        // Some other logics
    }
}

从关键的系统指标的角度看,在merge逻辑有收益时(接收QPS = 48k,75k,120k,150k),小包合并逻辑大大减少了上下文切换,执行指令数,icache/dcache miss,并且增加了IPC(instructions per cycle)见下表:

同时通过对前后火焰图的对比分析看,在合并数据包之后,原本用于调度线程池的cpu资源更多的消耗在了收包上,也解释了小包合并之后context switch减少的情况。

Case 3

用户在打点指标中的Tags,是拼接成字符串进行纯文本传递的,这样设计的主要目的是简化SDK和Agent之间的数据格式。但这种方式就要求Agent必须对字符串进行解析,将文本化的Tags反序列化出来,又由于在接收端收到的用户打点QPS很高,这也成为了Agent的性能热点。

早期Agent在实现这个解析操作时,采用了遍历字符串的方式,将字符串按|=分割成 key-value 对。在其成为性能瓶颈后,我们发现它很适合使用SIMD进行加速处理。

原版

inline bool is_tag_split(const char &c) {
    return c == '|' || c == ' ';
}

inline bool is_kv_split(const char &c) {
    return c == '=';
}

bool find_str_with_delimiters(const char *str, const std::size_t &cur_idx, const std::size_t &end_idx,
    const Process_State &state, std::size_t *str_end) {
    if (cur_idx >= end_idx) {
        return false;
    }
    std::size_t index = cur_idx;
    while (index < end_idx) {
        if (state == TAG_KEY) {
            if (is_kv_split(str[index])) {
                *str_end = index;
                return true;
            } else if (is_tag_split(str[index])) {
                return false;
            }
        } else {
            if (is_tag_split(str[index])) {
                *str_end = index;
                return true;
            }
        }
        index++;
    }
    if (state == TAG_VALUE) {
        *str_end = index;
        return true;
    }
    return false;
}

SIMD

#if defined(__SSE__)
static std::size_t find_key_simd(const char *str, std::size_t end, std::size_t idx) {
    if (idx >= end) { return 0; }

    for (; idx + 16 <= end; idx += 16) {
        __m128i v = _mm_loadu_si128((const __m128i*)(str + idx));
        __m128i is_tag = _mm_or_si128(_mm_cmpeq_epi8(v, _mm_set1_epi8('|')),
                                     _mm_cmpeq_epi8(v, _mm_set1_epi8(' ')));
        __m128i is_kv = _mm_cmpeq_epi8(v, _mm_set1_epi8('='));

        int tag_bits = _mm_movemask_epi8(is_tag);
        int kv_bits = _mm_movemask_epi8(is_kv);
        // has '|' or ' ' first
        bool has_tag_first = ((kv_bits - 1) & tag_bits) != 0;
        if (has_tag_first) { return 0; }
        if (kv_bits) { // found '='
            return idx + __builtin_ctz(kv_bits);
        }
    }

    for (; idx < end; ++idx) {
        if (is_kv_split(str[idx])) { return idx; } 
        else if (is_tag_split(str[idx])) { return 0; }
    }

    return 0;
}

static std::size_t find_value_simd(const char *str, std::size_t end, std::size_t idx) {
    if (idx >= end) { return 0; }

    for (; idx + 16 <= end; idx += 16) {
        __m128i v = _mm_loadu_si128((const __m128i*)(str + idx));
        __m128i is_tag = _mm_or_si128(_mm_cmpeq_epi8(v, _mm_set1_epi8('|')),
                                     _mm_cmpeq_epi8(v, _mm_set1_epi8(' ')));
        int tag_bits = _mm_movemask_epi8(is_tag);
        if (tag_bits) {
            return idx + __builtin_ctz(tag_bits);
        }
    }

    for (; idx < end; ++idx) {
        if (is_tag_split(str[idx])) { return idx; }
    }

    return idx;
}

构建的测试用例格式为 。text 则是测试例子里的 str_size,用来测试不同 str_size 下使用 simd 的收益。可以看到,在 str_size 较大时,simd 性能明显高于标量的实现。

str_sizesimdscalar
1109140
2145158
4147198
8143283
16155459
32168809
642201589
1282893216
2564776297
51288312494
1024168724410

数据处理

Case 1

Agent在数据聚合过程中,需要一个map来存储一个指标的所有序列,用于对一段时间内的打点值进行聚合计算,得到一个固定间隔的观测值。这个map的key是指标的tags,map的value是指标的值。我们通过采集火焰图发现,这个map的查找操作存在一定程度的热点。

下面是 _M_find_before_node 的实现:


这个函数作用是:算完 hash 后,在 hash 桶里找到匹配 key 的元素。这也意味着,即使命中了,hash 查找的时候也要进行一次 key 的比较操作。而在 Agent 里,这个 key 的比较操作定义为:

bool operator==(const TagSet &other) const {
        if (tags.size() != other.tags.size()) {
            return false;
        }
        for (size_t i = 0; i < tags.size(); ++i) {
            auto &left = tags[i];
            auto &right = other.tags[i];
            if (left.key_ != right.key_ || left.value_ != right.value_) {
                return false;
            }
        }
        return true;
    }

这里需要遍历整个 Tagset 的元素并比较他们是否相等。在查找较多的情况下,每次 hash 命中后都要进行这样一次操作是非常耗时的。可能导致时间开销增大的原因有:

  1. 每个 tag 的 key_ 和 value_ 是单独的内存(如果数据较短,stl 不会额外分配内存,这样的情况下就没有单独分配的内存了),存在着 cache miss 的开销,硬件预取效果也会变差;
  2. 需要频繁地调用 memcmp 函数;
  3. 按个比较每个 tag,分支较多。


因此,我们将 TagSet 的数据使用 string_view 表示,并将所有的 data 全部存放在同一块内存中。在 dictionary encode 的时候,再把 TagSet 转换成 string 的格式返回出去。

// TagView 
#include <functional>
#include <string>
#include <vector>

struct TagView {
    TagView() = default;
    TagView(std::string_view k, std::string_view v) : key_(k), value_(v) {}
    std::string_view key_;
    std::string_view value_;
};

struct TagViewSet {
    TagViewSet() = default;
    TagViewSet(const std::vector<TagView> &tgs, std::string&& buffer) : tags(tgs), 
        tags_buffer(std::move(buffer)) {}
    TagViewSet(std::vector<TagView> &&tgs, std::string&& buffer) { tags = std::move(tgs); }
    TagViewSet(const std::vector<TagView> &tgs, size_t buffer_assume_size) {
        tags.reserve(tgs.size());
        tags_buffer.reserve(buffer_assume_size);
        for (auto& tg : tgs) {
            tags_buffer += tg.key_;
            tags_buffer += tg.value_;
        }
        const char* start = tags_buffer.c_str();
        for (auto& tg : tgs) {
            std::string_view key(start, tg.key_.size());
            start += key.size();
            std::string_view value(start, tg.value_.size());
            start += value.size();
            tags.emplace_back(key, value);
        }
    }

    bool operator==(const TagViewSet &other) const {
        if (tags.size() != other.tags.size()) {
            return false;
        }
        // not compare every tag
        return tags_buffer == other.tags_buffer;
    }

    std::vector<TagView> tags;
    std::string tags_buffer;
};

struct TagViewSetPtrHash {
    inline std::size_t operator()(const TagViewSet *tgs) const {
        return std::hash<std::string>{}(tgs->tags_buffer);
    }
};

验证结果表明,当 Tagset 中 kv 的个数大于 2 的时候,新方法性能较好。

数据发送

Case 1

早期Agent使用zlib进行数据发送前的压缩,随着用户打点规模的增长,压缩逐步成为了Agent的性能热点。

因此我们通过构造满足线上用户数据特征的数据集,对常用的压缩库进行了测试:

zlib使用cloudflare


zlib使用1.2.11


通过测试结果我们可以看到,除bzip2外,其他压缩算法均在不同程度上优于zlib:

结合业务场景考虑,我们最终执行短期使用 zlib-cloudflare 替换,长期使用 zstd 替换的优化方案。

结论

上述优化取得了非常好的效果,经过上线验证得出:

综合分析以上性能热点和优化方案,可以看到我们对Agent优化的主要考量点是:

除此之外,我们还在开展 PGO 和 clang thinLTO 的验证工作,借助编译器的能力来进一步优化Agent性能。

参考引用

  1. v2_0_cpp_unpacker:https://github.com/msgpack/msgpack-c/wiki/v2_0_cpp_unpacker#memory-management
  2. messagepack-specification:https://github.com/msgpack/msgpack/blob/master/spec.md
  3. Cloudflare fork of zlib with massive performance improvements:https://github.com/RJVB/zlib-cloudflare
  4. Intel® Intrinsics Guide:https://www.intel.com/content/www/us/en/docs/intrinsics-guide/index.html
  5. Profile-guided optimization:https://en.wikipedia.org/wiki/Profile-guided_optimization
  6. ThinLTO:https://clang.llvm.org/docs/ThinLTO.html

Copyright© 2013-2019

京ICP备2023019179号-2