remove memory leak from member manager

rdf-connect · Mar 11, 2024 · 4f8a8d0 · 4f8a8d0
1 parent 7ccd41f
commit 4f8a8d0
Show file tree

Hide file tree

Showing 4 changed files with 685 additions and 1,263 deletions.
diff --git a/bin/cli.ts b/bin/cli.ts
@@ -117,20 +117,20 @@ async function main() {
 
   const reader = client.stream({ highWaterMark: 10 }).getReader();
   let el = await reader.read();
-  const seen = new Set();
+  let count = 0;
   while (el) {
     if (el.value) {
-      seen.add(el.value.id);
+      count += 1;
 
       if (!quiet) {
         if (verbose) {
           console.log(new Writer().quadsToString(el.value.quads));
         }
 
-        if (seen.size % 100 == 1) {
+        if (count % 100 == 1) {
           console.error(
             "Got member",
-            seen.size,
+            count,
             "with",
             el.value.quads.length,
             "quads",
@@ -147,7 +147,7 @@ async function main() {
   }
 
   if (!quiet) {
-    console.error("Found", seen.size, "members");
+    console.error("Found", count, "members");
   }
 }
 

diff --git a/lib/memberManager.ts b/lib/memberManager.ts
@@ -1,12 +1,11 @@
-import { Term, Quad } from "@rdfjs/types";
+import { Quad, Term } from "@rdfjs/types";
 import { Member } from "./page";
 import { FetchedPage } from "./pageFetcher";
 import { CBDShapeExtractor } from "extract-cbd-shape";
 import { RDF, TREE } from "@treecg/types";
-import Heap from "heap-js";
 import { LDESInfo } from "./client";
 import debug from "debug";
-import { Notifier } from "./utils";
+import { getObjects, Notifier } from "./utils";
 import { RdfStore } from "rdf-stores";
 
 const log = debug("manager");
@@ -26,13 +25,8 @@ export type MemberEvents = {
   extracted: Member;
   done: Member[];
 };
-const getObjects = function (store: RdfStore, subject: Term | null, predicate: Term | null, graph?: Term | null) {
-  return store.getQuads(subject, predicate, null, graph).map((quad) => {
-    return quad.object;
-  });
-}
+
 export class Manager {
-  private members: Heap<Member>;
   public queued: number = 0;
   private resolve?: () => void;
   private ldesId: Term;
@@ -56,15 +50,6 @@ export class Manager {
     this.shapeMap = info.shapeMap;
 
     logger("new %s %o", ldesId.value, info);
-
-    this.members = new Heap((a, b) => {
-      if (a.id.equals(b.id)) return 0;
-      if (a.timestamp == b.timestamp) return 0;
-      if (!a && b) return 1;
-      if (a && !b) return -1;
-      if (a.timestamp! < b.timestamp!) return -1;
-      return 1;
-    });
   }
 
   async close() {
@@ -85,13 +70,16 @@ export class Manager {
     member: Term,
     data: RdfStore,
   ): Promise<Member | undefined> {
-
     let quads: Quad[] = [];
 
     if (this.shapeMap) {
       if (this.shapeMap.size === 1) {
         // Use the only shape available
-        quads = await this.extractor.extract(data, member, Array.from(this.shapeMap.values())[0]);
+        quads = await this.extractor.extract(
+          data,
+          member,
+          Array.from(this.shapeMap.values())[0],
+        );
       } else if (this.shapeMap.size > 1) {
         // Find what is the proper shape for this member based on its rdf:type
         const memberType = getObjects(data, member, RDF.terms.type)[0];
@@ -140,11 +128,12 @@ export class Manager {
       if (this.isVersionOfPath) {
         isVersionOf = quads.find(
           (x) =>
-            x.subject.equals(member) && x.predicate.equals(this.isVersionOfPath),
+            x.subject.equals(member) &&
+            x.predicate.equals(this.isVersionOfPath),
         )?.object.value;
       }
 
-      this.members.push({ id: member, quads, timestamp, isVersionOf });
+      // HEAD
       return { id: member, quads, timestamp, isVersionOf };
     }
   }