Verify the Total Ordering Algorithm in peerServer in tcp mode (#11)

Verify the Total Ordering Algorithm in peerServer in tcp mode.

9 files changed, 328 insertions, 16 deletions

diff --git a/simulation/app.go b/simulation/app.go
index 02d1c1e..34b53f4 100644
--- a/simulation/app.go
+++ b/simulation/app.go
@@ -20,6 +20,7 @@ package simulation
 import (
 	"fmt"
 
+	"github.com/dexon-foundation/dexon-consensus-core/common"
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 )
 
@@ -28,12 +29,16 @@ type SimApp struct {
 	ValidatorID types.ValidatorID
 	Outputs     []*types.Block
 	Early       bool
+	Network     PeerServerNetwork
+	DeliverID   int
 }
 
 // NewSimApp returns point to a new instance of SimApp.
-func NewSimApp(id types.ValidatorID) *SimApp {
+func NewSimApp(id types.ValidatorID, Network PeerServerNetwork) *SimApp {
 	return &SimApp{
 		ValidatorID: id,
+		Network:     Network,
+		DeliverID:   0,
 	}
 }
 
@@ -47,4 +52,10 @@ func (a *SimApp) Deliver(blocks []*types.Block, early bool) {
 	a.Outputs = blocks
 	a.Early = early
 	fmt.Println("OUTPUT", a.ValidatorID, a.Early, a.Outputs)
+	blockHash := common.Hashes{}
+	for _, block := range blocks {
+		blockHash = append(blockHash, block.Hash)
+	}
+	a.Network.DeliverBlocks(blockHash, a.DeliverID)
+	a.DeliverID++
 }
diff --git a/simulation/block-list.go b/simulation/block-list.go
new file mode 100644
index 0000000..5bf9f20
--- /dev/null
+++ b/simulation/block-list.go
@@ -0,0 +1,60 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package simulation
+
+import (
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+)
+
+// BlockList is the list of blocks from the result of Total Ordering Algorithm.
+type BlockList struct {
+	ID        int           `json:"id"`
+	BlockHash common.Hashes `json:"blockhash"`
+	// The index is required by heap.Interface.
+	index int
+}
+
+// PendingBlockList is a PrioirtyQueue maintaining the BlockList received before the previous one (based on ID).
+type PendingBlockList []*BlockList
+
+// Len, Less and Swap are implementing heap.Interface
+func (p PendingBlockList) Len() int           { return len(p) }
+func (p PendingBlockList) Less(i, j int) bool { return p[i].ID < p[j].ID }
+func (p PendingBlockList) Swap(i, j int) {
+	p[i], p[j] = p[j], p[i]
+	p[i].index = i
+	p[j].index = j
+}
+
+// Push item in the Heap.
+func (p *PendingBlockList) Push(x interface{}) {
+	n := len(*p)
+	item := x.(*BlockList)
+	item.index = n
+	*p = append(*p, item)
+}
+
+// Pop the element from the Heap.
+func (p *PendingBlockList) Pop() interface{} {
+	old := *p
+	n := len(old)
+	item := old[n-1]
+	item.index = -1 // For safety.
+	*p = old[0 : n-1]
+	return item
+}
diff --git a/simulation/fake-network.go b/simulation/fake-network.go
index 923fab4..e85917e 100644
--- a/simulation/fake-network.go
+++ b/simulation/fake-network.go
@@ -22,11 +22,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/dexon-foundation/dexon-consensus-core/core"
+	"github.com/dexon-foundation/dexon-consensus-core/common"
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 )
 
-// FakeNetwork implements the core.Network interface.
+// FakeNetwork implements the Network interface.
 type FakeNetwork struct {
 	model Model
 
@@ -55,7 +55,7 @@ func (n *FakeNetwork) NumPeers() int {
 
 // Join allow a client to join the network. It reutnrs a interface{} channel for
 // the client to recieve information.
-func (n *FakeNetwork) Join(endpoint core.Endpoint) chan interface{} {
+func (n *FakeNetwork) Join(endpoint Endpoint) chan interface{} {
 	n.endpointMutex.Lock()
 	defer n.endpointMutex.Unlock()
 
@@ -93,3 +93,9 @@ func (n *FakeNetwork) BroadcastBlock(block *types.Block) {
 		n.Send(endpoint, block.Clone())
 	}
 }
+
+// DeliverBlocks sends blocks to peerServer.
+func (n *FakeNetwork) DeliverBlocks(blocks common.Hashes, id int) {
+	// TODO
+	return
+}
diff --git a/simulation/network.go b/simulation/network.go
new file mode 100644
index 0000000..7ce0dbc
--- /dev/null
+++ b/simulation/network.go
@@ -0,0 +1,42 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package simulation
+
+import (
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+// Endpoint is the interface for a client network endpoint.
+type Endpoint interface {
+	GetID() types.ValidatorID
+}
+
+// Network is the interface for network related functions.
+type Network interface {
+	PeerServerNetwork
+	Start()
+	NumPeers() int
+	Join(endpoint Endpoint) chan interface{}
+	BroadcastBlock(block *types.Block)
+}
+
+// PeerServerNetwork is the interface for peerServer network related functions
+type PeerServerNetwork interface {
+	DeliverBlocks(blocks common.Hashes, id int)
+}
diff --git a/simulation/peer-server.go b/simulation/peer-server.go
index 90bcf33..0d69f8e 100644
--- a/simulation/peer-server.go
+++ b/simulation/peer-server.go
@@ -20,6 +20,7 @@ package simulation
 import (
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"log"
 	"net"
 	"net/http"
@@ -31,14 +32,17 @@ import (
 
 // PeerServer is the main object for maintaining peer list.
 type PeerServer struct {
-	peers   map[types.ValidatorID]string
-	peersMu sync.Mutex
+	peers            map[types.ValidatorID]string
+	peersMu          sync.Mutex
+	peerTotalOrder   PeerTotalOrder
+	peerTotalOrderMu sync.Mutex
 }
 
 // NewPeerServer returns a new peer server.
 func NewPeerServer() *PeerServer {
 	return &PeerServer{
-		peers: make(map[types.ValidatorID]string),
+		peers:          make(map[types.ValidatorID]string),
+		peerTotalOrder: make(PeerTotalOrder),
 	}
 }
 
@@ -69,6 +73,7 @@ func (p *PeerServer) Run(configPath string) {
 
 		host, _, _ := net.SplitHostPort(r.RemoteAddr)
 		p.peers[id] = net.JoinHostPort(host, portString)
+		p.peerTotalOrder[id] = NewTotalOrderResult()
 		log.Printf("Peer %s joined from %s", id, p.peers[id])
 	}
 
@@ -105,10 +110,50 @@ func (p *PeerServer) Run(configPath string) {
 		w.Write(jsonText)
 	}
 
+	deliveryHandler := func(w http.ResponseWriter, r *http.Request) {
+		idString := r.Header.Get("ID")
+
+		defer r.Body.Close()
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		m := BlockList{}
+		if err := json.Unmarshal(body, &m); err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		id := types.ValidatorID{}
+		id.UnmarshalText([]byte(idString))
+
+		p.peerTotalOrderMu.Lock()
+		defer p.peerTotalOrderMu.Unlock()
+
+		readyForVerify := p.peerTotalOrder[id].PushBlocks(m)
+		if !readyForVerify {
+			return
+		}
+
+		// Verify the total order result.
+		go func(id types.ValidatorID) {
+			p.peerTotalOrderMu.Lock()
+			defer p.peerTotalOrderMu.Unlock()
+			var correct bool
+			p.peerTotalOrder, correct = VerifyTotalOrder(id, p.peerTotalOrder)
+			if !correct {
+				log.Printf("The result of Total Ordering Algorithm has error.\n")
+			}
+		}(id)
+	}
+
 	http.HandleFunc("/reset", resetHandler)
 	http.HandleFunc("/join", joinHandler)
 	http.HandleFunc("/peers", peersHandler)
 	http.HandleFunc("/info", infoHandler)
+	http.HandleFunc("/delivery", deliveryHandler)
 
 	addr := fmt.Sprintf("0.0.0.0:%d", peerPort)
 	log.Printf("Peer server started at %s", addr)
diff --git a/simulation/simulation.go b/simulation/simulation.go
index a219d61..8ec72ea 100644
--- a/simulation/simulation.go
+++ b/simulation/simulation.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 
 	"github.com/dexon-foundation/dexon-consensus-core/common"
-	"github.com/dexon-foundation/dexon-consensus-core/core"
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 	"github.com/dexon-foundation/dexon-consensus-core/simulation/config"
 )
@@ -39,7 +38,7 @@ func Run(configPath string) {
 		networkType == config.NetworkTypeTCPLocal {
 
 		var vs []*Validator
-		var network core.Network
+		var network Network
 
 		if networkType == config.NetworkTypeFake {
 			networkModel := &NormalNetwork{
diff --git a/simulation/tcp-network.go b/simulation/tcp-network.go
index 02475e1..464473b 100644
--- a/simulation/tcp-network.go
+++ b/simulation/tcp-network.go
@@ -28,15 +28,15 @@ import (
 	"sync"
 	"time"
 
-	"github.com/dexon-foundation/dexon-consensus-core/core"
+	"github.com/dexon-foundation/dexon-consensus-core/common"
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 )
 
-// TCPNetwork implements the core.Network interface.
+// TCPNetwork implements the Network interface.
 type TCPNetwork struct {
 	local    bool
 	port     int
-	endpoint core.Endpoint
+	endpoint Endpoint
 
 	peerServer    string
 	endpointMutex sync.RWMutex
@@ -116,7 +116,7 @@ func (n *TCPNetwork) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 // Join allow a client to join the network. It reutnrs a interface{} channel for
 // the client to recieve information.
-func (n *TCPNetwork) Join(endpoint core.Endpoint) chan interface{} {
+func (n *TCPNetwork) Join(endpoint Endpoint) chan interface{} {
 	n.endpointMutex.Lock()
 	defer n.endpointMutex.Unlock()
 
@@ -235,3 +235,41 @@ func (n *TCPNetwork) BroadcastBlock(block *types.Block) {
 		n.Send(endpoint, block.Clone())
 	}
 }
+
+// DeliverBlocks sends blocks to peerServer.
+func (n *TCPNetwork) DeliverBlocks(blocks common.Hashes, id int) {
+
+	message := BlockList{
+		ID:        id,
+		BlockHash: blocks,
+	}
+
+	messageJSON, err := json.Marshal(message)
+	if err != nil {
+		fmt.Printf("error: failed to marshal json: %v\n%+v\n", err, message)
+		return
+	}
+
+	msgURL := fmt.Sprintf("http://%s:%d/delivery", n.peerServer, peerPort)
+
+	go func() {
+		retries := 3
+		client := &http.Client{Timeout: 5 * time.Second}
+
+		for i := 0; i < retries; i++ {
+			req, err := http.NewRequest(
+				http.MethodPost, msgURL, strings.NewReader(string(messageJSON)))
+			if err != nil {
+				continue
+			}
+			req.Header.Add("ID", n.endpoint.GetID().String())
+
+			resp, err := client.Do(req)
+			if err == nil && resp.StatusCode == http.StatusOK {
+				runtime.Goexit()
+			}
+			time.Sleep(1 * time.Second)
+		}
+		fmt.Printf("failed to send message: %v\n", blocks)
+	}()
+}
diff --git a/simulation/validator.go b/simulation/validator.go
index c98efd2..b7ec2a0 100644
--- a/simulation/validator.go
+++ b/simulation/validator.go
@@ -31,7 +31,7 @@ import (
 
 // Validator represents a validator in DexCon.
 type Validator struct {
-	network core.Network
+	network Network
 	app     *SimApp
 
 	config     config.Validator
@@ -50,9 +50,9 @@ type Validator struct {
 func NewValidator(
 	id types.ValidatorID,
 	config config.Validator,
-	network core.Network,
+	network Network,
 	db *leveldb.DB) *Validator {
-	app := NewSimApp(id)
+	app := NewSimApp(id, network)
 	lattice := core.NewBlockLattice(blockdb.NewMemBackedBlockDB(), app)
 	return &Validator{
 		ID:      id,
diff --git a/simulation/verification.go b/simulation/verification.go
new file mode 100644
index 0000000..b32d719
--- /dev/null
+++ b/simulation/verification.go
@@ -0,0 +1,111 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package simulation
+
+import (
+	"container/heap"
+	"log"
+	"math"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+// TotalOrderResult is the object maintaining peer's result of
+// Total Ordering Algorithm.
+type TotalOrderResult struct {
+	hashList         common.Hashes
+	curID            int
+	pendingBlockList PendingBlockList
+}
+
+// PeerTotalOrder stores the TotalOrderResult of each validator.
+type PeerTotalOrder = map[types.ValidatorID]*TotalOrderResult
+
+// NewTotalOrderResult returns pointer to a a new TotalOrderResult instance.
+func NewTotalOrderResult() *TotalOrderResult {
+	totalOrder := &TotalOrderResult{}
+	heap.Init(&totalOrder.pendingBlockList)
+	return totalOrder
+}
+
+// PushBlocks push a BlockList into the TotalOrderResult and return true if
+// there is new blocks ready for verifiy
+func (totalOrder *TotalOrderResult) PushBlocks(blocks BlockList) (ready bool) {
+	if blocks.ID != totalOrder.curID {
+		heap.Push(&totalOrder.pendingBlockList, &blocks)
+		return false
+	}
+
+	// Append all of the consecutive blockList in the pendingBlockList.
+	for {
+		totalOrder.hashList = append(totalOrder.hashList, blocks.BlockHash...)
+		totalOrder.curID++
+		if len(totalOrder.pendingBlockList) == 0 ||
+			totalOrder.pendingBlockList[0].ID != totalOrder.curID {
+			break
+		}
+		blocks = *heap.Pop(&totalOrder.pendingBlockList).(*BlockList)
+	}
+	return true
+}
+
+// VerifyTotalOrder verifies if the result of Total Ordering Algorithm
+// returned by all validators are the same. However, the length of result
+// of each validators may not be the same, so only the common part is verified.
+func VerifyTotalOrder(id types.ValidatorID,
+	totalOrder PeerTotalOrder) (
+	unverifiedMap PeerTotalOrder, correct bool) {
+
+	hasError := false
+
+	// Get the common length from all validators.
+	length := math.MaxInt32
+	for _, peerTotalOrder := range totalOrder {
+		if len(peerTotalOrder.hashList) < length {
+			length = len(peerTotalOrder.hashList)
+		}
+	}
+
+	// Verify if the order of the blocks are the same by comparing
+	// the hash value.
+	for i := 0; i < length; i++ {
+		hash := totalOrder[id].hashList[i]
+		for vid, peerTotalOrder := range totalOrder {
+			if peerTotalOrder.hashList[i] != hash {
+				log.Printf("[%d] Unexpected hash %v from %v\n", i,
+					peerTotalOrder.hashList[i], vid)
+				hasError = true
+			}
+		}
+		if hasError {
+			log.Printf("[%d] Hash is %v from %v\n", i, hash, id)
+		} else {
+			log.Printf("Block %v confirmed\n", hash)
+		}
+	}
+
+	// Remove verified block from list.
+	if length > 0 {
+		for vid := range totalOrder {
+			totalOrder[vid].hashList =
+				totalOrder[vid].hashList[length:]
+		}
+	}
+	return totalOrder, !hasError
+}