Go 語言 實戰(zhàn): 封裝 qsort

原文鏈接：https://chai2010.cn/advanced-go-programming-book/ch2-cgo/ch2-06-qsort.html

2.6 實戰(zhàn): 封裝 qsort

qsort 快速排序函數(shù)是 C 語言的高階函數(shù)，支持用于自定義排序比較函數(shù)，可以對任意類型的數(shù)組進行排序。本節(jié)我們嘗試基于 C 語言的 qsort 函數(shù)封裝一個 Go 語言版本的 qsort 函數(shù)。

2.6.1 認識 qsort 函數(shù)

qsort 快速排序函數(shù)有 <stdlib.h> 標準庫提供，函數(shù)的聲明如下：

void qsort(
    void* base, size_t num, size_t size,
    int (*cmp)(const void*, const void*)
);

其中 base 參數(shù)是要排序數(shù)組的首個元素的地址，num 是數(shù)組中元素的個數(shù)，size 是數(shù)組中每個元素的大小。最關(guān)鍵是 cmp 比較函數(shù)，用于對數(shù)組中任意兩個元素進行排序。cmp 排序函數(shù)的兩個指針參數(shù)分別是要比較的兩個元素的地址，如果第一個參數(shù)對應(yīng)元素大于第二個參數(shù)對應(yīng)的元素將返回結(jié)果大于 0，如果兩個元素相等則返回 0，如果第一個元素小于第二個元素則返回結(jié)果小于 0。

下面的例子是用 C 語言的 qsort 對一個 int 類型的數(shù)組進行排序：

#include <stdio.h>
#include <stdlib.h>

#define DIM(x) (sizeof(x)/sizeof((x)[0]))

static int cmp(const void* a, const void* b) {
    const int* pa = (int*)a;
    const int* pb = (int*)b;
    return *pa - *pb;
}

int main() {
    int values[] = { 42, 8, 109, 97, 23, 25};
    int i;

    qsort(values, DIM(values), sizeof(values[0]), cmp);

    for(i = 0; i < DIM(values); i++) {
        printf ("%d",values[i]);
    }
    return 0;
}

其中 DIM(values) 宏用于計算數(shù)組元素的個數(shù)，sizeof(values[0]) 用于計算數(shù)組元素的大小。 cmp 是用于排序時比較兩個元素大小的回調(diào)函數(shù)。為了避免對全局名字空間的污染，我們將 cmp 回調(diào)函數(shù)定義為僅當前文件內(nèi)可訪問的靜態(tài)函數(shù)。

2.6.2 將 qsort 函數(shù)從 Go 包導(dǎo)出

為了方便 Go 語言的非 CGO 用戶使用 qsort 函數(shù)，我們需要將 C 語言的 qsort 函數(shù)包裝為一個外部可以訪問的 Go 函數(shù)。

用 Go 語言將 qsort 函數(shù)重新包裝為 qsort.Sort 函數(shù)：

package qsort

//typedef int (*qsort_cmp_func_t)(const void* a, const void* b);
import "C"
import "unsafe"

func Sort(
    base unsafe.Pointer, num, size C.size_t,
    cmp C.qsort_cmp_func_t,
) {
    C.qsort(base, num, size, cmp)
}

因為 Go 語言的 CGO 語言不好直接表達 C 語言的函數(shù)類型，因此在 C 語言空間將比較函數(shù)類型重新定義為一個 qsort_cmp_func_t 類型。

雖然 Sort 函數(shù)已經(jīng)導(dǎo)出了，但是對于 qsort 包之外的用戶依然不能直接使用該函數(shù)——Sort 函數(shù)的參數(shù)還包含了虛擬的 C 包提供的類型。 在 CGO 的內(nèi)部機制一節(jié)中我們已經(jīng)提過，虛擬的 C 包下的任何名稱其實都會被映射為包內(nèi)的私有名字。比如 C.size_t 會被展開為 _Ctype_size_t，C.qsort_cmp_func_t 類型會被展開為 _Ctype_qsort_cmp_func_t。

被 CGO 處理后的 Sort 函數(shù)的類型如下：

func Sort(
    base unsafe.Pointer, num, size _Ctype_size_t,
    cmp _Ctype_qsort_cmp_func_t,
)

這樣將會導(dǎo)致包外部用于無法構(gòu)造 _Ctype_size_t 和 _Ctype_qsort_cmp_func_t 類型的參數(shù)而無法使用 Sort 函數(shù)。因此，導(dǎo)出的 Sort 函數(shù)的參數(shù)和返回值要避免對虛擬 C 包的依賴。

重新調(diào)整 Sort 函數(shù)的參數(shù)類型和實現(xiàn)如下：

/*
#include <stdlib.h>

typedef int (*qsort_cmp_func_t)(const void* a, const void* b);
*/
import "C"
import "unsafe"

type CompareFunc C.qsort_cmp_func_t

func Sort(base unsafe.Pointer, num, size int, cmp CompareFunc) {
    C.qsort(base, C.size_t(num), C.size_t(size), C.qsort_cmp_func_t(cmp))
}

我們將虛擬 C 包中的類型通過 Go 語言類型代替，在內(nèi)部調(diào)用 C 函數(shù)時重新轉(zhuǎn)型為 C 函數(shù)需要的類型。因此外部用戶將不再依賴 qsort 包內(nèi)的虛擬 C 包。

以下代碼展示的 Sort 函數(shù)的使用方式：

package main

//extern int go_qsort_compare(void* a, void* b);
import "C"

import (
    "fmt"
    "unsafe"

    qsort "."
)

//export go_qsort_compare
func go_qsort_compare(a, b unsafe.Pointer) C.int {
    pa, pb := (*C.int)(a), (*C.int)(b)
    return C.int(*pa - *pb)
}

func main() {
    values := []int32{42, 9, 101, 95, 27, 25}

    qsort.Sort(unsafe.Pointer(&values[0]),
        len(values), int(unsafe.Sizeof(values[0])),
        qsort.CompareFunc(C.go_qsort_compare),
    )
    fmt.Println(values)
}

為了使用 Sort 函數(shù)，我們需要將 Go 語言的切片取首地址、元素個數(shù)、元素大小等信息作為調(diào)用參數(shù)，同時還需要提供一個 C 語言規(guī)格的比較函數(shù)。 其中 go_qsort_compare 是用 Go 語言實現(xiàn)的，并導(dǎo)出到 C 語言空間的函數(shù)，用于 qsort 排序時的比較函數(shù)。

目前已經(jīng)實現(xiàn)了對 C 語言的 qsort 初步包裝，并且可以通過包的方式被其它用戶使用。但是 qsort.Sort 函數(shù)已經(jīng)有很多不便使用之處：用戶要提供 C 語言的比較函數(shù)，這對許多 Go 語言用戶是一個挑戰(zhàn)。下一步我們將繼續(xù)改進 qsort 函數(shù)的包裝函數(shù)，嘗試通過閉包函數(shù)代替 C 語言的比較函數(shù)。

消除用戶對 CGO 代碼的直接依賴。

2.6.3 改進：閉包函數(shù)作為比較函數(shù)

在改進之前我們先回顧下 Go 語言 sort 包自帶的排序函數(shù)的接口：

func Slice(slice interface{}, less func(i, j int) bool)

標準庫的 sort.Slice 因為支持通過閉包函數(shù)指定比較函數(shù)，對切片的排序非常簡單：

import "sort"

func main() {
    values := []int32{42, 9, 101, 95, 27, 25}

    sort.Slice(values, func(i, j int) bool {
        return values[i] < values[j]
    })

    fmt.Println(values)
}

我們也嘗試將 C 語言的 qsort 函數(shù)包裝為以下格式的 Go 語言函數(shù)：

package qsort

func Sort(base unsafe.Pointer, num, size int, cmp func(a, b unsafe.Pointer) int)

閉包函數(shù)無法導(dǎo)出為 C 語言函數(shù)，因此無法直接將閉包函數(shù)傳入 C 語言的 qsort 函數(shù)。 為此我們可以用 Go 構(gòu)造一個可以導(dǎo)出為 C 語言的代理函數(shù)，然后通過一個全局變量臨時保存當前的閉包比較函數(shù)。

代碼如下：

var go_qsort_compare_info struct {
    fn func(a, b unsafe.Pointer) int
    sync.Mutex
}

//export _cgo_qsort_compare
func _cgo_qsort_compare(a, b unsafe.Pointer) C.int {
    return C.int(go_qsort_compare_info.fn(a, b))
}

其中導(dǎo)出的 C 語言函數(shù) _cgo_qsort_compare 是公用的 qsort 比較函數(shù)，內(nèi)部通過 go_qsort_compare_info.fn 來調(diào)用當前的閉包比較函數(shù)。

新的 Sort 包裝函數(shù)實現(xiàn)如下：

/*
#include <stdlib.h>

typedef int (*qsort_cmp_func_t)(const void* a, const void* b);
extern int _cgo_qsort_compare(void* a, void* b);
*/
import "C"

func Sort(base unsafe.Pointer, num, size int, cmp func(a, b unsafe.Pointer) int) {
    go_qsort_compare_info.Lock()
    defer go_qsort_compare_info.Unlock()

    go_qsort_compare_info.fn = cmp

    C.qsort(base, C.size_t(num), C.size_t(size),
        C.qsort_cmp_func_t(C._cgo_qsort_compare),
    )
}

每次排序前，對全局的 go_qsort_compare_info 變量加鎖，同時將當前的閉包函數(shù)保存到全局變量，然后調(diào)用 C 語言的 qsort 函數(shù)。

基于新包裝的函數(shù)，我們可以簡化之前的排序代碼：

func main() {
    values := []int32{42, 9, 101, 95, 27, 25}

    qsort.Sort(unsafe.Pointer(&values[0]), len(values), int(unsafe.Sizeof(values[0])),
        func(a, b unsafe.Pointer) int {
            pa, pb := (*int32)(a), (*int32)(b)
            return int(*pa - *pb)
        },
    )

    fmt.Println(values)
}

現(xiàn)在排序不再需要通過 CGO 實現(xiàn) C 語言版本的比較函數(shù)了，可以傳入 Go 語言閉包函數(shù)作為比較函數(shù)。 但是導(dǎo)入的排序函數(shù)依然依賴 unsafe 包，這是違背 Go 語言編程習(xí)慣的。

2.6.4 改進：消除用戶對 unsafe 包的依賴

前一個版本的 qsort.Sort 包裝函數(shù)已經(jīng)比最初的 C 語言版本的 qsort 易用很多，但是依然保留了很多 C 語言底層數(shù)據(jù)結(jié)構(gòu)的細節(jié)。 現(xiàn)在我們將繼續(xù)改進包裝函數(shù)，嘗試消除對 unsafe 包的依賴，并實現(xiàn)一個類似標準庫中 sort.Slice 的排序函數(shù)。

新的包裝函數(shù)聲明如下：

package qsort

func Slice(slice interface{}, less func(a, b int) bool)

首先，我們將 slice 作為接口類型參數(shù)傳入，這樣可以適配不同的切片類型。 然后切片的首個元素的地址、元素個數(shù)和元素大小可以通過 reflect 反射包從切片中獲取。

為了保存必要的排序上下文信息，我們需要在全局包變量增加要排序數(shù)組的地址、元素個數(shù)和元素大小等信息，比較函數(shù)改為 less：

var go_qsort_compare_info struct {
    base     unsafe.Pointer
    elemnum  int
    elemsize int
    less     func(a, b int) bool
    sync.Mutex
}

同樣比較函數(shù)需要根據(jù)元素指針、排序數(shù)組的開始地址和元素的大小計算出元素對應(yīng)數(shù)組的索引下標， 然后根據(jù) less 函數(shù)的比較結(jié)果返回 qsort 函數(shù)需要格式的比較結(jié)果。

//export _cgo_qsort_compare
func _cgo_qsort_compare(a, b unsafe.Pointer) C.int {
    var (
        // array memory is locked
        base     = uintptr(go_qsort_compare_info.base)
        elemsize = uintptr(go_qsort_compare_info.elemsize)
    )

    i := int((uintptr(a) - base) / elemsize)
    j := int((uintptr(b) - base) / elemsize)

    switch {
    case go_qsort_compare_info.less(i, j): // v[i] < v[j]
        return -1
    case go_qsort_compare_info.less(j, i): // v[i] > v[j]
        return +1
    default:
        return 0
    }
}

新的 Slice 函數(shù)的實現(xiàn)如下：

func Slice(slice interface{}, less func(a, b int) bool) {
    sv := reflect.ValueOf(slice)
    if sv.Kind() != reflect.Slice {
        panic(fmt.Sprintf("qsort called with non-slice value of type %T", slice))
    }
    if sv.Len() == 0 {
        return
    }

    go_qsort_compare_info.Lock()
    defer go_qsort_compare_info.Unlock()

    defer func() {
        go_qsort_compare_info.base = nil
        go_qsort_compare_info.elemnum = 0
        go_qsort_compare_info.elemsize = 0
        go_qsort_compare_info.less = nil
    }()

    // baseMem = unsafe.Pointer(sv.Index(0).Addr().Pointer())
    // baseMem maybe moved, so must saved after call C.fn
    go_qsort_compare_info.base = unsafe.Pointer(sv.Index(0).Addr().Pointer())
    go_qsort_compare_info.elemnum = sv.Len()
    go_qsort_compare_info.elemsize = int(sv.Type().Elem().Size())
    go_qsort_compare_info.less = less

    C.qsort(
        go_qsort_compare_info.base,
        C.size_t(go_qsort_compare_info.elemnum),
        C.size_t(go_qsort_compare_info.elemsize),
        C.qsort_cmp_func_t(C._cgo_qsort_compare),
    )
}

首先需要判斷傳入的接口類型必須是切片類型。然后通過反射獲取 qsort 函數(shù)需要的切片信息，并調(diào)用 C 語言的 qsort 函數(shù)。

基于新包裝的函數(shù)我們可以采用和標準庫相似的方式排序切片：

import (
    "fmt"

    qsort "."
)

func main() {
    values := []int64{42, 9, 101, 95, 27, 25}

    qsort.Slice(values, func(i, j int) bool {
        return values[i] < values[j]
    })

    fmt.Println(values)
}

為了避免在排序過程中，排序數(shù)組的上下文信息 go_qsort_compare_info 被修改，我們進行了全局加鎖。 因此目前版本的 qsort.Slice 函數(shù)是無法并發(fā)執(zhí)行的，讀者可以自己嘗試改進這個限制。