Сортировка и объединение нескольких связанных списков с отсортированными подразделами

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 всегда обновляемым, чтобы указать на конечный элемент списка.

Перемещение каждого списка, чтобы найти элемент хвоста, вероятно, является медленной частью здесь, но, к сожалению, я не вижу способа избежать этого. (Если элементы списка не являются последовательными в памяти, обход списка имеет тенденцию требовать много загрузок кеша из ОЗУ. Шаблоны доступа при сортировке массива намного легче прогнозировать ЦП (на текущих архитектурах), поэтому часть сортировки массива вероятно, это не самая медленная часть - но, конечно, можно профилировать код с практическим набором данных, а также рассмотреть вопрос, нужна ли быстрее реализация сортировки, чем библиотеки C qsort())

---

О.П. уточнил, что каждый человек. список, зависающий от указателя в массиве указателей, может содержать один или несколько «блоков», то есть последовательные отсортированные прогоны. Они могут быть обнаружены изменением 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, или вам необходимо ограничить объем используемой памяти, то я бы использовал небольшой мин -heap из

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

элементов, введенного пользователя с помощью count, число элементов в этом подсписке.

Идея состоит в том, что всякий раз, когда вы извлекаете блок из ввода, вы добавляете его в мини-кучу, если есть место; в противном случае вы объедините его с корневым списком в мини-куче. Обратите внимание, что согласно правилам OP это «слияние» на самом деле является одной вставкой, так как каждый блок является последовательным; сначала нужно найти только точку вставки. Обновлен count, чтобы отразить количество элементов в корневом списке, и, таким образом, вы повторно наследуете мини-кучу.

Цель кучи заключается в том, чтобы объединить два самых маленьких блока, сохранив обход списка, чтобы найти точку вставки до минимума.

Когда все блоки были вставлены, вы берете корневой каталог, объединяете этот список с новым корневым списком и повторно heapify, уменьшая размер кучи на каждый каждый раз, пока не останется один список. Это окончательный список результатов.

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), но вторая, вероятно, быстрее, потому что она не перемещает данные примерно так же. Какой алгоритм вы хотите использовать, будет зависеть от того, насколько велики ваши списки и как часто вы выполняете слияние.