initial commit of p2p package

1 files changed, 275 insertions, 0 deletions

diff --git a/p2p/connection.go b/p2p/connection.go
new file mode 100644
index 000000000..e999cbe55
--- /dev/null
+++ b/p2p/connection.go
@@ -0,0 +1,275 @@
+package p2p
+
+import (
+	"bytes"
+	// "fmt"
+	"net"
+	"time"
+
+	"github.com/ethereum/eth-go/ethutil"
+)
+
+type Connection struct {
+	conn net.Conn
+	// conn       NetworkConnection
+	timeout    time.Duration
+	in         chan []byte
+	out        chan []byte
+	err        chan *PeerError
+	closingIn  chan chan bool
+	closingOut chan chan bool
+}
+
+// const readBufferLength = 2 //for testing
+
+const readBufferLength = 1440
+const partialsQueueSize = 10
+const maxPendingQueueSize = 1
+const defaultTimeout = 500
+
+var magicToken = []byte{34, 64, 8, 145}
+
+func (self *Connection) Open() {
+	go self.startRead()
+	go self.startWrite()
+}
+
+func (self *Connection) Close() {
+	self.closeIn()
+	self.closeOut()
+}
+
+func (self *Connection) closeIn() {
+	errc := make(chan bool)
+	self.closingIn <- errc
+	<-errc
+}
+
+func (self *Connection) closeOut() {
+	errc := make(chan bool)
+	self.closingOut <- errc
+	<-errc
+}
+
+func NewConnection(conn net.Conn, errchan chan *PeerError) *Connection {
+	return &Connection{
+		conn:       conn,
+		timeout:    defaultTimeout,
+		in:         make(chan []byte),
+		out:        make(chan []byte),
+		err:        errchan,
+		closingIn:  make(chan chan bool, 1),
+		closingOut: make(chan chan bool, 1),
+	}
+}
+
+func (self *Connection) Read() <-chan []byte {
+	return self.in
+}
+
+func (self *Connection) Write() chan<- []byte {
+	return self.out
+}
+
+func (self *Connection) Error() <-chan *PeerError {
+	return self.err
+}
+
+func (self *Connection) startRead() {
+	payloads := make(chan []byte)
+	done := make(chan *PeerError)
+	pending := [][]byte{}
+	var head []byte
+	var wait time.Duration // initally 0 (no delay)
+	read := time.After(wait * time.Millisecond)
+
+	for {
+		// if pending empty, nil channel blocks
+		var in chan []byte
+		if len(pending) > 0 {
+			in = self.in // enable send case
+			head = pending[0]
+		} else {
+			in = nil
+		}
+
+		select {
+		case <-read:
+			go self.read(payloads, done)
+		case err := <-done:
+			if err == nil { // no error but nothing to read
+				if len(pending) < maxPendingQueueSize {
+					wait = 100
+				} else if wait == 0 {
+					wait = 100
+				} else {
+					wait = 2 * wait
+				}
+			} else {
+				self.err <- err // report error
+				wait = 100
+			}
+			read = time.After(wait * time.Millisecond)
+		case payload := <-payloads:
+			pending = append(pending, payload)
+			if len(pending) < maxPendingQueueSize {
+				wait = 0
+			} else {
+				wait = 100
+			}
+			read = time.After(wait * time.Millisecond)
+		case in <- head:
+			pending = pending[1:]
+		case errc := <-self.closingIn:
+			errc <- true
+			close(self.in)
+			return
+		}
+
+	}
+}
+
+func (self *Connection) startWrite() {
+	pending := [][]byte{}
+	done := make(chan *PeerError)
+	writing := false
+	for {
+		if len(pending) > 0 && !writing {
+			writing = true
+			go self.write(pending[0], done)
+		}
+		select {
+		case payload := <-self.out:
+			pending = append(pending, payload)
+		case err := <-done:
+			if err == nil {
+				pending = pending[1:]
+				writing = false
+			} else {
+				self.err <- err // report error
+			}
+		case errc := <-self.closingOut:
+			errc <- true
+			close(self.out)
+			return
+		}
+	}
+}
+
+func pack(payload []byte) (packet []byte) {
+	length := ethutil.NumberToBytes(uint32(len(payload)), 32)
+	// return error if too long?
+	// Write magic token and payload length (first 8 bytes)
+	packet = append(magicToken, length...)
+	packet = append(packet, payload...)
+	return
+}
+
+func avoidPanic(done chan *PeerError) {
+	if rec := recover(); rec != nil {
+		err := NewPeerError(MiscError, " %v", rec)
+		logger.Debugln(err)
+		done <- err
+	}
+}
+
+func (self *Connection) write(payload []byte, done chan *PeerError) {
+	defer avoidPanic(done)
+	var err *PeerError
+	_, ok := self.conn.Write(pack(payload))
+	if ok != nil {
+		err = NewPeerError(WriteError, " %v", ok)
+		logger.Debugln(err)
+	}
+	done <- err
+}
+
+func (self *Connection) read(payloads chan []byte, done chan *PeerError) {
+	//defer avoidPanic(done)
+
+	partials := make(chan []byte, partialsQueueSize)
+	errc := make(chan *PeerError)
+	go self.readPartials(partials, errc)
+
+	packet := []byte{}
+	length := 8
+	start := true
+	var err *PeerError
+out:
+	for {
+		// appends partials read via connection until packet is
+		// - either parseable (>=8bytes)
+		// - or complete (payload fully consumed)
+		for len(packet) < length {
+			partial, ok := <-partials
+			if !ok { // partials channel is closed
+				err = <-errc
+				if err == nil && len(packet) > 0 {
+					if start {
+						err = NewPeerError(PacketTooShort, "%v", packet)
+					} else {
+						err = NewPeerError(PayloadTooShort, "%d < %d", len(packet), length)
+					}
+				}
+				break out
+			}
+			packet = append(packet, partial...)
+		}
+		if start {
+			// at least 8 bytes read, can validate packet
+			if bytes.Compare(magicToken, packet[:4]) != 0 {
+				err = NewPeerError(MagicTokenMismatch, " received %v", packet[:4])
+				break
+			}
+			length = int(ethutil.BytesToNumber(packet[4:8]))
+			packet = packet[8:]
+
+			if length > 0 {
+				start = false // now consuming payload
+			} else { //penalize peer but read on
+				self.err <- NewPeerError(EmptyPayload, "")
+				length = 8
+			}
+		} else {
+			// packet complete (payload fully consumed)
+			payloads <- packet[:length]
+			packet = packet[length:] // resclice packet
+			start = true
+			length = 8
+		}
+	}
+
+	// this stops partials read via the connection, should we?
+	//if err != nil {
+	//  select {
+	//    case errc <- err
+	//  default:
+	//}
+	done <- err
+}
+
+func (self *Connection) readPartials(partials chan []byte, errc chan *PeerError) {
+	defer close(partials)
+	for {
+		// Give buffering some time
+		self.conn.SetReadDeadline(time.Now().Add(self.timeout * time.Millisecond))
+		buffer := make([]byte, readBufferLength)
+		// read partial from connection
+		bytesRead, err := self.conn.Read(buffer)
+		if err == nil || err.Error() == "EOF" {
+			if bytesRead > 0 {
+				partials <- buffer[:bytesRead]
+			}
+			if err != nil && err.Error() == "EOF" {
+				break
+			}
+		} else {
+			// unexpected error, report to errc
+			err := NewPeerError(ReadError, " %v", err)
+			logger.Debugln(err)
+			errc <- err
+			return // will close partials channel
+		}
+	}
+	close(errc)
+}