引用计数与对象树
cheungmine
2013-12-28
0 引言
我们经常在C语言中，用指针指向一个对象（Object）的结构，也称为句柄（Handle），利用不透明指针的技术把结构数据封装成对象，因此如果说在Java中，一切皆是对象的话，那么在C中，万物皆是指针，这么说是不过分的。
然而，C并没有提供垃圾回收等自动化的内存管理设施，我们需要对每一个创建（malloc）出来的对象调用（free），任何时候遗漏了free，或者多调用了一次free，都将造成不可挽回的损失。这也是很多程序员担惊受怕的地方。任何一个C语言的程序员可能都做过的梦魇就是访问了一个已经释放的指针（野指针）而造成程序崩溃。然而，这还不是最糟糕的情况。
当指针被跨线程访问，A线程创建了指针p，然后把p传递给线程B，然后A释放了p，可以想象此时线程B会发生情况。
struct OBJ
{
int data;
};
void threadB(struct OBJ *p)
{
printf("data=%d", p->data);
}
void threadA()
{
struct OBJ *p = (struct OBJ*) malloc(sizeof(*p));
p->data = 10101;
threadB(p);
free(p);
}
上面的代码并不会出问题，是因为threadA与threadB是同步执行的。
在C语言的多线程程序中，当一个线程获得了一个对象指针，我们能期待别的线程不会释放这个指针么？我们如何判断这个指针仍然是有效的？这时候这个指针的确像烫手的山芋，我们甚至不能剥开它的一点表皮来一探究竟。是的，如果在多线程环境下传递一个对象指针，这个指针就像烧红的木炭，你碰不得它，除非你指望别的线程能规范地使用这个指针。
一个简单的解决办法是给指针上一个锁（Mutex或Lock）。这情景可以想象指针就是医院妇科门诊室里的医生，同时只能有一个妇女看病，当病人甲进入诊室，门就锁上了，等她看完病出来，锁被打开，病人乙才可以进去。如果没有锁会怎么样？嘿嘿！
这样，一个指针要搭一把锁，1万个指针要1万把锁。这不是问题，问题的关键是我们传递对象的时候，一定要同时传递对应的锁。这个管理成本太大了，因为如果封装了这个指针和相应的锁，那么封装之后的对象又成为一个指针，又需要锁才可以访问，所以这是无解的问题。
更没可能是给所有指针共用一把锁，那样医院妇科门诊室里医生肯定清闲了，急坏了看病的妇女们。
回到问题的核心，我们想要构建这样的对象（指针）系统，在这个系统中，我们获得（Retain）了一个对象，我们就可以自信地访问这个对象，或者抛给别的访问者（线程），当我们访问结束，我们不需要同步等待其他访问者，我们直接撒手放开（Release）对象，而不会影响其他访问者，更不会造成内存泄露。于是我们能想象到一个名词：引用计数。
1 引用计数
是的，引用计数可以解决这个复杂的问题。对象内部维护一个计数器（不能小于0），当计数器是0的时候，对象销毁。当线程A创建一个对象o，o的引用计数=1，如果我们想把这个o传递给线程B，首先是给o的引用计数+1，此时o的引用计数=2，然后传递给线程B，B在使用完o之后，将o的引用计数-1，则此时o的引用计数=1。线程A任何时候释放o。任何时候当o的引用计数为o，对象自动释放。
struct OBJ
{
int refcount; // 引用计数
int data; // 其他数据
};
void UseObject(const char *user, struct OBJ *p)
{
printf("user=%s:\r\n refcount=%d\r\n data=%d\r\n",
user, p->refcount, p->data);
}
// 创建对象
struct OBJ * Create(int initData)
{
struct OBJ *p = (struct OBJ*) malloc(sizeof(*p));
if (p) {
// 创建成功,引用计数=1
p->refcount=1;
p->data = initData;
printf("Create OBJ.refcount=%d\r\n", p->refcount);
}
return p;
}
// 销毁对象,私有函数,用户永远不要直接调用这个函数
void internal_Free(struct OBJ *p)
{
printf("Free OBJ\r\n");
free(p);
}
struct OBJ * Retain(struct OBJ **pp)
{
struct OBJ *p = *pp;
if (p) {
p->refcount++;
}
return p;
}
void Release(struct OBJ **pp)
{
struct OBJ *p = *pp;
if (p) {
if (0 == --p->refcount) {
// 如果引用计数为0, 销毁对象
printf("OBJ.refcount=%d\r\n", p->refcount);
internal_Free(p);
*pp = 0;
}
}
}
void threadB(struct OBJ *p)
{
UseObject("threadB", p);
// 使用完必须释放
Release(&p);
}
void threadA()
{
// 创建对象o
struct OBJ *o = Create(350137278);
// 使用对象o
UseObject("threadA", o);
// 给对象o增加引用, 然后传递给其他线程(使用者)
threadB(Retain(&o));
// 线程A使用完对象o,释放它,不要管其他线程是否还在使用
Release(&o);
}
void main()
{
threadA();
}
上面的Release函数使用struct OBJ **，这样保证对象因为引用计数为0导致析构（Free）之后，指针o=NULL。
增加和减小引用计数并不像上面的代码那么简单，这个的确需要为refcount上锁，在Windows平台上有一对函数可以达到这个原子操作：InterlockedIncrement和InterlockedDecrement。在Linux平台上，__sync_sub_and_fetch和__sync_sub_and_fetch干类似的活。
在C++世界里，引用计数和所谓智能指针是共生的。如果对boost不陌生的话，那么智能指针shared_ptr就是封装了引用计数的实现。一个对象的裸指针ptr传递给shared_ptr，此时引用计数就增加1。shared_ptr在超出作用域时，由于C++栈上的类（shared_ptr）会自动析构，此时引用计数就减少1，如果引用计数为0，shared_ptr就调用delete ptr。
很显然，在C语言中，引用计数属于对象自身存储的一部分，需要我们写函数小心地维护，而在C++中，辅助类（shared_ptr）可以很好地帮助我们封装引用计数。C++把这种栈上的类会自动析构的特性玩弄得淋漓尽致。
2 对象树
上面的代码对于单个对象似乎可以工作的很好了，但是，如果对于对象系统，很多对象构成对象树——我们也称为对象模型，事情会变得稍微有点复杂，光有shared_ptr还是不够用的。
一个对象树有父和若干子对象组成，子对象还可以作为父对象拥有更多的孙对象等等。父对象持有儿女对象的指针，这个属于shared_ptr，因为没有父，就没有子，通过父对象访问子对象几乎是天经地义的。但是子对象也应该可以访问到父对象，这也是正常的。好像一颗树，给定任何一个树枝，我们不但能到达叶子，也能到达主干甚至树根。
父对象必须拥有儿女对象，也就是包含儿女的shared_ptr。但是儿女对象不能包含父对象的shared_ptr。因为如果儿女对象拥有了父对象的shared_ptr，将导致父对象引用计数增加，则产生效率问题和循环引用计数问题。如果儿女对象拥有父对象，而不增加父对象引用计数，那么父对象销毁（引用计数=0）后，子对象访问父对象将产生异常（对悬空指针访问）。许多情况下，为了防止递归的依赖关系，就要旁观一个共享资源而不能拥有所有权，或者为了避免悬空指针，这就是weak_ptr。
shared_ptr就是和weak_ptr对于创建对象树系统，是非常有用处的。父对象持有（未成年）儿女对象的指针，这个属于shared_ptr，因为没有父，就没有子；儿女对象持有父对象的指针，这个属于weak_ptr，因为父亲可以不存在，然而儿女依然存在。通过父对象可以得到子对象，这个很自然。通过子对象的weak_ptr，仍然可以得到父对象的shared_ptr。
weak_ptr 是 shared_ptr 的观察员。它不会干扰shared_ptr所共享的所有权。当一个被weak_ptr所观察的 shared_ptr 要释放它的资源时，它会把相关的 weak_ptr的指针设为空。这防止了 weak_ptr 持有悬空的指针。
这是C++中的概念，在C中，子对象持有父对象的weak_ptr，相当于持有父指针的地址，即指向父指针的指针。如下面的代码：
// 父对象
struct PARENT
{
int refcount; // 引用计数
int name; // 其他数据
int numberChilds;
struct CHILD **pChildList;
};
// 子对象
struct CHILD
{
struct PARENT **ppParent;
int refcount; // 引用计数
int name; // 其他数据
};
struct CHILD * CreateChild(struct PARENT **ppv, int name)
{
struct CHILD *p = (struct CHILD*) malloc(sizeof(*p));
p->refcount = 1;
p->name = name;
p->ppParent = ppv; // 子对象保持父对象的weak_ptr
return p;
}
struct PARENT * CreateParent(int numberChilds, int name)
{
struct CHILD *chld;
struct PARENT *p = (struct PARENT*) malloc(sizeof(*p));
// 创建子对象集合
p->pChildList = (struct CHILD**) malloc(sizeof(*chld)*numberChilds);
for (p->numberChilds=0; p->numberChilds<numberChilds; p->numberChilds++) {
// 创建子对象
chld = CreateChild(&p, p->name + p->numberChilds);
p->pChildList[p->numberChilds] = chld;
}
p->name = name;
p->refcount = 1;
return p;
}
struct PARENT * RetainParent(struct PARENT **pp)
{
struct PARENT *p = *pp;
if (p) {
p->refcount++;
}
return p;
}
struct CHILD * RetainChild(struct CHILD **pp)
{
struct CHILD *p = *pp;
if (p) {
p->refcount++;
}
return p;
}
struct CHILD * ParentRetainChild(struct PARENT *p, int index)
{
struct CHILD *chd;
assert(index>=0 && index<p->numberChilds);
chd = p->pChildList[index];
return RetainChild(&chd);
}
// 销毁对象,私有函数,用户永远不要直接调用这个函数
void internal_FreeChild(struct CHILD *p)
{
printf("Free CHILD\r\n");
// 如果子对象包含孙对象, 需要在此处释放
// ReleaseGrandSon(...);
free(p);
}
void ReleaseChild(struct CHILD **pp)
{
struct CHILD *p = *pp;
if (p) {
if (0 == --p->refcount) {
// 如果引用计数为0, 销毁子对象
printf("CHILD.refcount=%d\r\n", p->refcount);
internal_FreeChild(p);
*pp = 0;
}
}
}
// 销毁对象,私有函数,用户永远不要直接调用这个函数
void internal_FreeParent(struct PARENT *p)
{
printf("Free PARENT\r\n");
while (p->numberChilds-->0) {
// 释放子对象, 不是删除
struct CHILD *chd = p->pChildList[p->numberChilds];
ReleaseChild(&chd);
}
free(p->pChildList);
free(p);
}
void ReleaseParent(struct PARENT **pp)
{
struct PARENT *p = *pp;
if (p) {
if (0 == --p->refcount) {
// 如果引用计数为0, 销毁父对象
printf("PARENT.refcount=%d\r\n", p->refcount);
internal_FreeParent(p);
*pp = 0;
}
}
}
void main()
{
// 线程1: 创建父亲
p = CreateParent(3, 1000);
// 线程1: 取得父对象，传递给线程2
p2 = RetainParent(&p);
// 线程1: 使用父对象结束，释放之
ReleaseParent(&p);
// 线程2: 得到父对象，然后产生3个孩子
ch0 = ParentRetainChild(p2, 0);
ch1 = ParentRetainChild(p2, 1);
ch2 = ParentRetainChild(p2, 2);
// 线程2: 释放父对象
ReleaseParent(&p2);
// 线程2: 使用3个孩子，然后释放孩子
// UseChilds(...);
ReleaseChild(&ch0);
ReleaseChild(&ch1);
ReleaseChild(&ch2);
}
3 对于引用对象的一个封装代码
下面的代码试图把引用对象的一般操作提取出来，这样所有的对象（object）都包含一个公共的头部，我称之为RefHandleType，对这个RefHandle的访问将是一致的。下面的代码不保证可以运行，但是稍加改动，就可以使之为我们所用。
/**
* refhandle.h
* @brief
* RefHandle API
* @author
* ZhangLiang
* @since
* 2013-12-26
* @date
* 2013-12-26
*/
#ifndef REF_HANDLE_H_INCLUDED
#define REF_HANDLE_H_INCLUDED
#ifdef __cplusplus
extern "C" {
#endif
#ifdef _MSC_VER
#pragma warning (disable : 4996)
#endif
#if defined _MSC_VER || WIN32
#ifndef OS_PLATFORM_WIN
#define OS_PLATFORM_WIN
#endif
#endif
#ifdef OS_PLATFORM_WIN
#include <windows.h>
#include <process.h>
#else
#include <pthread.h>
#include <unistd.h>
#endif
/**
* ref count type
*/
#ifdef OS_PLATFORM_WIN
typedef volatile unsigned long ref_count_t;
#define __interlock_inc(add) InterlockedIncrement(add)
#define __interlock_dec(sub) InterlockedDecrement(sub)
#else
typedef volatile size_t ref_count_t;
#define __interlock_inc(add) __sync_add_and_fetch(add, 1)
#define __interlock_dec(sub) __sync_sub_and_fetch(sub, 1)
#endif
/**
* thread lock
*/
#ifdef OS_PLATFORM_WIN
typedef CRITICAL_SECTION thr_mutex_t;
#else
typedef pthread_mutex_t thr_mutex_t;
#endif
/**
* thread lock
*/
static int __thrmutex_init (thr_mutex_t* lock)
{
int ret = 0;
#ifdef OS_PLATFORM_WIN
InitializeCriticalSection(lock);
#else
/* Linux */
pthread_mutexattr_t attr;
ret = pthread_mutexattr_init(&attr);
if (ret == 0) {
/* PTHREAD_MUTEX_RECURSIVE_NP ? */
ret = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
if (ret == 0) {
ret = pthread_mutex_init(lock, &attr);
}
pthread_mutexattr_destroy(&attr);
}
#endif
return ret;
}
static void __thrmutex_uninit (thr_mutex_t* lock)
{
#ifdef OS_PLATFORM_WIN
DeleteCriticalSection(lock);
#else
pthread_mutex_lock(lock); /* do we need this? */
pthread_mutex_destroy(lock);
#endif
}
static void __thrmutex_lock (thr_mutex_t* lock)
{
#ifdef OS_PLATFORM_WIN
EnterCriticalSection(lock);
#else
pthread_mutex_lock(lock);
#endif
}
static int __thrmutex_trylock (thr_mutex_t* lock)
{
#ifdef OS_PLATFORM_WIN
return TryEnterCriticalSection(lock)? 0 : (-1);
#else
return pthread_mutex_trylock(lock);
#endif
}
static void __thrmutex_unlock (thr_mutex_t* lock)
{
#ifdef OS_PLATFORM_WIN
LeaveCriticalSection(lock);
#else
pthread_mutex_unlock(lock);
#endif
}
#define OBJECT_INVALID (-1)
#define EINDEX (-1)
#define REF_HANDLE_CAST(object,p) \
RefHandleType * (##p) = (RefHandleType *)(((char*)object)-sizeof(*##p))
/**
* Callback when object just before free self.
* Used by parent to release its children.
*/
typedef void (* ReleaseFreeDataCallback) (void *handle, void *param);
typedef struct _RefHandleType
{
ref_count_t __refcount;
/* DO NOT CHANGE THIS AFTER CREATION */
int __htype;
thr_mutex_t __lock;
volatile size_t __id;
void *__handle[0];
} RefHandleType, *RefHandle;
static void* _RefHandleCreate (int htype, size_t cbObjectSize)
{
char *hdl;
RefHandleType *p = (RefHandleType*) malloc(sizeof(*p) + cbObjectSize);
hdl = (char*) p;
hdl += sizeof(*p);
p->__refcount = 1L;
p->__htype = htype;
p->__id = EINDEX;
__thrmutex_init(&p->__lock);
return (void*) hdl;
}
static long _RefHandleRetain (void *object)
{
if (object) {
REF_HANDLE_CAST(object, p);
return __interlock_inc(&p->__refcount);
} else {
return 0;
}
}
static void _RefHandleRelease (void **ppObject, ReleaseFreeDataCallback freeDataFunc, void *param)
{
void *object = *ppObject;
if (object) {
REF_HANDLE_CAST(object, p);
if (0 == __interlock_dec(&p->__refcount)) {
if (freeDataFunc) {
freeDataFunc(object, param);
}
__thrmutex_uninit(&p->__lock);
free((void**) &p);
*ppObject = 0;
}
}
}
static int _RefHandleGetType (void *object)
{
if (object) {
REF_HANDLE_CAST(object, p);
return p->__htype;
} else {
return OBJECT_INVALID;
}
}
static thr_mutex_t * _RefHandleGetLock (void *object)
{
if (object) {
REF_HANDLE_CAST(object, p);
return &p->__lock;
} else {
return (thr_mutex_t*) 0;
}
}
static size_t _RefHandleGetId (void *object)
{
if (object) {
REF_HANDLE_CAST(object, p);
return p->__id;
} else {
return EINDEX;
}
}
static size_t _RefHandleSetId(void *object, size_t newId)
{
if (object) {
size_t oldId;
REF_HANDLE_CAST(object, p);
oldId = p->__id;
p->__id = newId;
return oldId;
} else {
return EINDEX;
}
}
#ifdef __cplusplus
}
#endif
#endif /* REF_HANDLE_H_INCLUDED */
4 结束语
上面代码说明了如何用C语言实现一个完整的跨线程调用的安全的对象系统。代码中对于每个对象的Retain和Release都可以提炼出共有部分，类似shaped_ptr，C语言用结构和不透明指针，可以写出十分精炼优美、高效健壮的架构代码。即使是一个复杂对象树，包含更多的层次，实现起来也并不复杂，不会有任何异常和内存泄露！这也许就是架构之美吧。然而美文要和懂得美的人共同欣赏，不可对牛弹琴。对于那些奉行“UIUE是核心竞争力”的人来说，“架构不重要，干嘛搞那么复杂？”，于是等待他们的将是最后一刻的突然崩溃。