有多个问题:
这是修改后的版本:
// function for inserting the digit in the last node
void insert_into_bin(int data, int rad) {
struct node *new_node = make_new_node(data);
struct node *temp = radix[rad];
if (temp == NULL) {
radix[rad] = new_node;
} else {
while (temp->next != NULL) {
temp = temp->next;
}
temp->next = new_node;
}
}
// function for radix sort
void radix_sort(int *arr, int n) {
int i, j = 0, k, pass, digit, div = 1;
pass = get_max_digit(arr, n);
for (i = 1; i <= pass; i++) {
printf("\ndiv = %d\n\n", div);
for (j = 0; j < n; j++) {
// getting the corresponding digit
digit = (arr[j] / div) % 10;
// inserting into the bin
insert_into_bin(arr[j], digit);
}
// now multiplying div with 10 and storing it in div
div *= 10;
// Now the list is sorted in the array of linked list
// Time to retrieve them
j = 0;
for (k = 0; k < 10; k++) {
struct node *temp = radix[k];
radix[k] = NULL;
while (temp != NULL) {
struct node *next = temp->next;
arr[j++] = temp->data;
free(temp);
temp = next;
}
}
}
}
研究您的方法,恐怕复杂度远高于预期的 O(n) 或更准确地说是 O(n.log(max_value))。以下是一些令人担忧的问题:
这里有一个仪表版本来说明这一点:
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#define DIVISOR 10
// function for radix sort
void radix_sort(int *arr, int n) {
int i, j, k, div, max;
struct node {
int data;
struct node *next;
} *radix[DIVISOR];
if (n < 2)
return;
for (i = 0; i < DIVISOR; i++)
radix[i] = NULL;
// This loop is for finding the maximum number
max = arr[0];
for (i = 1; i < n; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
div = 1;
while (max > 0) {
for (j = 0; j < n; j++) {
// allocate a new node
struct node *new_node = (struct node *)malloc(sizeof(struct node));
struct node *temp;
new_node->next = NULL;
new_node->data = arr[j];
// getting the corresponding digit
int digit = (arr[j] / div) % DIVISOR;
// inserting into the bin
temp = radix[digit];
if (temp == NULL) {
radix[digit] = new_node;
} else {
while (temp->next != NULL) {
temp = temp->next;
}
temp->next = new_node;
}
}
// now multiplying div with 10 and storing it in div
div *= DIVISOR;
max /= DIVISOR;
// Now the list is sorted in the array of linked list
// Time to retrieve them
j = 0;
for (k = 0; k < DIVISOR; k++) {
struct node *temp = radix[k];
radix[k] = NULL;
while (temp != NULL) {
struct node *next = temp->next;
arr[j++] = temp->data;
free(temp);
temp = next;
}
}
}
}
// driver function
int main() {
double last_clock = 0;
for (int p = 0; p < 28; p++) {
int n = 1 << p;
int *arr = malloc(sizeof(*arr) * n);
unsigned long long sum1 = 0, sum2 = 0;
for (int i = 0; i < n; i++) {
arr[i] = rand() % n;
sum1 += arr[i];
}
clock_t t = -clock();
radix_sort(arr, n);
t += clock();
sum2 = arr[0];
for (int i = 1; i < n; i++) {
sum2 += arr[i];
if (arr[i-1] > arr[i]) {
printf("%d: out of order at %d\n", n, i);
return 1;
}
}
if (sum1 != sum2) {
printf("%d: checksum mismatch: %llu != %llu\n", n, sum1, sum2);
return 1;
}
printf("%d: %.3fms ratio=%.2f\n",
n, t * 1000.0 / CLOCKS_PER_SEC,
last_clock > 0 ? t / last_clock : 0);
if (t > 7 * CLOCKS_PER_SEC)
break;
last_clock = t;
}
return 0;
}
输出清楚地显示了二次时间复杂度:
1: 0.002ms ratio=0.00
2: 0.007ms ratio=3.50
4: 0.001ms ratio=0.14
8: 0.002ms ratio=2.00
16: 0.006ms ratio=3.00
32: 0.010ms ratio=1.67
64: 0.021ms ratio=2.10
128: 0.067ms ratio=3.19
256: 0.134ms ratio=2.00
512: 0.314ms ratio=2.34
1024: 1.527ms ratio=4.86
2048: 3.508ms ratio=2.30
4096: 12.916ms ratio=3.68
8192: 56.545ms ratio=4.38
16384: 399.561ms ratio=7.07
32768: 1793.814ms ratio=4.49
65536: 7419.397ms ratio=4.14
您可以为列表头和尾使用单独的数组来解决瓶颈:
#define DIVISOR 10
// function for radix sort
void radix_sort(int *arr, int n) {
int i, j, k, div, max;
struct node {
int data;
struct node *next;
} *radix[DIVISOR], *tail[DIVISOR];
if (n < 2)
return;
for (i = 0; i < DIVISOR; i++)
tail[i] = radix[i] = NULL;
// This loop is for finding the maximum number
max = arr[0];
for (i = 1; i < n; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
div = 1;
while (max > 0) {
for (j = 0; j < n; j++) {
// allocate a new node
struct node *new_node = (struct node *)malloc(sizeof(struct node));
new_node->next = NULL;
new_node->data = arr[j];
// getting the corresponding digit
int digit = (arr[j] / div) % DIVISOR;
// inserting into the bin
if (radix[digit] == NULL) {
tail[digit] = radix[digit] = new_node;
} else {
tail[digit] = tail[digit]->next = new_node;
}
}
// now multiplying div with 10 and storing it in div
div *= DIVISOR;
max /= DIVISOR;
// Now the list is sorted in the array of linked list
// Time to retrieve them
j = 0;
for (k = 0; k < DIVISOR; k++) {
struct node *temp = radix[k];
tail[k] = radix[k] = NULL;
while (temp != NULL) {
struct node *next = temp->next;
arr[j++] = temp->data;
free(temp);
temp = next;
}
}
}
}
这解决了问题并允许快速排序更大的数组:
1: 0.003ms ratio=0.00
2: 0.007ms ratio=2.33
4: 0.001ms ratio=0.14
8: 0.002ms ratio=2.00
16: 0.005ms ratio=2.50
32: 0.010ms ratio=2.00
64: 0.018ms ratio=1.80
128: 0.054ms ratio=3.00
256: 0.150ms ratio=2.78
512: 0.236ms ratio=1.57
1024: 0.535ms ratio=2.27
2048: 1.469ms ratio=2.75
4096: 2.397ms ratio=1.63
8192: 4.536ms ratio=1.89
16384: 11.487ms ratio=2.53
32768: 26.473ms ratio=2.30
65536: 58.251ms ratio=2.20
131072: 120.385ms ratio=2.07
262144: 247.752ms ratio=2.06
524288: 535.602ms ratio=2.16
1048576: 1243.067ms ratio=2.32
2097152: 2651.775ms ratio=2.13
4194304: 5345.552ms ratio=2.02
8388608: 10962.960ms ratio=2.05
malloc 开销可以通过分配 2 个辅助数组来解决:
这些数组被分配一次,每次都使用并在最后被释放。
这里是修改后的radix_sort()函数:
#define DIVISOR 10
// function for radix sort
int radix_sort(int *arr, int n) {
int i, j, k, div, max;
int radix[DIVISOR], tail[DIVISOR];
int *data;
int *next;
if (n < 2)
return 0;
data = malloc(sizeof(*data) * n);
if (data == NULL)
return -1;
next = malloc(sizeof(*next) * (n + 1));
if (next == NULL) {
free(data);
return -1;
}
for (k = 0; k < DIVISOR; k++)
tail[k] = radix[k] = n;
// This loop is for finding the maximum number
max = arr[0];
for (i = 1; i < n; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
div = 1;
while (max > 0) {
for (j = 0; j < n; j++) {
data[j] = arr[j];
// getting the corresponding digit
int digit = (arr[j] / div) % DIVISOR;
// inserting into the bin
if (radix[digit] == n) {
tail[digit] = radix[digit] = j;
} else {
tail[digit] = next[tail[digit]] = j;
}
}
// now multiplying div with 10 and storing it in div
div *= DIVISOR;
max /= DIVISOR;
// Now the list is sorted in the array of linked list
// Time to retrieve them
j = 0;
for (k = 0; k < DIVISOR; k++) {
int temp = radix[k];
tail[k] = radix[k] = next[tail[k]] = n;
while (temp != n) {
arr[j++] = data[temp];
temp = next[temp];
}
}
}
free(data);
free(next);
return 0;
}
时间显示性能提高了 12 倍,同时具有相同的准线性复杂度:
1: 0.002ms ratio=0.00
2: 0.016ms ratio=8.00
4: 0.001ms ratio=0.06
8: 0.002ms ratio=2.00
16: 0.003ms ratio=1.50
32: 0.003ms ratio=1.00
64: 0.004ms ratio=1.33
128: 0.010ms ratio=2.50
256: 0.010ms ratio=1.00
512: 0.022ms ratio=2.20
1024: 0.055ms ratio=2.50
2048: 0.138ms ratio=2.51
4096: 0.215ms ratio=1.56
8192: 0.457ms ratio=2.13
16384: 1.084ms ratio=2.37
32768: 1.913ms ratio=1.76
65536: 3.426ms ratio=1.79
131072: 8.062ms ratio=2.35
262144: 26.584ms ratio=3.30
524288: 58.333ms ratio=2.19
1048576: 104.538ms ratio=1.79
2097152: 197.730ms ratio=1.89
4194304: 390.818ms ratio=1.98
8388608: 810.972ms ratio=2.08
16777216: 2057.180ms ratio=2.54
33554432: 3611.793ms ratio=1.76
67108864: 7858.513ms ratio=2.18
通过避免分裂可以获得进一步的改进:
- 对
DIVISOR 使用 2 的幂并使用 shift 和 mask 提取数字
- 使用单独的目标索引,而不是相同数字的条目链接列表。
这里是一个插图,将DIVISOR增加到64,使用了上一版本一半的内存:
#define SHIFT 6
#define MASK ((1 << SHIFT) - 1)
#define DIVISOR (1 << SHIFT)
// function for radix sort
int radix_sort(int *arr, int n) {
int i, j, k, shift, max;
int count[DIVISOR], index[DIVISOR];
int *data;
if (n < 2)
return 0;
data = malloc(sizeof(*data) * n);
if (data == NULL)
return -1;
// This loop is for finding the maximum number
max = arr[0];
for (i = 1; i < n; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
shift = 0;
while (max > 0) {
for (k = 0; k < DIVISOR; k++)
count[k] = 0;
for (j = 0; j < n; j++) {
// copy the value
data[j] = arr[j];
// getting the corresponding digit
int digit = (data[j] >> shift) & MASK;
// increase this digit's count
count[digit] += 1;
}
// compute the starting indexes
index[0] = 0;
for (k = 1; k < DIVISOR; k++) {
index[k] = index[k - 1] + count[k - 1];
}
// dispatch the values in the order of their current digit
for (j = 0; j < n; j++) {
int val = data[j];
// getting the corresponding digit
int digit = (val >> shift) & MASK;
arr[index[digit]++] = val;
}
// update shift and max
shift += SHIFT;
max >>= SHIFT;
}
free(data);
return 0;
}
输出显示另一个显着改进(7x 更快):
1: 0.002ms ratio=0.00
2: 0.027ms ratio=13.50
4: 0.002ms ratio=0.07
8: 0.003ms ratio=1.50
16: 0.003ms ratio=1.00
32: 0.002ms ratio=0.67
64: 0.002ms ratio=1.00
128: 0.003ms ratio=1.50
256: 0.004ms ratio=1.33
512: 0.010ms ratio=2.50
1024: 0.013ms ratio=1.30
2048: 0.022ms ratio=1.69
4096: 0.046ms ratio=2.09
8192: 0.114ms ratio=2.48
16384: 0.221ms ratio=1.94
32768: 0.366ms ratio=1.66
65536: 0.864ms ratio=2.36
131072: 1.184ms ratio=1.37
262144: 2.884ms ratio=2.44
524288: 12.569ms ratio=4.36
1048576: 21.896ms ratio=1.74
2097152: 34.809ms ratio=1.59
4194304: 61.134ms ratio=1.76
8388608: 105.110ms ratio=1.72
16777216: 212.954ms ratio=2.03
33554432: 548.368ms ratio=2.58
67108864: 1025.203ms ratio=1.87
134217728: 2340.717ms ratio=2.28
总有改进的余地:如果调度阶段在data 和arr 数组之间交替,则可以省略在数字计数阶段复制数据。我试过了,但它似乎根本没有太大的效果,实际上在很多情况下会变慢。
始终对假定的改进进行基准测试,并确保它们仍然产生正确的结果:正确性总是胜过性能。
谈到正确性,需要注意的是,上述所有实现都只能处理正数。要处理int 类型的全部范围,需要使用unsigned 类型并根据它们与最小值的差异对值进行排序:
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#define SHIFT 8
#define MASK ((1 << SHIFT) - 1)
#define DIVISOR (1 << SHIFT)
// function for radix sort
int radix_sort(int *arr, int n) {
int min, max;
unsigned k, shift, range;
int j;
int count[DIVISOR], index[DIVISOR];
unsigned *data, *src, *dst, *temp;
if (n < 2)
return 0;
data = malloc(sizeof(*data) * n);
if (data == NULL)
return -1;
src = (unsigned *)arr;
dst = data;
// This loop is for finding the minimum and maximum numbers
min = max = src[0];
for (j = 1; j < n; j++) {
int val = arr[j];
if (min > val) min = val;
if (max < val) max = val;
}
range = (unsigned)max - (unsigned)min;
for (shift = 0; range > 0; shift += SHIFT, range >>= SHIFT) {
for (k = 0; k < DIVISOR; k++)
count[k] = 0;
for (j = 0; j < n; j++) {
// getting the corresponding digit
unsigned val = src[j] - (unsigned)min;
unsigned digit = (val >> shift) & MASK;
// increase this digit's count
count[digit] += 1;
}
// check for trivial case
for (k = 0; count[k] == 0; k++) {
index[k] = 0;
}
if (count[k] == n) {
// all digits are identical
continue;
}
// compute the remaining starting indexes
index[k] = 0;
for (k += 1; k < DIVISOR; k++) {
index[k] = index[k - 1] + count[k - 1];
}
// dispatch the values in the order of their current digit
for (j = 0; j < n; j++) {
unsigned val = src[j] - (unsigned)min;
// getting the corresponding digit
unsigned digit = (val >> shift) & MASK;
dst[index[digit]++] = src[j];
}
// swap source and destination arrays
temp = src;
src = dst;
dst = temp;
}
if (src == data) {
// copy data back to the original array
memcpy(arr, data, sizeof(*arr) * n);
}
free(data);
return 0;
}
// driver function
int main() {
double last_clock = 0;
for (int p = 0; p < 28; p++) {
int n = 1 << p;
int *arr = malloc(sizeof(*arr) * n);
unsigned long long sum1 = 0, sum2 = 0;
for (int i = 0; i < n; i++) {
// mix positive and negative values
int v = rand();
int val = (v >> 1) % n;
arr[i] = (v & 1) ? -val : val;
sum1 += arr[i];
}
clock_t t = -clock();
radix_sort(arr, n);
t += clock();
sum2 = arr[0];
for (int i = 1; i < n; i++) {
sum2 += arr[i];
if (arr[i-1] > arr[i]) {
printf("%d: out of order at %d\n", n, i);
return 1;
}
}
if (sum1 != sum2) {
printf("%d: checksum mismatch: %llu != %llu\n", n, sum1, sum2);
return 1;
}
printf("%d: %.3fms ratio=%.2f\n",
n, t * 1000.0 / CLOCKS_PER_SEC,
last_clock > 0 ? t / last_clock : 0);
if (t > 7 * CLOCKS_PER_SEC)
break;
last_clock = t;
}
return 0;
}
这个小改动对整体性能影响不大,但提高了正确性:
1: 0.002ms ratio=0.00
2: 0.010ms ratio=5.00
4: 0.018ms ratio=1.80
8: 0.002ms ratio=0.11
16: 0.002ms ratio=1.00
32: 0.001ms ratio=0.50
64: 0.001ms ratio=1.00
128: 0.002ms ratio=2.00
256: 0.006ms ratio=3.00
512: 0.010ms ratio=1.67
1024: 0.014ms ratio=1.40
2048: 0.024ms ratio=1.71
4096: 0.049ms ratio=2.04
8192: 0.095ms ratio=1.94
16384: 0.201ms ratio=2.12
32768: 0.284ms ratio=1.41
65536: 0.969ms ratio=3.41
131072: 1.496ms ratio=1.54
262144: 3.034ms ratio=2.03
524288: 14.288ms ratio=4.71
1048576: 21.479ms ratio=1.50
2097152: 34.468ms ratio=1.60
4194304: 70.317ms ratio=2.04
8388608: 117.420ms ratio=1.67
16777216: 251.831ms ratio=2.14
33554432: 522.936ms ratio=2.08
67108864: 1076.859ms ratio=2.06
134217728: 2300.456ms ratio=2.14