Sortieren und Zusammenführen mehrerer verkettete Listen mit sortierten Unterabschnitten

Question

Ich habe ein Array von numlists verknüpften Listen. Knoten in den Listen sind von der Form:

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

ich alle Listen zu einer einzigen Liste verschmelzen müssen, die von location aufsteigend sortiert ist. Jede Liste besteht aus Blöcken, für die Knoten gleich blocknum sind, und jeder dieser Blöcke ist bereits sortiert. Listenblöcke mit größeren Werten von blocknum haben alle ihre Standortwerte größer als Blöcke mit kleineren blocknum. Blöcke in den Unterlisten sind bereits in der Reihenfolge blocknum lokal sortiert. Was praktisch bedeutet, dass dies auf das Sortieren von Blöcken durch blocknum in aufsteigender Reihenfolge hinausläuft, und ich muss mich nicht zu sehr um location kümmern, da das für sich selbst sorgen wird. Sie können davon ausgehen, dass das next Mitglied eines Arrays entweder gültig und zugewiesen ist oder explizit NULL deklariert ist.

Dies ist die Funktion, die ich mit

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; 
} 

Dies funktioniert kam. Meine Frage ist:

Kann ich besser in Bezug auf einen effizienten Sortieralgorithmus tun? (Fast sicher ja, ich bin nur neugierig, was die Leute für ein Sortierproblem mit den gegebenen Annahmen einfallen lassen).

Answer 1

Wenn von „Block“ Sie die Liste bedeuten, von jedem Zeiger in dem Zeigerfeld hängen, dann

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; 
} 

Die oben verwenden qsort() das Array von Zeigern zu sortieren, nach blocknum. Wir verwenden root als Handle für die resultierende Liste. Wir durchlaufen das Array von Zeigern und hängen jeden Nicht-Null-Zeiger an die tail der Ergebnisliste an, wobei tail immer aktualisiert wird, um auf das letzte Element der Liste zu zeigen.

Das Verfolgen jeder Liste, um das Schwanzelement zu finden, ist wahrscheinlich der langsame Teil hier, aber leider sehe ich keinen Weg, es zu vermeiden. (Wenn die Listenelemente nicht im Speicher aufeinanderfolgend sind, benötigt der Listendurchlauf viele Cache-Lasten aus dem RAM. Die Zugriffsmuster beim Sortieren des Arrays sind für die CPU viel einfacher vorhersagbar (bei aktuellen Architekturen), also der Array-Sortteil ist wahrscheinlich nicht der langsamste Teil - aber natürlich können Sie den Code mit einem praktischen Datensatz profilieren, und prüfen, ob Sie eine schnellere Art Implementierung als die C-Bibliothek qsort() benötigen)

---

OP, dass jeder einzelne geklärt. Eine Liste, die an einem Zeiger in dem Zeigerfeld hängt, kann einen oder mehrere "Blöcke" enthalten, dh aufeinanderfolgende sortierte Läufe. Diese können durch die sich ändernde Blocknummer erkannt werden.

Wenn zusätzlicher Speicherverbrauch kein Problem ist, würde ich eine Reihe von

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

erstellen, die dann von blockNum sortiert werden, und schließlich gefesselt. Das Speichern von Zeigern sowohl im ersten (Kopf) als auch im letzten (letzten) Element bedeutet, dass wir die Listen nur einmal scannen. Nachdem das edge_block-Array sortiert ist, reicht ein einfacher linearer Durchlauf darüber aus, um alle Unterlisten zu einer endgültigen Liste zu verketten.

Zum Beispiel (nur Kompilierung getestet):

#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; 
} 

---

Wenn es potenziell sehr viele blocknums sind, oder Sie müssen die Menge des verwendeten Speichers begrenzen, dann würde ich einen kleinen min verwenden -heap von

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

Elemente, verkeilt durch count, die Anzahl der Elemente in dieser Teilliste.

Die Idee ist, dass wenn Sie einen Block aus der Eingabe extrahieren, Sie es dem Min-Heap hinzufügen, wenn Platz vorhanden ist; Andernfalls verschmelzen Sie sie mit der Root-Liste im Min-Heap. Beachten Sie, dass gemäß den Regeln von OP dieses "Zusammenführen" tatsächlich eine einzelne Einfügung ist, da jeder Block aufeinanderfolgend ist; Nur der Einfügepunkt muss zuerst gefunden werden. Das count wird aktualisiert, um die Anzahl der Elemente in der Stammliste widerzuspiegeln, und Sie häufen so den Min-Heap erneut an.

Der Zweck des Heapspeichers besteht darin, sicherzustellen, dass Sie die zwei kürzesten Blöcke zusammenführen, wobei das Durchlaufen der Listen so gering wie möglich gehalten wird, um den Einfügepunkt zu finden.

Wenn alle Blöcke eingefügt wurden, nehmen Sie die Wurzel, fusionieren diese Liste mit der neuen Stammliste und haufen erneut, indem Sie die Größe des Heapspeichers jedes Mal um eins verringern, bis Sie eine einzelne Liste übrig haben. Das ist die Endergebnisliste.

Answer 2

So wie ich es verstehe, haben Sie mehrere sortierte Listen und Sie möchten sie zusammenführen, um eine einzige sortierte Liste zu erstellen.

Eine gängige Methode besteht darin, eine Warteschlange mit Listen zu erstellen und fortlaufend Paare zusammenzufassen, das Ergebnis zurück zur Warteschlange hinzuzufügen und zu wiederholen, bis nur noch eine Liste vorhanden ist. Zum Beispiel:

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 

Dies ist eine attraktive Option, weil es einfach ist und benötigt sehr wenig zusätzliche Speicher, und es ist ziemlich effizient.

Es gibt einen potenziell schnelleren Weg, der etwas mehr Speicher benötigt (O (log k), wobei k die Anzahl der Listen ist) und etwas mehr Codierung erfordert. Es umfasst das Erstellen eines Min-Heaps, das das erste Element aus jeder Liste enthält. Sie entfernen das unterste Element aus dem Heap, fügen es der neuen Liste hinzu und nehmen dann das nächste Element aus der Liste, aus der das niedrigste Element stammt, und fügen es in den Heap ein.

Beide dieser Algorithmen sind O (n log k) -Komplexität, aber die zweite ist wahrscheinlich schneller, weil sie Daten nicht so viel bewegt. Welchen Algorithmus Sie verwenden möchten, hängt davon ab, wie groß Ihre Listen sind und wie oft Sie die Zusammenführung durchführen.