並行程式設計: Reference Counting

貢獻者: csm1735, jserv

Reference counting

〈並行和多執行緒程式設計〉和〈Linux 核心模組運作原理〉提及 reference counting，refcnt 嘗試以 C11 Atomics 實作，參考執行輸出:

hread 1632626432, 0: Hello, world!
Thread 1632626432, 1: Hello, world!
Thread 1632626432, 2: Hello, world!
Thread 1632626432, 3: Hello, world!
...
Thread 1103886080, 97: Hello, world!
Thread 1103886080, 98: Hello, world!
Thread 1103886080, 99: Hello, world!

其中 1103886080 是執行緒的編號 (thread ID)。一旦 REFCNT_TRACE 事先定義，程式輸出如下:

...
main.c:141:(test_thread) refcnt_t_unref(str2)main.c:138:(test_thread) refcnt_t_ref(str)Thread 1174378240, 97: Hello, world!
main.c:141:(test_thread) refcnt_t_unref(str2)main.c:138:(test_thread) refcnt_t_ref(str)Thread 1174378240, 98: Hello, world!
main.c:141:(test_thread) refcnt_t_unref(str2)main.c:138:(test_thread) refcnt_t_ref(str)Thread 1174378240, 99: Hello, world!

程式碼運作原理

#define maybe_unused __attribute__((unused))

先定義 maybe_unused ，其作用為針對可能不會使用到的部份，消除編譯時的警告。

typedef struct {
#ifdef REFCNT_CHECK
    int magic;
#endif
    atomic_uint refcount;
    char data[];
} refcnt_t;

宣告了一個名為 refcnt_t 的 struct，可見其中除了 data 外，還有一個 atomic_uint 的變數 refcount ，用來以 atomic 的型態儲存 reference counting ，而如果程式中有定義 REFCNT_CHECK ， struct 中還會再多一個 magic 的變數，用途為檢查傳入 refcnt_ref 和 refcnt_unref 的 reference 是否由 refcnt_malloc 或 refcnt_strdup 所建立。

static maybe_unused void *refcnt_malloc(size_t len)
{
    refcnt_t *ref = malloc(sizeof(refcnt_t) + len);
    if (!ref)
        return NULL;
#ifdef REFCNT_CHECK
    ref->magic = REFCNT_MAGIC;
#endif
    atomic_init(&ref->refcount, 1);
    return ref->data;
}

先透過 malloc 配置大小為 sizeof(refcnt_t) + len 的記憶體空間，如果配置失敗則直接回傳 NULL ，如果有定義 REFCNT_CHECK 的話，則將 ref->magic 設為事先定義好的魔術數 REFCNT_MAGIC ，接著再透過 atomic_init 將代表 reference counting 的 ref->refcount 初始化為 1，最後會回傳 ref->data。

static maybe_unused void *refcnt_realloc(void *ptr, size_t len)
{
    refcnt_t *ref = (void *) (ptr - offsetof(refcnt_t, data));
#ifdef REFCNT_CHECK
    assert(ref->magic == REFCNT_MAGIC);
#endif
    ref = realloc(ref, sizeof(refcnt_t) + len);
    if (!ref)
        return NULL;
    return ref->data;
}

使用 offsetof 計算 refcnt_t 的成員 data 距離結構起始位址的偏移量，以透過 (ptr - offsetof(refcnt_t, data)) 取得結構起始位址，如果有定義 REFCNT_CHECK 的話，則會去檢查 ref->magic ，接著透過 realloc 重新調整 ref 所指向的記憶體空間的大小為 sizeof(refcnt_t) + len ，如果配置失敗則回傳 NULL ，最後會回傳 ref->data。

static maybe_unused void *refcnt_ref(void *ptr)
{
    refcnt_t *ref = (void *) (ptr - offsetof(refcnt_t, data));
#ifdef REFCNT_CHECK
    assert(ref->magic == REFCNT_MAGIC && "Invalid refcnt pointer");
#endif
    atomic_fetch_add(&ref->refcount, 1);
    return ref->data;
}

使用 offsetof 計算 refcnt_t 的成員 data 距離結構起始位址的偏移量，以透過 (ptr - offsetof(refcnt_t, data)) 取得結構起始位址，如果有定義 REFCNT_CHECK 的話，則會去檢查 ref->magic ，接著再透過 atomic_fetch_add 去將 ref->refcount 做加一的動作，最後會回傳 ref->data。

static maybe_unused void refcnt_unref(void *ptr)
{
    refcnt_t *ref = (void *) (ptr - offsetof(refcnt_t, data));
#ifdef REFCNT_CHECK
    assert(ref->magic == REFCNT_MAGIC && "Invalid refcnt pointer");
#endif
    if (atomic_fetch_sub(&ref->refcount, 1) == 1)
        free(ref);
}

使用 offsetof 計算 refcnt_t 的成員 data 距離結構起始位址的偏移量，以透過 (ptr - offsetof(refcnt_t, data)) 取得結構起始位址，如果有定義 REFCNT_CHECK 的話，則會去檢查 ref->magic ，接著再透過 atomic_fetch_sub 去將 ref->refcount 做減一的動作，如果 ref->refcount 的值在減一之前為 1 ，即做完減一的動作後會變成 0 的話，則將 ref 做 free 的動作。

static maybe_unused char *refcnt_strdup(char *str)
{
    refcnt_t *ref = malloc(sizeof(refcnt_t) + strlen(str) + 1);
    if (!ref)
        return NULL;
#ifdef REFCNT_CHECK
    ref->magic = REFCNT_MAGIC;
#endif
    atomic_init(&ref->refcount, 1);
    strcpy(ref->data, str);
    return ref->data;
}

先透過 malloc 配置大小為 sizeof(refcnt_t) + strlen(str) + 1 的記憶體空間，如果配置失敗則直接回傳 NULL ，如果有定義 REFCNT_CHECK 的話，則將 ref->magic 設為事先定義好的魔術數 REFCNT_MAGIC ，接著再透過 atomic_init 將代表 reference counting 的 ref->refcount 初始化為 1，再使用 strcpy 將字串 str 複製到 ref->data，最後會回傳 ref->data。

#define N_ITERATIONS 100

static void *test_thread(void *arg)
{
    char *str = arg;
    for (int i = 0; i < N_ITERATIONS; i++) {
        char *str2 = refcnt_ref(str);
        fprintf(stderr, "Thread %u, %i: %s\n", (unsigned int) pthread_self(), i,
                str2);
        refcnt_unref(str2);
    }
    refcnt_unref(str);
    return NULL;
}

#define N_THREADS 64

int main(int argc, char **argv)
{
    /* Create threads */
    pthread_t threads[N_THREADS];

    /* Create a new string that is count referenced */
    char *str = refcnt_strdup("Hello, world!");

    /* Start the threads, passing a new counted copy of the referece */
    for (int i = 0; i < N_THREADS; i++)
        pthread_create(&threads[i], NULL, test_thread, refcnt_ref(str));

    /* We no longer own the reference */
    refcnt_unref(str);

    /* Whichever thread finishes last will free the string */
    for (int i = 0; i < N_THREADS; i++)
        pthread_join(threads[i], NULL);

    void *ptr = malloc(100);
    /* This should cause a heap overflow while checking the magic num which the
     * sanitizer checks.
     * Leaving commented out for now
     */
    // refcnt_ref(ptr);

    free(ptr);
    return 0;
}

main 中使用 pthread_create 來建立 N_THREADS 個執行緒，而新建立的執行緒會去調用 test_thread 來開始執行，refcnt_ref(str) 則是作為 test_thread 的參數傳遞。
在 test_thread 中可發現每次做 fprintf 前都會做 refcnt_ref 的動作，在fprintf 完後再做 refcnt_unref 的動作，而在 fprintf 中可發現有使用到 pthread_self() ，其功能為取得執行緒自身的 ID 。
此外， main 中也使用了 pthread_join ，其作用為去等待指定的執行緒執行完畢，如果沒有去使用 pthread_join 可能會造成我們所建立的執行緒沒有執行完的問題。

在 Linux 核心原始程式碼中找出相關 reference counting 的程式碼

在取得 Linux 核心原始程式碼後，可用以下命令去搜尋包含關鍵字 reference count 的 commit。

$ git log --grep="reference count"

可發現在 commit ddaf098 提及 reference count ，而相關的程式碼可以在 drivers/base/class.c 及 include/linux/device/class.h 裡面查看。
根據註解及程式碼推測，subsys_get 的功能與 reference count 的 INCREMENT 有關，而 subsys_put 的功能則與 reference count 的 DECREMENT 有關。
在 lxr 搜尋 subsys_get

static inline struct subsys_private *subsys_get(struct subsys_private *sp)
{
	if (sp)
		kset_get(&sp->subsys);
	return sp;
}

在 subsys_get 中呼叫 kset_get

static inline struct kset *kset_get(struct kset *k)
{
	return k ? to_kset(kobject_get(&k->kobj)) : NULL;
}

在 kset_get 中呼叫 kobject_get

struct kobject *kobject_get(struct kobject *kobj)
{
	if (kobj) {
		if (!kobj->state_initialized)
			WARN(1, KERN_WARNING
				"kobject: '%s' (%p): is not initialized, yet kobject_get() is being called.\n",
			     kobject_name(kobj), kobj);
		kref_get(&kobj->kref);
	}
	return kobj;
}

在 kobject_get 中呼叫 kref_get

/**
 * kref_get - increment refcount for object.
 * @kref: object.
 */
static inline void kref_get(struct kref *kref)
{
	refcount_inc(&kref->refcount);
}

在 kref_get 中會呼叫 refcount_inc 去做 reference count 的 INCREMENT。
由此可見，subsys_get 確實與增加 reference count 有所關聯。

接著搜尋 subsys_put

static inline void subsys_put(struct subsys_private *sp)
{
	if (sp)
		kset_put(&sp->subsys);
}

在 subsys_put 中呼叫 kset_put

static inline void kset_put(struct kset *k)
{
	kobject_put(&k->kobj);
}

在 kset_put 中呼叫 kobject_put

void kobject_put(struct kobject *kobj)
{
	if (kobj) {
		if (!kobj->state_initialized)
			WARN(1, KERN_WARNING
				"kobject: '%s' (%p): is not initialized, yet kobject_put() is being called.\n",
			     kobject_name(kobj), kobj);
		kref_put(&kobj->kref, kobject_release);
	}
}

在 kobject_put 中呼叫 kref_put

/**
 * kref_put - decrement refcount for object.
 * @kref: object.
 * @release: pointer to the function that will clean up the object when the
 *	     last reference to the object is released.
 *	     This pointer is required, and it is not acceptable to pass kfree
 *	     in as this function.
 *
 * Decrement the refcount, and if 0, call release().
 * Return 1 if the object was removed, otherwise return 0.  Beware, if this
 * function returns 0, you still can not count on the kref from remaining in
 * memory.  Only use the return value if you want to see if the kref is now
 * gone, not present.
 */
static inline int kref_put(struct kref *kref, void (*release)(struct kref *kref))
{
	if (refcount_dec_and_test(&kref->refcount)) {
		release(kref);
		return 1;
	}
	return 0;
}

kref_put 中會去做 reference count 的 DECREMENT，而當最後的 reference 被釋放之後，該物件也會被釋放。
由此可見，subsys_put 也確實與減少 reference count 有所關聯。

在 drivers/base/class.c 中 class_to_subsys 的功能是將 struct class 轉換為 struct subsys_private，做這樣的轉換是因為 driver core 內部是需要處理 struct subsys_private，不是外部的 struct class，而在找到匹配的 class 後就會 return 與該 class 相關的內部結構 subsys_private，如果沒有找到匹配的 class 則會 return NULL，而在不為 NULL 的時候，就會呼叫 subsys_get 去做 reference count 的 INCREMENT，因此在使用完畢後必須呼叫 subsys_put() 才能正確地去做釋放的動作。

而 commit ddaf098 所修正的問題就是當呼叫 class_dev_iter_init 函式去初始化 class_dev_iter 的時候使用了 class_to_subsys 去取得 subsys_private 結構並使其 reference count 增加，但當 class_dev_iter 使用完畢後，卻沒有將 subsys_private 的 reference count 做減少的動作，由於這個缺失，漸漸地就會造成 memory leak 的問題。

為何 Linux 核心的同步機制不依賴 reference counting？

參閱 RCU 同步機制，RCU 與 reference counting 的對比如下

	Reference Counting	RCU
Unreclaimed objects	Bounded	Unbounded
Contention among readers	High	None
Traversal forward progress	lock-free	wait-free
Reclamation forward progress	lock-free	blocking
Traversal speed	atomic	no-overhead
Reference acquisition	unconditional	unconditional
Automatic reclamation	yes	no
Purpose of domains	N/A	isolate slow reader

首先，reference counting 的主要問題是 atomic。多個執行緒可能同時增加或減少 reference counting，這就需要使用 atomic operation 來確保計數的一致性。然而，atomic operation 是相對高成本的操作，特別是當多個執行緒同時競爭一個資源時。在高度並行的情況下，可能會導致性能下降。

再來，reference counting 可能會在多個 readers 的情況下導致高度競爭，如果多個執行緒同時增加或減少引用計數，可能會產生競爭，導致計數結果不正確，而為了解決這個問題，需要使用其他同步機制，但也會使得整體的複雜性增加。

另外，當 reference counting 存在 reference cycles (an object which refers directly or indirectly to itself) 的情況時，會導致計數保持非零，無法正確釋放資源。這種情況需要特殊的機制來檢測和處理，以確保資源能夠正確地釋放，但也會增加 reference counting 的成本和複雜性。

基於以上這些性能、競爭、和資源釋放等方面的考慮，這也是為什麼 Linux 核心的同步機制不依賴 reference counting。