Browse Source

generate WAL on the fly (Refs #468)

pull/930/head
Anton Kaliaev 7 years ago
parent
commit
c6f025f40e
No known key found for this signature in database GPG Key ID: 7B6881D965918214
8 changed files with 188 additions and 255 deletions
  1. +2
    -3
      consensus/replay_test.go
  2. +0
    -36
      consensus/test_data/README.md
  3. +0
    -148
      consensus/test_data/build.sh
  4. BIN
      consensus/test_data/many_blocks.cswal
  5. +1
    -1
      consensus/wal.go
  6. +7
    -2
      consensus/wal_test.go
  7. +178
    -0
      consensus/walgen.go
  8. +0
    -65
      scripts/cutWALUntil/main.go

+ 2
- 3
consensus/replay_test.go View File

@ -265,7 +265,7 @@ func (w *crashingWAL) Wait() { w.next.Wait() }
// Handshake Tests // Handshake Tests
var ( var (
NUM_BLOCKS = 6 // number of blocks in the test_data/many_blocks.cswal
NUM_BLOCKS = 6
mempool = types.MockMempool{} mempool = types.MockMempool{}
) )
@ -324,8 +324,7 @@ func writeWAL(walMsgs []byte) string {
func testHandshakeReplay(t *testing.T, nBlocks int, mode uint) { func testHandshakeReplay(t *testing.T, nBlocks int, mode uint) {
config := ResetConfig("proxy_test_") config := ResetConfig("proxy_test_")
// copy the many_blocks file
walBody, err := cmn.ReadFile(path.Join(data_dir, "many_blocks.cswal"))
walBody, err := GenWAL(NUM_BLOCKS)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }


+ 0
- 36
consensus/test_data/README.md View File

@ -1,36 +0,0 @@
# Generating test data
To generate the data, run `build.sh`. See that script for more details.
Make sure to adjust the stepChanges in the testCases if the number of messages changes.
This sometimes happens for the `small_block2.cswal`, where the number of block parts changes between 4 and 5.
If you need to change the signatures, you can use a script as follows:
The privBytes comes from `config/tendermint_test/...`:
```
package main
import (
"encoding/hex"
"fmt"
"github.com/tendermint/go-crypto"
)
func main() {
signBytes, err := hex.DecodeString("7B22636861696E5F6964223A2274656E6465726D696E745F74657374222C22766F7465223A7B22626C6F636B5F68617368223A2242453544373939433846353044354645383533364334333932464443384537423342313830373638222C22626C6F636B5F70617274735F686561646572223A506172745365747B543A31204236323237323535464632307D2C22686569676874223A312C22726F756E64223A302C2274797065223A327D7D")
if err != nil {
panic(err)
}
privBytes, err := hex.DecodeString("27F82582AEFAE7AB151CFB01C48BB6C1A0DA78F9BDDA979A9F70A84D074EB07D3B3069C422E19688B45CBFAE7BB009FC0FA1B1EA86593519318B7214853803C8")
if err != nil {
panic(err)
}
privKey := crypto.PrivKeyEd25519{}
copy(privKey[:], privBytes)
signature := privKey.Sign(signBytes)
fmt.Printf("Signature Bytes: %X\n", signature.Bytes())
}
```

+ 0
- 148
consensus/test_data/build.sh View File

@ -1,148 +0,0 @@
#!/usr/bin/env bash
# Requires: killall command and jq JSON processor.
# Get the parent directory of where this script is.
SOURCE="${BASH_SOURCE[0]}"
while [ -h "$SOURCE" ] ; do SOURCE="$(readlink "$SOURCE")"; done
DIR="$( cd -P "$( dirname "$SOURCE" )/../.." && pwd )"
# Change into that dir because we expect that.
cd "$DIR" || exit 1
# Make sure we have a tendermint command.
if ! hash tendermint 2>/dev/null; then
make install
fi
# Make sure we have a cutWALUntil binary.
cutWALUntil=./scripts/cutWALUntil/cutWALUntil
cutWALUntilDir=$(dirname $cutWALUntil)
if ! hash $cutWALUntil 2>/dev/null; then
cd "$cutWALUntilDir" && go build && cd - || exit 1
fi
TMHOME=$(mktemp -d)
export TMHOME="$TMHOME"
if [[ ! -d "$TMHOME" ]]; then
echo "Could not create temp directory"
exit 1
else
echo "TMHOME: ${TMHOME}"
fi
# TODO: eventually we should replace with `tendermint init --test`
DIR_TO_COPY=$HOME/.tendermint_test/consensus_state_test
if [ ! -d "$DIR_TO_COPY" ]; then
echo "$DIR_TO_COPY does not exist. Please run: go test ./consensus"
exit 1
fi
echo "==> Copying ${DIR_TO_COPY} to ${TMHOME} directory..."
cp -r "$DIR_TO_COPY"/* "$TMHOME"
# preserve original genesis file because later it will be modified (see small_block2)
cp "$TMHOME/genesis.json" "$TMHOME/genesis.json.bak"
function reset(){
echo "==> Resetting tendermint..."
tendermint unsafe_reset_all
cp "$TMHOME/genesis.json.bak" "$TMHOME/genesis.json"
}
reset
# function empty_block(){
# echo "==> Starting tendermint..."
# tendermint node --proxy_app=persistent_dummy &> /dev/null &
# sleep 5
# echo "==> Killing tendermint..."
# killall tendermint
# echo "==> Copying WAL log..."
# $cutWALUntil "$TMHOME/data/cs.wal/wal" 1 consensus/test_data/new_empty_block.cswal
# mv consensus/test_data/new_empty_block.cswal consensus/test_data/empty_block.cswal
# reset
# }
function many_blocks(){
bash scripts/txs/random.sh 1000 36657 &> /dev/null &
PID=$!
echo "==> Starting tendermint..."
tendermint node --proxy_app=persistent_dummy &> /dev/null &
sleep 10
echo "==> Killing tendermint..."
kill -9 $PID
killall tendermint
echo "==> Copying WAL log..."
$cutWALUntil "$TMHOME/data/cs.wal/wal" 6 consensus/test_data/new_many_blocks.cswal
mv consensus/test_data/new_many_blocks.cswal consensus/test_data/many_blocks.cswal
reset
}
# function small_block1(){
# bash scripts/txs/random.sh 1000 36657 &> /dev/null &
# PID=$!
# echo "==> Starting tendermint..."
# tendermint node --proxy_app=persistent_dummy &> /dev/null &
# sleep 10
# echo "==> Killing tendermint..."
# kill -9 $PID
# killall tendermint
# echo "==> Copying WAL log..."
# $cutWALUntil "$TMHOME/data/cs.wal/wal" 1 consensus/test_data/new_small_block1.cswal
# mv consensus/test_data/new_small_block1.cswal consensus/test_data/small_block1.cswal
# reset
# }
# # block part size = 512
# function small_block2(){
# cat "$TMHOME/genesis.json" | jq '. + {consensus_params: {block_size_params: {max_bytes: 22020096}, block_gossip_params: {block_part_size_bytes: 512}}}' > "$TMHOME/new_genesis.json"
# mv "$TMHOME/new_genesis.json" "$TMHOME/genesis.json"
# bash scripts/txs/random.sh 1000 36657 &> /dev/null &
# PID=$!
# echo "==> Starting tendermint..."
# tendermint node --proxy_app=persistent_dummy &> /dev/null &
# sleep 5
# echo "==> Killing tendermint..."
# kill -9 $PID
# killall tendermint
# echo "==> Copying WAL log..."
# $cutWALUntil "$TMHOME/data/cs.wal/wal" 1 consensus/test_data/new_small_block2.cswal
# mv consensus/test_data/new_small_block2.cswal consensus/test_data/small_block2.cswal
# reset
# }
case "$1" in
# "small_block1")
# small_block1
# ;;
# "small_block2")
# small_block2
# ;;
# "empty_block")
# empty_block
# ;;
"many_blocks")
many_blocks
;;
*)
# small_block1
# small_block2
# empty_block
many_blocks
esac
echo "==> Cleaning up..."
rm -rf "$TMHOME"

BIN
consensus/test_data/many_blocks.cswal View File


+ 1
- 1
consensus/wal.go View File

@ -30,7 +30,7 @@ type TimedWALMessage struct {
} }
// EndHeightMessage marks the end of the given height inside WAL. // EndHeightMessage marks the end of the given height inside WAL.
// @internal used by scripts/cutWALUntil util.
// @internal used by scripts/wal2json util.
type EndHeightMessage struct { type EndHeightMessage struct {
Height int64 `json:"height"` Height int64 `json:"height"`
} }


+ 7
- 2
consensus/wal_test.go View File

@ -3,7 +3,6 @@ package consensus
import ( import (
"bytes" "bytes"
"crypto/rand" "crypto/rand"
"path"
"sync" "sync"
"testing" "testing"
"time" "time"
@ -43,7 +42,13 @@ func TestWALEncoderDecoder(t *testing.T) {
} }
func TestSearchForEndHeight(t *testing.T) { func TestSearchForEndHeight(t *testing.T) {
wal, err := NewWAL(path.Join(data_dir, "many_blocks.cswal"), false)
walBody, err := GenWAL(6)
if err != nil {
t.Fatal(err)
}
walFile := writeWAL(walBody)
wal, err := NewWAL(walFile, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }


+ 178
- 0
consensus/walgen.go View File

@ -0,0 +1,178 @@
// walgen provides an utility function for generating WAL on the fly.
package consensus
import (
"bufio"
"bytes"
"fmt"
"math/rand"
"os"
"path/filepath"
"strings"
"time"
"github.com/pkg/errors"
"github.com/tendermint/abci/example/dummy"
bc "github.com/tendermint/tendermint/blockchain"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/proxy"
sm "github.com/tendermint/tendermint/state"
"github.com/tendermint/tendermint/types"
auto "github.com/tendermint/tmlibs/autofile"
"github.com/tendermint/tmlibs/db"
"github.com/tendermint/tmlibs/log"
)
// GenWAL generates a consensus WAL. It does this by spining up a new node with a
// dummy application and special consensus wal instance (byteBufferWAL) and
// waits until numBlocks are created. Then it returns a WAL body.
func GenWAL(numBlocks int) (body []byte, err error) {
config := getConfig()
app := dummy.NewPersistentDummyApplication(filepath.Join(config.DBDir(), "genwal"))
logger := log.TestingLogger().With("walgen", "walgen")
/////////////////////////////////////////////////////////////////////////////
// COPY PASTE FROM node.go WITH A FEW MODIFICATIONS
// NOTE: we can't import node package because of circular dependency
privValidatorFile := config.PrivValidatorFile()
privValidator := types.LoadOrGenPrivValidatorFS(privValidatorFile)
genDoc, err := types.GenesisDocFromFile(config.GenesisFile())
if err != nil {
return nil, errors.Wrap(err, "failed to read genesis file")
}
stateDB := db.NewMemDB()
blockStoreDB := db.NewMemDB()
state, err := sm.MakeGenesisState(stateDB, genDoc)
state.SetLogger(logger.With("module", "state"))
if err != nil {
return nil, errors.Wrap(err, "failed to make genesis state")
}
blockStore := bc.NewBlockStore(blockStoreDB)
handshaker := NewHandshaker(state, blockStore)
proxyApp := proxy.NewAppConns(proxy.NewLocalClientCreator(app), handshaker)
proxyApp.SetLogger(logger.With("module", "proxy"))
if err := proxyApp.Start(); err != nil {
return nil, errors.Wrap(err, "failed to start proxy app connections")
}
defer proxyApp.Stop()
eventBus := types.NewEventBus()
eventBus.SetLogger(logger.With("module", "events"))
if err := eventBus.Start(); err != nil {
return nil, errors.Wrap(err, "failed to start event bus")
}
mempool := types.MockMempool{}
consensusState := NewConsensusState(config.Consensus, state.Copy(), proxyApp.Consensus(), blockStore, mempool)
consensusState.SetLogger(logger)
consensusState.SetEventBus(eventBus)
if privValidator != nil {
consensusState.SetPrivValidator(privValidator)
}
// END OF COPY PASTE
/////////////////////////////////////////////////////////////////////////////
// set consensus wal to buffered WAL, which will write all incoming msgs to buffer
var b bytes.Buffer
wr := bufio.NewWriter(&b)
numBlocksWritten := make(chan struct{})
wal := &byteBufferWAL{enc: NewWALEncoder(wr), heightToStop: int64(numBlocks), signalWhenStopsTo: numBlocksWritten}
wal.Save(EndHeightMessage{0})
consensusState.wal = wal
if err := consensusState.Start(); err != nil {
return nil, errors.Wrap(err, "failed to start consensus state")
}
defer consensusState.Stop()
select {
case <-numBlocksWritten:
wr.Flush()
return b.Bytes(), nil
case <-time.After(time.Duration(2*numBlocks) * time.Second):
return b.Bytes(), fmt.Errorf("waited too long for tendermint to produce %d blocks", numBlocks)
}
}
// f**ing long, but unique for each test
func makePathname() string {
// get path
p, err := os.Getwd()
if err != nil {
panic(err)
}
fmt.Println(p)
sep := string(filepath.Separator)
return strings.Replace(p, sep, "_", -1)
}
func randPort() int {
// returns between base and base + spread
base, spread := 20000, 20000
return base + rand.Intn(spread)
}
func makeAddrs() (string, string, string) {
start := randPort()
return fmt.Sprintf("tcp://0.0.0.0:%d", start),
fmt.Sprintf("tcp://0.0.0.0:%d", start+1),
fmt.Sprintf("tcp://0.0.0.0:%d", start+2)
}
// getConfig returns a config for test cases
func getConfig() *cfg.Config {
pathname := makePathname()
c := cfg.ResetTestRoot(pathname)
// and we use random ports to run in parallel
tm, rpc, grpc := makeAddrs()
c.P2P.ListenAddress = tm
c.RPC.ListenAddress = rpc
c.RPC.GRPCListenAddress = grpc
return c
}
// byteBufferWAL is a WAL which writes all msgs to a byte buffer. Writing stops
// when the heightToStop is reached. Client will be notified via
// signalWhenStopsTo channel.
type byteBufferWAL struct {
enc *WALEncoder
stopped bool
heightToStop int64
signalWhenStopsTo chan struct{}
}
var fixedTime, err = time.Parse(time.RFC3339, "2017-01-02T15:04:05Z")
// Save writes message to the internal buffer except when heightToStop is
// reached, in which case it signal the caller via signalWhenStopsTo and skip
// writing.
func (w *byteBufferWAL) Save(m WALMessage) {
if w.stopped {
return
}
if endMsg, ok := m.(EndHeightMessage); ok {
if endMsg.Height == w.heightToStop {
w.signalWhenStopsTo <- struct{}{}
w.stopped = true
return
}
}
err := w.enc.Encode(&TimedWALMessage{fixedTime, m})
if err != nil {
panic(fmt.Sprintf("failed to encode the msg %v", m))
}
}
func (w *byteBufferWAL) Group() *auto.Group {
panic("not implemented")
}
func (w *byteBufferWAL) SearchForEndHeight(height int64) (gr *auto.GroupReader, found bool, err error) {
return nil, false, nil
}
func (w *byteBufferWAL) Start() error { return nil }
func (w *byteBufferWAL) Stop() error { return nil }
func (w *byteBufferWAL) Wait() {}

+ 0
- 65
scripts/cutWALUntil/main.go View File

@ -1,65 +0,0 @@
/*
cutWALUntil is a small utility for cutting a WAL until the given height
(inclusively). Note it does not include last cs.EndHeightMessage.
Usage:
cutWALUntil <path-to-wal> height-to-stop <output-wal>
*/
package main
import (
"fmt"
"io"
"os"
"strconv"
cs "github.com/tendermint/tendermint/consensus"
)
func main() {
if len(os.Args) < 4 {
fmt.Println("3 arguments required: <path-to-wal> <height-to-stop> <output-wal>")
os.Exit(1)
}
var heightToStop int64
var err error
if heightToStop, err = strconv.ParseInt(os.Args[2], 10, 64); err != nil {
panic(fmt.Errorf("failed to parse height: %v", err))
}
in, err := os.Open(os.Args[1])
if err != nil {
panic(fmt.Errorf("failed to open input WAL file: %v", err))
}
defer in.Close()
out, err := os.Create(os.Args[3])
if err != nil {
panic(fmt.Errorf("failed to open output WAL file: %v", err))
}
defer out.Close()
enc := cs.NewWALEncoder(out)
dec := cs.NewWALDecoder(in)
for {
msg, err := dec.Decode()
if err == io.EOF {
break
} else if err != nil {
panic(fmt.Errorf("failed to decode msg: %v", err))
}
if m, ok := msg.Msg.(cs.EndHeightMessage); ok {
if m.Height == heightToStop {
break
}
}
err = enc.Encode(msg)
if err != nil {
panic(fmt.Errorf("failed to encode msg: %v", err))
}
}
}

Loading…
Cancel
Save