c - 高效的列表压缩

Question

假设您有一个无符号整数列表。假设某些元素等于 0，并且您想将它们推回。目前我使用此代码(列表是指向大小为 n 的无符号整数列表的指针

for (i = 0; i < n; ++i) 
{    
    if (list[i])
        continue;
    int j;
    for (j = i + 1; j < n && !list[j]; ++j);
    int z;
    for (z = j + 1; z < n && list[z]; ++z);
    if (j == n)
        break;
    memmove(&(list[i]), &(list[j]), sizeof(unsigned int) * (z - j)));
    int s = z - j + i;
    for(j = s; j < z; ++j) 
        list[j] = 0;
    i = s - 1;
} 

你能想出一种更有效的方法来执行这项任务吗？

该片段纯属理论，在生产代码中，列表的每个元素都是一个 64 字节的结构

编辑:
我会发布我的解决方案。非常感谢乔纳森·莱弗勒。

void RemoveDeadParticles(int * list, int * n)
{
    int i, j = *n - 1;
    for (; j >= 0 && list[j] == 0; --j);
    for (i = 0; i <= j; ++i)
    {   
        if (list[i])
            continue;
        memcpy(&(list[i]), &(list[j]), sizeof(int));
        list[j] = 0;
        for (; j >= 0 && list[j] == 0; --j);
        if (i == j)
            break;
    }   

    *n = i;
}

Answer 1

下面的代码实现了我在评论中概述的线性算法:

There's a straight linear O(N) algorithm if you're careful; your's is O(N²). Given that order is irrelevant, every time you encounter a zero going forwards through the array, you swap it with the last element that might not be a zero. That's one pass through the array. Care is required on the boundary conditions.

需要小心； list3[] 的酸性测试在测试代​​码中引起了悲伤，直到我得到了正确的边界条件。请注意，大小为 0 或 1 的列表的顺序已经正确。

#include <stdio.h>
#define DIM(x)  (sizeof(x)/sizeof(*(x)))

extern void shunt_zeroes(int *list, size_t n);

void shunt_zeroes(int *list, size_t n)
{
    if (n > 1)
    {
        size_t tail = n;
        for (size_t i = 0; i < tail - 1; i++)
        {
            if (list[i] == 0)
            {
                while (--tail > i + 1 && list[tail] == 0)
                    ;
                if (tail > i)
                {
                    int t = list[i];
                    list[i] = list[tail];
                    list[tail] = t;
                }
            }
        }
    }
}

static void dump_list(const char *tag, int *list, size_t n)
{
    printf("%-8s", tag);
    for (size_t i = 0; i < n; i++)
    {
        printf("%d ", list[i]);
    }
    putchar('\n');
    fflush(0);
}

static void test_list(int *list, size_t n)
{
    dump_list("Before:", list, n);
    shunt_zeroes(list, n);
    dump_list("After:", list, n);
}

int main(void)
{
    int list1[] = { 1, 0, 2, 0, 3, 0, 4, 0, 5 };
    int list2[] = { 1, 2, 2, 0, 3, 0, 4, 0, 0 };
    int list3[] = { 0, 0, 0, 0, 0, 0, 0, 0, 0 };
    int list4[] = { 0, 1 };
    int list5[] = { 0, 0 };
    int list6[] = { 0 };
    test_list(list1, DIM(list1));
    test_list(list2, DIM(list2));
    test_list(list3, DIM(list3));
    test_list(list4, DIM(list4));
    test_list(list5, DIM(list5));
    test_list(list6, DIM(list6));
}

示例运行:

$ shuntzeroes
Before: 1 0 2 0 3 0 4 0 5 
After:  1 5 2 4 3 0 0 0 0 
Before: 1 2 2 0 3 0 4 0 0 
After:  1 2 2 4 3 0 0 0 0 
Before: 0 0 0 0 0 0 0 0 0 
After:  0 0 0 0 0 0 0 0 0 
Before: 0 1 
After:  1 0 
Before: 0 0 
After:  0 0 
Before: 0 
After:  0 
$

代码的复杂性

我已经断言问题和 answer 中的原始代码UmNyobe 是 O(N2)，但这是 O(N)。但是，在所有三种情况下，循环中都有一个循环。当其他答案为 O(N2) 时，为什么这个答案是线性的？

好问题!

不同之处在于我的代码中的内部循环向后扫描数组，找到一个非零值来与刚刚找到的零交换。这样做，它减少了外部循环要做的工作。所以， i索引向前扫描一次， tail index 向后扫描一次，直到两者在中间相遇。相比之下，在原始代码中，内部循环从当前索引开始，每次都向前工作到结束，这导致了二次行为。

展示复杂性

UmNyobe 和 Argeman 都声称 code在 UmNyobe的答案是线性的，O(N)，而不是二次的，O(N2)，正如我在对答案的评论中断言的那样。鉴于两种相反的观点，我编写了一个程序来检查我的断言。

这是一个充分证明了这一点的测试结果。 "timer.h" 描述的代码是我的平台中立计时接口(interface)；它的代码可以根据要求提供(见我的 profile)。该测试是在配备 2.3 GHz Intel Core i7、Mac OS X 10.7.5、GCC 4.7.1 的 MacBook Pro 上进行的。

我对 UmNyobe 的代码所做的唯一更改是将数组索引从 int 更改为至 size_t使外部函数接口(interface)和我的一样，并且为了内部的一致性。

测试代码包括一个热身练习，以显示这些函数产生相同的答案； UmNyobe 的答案保留了数组中的顺序，而我的则没有。我已经从计时数据中省略了这些信息。

$ make on2
gcc -O3 -g -I/Users/jleffler/inc -std=c99 -Wall -Wextra -L/Users/jleffler/lib/64 on2.c -ljl -o on2
$

定时

第 1 组:在没有 UmNyobe 修正算法的旧版本测试工具上。

shunt_zeroes:        100    0.000001
shunt_zeroes:       1000    0.000005
shunt_zeroes:      10000    0.000020
shunt_zeroes:     100000    0.000181
shunt_zeroes:    1000000    0.001468
pushbackzeros:       100    0.000001
pushbackzeros:      1000    0.000086
pushbackzeros:     10000    0.007003
pushbackzeros:    100000    0.624870
pushbackzeros:   1000000   46.928721
shunt_zeroes:        100    0.000000
shunt_zeroes:       1000    0.000002
shunt_zeroes:      10000    0.000011
shunt_zeroes:     100000    0.000113
shunt_zeroes:    1000000    0.000923
pushbackzeros:       100    0.000001
pushbackzeros:      1000    0.000097
pushbackzeros:     10000    0.007077
pushbackzeros:    100000    0.628327
pushbackzeros:   1000000   41.512151

机器上最多只有非常轻的背景负载；例如，我暂停了通常在后台运行的 Boinc 计算。详细的时间安排并没有我想的那么稳定，但结论很清楚。

我的算法是 O(N)

UmNyobe 的(原始)算法为 O(N2)

第 2 组:使用 UmNyobe 的修正算法

还包括 Patrik 的前后算法，以及 Wildplasser 的算法(参见下面的源代码)；测试程序从 on2 重命名至 timezeromoves .

$ ./timezeromoves -c -m 100000 -n 1
shunt_zeroes: (Jonathan)
shunt_zeroes:        100    0.000001
shunt_zeroes:       1000    0.000003
shunt_zeroes:      10000    0.000018
shunt_zeroes:     100000    0.000155
RemoveDead: (Patrik)
RemoveDead:          100    0.000001
RemoveDead:         1000    0.000004
RemoveDead:        10000    0.000018
RemoveDead:       100000    0.000159
pushbackzeros2: (UmNyobe)
pushbackzeros2:      100    0.000001
pushbackzeros2:     1000    0.000005
pushbackzeros2:    10000    0.000031
pushbackzeros2:   100000    0.000449
list_compact: (Wildplasser)
list_compact:        100    0.000004
list_compact:       1000    0.000005
list_compact:      10000    0.000036
list_compact:     100000    0.000385
shufflezeroes: (Patrik)
shufflezeroes:       100    0.000003
shufflezeroes:      1000    0.000069
shufflezeroes:     10000    0.006685
shufflezeroes:    100000    0.504551
pushbackzeros: (UmNyobe)
pushbackzeros:       100    0.000003
pushbackzeros:      1000    0.000126
pushbackzeros:     10000    0.011719
pushbackzeros:    100000    0.480458
$

这表明 UmNyobe 的修正算法是 O(N)，其他解决方案也是如此。原始代码显示为 O(N2)，UmNyobe 的原始算法也是如此。

来源

这是修改后的测试程序(重命名为 testzeromoves.c )。算法实现位于顶部。测试工具在评论“测试工具”之后。该命令可以进行检查或计时或两者(默认)；它默认进行两次迭代；默认情况下，它的大小可达一百万个条目。您可以使用 -c省略检查的标志， -t标志省略计时， -n用于指定迭代次数的标志，以及 -m标志来指定最大尺寸。过百万要谨慎；您可能会遇到 VLA(可变长度数组)炸毁堆栈的问题。修改代码以使用 malloc() 很容易和 free()反而;不过，这似乎没有必要。

#include <string.h>

#define MAX(x, y)   (((x) > (y)) ? (x) : (y))

extern void shunt_zeroes(int *list, size_t n);
extern void pushbackzeros(int *list, size_t n);
extern void pushbackzeros2(int *list, size_t n);
extern void shufflezeroes(int *list, size_t n);
extern void RemoveDead(int *list, size_t n);
extern void list_compact(int *arr, size_t cnt);

void list_compact(int *arr, size_t cnt)
{
    size_t dst,src,pos;

    /* Skip leading filled area; find start of blanks */
    for (pos=0; pos < cnt; pos++) {
        if ( !arr[pos] ) break;
    }
    if (pos == cnt) return;

    for(dst= pos; ++pos < cnt; ) { 
        /* Skip blanks; find start of filled area */
        if ( !arr[pos] ) continue;

        /* Find end of filled area */
        for(src = pos; ++pos < cnt; ) {
            if ( !arr[pos] ) break;
        }   
        if (pos > src) {
            memmove(arr+dst, arr+src, (pos-src) * sizeof arr[0] );
            dst += pos-src;
        }   
    }
}

/* Cannot change j to size_t safely; algorithm relies on it going negative */
void RemoveDead(int *list, size_t n)
{
    int i, j = n - 1;
    for (; j >= 0 && list[j] == 0; --j)
        ;
    for (i = 0; i <= j; ++i)
    {   
        if (list[i])
            continue;
        memcpy(&(list[i]), &(list[j]), sizeof(int));
        list[j] = 0;
        for (; j >= 0 && list[j] == 0; --j);
        if (i == j)
            break;
    }   
}

void shufflezeroes(int *list, size_t n)
{
    for (size_t i = 0; i < n; ++i) 
    {    
        if (list[i])
            continue;
        size_t j;
        for (j = i + 1; j < n && !list[j]; ++j)
            ;
        size_t z;
        for (z = j + 1; z < n && list[z]; ++z)
            ;
        if (j == n)
            break;
        memmove(&(list[i]), &(list[j]), sizeof(int) * (z - j));
        size_t s = z - j + i;
        for(j = s; j < z; ++j) 
            list[j] = 0;
        i = s - 1;
    } 
}

static int nextZero(int* list, size_t start, size_t n){
   size_t i = start;
   while(i < n && list[i])
        i++;
   return i;
}

static int nextNonZero(int* list, size_t start, size_t n){
   size_t i = start;
   while(i < n && !list[i])
        i++;
   return i;
}

void pushbackzeros(int* list, size_t n){
    size_t i = 0;
    size_t j = 0;
    while(i < n && j < n){
         i = nextZero(list,i, n);
         j = nextNonZero(list,i, n);
         if(i >= n || j >=n)
             return;
         list[i] = list[j];
         list[j] = 0;
    }
}

/* Amended algorithm */
void pushbackzeros2(int* list, size_t n){
    size_t i = 0;
    size_t j = 0;
    while(i < n && j < n){
         i = nextZero(list, i, n);
         j = nextNonZero(list, MAX(i,j), n);
         if(i >= n || j >=n)
             return;
         list[i] = list[j];
         list[j] = 0;
    }
}

void shunt_zeroes(int *list, size_t n)
{
    if (n > 1)
    {
        size_t tail = n;
        for (size_t i = 0; i < tail - 1; i++)
        {
            if (list[i] == 0)
            {
                while (--tail > i + 1 && list[tail] == 0)
                    ;
                if (tail > i)
                {
                    int t = list[i];
                    list[i] = list[tail];
                    list[tail] = t;
                }
            }
        }
    }
}

/* Test Harness */

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include "timer.h"

#define DIM(x)      (sizeof(x)/sizeof(*(x)))

typedef void (*Shunter)(int *list, size_t n);

typedef struct FUT      /* FUT = Function Under Test */
{
    Shunter function;
    const char *name;
    const char *author;
} FUT;

static int tflag = 1;   /* timing */
static int cflag = 1;   /* checking */
static size_t maxsize = 1000000;

static void dump_list(const char *tag, int *list, size_t n)
{
    printf("%-8s", tag);
    for (size_t i = 0; i < n; i++)
    {
        printf("%d ", list[i]);
    }
    putchar('\n');
    fflush(0);
}

static void test_list(int *list, size_t n, Shunter s)
{
    dump_list("Before:", list, n);
    (*s)(list, n);
    dump_list("After:", list, n);
}

static void list_of_tests(const FUT *f)
{
    int list1[] = { 1, 0, 2, 0, 3, 0, 4, 0, 5 };
    int list2[] = { 1, 2, 2, 0, 3, 0, 4, 0, 0 };
    int list3[] = { 0, 0, 0, 0, 0, 0, 0, 0, 0 };
    int list4[] = { 0, 1 };
    int list5[] = { 0, 0 };
    int list6[] = { 0 };

    test_list(list1, DIM(list1), f->function);
    test_list(list2, DIM(list2), f->function);
    test_list(list3, DIM(list3), f->function);
    test_list(list4, DIM(list4), f->function);
    test_list(list5, DIM(list5), f->function);
    test_list(list6, DIM(list6), f->function);
}

static void test_timer(int *list, size_t n, const FUT *f)
{
    Clock t;
    clk_init(&t);
    clk_start(&t);
    f->function(list, n);
    clk_stop(&t);
    char buffer[32];
    printf("%-15s  %7zu  %10s\n", f->name, n, clk_elapsed_us(&t, buffer, sizeof(buffer)));
    fflush(0);
}

static void gen_test(size_t n, const FUT *f)
{
    int list[n];
    for (size_t i = 0; i < n/2; i += 2)
    {
        list[2*i+0] = i;
        list[2*i+1] = 0;
    }   
    test_timer(list, n, f);
}

static void timed_run(const FUT *f)
{
    printf("%s (%s)\n", f->name, f->author);
    if (cflag)
        list_of_tests(f);
    if (tflag)
    {
        for (size_t n = 100; n <= maxsize; n *= 10)
            gen_test(n, f);
    }
}

static const char optstr[] = "cm:n:t";
static const char usestr[] = "[-ct][-m maxsize][-n iterations]";

int main(int argc, char **argv)
{
    FUT functions[] =
    {
        { shunt_zeroes,   "shunt_zeroes:",   "Jonathan"    },   /* O(N) */
        { RemoveDead,     "RemoveDead:",     "Patrik"      },   /* O(N) */
        { pushbackzeros2, "pushbackzeros2:", "UmNyobe"     },   /* O(N) */
        { list_compact,   "list_compact:",   "Wildplasser" },   /* O(N) */
        { shufflezeroes,  "shufflezeroes:",  "Patrik"      },   /* O(N^2) */
        { pushbackzeros,  "pushbackzeros:",  "UmNyobe"     },   /* O(N^2) */
    };
    enum { NUM_FUNCTIONS = sizeof(functions)/sizeof(functions[0]) };
    int opt;
    int itercount = 2;

    while ((opt = getopt(argc, argv, optstr)) != -1)
    {
        switch (opt)
        {
        case 'c':
            cflag = 0;
            break;
        case 't':
            tflag = 0;
            break;
        case 'n':
            itercount = atoi(optarg);
            break;
        case 'm':
            maxsize = strtoull(optarg, 0, 0);
            break;
        default:
            fprintf(stderr, "Usage: %s %s\n", argv[0], usestr);
            return(EXIT_FAILURE);
        }
    }

    for (int i = 0; i < itercount; i++)
    {
        for (int j = 0; j < NUM_FUNCTIONS; j++)
            timed_run(&functions[j]);
        if (tflag == 0)
            break;
        cflag = 0;  /* Don't check on subsequent iterations */
    }

    return 0;
}