深入理解Python虛擬機(jī)中字節(jié)(bytes)的實(shí)現(xiàn)原理及源碼剖析

更新時(shí)間：2023年03月24日 08:20:57 作者：一無是處的研究僧

在本篇文章當(dāng)中主要給大家介紹在?cpython?內(nèi)部，bytes?的實(shí)現(xiàn)原理、內(nèi)存布局以及與?bytes?相關(guān)的一個比較重要的優(yōu)化點(diǎn)——?bytes?的拼接,需要的可以參考一下

數(shù)據(jù)結(jié)構(gòu)

typedef struct {
    PyObject_VAR_HEAD
    Py_hash_t ob_shash;
    char ob_sval[1];
 
    /* Invariants:
     *     ob_sval contains space for 'ob_size+1' elements.
     *     ob_sval[ob_size] == 0.
     *     ob_shash is the hash of the string or -1 if not computed yet.
     */
} PyBytesObject;
 
typedef struct {
    PyObject ob_base;
    Py_ssize_t ob_size; /* Number of items in variable part */
} PyVarObject;
 
typedef struct _object {
    Py_ssize_t ob_refcnt;
    struct _typeobject *ob_type;
} PyObject;

上面的數(shù)據(jù)結(jié)構(gòu)用圖示如下所示：

現(xiàn)在我們來解釋一下上面的數(shù)據(jù)結(jié)構(gòu)各個字段的含義：

ob_refcnt，這個還是對象的引用計(jì)數(shù)的個數(shù)，主要是在垃圾回收的時(shí)候有用。
ob_type，這個是對象的數(shù)據(jù)類型。
ob_size，表示這個對象當(dāng)中字節(jié)的個數(shù)。
ob_shash，對象的哈希值，如果還沒有計(jì)算，哈希值為 -1 。
ob_sval，一個數(shù)據(jù)存儲一個字節(jié)的數(shù)據(jù)，需要注意的是 ob_sval[size] 一定等于 '\0' ，表示字符串的結(jié)尾。

可能你會有疑問上面的結(jié)構(gòu)體當(dāng)中并沒有后面的那么多字節(jié)啊，數(shù)組只有一個字節(jié)的數(shù)據(jù)啊，這是因?yàn)樵?cpython 的實(shí)現(xiàn)當(dāng)中除了申請 PyBytesObject 大的小內(nèi)存空間之外，還會在這個基礎(chǔ)之上申請連續(xù)的額外的內(nèi)存空間用于保存數(shù)據(jù)，在后續(xù)的源碼分析當(dāng)中可以看到這一點(diǎn)。

下面我們舉幾個例子來說明一下上面的布局：

上面是空和字符串 abc 的字節(jié)表示。

創(chuàng)建字節(jié)對象

下面是在 cpython 當(dāng)中通過字節(jié)數(shù)創(chuàng)建 PyBytesObject 對象的函數(shù)。下面的函數(shù)的主要功能是創(chuàng)建一個能夠存儲 size 個字節(jié)大小的數(shù)據(jù)的 PyBytesObject 對象，下面的函數(shù)最重要的一個步驟就是申請內(nèi)存空間。

static PyObject *
_PyBytes_FromSize(Py_ssize_t size, int use_calloc)
{
    PyBytesObject *op;
    assert(size >= 0);
 
    if (size == 0 && (op = nullstring) != NULL) {
#ifdef COUNT_ALLOCS
        null_strings++;
#endif
        Py_INCREF(op);
        return (PyObject *)op;
    }
 
    if ((size_t)size > (size_t)PY_SSIZE_T_MAX - PyBytesObject_SIZE) {
        PyErr_SetString(PyExc_OverflowError,
                        "byte string is too large");
        return NULL;
    }
 
    /* Inline PyObject_NewVar */
    // PyBytesObject_SIZE + size 就是實(shí)際申請的內(nèi)存空間的大小 PyBytesObject_SIZE 就是表示 PyBytesObject 各個字段占用的實(shí)際的內(nèi)存空間大小
    if (use_calloc)
        op = (PyBytesObject *)PyObject_Calloc(1, PyBytesObject_SIZE + size);
    else
        op = (PyBytesObject *)PyObject_Malloc(PyBytesObject_SIZE + size);
    if (op == NULL)
        return PyErr_NoMemory();
    // 將對象的 ob_size 字段賦值成 size 
    (void)PyObject_INIT_VAR(op, &PyBytes_Type, size);
    // 由于對象的哈希值還沒有進(jìn)行計(jì)算 因此現(xiàn)將哈希值賦值成 -1
    op->ob_shash = -1;
    if (!use_calloc)
        op->ob_sval[size] = '\0';
    /* empty byte string singleton */
    if (size == 0) {
        nullstring = op;
        Py_INCREF(op);
    }
    return (PyObject *) op;
}

我們可以使用一個寫例子來看一下實(shí)際的 PyBytesObject 內(nèi)存空間的大小。

>>> import sys
>>> a = b"hello world"
>>> sys.getsizeof(a)
44
>>>

上面的 44 = 32 + 11 + 1 。

其中 32 是 PyBytesObject 4 個字段所占用的內(nèi)存空間，ob_refcnt、ob_type、ob_size和 ob_shash 各占 8 個字節(jié)。11 是表示字符串 "hello world" 占用 11 個字節(jié)，最后一個字節(jié)是 '\0' 。

查看字節(jié)長度

這個函數(shù)主要是返回 PyBytesObject 對象的字節(jié)長度，也就是直接返回 ob_size 的值。

static Py_ssize_t
bytes_length(PyBytesObject *a)
{
    // (((PyVarObject*)(ob))->ob_size)
    return Py_SIZE(a);
}

字節(jié)拼接

在 python 當(dāng)中執(zhí)行下面的代碼就會執(zhí)行字節(jié)拼接函數(shù)：

>>> b"abc" + b"edf"

下方就是具體的執(zhí)行字節(jié)拼接的函數(shù)：

/* This is also used by PyBytes_Concat() */
static PyObject *
bytes_concat(PyObject *a, PyObject *b)
{
    Py_buffer va, vb;
    PyObject *result = NULL;
 
    va.len = -1;
    vb.len = -1;
    // Py_buffer 當(dāng)中有一個指針字段 buf 可以用戶保存 PyBytesObject 當(dāng)中字節(jié)數(shù)據(jù)的首地址
    // PyObject_GetBuffer 函數(shù)的主要作用是將 對象 a 當(dāng)中的字節(jié)數(shù)組賦值給 va 當(dāng)中的 buf
    if (PyObject_GetBuffer(a, &va, PyBUF_SIMPLE) != 0 ||
        PyObject_GetBuffer(b, &vb, PyBUF_SIMPLE) != 0) {
        PyErr_Format(PyExc_TypeError, "can't concat %.100s to %.100s",
                     Py_TYPE(b)->tp_name, Py_TYPE(a)->tp_name);
        goto done;
    }
 
    /* Optimize end cases */
    if (va.len == 0 && PyBytes_CheckExact(b)) {
        result = b;
        Py_INCREF(result);
        goto done;
    }
    if (vb.len == 0 && PyBytes_CheckExact(a)) {
        result = a;
        Py_INCREF(result);
        goto done;
    }
 
    if (va.len > PY_SSIZE_T_MAX - vb.len) {
        PyErr_NoMemory();
        goto done;
    }
    result = PyBytes_FromStringAndSize(NULL, va.len + vb.len);
    // 下方就是將對象 a b 當(dāng)中的字節(jié)數(shù)據(jù)拷貝到新的
    if (result != NULL) {
        // PyBytes_AS_STRING 宏定義在下方當(dāng)中 主要就是使用 PyBytesObject 對象當(dāng)中的
        // ob_sval 字段 也就是將 buf 數(shù)據(jù)（也就是 a 或者 b 當(dāng)中的字節(jié)數(shù)據(jù)）拷貝到 ob_sval當(dāng)中
        memcpy(PyBytes_AS_STRING(result), va.buf, va.len);
        memcpy(PyBytes_AS_STRING(result) + va.len, vb.buf, vb.len);
    }
 
  done:
    if (va.len != -1)
        PyBuffer_Release(&va);
    if (vb.len != -1)
        PyBuffer_Release(&vb);
    return result;
}

#define PyBytes_AS_STRING(op) (assert(PyBytes_Check(op)), \
                                (((PyBytesObject *)(op))->ob_sval))

我們修改一個這個函數(shù)，在其中加入一條打印語句，然后重新編譯 python 執(zhí)行結(jié)果如下所示：

Python 3.9.0b1 (default, Mar 23 2023, 08:35:33) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> b"abc" + b"edf"
In concat function: abc <> edf
b'abcedf'
>>>

在上面的拼接函數(shù)當(dāng)中會拷貝原來的兩個字節(jié)對象，因此需要謹(jǐn)慎使用，一旦發(fā)生非常多的拷貝的話是非常耗費(fèi)內(nèi)存的。因此需要警惕使用循環(huán)內(nèi)的內(nèi)存拼接。比如對于 [b"a", b"b", b"c"] 來說，如果使用循環(huán)拼接的話，那么會將 b"a" 拷貝兩次。

>>> res = b""
>>> for item in  [b"a", b"b", b"c"]:
...     res += item
...
>>> res
b'abc'
>>>

因?yàn)?b"a", b"b" 在拼接的時(shí)候會將他們分別拷貝一次，在進(jìn)行 b"ab"，b"c" 拼接的時(shí)候又會將 ab 和 c 拷貝一次，那么具體的拷貝情況如下所示：

"a" 拷貝了一次。
"b" 拷貝了一次。
"ab" 拷貝了一次。
"c" 拷貝了一次。

但是實(shí)際上我們的需求是只需要對 [b"a", b"b", b"c"] 當(dāng)中的數(shù)據(jù)各拷貝一次，如果我們要實(shí)現(xiàn)這一點(diǎn)可以使用 b"".join([b"a", b"b", b"c"])，直接將 [b"a", b"b", b"c"] 作為參數(shù)傳遞，然后各自只拷貝一次，具體的實(shí)現(xiàn)代碼如下所示，在這個例子當(dāng)中 sep 就是空串 b""，iterable 就是 [b"a", b"b", b"c"] 。

Py_LOCAL_INLINE(PyObject *)
STRINGLIB(bytes_join)(PyObject *sep, PyObject *iterable)
{
    char *sepstr = STRINGLIB_STR(sep);
    const Py_ssize_t seplen = STRINGLIB_LEN(sep);
    PyObject *res = NULL;
    char *p;
    Py_ssize_t seqlen = 0;
    Py_ssize_t sz = 0;
    Py_ssize_t i, nbufs;
    PyObject *seq, *item;
    Py_buffer *buffers = NULL;
#define NB_STATIC_BUFFERS 10
    Py_buffer static_buffers[NB_STATIC_BUFFERS];
 
    seq = PySequence_Fast(iterable, "can only join an iterable");
    if (seq == NULL) {
        return NULL;
    }
 
    seqlen = PySequence_Fast_GET_SIZE(seq);
    if (seqlen == 0) {
        Py_DECREF(seq);
        return STRINGLIB_NEW(NULL, 0);
    }
#ifndef STRINGLIB_MUTABLE
    if (seqlen == 1) {
        item = PySequence_Fast_GET_ITEM(seq, 0);
        if (STRINGLIB_CHECK_EXACT(item)) {
            Py_INCREF(item);
            Py_DECREF(seq);
            return item;
        }
    }
#endif
    if (seqlen > NB_STATIC_BUFFERS) {
        buffers = PyMem_NEW(Py_buffer, seqlen);
        if (buffers == NULL) {
            Py_DECREF(seq);
            PyErr_NoMemory();
            return NULL;
        }
    }
    else {
        buffers = static_buffers;
    }
 
    /* Here is the general case.  Do a pre-pass to figure out the total
     * amount of space we'll need (sz), and see whether all arguments are
     * bytes-like.
     */
    for (i = 0, nbufs = 0; i < seqlen; i++) {
        Py_ssize_t itemlen;
        item = PySequence_Fast_GET_ITEM(seq, i);
        if (PyBytes_CheckExact(item)) {
            /* Fast path. */
            Py_INCREF(item);
            buffers[i].obj = item;
            buffers[i].buf = PyBytes_AS_STRING(item);
            buffers[i].len = PyBytes_GET_SIZE(item);
        }
        else if (PyObject_GetBuffer(item, &buffers[i], PyBUF_SIMPLE) != 0) {
            PyErr_Format(PyExc_TypeError,
                         "sequence item %zd: expected a bytes-like object, "
                         "%.80s found",
                         i, Py_TYPE(item)->tp_name);
            goto error;
        }
        nbufs = i + 1;  /* for error cleanup */
        itemlen = buffers[i].len;
        if (itemlen > PY_SSIZE_T_MAX - sz) {
            PyErr_SetString(PyExc_OverflowError,
                            "join() result is too long");
            goto error;
        }
        sz += itemlen;
        if (i != 0) {
            if (seplen > PY_SSIZE_T_MAX - sz) {
                PyErr_SetString(PyExc_OverflowError,
                                "join() result is too long");
                goto error;
            }
            sz += seplen;
        }
        if (seqlen != PySequence_Fast_GET_SIZE(seq)) {
            PyErr_SetString(PyExc_RuntimeError,
                            "sequence changed size during iteration");
            goto error;
        }
    }
 
    /* Allocate result space. */
    res = STRINGLIB_NEW(NULL, sz);
    if (res == NULL)
        goto error;
 
    /* Catenate everything. */
    p = STRINGLIB_STR(res);
    if (!seplen) {
        /* fast path */
        for (i = 0; i < nbufs; i++) {
            Py_ssize_t n = buffers[i].len;
            char *q = buffers[i].buf;
            Py_MEMCPY(p, q, n);
            p += n;
        }
        goto done;
    }
    // 具體的實(shí)現(xiàn)邏輯就是在這里
    for (i = 0; i < nbufs; i++) {
        Py_ssize_t n;
        char *q;
        if (i) {
            // 首先現(xiàn)將 sepstr 拷貝到新的數(shù)組里面但是在我們舉的例子當(dāng)中是空串 b""
            Py_MEMCPY(p, sepstr, seplen);
            p += seplen;
        }
        n = buffers[i].len;
        q = buffers[i].buf;
        // 然后將列表當(dāng)中第 i 個 bytes 的數(shù)據(jù)拷貝到 p 當(dāng)中 這樣就是實(shí)現(xiàn)了我們所需要的效果
        Py_MEMCPY(p, q, n);
        p += n;
    }
    goto done;
 
error:
    res = NULL;
done:
    Py_DECREF(seq);
    for (i = 0; i < nbufs; i++)
        PyBuffer_Release(&buffers[i]);
    if (buffers != static_buffers)
        PyMem_FREE(buffers);
    return res;
}

單字節(jié)字符

在 cpython 的內(nèi)部實(shí)現(xiàn)當(dāng)中給單字節(jié)的字符做了一個小的緩沖池：

static PyBytesObject *characters[UCHAR_MAX + 1]; // UCHAR_MAX 在 64 位系統(tǒng)當(dāng)中等于 255

當(dāng)創(chuàng)建的 bytes 只有一個字符的時(shí)候就可以檢查是否 characters 當(dāng)中已經(jīng)存在了，如果存在就直接返回這個已經(jīng)創(chuàng)建好的 PyBytesObject 對象，否則再進(jìn)行創(chuàng)建。新創(chuàng)建的 PyBytesObject 對象如果長度等于 1 的話也會被加入到這個數(shù)組當(dāng)中。下面是 PyBytesObject 的另外一個創(chuàng)建函數(shù)：

PyObject *
PyBytes_FromStringAndSize(const char *str, Py_ssize_t size)
{
    PyBytesObject *op;
    if (size < 0) {
        PyErr_SetString(PyExc_SystemError,
            "Negative size passed to PyBytes_FromStringAndSize");
        return NULL;
    }
    // 如果創(chuàng)建長度等于 1 而且對象在 characters 當(dāng)中存在的話那么就直接返回
    if (size == 1 && str != NULL &&
        (op = characters[*str & UCHAR_MAX]) != NULL)
    {
#ifdef COUNT_ALLOCS
        one_strings++;
#endif
        Py_INCREF(op);
        return (PyObject *)op;
    }
 
    op = (PyBytesObject *)_PyBytes_FromSize(size, 0);
    if (op == NULL)
        return NULL;
    if (str == NULL)
        return (PyObject *) op;
 
    Py_MEMCPY(op->ob_sval, str, size);
    /* share short strings */
    // 如果創(chuàng)建的對象的長度等于 1 那么久將這個對象保存到 characters 當(dāng)中
    if (size == 1) {
        characters[*str & UCHAR_MAX] = op;
        Py_INCREF(op);
    }
    return (PyObject *) op;
}

我們可以使用下面的代碼進(jìn)行驗(yàn)證：

>>> a = b"a"
>>> b  =b"a"
>>> a == b
True
>>> a is b
True
>>> a = b"aa"
>>> b = b"aa"
>>> a == b
True
>>> a is b
False

從上面的代碼可以知道，確實(shí)當(dāng)我們創(chuàng)建的 bytes 的長度等于 1 的時(shí)候?qū)ο蟠_實(shí)是同一個對象。

總結(jié)

在本篇文章當(dāng)中主要給大家介紹了在 cpython 內(nèi)部對于 bytes 的實(shí)現(xiàn)，重點(diǎn)介紹了 cpython 當(dāng)中 PyBytesObject 的內(nèi)存布局和創(chuàng)建 PyBytesObject 的函數(shù)，以及對于 bytes 對象的拼接細(xì)節(jié)和 cpython 內(nèi)部單字節(jié)字符的緩沖池。在程序當(dāng)中最好使用 join 操作進(jìn)行 btyes 的拼接操作，否則效率會比較低。

到此這篇關(guān)于深入理解Python虛擬機(jī)中字節(jié)(bytes)的實(shí)現(xiàn)原理及源碼剖析的文章就介紹到這了,更多相關(guān)Python虛擬機(jī)字節(jié)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: