使用排序的子部分对多个链表进行排序和合并

Question

我有一个numlists 链表数组。列表中的节点采用以下形式：

struct Edge
{
    int64_t blocknum;
    int64_t location;
    struct Edge *next;
};
typedef struct Edge edge;

我需要将所有列表合并成一个链表，该链表按location 升序排序。每个列表由节点具有相等blocknum 的块组成，并且这些块中的每一个都已排序。具有较大blocknum 值的列表块的所有位置值都大于具有较小blocknum 的块。子列表中的块已经在本地按blocknum 的顺序排序。这实际上意味着，这归结为按blocknum 升序对块进行排序，我不必太担心location，因为它会自行解决。您可以假设数组的 next 成员有效且已分配，或者显式声明为 NULL。

这是我想出的功能

edge *sort_edges(edge **unsorted, int numlists)
{
    edge *sorted_head = NULL;
    edge *sorted_current = NULL;
    edge *current_edge = NULL;
    edge *temp = NULL;
    int64_t blocknum;

    int i;
    int64_t minblock;
    int remaining = numlists;
    int first = 1;
    int minblock_index;
    while(remaining) //while there are still more lists to process
    {
        minblock = LLONG_MAX;
        temp = NULL;
        minblock_index = INT_MAX;
        remaining = numlists;
        for (i=0; i<numlists; i++) //loop over the list of head nodes to find the one with the smallest blocknum
        {
            if (!unsorted[i]) //when a lists is exhausted the lead node becomes NULL, and we decrement the counter
            {
                remaining--;
            } 
            else //a simple minimum finding algorithm
            {
                current_edge = unsorted[i];
                if (current_edge->blocknum < minblock)
                {
                    temp = current_edge;
                    minblock = current_edge->blocknum;
                    minblock_index = i;
                }
            }
        }
        if (remaining == 0)
        {
            break;
        }
        if (first) //if we have not yet set up the head of the list, we have to save a pointer to the head
        {
            sorted_head = temp;
            sorted_current = sorted_head;
            first = 0;
        }
        else 
        {
            sorted_current->next = temp;
        }
        blocknum = sorted_current->blocknum;
        while (sorted_current->blocknum == blocknum && sorted_current->next) //skip through to the end of the block so that the next section we append will go on the end
        {
            sorted_current = sorted_current->next;
        }
        unsorted[minblock_index] = sorted_current->next; //reset the head of the unsorted list to the node after the block
    }
    return sorted_head;
}

这行得通。我的问题是：

我可以在高效的排序算法方面做得更好吗？ （几乎可以肯定，我只是好奇人们在给定假设的情况下会提出什么排序问题）。

Answer 1

如果“块”是指从指针数组中的每个指针挂起的列表，那么

int compare_edge_blocknum(const void *e1, const void *e2)
{
    if (!e1 && !e2)
        return 0;
    else
    if (!e1)
        return +1;
    else
    if (!e2)
        return -1;
    else {
        const int64_t b1 = ((edge *)e1)->blocknum;
        const int64_t b2 = ((edge *)e2)->blocknum;
        return (b1 < b2) ? -1 :
               (b1 > b2) ? +1 : 0;
    }
}

edge *last_in_list(edge *list)
{
    if (list)
        while (list->next)
            list = list->next;
    return list;
}

edge *sort_edges(edge **array, size_t count)
{
    edge   root = { 0, 0, NULL };
    edge  *tail = &root;
    size_t i;

    if (!array || count < 1)
        return NULL;
    if (count == 1)
        return array[0];

    qsort(array, count, sizeof *array, compare_edge_blocknum);

    for (i = 0; i < count; i++)
        if (array[i]) {
            tail->next = array[i];
            tail = last_in_list(array[i]);
        }

    return root->next;
}

上面使用qsort()对指针数组进行排序，根据blocknum。我们使用root 作为结果列表的句柄。我们遍历指针数组，将每个非 NULL 指针附加到结果列表的tail，而tail 始终更新为指向列表的最后一个元素。

遍历每个列表以找到尾部元素可能是这里的缓慢部分，但不幸的是我没有看到任何避免它的方法。 （如果列表元素在内存中不连续，则列表遍历往往需要从 RAM 中加载许多缓存。数组排序时的访问模式对于 CPU 来说更容易预测（在当前架构上），因此数组排序部分可能不是最慢的部分——但当然您可以使用实际数据集分析代码，并考虑是否需要比 C 库qsort() 更快的排序实现。）

---

OP 澄清了每个单独的列表挂在指针数组中的一个指针上可能包含一个或多个“块”，即连续排序运行。这些可以通过改变blocknum来检测。

如果额外的内存使用不是问题，我会创建一个数组

typedef struct {
    int64_t  blocknum;
    edge    *head;
    edge    *tail;
} edge_block;

然后按blocknum排序，最后链接。保存指向第一个（头）和最后一个（尾）元素的指针意味着我们只扫描列表一次。在对 edge_block 数组进行排序后，对其进行简单的线性传递就足以将所有子列表链接到最终列表中。

例如（仅经过编译测试）：

#include <stdlib.h>
#include <stdint.h>
#include <errno.h>

typedef struct Edge edge;
struct Edge {
    int64_t      blocknum;
    int64_t      location;
    struct Edge *next;
};

typedef struct {
    int64_t      blocknum;
    struct Edge *head;
    struct Edge *tail;
} edge_block;

static int cmp_edge_block(const void *ptr1, const void *ptr2)
{
    const int64_t b1 = ((const edge_block *)ptr1)->blocknum;
    const int64_t b2 = ((const edge_block *)ptr2)->blocknum;
    return (b1 < b2) ? -1 :
           (b1 > b2) ? +1 : 0;
}

edge *sort_edges(edge **array, size_t count)
{
    edge_block *block = NULL;
    size_t      blocks = 0;
    size_t      blocks_max = 0;
    edge       *root, *curr;
    size_t      i;

    if (count < 1) {
        errno = 0;
        return NULL;
    }

    if (!array) {
        errno = EINVAL;
        return NULL;
    }

    for (i = 0; i < count; i++) {
        curr = array[i];

        while (curr) {

            if (blocks >= blocks_max) {
                edge_block *old = block;

                if (blocks < 512)
                    blocks_max = 1024;
                else
                if (blocks < 1048576)
                    blocks_max = ((blocks * 3 / 2) | 1023) + 1; 
                else
                    blocks_max = (blocks | 1048575) + 1048577;

                block = realloc(block, blocks_max * sizeof block[0]);
                if (!block) {
                    free(old);
                    errno = ENOMEM;
                    return NULL;
                }
            }

            block[blocks].blocknum = curr->blocknum;
            block[blocks].head = curr;

            while (curr->next && curr->next->blocknum == block[blocks].blocknum)
                curr = curr->next;

            block[blocks].tail = curr;
            blocks++;
            curr = curr->next;
        }
    }

    if (blocks < 1) {
        /* Note: block==NULL here, so no free(block) needed. */
        errno = 0;
        return NULL;
    }

    qsort(block, blocks, sizeof block[0], cmp_edge_block);

    root = block[0].head;
    curr = block[0].tail;
    for (i = 1; i < blocks; i++) {
        curr->next = block[i].head;
        curr = block[i].tail;
    }

    free(block);

    errno = 0;
    return root;
}

---

如果可能有非常多的blocknums，或者您需要限制使用的内存量，那么我会使用一个小的最小堆

typedef struct {
    size_t   count;
    edge    *head;
    edge    *tail;
} edge_block;

elements，以count 为键，该子列表中的元素数。

这个想法是，每当你从输入中提取一个块时，如果有空间，你就将它添加到最小堆中；否则，将它与最小堆中的根列表合并。请注意，根据 OP 的规则，这种“合并”实际上是一次插入，因为每个块都是连续的；只需要先找到插入点。 count 已更新以反映根列表中的元素数量，因此您重新堆放最小堆。

堆的目的是确保您合并两个最短的块，使遍历列表以找到插入点的次数最少。

当所有块都插入后，你取根，将该列表与新的根列表合并，然后重新堆，每次将堆的大小减一，直到剩下一个列表。这就是最终的结果列表。

Answer 2

据我了解，您有多个排序列表，您希望将它们合并在一起以创建一个排序列表。

执行此操作的一种常见方法是创建一个列表队列并不断合并对，将结果添加回队列，然后重复直到只剩下一个列表。例如：

listQueue = queue of lists to be merged
while listQueue.count > 1
{
    list1 = listQueue.dequeue
    list2 = listQueue.dequeue
    newList = new list
    // do standard merge here
    while (list1 != null && list2 != null)
    {
        if (list1.item <= list2.item)
        {
            newList.append(list1.item)
            list1 = list1.next
        }
        else
        {
            newList.append(list2.item)
            list2 = list2.next
        }
    }
    // clean up the stragglers, if any
    while (list1 != null)
    {
        newList.append(list1.item)
        list1 = list1.next
    }
    while (list2 != null)
    {
        newList.append(list2.item)
        list2 = list2.next
    }
    listQueue.enqueue(newList)
}
mergedList = listQueue.dequeue

这是一个很有吸引力的选择，因为它很简单，只需要很少的额外内存，而且效率相当高。

有一种可能更快的方法需要更多的内存（O(log k)，其中 k 是列表的数量），并且需要更多的编码。它涉及创建一个最小堆，其中包含每个列表中的第一项。您从堆中删除最低项，将其添加到新列表中，然后从列表中取出最低项所在的下一项，并将其插入到堆中。

这两种算法都是 O(n log k) 复杂度，但第二种可能更快，因为它不会移动太多数据。您要使用哪种算法取决于您的列表有多大以及您进行合并的频率。