Merge branch 'release/1.17.2'

maurizi · maurizi · commit 8211abcbb431 · 2022-05-10T11:46:59.000-04:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,11 +13,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 
-## [1.17.1]
+## [1.17.2]
 ### Changed
- - Improved PlanScore error logging & double timeout [#1204](https://github.com/PublicMapping/districtbuilder/pull/1204)
- - Revert to using JSON for data serialization [#1206](https://github.com/PublicMapping/districtbuilder/pull/1206)
+ - Reduce max cache size [#1213](https://github.com/PublicMapping/districtbuilder/pull/1213)
+
+### Fixed
+- Fixed typos in process-geojson & serialize-topojson commands [#1209](https://github.com/PublicMapping/districtbuilder/pull/1209)
+- Recreate workers when they time out [#1214](https://github.com/PublicMapping/districtbuilder/pull/1214)
 
+## [1.17.1]
+### Changed
+- Improved PlanScore error logging & double timeout [#1204](https://github.com/PublicMapping/districtbuilder/pull/1204)
+- Revert to using JSON for data serialization [#1206](https://github.com/PublicMapping/districtbuilder/pull/1206)
 
 ## [1.17.0]
 
@@ -499,7 +506,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Initial release.
 
-[unreleased]: https://github.com/publicmapping/districtbuilder/compare/1.17.1...HEAD
+[unreleased]: https://github.com/publicmapping/districtbuilder/compare/1.17.2...HEAD
+[1.17.2]: https://github.com/publicmapping/districtbuilder/compare/1.17.1...1.17.2
 [1.17.1]: https://github.com/publicmapping/districtbuilder/compare/1.17.0...1.17.1
 [1.17.0]: https://github.com/publicmapping/districtbuilder/compare/1.16.1...1.17.0
 [1.16.1]: https://github.com/publicmapping/districtbuilder/compare/1.16.0...1.16.1
diff --git a/src/manage/src/commands/process-geojson.ts b/src/manage/src/commands/process-geojson.ts
@@ -522,8 +522,8 @@ it when necessary (file sizes ~1GB+).
   ) {
     const filteredTopojson = this.filterTopoJson(topology, demographics, voting);
     this.log("Writing topojson file");
-    const path = join(dir, "topo.buf");
-    const output = createWriteStream(path, { encoding: "binary" });
+    const path = join(dir, "topo.json");
+    const output = createWriteStream(path, { encoding: "utf-8" });
     output.write(JSON.stringify(filteredTopojson));
     output.close();
   }
diff --git a/src/manage/src/commands/serialize-topojson.ts b/src/manage/src/commands/serialize-topojson.ts
@@ -43,7 +43,7 @@ export default class SerializeTopojson extends Command {
       cli.action.start(`Reading base TopoJSON: ${s3URI}`);
       const baseTopojson = await (flags.input === "json"
         ? this.readJson(s3URI)
-        : flags.output === "buf"
+        : flags.input === "buf"
         ? this.readBuf(s3URI)
         : this.readPbf(s3URI));
       cli.action.stop();
@@ -61,15 +61,7 @@ export default class SerializeTopojson extends Command {
   // Reads a TopoJSON file from S3, given the S3 run directory
   async readJson(inputS3Dir: string): Promise<Topology<Objects<{}>>> {
     this.log("Reading topo.json");
-    const uriComponents = inputS3Dir.split("/");
-    const bucket = uriComponents[2];
-    const key = `${uriComponents.slice(3).join("/")}topo.json`;
-    const response: any = await new S3()
-      .getObject({
-        Bucket: bucket,
-        Key: key
-      })
-      .promise();
+    const response: any = await new S3().getObject(this.s3Options(inputS3Dir, "json")).promise();
 
     const objects = await new Promise(resolve =>
       createReadStream(response.Body as Buffer)
@@ -120,76 +112,50 @@ export default class SerializeTopojson extends Command {
     } as Topology<Objects<{}>>;
   }
 
-  jsonOptions(inputS3Dir: string) {
-    const uriComponents = inputS3Dir.split("/");
-    return {
-      Bucket: uriComponents[2],
-      Key: `${uriComponents.slice(3).join("/")}topo.json`
-    };
-  }
-
-  bufOptions(inputS3Dir: string) {
-    const uriComponents = inputS3Dir.split("/");
-    return {
-      Bucket: uriComponents[2],
-      Key: `${uriComponents.slice(3).join("/")}topo.buf`
-    };
+  read(inputS3Dir: string, ext: string) {
+    this.log(`Reading topo.${ext}`);
+    const s3Client = new S3();
+    return s3Client.getObject(this.s3Options(inputS3Dir, ext)).promise();
   }
 
-  pbfOptions(inputS3Dir: string) {
+  s3Options(inputS3Dir: string, ext: string) {
     const uriComponents = inputS3Dir.split("/");
     return {
       Bucket: uriComponents[2],
-      Key: `${uriComponents.slice(3).join("/")}topo.pbf`
+      Key: `${uriComponents.slice(3).join("/")}topo.${ext}`
     };
   }
 
   async readBuf(inputS3Dir: string) {
-    this.log("Reading topo.buf");
-
-    const s3Client = new S3();
-    const resp = await s3Client.getObject(this.bufOptions(inputS3Dir)).promise();
+    const resp = await this.read(inputS3Dir, "buf");
     return deserialize(resp.Body as Buffer);
   }
 
   async readPbf(inputS3Dir: string) {
-    this.log("Reading topo.buf");
-
-    const s3Client = new S3();
-    const resp = await s3Client.getObject(this.bufOptions(inputS3Dir)).promise();
+    const resp = await this.read(inputS3Dir, "pbf");
     return decode(new Pbf(resp.Body as Buffer)) as Topology;
   }
 
-  writeJson(inputS3Dir: string, topology: Topology<Objects<{}>>) {
-    this.log("Streaming topo.json");
+  write(inputS3Dir: string, ext: string, body: string | Buffer | Uint8Array) {
+    this.log(`Writing topo.${ext}`);
     const s3Client = new S3();
     return s3Client
       .upload({
-        Body: JSON.stringify(topology),
-        ...this.jsonOptions(inputS3Dir)
+        Body: body,
+        ...this.s3Options(inputS3Dir, ext)
       })
       .promise();
   }
 
+  writeJson(inputS3Dir: string, topology: Topology<Objects<{}>>) {
+    return this.write(inputS3Dir, "json", JSON.stringify(topology));
+  }
+
   writeBuf(inputS3Dir: string, topology: Topology<Objects<{}>>) {
-    this.log("Streaming topo.buf");
-    const s3Client = new S3();
-    return s3Client
-      .upload({
-        Body: serialize(topology),
-        ...this.bufOptions(inputS3Dir)
-      })
-      .promise();
+    return this.write(inputS3Dir, "buf", serialize(topology));
   }
 
   writePbf(inputS3Dir: string, topology: Topology<Objects<{}>>) {
-    this.log("Streaming topo.pbf");
-    const s3Client = new S3();
-    return s3Client
-      .upload({
-        Body: encode(topology, new Pbf()),
-        ...this.pbfOptions(inputS3Dir)
-      })
-      .promise();
+    return this.write(inputS3Dir, "pbf", encode(topology, new Pbf()));
   }
 }
diff --git a/src/server/src/districts/services/worker-pool.service.ts b/src/server/src/districts/services/worker-pool.service.ts
@@ -25,11 +25,10 @@ const dockerMemLimit = existsSync("/sys/fs/cgroup/memory.max")
 const hostmem = os.totalmem();
 const totalmem = Math.min(hostmem, dockerMemLimit);
 // Targets:
-// 1.5Gb of cache for 12Gb total (dev)
-//   3Gb of cache for 15Gb total (16Gb instance w/ ~1Gb ECS agent)
-//  11Gb of cache for 31Gb total (32Gb instance w/ ~1Gb ECS agent)
-//  27Gb of cache for 63Gb total (64Gb instance w/ ~1Gb ECS agent)
-export const cacheSize = totalmem * 0.5 - 4.5 * 1024 * 1024 * 1024;
+// 1.8Gb of cache for 12Gb total (dev)
+// 3.0Gb of cache for 15Gb total (16Gb instance w/ ~1Gb ECS agent)
+// 9.5Gb of cache for 31Gb total (32Gb instance w/ ~1Gb ECS agent)
+export const cacheSize = totalmem * 0.4 - 3 * 1024 * 1024 * 1024;
 const maxWorkerCacheSize = Math.ceil(cacheSize / NUM_WORKERS);
 
 // Not that important to limit small regions, but large regions in every worker will eat up our cache
@@ -97,7 +96,10 @@ export class WorkerPoolService {
       (!this.pendingWorkersByRegion[regionConfig.id] ||
         !this.pendingWorkersByRegion[regionConfig.id].includes(worker));
     const availableWorkerIndexes = [...this.workerQueues.keys()].filter(workerLacksRegion);
-    const lastUsed = this.workersByRegion[regionConfig.id] || [];
+    const lastUsed = [
+      ...(this.workersByRegion[regionConfig.id] || []),
+      ...(this.pendingWorkersByRegion[regionConfig.id] || [])
+    ];
 
     const lastUsedSettled = getSettledQueues(this.workerQueues, lastUsed);
     const settledQueues = getSettledQueues(this.workerQueues, availableWorkerIndexes);
@@ -203,7 +205,7 @@ export class WorkerPoolService {
         const [workerIndex, addedToRegion] = this.findQueue(regionConfig);
         const task = this.workerQueues[workerIndex].add(() =>
           this.workers[workerIndex]
-            .then((worker: undefined | WorkerThread | Promise<WorkerThread>) => {
+            .then(async (worker: WorkerThread | undefined) => {
               // If this region wasn't already in this workers cache, update the worker size
               // This may trigger recreating the worker thread if we would exceed the max size
               if (addedToRegion) {
@@ -213,7 +215,7 @@ export class WorkerPoolService {
 
                 if (this.workerSizes[workerIndex] + size > maxWorkerCacheSize) {
                   // eslint-disable-next-line no-param-reassign
-                  worker = this.createWorker(workerIndex, "OoM");
+                  worker = await this.createWorker(workerIndex, "OoM");
                   // Reset tracking info for this worker (any pending data stays the same)
                   this.workerSizes[workerIndex] = 0;
                   this.workersByRegion = _.mapValues(this.workersByRegion, workers =>
@@ -231,26 +233,30 @@ export class WorkerPoolService {
               }
               return worker;
             })
-            .then(worker =>
-              // If we haven't created the worker yet, do so now
-              worker ? cb(worker) : this.createWorker(workerIndex).then(cb)
+            .then(
+              worker =>
+                new Promise<R>((resolve, reject) => {
+                  // Clear pre-existing timeouts if we're adding a new one before cleanup could happen
+                  this.clearQueueTimeout(workerIndex);
+                  if (timeout) {
+                    // eslint-disable-next-line functional/immutable-data
+                    this.timeouts[workerIndex] = setTimeout(() => {
+                      reject("Worker request timed out");
+                    }, timeout);
+                  }
+                  // If we haven't created the worker yet, do so now
+                  const result = worker ? cb(worker) : this.createWorker(workerIndex).then(cb);
+                  result.then(resolve).catch(reject);
+                })
             )
         );
         resolve(
           new Promise((resolve, reject) => {
-            // Clear pre-existing timeouts if we're adding a new one before cleanup could happen
-            this.clearQueueTimeout(workerIndex);
-            if (timeout) {
-              // eslint-disable-next-line functional/immutable-data
-              this.timeouts[workerIndex] = setTimeout(() => {
-                this.terminateWorker(workerIndex, "timeout").finally(() => reject());
-              }, timeout);
-            }
             task
-              .then(worker => {
+              .then(result => {
                 // Clear out any timeouts after the task completes
                 this.clearQueueTimeout(workerIndex);
-                resolve(worker);
+                resolve(result);
               })
               .catch(error => {
                 this.terminateWorker(workerIndex, "error").finally(() => reject(error));
