學校課程/平行程式

數據分割 區間分塊小技巧

contents

  1. 1. 前言
  2. 2. 數學分析
    1. 2.1. 直觀
    2. 2.2. 程序流
    3. 2.3. ceil
    4. 2.4. floor
    5. 2.5. 補充流
  3. 3. 測試

前言

運行平行程式之前，要將資料切成很多塊，每一塊之間獨立。從邏輯上看來有 row-based, column-based, matrix-based, tree-based 等平行方法。在許多方法中，最容易入手的就是 row-based。

假設有 $n$ 個元素，編號介於 $[0, n-1]$ 之間，要分成 $m$ 組。每一組得編號都要連續，以增加資料局部性 (data locality) 並且每一塊大小盡可能一樣，請問要怎麼分割才好？這樣的數學問題要做出來並不是件難事，但強迫症的人來說，在程式中零碎的判斷相當折騰。

數學分析

直觀

若有 $n$ 個元素分成 $m$ 組，每一個組至少有 $\lfloor n / m \rfloor$ 個元素，其中 $n \mod m$ 組會額外多一個元素。因此，對於第 $i$ 組滿足 $i < n \mod m$ 都會多一個元素。

1
2
3
4
5
6
7
8
9
10
void method0(int n, int m) {
    printf("In %s\n", __func__);
    int bsz = n/m, rm = n%m, sz;
    for (int i = 0, L = 0, R; i < m; i++) {
        sz = bsz + (i < rm);
        R = L + sz - 1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
        L = R + 1;
    }
}

程序流

這方式屬於懶惰程序員，根據整數的性質會得到刻度，在計算第 $i$ 個刻度採用 $\lfloor \frac{in}{m} \rfloor$，自然而然地可以與下一個刻度形成一個區間。不幸地，若 $m > n$ 時，有些刻度會重疊，導致重複計算的情況發生。

1
2
3
4
5
6
7
void method1(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L, R; i < m; i++) {
        L = i*n/m, R = (i+1)*n/m - 1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
    }
}

ceil

從《具體數學》第三章節的介紹，得到幾個簡單的數學公式

$$\begin{align} n = \left \lceil \frac{n}{m} \right \rceil + \left \lceil \frac{n-1}{m} \right \rceil + \cdots + \left \lceil \frac{n-m-1}{m} \right \rceil \end{align}$$

第 $i$ 團 ($0 \le i < m$)，要處理 $\left \lceil \frac{n-i}{m} \right \rceil$ 個元素。

1
2
3
4
5
6
7
8
9
void method2(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L = 0, R, sz; i < m; i++) {
        sz = (n + m-i-1) / m;
        R = L + sz-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, sz);
        L = R + 1;
    }
}

floor

又或者使用 floor 表示法

$$\begin{align} n = \left \lfloor \frac{n}{m} \right \rfloor + \left \lfloor \frac{n+1}{m} \right \rfloor + \cdots + \left \lfloor \frac{n+m-1}{m} \right \rfloor \end{align}$$

第 $i$ 團 ($0 \le i < m$)，要處理 $\left \lceil \frac{n+i}{m} \right \rceil$ 個元素。

1
2
3
4
5
6
7
8
9
void method3(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L = 0, R, sz; i < m; i++) {
        sz = (n + i) / m;
        R = L + sz-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, sz);
        L = R + 1;
    }
}

補充流

不管怎樣，前 $i$ 就拿取 $\lceil \frac{n}{m} \rceil$，最後一組拿少一點。

1
2
3
4
5
6
7
8
9
void method4(int n, int m) {
    printf("In %s\n", __func__);
    int bsz = (n + m-1)/m, sz;
    for (int i = 0, L = 0, R; i < m; i++) {
        L = i*bsz, R = L + bsz - 1;
        if (R >= n)	R = n-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
    }
}

測試

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
#include <bits/stdc++.h>
using namespace std;
void method0(int n, int m) {
    printf("In %s\n", __func__);
    int bsz = n/m, rm = n%m, sz;
    for (int i = 0, L = 0, R; i < m; i++) {
        sz = bsz + (i < rm);
        R = L + sz - 1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
        L = R + 1;
    }
}
void method1(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L, R; i < m; i++) {
        L = i*n/m, R = (i+1)*n/m - 1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
    }
}
void method2(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L = 0, R, sz; i < m; i++) {
        sz = (n + m-i-1) / m;
        R = L + sz-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, sz);
        L = R + 1;
    }
}
void method3(int n, int m) {
    printf("In %s\n", __func__);
    for (int i = 0, L = 0, R, sz; i < m; i++) {
        sz = (n + i) / m;
        R = L + sz-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, sz);
        L = R + 1;
    }
}
void method4(int n, int m) {
    printf("In %s\n", __func__);
    int bsz = (n + m-1)/m, sz;
    for (int i = 0, L = 0, R; i < m; i++) {
        L = i*bsz, R = L + bsz - 1;
        if (R >= n)	R = n-1;
        printf("%d len([%d, %d]) = %d\n", i, L, R, (R - L + 1));
    }
}
int main() {
    void (*FUNC[])(int, int) = {method0, method1, method2, method3, method4};
    int n, m;
    while (scanf("%d %d", &n, &m) == 2) {
        for (int i = 0; i < sizeof(FUNC)/sizeof(FUNC[0]); i++)
            (*FUNC[i])(n, m);
    }
    return 0;
}