之前写过几篇自己动手系列的文章,简要实现了栈,二叉堆,malloc等函数,对于垃圾收集器,一直也有所耳闻。像python中主要使用引用计数手段来管理内存,为了解决循环引用的问题,引入了分代收集和标记-清除方式。当然python中可能产生循环引用的只可能是容器类对象如list,dict,class等,而像int,string是不可能产生循环引用的。当然python中的垃圾收集器实现是比较复杂的,我也没有看过具体代码,昨天在网上看到一篇简单实现垃圾收集器的好文章,于是,翻译到这里,供大家参阅。这篇文章中实现了一个mark-sweep(标记-清除)的垃圾收集器,代码不多,我这里没有照文翻译,只是把核心的实现翻译了过来,原文链接Baby’s First Garbage Collector。
1 Reduce, reuse, recycle
假想一台机器有无限的内存,这样开发人员只需要不停的分配内存即可,不用考虑回收内存的的问题。当然,理想是丰满的,现实是骨感的。机器没有无限内存,当你在分配了许多内存后,程序运行慢下来了,需要考虑回收垃圾了。
在本文上下文中,垃圾就是指的就是之前分配过的现在不再使用的内存。为了让程序能够在有限的内存里面工作,需要保证“不再使用”是非常安全的,一旦准备收集垃圾,这些待收集对象一定要能够保证不再使用,也不能通过任何方式引用到。我们要收集的是“不再使用”的对象,这里先给出”在使用”的定义(注:因为我们是要标记在使用的对象,然后清除没有标记的对象即不再使用的对象):
- 1任何在代码范围内被变量引用的对象是在使用中的。
- 2任何被其他在使用中的对象引用的对象是在使用中的。
第2条规则是递归的。如果对象A被一个变量引用,且对象A有字段引用了对象B,那么对象B是使用中的,因为你可以从A引用到B。
最终的结果就是一个可达对象图(reachable objects graph)-那些你可以通过变量遍历到的对象。而任何不在可达对象集合中的对象都已经成为了垃圾,它们占用的内存可以被回收了。
2 Marking and sweeping
回收不再使用的对象有很多种方法,不过最早和最简单的算法就是标记-清除法。发明该方法的人是John McCarthy大牛,他还发明了Lisp语言。
标记-清除方法的流程很简单:
- 从根出发,遍历可达对象图,每次遇到一个可达对象,设置对象标记为true。
- 然后,遍历目前分配的所有对象,如果该对象标记没有设置为true,则该对象不可达,需要删除来释放内存。
3 A pair of objects
在实现标记-清除算法之前,先来定义几个对象。假想我们是在实现一个语言解释器,该语言只有两种对象类型,因此,我们用枚举类型定义对象类型ObjectType如下:
typedef enum {
OBJ_INT,
OBJ_PAIR
} ObjectType;
OBJ_INT是一个整数类型对象,而OBJ_PAIR是一个pair对象,它可以包含两个整数,也可以是一个整数和一个pair,因为我们定义中只有这两种对象类型,要么int要么pair,因此采用union类型来定义对象是十分合适的,定义如下:
typedef struct sObject {
ObjectType type;
union {
/* OBJ_INT */
int value;
/* OBJ_PAIR */
struct {
struct sObject* head;
struct sObject* tail;
};
};
} Object;
其中type字段说明了对象类型,而union对象用于存储对象内容,要么是int的值,或者是一个pair结构体。
4 A minimal virtual machine
接下来实现一个虚拟机对象,它的角色就是用栈来存储我们当前范围内的对象。许多语言的虚拟机都是基于栈的,如JVM和CLR,也有基于寄存器的,如Lua。在我们的例子中,采用的是基于栈的,它用于存储局部变量和临时变量。代码如下:
#define STACK_MAX 256
typedef struct {
Object* stack[STACK_MAX];
int stackSize;
} VM;
有了虚拟机的数据结构,我们看下创建虚拟机以及初始化的代码,以及操作虚拟机的代码:
VM* newVM() {
VM* vm = malloc(sizeof(VM));
vm->stackSize = 0;
return vm;
}
void push(VM* vm, Object* value) {
assert(vm->stackSize < STACK_MAX, "Stack overflow!");
vm->stack[vm->stackSize++] = value;
}
Object* pop(VM* vm) {
assert(vm->stackSize > 0, "Stack underflow!");
return vm->stack[--vm->stackSize];
}
有了虚拟机以及虚拟机操作,现在可以来创建对象了,相关代码如下:
Object* newObject(VM* vm, ObjectType type) {
Object* object = malloc(sizeof(Object));
object->type = type;
return object;
}
创建对象后,需要压入到虚拟机的栈中,由于有两种不同对象int和pair,因此有两个不同函数,实现代码如下:
void pushInt(VM* vm, int intValue) {
Object* object = newObject(vm, OBJ_INT);
object->value = intValue;
push(vm, object);
}
Object* pushPair(VM* vm) {
Object* object = newObject(vm, OBJ_PAIR);
object->tail = pop(vm);
object->head = pop(vm);
push(vm, object);
return object;
}
5 Marky mark
为了实现标记,我们需要在之前的Object定义中加一个marked字段,用于标识该对象是否可达。修改后定义如下:
typedef struct sObject {
unsigned char marked; //新增的标记字段
ObjectType type;
union {
/* OBJ_INT */
int value;
/* OBJ_PAIR */
struct {
struct sObject* head;
struct sObject* tail;
};
};
} Object;
每当我们创建一个对象,我们修改newObject()函数设置marked为0。为了标记所有可达对象,我们需要遍历可达对象栈。代码如下:
void markAll(VM* vm)
{
for (int i = 0; i < vm->stackSize; i++) {
mark(vm->stack[i]);
}
}
在markAll中我们调用了mark函数,它的实现如下:
void mark(Object* object) {
object->marked = 1;
}
需要注意到的是,对象可达是递归的,因为我们还有pair类型的对象,因此mark函数修改如下:
void mark(Object* object) {
/* If already marked, we‘re done. Check this first
to avoid recursing on cycles in the object graph. */
if (object->marked) return;
object->marked = 1;
if (object->type == OBJ_PAIR) {
mark(object->head);
mark(object->tail);
}
}
6 Sweepy sweep
标记完之后,下一步就是清除那些没有被标记的对象(也就是不可达对象)。但是现在有个问题是,我们找不到这些不可达的对象。
因此,我们还需要跟踪对象。最简单的方法就是在对象中加入一个链表用于跟踪我们分配过的对象,虚拟机结构中保存链表头指针,因此,Object和VM定义修改如下:
typedef struct sObject {
/* The next object in the list of all objects. */
struct sObject* next; //新增链表指针
unsigned char marked;
ObjectType type;
union {
/* OBJ_INT */
int value;
/* OBJ_PAIR */
struct {
struct sObject* head;
struct sObject* tail;
};
};
} Object;
typedef struct {
/* The first object in the list of all objects. */
Object* firstObject; //新增对象链表头
Object* stack[STACK_MAX];
int stackSize;
} VM;
创建对象的时候,需要将其加入到对象链(插入到头部),同时更新虚拟机的链表头指针的值。
Object* newObject(VM* vm, ObjectType type) {
Object* object = malloc(sizeof(Object));
object->type = type;
object->marked = 0;
/* Insert it into the list of allocated objects. */
object->next = vm->firstObject;
vm->firstObject = object;
return object;
}
接下来,我们就可以来清除不可达对象了。先标记可达对象,然后遍历对象链表,如果对象没有标记,则清除,有标记,则去掉标记,并从下一个对象接着遍历。
void sweep(VM* vm)
{
Object** object = &vm->firstObject;
while (*object) {
if (!(*object)->marked) {
/* This object wasn‘t reached, so remove it from the list
and free it. */
Object* unreached = *object;
*object = unreached->next;
free(unreached);
} else {
/* This object was reached, so unmark it (for the next GC)
and move on to the next. */
(*object)->marked = 0;
object = &(*object)->next;
}
}
}
至此,垃圾收集器基本完成了。首先来组合一下标记和清除这两个函数。那么你可能会有疑惑的是我们应该在什么时候调用gc这个函数呢?这里我们采用一种简单的策略,只要分配对象超过了我们设定的最大数目就调用gc。为此,我们需要在VM中加入两个变量来记录已分配的对象数目和最大对象数目。
void gc(VM* vm) {
markAll(vm);
sweep(vm);
}
typedef struct {
/* The total number of currently allocated objects. */
int numObjects;
/* The number of objects required to trigger a GC. */
int maxObjects;
/* Previous stuff... */
} VM;
VM* newVM() {
/* Previous stuff... */
vm->numObjects = 0;
vm->maxObjects = INITIAL_GC_THRESHOLD;
return vm;
}
代码中的INITIAL_GC_THRESHOLD就是我们首次开启gc的对象数目,这个值可以根据内存大小进行调整。因此,在newObject函数中,如果分配对象大于了最大对象数目,我们需要运行gc来清除不可达的对象。
Object* newObject(VM* vm, ObjectType type) {
if (vm->numObjects == vm->maxObjects) gc(vm);
/* Create object... */
vm->numObjects++;
return object;
}
此外,我们在每次gc之后,会更新下最大对象数。修改后的gc函数代码如下:
void gc(VM* vm) {
int numObjects = vm->numObjects;
markAll(vm);
sweep(vm);
vm->maxObjects = vm->numObjects * 2;
}
完整版示例代码:https://github.com/munificent/mark-sweep
7 译者注
个人觉得容易错的是sweep函数,代码中使用了指向指针的指针来遍历链表,确实便捷很多。另外,在每次遍历时,如果对象可达,则需要设置mark标记为0并将遍历指针指向下一个对象.这里设置可达对象的mark标记为0十分重要,因为你每次sweep之前都会markAll,如果这里不标记为0,那么后续如果这个对象确实不可达了,由于mark标记没有复位为0,则以后都会收不到了。
另外,一定要看下完整代码,里面有完整的实例,可以解决你在看本文中的一些疑惑。