Add Priority Queue

Required for #46 and #51

engines/config-query-sparql-link-traversal/config/config-solid-base-priority-queue.json

@@ -0,0 +1,67 @@
+{
+  "@context": [
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/config-query-sparql/^3.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/config-query-sparql-link-traversal/^0.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/config-query-sparql-solid/^3.0.0/components/context.jsonld"
+  ],
+  "import": [
+    "ccqs:config/context-preprocess/actors.json",
+    "ccqslt:config/context-preprocess/actors.json",
+    "ccqs:config/context-preprocess/mediators.json",
+    "ccqslt:config/extract-links/mediators.json",
+    "ccqs:config/hash-bindings/actors.json",
+    "ccqs:config/hash-bindings/mediators.json",
+    "ccqss:config/http/actors.json",
+    "ccqs:config/http/mediators.json",
+    "ccqs:config/http-invalidate/actors.json",
+    "ccqs:config/http-invalidate/mediators.json",
+    "ccqslt:config/init/actors.json",
+    "ccqs:config/merge-bindings-context/actors.json",
+    "ccqs:config/merge-bindings-context/mediators.json",
+    "ccqslt:config/optimize-query-operation/actors.json",
+    "ccqs:config/optimize-query-operation/mediators.json",
+    "ccqs:config/query-operation/actors.json",
+    "ccqs:config/query-operation/mediators.json",
+    "ccqs:config/query-parse/actors.json",
+    "ccqs:config/query-parse/mediators.json",
+    "ccqs:config/query-process/actors.json",
+    "ccqs:config/query-process/mediators.json",
+    "ccqs:config/query-result-serialize/actors.json",
+    "ccqs:config/query-result-serialize/mediators.json",
+    "ccqslt:config/query-source-identify/actors.json",
+    "ccqs:config/query-source-identify/mediators.json",
+    "ccqs:config/query-source-identify-hypermedia/actors.json",
+    "ccqs:config/query-source-identify-hypermedia/mediators.json",
+    "ccqs:config/dereference/actors.json",
+    "ccqs:config/dereference/mediators.json",
+    "ccqs:config/dereference-rdf/actors.json",
+    "ccqs:config/dereference-rdf/mediators.json",
+    "ccqslt:config/rdf-join/actors.json",
+    "ccqs:config/rdf-join/mediators.json",
+    "ccqslt:config/rdf-join-entries-sort/actors.json",
+    "ccqs:config/rdf-join-entries-sort/mediators.json",
+    "ccqs:config/rdf-join-selectivity/actors.json",
+    "ccqs:config/rdf-join-selectivity/mediators.json",
+    "ccqs:config/rdf-metadata/actors.json",
+    "ccqs:config/rdf-metadata/mediators.json",
+    "ccqs:config/rdf-metadata-accumulate/actors.json",
+    "ccqs:config/rdf-metadata-accumulate/mediators.json",
+    "ccqs:config/rdf-metadata-extract/actors.json",
+    "ccqs:config/rdf-metadata-extract/mediators.json",
+    "ccqslt:config/rdf-metadata-extract/actors/traverse.json",
+    "ccqs:config/rdf-parse/actors.json",
+    "ccqs:config/rdf-parse/mediators.json",
+    "ccqs:config/rdf-parse-html/actors.json",
+    "ccqs:config/rdf-resolve-hypermedia-links/actors.json",
+    "ccqslt:config/rdf-resolve-hypermedia-links/actors/traverse.json",
+    "ccqs:config/rdf-resolve-hypermedia-links/mediators.json",
+    "ccqslt:config/rdf-resolve-hypermedia-links-queue/actors/priority-queue.json",
+    "ccqs:config/rdf-resolve-hypermedia-links-queue/mediators.json",
+    "ccqs:config/rdf-serialize/actors.json",
+    "ccqs:config/rdf-serialize/mediators.json",
+    "ccqs:config/rdf-update-hypermedia/actors.json",
+    "ccqs:config/rdf-update-hypermedia/mediators.json",
+    "ccqs:config/rdf-update-quads/actors.json",
+    "ccqs:config/rdf-update-quads/mediators.json"
+  ]
+}

engines/config-query-sparql-link-traversal/config/rdf-resolve-hypermedia-links-queue/actors/priority-queue.json

@@ -0,0 +1,15 @@
+{
+  "@context": [
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/runner/^3.0.0/components/context.jsonld",
+
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-resolve-hypermedia-links-queue-priority/^0.0.0/components/context.jsonld"
+  ],
+  "@id": "urn:comunica:default:Runner",
+  "@type": "Runner",
+  "actors": [
+    {
+      "@id": "urn:comunica:default:rdf-resolve-hypermedia-links-queue/actors#priority",
+      "@type": "ActorRdfResolveHypermediaLinksQueuePriority"
+    }
+  ]
+}

packages/actor-rdf-resolve-hypermedia-links-queue-priority/README.md

@@ -0,0 +1,37 @@
+# Comunica Priority RDF Resolve Hypermedia Links Queue Actor
+
+[![npm version](https://badge.fury.io/js/%40comunica%2Factor-rdf-resolve-hypermedia-links-queue-priority.svg)](https://www.npmjs.com/package/@comunica/actor-rdf-resolve-hypermedia-links-queue-priority)
+
+An [RDF Resolve Hypermedia Links Queue](https://github.com/comunica/comunica/tree/master/packages/bus-rdf-resolve-hypermedia-links-queue) actor
+that provides a [`ILinkQueue`](https://comunica.github.io/comunica/interfaces/_comunica_bus_rdf_resolve_hypermedia_links_queue.ILinkQueue.html)
+following the priority of links in the queue. The priority must be set by an actor outside the queue, for example in a link queue wrapper.
+
+This module is part of the [Comunica framework](https://github.com/comunica/comunica),
+and should only be used by [developers that want to build their own query engine](https://comunica.dev/docs/modify/).
+
+[Click here if you just want to query with Comunica](https://comunica.dev/docs/query/).
+
+## Install
+
+```bash
+$ yarn add @comunica/actor-rdf-resolve-hypermedia-links-queue-priority
+```
+
+## Configure
+
+After installing, this package can be added to your engine's configuration as follows:
+```text
+{
+  "@context": [
+    ...
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-resolve-hypermedia-links-queue-priority/^3.0.0/components/context.jsonld"
+  ],
+  "actors": [
+    ...
+    {
+      "@id": "urn:comunica:default:rdf-resolve-hypermedia-links-queue/actors#priority",
+      "@type": "ActorRdfResolveHypermediaLinksQueuePriority
+    }
+  ]
+}
+```

packages/actor-rdf-resolve-hypermedia-links-queue-priority/lib/ActorRdfResolveHypermediaLinksQueuePriority.ts

@@ -0,0 +1,26 @@
+import type {
+  IActionRdfResolveHypermediaLinksQueue,
+  IActorRdfResolveHypermediaLinksQueueArgs,
+  IActorRdfResolveHypermediaLinksQueueOutput,
+} from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import { ActorRdfResolveHypermediaLinksQueue } from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import type { IActorTest } from '@comunica/core';
+import { LinkQueuePriority } from './LinkQueuePriority';
+
+/**
+ * A comunica Priority RDF Resolve Hypermedia Links Queue Actor.
+ */
+export class ActorRdfResolveHypermediaLinksQueuePriority extends ActorRdfResolveHypermediaLinksQueue {
+  public constructor(args: IActorRdfResolveHypermediaLinksQueueArgs) {
+    super(args);
+  }
+
+  public async test(_action: IActionRdfResolveHypermediaLinksQueue): Promise<IActorTest> {
+    return true;
+  }
+
+  public async run(_action: IActionRdfResolveHypermediaLinksQueue):
+  Promise<IActorRdfResolveHypermediaLinksQueueOutput> {
+    return { linkQueue: new LinkQueuePriority() };
+  }
+}

packages/actor-rdf-resolve-hypermedia-links-queue-priority/lib/LinkQueuePriority.ts

@@ -0,0 +1,210 @@
+import type { ILink, ILinkQueue } from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+
+/**
+ * A link queue based on priority, using binary heap.
+ */
+export class LinkQueuePriority implements ILinkQueue {
+  /**
+   * Max heap with links and their priorities
+   */
+  public readonly links: ILink[] = [];
+  /**
+   * Data structure to track URLs in the queue and allow for constant time lookup of links in queue based
+   * on link URL
+   */
+  public readonly urlToLink: Record<string, ILink> = {};
+
+  /**
+   * Pushes element to heap by appending it to array and up-heaping the new element
+   */
+  public push(link: ILink): boolean {
+    const idx: number = this.links.length;
+
+    // If we push a link that has no metadata or has metadata but no priority
+    // we set priority to 0 and always set index to end of array (and upheap from there)
+    if (!link.metadata || !link.metadata.priority) {
+      link.metadata = { priority: 0, index: idx };
+    } else {
+      link.metadata.index = idx;
+    }
+    this.links.push(link);
+
+    // Add to Records to allow fast updates in priority
+    this.urlToLink[link.url] = link;
+    this.upHeap(idx);
+    return true;
+  }
+
+  /**
+   * Pops the highest priority link and returns it. Then it sets the last element as root and
+   * down-heaps it.
+   * @returns popped element
+   */
+  public pop(): ILink | undefined {
+    if (this.getSize() > 0) {
+      const max = this.links[0];
+      const endArray = this.links.pop();
+
+      // If we remove link, remove it from records to reflect new state of queue
+      if (max) {
+        delete this.urlToLink[max.url];
+      }
+
+      // Set last element to root and downheap to maintain heap property
+      if (this.links.length > 0) {
+        this.links[0] = endArray!;
+        this.downHeap(0);
+      }
+      return max;
+    }
+  }
+
+  /**
+   * Changes all priorities in queue in O(n + m*log m ) time, with n is size of urlPriorities entry
+   * and m size of the queue. Will only update priorities of urls that are actually in queue.
+   * @param urlPriorities A record with url and new priority
+   */
+  public setAllPriority(urlPriorities: Record<string, number>): void {
+    for (const url in urlPriorities) {
+      if (this.urlToLink[url]) {
+        this.setPriority(url, urlPriorities[url]);
+      }
+    }
+  }
+
+  /**
+   * Update priority of link in queue if it is in queue
+   * @param nodeUrl URL of link to update
+   * @param newValue new priority value
+   * @returns boolean indicating if priority was changed successfully
+   */
+  public setPriority(nodeUrl: string, newValue: number): boolean {
+    const link = this.urlToLink[nodeUrl];
+    if (link) {
+      const delta = newValue - link.metadata!.priority;
+      return this.modifyPriority(nodeUrl, delta);
+    }
+    return false;
+  }
+
+  /**
+   * Function to change priority of element of heap with certain number.
+   * Based on whether the delta is positive or negative, we upheap or downheap.
+   */
+
+  public modifyPriority(nodeUrl: string, delta: number): boolean {
+    const link = this.urlToLink[nodeUrl];
+    if (link) {
+      const idx = link.metadata!.index;
+      this.links[idx].metadata!.priority += delta;
+
+      if (delta < 0) {
+        this.downHeap(idx);
+        return true;
+      }
+      if (delta > 0) {
+        this.upHeap(idx);
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Bubbles up the element at index until the max-heap property is satifisfied
+   * @param idx Index of element to up-heap
+   */
+  public upHeap(idx: number): void {
+    if (idx < 0 || idx > this.links.length - 1) {
+      throw new Error(`Invalid index passed to upheap in priority queue`);
+    }
+    if (idx === 0 && !this.links[idx].metadata!.index) {
+      this.links[idx].metadata!.index = 0;
+    }
+    const element: ILink = this.links[idx];
+    while (idx > 0) {
+      const parentIdx = Math.floor((idx - 1) / 2);
+      const parent = this.links[parentIdx];
+      if (element.metadata!.priority <= parent.metadata!.priority) {
+        element.metadata!.index = idx;
+        break;
+      }
+      this.links[parentIdx] = element;
+      // Update indices
+      element.metadata!.index = parentIdx;
+      this.links[idx] = parent;
+      parent.metadata!.index = idx;
+      idx = parentIdx;
+    }
+  }
+
+  /**
+   * Bubbles down the element at input index until max-heap property is satisifed
+   * @param idx Index of element to down-heap
+   */
+  public downHeap(idx: number): void {
+    if (idx < 0 || idx > this.links.length - 1) {
+      throw new Error(`Invalid index passed to upheap in priority queue`);
+    }
+
+    const length = this.links.length;
+    const element = this.links[idx];
+    let performedSwap = false;
+    let keepSwapping = true;
+
+    while (keepSwapping) {
+      const leftChildIdx = 2 * idx + 1;
+      const rightChildIdx = 2 * idx + 2;
+      let leftChild,
+        rightChild;
+      let swap = null;
+
+      // If there exist a left/right child we do comparison
+      if (leftChildIdx < length) {
+        leftChild = this.links[leftChildIdx];
+        if (leftChild.metadata!.priority > element.metadata!.priority) {
+          swap = leftChildIdx;
+        }
+      }
+      if (rightChildIdx < length) {
+        rightChild = this.links[rightChildIdx];
+        // Only swap with right child if we either: don't swap a left child and the right child has higher
+        // priority or if we do swap and left child has lower priority than right
+        if (
+          (swap === null && rightChild.metadata!.priority > element.metadata!.priority) ||
+          (swap !== null && leftChild && rightChild.metadata!.priority > leftChild.metadata!.priority)
+        ) {
+          swap = rightChildIdx;
+        }
+      }
+      if (swap === null) {
+        // If we don't perform any swap operations we update index
+        if (!performedSwap) {
+          element.metadata!.index = idx;
+        }
+        // This is only for linter..
+        keepSwapping = false;
+        break;
+      }
+      performedSwap = true;
+      // We swap the elements and their stored indexes
+      this.links[idx] = this.links[swap];
+      this.links[idx].metadata!.index = idx;
+      this.links[swap] = element;
+      this.links[swap].metadata!.index = swap;
+      idx = swap;
+    }
+  }
+
+  public getSize(): number {
+    return this.links.length;
+  }
+
+  public isEmpty(): boolean {
+    return this.links.length === 0;
+  }
+
+  public peek(): ILink | undefined {
+    return this.links[0];
+  }
+}

packages/actor-rdf-resolve-hypermedia-links-queue-priority/lib/index.ts

@@ -0,0 +1,2 @@
+export * from './ActorRdfResolveHypermediaLinksQueuePriority';
+export * from './LinkQueuePriority';