gperftools源碼分析和項目應用 - CPU Profiler

原文：https://blog.csdn.net/yubo112002/article/details/81076821 

原文鏈接：http://www.tealcode.com/gperftool_source_analysis/

 

Google的gperftool是一款非常好用的服務器程序性能分析工具，能提供非常直觀和相對准確的性能數據，讓開發者可以進行更有方向能的優化。關於工具的使用方法，用gperftool作關鍵字搜索，會有很多的結果，這里就不多講了。本文的重點在於深入到工具源碼的內部，了解一下這個工具的實現原理和數據格式，然后介紹一下我從事的一個商業項目集成使用這個工具的一點小技巧。

 

工作原理
這一部分會重點解答這么幾個問題：

1、這個工具是如何收集程序的性能數據的？

2、這個工具使用的時候，不需要在產品代碼中插入任何的額外代碼，那么它怎么能知道哪個函數執行了多長時間呢？

3、工具的介紹上說，這個工具不工作的時候，對目標程序的執行性能幾乎沒有任何影響。可信嗎？幾乎沒影響到底是多大的影響？產品能接受這樣的影響嗎？

如果上面三個問題你已經能非常清楚地解答了，那這篇文章你可以直接跳到最后一部分：項目應用小技巧那里了，看看這個技巧對你是不是有點用處。

 

廢話少說，直接到工具的源碼中去找答案吧。考慮到貼太多的代碼在這里容易迷失在不必要的細節里，我這里就只放最核心的功能代碼了，為了讓邏輯看上去更清晰，下面貼出的代碼都刪除了一些錯誤檢查類的容錯代碼。

extern “C” PERFTOOLS_DLL_DECL int ProfilerStart(const char* fname) {
    return CpuProfiler::instance_.Start(fname, NULL);
}

bool CpuProfiler::Start(const char* fname, const ProfilerOptions* options) {
    collector_.Start(fname, collector_options);
    // Setup handler for SIGPROF interrupts
    EnableHandler();
    return true;
}
CpuProfiler啟動的時候，核心功能就是啟動數據收集器（collector_），這個數據收集器的Start函數的功能就是初始化數據收集需要的數據結構，並創建數據收集文件：

bool ProfileData::Start(const char* fname, const ProfileData::Options& options) {

    // Open output file and initialize various data structures
    int fd =open(fname, O_CREAT | O_WRONLY | O_TRUNC, 0666);
    start_time_ = time(NULL);
    fname_ = strdup(fname);

    // Reset counters
    num_evicted_ = 0;
    count_ = 0;
    evictions_ = 0;
    total_bytes_ = 0;

    hash_ = new Bucket[kBuckets];
    evict_ = new Slot[kBufferLength];
    memset(hash_, 0, sizeof(hash_[0]) * kBuckets);

    // Record special entries
    evict_[num_evicted_++] = 0; // count for header
    evict_[num_evicted_++] = 3; // depth for header
    evict_[num_evicted_++] = 0; // Version number
    CHECK_NE(0, options.frequency());
    int period =1000000/ options.frequency();
    evict_[num_evicted_++] = period; // Period (microseconds)
    evict_[num_evicted_++] = 0; // Padding
    out_ = fd;
    return true;
}
然后就是開啟了CpuProfiler的一個處理函數，而這個函數做的事情就是把prof_handler這個函數注冊到了某個地方。

void CpuProfiler::EnableHandler() {
    prof_handler_token_ = ProfileHandlerRegisterCallback(prof_handler, this);
}
注冊這個函數是干什么用的呢？

ProfileHandlerToken* ProfileHandlerRegisterCallback(
    ProfileHandlerCallback callback, void* callback_arg) {
    return ProfileHandler::Instance()->RegisterCallback(callback, callback_arg);
}
好吧，看來功能都在ProfileHandler里面了。ProfileHandler又是一個單例類，來看它的構造函數：

ProfileHandler::ProfileHandler() {

    timer_type_ = (getenv(“CPUPROFILE_REALTIME”) ? ITIMER_REAL : ITIMER_PROF);
    signal_number_ = (timer_type_ == ITIMER_PROF ? SIGPROF : SIGALRM);

    // Get frequency of interrupts (if specified)
    char junk;
    constchar* fr =getenv(“CPUPROFILE_FREQUENCY”);

    if (fr != NULL && (sscanf(fr, "%u%c", &frequency_, &junk) == 1) && (frequency_ > 0)) {
        // Limit to kMaxFrequency
        frequency_ = (frequency_ > kMaxFrequency) ? kMaxFrequency : frequency_;
    } else {
        frequency_ = kDefaultFrequency;
    }

    // Install the signal handler.
    structsigaction sa;
    sa.sa_sigaction = SignalHandler;
    sa.sa_flags = SA_RESTART | SA_SIGINFO;
    sigemptyset(&sa.sa_mask);
    sigaction(signal_number_, &sa, NULL);
}
這個構造函數，根據環境變量CPUPROFILE_REALTIME的配置，來決定讓SIGPROF還是SIGALRM信號來觸發SignalHandler信號處理函數，並根據環境變量CPUPROFILE_FREQUENCY的配置來設置自己的一個頻率變量 frequency_，如果沒有設置，就使用默認值，這個默認值是100，而最大值是4000.

然后ProfileHandler的RegisterCallback函數的實現如下：

ProfileHandlerToken* ProfileHandler::RegisterCallback(ProfileHandlerCallback callback, void* callback_arg) {

    ProfileHandlerToken* token = new ProfileHandlerToken(callback, callback_arg);
    SpinLockHolder cl(&control_lock_);
    DisableHandler();
    {
        SpinLockHolder sl(&signal_lock_);
        callbacks_.push_back(token);
    }

    // Start the timer if timer is shared and this is a first callback.
    if ((callback_count_ == 0) && (timer_sharing_ == TIMERS_SHARED)) {
        StartTimer();
    }
    ++callback_count_;
    EnableHandler();
    return token;
}
這個函數就如其函數名字，把指定的回調函數添加到callbacks_里面去，然后在加入第一個callback的時候調用StartTimer（）函數來啟動定時器，然后調用EnableHander函數來開啟回調。StartTimer()的實現如下：

void ProfileHandler::StartTimer() {
    struct itimerval timer;
    timer.it_interval.tv_sec = 0;
    timer.it_interval.tv_usec = 1000000 / frequency_;
    timer.it_value = timer.it_interval;
    setitimer(timer_type_, &timer, 0);
}
而EnableHandler()的實現如下：

void ProfileHandler::EnableHandler() {
    struct sigaction sa;
    sa.sa_sigaction = SignalHandler;
    sa.sa_flags = SA_RESTART | SA_SIGINFO;
    sigemptyset(&sa.sa_mask);
    const int signal_number = (timer_type_ == ITIMER_PROF ? SIGPROF : SIGALRM);
    RAW_CHECK(sigaction(signal_number, &sa, NULL) == 0, "sigprof (enable)");
}
好了，到這里，這個工具的基本工作原理已經可以猜出個大概了。它用setitimer啟動一個系統定時器，這個定時器會每秒鍾執行觸發frequency次SIGPROF或者SIGALRM信號，從而去觸發上面注冊的信號處理函數。那么猜想，信號處理函數里面應該會用backtrace去檢查一下目標程序執行到什么位置了。那么繼續看信號處理函數里面都做了些什么事情吧。

void CpuProfiler::prof_handler(int sig, siginfo_t*, void* signal_ucontext, void* cpu_profiler) {
    CpuProfiler* instance = static_cast<CpuProfiler*>(cpu_profiler);
    if (instance->filter_==NULL||(*instance->filter_)(instance->filter_arg_)) {
        void* stack[ProfileData::kMaxStackDepth];
        stack[0] = GetPC(*reinterpret_cast<ucontext_t*>(signal_ucontext));
        int depth = GetStackTraceWithContext(stack +1, arraysize(stack) -1, 3, signal_ucontext);

        void**used_stack;
        if (depth >0&& stack[1] == stack[0]) {
            // in case of non-frame-pointer-based unwinding we will get
            // duplicate of PC in stack[1], which we don’t want
            used_stack = stack + 1;
        } else {
            used_stack = stack;
            depth++; // To account for pc value in stack[0];
        }
        instance->collector_.Add(depth, used_stack);
    }
}
果然是獲取backtrace，然后記錄到colloector_里面去。另外這里為了讓代碼邏輯看起來更清晰，沒有貼出來源代碼中的大段注釋，那些注釋詳細解釋了對stack數組下標的那幾個加減值，感興趣的話可以自行前往源代碼去進一步閱讀。

到此為止，本文開頭的三個問題都可以有答案了。

1、這個工具是用系統定時器定時產生信號的方式，在信號處理函數里面獲取當前的調用堆棧來確定當前落在哪個函數里面的。獲取頻率默認是每10ms采樣一次，參數是可調的，但是最大頻率是4000，也就是支持的最小采樣間隔是250微秒；

2、這個工具獲取到的性能數據是基於統計數據的，也就是他並不真正跟蹤函數的每一次調用過程，而是均勻地采樣並記錄采樣點所落在的函數調用位置，用這些統計數據來計算每個函數的執行時間占比。這個數據並不是准確的數據，但是只要運行時間相對比較長，統計數據還是能比較准確地說明問題的。而這也是為什么說這個工具是比較好的服務器程序性能分析工具，而對一些客戶端程序，比如游戲客戶端並不是非常合適。因為游戲客戶端上，相比長時間的統計數據，它們通常更加關心的是某些幀內的具體負載情況。

3、這個工具不工作的時候，就會把系統定時器取消掉，不會定時產生中斷信號，不會觸發中斷處理程序，所以對運行程序的影響真的是很小，運行效率上可以說完全沒有影響。而對產品的影響只是多占用一些鏈接profiler庫的內存而已。

收集器中的數據格式
先來看ProfileData類中相關的結構定義：

static const int kAssociativity =4; // For hashtable
static const int kBuckets =1<<10; // For hashtable
static const int kBufferLength =1<<18; // For eviction buffer

// Type of slots: each slot can be either a count, or a PC value
typedef uintptr_t Slot;

// Hash-table/eviction-buffer entry (a.k.a. a sample)
struct Entry {
  Slot count; // Number of hits
  Slot depth; // Stack depth
  Slot stack[kMaxStackDepth]; // Stack contents
};

// Hash table bucket
struct Bucket {
  Entry entry[kAssociativity];
};
使用這些結構的成員如下：

Bucket* hash_; // hash table
Slot* evict_; // evicted entries
int num_evicted_; // how many evicted entries?
創建代碼：

hash_ = new Bucket[kBuckets];             //長度1024的hash表
evict_ = new Slot[kBufferLength];         //256K的移除buffer
memset(hash_, 0, sizeof(hash_[0]) * kBuckets);

// Record special entries
evict_[num_evicted_++] = 0; // count for header
evict_[num_evicted_++] = 3; // depth for header
evict_[num_evicted_++] = 0; // Version number

CHECK_NE(0, options.frequency());
int period =1000000/ options.frequency();
evict_[num_evicted_++] = period; // Period (microseconds)
evict_[num_evicted_++] = 0; // Padding
收集數據的邏輯：

//1. Make hash-value
Slot h = 0;
for (int i =0; i < depth; i++) {
    Slot slot = reinterpret_cast<Slot>(stack[i]);
    h = (h << 8) | (h >> (8*(sizeof(h)-1)));
    h += (slot * 31) + (slot * 7) + (slot * 3);
}
count_++;

//2. See if table already has an entry for this trace
bool done =false;
Bucket* bucket = &hash_[h % kBuckets];

for (int a =0; a < kAssociativity; a++) {
    Entry* e = &bucket->entry[a];
    if (e->depth== depth) {
        bool match =true;
        for (int i =0; i < depth; i++) {
            if (e->stack[i] !=reinterpret_cast<Slot>(stack[i])) {
                match = false;
                break;
            }
        }

        if (match) {
            e->count++;
            done = true;
            break;
        }
    }
}

// 3.          
if (!done) {
    // Evict entry with smallest count
    Entry* e = &bucket->entry[0];
    for (int a =1; a < kAssociativity; a++) {
        if (bucket->entry[a].count< e->count) {
            e = &bucket->entry[a];
        }
    }

    if (e->count>0) {
        evictions_++;
        Evict(*e);
    }

    // Use the newly evicted entry
    e->depth = depth;
    e->count = 1;
    for (int i =0; i < depth; i++) {
        e->stack[i] = reinterpret_cast<Slot>(stack[i]);
    }
}
可以看到，它使用了長度為1024的Bucket數組來存放性能收集的記錄，每個Bucke能最多存放四條hash沖突的記錄。

拿到性能記錄之后，第一步先對記錄中的backtrace計算hash值，hash值模余1024確定存儲該條記錄使用的Bucket，然后在Bucket的四個位置中查看能不能找到一個完全一樣的backtrace，如果能找到，就直接在這個位置上累加計數；如果找不到，說明遇到了一個全新的backtrace，那么就在四個位置中找一個當前計數最少的位置來存儲當前的記錄。如果目標位置原來沒有計數，那就直接當做一條新的記錄添加進去，而如果目標位置處已經有計數了，說明當前的Bucket已經滿了，那么就把當前位置處的記錄驅逐到evict_數組中，而把新的記錄保存到當前的位置上。

驅逐邏輯的代碼是這樣的：

void ProfileData::Evict(const Entry& entry) {
  const int d = entry.depth;
  const int nslots = d +2; // Number of slots needed in eviction buffer
  if (num_evicted_ + nslots > kBufferLength) {
    FlushEvicted();
    assert(num_evicted_ ==0);
    assert(nslots <= kBufferLength);
  }

  evict_[num_evicted_++] = entry.count;
  evict_[num_evicted_++] = d;
  memcpy(&evict_[num_evicted_], entry.stack, d *sizeof(Slot));
  num_evicted_ += d;
}
如果當前evict_數組已經放不下當前的記錄了，那就先用FlushEvicted方法把當前的內容都寫入到文件中去，然后清空當前的evict_數組，從頭開始放這些被驅逐出來的記錄。結合初始化的時候注釋為“Record special entries”的代碼塊，可以看到，寫入到文件中的結構是開頭的固定的五個slot的文件頭，slot的大小取決於目標程序是32位的還是64位的，然后后面會跟着多塊采樣數據，每塊數據都是固定的兩個slot分別存放采樣點命中的次數和backtrace的深度，然后后面跟着可變長度的N個PC值，N由backtrace的深度值來決定，每個Bucket中的Entry的結構與此也是一樣的，而Bucket中的Entry，是在性能數據收集完成之后，統一Flush到文件中。在所有采樣點數據dump完成之后，會用三個slot來作為數據結束的標記，分別設置為0,1,0，最后還會把當前進程的maps信息輸出到最終的文件中。輸出maps信息的作用，是幫助后期定位到某個PC值來源於哪個動態鏈接庫，並可以根據偏移量來取得它對應的函數名。

void ProfileData::Stop() {

    if (!enabled()) {
        return;
    }

// Move data from hash table to eviction buffer
for (int b =0; b < kBuckets; b++) {
    Bucket* bucket = &hash_[b];
    for (int a =0; a < kAssociativity; a++) {
        if (bucket->entry[a].count>0) {
            Evict(bucket->entry[a]);
        }
    }
}

if (num_evicted_ +3> kBufferLength) {
    // Ensure there is enough room for end of data marker
    FlushEvicted();
}

// Write end of data marker
evict_[num_evicted_++] = 0; // count
evict_[num_evicted_++] = 1; // depth
evict_[num_evicted_++] = 0; // end of data marker
FlushEvicted();

// Dump “/proc/self/maps” so we get list of mapped shared libraries
DumpProcSelfMaps(out_);
Reset();
fprintf(stderr, “PROFILE: interrupts/evictions/bytes = %d/%d/%” PRIuS “\n”,
    count_, evictions_, total_bytes_);
}
下面一張圖是dump了一個真實的性能數據，可以來對比驗證一下：

 

這是一個在64位機器上運行的Linux程序，所以每個slot是8個字節，開始時5個Slot的文件頭，其中第四個Slot指示采樣的間隔是10000(0x2710)微秒，也就是默認的每秒采樣100次。然后后面可以找到兩塊采樣點數據，第一個塊命中了三次，backtrace深度是10；第二塊命中了一次，backtrace深度是7。然后是值分別為0，1，0的采樣數據結束標志。在后面就是ascii字符形式保存的maps文本。結合pprof的文本方式的分析結果，也可以驗證我們上面的觀察：

 

這下文件結構應該很清楚了，甚至pprof分析工具應該如何處理的邏輯也能想出個大概來了。

項目應用的小技巧
使用gperftools收集運行數據的時候，需要在需要開始收集的位置調用ProfilerStart(),並在結束收集的時候調用profilerStop()，收集到的數據才會被寫入到文件里面去。但是有時我們希望能動態地控制性能數據收集的開始和結束時間，而不想頻繁地修改代碼中ProfilerStart() 和 ProfilerStop()的插入位置。

有兩種方法：

1、在產品中添加自定義信號處理函數，比如可以分別在SIGUSR1和 SIGUSR2信號的處理函數中執行ProfilerStart()和ProfilerStop()，使用的時候用kill程序發送指定的信號來開啟和結束數據收集就可以了；

2、產品中啟動一個專門監聽外部命令的線程，接收到指定命令時開啟和結束性能收集。比如監聽一個本地Socket，在這個socket上接收到命令時就執行，並把輸出也都反饋到這個本地socket中去。這樣只要再寫另外一個簡單的讀寫這個socket的小程序，就可以很方便地實現動態控制服務器進程的效果。

易用性上的考慮，推薦使用第二種方法。這樣可以根據自己的需要靈活擴展這個監聽線程的功能，控制客戶端工具也能做到非常人性化的交互接口。監聽線程還可以擴展很多其他的調試或監控功能能，而這個線程在沒有命令需求的時候，只是阻塞在一個Socket監聽事件上，對產品的運行沒有任何其他影響。

對CPUProfiler的分析就到這里了，后面還會整理一個隊TCMalloc的源碼級分析，看看google是如何加速多線程應用的內存分配性能的，敬請期待。

        任何問題，歡迎在評論區留言討論。

====================== End