Skip to content

Commit 3a5936a

Browse files
michaelfigmichaelfig
committed
feat: implement the Cosmos block manager
Eliminate some technical debt FTW!
1 parent 2e8d22d commit 3a5936a

File tree

5 files changed

+224
-96
lines changed

5 files changed

+224
-96
lines changed

‎packages/cosmic-swingset/lib/ag-solo/web.js

+5-1
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,13 @@ import path from 'path';
33
import http from 'http';
44
import express from 'express';
55
import WebSocket from 'ws';
6-
import morgan from 'morgan';
76
import fs from 'fs';
87

8+
// We need to CommonJS require morgan or else it warns, until:
9+
// https://github.com/expressjs/morgan/issues/190
10+
// is fixed.
11+
const morgan = require('morgan');
12+
913
const points = new Map();
1014
const broadcasts = new Map();
1115

‎packages/cosmic-swingset/lib/block-manager.js

+165
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
// TODO: Put this somewhere else.
2+
function deepEquals(a, b, already = new WeakSet()) {
3+
if (Object.is(a, b)) {
4+
return true;
5+
}
6+
7+
// Must both be objects.
8+
if (Object(a) !== a || Object(b) !== b) {
9+
return false;
10+
}
11+
12+
// That we haven't seen before.
13+
if (already.has(a) || already.has(b)) {
14+
return false;
15+
}
16+
already.add(a);
17+
already.add(b);
18+
19+
// With the same prototype.
20+
if (Object.getPrototypeOf(a) !== Object.getPrototypeOf(b)) {
21+
return false;
22+
}
23+
24+
// And deepEquals entries.
25+
const amap = new Map(Object.entries(a));
26+
for (const [key, bval] of Object.entries(b)) {
27+
if (!amap.has(key)) {
28+
return false;
29+
}
30+
if (!deepEquals(amap.get(key), bval, already)) {
31+
return false;
32+
}
33+
amap.delete(key);
34+
}
35+
36+
// And no extra keys in b.
37+
if (amap.size > 0) {
38+
return false;
39+
}
40+
return true;
41+
}
42+
43+
const BEGIN_BLOCK = 'BEGIN_BLOCK';
44+
const DELIVER_INBOUND = 'DELIVER_INBOUND';
45+
const END_BLOCK = 'END_BLOCK';
46+
47+
export default function makeBlockManager({
48+
deliverInbound,
49+
beginBlock,
50+
saveChainState,
51+
saveOutsideState,
52+
savedActions,
53+
savedHeight,
54+
}) {
55+
let runTime = 0;
56+
async function kernelPerformAction(action) {
57+
const start = Date.now();
58+
const finish = _ => (runTime += Date.now() - start);
59+
60+
let p;
61+
switch (action.type) {
62+
case BEGIN_BLOCK:
63+
p = beginBlock(action.blockHeight, action.blockTime);
64+
break;
65+
66+
case DELIVER_INBOUND:
67+
p = deliverInbound(
68+
action.peer,
69+
action.messages,
70+
action.ack,
71+
action.blockHeight,
72+
action.blockTime,
73+
);
74+
break;
75+
76+
case END_BLOCK:
77+
return true;
78+
79+
default:
80+
throw new Error(`${action.type} not recognized`);
81+
}
82+
p.then(finish, finish);
83+
return p;
84+
}
85+
86+
let currentActions;
87+
let currentIndex;
88+
let replaying;
89+
let decohered;
90+
91+
async function blockManager(action) {
92+
if (decohered) {
93+
throw decohered;
94+
}
95+
96+
if (action.type === BEGIN_BLOCK) {
97+
// Start a new block, or possibly replay the prior one.
98+
replaying = action.blockHeight === savedHeight;
99+
currentIndex = 0;
100+
currentActions = [];
101+
runTime = 0;
102+
} else {
103+
// We're working on a subsequent actions.
104+
currentIndex += 1;
105+
}
106+
107+
currentActions.push(action);
108+
109+
if (!replaying) {
110+
// Compute new state by running the kernel.
111+
await kernelPerformAction(action);
112+
} else if (!deepEquals(action, savedActions[currentIndex])) {
113+
// Divergence of the inbound messages, so rewind the state if we need to.
114+
console.log(action, 'and', savedActions[currentIndex], 'are not equal');
115+
replaying = false;
116+
117+
// We only handle the trivial case.
118+
const restoreHeight = action.blockHeight - 1;
119+
if (restoreHeight !== savedHeight) {
120+
// Keep throwing forever.
121+
decohered = Error(
122+
`Cannot reset state from ${savedHeight} to ${restoreHeight}; unimplemented`,
123+
);
124+
throw decohered;
125+
}
126+
127+
// Replay the saved actions.
128+
for (const a of currentActions) {
129+
// eslint-disable-next-line no-await-in-loop
130+
await kernelPerformAction(a);
131+
}
132+
}
133+
134+
if (action.type !== END_BLOCK) {
135+
return;
136+
}
137+
138+
// Commit all the keeper state, even on replay.
139+
// This is necessary since the block proposer will be asked to validate
140+
// the actions it just proposed (in Tendermint v0.33.0).
141+
let start = Date.now();
142+
const { mailboxSize } = saveChainState();
143+
let now = Date.now();
144+
145+
const mbTime = now - start;
146+
start = now;
147+
148+
// Advance our saved state variables.
149+
savedActions = currentActions;
150+
savedHeight = action.blockHeight;
151+
152+
if (!replaying) {
153+
// Save the kernel's new state.
154+
saveOutsideState(savedHeight, savedActions);
155+
}
156+
now = Date.now();
157+
const saveTime = now - start;
158+
159+
console.log(
160+
`wrote SwingSet checkpoint (mailbox=${mailboxSize}), [run=${runTime}ms, mb=${mbTime}ms, save=${saveTime}ms]`,
161+
);
162+
}
163+
164+
return blockManager;
165+
}

‎packages/cosmic-swingset/lib/chain-main.js

+11-52
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,9 @@
11
import stringify from '@agoric/swingset-vat/src/kernel/json-stable-stringify';
22

33
import { launch } from './launch-chain';
4+
import makeBlockManager from './block-manager';
45

56
const AG_COSMOS_INIT = 'AG_COSMOS_INIT';
6-
const BEGIN_BLOCK = 'BEGIN_BLOCK';
7-
const DELIVER_INBOUND = 'DELIVER_INBOUND';
8-
const END_BLOCK = 'END_BLOCK';
97

108
export default async function main(progname, args, { path, env, agcc }) {
119
const bootAddress = env.BOOT_ADDRESS;
@@ -60,7 +58,7 @@ export default async function main(progname, args, { path, env, agcc }) {
6058
const p = Promise.resolve(handler(action));
6159
p.then(
6260
res => replier.resolve(`${res}`),
63-
rej => replier.reject(`rejection ${rej} ignored`),
61+
rej => replier.reject(`rejection ignored: ${rej.stack || rej}`),
6462
);
6563
}
6664

@@ -72,31 +70,11 @@ export default async function main(progname, args, { path, env, agcc }) {
7270
setInterval(() => undefined, 30000);
7371
agcc.runAgCosmosDaemon(nodePort, fromGo, [progname, ...args]);
7472

75-
let deliverInbound;
76-
let deliverStartBlock;
77-
let deliveryFunctionsInitialized = false;
78-
7973
// this storagePort changes for every single message. We define it out here
8074
// so the 'externalStorage' object can close over the single mutable
8175
// instance, and we update the 'sPort' value each time toSwingSet is called
8276
let sPort;
83-
84-
function toSwingSet(action, replier) {
85-
// console.log(`toSwingSet`, action, replier);
86-
// eslint-disable-next-line no-use-before-define
87-
return blockManager(action, replier).then(
88-
ret => {
89-
// console.log(`blockManager returning:`, ret);
90-
return ret;
91-
},
92-
err => {
93-
console.log('blockManager threw error:', err);
94-
throw err;
95-
},
96-
);
97-
}
98-
99-
async function launchAndInitializeDeliverInbound() {
77+
async function launchAndInitializeSwingSet() {
10078
// this object is used to store the mailbox state. we only ever use
10179
// key='mailbox'
10280
const mailboxStorage = {
@@ -143,7 +121,9 @@ export default async function main(progname, args, { path, env, agcc }) {
143121
return s;
144122
}
145123

146-
async function blockManager(action, _replier) {
124+
let blockManager;
125+
async function toSwingSet(action, _replier) {
126+
// console.log(`toSwingSet`, action, replier);
147127
if (action.type === AG_COSMOS_INIT) {
148128
return true;
149129
}
@@ -152,34 +132,13 @@ export default async function main(progname, args, { path, env, agcc }) {
152132
// Initialize the storage for this particular transaction.
153133
// console.log(` setting sPort to`, action.storagePort);
154134
sPort = action.storagePort;
155-
}
156135

157-
// launch the swingset once
158-
if (!deliveryFunctionsInitialized) {
159-
const deliveryFunctions = await launchAndInitializeDeliverInbound();
160-
deliverInbound = deliveryFunctions.deliverInbound;
161-
deliverStartBlock = deliveryFunctions.deliverStartBlock;
162-
deliveryFunctionsInitialized = true;
136+
if (!blockManager) {
137+
const fns = await launchAndInitializeSwingSet();
138+
blockManager = makeBlockManager(fns);
139+
}
163140
}
164141

165-
switch (action.type) {
166-
case BEGIN_BLOCK:
167-
return deliverStartBlock(action.blockHeight, action.blockTime);
168-
case DELIVER_INBOUND:
169-
return deliverInbound(
170-
action.peer,
171-
action.messages,
172-
action.ack,
173-
action.blockHeight,
174-
action.blockTime,
175-
);
176-
case END_BLOCK:
177-
return true;
178-
179-
default:
180-
throw new Error(
181-
`${action.type} not recognized. must be BEGIN_BLOCK, DELIVER_INBOUND, or END_BLOCK`,
182-
);
183-
}
142+
return blockManager(action);
184143
}
185144
}

‎packages/cosmic-swingset/lib/launch-chain.js

+28-38
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import fs from 'fs';
22

33
import djson from 'deterministic-json';
4-
import readlines from 'n-readlines';
54
import {
65
buildMailbox,
76
buildMailboxStateMap,
@@ -13,6 +12,8 @@ import {
1312
} from '@agoric/swingset-vat';
1413
import { openSwingStore } from '@agoric/swing-store-simple';
1514

15+
const SWING_STORE_META_KEY = 'cosmos/meta';
16+
1617
async function buildSwingset(withSES, mailboxState, storage, vatsDir, argv) {
1718
const config = {};
1819
const mbs = buildMailboxStateMap();
@@ -69,17 +70,12 @@ export async function launch(kernelStateDBDir, mailboxStorage, vatsDir, argv) {
6970
argv,
7071
);
7172

72-
let mailboxLastData = djson.stringify(mbs.exportToData());
73-
function saveState(runTime = undefined) {
74-
let start = Date.now();
75-
73+
function saveChainState() {
7674
// now check mbs
7775
const newState = mbs.exportToData();
7876
const newData = djson.stringify(newState);
79-
if (newData !== mailboxLastData) {
80-
console.log(`outbox changed`);
81-
}
8277

78+
// Save the mailbox state.
8379
for (const peer of Object.getOwnPropertyNames(newState)) {
8480
const data = {
8581
outbox: newState[peer].outbox,
@@ -88,51 +84,45 @@ export async function launch(kernelStateDBDir, mailboxStorage, vatsDir, argv) {
8884
mailboxStorage.set(`mailbox.${peer}`, djson.stringify(data));
8985
}
9086
mailboxStorage.set('mailbox', newData);
91-
mailboxLastData = newData;
92-
93-
const mbTime = Date.now() - start;
94-
95-
// Save the rest of the kernel state.
96-
start = Date.now();
97-
commit();
98-
const saveTime = Date.now() - start;
99-
100-
const mailboxSize = mailboxLastData.length;
101-
const runTimeStr = runTime === undefined ? '' : `run=${runTime}ms, `;
102-
console.log(
103-
`wrote SwingSet checkpoint (mailbox=${mailboxSize}), [${runTimeStr}mb=${mbTime}ms, save=${saveTime}ms]`,
104-
);
87+
return { mailboxSize: newData.length };
10588
}
10689

107-
// save the initial state immediately
108-
saveState();
109-
110-
// then arrange for inbound messages to be processed, after which we save
111-
async function turnCrank() {
112-
const start = Date.now();
113-
await controller.run();
114-
const runTime = Date.now() - start;
115-
// Have to save state every time.
116-
saveState(runTime);
90+
function saveOutsideState(savedHeight, savedActions) {
91+
storage.set(
92+
SWING_STORE_META_KEY,
93+
JSON.stringify([savedHeight, savedActions]),
94+
);
95+
commit();
11796
}
11897

11998
async function deliverInbound(sender, messages, ack) {
12099
if (!(messages instanceof Array)) {
121100
throw new Error(`inbound given non-Array: ${messages}`);
122101
}
123-
if (mb.deliverInbound(sender, messages, ack)) {
124-
console.log(`mboxDeliver: ADDED messages`);
102+
if (!mb.deliverInbound(sender, messages, ack)) {
103+
return;
125104
}
126-
await turnCrank();
105+
console.log(`mboxDeliver: ADDED messages`);
106+
await controller.run();
127107
}
128108

129-
async function deliverStartBlock(blockHeight, blockTime) {
109+
async function beginBlock(blockHeight, blockTime) {
130110
const addedToQueue = timer.poll(blockTime);
131111
console.log(
132112
`polled; blockTime:${blockTime}, h:${blockHeight} ADDED: ${addedToQueue}`,
133113
);
134-
await turnCrank();
114+
await controller.run();
135115
}
136116

137-
return { deliverInbound, deliverStartBlock };
117+
const [savedHeight, savedActions] = JSON.parse(
118+
storage.get(SWING_STORE_META_KEY) || '[0, []]',
119+
);
120+
return {
121+
deliverInbound,
122+
beginBlock,
123+
saveChainState,
124+
saveOutsideState,
125+
savedHeight,
126+
savedActions,
127+
};
138128
}

0 commit comments

Comments
 (0)