2011年2月12日 星期六

幫 C 語言加上 Garbage Collector

前幾天讀 Dragon Book 的時候突發奇想,何不自己實作 Conservative Garbage Collector?

一般來說 Tracing Garbage Collector 分成以下三個步驟:
  1. 先從 call stack 或者 global variables 的 references (或者 pointers) 蒐集 object 的位址作為 root set,並標記 root set 當中的 object。
  2. 檢視 root set 當中的 object。如果 object 也包含 reference,而且這個 reference 指向的 object' 沒有被標記過,就標記並檢視 object'。(不斷遞迴直到沒有新的 object 被標記)
  3. 清除並回收沒有被標記的 object。
然而為了有效率地計算 root set 與走訪被標記的 objects,通常我們會需要編譯器加入一些額外的資訊。而且程式語言必需是 type-safe 的!著名的例子如 Java 或者是 Haskell、Common Lisp 等等。而 C/C++ 就不適合當 Garbage Collector 的目標語言,因為 C/C++ 支援 Pointer Arithmetic、Union Type 等語言構件,使得我們幾乎不可能在 C/C++ 之上實作 (Accurate) Garbage Collector。

不過 Conservative Garbage Collector 的做法不太一樣。Conservative Garbage Collector 的想法是:我不去分辨誰是 Pointer Type 誰是 Integer Type,只要「看起來」像是 Pointer 的 bit pattern,我就把他當成 Pointer。這有一個好處:我們不需要編譯器的幫忙也可以寫 Garbage Collector,所以可以和其他的程式碼混用;然而也有一個壞處:有時候 Conservative Garbage Collector 過於保守,以至於無法回收一些可以回收的記憶體。以下我就以一個小程式展現 Conservative Garbage Collector 的基本概念。

首先我們的第一個問題是如何找到 root set?我們可以取 main 函式 argc 的位址,另外在 gc_cleanup_ 函式隨意傳入一個整數,再取該整數的位址。所有還可能會用到的 pointer 就會在這二個整數之間:

void *stack_top;

void gc_init(int *argc_ptr) {
  stack_top = argc_ptr;
}

void gc_cleanup_(int stack_indicator) {
  void *stack_bottom = &stack_indicator; 
  /* Stack 的有效範圍就介於 [stack_bottom, stack_top] 之間 */
}

int main(int argc, char **argv) {
  gc_init(&argc);
}

當然我們還必需考慮 alignment 的問題,所以要把 stack_top 與 stack_bottom 對齊到 sizeof(void *) 的某倍。這可以用以下二個巨集來達成:

#define FLOOR_ALIGN_TO_WORD(X) \
    (((uintptr_t)(X)) / sizeof(void *) * sizeof(void *))

#define CEIL_ALIGN_TO_WORD(X) \
    (((uintptr_t)(X) + sizeof(void *) - 1) / \
        sizeof(void *) * sizeof(void *))

接著我們要準備一些空間來當我們的 Heap,所以我們在 gc_init 之中加入一些程式碼:

static char *heap_begin = NULL;
static char *heap_free = NULL;
static char *heap_end = NULL;

/* Initialize the internal data structure of the
   garbage collector */
void gc_init(int *argc_ptr) {
    /* We assume that argc is the variable in the
       stack having the highest address. */
    stack_top = (void **)CEIL_ALIGN_TO_WORD(argc_ptr);

    /* Allocate the heap using mmap. */
    heap_begin = (char *)mmap(0, GC_HEAP_SIZE,
                              PROT_READ | PROT_WRITE,
                              MAP_PRIVATE | MAP_ANONYMOUS,
                              -1, 0);

    if (!heap_begin || heap_begin == MAP_FAILED) {
        fprintf(stderr,
                "Unable to allocate the heap (size=%lu).\n",
                (unsigned long)GC_HEAP_SIZE);
        exit(EXIT_FAILURE);
    }

    heap_free = heap_begin;
    heap_end = heap_begin + GC_HEAP_SIZE;
}

接著我們需要一些資料結構用以管理我們分配出去的記憶體。做為一個簡單的範例,我只用了一個陣列依序記錄分配出去的記憶體,我的陣列會從 heap_end 開始往 heap_begin 的方向生長。

當然這是一個很沒有效率的作法,比較有效率的方法是使用類似 buddy system 之類的資料結構來管理 heap。

enum {
    ALLOC_STATUS_UNKNOWN,
    ALLOC_STATUS_TOUCHED,
    ALLOC_STATUS_REFERRED,
    ALLOC_STATUS_DEALLOCATED,
};

typedef struct alloc_record_ {
    char *addr;
    size_t size;
    unsigned int status;
} alloc_record;

為了操作 alloc_record 資料結構,我寫了三個函式,其功能分述如下(程式碼就不再贅述):

static alloc_record *find_alloc_record(char *addr) {
  /* 給定一個位址,找出它是屬於哪一次 allocation */
}

static void insert_alloc_record(char *addr, size_t size) {
  /* 在 alloc_record 的尾端加上一筆記錄 */
}

static void *allocate_deallocated_block(size_t size) {
  /* 重新把 deallocated 過的位址配置給不同的人 */
}

接下來是 Malloc 的程式碼。我們先嘗試直接從 heap_free 配置記憶體,如果記憶體不夠,再嘗試使用別人 deallocate 的記憶體。如果以上二者都沒有用,我們就必需進行 Garbage Collection。(備註:gc_cleanup(); 即 gc_cleanup_(0);)

/* Allocate a block with given bytes. */
void *gc_malloc(size_t size) {
    /* Make sure that all of returned address is aligned */
    size = CEIL_ALIGN_TO_WORD(size);

    /* Try to allocate if free space available */
    if (is_free_space_avail(size)) {
        char *result = heap_free;
        heap_free += size;
        insert_alloc_record(result, size);
        return result;
    }

    /* Try to allocate from deallocated address. */
    void *result = allocate_deallocated_block(size);
    if (result) {
        return result;
    }

    /* Try to collect garbage and retry. */
    while (gc_cleanup()) {
        void *result = allocate_deallocated_block(size);
        if (result) {
            return result;
        }
    }

    return NULL;
}

當然有 Malloc 就要有 Free,不過值得注意的是:為了提高 Conservative Garbage Collector 的回收率,減少「應該 deallocate 而沒有 deallocate」的情況,我們應該要抹除整個 object 與傳入的 Pointer

/* Deallocate and wipe the allocated block. */
static void deallocate(alloc_record *record) {
    /* Mark this block as deallocated */
    record->status = ALLOC_STATUS_DEALLOCATED;

    /* Wipe the memory block for cleaning the possible
       pointer address */
    memset(record->addr, '\0', record->size);
}

/* Deallocate the given address and wipe the pointer. */
void gc_free_(void **addr_ptr) {
    alloc_record *record = find_alloc_record(*addr_ptr);
    if (!record) {
        fprintf(stderr, "Memory corrupted\n");
        return;
    }

    deallocate(record);
    *addr_ptr = NULL;
}

最後就是最關鍵的 Garbage Collection 的程式碼 gc_cleanup

/* Garbage collection and deallocate unused memory. */
size_t gc_cleanup_(int stack_indicator) {
    alloc_record *records =
        (alloc_record *)heap_end - alloc_record_count;

    size_t i, j;

    /* Mark alloc record as UNKNOWN */
    for (i = 0; i < alloc_record_count; ++i) {
        if (records[i].status != ALLOC_STATUS_DEALLOCATED) {
            records[i].status = ALLOC_STATUS_UNKNOWN;
        }
    }

    /* Scan the stack */
    void **stack_bottom =
        (void **)CEIL_ALIGN_TO_WORD(&stack_indicator);
    assert(stack_top != NULL);
    assert(stack_bottom <= stack_top);
    scan_and_touch(stack_bottom, (void **)stack_top);

    /* Scan the heap */
    scan_touched_objects();

    /* Deallocate and reset the status */
    static unsigned int gc_count = 0;
    static char const sep[] =
    "-------------------------------------------------------";

    fprintf(stderr, "%s\n", sep);
    fprintf(stderr,
            "GARBAGE COLLECTION ROUND #%u\n", ++gc_count);

    size_t deallocated_count = 0;
    for (j = alloc_record_count, i = j - 1; j > 0; --i, --j) {
        if (records[i].status == ALLOC_STATUS_UNKNOWN) {
            fprintf(stderr,
                    "  Reclaim [addr: %p, size: %lu]\n",
                    records[i].addr,
                    (unsigned long)records[i].size);

            deallocate(&records[i]);
            ++deallocated_count;
        }
    }

    fprintf(stderr, "%s\n\n", sep);

    return deallocated_count;
}

/* Scan for address between [begin, end) */
static void scan_and_touch(void *begin_, void *end_) {
    void **begin = (void **)FLOOR_ALIGN_TO_WORD(begin_);
    void **end = (void **)FLOOR_ALIGN_TO_WORD(end_);

    for (; begin < end; ++begin) {
        char *addr = (char *)*begin;

        if (addr >= heap_begin && addr < heap_free) {
            alloc_record *record = find_alloc_record(addr);

            if (record && record->status ==
                    ALLOC_STATUS_UNKNOWN) {
                record->status = ALLOC_STATUS_TOUCHED;
            }
        }
    }
}

/* Scan the touched objects */
static void scan_touched_objects() {
    alloc_record *records =
        (alloc_record *)heap_end - alloc_record_count;

    size_t count, i;

    do {
        count = 0;
        for (i = 0; i < alloc_record_count; ++i) {
            if (records[i].status != ALLOC_STATUS_TOUCHED) {
                /* Not in touched state, skip it. */
                continue;
            }

            /* Mark as referred. */
            records[i].status = ALLOC_STATUS_REFERRED;

            /* Scan this object */
            scan_and_touch(
                (void **)(records[i].addr),
                (void **)(records[i].addr + records[i].size));

            count++;
        }
    } while (count > 0);
}

以上程式碼就是一個具體而微的 Conservative Garbage Collector,完整的程式碼與測試程式可以從這裡下載:conservativegc.h , conservativegc.c , test_tree.c , test_many.c 。當然這個 Conservative Garbage Collector 還少了很多東西:包括計算 static storage 的 root set(在 Linux 之下可以檢查 etext 與 end 二者之間的值),還有更有效率的 allocation policy 等等。

備註:如果你真的有在 C/C++ 使用 Garbage Collector 的需求可以參考 libgc,一個由 BoehmDemersWeiser 等前輩撰寫 (他們是提出 Conservative Garbage Collector 的重要前輩),並被移植到多個平台的 Conservative Garbage Collector 函式庫。

2011年1月25日 星期二

Rvalue Reference 與 String 的實作

Rvalue ReferenceC++0x 當中,一個非常重要的新功能。有了 Rvalue Reference,我們就可以容易地寫出具有 Moving Semantics 的 class。如果善用它,我們可以寫出更有效率的 C++ 程式碼。目前比較有名的編譯器如 GCC 或者 Visual C++ 都已經有 Rvalue Reference。

說來說去,Rvalue Reference 究竟是什麼?

首先我們必須要介紹 Lvalue 與 Rvalue。C++ 程式語言之中,Rvalue 與 Lvalue 相反的概念,Rvalue 的定義是「不是 Lvalue 的 Object(或曰 Variable),就是 Rvalue」。那 Lvalue 又是什麼?所謂的 Lvalue 是指在記憶體上有固定位置可以取址,可以放在指派運算子左邊的值。例如:

int a, b, c;
a = 1; b = 2; c = 3;  // (1)
a = b + c;  // (2)

第一行當中 a、b、c 三個變數都是放在指派運算子的左邊,所以都是 Lvalue。而第二行當中,a 是 Lvalue,(b+c) 這個表達式是 Rvalue。下面的程式很明顯是不合法的:

b+c = 0; // X

因為 b+c 這個 Rvalue 在記憶體上沒有固定的位址(翻譯為機器碼後,b+c 通常會被儲存在暫存器),所以自然也不能把 0 指派給他。但是這和 Rvalue Reference 有什麼關係呢?我們看看這個例子:

string a("abc");
string b("def");
string c("ghi");

string all = a + b + c + a;  // (3)

第三行的 (a+b) 、((a+b)+c)、(((a+b)+c)+a) 都是 Rvalue。在 C++98 當中,我們只能把 Rvalue 綁定到 const 修飾過的 Reference,所以我們的 operator+ 通常會這樣寫:

string operator+(string const &lhs, string const &rhs)  {
  string result(lhs); // (4) copy lhs
  result += rhs; // (5) concat rhs
  return result;
}

每當我們呼叫 operator+,我們就要複製一份 lhs,然後再把 rhs 串接上去。我們仔細地看一下上述程式的執行過程:
  1. 首先執行 a+b 的時候,我們要先複製一份 a(稱為 result1),再把 b 串接到 result1 的後面,最後回傳 result1。
  2. 接著我們要計算 result1+c 的結果。我們必需先複製一份 result1(稱之為 result2),再把 c 串接到 result2 的後面,最後回傳 result2。
  3. 最後我們要計算 result2+a 的結果,我們必需要先複製一份 result2(稱之為 result3),再把 a 串接到 result3 的後面,最後回傳 result3。
這個程式非常沒有效率。它會複製一份 result1,一份 result2,可是複製完就忘掉 result1 與 result2!為了減少這樣的浪費,有人提出了 Move Semantics:

struct string_tmp {
  string *ptr;
  string_tmp(string *s) : ptr(s) { }
  ~string_tmp() { delete ptr; }
};

// Move Constructor
string::string(string_tmp rhs) {
  internal_move(rhs); 
}

// operator=
string &string::operator=(string_tmp rhs) {
  internal_move(rhs);
  return *this;
}

void string::internal_move(string_tmp with) {
  if (buf) {
    delete [] buf;
  }

  buf = with.ptr->buf;
  buf_size = with.ptr->buf_size;
  str_length = with.ptr->str_length;

  with.ptr->buf = NULL;
  with.ptr->buf_size = 0;
  with.ptr->str_length = 0; 
}

string_tmp operator+(string const &lhs, string const &rhs) {
  string_tmp result(new string(lhs));
  *(result.ptr) += rhs;
  return result;
}

string_tmp operator+(string_tmp lhs, string const &rhs) {
  *(lhs.ptr) += rhs;
  return lhs;
}

上面的程式碼最重要的概念:operator+ 回傳的型別變成 string_tmp。然後多載 operator+ 與 operator=。這二個 operator 如果看到 string_tmp,就會想辦法把裡面的東西偷出來用,從而避免無謂的複製!

雖然在 C++98 當中,我們可以模仿出 Move Semantics,不過寫起來即為痛苦,不但需要黑魔法(上面的程式碼並不完整,詳情請參閱 moving_string.hpp),而且難以維護,如果 C++ 可以讓我們把 Rvalue 偷出來用就好了!

這就是 Rvalue Reference 可以讓我們做的事!

Rvalue Reference 在 C++ 的 notation 如下:

T &&rref = ... ;

我們可以把 Rvalue 綁定到 Rvalue Reference 上,例如:

string &&rref = a + b;

但是我們沒有辦法直接把 Lvalue 綁定到 Rvalue Reference 上,下面這個 statement 是不合法的:

string a;
string &&rref = a; // X

這是為了防止我們錯誤地把 Lvalue 當成 Rvalue。還記得嗎?Lvalue 是在記憶體上有特定位置,程序員碰得到 Lvalue,因此我們不能偷  Lvalue 的東西。如果我想告訴編譯器:「沒關係,這個變數我不在乎你去偷東西,你把它當成 Rvalue 就可以了!」可以使用 std::move:

string &&rref = std::move(a);

說了這麼多,到底要怎麼用 Rvalue Reference?和前面的 string_tmp 一樣,我們必需要多載 operator+、operator=、move constructor:

// Move Constructor
string::string(string &&rhs)
  : buf(NULL), buf_size(0), str_length(0) {

  internal_move(rhs);
}

// operator=
string &string::operator=(string &&rhs) {
  if (this == &rhs) {
    return *this;
  }
  internal_move(rhs);
  return *this;
}

void string::internal_move(string &with) throw () {
  if (buf) {
    delete [] buf;
  }

  buf = with.buf;
  buf_size = with.buf_size;
  str_length = with.str_length;

  with.buf = nullptr;
  with.buf_size = 0;
  with.str_length = 0;
}

// operator+
string &&operator+(string &&lhs, string const &rhs) {
  lhs += rhs;
  return std::move(lhs); // convert lhs to r-value again
}

我們可以注意到:我們不再需要 string_tmp,編譯器可以幫我們分辨何者是 Rvalue,並告訴我們可不可以從該物件偷取資源!

結語:有了 Rvalue Reference,我們可以寫出更有效率的程式碼。到了 C++0x 的時代,一個好的 C++ 程序員應該要能掌握 Rvalue Reference 的威力!

完整的程式碼:test.cppsimple_string.hpp(一般的字串)、moving_string.hpp(有 move semantics 的字串)、rrefopt_string.hpp(使用 rvalue reference 的字串)。

2010年12月3日 星期五

合併 Git 的 Commit

Git 是一個很強大的分散式版本管理系統,因為有了 git,我習慣寫二三十行 code 就 commit 到 local repository。然而如此一來,repository 會充滿一大堆不能正常 compile 的 commit,而且會有一大堆 Fix Typo 之類很沒有用的 commit。有時候會想要把一些 commits 合併之後再上傳給別人,這時候就可以使用 rebase 來合併修改計錄。

(不過以下的流程只適合在 local repository 做,一旦上傳到 public repository,就不適合再 rebase 了!)

首先我們要知道過去的 commit 記錄:

$ git log  # 先看每一個 commit 的 log 與 sha1 編號

假設我們要把 branch1 的  commitA..commitB 的 commits 合併起來,我們就這樣做(以下的 commitA 與 commitB 皆是指這二個 commit 的 sha1 編號):

$ git checkout commitB  # 先回到過去某個 commit

$ git reset --soft commitA  # 修改 working directory 的 commit index

$ git commit --amend  # 更新 commit 訊息與時間 (合併 commitA..commitB)

$ git tag tmp  # 記錄合併後的 commit 的 sha1 編號(你直接記 sha1 編號也是可以)

$ git checkout branch1  # 回到原本 branch 的 HEAD

$ git rebase --onto tmp commitB
# 把 commitB 之前的 commit history 換成合併之後的 commit。

$ git tag -d tmp  # 刪除暫時的 tag
--
資料來源:
How do I combine the first two commits of a Git repository?