Merge pull request #2156 from filecoin-project/fix/bench-analyze

Improve performance of bench import analyze

.gitignore

@@ -35,3 +35,4 @@ build/paramfetch.sh
 bin/ipget
 bin/tmp/*
 .idea
+scratchpad

chain/vm/gas.go

@@ -143,30 +143,40 @@ func (ps pricedSyscalls) VerifySignature(signature crypto.Signature, signer addr
 		return err
 	}
 	ps.chargeGas(c)
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.VerifySignature(signature, signer, plaintext)
 }
 
 // Hashes input data using blake2b with 256 bit output.
 func (ps pricedSyscalls) HashBlake2b(data []byte) [32]byte {
 	ps.chargeGas(ps.pl.OnHashing(len(data)))
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.HashBlake2b(data)
 }
 
 // Computes an unsealed sector CID (CommD) from its constituent piece CIDs (CommPs) and sizes.
 func (ps pricedSyscalls) ComputeUnsealedSectorCID(reg abi.RegisteredSealProof, pieces []abi.PieceInfo) (cid.Cid, error) {
 	ps.chargeGas(ps.pl.OnComputeUnsealedSectorCid(reg, pieces))
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.ComputeUnsealedSectorCID(reg, pieces)
 }
 
 // Verifies a sector seal proof.
 func (ps pricedSyscalls) VerifySeal(vi abi.SealVerifyInfo) error {
 	ps.chargeGas(ps.pl.OnVerifySeal(vi))
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.VerifySeal(vi)
 }
 
 // Verifies a proof of spacetime.
 func (ps pricedSyscalls) VerifyPoSt(vi abi.WindowPoStVerifyInfo) error {
 	ps.chargeGas(ps.pl.OnVerifyPost(vi))
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.VerifyPoSt(vi)
 }
 
@@ -182,19 +192,21 @@ func (ps pricedSyscalls) VerifyPoSt(vi abi.WindowPoStVerifyInfo) error {
 // Returns nil and an error if the headers don't prove a fault.
 func (ps pricedSyscalls) VerifyConsensusFault(h1 []byte, h2 []byte, extra []byte) (*runtime.ConsensusFault, error) {
 	ps.chargeGas(ps.pl.OnVerifyConsensusFault())
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.VerifyConsensusFault(h1, h2, extra)
 }
 
 func (ps pricedSyscalls) BatchVerifySeals(inp map[address.Address][]abi.SealVerifyInfo) (map[address.Address][]bool, error) {
 	var gasChargeSum GasCharge
 	gasChargeSum.Name = "BatchVerifySeals"
-	ps.chargeGas(gasChargeSum) // TODO: this is only called by the cron actor. Should we even charge gas?
+	count := int64(0)
-
 	for _, svis := range inp {
-		for _, svi := range svis {
+		count += int64(len(svis))
-			ch := ps.pl.OnVerifySeal(svi)
-			ps.chargeGas(newGasCharge("BatchVerifySingle", 0, 0).WithVirtual(ch.VirtualCompute+ch.ComputeGas, 0))
-		}
 	}
+	gasChargeSum = gasChargeSum.WithExtra(count).WithVirtual(129778623*count+716683250, 0)
+	ps.chargeGas(gasChargeSum) // TODO: this is only called by the cron actor. Should we even charge gas?
+	defer ps.chargeGas(gasOnActorExec)
+
 	return ps.under.BatchVerifySeals(inp)
 }

chain/vm/gas_v0.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 	"github.com/filecoin-project/specs-actors/actors/crypto"
 )
@@ -86,39 +87,57 @@ var _ Pricelist = (*pricelistV0)(nil)
 
 // OnChainMessage returns the gas used for storing a message of a given size in the chain.
 func (pl *pricelistV0) OnChainMessage(msgSize int) GasCharge {
-	return newGasCharge("OnChainMessage", 0, pl.onChainMessageBase+pl.onChainMessagePerByte*int64(msgSize))
+	return newGasCharge("OnChainMessage", 0, pl.onChainMessageBase+pl.onChainMessagePerByte*int64(msgSize)).WithVirtual(77302, 0)
 }
 
 // OnChainReturnValue returns the gas used for storing the response of a message in the chain.
 func (pl *pricelistV0) OnChainReturnValue(dataSize int) GasCharge {
-	return newGasCharge("OnChainReturnValue", 0, int64(dataSize)*pl.onChainReturnValuePerByte)
+	return newGasCharge("OnChainReturnValue", 0, int64(dataSize)*pl.onChainReturnValuePerByte).WithVirtual(107294, 0)
 }
 
 // OnMethodInvocation returns the gas used when invoking a method.
 func (pl *pricelistV0) OnMethodInvocation(value abi.TokenAmount, methodNum abi.MethodNum) GasCharge {
 	ret := pl.sendBase
+	extra := ""
+	virtGas := int64(1072944)
+
 	if value != abi.NewTokenAmount(0) {
+		// TODO: fix this, it is comparing pointers instead of values
+		// see vv
 		ret += pl.sendTransferFunds
 	}
+	if big.Cmp(value, abi.NewTokenAmount(0)) != 0 {
+		virtGas += 497495
+		if methodNum == builtin.MethodSend {
+			// transfer only
+			virtGas += 973940
+		}
+		extra += "t"
+	}
 	if methodNum != builtin.MethodSend {
 		ret += pl.sendInvokeMethod
+		extra += "i"
+		// running actors is cheaper becase we hand over to actors
+		virtGas += -295779
 	}
-	return newGasCharge("OnMethodInvocation", ret, 0).WithVirtual(ret*15000, 0)
+	return newGasCharge("OnMethodInvocation", ret, 0).WithVirtual(virtGas, 0).WithExtra(extra)
 }
 
 // OnIpldGet returns the gas used for storing an object
 func (pl *pricelistV0) OnIpldGet(dataSize int) GasCharge {
-	return newGasCharge("OnIpldGet", pl.ipldGetBase+int64(dataSize)*pl.ipldGetPerByte, 0).WithExtra(dataSize).WithVirtual(pl.ipldGetBase*13750+(pl.ipldGetPerByte*100), 0)
+	return newGasCharge("OnIpldGet", pl.ipldGetBase+int64(dataSize)*pl.ipldGetPerByte, 0).
+		WithExtra(dataSize).WithVirtual(433685, 0)
 }
 
 // OnIpldPut returns the gas used for storing an object
 func (pl *pricelistV0) OnIpldPut(dataSize int) GasCharge {
-	return newGasCharge("OnIpldPut", pl.ipldPutBase, int64(dataSize)*pl.ipldPutPerByte).WithExtra(dataSize).WithVirtual(pl.ipldPutBase*8700+(pl.ipldPutPerByte*100), 0)
+	return newGasCharge("OnIpldPut", pl.ipldPutBase, int64(dataSize)*pl.ipldPutPerByte).
+		WithExtra(dataSize).WithVirtual(88970, 0)
 }
 
 // OnCreateActor returns the gas used for creating an actor
 func (pl *pricelistV0) OnCreateActor() GasCharge {
-	return newGasCharge("OnCreateActor", pl.createActorBase, pl.createActorExtra)
+	return newGasCharge("OnCreateActor", pl.createActorBase, pl.createActorExtra).WithVirtual(65636, 0)
 }
 
 // OnDeleteActor returns the gas used for deleting an actor
@@ -127,39 +146,52 @@ func (pl *pricelistV0) OnDeleteActor() GasCharge {
 }
 
 // OnVerifySignature
+
 func (pl *pricelistV0) OnVerifySignature(sigType crypto.SigType, planTextSize int) (GasCharge, error) {
 	costFn, ok := pl.verifySignature[sigType]
 	if !ok {
 		return GasCharge{}, fmt.Errorf("cost function for signature type %d not supported", sigType)
 	}
 	sigName, _ := sigType.Name()
-	return newGasCharge("OnVerifySignature", costFn(int64(planTextSize)), 0).WithExtra(sigName), nil
+	virtGas := int64(0)
+	switch sigType {
+	case crypto.SigTypeBLS:
+		virtGas = 220138570
+	case crypto.SigTypeSecp256k1:
+		virtGas = 7053730
+	}
+
+	return newGasCharge("OnVerifySignature", costFn(int64(planTextSize)), 0).
+		WithExtra(map[string]interface{}{
+			"type": sigName,
+			"size": planTextSize,
+		}).WithVirtual(virtGas, 0), nil
 }
 
 // OnHashing
 func (pl *pricelistV0) OnHashing(dataSize int) GasCharge {
-	return newGasCharge("OnHashing", pl.hashingBase+int64(dataSize)*pl.hashingPerByte, 0)
+	return newGasCharge("OnHashing", pl.hashingBase+int64(dataSize)*pl.hashingPerByte, 0).WithExtra(dataSize).WithVirtual(77300, 0)
 }
 
 // OnComputeUnsealedSectorCid
 func (pl *pricelistV0) OnComputeUnsealedSectorCid(proofType abi.RegisteredSealProof, pieces []abi.PieceInfo) GasCharge {
 	// TODO: this needs more cost tunning, check with @lotus
-	return newGasCharge("OnComputeUnsealedSectorCid", pl.computeUnsealedSectorCidBase, 0).WithVirtual(pl.computeUnsealedSectorCidBase*24500, 0)
+	return newGasCharge("OnComputeUnsealedSectorCid", pl.computeUnsealedSectorCidBase, 0).WithVirtual(382370, 0)
 }
 
 // OnVerifySeal
 func (pl *pricelistV0) OnVerifySeal(info abi.SealVerifyInfo) GasCharge {
 	// TODO: this needs more cost tunning, check with @lotus
-	return newGasCharge("OnVerifySeal", pl.verifySealBase, 0).WithVirtual(pl.verifySealBase*177500, 0)
+	return newGasCharge("OnVerifySeal", pl.verifySealBase, 0).WithVirtual(199954003, 0)
 }
 
 // OnVerifyPost
 func (pl *pricelistV0) OnVerifyPost(info abi.WindowPoStVerifyInfo) GasCharge {
 	// TODO: this needs more cost tunning, check with @lotus
-	return newGasCharge("OnVerifyPost", pl.verifyPostBase, 0)
+	return newGasCharge("OnVerifyPost", pl.verifyPostBase, 0).WithVirtual(2629471704, 0).WithExtra(len(info.ChallengedSectors))
 }
 
 // OnVerifyConsensusFault
 func (pl *pricelistV0) OnVerifyConsensusFault() GasCharge {
-	return newGasCharge("OnVerifyConsensusFault", pl.verifyConsensusFault, 0)
+	return newGasCharge("OnVerifyConsensusFault", pl.verifyConsensusFault, 0).WithVirtual(551935, 0)
 }

chain/vm/runtime.go

@@ -264,6 +264,7 @@ func (rt *Runtime) CreateActor(codeID cid.Cid, address address.Address) {
 	if err != nil {
 		panic(aerrors.Fatalf("creating actor entry: %v", err))
 	}
+	_ = rt.chargeGasSafe(gasOnActorExec)
 }
 
 func (rt *Runtime) DeleteActor(addr address.Address) {
@@ -284,10 +285,10 @@ func (rt *Runtime) DeleteActor(addr address.Address) {
 	if err := rt.state.DeleteActor(rt.Message().Receiver()); err != nil {
 		panic(aerrors.Fatalf("failed to delete actor: %s", err))
 	}
+	_ = rt.chargeGasSafe(gasOnActorExec)
 }
 
 func (rt *Runtime) Syscalls() vmr.Syscalls {
-	// TODO: Make sure this is wrapped in something that charges gas for each of the calls
 	return rt.sys
 }
 
@@ -367,6 +368,7 @@ func (rt *Runtime) Send(to address.Address, method abi.MethodNum, m vmr.CBORMars
 		log.Warnf("vmctx send failed: to: %s, method: %d: ret: %d, err: %s", to, method, ret, err)
 		return nil, err.RetCode()
 	}
+	_ = rt.chargeGasSafe(gasOnActorExec)
 	return &dumbWrapperType{ret}, 0
 }
 

chain/vm/vm.go

@@ -35,6 +35,7 @@ import (
 )
 
 var log = logging.Logger("vm")
+var gasOnActorExec = newGasCharge("OnActorExec", 0, 0)
 
 // ResolveToKeyAddr returns the public key type of address (`BLS`/`SECP256K1`) of an account actor identified by `addr`.
 func ResolveToKeyAddr(state types.StateTree, cst cbor.IpldStore, addr address.Address) (address.Address, aerrors.ActorError) {
@@ -68,11 +69,13 @@ type gasChargingBlocks struct {
 }
 
 func (bs *gasChargingBlocks) Get(c cid.Cid) (block.Block, error) {
+	bs.chargeGas(newGasCharge("OnIpldGetStart", 0, 0))
 	blk, err := bs.under.Get(c)
 	if err != nil {
 		return nil, aerrors.Escalate(err, "failed to get block from blockstore")
 	}
 	bs.chargeGas(bs.pricelist.OnIpldGet(len(blk.RawData())))
+	bs.chargeGas(gasOnActorExec)
 
 	return blk, nil
 }
@@ -83,6 +86,7 @@ func (bs *gasChargingBlocks) Put(blk block.Block) error {
 	if err := bs.under.Put(blk); err != nil {
 		return aerrors.Escalate(err, "failed to write data to disk")
 	}
+	bs.chargeGas(gasOnActorExec)
 	return nil
 }
 
@@ -231,7 +235,9 @@ func (vm *VM) send(ctx context.Context, msg *types.Message, parent *Runtime,
 
 		if msg.Method != 0 {
 			var ret []byte
+			_ = rt.chargeGasSafe(gasOnActorExec)
 			ret, err := vm.Invoke(toActor, rt, msg.Method, msg.Params)
+			_ = rt.chargeGasSafe(newGasCharge("OnActorExecDone", 0, 0))
 			return ret, err
 		}
 		return nil, nil

cmd/lotus-bench/import.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"math"
+	"net/http"
+	_ "net/http/pprof"
 	"os"
 	"runtime"
 	"runtime/pprof"
@@ -226,19 +229,222 @@ func compStats(vals []float64) (float64, float64) {
 	return av, math.Sqrt(varsum / float64(len(vals)))
 }
 
-func tallyGasCharges(charges map[string][]float64, et *types.ExecutionTrace) {
+type stats struct {
-	for _, gc := range et.GasCharges {
+	timeTaken meanVar
+	gasRatio  meanVar
 
-		compGas := gc.ComputeGas + gc.VirtualComputeGas
+	extraCovar *covar
-		ratio := float64(compGas) / float64(gc.TimeTaken.Nanoseconds())
+}
 
-		charges[gc.Name] = append(charges[gc.Name], 1/(ratio/GasPerNs))
+type covar struct {
-		//fmt.Printf("%s: %d, %s: %0.2f\n", gc.Name, compGas, gc.TimeTaken, 1/(ratio/GasPerNs))
+	meanX float64
-		for _, sub := range et.Subcalls {
+	meanY float64
-			tallyGasCharges(charges, &sub)
+	c     float64
-		}
+	n     float64
+	m2x   float64
+	m2y   float64
+}
+
+func (cov1 *covar) Covariance() float64 {
+	return cov1.c / (cov1.n - 1)
+}
+
+func (cov1 *covar) VarianceX() float64 {
+	return cov1.m2x / (cov1.n - 1)
+}
+
+func (v1 *covar) StddevX() float64 {
+	return math.Sqrt(v1.VarianceX())
+}
+
+func (cov1 *covar) VarianceY() float64 {
+	return cov1.m2y / (cov1.n - 1)
+}
+
+func (v1 *covar) StddevY() float64 {
+	return math.Sqrt(v1.VarianceY())
+}
+
+func (cov1 *covar) AddPoint(x, y float64) {
+	cov1.n += 1
+
+	dx := x - cov1.meanX
+	cov1.meanX += dx / cov1.n
+	dx2 := x - cov1.meanX
+	cov1.m2x += dx * dx2
+
+	dy := y - cov1.meanY
+	cov1.meanY += dy / cov1.n
+	dy2 := y - cov1.meanY
+	cov1.m2y += dy * dy2
+
+	cov1.c += dx * dy
+}
+
+func (cov1 *covar) Combine(cov2 *covar) {
+	if cov1.n == 0 {
+		*cov1 = *cov2
+		return
+	}
+	if cov2.n == 0 {
+		return
 	}
 
+	if cov1.n == 1 {
+		cpy := *cov2
+		cpy.AddPoint(cov2.meanX, cov2.meanY)
+		*cov1 = cpy
+		return
+	}
+	if cov2.n == 1 {
+		cov1.AddPoint(cov2.meanX, cov2.meanY)
+	}
+
+	out := covar{}
+	out.n = cov1.n + cov2.n
+
+	dx := cov1.meanX - cov2.meanX
+	out.meanX = cov1.meanX - dx*cov2.n/out.n
+	out.m2x = cov1.m2x + cov2.m2x + dx*dx*cov1.n*cov2.n/out.n
+
+	dy := cov1.meanY - cov2.meanY
+	out.meanY = cov1.meanY - dy*cov2.n/out.n
+	out.m2y = cov1.m2y + cov2.m2y + dy*dy*cov1.n*cov2.n/out.n
+
+	out.c = cov1.c + cov2.c + dx*dy*cov1.n*cov2.n/out.n
+	*cov1 = out
+}
+
+func (cov1 *covar) A() float64 {
+	return cov1.Covariance() / cov1.VarianceX()
+}
+func (cov1 *covar) B() float64 {
+	return cov1.meanY - cov1.meanX*cov1.A()
+}
+func (cov1 *covar) Correl() float64 {
+	return cov1.Covariance() / cov1.StddevX() / cov1.StddevY()
+}
+
+type meanVar struct {
+	n    float64
+	mean float64
+	m2   float64
+}
+
+func (v1 *meanVar) AddPoint(value float64) {
+	// based on https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm
+	v1.n += 1
+	delta := value - v1.mean
+	v1.mean += delta / v1.n
+	delta2 := value - v1.mean
+	v1.m2 += delta * delta2
+}
+
+func (v1 *meanVar) Variance() float64 {
+	return v1.m2 / (v1.n - 1)
+}
+func (v1 *meanVar) Mean() float64 {
+	return v1.mean
+}
+func (v1 *meanVar) Stddev() float64 {
+	return math.Sqrt(v1.Variance())
+}
+
+func (v1 *meanVar) Combine(v2 *meanVar) {
+	if v1.n == 0 {
+		*v1 = *v2
+		return
+	}
+	if v2.n == 0 {
+		return
+	}
+	if v1.n == 1 {
+		cpy := *v2
+		cpy.AddPoint(v1.mean)
+		*v1 = cpy
+		return
+	}
+	if v2.n == 1 {
+		v1.AddPoint(v2.mean)
+		return
+	}
+
+	newCount := v1.n + v2.n
+	delta := v2.mean - v1.mean
+	meanDelta := delta * v2.n / newCount
+	m2 := v1.m2 + v2.m2 + delta*meanDelta*v1.n
+	v1.n = newCount
+	v1.mean += meanDelta
+	v1.m2 = m2
+}
+
+func getExtras(ex interface{}) (*string, *float64) {
+	if t, ok := ex.(string); ok {
+		return &t, nil
+	}
+	if size, ok := ex.(float64); ok {
+		return nil, &size
+	}
+	if exMap, ok := ex.(map[string]interface{}); ok {
+		t, tok := exMap["type"].(string)
+		size, sok := exMap["size"].(float64)
+		if tok && sok {
+			return &t, &size
+		}
+		if tok {
+			return &t, nil
+		}
+		if sok {
+			return nil, &size
+		}
+		return nil, nil
+	}
+	return nil, nil
+}
+
+func tallyGasCharges(charges map[string]*stats, et types.ExecutionTrace) {
+	for i, gc := range et.GasCharges {
+		name := gc.Name
+		if name == "OnIpldGetStart" {
+			continue
+		}
+		tt := float64(gc.TimeTaken.Nanoseconds())
+		if name == "OnIpldGet" {
+			prev := et.GasCharges[i-1]
+			if prev.Name != "OnIpldGetStart" {
+				log.Warn("OnIpldGet without OnIpldGetStart")
+			}
+			tt += float64(prev.TimeTaken.Nanoseconds())
+		}
+		eType, eSize := getExtras(gc.Extra)
+		if eType != nil {
+			name += "-" + *eType
+		}
+		compGas := gc.VirtualComputeGas
+		if compGas == 0 {
+			compGas = 1
+		}
+		s := charges[name]
+		if s == nil {
+			s = new(stats)
+			charges[name] = s
+		}
+
+		if eSize != nil {
+			if s.extraCovar == nil {
+				s.extraCovar = &covar{}
+			}
+			s.extraCovar.AddPoint(*eSize, tt)
+		}
+
+		s.timeTaken.AddPoint(tt)
+
+		ratio := tt / float64(compGas) * GasPerNs
+		s.gasRatio.AddPoint(ratio)
+	}
+	for _, sub := range et.Subcalls {
+		tallyGasCharges(charges, sub)
+	}
 }
 
 var importAnalyzeCmd = &cli.Command{
@@ -249,57 +455,143 @@ var importAnalyzeCmd = &cli.Command{
 			return nil
 		}
 
+		go func() {
+			http.ListenAndServe("localhost:6060", nil)
+		}()
+
 		fi, err := os.Open(cctx.Args().First())
 		if err != nil {
 			return err
 		}
 
-		var results []TipSetExec
+		const nWorkers = 16
-		for {
+		jsonIn := make(chan []byte, 2*nWorkers)
-			var tse TipSetExec
+		type result struct {
-			if err := json.NewDecoder(fi).Decode(&tse); err != nil {
+			totalTime       time.Duration
-				if err != io.EOF {
+			chargeStats     map[string]*stats
-					return err
+			expensiveInvocs []Invocation
-				}
-				break
-			}
-			results = append(results, tse)
 		}
 
-		chargeDeltas := make(map[string][]float64)
+		results := make(chan result, nWorkers)
+
+		for i := 0; i < nWorkers; i++ {
+			go func() {
+				chargeStats := make(map[string]*stats)
+				var totalTime time.Duration
+				const invocsKeep = 32
+				var expensiveInvocs = make([]Invocation, 0, 8*invocsKeep)
+				var leastExpensiveInvoc = time.Duration(0)
+
+				for {
+					b, ok := <-jsonIn
+					if !ok {
+						results <- result{
+							totalTime:       totalTime,
+							chargeStats:     chargeStats,
+							expensiveInvocs: expensiveInvocs,
+						}
+						return
+					}
+
+					var tse TipSetExec
+					err := json.Unmarshal(b, &tse)
+					if err != nil {
+						log.Warnf("error unmarshaling tipset: %+v", err)
+						continue
+					}
+
+					totalTime += tse.Duration
+					for _, inv := range tse.Trace {
+						if inv.Duration > leastExpensiveInvoc {
+							expensiveInvocs = append(expensiveInvocs, Invocation{
+								TipSet: tse.TipSet,
+								Invoc:  inv,
+							})
+						}
+
+						tallyGasCharges(chargeStats, inv.ExecutionTrace)
+					}
+					if len(expensiveInvocs) > 4*invocsKeep {
+						sort.Slice(expensiveInvocs, func(i, j int) bool {
+							return expensiveInvocs[i].Invoc.Duration > expensiveInvocs[j].Invoc.Duration
+						})
+						leastExpensiveInvoc = expensiveInvocs[len(expensiveInvocs)-1].Invoc.Duration
+						n := 30
+						if len(expensiveInvocs) < n {
+							n = len(expensiveInvocs)
+						}
+						expensiveInvocs = expensiveInvocs[:n]
+					}
+				}
+			}()
+		}
+
+		var totalTipsets int64
+		reader := bufio.NewReader(fi)
+		for {
+			b, err := reader.ReadBytes('\n')
+			if err != nil && err != io.EOF {
+				if e, ok := err.(*json.SyntaxError); ok {
+					log.Warnf("syntax error at byte offset %d", e.Offset)
+				}
+				return err
+			}
+			totalTipsets++
+			jsonIn <- b
+			fmt.Fprintf(os.Stderr, "\rProcessed %d tipsets", totalTipsets)
+			if err == io.EOF {
+				break
+			}
+		}
+		close(jsonIn)
+		fmt.Fprintf(os.Stderr, "\n")
+		fmt.Fprintf(os.Stderr, "Collecting results\n")
 
 		var invocs []Invocation
 		var totalTime time.Duration
-		for i, r := range results {
+		var keys []string
-			_ = i
+		var charges = make(map[string]*stats)
-			totalTime += r.Duration
+		for i := 0; i < nWorkers; i++ {
-
+			fmt.Fprintf(os.Stderr, "\rProcessing results from worker %d/%d", i+1, nWorkers)
-			for _, inv := range r.Trace {
+			res := <-results
-				invocs = append(invocs, Invocation{
+			invocs = append(invocs, res.expensiveInvocs...)
-					TipSet: r.TipSet,
+			for k, v := range res.chargeStats {
-					Invoc:  inv,
+				s := charges[k]
-				})
+				if s == nil {
-
+					s = new(stats)
-				cgas, vgas := countGasCosts(&inv.ExecutionTrace)
+					charges[k] = s
-				fmt.Printf("Invocation: %d %s: %s %d -> %0.2f\n", inv.Msg.Method, inv.Msg.To, inv.Duration, cgas+vgas, float64(GasPerNs*inv.Duration.Nanoseconds())/float64(cgas+vgas))
+				}
-
+				s.timeTaken.Combine(&v.timeTaken)
-				tallyGasCharges(chargeDeltas, &inv.ExecutionTrace)
+				s.gasRatio.Combine(&v.gasRatio)
 
+				if v.extraCovar != nil {
+					if s.extraCovar == nil {
+						s.extraCovar = &covar{}
+					}
+					s.extraCovar.Combine(v.extraCovar)
+				}
 			}
+			totalTime += res.totalTime
 		}
 
-		var keys []string
+		fmt.Fprintf(os.Stderr, "\nCollecting gas keys\n")
-		for k := range chargeDeltas {
+		for k := range charges {
 			keys = append(keys, k)
 		}
 
 		fmt.Println("Gas Price Deltas")
 		sort.Strings(keys)
 		for _, k := range keys {
-			vals := chargeDeltas[k]
+			s := charges[k]
-			av, stdev := compStats(vals)
+			fmt.Printf("%s: incr by %.4f~%.4f; tt %.4f~%.4f\n", k, s.gasRatio.Mean(), s.gasRatio.Stddev(),
-
+				s.timeTaken.Mean(), s.timeTaken.Stddev())
-			fmt.Printf("%s: incr by %f (%f)\n", k, av, stdev)
+			if s.extraCovar != nil {
+				fmt.Printf("\t correll: %.2f, tt = %.2f * extra + %.2f\n", s.extraCovar.Correl(),
+					s.extraCovar.A(), s.extraCovar.B())
+				fmt.Printf("\t covar: %.2f, extra: %.2f~%.2f, tt2: %.2f~%.2f, count %.0f\n",
+					s.extraCovar.Covariance(), s.extraCovar.meanX, s.extraCovar.StddevX(),
+					s.extraCovar.meanY, s.extraCovar.StddevY(), s.extraCovar.n)
+			}
 		}
 
 		sort.Slice(invocs, func(i, j int) bool {
@@ -307,9 +599,20 @@ var importAnalyzeCmd = &cli.Command{
 		})
 
 		fmt.Println("Total time: ", totalTime)
-		fmt.Println("Average time per epoch: ", totalTime/time.Duration(len(results)))
+		fmt.Println("Average time per epoch: ", totalTime/time.Duration(totalTipsets))
+		if actorExec, ok := charges["OnActorExec"]; ok {
+			timeInActors := actorExec.timeTaken.Mean() * actorExec.timeTaken.n
+			fmt.Printf("Avarage time per epoch in actors: %s (%.1f%%)\n", time.Duration(timeInActors)/time.Duration(totalTipsets), timeInActors/float64(totalTime)*100)
+		}
+		if actorExecDone, ok := charges["OnActorExecDone"]; ok {
+			timeInActors := actorExecDone.timeTaken.Mean() * actorExecDone.timeTaken.n
+			fmt.Printf("Avarage time per epoch in OnActorExecDone %s (%.1f%%)\n", time.Duration(timeInActors)/time.Duration(totalTipsets), timeInActors/float64(totalTime)*100)
+		}
 
 		n := 30
+		if len(invocs) < n {
+			n = len(invocs)
+		}
 		fmt.Printf("Top %d most expensive calls:\n", n)
 		for i := 0; i < n; i++ {
 			inv := invocs[i].Invoc

cmd/lotus-bench/stats_test.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"math/rand"
+	"testing"
+)
+
+func TestMeanVar(t *testing.T) {
+	N := 16
+	ss := make([]*meanVar, N)
+	rng := rand.New(rand.NewSource(1))
+	for i := 0; i < N; i++ {
+		ss[i] = &meanVar{}
+		maxJ := rng.Intn(1000)
+		for j := 0; j < maxJ; j++ {
+			ss[i].AddPoint(rng.NormFloat64()*5 + 500)
+		}
+		t.Logf("mean: %f, stddev: %f, count %f", ss[i].mean, ss[i].Stddev(), ss[i].n)
+	}
+	out := &meanVar{}
+	for i := 0; i < N; i++ {
+		out.Combine(ss[i])
+		t.Logf("combine: mean: %f, stddev: %f", out.mean, out.Stddev())
+	}
+}
+
+func TestCovar(t *testing.T) {
+	N := 16
+	ss := make([]*covar, N)
+	rng := rand.New(rand.NewSource(1))
+	for i := 0; i < N; i++ {
+		ss[i] = &covar{}
+		maxJ := rng.Intn(1000) + 500
+		for j := 0; j < maxJ; j++ {
+			x := rng.NormFloat64()*5 + 500
+			ss[i].AddPoint(x, x*2-1000)
+		}
+		t.Logf("corell: %f, y = %f*x+%f @%.0f", ss[i].Correl(), ss[i].A(), ss[i].B(), ss[i].n)
+		t.Logf("\txVar: %f yVar: %f covar: %f", ss[i].StddevX(), ss[i].StddevY(), ss[i].Covariance())
+	}
+	out := &covar{}
+	for i := 0; i < N; i++ {
+		out.Combine(ss[i])
+		t.Logf("combine: corell: %f, y = %f*x+%f", out.Correl(), out.A(), out.B())
+		t.Logf("\txVar: %f yVar: %f covar: %f", out.StddevX(), out.StddevY(), out.Covariance())
+	}
+}