make instance type (on-demand) configurable. run demo-4 on-demand

README.md

@@ -52,6 +52,15 @@ inputFileUrl = await txUploader.end();
 const accumulatedProof = await client.process(inputFileUrl, transactions.length);
 ```
 
+## Instance Types
+By default the Hadoop Cluster runs on [AWS EC2 Spot Instances](https://aws.amazon.com/ec2/spot/) which are cheaper than normal (i.e. on-demand) instances. You can configure the spot instance bid price (SPOT_BID_PRICE_DOLLARS) [in this file](sequencer/src/map-reduce/constants.ts). It is currently set to $0.5.
+
+For large transaction batches (>1000 txs), it is more reliable to use on-demand instances though. This is because AWS often does not have enough spot instances to allocate for the job. You can select to use on-demand instances by setting:
+```typescript
+const client = new MapReduceClient<RollupProof>(MODE, REGION);
+client.onDemandInstances = true;
+```
+
 ## Acceptable ZkPrograms
 Your rollup (which is simply an o1-js ZkProgram) should work with Hazook almost out of the box, but it should be built as a Zk-Rollup as described in [Mina's documentation here](https://docs.minaprotocol.com/zkapps/tutorials/recursion#scaling-throughput-with-zkrollups-and-app-chains). To summarize, the ZkProgram should enable two functionalities: (1) a baseFunction which should generate and validate a base zk proof per transaction and (2) a mergeFunction which validates that two baseFunctions form a valid sequence and generates a proof of that. 
 

sequencer/src/map-reduce/client/client.ts

@@ -29,6 +29,7 @@ import { logger } from '../../utils';
 
 export class MapReduceClient<RollupProof extends RollupProofBase> {
   public uploader: Uploader<RollupProof>;
+  public onDemandInstances: boolean; // set to true to use on-demand instances instead of spot instances (for better reliability but for higher cost)
   private mode: Mode;
   private emrClient?: EMRClient;
 
@@ -38,6 +39,7 @@ export class MapReduceClient<RollupProof extends RollupProofBase> {
 
     if (this.mode == Mode.EMR) {
       this.emrClient = new EMRClient({ region });
+      this.onDemandInstances = false;
     }
   }
 
@@ -206,7 +208,8 @@ export class MapReduceClient<RollupProof extends RollupProofBase> {
       await this.autoScale({
         clusterId,
         instanceFleetId: taskFleetDetails.Id,
-        targetSpotNodes: Math.round(numberOfProofs / PROOFS_PER_TASK_NODE) + 1,
+        targetInstanceCount:
+          Math.round(numberOfProofs / PROOFS_PER_TASK_NODE) + 1,
       });
 
       const start = Date.now();
@@ -248,23 +251,25 @@ export class MapReduceClient<RollupProof extends RollupProofBase> {
       await this.autoScale({
         clusterId,
         instanceFleetId: taskFleetDetails.Id,
-        targetSpotNodes: TASK_NODE_FLEET_IDLE_TARGET_CAPACITY,
+        targetInstanceCount: TASK_NODE_FLEET_IDLE_TARGET_CAPACITY,
       });
     }
   }
 
   private async autoScale(args: {
     clusterId: string;
     instanceFleetId: string;
-    targetSpotNodes: number;
+    targetInstanceCount: number;
   }): Promise<void> {
-    const { clusterId, instanceFleetId, targetSpotNodes } = args;
+    const { clusterId, instanceFleetId, targetInstanceCount } = args;
     const command = new ModifyInstanceFleetCommand({
       ClusterId: clusterId,
       InstanceFleet: {
         InstanceFleetId: instanceFleetId,
-        TargetSpotCapacity: targetSpotNodes,
-        TargetOnDemandCapacity: 0,
+        TargetSpotCapacity: this.onDemandInstances ? 0 : targetInstanceCount,
+        TargetOnDemandCapacity: this.onDemandInstances
+          ? targetInstanceCount
+          : 0,
         ResizeSpecifications: {
           SpotResizeSpecification: {
             TimeoutDurationMinutes: 20,
@@ -275,7 +280,11 @@ export class MapReduceClient<RollupProof extends RollupProofBase> {
         },
       },
     });
-    logger.info(`EMR: autoscaling cluster to ${targetSpotNodes} spot nodes`);
+    logger.info(
+      `EMR: autoscaling cluster to ${targetInstanceCount} ${
+        this.onDemandInstances ? 'on-demand' : 'spot'
+      } instances`,
+    );
 
     try {
       await this.emrClient.send(command);
@@ -339,7 +348,12 @@ export class MapReduceClient<RollupProof extends RollupProofBase> {
           {
             Name: TASK_NODE_FLEET_NAME,
             InstanceFleetType: 'TASK',
-            TargetSpotCapacity: TASK_NODE_FLEET_IDLE_TARGET_CAPACITY, // Number of task instances
+            TargetSpotCapacity: this.onDemandInstances
+              ? 0
+              : TASK_NODE_FLEET_IDLE_TARGET_CAPACITY,
+            TargetOnDemandCapacity: this.onDemandInstances
+              ? TASK_NODE_FLEET_IDLE_TARGET_CAPACITY
+              : 0,
             InstanceTypeConfigs: INSTANCE_TYPES,
           },
         ],

sequencer/src/map-reduce/constants.ts

@@ -41,34 +41,39 @@ export const PROOFS_PER_TASK_NODE =
  */
 export const REDUCER_SEQUENTIALISM = 2;
 
+/**
+ * The bid price for spot instances in US Dollars (modify this as you please)
+ */
+export const SPOT_BID_PRICE_DOLLARS = '0.5';
+
 /**
  * The instance types the Hadoop cluster can use when it provisions / autoscales itself.
  * For now we only support instances of different families but similar size.
  */
 export const INSTANCE_TYPES: InstanceTypeConfig[] = [
   {
     InstanceType: 'm5.xlarge',
-    BidPrice: '0.5',
+    BidPrice: SPOT_BID_PRICE_DOLLARS,
     EbsConfiguration,
   },
   {
     InstanceType: 'm5d.xlarge',
-    BidPrice: '0.5',
+    BidPrice: SPOT_BID_PRICE_DOLLARS,
     EbsConfiguration,
   },
   {
     InstanceType: 'm6a.xlarge',
-    BidPrice: '0.5',
+    BidPrice: SPOT_BID_PRICE_DOLLARS,
     EbsConfiguration,
   },
   {
     InstanceType: 'm6g.xlarge',
-    BidPrice: '0.5',
+    BidPrice: SPOT_BID_PRICE_DOLLARS,
     EbsConfiguration,
   },
   {
     InstanceType: 'm6i.xlarge',
-    BidPrice: '0.5',
+    BidPrice: SPOT_BID_PRICE_DOLLARS,
     EbsConfiguration,
   },
 ];

sequencer/src/server/services/sequencer.ts

@@ -46,6 +46,7 @@ class Sequencer<RollupProof extends RollupProofBase>
         break;
       case Case.CASE_RUN_4:
         txCount = 1024;
+        this.mapReduce.onDemandInstances = true;
         break;
       default:
     }
@@ -69,6 +70,10 @@ class Sequencer<RollupProof extends RollupProofBase>
 
     response.result = JSON.stringify(proof);
 
+    if (request.case == Case.CASE_RUN_4) {
+      this.mapReduce.onDemandInstances = false;
+    }
+
     return response;
   };
 }