深入理解 Python 虛擬機器器:程序、執行緒和協程

2023-10-20 18:00:45

深入理解 Python 虛擬機器器:程序、執行緒和協程

在本篇文章當中深入分析在 Python 當中 程序、執行緒和協程的區別,這三個概念會讓人非常迷惑。如果沒有深入瞭解這三者的實現原理,只是看一些文字說明,也很難理解。在本篇文章當中我們將通過分析部分原始碼來詳細分析一下這三者根本的區別是什麼,重點是協程的應用場景和在 Python 當中是如何使用協程的,至於協程的實現原理在前面的文章當中已經詳細討論過了 深入理解 Python 虛擬機器器:協程初探——不過是生成器而已深入理解 Python 虛擬機器器:生成器停止背後的魔法

程序和執行緒

程序是一個非常古老的概念,根據 wiki 的描述,程序是一個正在執行的計算機程式,這裡說的計算機程式是指的是能夠直接被作業系統載入執行的程式,比如你通過編譯器編譯之後的 c/c++ 程式。

舉個例子,你在 shell 當中敲出的 ./a.out 在按下回車之後,a.out 就會被執行起來,這個被作業系統執行的程式就是一個程序。在一個程序內部會有很多的資源,比如開啟的檔案,申請的記憶體,接收到的訊號等等,這些資訊都是由核心來維護。關於程序有一個非常重要的概念,就是程序的記憶體地址空間,一個程序當中主要有程式碼、資料、堆和執行棧:

這裡我們不過多的去分析這一點,現在就需要知道在一個程序當中主要有這 4 個東西,而且在核心當中會有資料結構去儲存他們。程式被作業系統載入之後可以被作業系統放到 CPU 上執行。我們可以同時啟動多個程序,讓作業系統去排程,而且隨著體系結構的發展,現在的機器上都是多核機器,同時啟動多個程序可以讓他們同時執行。

在程式設計時我們會有一個需求,我們希望並行的去執行程式,而且他們可以修改共有的記憶體,當一個程序修改之後能夠被另外一個程序看到,從這個角度來說他們就需要有同一個地址空間,這樣就可以實現這一點了,而且這種方式有一個好處就是節省記憶體資源,比如只需要儲存一份記憶體的地址空間了。

上面談到的實現程序的方式實際上被稱作輕量級程序,也被叫做執行緒。具體來說就是可以在一個程序內部啟動多個執行緒,這些執行緒之前有這相同的記憶體地址空間,這些執行緒能夠同時被作業系統排程到不同的核心上同時執行。我們現在在 linux 上使用的執行緒是NPTL (Native POSIX Threads Library),從 glibc2.3.2 開始支援,而且要求 linux 2.6 之後的特性。在前面的內容我們談到了,在同一個程序內部的執行緒是可以共用一些程序擁有的資料的,比如:

  • 程序號。
  • 父程序號。
  • 行程群組號和對談號。
  • 控制終端。
  • 開啟的檔案描述符表。
  • 當前工作目錄。
  • 虛擬地址空間。

執行緒也有自己的私有資料,比如:

  • 程式執行棧空間。
  • 暫存器狀態。
  • 執行緒的執行緒號。

在 linux 當中建立執行緒和程序的系統呼叫分別為 clonefork,如果為了建立執行緒的話我們可以不使用這麼低層級的 API,我們可以通過 NPTL 提供的 pthread_create 方法建立執行緒執行相應的方法。

#include <stdio.h>
#include <pthread.h>

void* func(void* arg) {
  printf("Hello World\n");
  return NULL;
}

int main() {

  pthread_t t; // 定義一個執行緒
  pthread_create(&t, NULL, func, NULL); // 建立執行緒並且執行函數 func 

  // wait unit thread t finished
  pthread_join(t, NULL); // 主執行緒等待執行緒 t 執行完成然後主執行緒才繼續往下執行

  printf("thread t has finished\n");
  return 0;
}

編譯上述程式:

clang helloworld.c -o helloworld.out -lpthread
或者
gcc helloworld.c -o helloworld.out -lpthread

在上面的程式碼當中主執行緒(可以認為是執行主函數的執行緒)首先定義一個執行緒,然後建立執行緒並且執行函數 func ,當建立完成之後,主執行緒使用 pthread_join 阻塞自己,直到等待執行緒 t 執行完成之後主執行緒才會繼續往下執行。

我們現在仔細分析一下 pthread_create 的函數簽名,並且對他的引數進行詳細分析:

int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                          void *(*start_routine) (void *), void *arg);
  • 引數 thread 是一個型別為 pthread_t 的指標物件,將這個物件會在 pthread_create 內部會被賦值為存放執行緒 id 的地址,在後文當中我們將使用一個例子仔細的介紹這個引數的含義。
  • 引數 attr 是一個型別為 pthread_attr_t 的指標物件,我們可以在這個物件當中設定執行緒的各種屬性,比如說執行緒取消的狀態和類別,執行緒使用的棧的大小以及棧的初始位置等等,在後文當中我們將詳細介紹這個屬性的使用方法,當這個屬性為 NULL 的時候,使用預設的屬性值。
  • 引數 start_routine 是一個返回型別為 void,引數型別為 void 的函數指標,指向執行緒需要執行的函數,執行緒執行完成這個函數之後執行緒就會退出。
  • 引數 arg ,傳遞給函數 start_routine 的一個引數,在上一條當中我們提到了 start_routine 有一個引數,是一個 void 型別的指標,這個引數也是一個 void 型別的指標,在後文當中我們使用一個例子說明這個引數的使用方法。

在 Python 當中可以通過 threading 來建立一個執行緒:

import threading

def func():
	print("Hello World")


if __name__ == '__main__':
	t = threading.Thread(target=func)
	t.start()
	t.join()

現在有一個問題是,在 Python 當中真的是使用 pthread_create 來建立執行緒的嗎(在 Linux 當中)?Python 當中的執行緒和我們常說的執行緒是一致的嗎?

我們現在來分析一下 threading 的原始碼,執行緒的 start (也就是 Thread 類的 start 方法)方法如下:

    def start(self):
        if not self._initialized:
            raise RuntimeError("thread.__init__() not called")

        if self._started.is_set():
            raise RuntimeError("threads can only be started once")

        with _active_limbo_lock:
            _limbo[self] = self
        try:
            _start_new_thread(self._bootstrap, ())
        except Exception:
            with _active_limbo_lock:
                del _limbo[self]
            raise
        self._started.wait()
 

在上面的程式碼當中最核心的一行程式碼就是 _start_new_thread(self._bootstrap, ()),這行程式碼的含義是啟動一個新的執行緒去執行 self._bootstrap ,在 self._bootstrap 當中會呼叫 _bootstrap_inner,在 _bootstrap_inner 當中會呼叫 Thread 的 run 方法,而在run方法當中最終呼叫了我們傳遞給 Thread 類的函數。

    def run(self):
        try:
            if self._target is not None:
                self._target(*self._args, **self._kwargs)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs

    def _bootstrap(self):
        try:
            self._bootstrap_inner()
        except:
            if self._daemonic and _sys is None:
                return
            raise

    def _bootstrap_inner(self):
        try:
            self._set_ident()
            self._set_tstate_lock()
            if _HAVE_THREAD_NATIVE_ID:
                self._set_native_id()
            self._started.set()
            with _active_limbo_lock:
                _active[self._ident] = self
                del _limbo[self]

            if _trace_hook:
                _sys.settrace(_trace_hook)
            if _profile_hook:
                _sys.setprofile(_profile_hook)

            try:
                self.run()
            except:
                self._invoke_excepthook(self)
        finally:
            self._delete()

現在的問題是 _start_new_thread 是如何實現的?這個方法是 CPython 內部使用 C 語言實現的方法,在這裡我們不再將全部的細節進行分析,只討論大致的流程。

在執行 _start_new_thread 時,最終會呼叫PyThread_start_new_thread 這個方法,第一個引數是一個函數,這個函數為 t_bootstrap,在PyThread_start_new_thread 當中會使用 pthread_create 建立一個新的執行緒執行 t_bootstrap 函數,在函數 t_bootstrap 當中會呼叫從 Python 層面當中傳遞過來的 _bootstrap 方法。

long
PyThread_start_new_thread(void (*func)(void *), void *arg)
{
    pthread_t th;
    int status;
    pthread_attr_t attrs;
    size_t      tss;

    if (!initialized)
        PyThread_init_thread();

    if (pthread_attr_init(&attrs) != 0)
        return -1;
    tss = (_pythread_stacksize != 0) ? _pythread_stacksize
                                     : THREAD_STACK_SIZE;
    if (tss != 0) {
        if (pthread_attr_setstacksize(&attrs, tss) != 0) {
            pthread_attr_destroy(&attrs);
            return -1;
        }
    }
    pthread_attr_setscope(&attrs, PTHREAD_SCOPE_SYSTEM);

    status = pthread_create(&th,
                             &attrs,
                             (void* (*)(void *))func,
                             (void *)arg
                             ); // 建立新執行緒執行函數 func,也就是傳遞過來的函數 t_bootstrap(函數內容見下方)
    // 在執行完上面的程式碼之後執行緒就會立即執行了不需要像 Python 當中的執行緒一樣需要呼叫 start
    pthread_attr_destroy(&attrs);
    if (status != 0)
        return -1;

    pthread_detach(th);

    return (long) th;
}

static void
t_bootstrap(void *boot_raw)
{
    struct bootstate *boot = (struct bootstate *) boot_raw;
    PyThreadState *tstate;
    PyObject *res;

    tstate = boot->tstate;
    tstate->thread_id = PyThread_get_thread_ident();
    _PyThreadState_Init(tstate);
    PyEval_AcquireThread(tstate);
    nb_threads++;
    // boot->func 就是從 Python 層面傳遞過來的 _bootstrap 
    // PyEval_CallObjectWithKeywords 就是呼叫 Python 層面的函數
    // 下面這行程式碼就是在建立執行緒後執行的 Python 程式碼
    res = PyEval_CallObjectWithKeywords(
        boot->func, boot->args, boot->keyw);
    if (res == NULL) {
        if (PyErr_ExceptionMatches(PyExc_SystemExit))
            PyErr_Clear();
        else {
            PyObject *file;
            PySys_WriteStderr(
                "Unhandled exception in thread started by ");
            file = PySys_GetObject("stderr");
            if (file != NULL && file != Py_None)
                PyFile_WriteObject(boot->func, file, 0);
            else
                PyObject_Print(boot->func, stderr, 0);
            PySys_WriteStderr("\n");
            PyErr_PrintEx(0);
        }
    }
    else
        Py_DECREF(res);
    Py_DECREF(boot->func);
    Py_DECREF(boot->args);
    Py_XDECREF(boot->keyw);
    PyMem_DEL(boot_raw);
    nb_threads--;
    PyThreadState_Clear(tstate);
    PyThreadState_DeleteCurrent();
    PyThread_exit_thread();
}

從上面的整個建立執行緒的流程來看,當我們在 Python 層面建立一個執行緒之後,最終會呼叫 pthread_create 函數,真正建立一個執行緒(我們在前面已經討論過這種執行緒能夠被作業系統排程在 CPU 上執行,如果是多核機器的話,這兩個執行緒可以在同一個時刻執行)去執行相應的 Python 程式碼,也就是說當我們使用 threading 模組建立一個執行緒的時候,最終確實使用了 pthread_create 建立了一個執行緒。

協程

Coroutines are computer program components that allow execution to be suspended and resumed, generalizing subroutines for cooperative multitasking.

根據 wiki 的描述,協程是一個允許停下來和恢復執行的程式。在 Python 當中協程是基於生成器實現的(如果想具體瞭解生成器和協程的實現原理,可以參考這兩篇文章 深入理解 Python 虛擬機器器:協程初探——不過是生成器而已深入理解 Python 虛擬機器器:生成器停止背後的魔法),因為生成器是滿足這個要求的,他可以讓程式執行到函數的某一部分停下來,然後還能夠繼續恢復執行。

在繼續分析協程之前我們來討論一下協程的應用場景。現在假如需要處理很多網路請求,一個執行緒處理一個請求,當處理一個請求的時候我們需要等待使用者端的響應,執行緒在等待使用者端響應的時候是處於阻塞狀態不需要使用 CPU,假設 CPU 的使用率為 0.0001%,那麼我們大概需要 1000000 個執行緒才能夠將 CPU 的使用率達到 100%,而通常我們在核心建立一個執行緒大概需要 2MB 的記憶體,4GB 記憶體大概能夠建立 2048 個執行緒,這遠遠達不到我們需要建立的執行緒個數。而我們可以通過建立協程來達到這一點要求,因為協程需要的記憶體比執行緒小的多,而且協程是在使用者態實現的,不同的程式語言可以根據語言本身的情況進行實現。而我們在前面說明了一個執行緒可以被掛起,掛起之後也可以被繼續執行,我們可以利用這一點,當協程傳送一個網路請求之後就被掛起,這個時候切換到其他協程繼續執行,這樣就可以讓一個執行緒充分利用 CPU 的資源。對應的虛擬碼如下:

def recv(socket):
  while True:
    try:
      data = socket.recv() # 接收到資料了
    	return data
    except BlockingIOError:
      yield # 讓出 CPU 的執行權,也就是將協程暫停,讓其他協程執行起來

在 Python 當中和協程非常相關的另外一個概念就是事件迴圈 (Eventloop),我們將需要執行的協程都加入到事件迴圈當中,當有協程讓出 CPU 的執行權的之後,整個程式的流程就退回到了事件迴圈上,此時事件迴圈再執行另外一個協程,這樣就能夠充分利用 CPU 的效能了。事件迴圈的執行流程大致如下所示:

def event_loop():
  coroutines = [...]
  while coroutines.is_not_empty():
    coroutine = get_a_coroutine(coroutines)
    res = coroutine.run() # 當程式從這裡返回的時候要麼是協程停下來了,要麼是協程執行完成了
    if coroutine.is_not_finished():
      append(coroutines)

執行緒和程序的概念相對來說比較容易理解,協程比較困難,協程是使用者態實現的,它是由程式語言自己來進行排程,而不是由作業系統進行排程的,這是他和執行緒和程序最大的區別,而且協程相比起執行緒和程序來說需要的記憶體資源更少(如果想具體瞭解生成器和協程的實現原理,可以參考這兩篇文章 深入理解 Python 虛擬機器器:協程初探——不過是生成器而已深入理解 Python 虛擬機器器:生成器停止背後的魔法)。

對於我們在實際程式設計當中來說,只有當你的程式由很多 IO 密集型的程式的時候才需要考慮使用協程,比如伺服器開發。這是因為只有在這種場景下才能夠發揮協程的效能,如果你的程式是計算密集型的程式就不需要使用協程了,因為協程相對於執行緒來說還會有協程切換的開銷。

總結

在本篇文章當中主要討論了程序、執行緒和協程的區別,以及在 Linux 當中建立執行緒的 API,以及 CPython 當中建立執行緒的流程,最後討論了一下協程的使用場景,為什麼需要使用協程以及在 Python 當中是如何使用協程的。只有當你的程式是有比較多的 IO 操作的時候,你才需要考慮使用協程,因為協程提升的是 CPU 的利用率,如果你的程式本來 CPU 利用率就很高了,比如有很多的數學計算,你就不需要使用協程了,這樣做就可以避免額外的切換開銷了。


本篇文章是深入理解 python 虛擬機器器系列文章之一,文章地址:https://github.com/Chang-LeHung/dive-into-cpython

更多精彩內容合集可存取專案:https://github.com/Chang-LeHung/CSCore

關注公眾號:一無是處的研究僧,瞭解更多計算機(Java、Python、計算機系統基礎、演演算法與資料結構)知識。