Vue 3 的Composition API + ``` ``` 這就把清單功能獨立出來,可在任意需要的地方復用。 基於組件去搭建應用,可實現對業務邏輯的復用。如有其他頁面也需要用到這功能,直接復用。 然後,就可基於新語法實現清單應用。 把之前的代碼移植過來後,使用ref包裹的響應式數據。修改tit ...
Python版本
v3.9.17
分析代碼的過程比較枯燥,可以直接跳轉到總結。
只能被其他對象引用類型
比如:longobject、floatobject
floatobject
以floatobject為例子來分析,先看看結構定義
typedef struct {
PyObject_HEAD
double ob_fval;
} PyFloatObject;
// 展開PyObject_HEAD後
typedef struct {
PyObject ob_base;
double ob_fval;
} PyFloatObject;
typedef struct _object {
_PyObject_HEAD_EXTRA
Py_ssize_t ob_refcnt;
PyTypeObject *ob_type;
} PyObject;
在PyObject中的_PyObject_HEAD_EXTRA
,只有在編譯時指定--with-trace-refs
才有效,這裡忽略即可。
./configure --with-trace-refs
可以看到在PyObject里有一個ob_refcnt
的屬性,這個就是引用計數。
當對引用計數減為0時,就會調用各類型對應的析構函數。
define Py_DECREF(op) _Py_DECREF(_PyObject_CAST(op))
void _Py_Dealloc(PyObject *op)
{
destructor dealloc = Py_TYPE(op)->tp_dealloc;
(*dealloc)(op);
}
static inline void _Py_DECREF(PyObject *op)
{
if (--op->ob_refcnt != 0) {
}
else {
_Py_Dealloc(op);
}
}
能引用其他對象的類型
比如listobject,dictobject...
listobject
以listobject為例子來分析,先看看結構定義
typedef struct {
PyObject_VAR_HEAD
PyObject **ob_item;
Py_ssize_t allocated;
} PyListObject;
// 展開 PyObject_VAR_HEAD
typedef struct {
PyVarObject ob_base;
PyObject **ob_item;
Py_ssize_t allocated;
} PyListObject;
typedef struct {
PyObject ob_base;
Py_ssize_t ob_size; /* Number of items in variable part */
} PyVarObject;
可以看出,PyObject_VAR_HEAD
也就比PyObject_HEAD
多了一個Py_ssize_t ob_size
而已,這個屬性是用來表示這個可變對象里元素數量。
因為可以引用其他對象,就有可能會出現環引用問題,這種問題如果再使用引用計數來作為GC就會出現問題。
lst1 = []
lst2 = []
lst1.append(lst2)
lst2.append(lst1)
當然這種情況可以使用弱引用,或者手動解除環引用。這些解決方案這裡不深入,現在主要看看python是怎樣應對這種情況。
對於這類型的對象在申請記憶體的時候調用的是PyObject_GC_New
,而不可變類型是用PyObject_MALLOC
。為了減少篇幅,刪掉了一些判斷邏輯。
typedef struct {
// Pointer to next object in the list.
// 0 means the object is not tracked
uintptr_t _gc_next;
// Pointer to previous object in the list.
// Lowest two bits are used for flags documented later.
uintptr_t _gc_prev;
} PyGC_Head;
#define FROM_GC(g) ((PyObject *)(((PyGC_Head *)g)+1))
static PyObject * _PyObject_GC_Alloc(int use_calloc, size_t basicsize)
{
PyThreadState *tstate = _PyThreadState_GET();
GCState *gcstate = &tstate->interp->gc;
size_t size = sizeof(PyGC_Head) + basicsize;
PyGC_Head *g;
g = (PyGC_Head *)PyObject_Malloc(size);
g->_gc_next = 0;
g->_gc_prev = 0;
gcstate->generations[0].count++; /* number of allocated GC objects */
if (/* 判斷是否可以執行GC */)
{
gcstate->collecting = 1;
collect_generations(tstate);
gcstate->collecting = 0;
}
PyObject *op = FROM_GC(g);
return op;
}
在可變對象中,python又加上了一個PyGC_Head
。通過這個PyGC_Head
將listobject鏈接到gc列表中。
在分配完listobject記憶體後,緊接著調用_PyObject_GC_TRACK
,鏈接到gc列表中。
static inline void _PyObject_GC_TRACK_impl(const char *filename, int lineno,
PyObject *op)
{
PyGC_Head *gc = _Py_AS_GC(op);
PyThreadState *tstate = _PyThreadState_GET();
PyGC_Head *generation0 = tstate->interp->gc.generation0;
PyGC_Head *last = (PyGC_Head*)(generation0->_gc_prev);
_PyGCHead_SET_NEXT(last, gc);
_PyGCHead_SET_PREV(gc, last);
_PyGCHead_SET_NEXT(gc, generation0);
generation0->_gc_prev = (uintptr_t)gc;
}
通過這裡的變數名,可以猜測使用到了分代垃圾回收。
分代回收
python手動執行垃圾回收一般調用gc.collect(generation=2)函數。
#define NUM_GENERATIONS 3
#define GC_COLLECT_METHODDEF \
{"collect", (PyCFunction)(void(*)(void))gc_collect, METH_FASTCALL|METH_KEYWORDS, gc_collect__doc__},
static PyObject *
gc_collect(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
{
PyObject *return_value = NULL;
int generation = NUM_GENERATIONS - 1;
Py_ssize_t _return_value;
_return_value = gc_collect_impl(module, generation);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyLong_FromSsize_t(_return_value);
exit:
return return_value;
}
具體執行在gc_collect_impl
函數中,接著往下
static Py_ssize_t gc_collect_impl(PyObject *module, int generation)
{
PyThreadState *tstate = _PyThreadState_GET();
GCState *gcstate = &tstate->interp->gc;
Py_ssize_t n;
if (gcstate->collecting) {
/* already collecting, don't do anything */
n = 0;
}
else {
gcstate->collecting = 1;
n = collect_with_callback(tstate, generation);
gcstate->collecting = 0;
}
return n;
}
可以看到,如果已經在執行GC,則直接返回。接著看collect_with_callback
static Py_ssize_t
collect_with_callback(PyThreadState *tstate, int generation)
{
assert(!_PyErr_Occurred(tstate));
Py_ssize_t result, collected, uncollectable;
invoke_gc_callback(tstate, "start", generation, 0, 0);
result = collect(tstate, generation, &collected, &uncollectable, 0);
invoke_gc_callback(tstate, "stop", generation, collected, uncollectable);
assert(!_PyErr_Occurred(tstate));
return result;
}
其中invoke_gc_callback
是調用通過gc.callbacks
註冊的回調函數,這裡我們忽略,重點分析collect
函數。
collect函數簽名
這段代碼很長,我們拆分開來分析,這裡會去除掉一些DEBUG相關的邏輯。
static Py_ssize_t collect(PyThreadState *tstate, int generation,Py_ssize_t *n_collected, Py_ssize_t *n_uncollectable, int nofail);
- 將新生代的對象合併到指定代的對象列表中。
/* merge younger generations with one we are currently collecting */
for (i = 0; i < generation; i++) {
gc_list_merge(GEN_HEAD(gcstate, i), GEN_HEAD(gcstate, generation));
}
比如調用gc.collect(2)
,就表示啟動全部的垃圾回收。這裡就會將第0、1代的對象合併到第2代上。合併之後第0、1代上就空了,全部可GC的對象都在第2代上。
- 推斷不可達對象
/* handy references */
young = GEN_HEAD(gcstate, generation);
if (generation < NUM_GENERATIONS-1)
old = GEN_HEAD(gcstate, generation+1);
else
old = young;
validate_list(old, collecting_clear_unreachable_clear);
deduce_unreachable(young, &unreachable);
這裡的young指針指向第2代的鏈表頭,validate_list
做校驗,這裡忽略,重點在deduce_unreachable
函數中。
static inline void
deduce_unreachable(PyGC_Head *base, PyGC_Head *unreachable) {
validate_list(base, collecting_clear_unreachable_clear);
update_refs(base); // gc_prev is used for gc_refs
subtract_refs(base);
gc_list_init(unreachable);
move_unreachable(base, unreachable); // gc_prev is pointer again
validate_list(base, collecting_clear_unreachable_clear);
validate_list(unreachable, collecting_set_unreachable_set);
}
首先調用update_refs更新引用計數
static inline void
gc_reset_refs(PyGC_Head *g, Py_ssize_t refs)
{
g->_gc_prev = (g->_gc_prev & _PyGC_PREV_MASK_FINALIZED)
| PREV_MASK_COLLECTING
| ((uintptr_t)(refs) << _PyGC_PREV_SHIFT);
}
static void
update_refs(PyGC_Head *containers)
{
PyGC_Head *gc = GC_NEXT(containers);
for (; gc != containers; gc = GC_NEXT(gc)) {
gc_reset_refs(gc, Py_REFCNT(FROM_GC(gc)));
_PyObject_ASSERT(FROM_GC(gc), gc_get_refs(gc) != 0);
}
}
這裡的邏輯就是遍歷所有對象,然後賦值_gc_prev,設置為收集中的標識PREV_MASK_COLLECTING,然後將引用計數賦值給_gc_prev 。最後_gc_prev的內容如下。
更新完_gc_prev後,就開始調用subtrace_refs,遍歷對象中的元素,判斷元素是否也是可GC對象並且有收集中標記,如果是則減去該對象的計數。註意這裡減去的是_gc_prev中的計數,而不是真正的計數ob_refcnt。
static int
visit_decref(PyObject *op, void *parent)
{
_PyObject_ASSERT(_PyObject_CAST(parent), !_PyObject_IsFreed(op));
if (_PyObject_IS_GC(op)) {
PyGC_Head *gc = AS_GC(op);
/* We're only interested in gc_refs for objects in the
* generation being collected, which can be recognized
* because only they have positive gc_refs.
*/
if (gc_is_collecting(gc)) {
gc_decref(gc);
}
}
return 0;
}
static void
subtract_refs(PyGC_Head *containers)
{
traverseproc traverse;
PyGC_Head *gc = GC_NEXT(containers);
for (; gc != containers; gc = GC_NEXT(gc)) {
PyObject *op = FROM_GC(gc);
traverse = Py_TYPE(op)->tp_traverse;
(void) traverse(FROM_GC(gc),
(visitproc)visit_decref,
op);
}
}
更新計數值之後,就開始收集不可達對象,將對象移入到不可達列表中。unreachable。
/* A traversal callback for move_unreachable. */
static int
visit_reachable(PyObject *op, PyGC_Head *reachable)
{
if (!_PyObject_IS_GC(op)) {
return 0;
}
PyGC_Head *gc = AS_GC(op);
const Py_ssize_t gc_refs = gc_get_refs(gc);
if (! gc_is_collecting(gc)) {
return 0;
}
assert(gc->_gc_next != 0);
if (gc->_gc_next & NEXT_MASK_UNREACHABLE) {
PyGC_Head *prev = GC_PREV(gc);
PyGC_Head *next = (PyGC_Head*)(gc->_gc_next & ~NEXT_MASK_UNREACHABLE);
_PyObject_ASSERT(FROM_GC(prev),
prev->_gc_next & NEXT_MASK_UNREACHABLE);
_PyObject_ASSERT(FROM_GC(next),
next->_gc_next & NEXT_MASK_UNREACHABLE);
prev->_gc_next = gc->_gc_next; // copy NEXT_MASK_UNREACHABLE
_PyGCHead_SET_PREV(next, prev);
gc_list_append(gc, reachable);
gc_set_refs(gc, 1);
}
else if (gc_refs == 0) {
gc_set_refs(gc, 1);
}
else {
_PyObject_ASSERT_WITH_MSG(op, gc_refs > 0, "refcount is too small");
}
return 0;
}
static void
move_unreachable(PyGC_Head *young, PyGC_Head *unreachable)
{
PyGC_Head *prev = young;
PyGC_Head *gc = GC_NEXT(young);
while (gc != young) {
if (gc_get_refs(gc)) {
PyObject *op = FROM_GC(gc);
traverseproc traverse = Py_TYPE(op)->tp_traverse;
_PyObject_ASSERT_WITH_MSG(op, gc_get_refs(gc) > 0,
"refcount is too small");
(void) traverse(op,
(visitproc)visit_reachable,
(void *)young);
_PyGCHead_SET_PREV(gc, prev);
gc_clear_collecting(gc);
prev = gc;
}
else {
prev->_gc_next = gc->_gc_next;
PyGC_Head *last = GC_PREV(unreachable);
last->_gc_next = (NEXT_MASK_UNREACHABLE | (uintptr_t)gc);
_PyGCHead_SET_PREV(gc, last);
gc->_gc_next = (NEXT_MASK_UNREACHABLE | (uintptr_t)unreachable);
unreachable->_gc_prev = (uintptr_t)gc;
}
gc = (PyGC_Head*)prev->_gc_next;
}
// young->_gc_prev must be last element remained in the list.
young->_gc_prev = (uintptr_t)prev;
// don't let the pollution of the list head's next pointer leak
unreachable->_gc_next &= ~NEXT_MASK_UNREACHABLE;
}
這段代碼的邏輯是,遍歷收集代中的所有對象,判斷對象的計數值是否為0
如果等於0,則從收集代中移除,加入不可達列表中,然後打上不可達標記。
如果不等於0,則遍歷對象的所有元素,如果元素已經被打上不可達標記,則把該元素從不可達列表中移除,重新加入收集代列表中,並且將計數值設置為1。這是因為父對象可以被訪問,那麼子對象一定可以被訪問。
- 把定義了__del__的對象從不可達對象中移除
static int
has_legacy_finalizer(PyObject *op)
{
return Py_TYPE(op)->tp_del != NULL;
}
static void
move_legacy_finalizers(PyGC_Head *unreachable, PyGC_Head *finalizers)
{
PyGC_Head *gc, *next;
assert((unreachable->_gc_next & NEXT_MASK_UNREACHABLE) == 0);
for (gc = GC_NEXT(unreachable); gc != unreachable; gc = next) {
PyObject *op = FROM_GC(gc);
_PyObject_ASSERT(op, gc->_gc_next & NEXT_MASK_UNREACHABLE);
gc->_gc_next &= ~NEXT_MASK_UNREACHABLE;
next = (PyGC_Head*)gc->_gc_next;
if (has_legacy_finalizer(op)) {
gc_clear_collecting(gc);
gc_list_move(gc, finalizers);
}
}
}
這裡的邏輯就比較簡單,判斷是否定義了__del__函數,如果有,則從不可達列表中刪除,加入finalizers列表,並且清除收集中標記。
/* A traversal callback for move_legacy_finalizer_reachable. */
static int
visit_move(PyObject *op, PyGC_Head *tolist)
{
if (_PyObject_IS_GC(op)) {
PyGC_Head *gc = AS_GC(op);
if (gc_is_collecting(gc)) {
gc_list_move(gc, tolist);
gc_clear_collecting(gc);
}
}
return 0;
}
/* Move objects that are reachable from finalizers, from the unreachable set
* into finalizers set.
*/
static void
move_legacy_finalizer_reachable(PyGC_Head *finalizers)
{
traverseproc traverse;
PyGC_Head *gc = GC_NEXT(finalizers);
for (; gc != finalizers; gc = GC_NEXT(gc)) {
/* Note that the finalizers list may grow during this. */
traverse = Py_TYPE(FROM_GC(gc))->tp_traverse;
(void) traverse(FROM_GC(gc),
(visitproc)visit_move,
(void *)finalizers);
}
}
然後再遍歷finalizers列表中的所有對象,判斷對象的每個元素是否也是可GC對象,並且也有收集中標記,如果滿足條件,則從不可達列表中刪除,加入finalizers列表,並且清除收集中標記。
- 遍歷不可達對象列表,處理弱引用。
- 遍歷不可達對象列表,為每個對象調用tp_finalize函數,如果沒有則跳過。
static void
finalize_garbage(PyThreadState *tstate, PyGC_Head *collectable)
{
destructor finalize;
PyGC_Head seen;
gc_list_init(&seen);
while (!gc_list_is_empty(collectable)) {
PyGC_Head *gc = GC_NEXT(collectable);
PyObject *op = FROM_GC(gc);
gc_list_move(gc, &seen);
if (!_PyGCHead_FINALIZED(gc) &&
(finalize = Py_TYPE(op)->tp_finalize) != NULL) {
_PyGCHead_SET_FINALIZED(gc);
Py_INCREF(op);
finalize(op);
assert(!_PyErr_Occurred(tstate));
Py_DECREF(op);
}
}
gc_list_merge(&seen, collectable);
}
- 處理複活的對象
static inline void
handle_resurrected_objects(PyGC_Head *unreachable, PyGC_Head* still_unreachable,
PyGC_Head *old_generation)
{
// Remove the PREV_MASK_COLLECTING from unreachable
// to prepare it for a new call to 'deduce_unreachable'
gc_list_clear_collecting(unreachable);
// After the call to deduce_unreachable, the 'still_unreachable' set will
// have the PREV_MARK_COLLECTING set, but the objects are going to be
// removed so we can skip the expense of clearing the flag.
PyGC_Head* resurrected = unreachable;
deduce_unreachable(resurrected, still_unreachable);
clear_unreachable_mask(still_unreachable);
// Move the resurrected objects to the old generation for future collection.
gc_list_merge(resurrected, old_generation);
}
這裡主要是上一步會調用tp_finalize
函數,有可能會把一些對象複活,所以需要重新收集一次不可達對象,然後將複活的對象移入老年代中。
- 刪除不可達對象
static void
delete_garbage(PyThreadState *tstate, GCState *gcstate,
PyGC_Head *collectable, PyGC_Head *old)
{
assert(!_PyErr_Occurred(tstate));
while (!gc_list_is_empty(collectable)) {
PyGC_Head *gc = GC_NEXT(collectable);
PyObject *op = FROM_GC(gc);
_PyObject_ASSERT_WITH_MSG(op, Py_REFCNT(op) > 0,
"refcount is too small");
if (gcstate->debug & DEBUG_SAVEALL) {
assert(gcstate->garbage != NULL);
if (PyList_Append(gcstate->garbage, op) < 0) {
_PyErr_Clear(tstate);
}
}
else {
inquiry clear;
if ((clear = Py_TYPE(op)->tp_clear) != NULL) {
Py_INCREF(op);
(void) clear(op);
if (_PyErr_Occurred(tstate)) {
_PyErr_WriteUnraisableMsg("in tp_clear of",
(PyObject*)Py_TYPE(op));
}
Py_DECREF(op);
}
}
if (GC_NEXT(collectable) == gc) {
/* object is still alive, move it, it may die later */
gc_clear_collecting(gc);
gc_list_move(gc, old);
}
}
}
其中的邏輯也簡單,遍歷最終不可達列表,然後調用每個對象的tp_clear函數。調用後,如果對象可以被釋放,則也會從GC列表中移除。所以在後面有一個判斷if (GC_NEXT(collectable) == gc),也就是該對象還沒有被移除,這種情況則清除該對象的收集中標記,然後移入老年代中。
- 將finalizers列表中的對象移入老年代中
static void
handle_legacy_finalizers(PyThreadState *tstate,
GCState *gcstate,
PyGC_Head *finalizers, PyGC_Head *old)
{
assert(!_PyErr_Occurred(tstate));
assert(gcstate->garbage != NULL);
PyGC_Head *gc = GC_NEXT(finalizers);
for (; gc != finalizers; gc = GC_NEXT(gc)) {
PyObject *op = FROM_GC(gc);
if ((gcstate->debug & DEBUG_SAVEALL) || has_legacy_finalizer(op)) {
if (PyList_Append(gcstate->garbage, op) < 0) {
_PyErr_Clear(tstate);
break;
}
}
}
gc_list_merge(finalizers, old);
}
所以說,定義了__del__的對象,有可能出現無法回收的情況。需要仔細編碼。
總結
python的垃圾回收主要用到了
- 引用計數
- 標記清除
- 分代回收
其中分代回收步驟為
- 將年輕代的對象移動到指定回收代的列表後。
- 遍歷回收代列表,將對象設置為收集中PREV_MASK_COLLECTING標記,然後將引用計數複製一份到_gc_prev中
- 然後遍歷每個對象中的每個元素,如果這個元素也是可GC對象,並且也有收集中標記,則將_gc_prev中的計數值減1
- 再遍歷回收代列表,判斷_gc_prev計數值是否為0,
- 如果為0,則標記為不可達,然後移動到不可達列表中。
- 如果不為0,則遍歷該對象的元素,如果該元素已經標記為清除,就把該元素移動到原回收代列表中。(也就是父對象仍然可達,則子對象也可達)。然後清除該對象的收集中標記。
- 遍歷不可達列表,清除不可達標記,判斷是否定義了__del__函數,如果有,則將清除收集中標記,並移入finalizers列表中。
- 遍歷finalizers列表的每個對象,判斷對象中的元素是否是可GC對象,並且有收集中標記,將該元素清除收集中標記,移入finalizers列表中。
- 遍歷不可達列表, 處理弱引用
- 遍歷不可達列表的每個對象,調用對象的tp_finalize函數,如果沒有則跳過。
- 遍歷不可達列表,將複活對象移到老年代列表中,其他對象移動到仍然不可達列表final_unreachable
- 最後遍歷 final_unreachable 列表,為每個對象調用tp_clear函數
- 如果真的可以刪除,則把自己從對應GC列表中摘除
- 如果還不能刪除,則清除對象的收集中標記,對象重新加入老年代中。
- 將finalizers列表中的每個對象重新加入老年代列表中。
例子
說到這裡好像還沒有具體分析環引用的情況
import sys
import gc
def a():
lst1 = []
lst2 = []
lst1.append(lst2)
lst2.append(lst1)
print("lst1 refcnt: {}".format(sys.getrefcount(lst1)))
print("lst2 refcnt: {}".format(sys.getrefcount(lst2)))
before_collect_cnt = gc.collect(2)
a()
after_collect_cnt = gc.collect(2)
print("before({}), after({})".format(before_collect_cnt, after_collect_cnt))
在筆者的電腦上輸出
hejs@ubuntu:~$ python main.py
lst1 refcnt: 3
lst2 refcnt: 3
before(0), after(2)
可以看到,在執行a函數時,lst1和lst2的引用計數為2(因為sys.getrefcount也會引用一次,所以輸出的值是真實計數+1)。
當a函數調用結束後,由於函數內的lst1、lst2變數解除了引用,所以此時兩個列表的計數值就為1了。出現環引用,無法釋放。
這個時候就輪到標記清楚和分代回收解決了。
- 首先會將第0、1代的元素移到第2代上。因為gc.collect(2)
- 然後遍歷第2代列表,為每個對象設置收集中標記,將對象的真實計數複製到_gc_prev中。
- 再遍歷第2代列表,判斷對象的子元素是否也是 可GC對象、也有收集中標記,如果有則將該元素計數值減1。
- 此時 lst1、lst2的_gc_prev計數值都為0
- 然後將_gc_prev計數值為0的對象移入不可達列表中。
- 因為listobject沒有__del__函數,也沒有tp_finalize函數,所以直接到第10步,調用tp_clear函數。
static int _list_clear(PyListObject *a)
{
Py_ssize_t i;
PyObject **item = a->ob_item;
if (item != NULL) {
i = Py_SIZE(a);
Py_SET_SIZE(a, 0);
a->ob_item = NULL;
a->allocated = 0;
while (--i >= 0) {
Py_XDECREF(item[i]);
}
PyMem_FREE(item);
}
/* Never fails; the return value can be ignored.
Note that there is no guarantee that the list is actually empty
at this point, because XDECREF may have populated it again! */
return 0;
}
也就是會為每個元素的引用計數減1。從之前分析可知,當計數減為0時,會調用對象的tp_dealloc函數,再看看listobject的tp_dealloc實現。
static void
list_dealloc(PyListObject *op)
{
Py_ssize_t i;
PyObject_GC_UnTrack(op);
Py_TRASHCAN_BEGIN(op, list_dealloc)
if (op->ob_item != NULL) {
i = Py_SIZE(op);
while (--i >= 0) {
Py_XDECREF(op->ob_item[i]);
}
PyMem_FREE(op->ob_item);
}
if (numfree < PyList_MAXFREELIST && PyList_CheckExact(op))
free_list[numfree++] = op;
else
Py_TYPE(op)->tp_free((PyObject *)op);
Py_TRASHCAN_END
}
首先會調用PyObject_GC_UnTrack,就是將該對象從GC鏈表中摘除。然後再遍歷子元素,將子元素的計數減1。計數減為0時,又會調用對象的tp_dealloc函數。
此番調用下來,lst1和lst2的計數都會被減為0,都會從GC鏈表中摘除,並且都能釋放。解除了環引用。