mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-10-01 04:05:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

go fmt

Browse Source
This commit is contained in:
Jae Kwon
2014-07-01 14:50:24 -07:00
parent fa07748d23
commit c40fc65e6b
41 changed files with 3176 additions and 2938 deletions
Download Patch File Download Diff File Expand all files Collapse all files
peer/addrbook.go
+388 -359
View File
@@ -5,217 +5,236 @@
package peer
import (
. "github.com/tendermint/tendermint/binary"
crand "crypto/rand" // for seeding
"encoding/binary"
"encoding/json"
"io"
"math"
"math/rand"
"net"
"sync"
"sync/atomic"
"time"
"os"
"fmt"
crand "crypto/rand" // for seeding
"encoding/binary"
"encoding/json"
"fmt"
. "github.com/tendermint/tendermint/binary"
"io"
"math"
"math/rand"
"net"
"os"
"sync"
"sync/atomic"
"time"
)
/* AddrBook - concurrency safe peer address manager */
type AddrBook struct {
filePath string
filePath string
mtx sync.Mutex
rand *rand.Rand
key [32]byte
addrIndex map[string]*KnownAddress // addr.String() -> KnownAddress
addrNew [newBucketCount]map[string]*KnownAddress
addrOld [oldBucketCount][]*KnownAddress
started int32
shutdown int32
wg sync.WaitGroup
quit chan struct{}
nOld int
nNew int
mtx sync.Mutex
rand *rand.Rand
key [32]byte
addrIndex map[string]*KnownAddress // addr.String() -> KnownAddress
addrNew [newBucketCount]map[string]*KnownAddress
addrOld [oldBucketCount][]*KnownAddress
started int32
shutdown int32
wg sync.WaitGroup
quit chan struct{}
nOld int
nNew int
}
const (
// addresses under which the address manager will claim to need more addresses.
needAddressThreshold = 1000
// addresses under which the address manager will claim to need more addresses.
needAddressThreshold = 1000
// interval used to dump the address cache to disk for future use.
dumpAddressInterval = time.Minute * 2
// interval used to dump the address cache to disk for future use.
dumpAddressInterval = time.Minute * 2
// max addresses in each old address bucket.
oldBucketSize = 64
// max addresses in each old address bucket.
oldBucketSize = 64
// buckets we split old addresses over.
oldBucketCount = 64
// buckets we split old addresses over.
oldBucketCount = 64
// max addresses in each new address bucket.
newBucketSize = 64
// max addresses in each new address bucket.
newBucketSize = 64
// buckets that we spread new addresses over.
newBucketCount = 256
// buckets that we spread new addresses over.
newBucketCount = 256
// old buckets over which an address group will be spread.
oldBucketsPerGroup = 4
// old buckets over which an address group will be spread.
oldBucketsPerGroup = 4
// new buckets over which an source address group will be spread.
newBucketsPerGroup = 32
// new buckets over which an source address group will be spread.
newBucketsPerGroup = 32
// buckets a frequently seen new address may end up in.
newBucketsPerAddress = 4
// buckets a frequently seen new address may end up in.
newBucketsPerAddress = 4
// days before which we assume an address has vanished
// if we have not seen it announced in that long.
numMissingDays = 30
// days before which we assume an address has vanished
// if we have not seen it announced in that long.
numMissingDays = 30
// tries without a single success before we assume an address is bad.
numRetries = 3
// tries without a single success before we assume an address is bad.
numRetries = 3
// max failures we will accept without a success before considering an address bad.
maxFailures = 10
// max failures we will accept without a success before considering an address bad.
maxFailures = 10
// days since the last success before we will consider evicting an address.
minBadDays = 7
// days since the last success before we will consider evicting an address.
minBadDays = 7
// max addresses that we will send in response to a getAddr
// (in practise the most addresses we will return from a call to AddressCache()).
getAddrMax = 2500
// max addresses that we will send in response to a getAddr
// (in practise the most addresses we will return from a call to AddressCache()).
getAddrMax = 2500
// % of total addresses known that we will share with a call to AddressCache.
getAddrPercent = 23
// % of total addresses known that we will share with a call to AddressCache.
getAddrPercent = 23
// current version of the on-disk format.
serialisationVersion = 1
// current version of the on-disk format.
serialisationVersion = 1
)
// Use Start to begin processing asynchronous address updates.
func NewAddrBook(filePath string) *AddrBook {
am := AddrBook{
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
quit: make(chan struct{}),
filePath: filePath,
}
am.init()
return &am
am := AddrBook{
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
quit: make(chan struct{}),
filePath: filePath,
}
am.init()
return &am
}
// When modifying this, don't forget to update loadFromFile()
func (a *AddrBook) init() {
a.addrIndex = make(map[string]*KnownAddress)
io.ReadFull(crand.Reader, a.key[:])
for i := range a.addrNew {
a.addrNew[i] = make(map[string]*KnownAddress)
}
for i := range a.addrOld {
a.addrOld[i] = make([]*KnownAddress, 0, oldBucketSize)
}
a.addrIndex = make(map[string]*KnownAddress)
io.ReadFull(crand.Reader, a.key[:])
for i := range a.addrNew {
a.addrNew[i] = make(map[string]*KnownAddress)
}
for i := range a.addrOld {
a.addrOld[i] = make([]*KnownAddress, 0, oldBucketSize)
}
}
func (a *AddrBook) Start() {
if atomic.AddInt32(&a.started, 1) != 1 { return }
log.Trace("Starting address manager")
a.loadFromFile(a.filePath)
a.wg.Add(1)
go a.addressHandler()
if atomic.AddInt32(&a.started, 1) != 1 {
return
}
log.Trace("Starting address manager")
a.loadFromFile(a.filePath)
a.wg.Add(1)
go a.addressHandler()
}
func (a *AddrBook) Stop() {
if atomic.AddInt32(&a.shutdown, 1) != 1 { return }
log.Infof("Address manager shutting down")
close(a.quit)
a.wg.Wait()
if atomic.AddInt32(&a.shutdown, 1) != 1 {
return
}
log.Infof("Address manager shutting down")
close(a.quit)
a.wg.Wait()
}
func (a *AddrBook) AddAddress(addr *NetAddress, src *NetAddress) {
a.mtx.Lock(); defer a.mtx.Unlock()
a.addAddress(addr, src)
a.mtx.Lock()
defer a.mtx.Unlock()
a.addAddress(addr, src)
}
func (a *AddrBook) NeedMoreAddresses() bool {
return a.NumAddresses() < needAddressThreshold
return a.NumAddresses() < needAddressThreshold
}
func (a *AddrBook) NumAddresses() int {
a.mtx.Lock(); defer a.mtx.Unlock()
return a.nOld + a.nNew
a.mtx.Lock()
defer a.mtx.Unlock()
return a.nOld + a.nNew
}
// Pick a new address to connect to.
func (a *AddrBook) PickAddress(class string, newBias int) *KnownAddress {
a.mtx.Lock(); defer a.mtx.Unlock()
a.mtx.Lock()
defer a.mtx.Unlock()
if a.nOld == 0 && a.nNew == 0 { return nil }
if newBias > 100 { newBias = 100 }
if newBias < 0 { newBias = 0 }
if a.nOld == 0 && a.nNew == 0 {
return nil
}
if newBias > 100 {
newBias = 100
}
if newBias < 0 {
newBias = 0
}
// Bias between new and old addresses.
oldCorrelation := math.Sqrt(float64(a.nOld)) * (100.0 - float64(newBias))
newCorrelation := math.Sqrt(float64(a.nNew)) * float64(newBias)
// Bias between new and old addresses.
oldCorrelation := math.Sqrt(float64(a.nOld)) * (100.0 - float64(newBias))
newCorrelation := math.Sqrt(float64(a.nNew)) * float64(newBias)
if (newCorrelation+oldCorrelation)*a.rand.Float64() < oldCorrelation {
// pick random Old bucket.
var bucket []*KnownAddress = nil
for len(bucket) == 0 {
bucket = a.addrOld[a.rand.Intn(len(a.addrOld))]
}
// pick a random ka from bucket.
return bucket[a.rand.Intn(len(bucket))]
} else {
// pick random New bucket.
var bucket map[string]*KnownAddress = nil
for len(bucket) == 0 {
bucket = a.addrNew[a.rand.Intn(len(a.addrNew))]
}
// pick a random ka from bucket.
randIndex := a.rand.Intn(len(bucket))
for _, ka := range bucket {
randIndex--
if randIndex == 0 {
return ka
}
}
panic("Should not happen")
}
return nil
if (newCorrelation+oldCorrelation)*a.rand.Float64() < oldCorrelation {
// pick random Old bucket.
var bucket []*KnownAddress = nil
for len(bucket) == 0 {
bucket = a.addrOld[a.rand.Intn(len(a.addrOld))]
}
// pick a random ka from bucket.
return bucket[a.rand.Intn(len(bucket))]
} else {
// pick random New bucket.
var bucket map[string]*KnownAddress = nil
for len(bucket) == 0 {
bucket = a.addrNew[a.rand.Intn(len(a.addrNew))]
}
// pick a random ka from bucket.
randIndex := a.rand.Intn(len(bucket))
for _, ka := range bucket {
randIndex--
if randIndex == 0 {
return ka
}
}
panic("Should not happen")
}
return nil
}
func (a *AddrBook) MarkGood(addr *NetAddress) {
a.mtx.Lock(); defer a.mtx.Unlock()
ka := a.addrIndex[addr.String()]
if ka == nil { return }
ka.MarkAttempt(true)
if ka.OldBucket == -1 {
a.moveToOld(ka)
}
a.mtx.Lock()
defer a.mtx.Unlock()
ka := a.addrIndex[addr.String()]
if ka == nil {
return
}
ka.MarkAttempt(true)
if ka.OldBucket == -1 {
a.moveToOld(ka)
}
}
func (a *AddrBook) MarkAttempt(addr *NetAddress) {
a.mtx.Lock(); defer a.mtx.Unlock()
ka := a.addrIndex[addr.String()]
if ka == nil { return }
ka.MarkAttempt(false)
a.mtx.Lock()
defer a.mtx.Unlock()
ka := a.addrIndex[addr.String()]
if ka == nil {
return
}
ka.MarkAttempt(false)
}
/* Loading & Saving */
type addrBookJSON struct {
Key [32]byte
AddrNew [newBucketCount]map[string]*KnownAddress
AddrOld [oldBucketCount][]*KnownAddress
NOld int
NNew int
Key [32]byte
AddrNew [newBucketCount]map[string]*KnownAddress
AddrOld [oldBucketCount][]*KnownAddress
NOld int
NNew int
}
func (a *AddrBook) saveToFile(filePath string) {
aJSON := &addrBookJSON{
Key: a.key,
AddrNew: a.addrNew,
AddrOld: a.addrOld,
NOld: a.nOld,
NNew: a.nNew,
}
aJSON := &addrBookJSON{
Key: a.key,
AddrNew: a.addrNew,
AddrOld: a.addrOld,
NOld: a.nOld,
NNew: a.nNew,
}
w, err := os.Create(filePath)
if err != nil {
@@ -225,296 +244,306 @@ func (a *AddrBook) saveToFile(filePath string) {
enc := json.NewEncoder(w)
defer w.Close()
err = enc.Encode(&aJSON)
if err != nil { panic(err) }
if err != nil {
panic(err)
}
}
func (a *AddrBook) loadFromFile(filePath string) {
// If doesn't exist, do nothing.
// If doesn't exist, do nothing.
_, err := os.Stat(filePath)
if os.IsNotExist(err) { return }
if os.IsNotExist(err) {
return
}
// Load addrBookJSON{}
// Load addrBookJSON{}
r, err := os.Open(filePath)
if err != nil {
panic(fmt.Errorf("%s error opening file: %v", filePath, err))
panic(fmt.Errorf("%s error opening file: %v", filePath, err))
}
defer r.Close()
aJSON := &addrBookJSON{}
aJSON := &addrBookJSON{}
dec := json.NewDecoder(r)
err = dec.Decode(aJSON)
if err != nil {
panic(fmt.Errorf("error reading %s: %v", filePath, err))
}
// Now we need to initialize self.
// Now we need to initialize self.
copy(a.key[:], aJSON.Key[:])
a.addrNew = aJSON.AddrNew
for i, oldBucket := range aJSON.AddrOld {
copy(a.addrOld[i], oldBucket)
}
a.nNew = aJSON.NNew
a.nOld = aJSON.NOld
copy(a.key[:], aJSON.Key[:])
a.addrNew = aJSON.AddrNew
for i, oldBucket := range aJSON.AddrOld {
copy(a.addrOld[i], oldBucket)
}
a.nNew = aJSON.NNew
a.nOld = aJSON.NOld
a.addrIndex = make(map[string]*KnownAddress)
for _, newBucket := range a.addrNew {
for key, ka := range newBucket {
a.addrIndex[key] = ka
}
}
a.addrIndex = make(map[string]*KnownAddress)
for _, newBucket := range a.addrNew {
for key, ka := range newBucket {
a.addrIndex[key] = ka
}
}
}
/* Private methods */
func (a *AddrBook) addressHandler() {
dumpAddressTicker := time.NewTicker(dumpAddressInterval)
dumpAddressTicker := time.NewTicker(dumpAddressInterval)
out:
for {
select {
case <-dumpAddressTicker.C:
a.saveToFile(a.filePath)
case <-a.quit:
break out
}
}
dumpAddressTicker.Stop()
a.saveToFile(a.filePath)
a.wg.Done()
log.Trace("Address handler done")
for {
select {
case <-dumpAddressTicker.C:
a.saveToFile(a.filePath)
case <-a.quit:
break out
}
}
dumpAddressTicker.Stop()
a.saveToFile(a.filePath)
a.wg.Done()
log.Trace("Address handler done")
}
func (a *AddrBook) addAddress(addr, src *NetAddress) {
if !addr.Routable() { return }
if !addr.Routable() {
return
}
key := addr.String()
ka := a.addrIndex[key]
key := addr.String()
ka := a.addrIndex[key]
if ka != nil {
// Already added
if ka.OldBucket != -1 { return }
if ka.NewRefs == newBucketsPerAddress { return }
if ka != nil {
// Already added
if ka.OldBucket != -1 {
return
}
if ka.NewRefs == newBucketsPerAddress {
return
}
// The more entries we have, the less likely we are to add more.
factor := int32(2 * ka.NewRefs)
if a.rand.Int31n(factor) != 0 {
return
}
} else {
ka = NewKnownAddress(addr, src)
a.addrIndex[key] = ka
a.nNew++
}
// The more entries we have, the less likely we are to add more.
factor := int32(2 * ka.NewRefs)
if a.rand.Int31n(factor) != 0 {
return
}
} else {
ka = NewKnownAddress(addr, src)
a.addrIndex[key] = ka
a.nNew++
}
bucket := a.getNewBucket(addr, src)
bucket := a.getNewBucket(addr, src)
// Already exists?
if _, ok := a.addrNew[bucket][key]; ok {
return
}
// Already exists?
if _, ok := a.addrNew[bucket][key]; ok {
return
}
// Enforce max addresses.
if len(a.addrNew[bucket]) > newBucketSize {
log.Tracef("new bucket is full, expiring old ")
a.expireNew(bucket)
}
// Enforce max addresses.
if len(a.addrNew[bucket]) > newBucketSize {
log.Tracef("new bucket is full, expiring old ")
a.expireNew(bucket)
}
// Add to new bucket.
ka.NewRefs++
a.addrNew[bucket][key] = ka
// Add to new bucket.
ka.NewRefs++
a.addrNew[bucket][key] = ka
log.Tracef("Added new address %s for a total of %d addresses", addr, a.nOld+a.nNew)
log.Tracef("Added new address %s for a total of %d addresses", addr, a.nOld+a.nNew)
}
// Make space in the new buckets by expiring the really bad entries.
// If no bad entries are available we look at a few and remove the oldest.
func (a *AddrBook) expireNew(bucket int) {
var oldest *KnownAddress
for k, v := range a.addrNew[bucket] {
// If an entry is bad, throw it away
if v.Bad() {
log.Tracef("expiring bad address %v", k)
delete(a.addrNew[bucket], k)
v.NewRefs--
if v.NewRefs == 0 {
a.nNew--
delete(a.addrIndex, k)
}
return
}
// or, keep track of the oldest entry
if oldest == nil {
oldest = v
} else if v.LastAttempt.Before(oldest.LastAttempt.Time) {
oldest = v
}
}
var oldest *KnownAddress
for k, v := range a.addrNew[bucket] {
// If an entry is bad, throw it away
if v.Bad() {
log.Tracef("expiring bad address %v", k)
delete(a.addrNew[bucket], k)
v.NewRefs--
if v.NewRefs == 0 {
a.nNew--
delete(a.addrIndex, k)
}
return
}
// or, keep track of the oldest entry
if oldest == nil {
oldest = v
} else if v.LastAttempt.Before(oldest.LastAttempt.Time) {
oldest = v
}
}
// If we haven't thrown out a bad entry, throw out the oldest entry
if oldest != nil {
key := oldest.Addr.String()
log.Tracef("expiring oldest address %v", key)
delete(a.addrNew[bucket], key)
oldest.NewRefs--
if oldest.NewRefs == 0 {
a.nNew--
delete(a.addrIndex, key)
}
}
// If we haven't thrown out a bad entry, throw out the oldest entry
if oldest != nil {
key := oldest.Addr.String()
log.Tracef("expiring oldest address %v", key)
delete(a.addrNew[bucket], key)
oldest.NewRefs--
if oldest.NewRefs == 0 {
a.nNew--
delete(a.addrIndex, key)
}
}
}
func (a *AddrBook) moveToOld(ka *KnownAddress) {
// Remove from all new buckets.
// Remember one of those new buckets.
addrKey := ka.Addr.String()
freedBucket := -1
for i := range a.addrNew {
// we check for existance so we can record the first one
if _, ok := a.addrNew[i][addrKey]; ok {
delete(a.addrNew[i], addrKey)
ka.NewRefs--
if freedBucket == -1 {
freedBucket = i
}
}
}
a.nNew--
if freedBucket == -1 { panic("Expected to find addr in at least one new bucket") }
// Remove from all new buckets.
// Remember one of those new buckets.
addrKey := ka.Addr.String()
freedBucket := -1
for i := range a.addrNew {
// we check for existance so we can record the first one
if _, ok := a.addrNew[i][addrKey]; ok {
delete(a.addrNew[i], addrKey)
ka.NewRefs--
if freedBucket == -1 {
freedBucket = i
}
}
}
a.nNew--
if freedBucket == -1 {
panic("Expected to find addr in at least one new bucket")
}
oldBucket := a.getOldBucket(ka.Addr)
oldBucket := a.getOldBucket(ka.Addr)
// If room in oldBucket, put it in.
if len(a.addrOld[oldBucket]) < oldBucketSize {
ka.OldBucket = Int16(oldBucket)
a.addrOld[oldBucket] = append(a.addrOld[oldBucket], ka)
a.nOld++
return
}
// If room in oldBucket, put it in.
if len(a.addrOld[oldBucket]) < oldBucketSize {
ka.OldBucket = Int16(oldBucket)
a.addrOld[oldBucket] = append(a.addrOld[oldBucket], ka)
a.nOld++
return
}
// No room, we have to evict something else.
rmkaIndex := a.pickOld(oldBucket)
rmka := a.addrOld[oldBucket][rmkaIndex]
// No room, we have to evict something else.
rmkaIndex := a.pickOld(oldBucket)
rmka := a.addrOld[oldBucket][rmkaIndex]
// Find a new bucket to put rmka in.
newBucket := a.getNewBucket(rmka.Addr, rmka.Src)
if len(a.addrNew[newBucket]) >= newBucketSize {
newBucket = freedBucket
}
// Find a new bucket to put rmka in.
newBucket := a.getNewBucket(rmka.Addr, rmka.Src)
if len(a.addrNew[newBucket]) >= newBucketSize {
newBucket = freedBucket
}
// replace with ka in list.
ka.OldBucket = Int16(oldBucket)
a.addrOld[oldBucket][rmkaIndex] = ka
rmka.OldBucket = -1
// replace with ka in list.
ka.OldBucket = Int16(oldBucket)
a.addrOld[oldBucket][rmkaIndex] = ka
rmka.OldBucket = -1
// put rmka into new bucket
rmkey := rmka.Addr.String()
log.Tracef("Replacing %s with %s in old", rmkey, addrKey)
a.addrNew[newBucket][rmkey] = rmka
rmka.NewRefs++
a.nNew++
// put rmka into new bucket
rmkey := rmka.Addr.String()
log.Tracef("Replacing %s with %s in old", rmkey, addrKey)
a.addrNew[newBucket][rmkey] = rmka
rmka.NewRefs++
a.nNew++
}
// Returns the index in old bucket of oldest entry.
func (a *AddrBook) pickOld(bucket int) int {
var oldest *KnownAddress
var oldestIndex int
for i, ka := range a.addrOld[bucket] {
if oldest == nil || ka.LastAttempt.Before(oldest.LastAttempt.Time) {
oldest = ka
oldestIndex = i
}
}
return oldestIndex
var oldest *KnownAddress
var oldestIndex int
for i, ka := range a.addrOld[bucket] {
if oldest == nil || ka.LastAttempt.Before(oldest.LastAttempt.Time) {
oldest = ka
oldestIndex = i
}
}
return oldestIndex
}
// doublesha256(key + sourcegroup +
// int64(doublesha256(key + group + sourcegroup))%bucket_per_source_group) % num_new_buckes
func (a *AddrBook) getNewBucket(addr, src *NetAddress) int {
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(GroupKey(addr))...)
data1 = append(data1, []byte(GroupKey(src))...)
hash1 := DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= newBucketsPerGroup
var hashbuf [8]byte
binary.LittleEndian.PutUint64(hashbuf[:], hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(src)...)
data2 = append(data2, hashbuf[:]...)
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(GroupKey(addr))...)
data1 = append(data1, []byte(GroupKey(src))...)
hash1 := DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= newBucketsPerGroup
var hashbuf [8]byte
binary.LittleEndian.PutUint64(hashbuf[:], hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(src)...)
data2 = append(data2, hashbuf[:]...)
hash2 := DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % newBucketCount)
hash2 := DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % newBucketCount)
}
// doublesha256(key + group + truncate_to_64bits(doublesha256(key + addr))%buckets_per_group) % num_buckets
func (a *AddrBook) getOldBucket(addr *NetAddress) int {
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(addr.String())...)
hash1 := DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= oldBucketsPerGroup
var hashbuf [8]byte
binary.LittleEndian.PutUint64(hashbuf[:], hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(addr)...)
data2 = append(data2, hashbuf[:]...)
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(addr.String())...)
hash1 := DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= oldBucketsPerGroup
var hashbuf [8]byte
binary.LittleEndian.PutUint64(hashbuf[:], hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(addr)...)
data2 = append(data2, hashbuf[:]...)
hash2 := DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % oldBucketCount)
hash2 := DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % oldBucketCount)
}
// Return a string representing the network group of this address.
// This is the /16 for IPv6, the /32 (/36 for he.net) for IPv6, the string
// "local" for a local address and the string "unroutable for an unroutable
// address.
func GroupKey (na *NetAddress) string {
if na.Local() {
return "local"
}
if !na.Routable() {
return "unroutable"
}
func GroupKey(na *NetAddress) string {
if na.Local() {
return "local"
}
if !na.Routable() {
return "unroutable"
}
if ipv4 := na.IP.To4(); ipv4 != nil {
return (&net.IPNet{IP: na.IP, Mask: net.CIDRMask(16, 32)}).String()
}
if na.RFC6145() || na.RFC6052() {
// last four bytes are the ip address
ip := net.IP(na.IP[12:16])
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
}
if ipv4 := na.IP.To4(); ipv4 != nil {
return (&net.IPNet{IP: na.IP, Mask: net.CIDRMask(16, 32)}).String()
}
if na.RFC6145() || na.RFC6052() {
// last four bytes are the ip address
ip := net.IP(na.IP[12:16])
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
}
if na.RFC3964() {
ip := net.IP(na.IP[2:7])
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
if na.RFC3964() {
ip := net.IP(na.IP[2:7])
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
}
if na.RFC4380() {
// teredo tunnels have the last 4 bytes as the v4 address XOR
// 0xff.
ip := net.IP(make([]byte, 4))
for i, byte := range na.IP[12:16] {
ip[i] = byte ^ 0xff
}
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
}
}
if na.RFC4380() {
// teredo tunnels have the last 4 bytes as the v4 address XOR
// 0xff.
ip := net.IP(make([]byte, 4))
for i, byte := range na.IP[12:16] {
ip[i] = byte ^ 0xff
}
return (&net.IPNet{IP: ip, Mask: net.CIDRMask(16, 32)}).String()
}
// OK, so now we know ourselves to be a IPv6 address.
// bitcoind uses /32 for everything, except for Hurricane Electric's
// (he.net) IP range, which it uses /36 for.
bits := 32
heNet := &net.IPNet{IP: net.ParseIP("2001:470::"),
Mask: net.CIDRMask(32, 128)}
if heNet.Contains(na.IP) {
bits = 36
}
// OK, so now we know ourselves to be a IPv6 address.
// bitcoind uses /32 for everything, except for Hurricane Electric's
// (he.net) IP range, which it uses /36 for.
bits := 32
heNet := &net.IPNet{IP: net.ParseIP("2001:470::"),
Mask: net.CIDRMask(32, 128)}
if heNet.Contains(na.IP) {
bits = 36
}
return (&net.IPNet{IP: na.IP, Mask: net.CIDRMask(bits, 128)}).String()
return (&net.IPNet{IP: na.IP, Mask: net.CIDRMask(bits, 128)}).String()
}
peer/client.go
+122 -108
View File
@@ -1,12 +1,12 @@
package peer
import (
. "github.com/tendermint/tendermint/common"
. "github.com/tendermint/tendermint/binary"
"github.com/tendermint/tendermint/merkle"
"sync/atomic"
"sync"
"errors"
"errors"
. "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
"github.com/tendermint/tendermint/merkle"
"sync"
"sync/atomic"
)
/* Client
@@ -21,147 +21,161 @@ import (
XXX what about peer disconnects?
*/
type Client struct {
addrBook *AddrBook
targetNumPeers int
makePeerFn func(*Connection) *Peer
self *Peer
recvQueues map[String]chan *InboundPacket
addrBook *AddrBook
targetNumPeers int
makePeerFn func(*Connection) *Peer
self *Peer
recvQueues map[String]chan *InboundPacket
mtx sync.Mutex
peers merkle.Tree // addr -> *Peer
quit chan struct{}
stopped uint32
mtx sync.Mutex
peers merkle.Tree // addr -> *Peer
quit chan struct{}
stopped uint32
}
var (
CLIENT_STOPPED_ERROR = errors.New("Client already stopped")
CLIENT_DUPLICATE_PEER_ERROR = errors.New("Duplicate peer")
CLIENT_STOPPED_ERROR = errors.New("Client already stopped")
CLIENT_DUPLICATE_PEER_ERROR = errors.New("Duplicate peer")
)
func NewClient(makePeerFn func(*Connection) *Peer) *Client {
self := makePeerFn(nil)
if self == nil {
Panicf("makePeerFn(nil) must return a prototypical peer for self")
}
self := makePeerFn(nil)
if self == nil {
Panicf("makePeerFn(nil) must return a prototypical peer for self")
}
recvQueues := make(map[String]chan *InboundPacket)
for chName, _ := range self.channels {
recvQueues[chName] = make(chan *InboundPacket)
}
recvQueues := make(map[String]chan *InboundPacket)
for chName, _ := range self.channels {
recvQueues[chName] = make(chan *InboundPacket)
}
c := &Client{
addrBook: nil, // TODO
targetNumPeers: 0, // TODO
makePeerFn: makePeerFn,
self: self,
recvQueues: recvQueues,
c := &Client{
addrBook: nil, // TODO
targetNumPeers: 0, // TODO
makePeerFn: makePeerFn,
self: self,
recvQueues: recvQueues,
peers: merkle.NewIAVLTree(nil),
quit: make(chan struct{}),
stopped: 0,
}
return c
peers: merkle.NewIAVLTree(nil),
quit: make(chan struct{}),
stopped: 0,
}
return c
}
func (c *Client) Stop() {
log.Infof("Stopping client")
// lock
c.mtx.Lock()
if atomic.CompareAndSwapUint32(&c.stopped, 0, 1) {
close(c.quit)
// stop each peer.
for peerValue := range c.peers.Values() {
peer := peerValue.(*Peer)
peer.Stop()
}
// empty tree.
c.peers = merkle.NewIAVLTree(nil)
}
c.mtx.Unlock()
// unlock
log.Infof("Stopping client")
// lock
c.mtx.Lock()
if atomic.CompareAndSwapUint32(&c.stopped, 0, 1) {
close(c.quit)
// stop each peer.
for peerValue := range c.peers.Values() {
peer := peerValue.(*Peer)
peer.Stop()
}
// empty tree.
c.peers = merkle.NewIAVLTree(nil)
}
c.mtx.Unlock()
// unlock
}
func (c *Client) AddPeerWithConnection(conn *Connection, outgoing bool) (*Peer, error) {
if atomic.LoadUint32(&c.stopped) == 1 { return nil, CLIENT_STOPPED_ERROR }
if atomic.LoadUint32(&c.stopped) == 1 {
return nil, CLIENT_STOPPED_ERROR
}
log.Infof("Adding peer with connection: %v, outgoing: %v", conn, outgoing)
peer := c.makePeerFn(conn)
peer.outgoing = outgoing
err := c.addPeer(peer)
if err != nil { return nil, err }
log.Infof("Adding peer with connection: %v, outgoing: %v", conn, outgoing)
peer := c.makePeerFn(conn)
peer.outgoing = outgoing
err := c.addPeer(peer)
if err != nil {
return nil, err
}
go peer.Start(c.recvQueues)
go peer.Start(c.recvQueues)
return peer, nil
return peer, nil
}
func (c *Client) Broadcast(pkt Packet) {
if atomic.LoadUint32(&c.stopped) == 1 { return }
if atomic.LoadUint32(&c.stopped) == 1 {
return
}
log.Tracef("Broadcast on [%v] len: %v", pkt.Channel, len(pkt.Bytes))
for v := range c.Peers().Values() {
peer := v.(*Peer)
success := peer.TrySend(pkt)
log.Tracef("Broadcast for peer %v success: %v", peer, success)
if !success {
// TODO: notify the peer
}
}
log.Tracef("Broadcast on [%v] len: %v", pkt.Channel, len(pkt.Bytes))
for v := range c.Peers().Values() {
peer := v.(*Peer)
success := peer.TrySend(pkt)
log.Tracef("Broadcast for peer %v success: %v", peer, success)
if !success {
// TODO: notify the peer
}
}
}
// blocks until a message is popped.
func (c *Client) Receive(chName String) *InboundPacket {
if atomic.LoadUint32(&c.stopped) == 1 { return nil }
if atomic.LoadUint32(&c.stopped) == 1 {
return nil
}
log.Tracef("Receive on [%v]", chName)
q := c.recvQueues[chName]
if q == nil { Panicf("Expected recvQueues[%f], found none", chName) }
log.Tracef("Receive on [%v]", chName)
q := c.recvQueues[chName]
if q == nil {
Panicf("Expected recvQueues[%f], found none", chName)
}
for {
select {
case <-c.quit:
return nil
case inPacket := <-q:
return inPacket
}
}
for {
select {
case <-c.quit:
return nil
case inPacket := <-q:
return inPacket
}
}
}
func (c *Client) Peers() merkle.Tree {
// lock & defer
c.mtx.Lock(); defer c.mtx.Unlock()
return c.peers.Copy()
// unlock deferred
// lock & defer
c.mtx.Lock()
defer c.mtx.Unlock()
return c.peers.Copy()
// unlock deferred
}
func (c *Client) StopPeer(peer *Peer) {
// lock
c.mtx.Lock()
peerValue, _ := c.peers.Remove(peer.RemoteAddress())
c.mtx.Unlock()
// unlock
// lock
c.mtx.Lock()
peerValue, _ := c.peers.Remove(peer.RemoteAddress())
c.mtx.Unlock()
// unlock
peer_ := peerValue.(*Peer)
if peer_ != nil {
peer_.Stop()
}
peer_ := peerValue.(*Peer)
if peer_ != nil {
peer_.Stop()
}
}
func (c *Client) addPeer(peer *Peer) error {
addr := peer.RemoteAddress()
addr := peer.RemoteAddress()
// lock & defer
c.mtx.Lock(); defer c.mtx.Unlock()
if c.stopped == 1 { return CLIENT_STOPPED_ERROR }
if !c.peers.Has(addr) {
log.Tracef("Actually putting addr: %v, peer: %v", addr, peer)
c.peers.Put(addr, peer)
return nil
} else {
// ignore duplicate peer for addr.
log.Infof("Ignoring duplicate peer for addr %v", addr)
return CLIENT_DUPLICATE_PEER_ERROR
}
// unlock deferred
// lock & defer
c.mtx.Lock()
defer c.mtx.Unlock()
if c.stopped == 1 {
return CLIENT_STOPPED_ERROR
}
if !c.peers.Has(addr) {
log.Tracef("Actually putting addr: %v, peer: %v", addr, peer)
c.peers.Put(addr, peer)
return nil
} else {
// ignore duplicate peer for addr.
log.Infof("Ignoring duplicate peer for addr %v", addr)
return CLIENT_DUPLICATE_PEER_ERROR
}
// unlock deferred
}
peer/client_test.go
+69 -70
View File
@@ -1,106 +1,105 @@
package peer
import (
. "github.com/tendermint/tendermint/binary"
"testing"
"time"
. "github.com/tendermint/tendermint/binary"
"testing"
"time"
)
// convenience method for creating two clients connected to each other.
func makeClientPair(t *testing.T, bufferSize int, channels []string) (*Client, *Client) {
peerMaker := func(conn *Connection) *Peer {
p := NewPeer(conn)
p.channels = map[String]*Channel{}
for chName := range channels {
p.channels[String(chName)] = NewChannel(String(chName), bufferSize)
}
return p
}
peerMaker := func(conn *Connection) *Peer {
p := NewPeer(conn)
p.channels = map[String]*Channel{}
for chName := range channels {
p.channels[String(chName)] = NewChannel(String(chName), bufferSize)
}
return p
}
// Create two clients that will be interconnected.
c1 := NewClient(peerMaker)
c2 := NewClient(peerMaker)
// Create two clients that will be interconnected.
c1 := NewClient(peerMaker)
c2 := NewClient(peerMaker)
// Create a server for the listening client.
s1 := NewServer("tcp", ":8001", c1)
// Create a server for the listening client.
s1 := NewServer("tcp", ":8001", c1)
// Dial the server & add the connection to c2.
s1laddr := s1.LocalAddress()
conn, err := s1laddr.Dial()
if err != nil {
t.Fatalf("Could not connect to server address %v", s1laddr)
} else {
t.Logf("Created a connection to local server address %v", s1laddr)
}
// Dial the server & add the connection to c2.
s1laddr := s1.LocalAddress()
conn, err := s1laddr.Dial()
if err != nil {
t.Fatalf("Could not connect to server address %v", s1laddr)
} else {
t.Logf("Created a connection to local server address %v", s1laddr)
}
c2.AddPeerWithConnection(conn, true)
c2.AddPeerWithConnection(conn, true)
// Wait for things to happen, peers to get added...
time.Sleep(100 * time.Millisecond)
// Wait for things to happen, peers to get added...
time.Sleep(100 * time.Millisecond)
return c1, c2
return c1, c2
}
func TestClients(t *testing.T) {
c1, c2 := makeClientPair(t, 10, []string{"ch1", "ch2", "ch3"})
c1, c2 := makeClientPair(t, 10, []string{"ch1", "ch2", "ch3"})
// Lets send a message from c1 to c2.
if c1.Peers().Size() != 1 {
t.Errorf("Expected exactly 1 peer in c1, got %v", c1.Peers().Size())
}
if c2.Peers().Size() != 1 {
t.Errorf("Expected exactly 1 peer in c2, got %v", c2.Peers().Size())
}
// Lets send a message from c1 to c2.
if c1.Peers().Size() != 1 {
t.Errorf("Expected exactly 1 peer in c1, got %v", c1.Peers().Size())
}
if c2.Peers().Size() != 1 {
t.Errorf("Expected exactly 1 peer in c2, got %v", c2.Peers().Size())
}
// Broadcast a message on ch1
c1.Broadcast(NewPacket("ch1", ByteSlice("channel one")))
// Broadcast a message on ch2
c1.Broadcast(NewPacket("ch2", ByteSlice("channel two")))
// Broadcast a message on ch3
c1.Broadcast(NewPacket("ch3", ByteSlice("channel three")))
// Broadcast a message on ch1
c1.Broadcast(NewPacket("ch1", ByteSlice("channel one")))
// Broadcast a message on ch2
c1.Broadcast(NewPacket("ch2", ByteSlice("channel two")))
// Broadcast a message on ch3
c1.Broadcast(NewPacket("ch3", ByteSlice("channel three")))
// Wait for things to settle...
time.Sleep(100 * time.Millisecond)
// Wait for things to settle...
time.Sleep(100 * time.Millisecond)
// Receive message from channel 2 and check
inMsg := c2.Receive("ch2")
if string(inMsg.Bytes) != "channel two" {
t.Errorf("Unexpected received message bytes: %v", string(inMsg.Bytes))
}
// Receive message from channel 2 and check
inMsg := c2.Receive("ch2")
if string(inMsg.Bytes) != "channel two" {
t.Errorf("Unexpected received message bytes: %v", string(inMsg.Bytes))
}
// Receive message from channel 1 and check
inMsg = c2.Receive("ch1")
if string(inMsg.Bytes) != "channel one" {
t.Errorf("Unexpected received message bytes: %v", string(inMsg.Bytes))
}
// Receive message from channel 1 and check
inMsg = c2.Receive("ch1")
if string(inMsg.Bytes) != "channel one" {
t.Errorf("Unexpected received message bytes: %v", string(inMsg.Bytes))
}
s1.Stop()
c2.Stop()
s1.Stop()
c2.Stop()
}
func BenchmarkClients(b *testing.B) {
b.StopTimer()
b.StopTimer()
// TODO: benchmark the random functions, which is faster?
// TODO: benchmark the random functions, which is faster?
c1, c2 := makeClientPair(t, 10, []string{"ch1", "ch2", "ch3"})
c1, c2 := makeClientPair(t, 10, []string{"ch1", "ch2", "ch3"})
// Create a sink on either channel to just pop off messages.
// TODO: ensure that when clients stop, this goroutine stops.
func recvHandler(c *Client) {
}
// Create a sink on either channel to just pop off messages.
// TODO: ensure that when clients stop, this goroutine stops.
recvHandler := func(c *Client) {
}
go recvHandler(c1)
go recvHandler(c2)
go recvHandler(c1)
go recvHandler(c2)
b.StartTimer()
b.StartTimer()
// Send random message from one channel to another
for i := 0; i < b.N; i++ {
}
// Send random message from one channel to another
for i := 0; i < b.N; i++ {
}
}
peer/connection.go
+130 -129
View File
@@ -1,191 +1,192 @@
package peer
import (
. "github.com/tendermint/tendermint/common"
. "github.com/tendermint/tendermint/binary"
"sync/atomic"
"net"
"time"
"fmt"
"fmt"
. "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
"net"
"sync/atomic"
"time"
)
const (
OUT_QUEUE_SIZE = 50
IDLE_TIMEOUT_MINUTES = 5
PING_TIMEOUT_MINUTES = 2
OUT_QUEUE_SIZE = 50
IDLE_TIMEOUT_MINUTES = 5
PING_TIMEOUT_MINUTES = 2
)
/* Connnection */
type Connection struct {
ioStats IOStats
ioStats IOStats
sendQueue chan Packet // never closes
conn net.Conn
quit chan struct{}
stopped uint32
pingDebouncer *Debouncer
pong chan struct{}
sendQueue chan Packet // never closes
conn net.Conn
quit chan struct{}
stopped uint32
pingDebouncer *Debouncer
pong chan struct{}
}
var (
PACKET_TYPE_PING = UInt8(0x00)
PACKET_TYPE_PONG = UInt8(0x01)
PACKET_TYPE_MSG = UInt8(0x10)
PACKET_TYPE_PING = UInt8(0x00)
PACKET_TYPE_PONG = UInt8(0x01)
PACKET_TYPE_MSG = UInt8(0x10)
)
func NewConnection(conn net.Conn) *Connection {
return &Connection{
sendQueue: make(chan Packet, OUT_QUEUE_SIZE),
conn: conn,
quit: make(chan struct{}),
pingDebouncer: NewDebouncer(PING_TIMEOUT_MINUTES * time.Minute),
pong: make(chan struct{}),
}
return &Connection{
sendQueue: make(chan Packet, OUT_QUEUE_SIZE),
conn: conn,
quit: make(chan struct{}),
pingDebouncer: NewDebouncer(PING_TIMEOUT_MINUTES * time.Minute),
pong: make(chan struct{}),
}
}
// returns true if successfully queued,
// returns false if connection was closed.
// blocks.
func (c *Connection) Send(pkt Packet) bool {
select {
case c.sendQueue <- pkt:
return true
case <-c.quit:
return false
}
select {
case c.sendQueue <- pkt:
return true
case <-c.quit:
return false
}
}
func (c *Connection) Start(channels map[String]*Channel) {
log.Debugf("Starting %v", c)
go c.sendHandler()
go c.recvHandler(channels)
log.Debugf("Starting %v", c)
go c.sendHandler()
go c.recvHandler(channels)
}
func (c *Connection) Stop() {
if atomic.CompareAndSwapUint32(&c.stopped, 0, 1) {
log.Debugf("Stopping %v", c)
close(c.quit)
c.conn.Close()
c.pingDebouncer.Stop()
// We can't close pong safely here because
// recvHandler may write to it after we've stopped.
// Though it doesn't need to get closed at all,
// we close it @ recvHandler.
// close(c.pong)
}
if atomic.CompareAndSwapUint32(&c.stopped, 0, 1) {
log.Debugf("Stopping %v", c)
close(c.quit)
c.conn.Close()
c.pingDebouncer.Stop()
// We can't close pong safely here because
// recvHandler may write to it after we've stopped.
// Though it doesn't need to get closed at all,
// we close it @ recvHandler.
// close(c.pong)
}
}
func (c *Connection) LocalAddress() *NetAddress {
return NewNetAddress(c.conn.LocalAddr())
return NewNetAddress(c.conn.LocalAddr())
}
func (c *Connection) RemoteAddress() *NetAddress {
return NewNetAddress(c.conn.RemoteAddr())
return NewNetAddress(c.conn.RemoteAddr())
}
func (c *Connection) String() string {
return fmt.Sprintf("Connection{%v}", c.conn.RemoteAddr())
return fmt.Sprintf("Connection{%v}", c.conn.RemoteAddr())
}
func (c *Connection) flush() {
// TODO flush? (turn off nagel, turn back on, etc)
// TODO flush? (turn off nagel, turn back on, etc)
}
func (c *Connection) sendHandler() {
log.Tracef("%v sendHandler", c)
log.Tracef("%v sendHandler", c)
// TODO: catch panics & stop connection.
// TODO: catch panics & stop connection.
FOR_LOOP:
for {
var err error
select {
case <-c.pingDebouncer.Ch:
_, err = PACKET_TYPE_PING.WriteTo(c.conn)
case sendPkt := <-c.sendQueue:
log.Tracef("Found pkt from sendQueue. Writing pkt to underlying connection")
_, err = PACKET_TYPE_MSG.WriteTo(c.conn)
if err != nil { break }
_, err = sendPkt.WriteTo(c.conn)
case <-c.pong:
_, err = PACKET_TYPE_PONG.WriteTo(c.conn)
case <-c.quit:
break FOR_LOOP
}
FOR_LOOP:
for {
var err error
select {
case <-c.pingDebouncer.Ch:
_, err = PACKET_TYPE_PING.WriteTo(c.conn)
case sendPkt := <-c.sendQueue:
log.Tracef("Found pkt from sendQueue. Writing pkt to underlying connection")
_, err = PACKET_TYPE_MSG.WriteTo(c.conn)
if err != nil {
break
}
_, err = sendPkt.WriteTo(c.conn)
case <-c.pong:
_, err = PACKET_TYPE_PONG.WriteTo(c.conn)
case <-c.quit:
break FOR_LOOP
}
if err != nil {
log.Infof("%v failed @ sendHandler:\n%v", c, err)
c.Stop()
break FOR_LOOP
}
if err != nil {
log.Infof("%v failed @ sendHandler:\n%v", c, err)
c.Stop()
break FOR_LOOP
}
c.flush()
}
c.flush()
}
log.Tracef("%v sendHandler done", c)
// cleanup
log.Tracef("%v sendHandler done", c)
// cleanup
}
func (c *Connection) recvHandler(channels map[String]*Channel) {
log.Tracef("%v recvHandler with %v channels", c, len(channels))
log.Tracef("%v recvHandler with %v channels", c, len(channels))
// TODO: catch panics & stop connection.
// TODO: catch panics & stop connection.
FOR_LOOP:
for {
pktType, err := ReadUInt8Safe(c.conn)
if err != nil {
if atomic.LoadUint32(&c.stopped) != 1 {
log.Infof("%v failed @ recvHandler", c)
c.Stop()
}
break FOR_LOOP
} else {
log.Tracef("Found pktType %v", pktType)
}
FOR_LOOP:
for {
pktType, err := ReadUInt8Safe(c.conn)
if err != nil {
if atomic.LoadUint32(&c.stopped) != 1 {
log.Infof("%v failed @ recvHandler", c)
c.Stop()
}
break FOR_LOOP
} else {
log.Tracef("Found pktType %v", pktType)
}
switch pktType {
case PACKET_TYPE_PING:
c.pong <- struct{}{}
case PACKET_TYPE_PONG:
// do nothing
case PACKET_TYPE_MSG:
pkt, err := ReadPacketSafe(c.conn)
if err != nil {
if atomic.LoadUint32(&c.stopped) != 1 {
log.Infof("%v failed @ recvHandler", c)
c.Stop()
}
break FOR_LOOP
}
channel := channels[pkt.Channel]
if channel == nil {
Panicf("Unknown channel %v", pkt.Channel)
}
channel.recvQueue <- pkt
default:
Panicf("Unknown message type %v", pktType)
}
switch pktType {
case PACKET_TYPE_PING:
c.pong <- struct{}{}
case PACKET_TYPE_PONG:
// do nothing
case PACKET_TYPE_MSG:
pkt, err := ReadPacketSafe(c.conn)
if err != nil {
if atomic.LoadUint32(&c.stopped) != 1 {
log.Infof("%v failed @ recvHandler", c)
c.Stop()
}
break FOR_LOOP
}
channel := channels[pkt.Channel]
if channel == nil {
Panicf("Unknown channel %v", pkt.Channel)
}
channel.recvQueue <- pkt
default:
Panicf("Unknown message type %v", pktType)
}
c.pingDebouncer.Reset()
}
c.pingDebouncer.Reset()
}
log.Tracef("%v recvHandler done", c)
// cleanup
close(c.pong)
for _ = range c.pong {
// drain
}
log.Tracef("%v recvHandler done", c)
// cleanup
close(c.pong)
for _ = range c.pong {
// drain
}
}
/* IOStats */
type IOStats struct {
TimeConnected Time
LastSent Time
LastRecv Time
BytesRecv UInt64
BytesSent UInt64
PktsRecv UInt64
PktsSent UInt64
TimeConnected Time
LastSent Time
LastRecv Time
BytesRecv UInt64
BytesSent UInt64
PktsRecv UInt64
PktsSent UInt64
}
peer/knownaddress.go
+71 -71
View File
@@ -1,104 +1,104 @@
package peer
import (
. "github.com/tendermint/tendermint/binary"
"time"
"io"
. "github.com/tendermint/tendermint/binary"
"io"
"time"
)
/*
KnownAddress
KnownAddress
tracks information about a known network address that is used
to determine how viable an address is.
tracks information about a known network address that is used
to determine how viable an address is.
*/
type KnownAddress struct {
Addr *NetAddress
Src *NetAddress
Attempts UInt32
LastAttempt Time
LastSuccess Time
NewRefs UInt16
OldBucket Int16 // TODO init to -1
Addr *NetAddress
Src *NetAddress
Attempts UInt32
LastAttempt Time
LastSuccess Time
NewRefs UInt16
OldBucket Int16 // TODO init to -1
}
func NewKnownAddress(addr *NetAddress, src *NetAddress) *KnownAddress {
return &KnownAddress{
Addr: addr,
Src: src,
OldBucket: -1,
LastAttempt: Time{time.Now()},
Attempts: 0,
}
return &KnownAddress{
Addr: addr,
Src: src,
OldBucket: -1,
LastAttempt: Time{time.Now()},
Attempts: 0,
}
}
func ReadKnownAddress(r io.Reader) *KnownAddress {
return &KnownAddress{
Addr: ReadNetAddress(r),
Src: ReadNetAddress(r),
Attempts: ReadUInt32(r),
LastAttempt: ReadTime(r),
LastSuccess: ReadTime(r),
NewRefs: ReadUInt16(r),
OldBucket: ReadInt16(r),
}
return &KnownAddress{
Addr: ReadNetAddress(r),
Src: ReadNetAddress(r),
Attempts: ReadUInt32(r),
LastAttempt: ReadTime(r),
LastSuccess: ReadTime(r),
NewRefs: ReadUInt16(r),
OldBucket: ReadInt16(r),
}
}
func (ka *KnownAddress) WriteTo(w io.Writer) (n int64, err error) {
n, err = WriteOnto(ka.Addr, w, n, err)
n, err = WriteOnto(ka.Src, w, n, err)
n, err = WriteOnto(ka.Attempts, w, n, err)
n, err = WriteOnto(ka.LastAttempt, w, n, err)
n, err = WriteOnto(ka.LastSuccess, w, n, err)
n, err = WriteOnto(ka.NewRefs, w, n, err)
n, err = WriteOnto(ka.OldBucket, w, n, err)
return
n, err = WriteOnto(ka.Addr, w, n, err)
n, err = WriteOnto(ka.Src, w, n, err)
n, err = WriteOnto(ka.Attempts, w, n, err)
n, err = WriteOnto(ka.LastAttempt, w, n, err)
n, err = WriteOnto(ka.LastSuccess, w, n, err)
n, err = WriteOnto(ka.NewRefs, w, n, err)
n, err = WriteOnto(ka.OldBucket, w, n, err)
return
}
func (ka *KnownAddress) MarkAttempt(success bool) {
now := Time{time.Now()}
ka.LastAttempt = now
if success {
ka.LastSuccess = now
ka.Attempts = 0
} else {
ka.Attempts += 1
}
now := Time{time.Now()}
ka.LastAttempt = now
if success {
ka.LastSuccess = now
ka.Attempts = 0
} else {
ka.Attempts += 1
}
}
/*
An address is bad if the address in question has not been tried in the last
minute and meets one of the following criteria:
An address is bad if the address in question has not been tried in the last
minute and meets one of the following criteria:
1) It claims to be from the future
2) It hasn't been seen in over a month
3) It has failed at least three times and never succeeded
4) It has failed ten times in the last week
1) It claims to be from the future
2) It hasn't been seen in over a month
3) It has failed at least three times and never succeeded
4) It has failed ten times in the last week
All addresses that meet these criteria are assumed to be worthless and not
worth keeping hold of.
All addresses that meet these criteria are assumed to be worthless and not
worth keeping hold of.
*/
func (ka *KnownAddress) Bad() bool {
// Has been attempted in the last minute --> good
if ka.LastAttempt.Before(time.Now().Add(-1 * time.Minute)) {
return false
}
// Has been attempted in the last minute --> good
if ka.LastAttempt.Before(time.Now().Add(-1 * time.Minute)) {
return false
}
// Over a month old?
if ka.LastAttempt.After(time.Now().Add(-1 * numMissingDays * time.Hour * 24)) {
return true
}
// Over a month old?
if ka.LastAttempt.After(time.Now().Add(-1 * numMissingDays * time.Hour * 24)) {
return true
}
// Never succeeded?
if ka.LastSuccess.IsZero() && ka.Attempts >= numRetries {
return true
}
// Never succeeded?
if ka.LastSuccess.IsZero() && ka.Attempts >= numRetries {
return true
}
// Hasn't succeeded in too long?
if ka.LastSuccess.Before(time.Now().Add(-1*minBadDays*time.Hour*24)) &&
ka.Attempts >= maxFailures {
return true
}
// Hasn't succeeded in too long?
if ka.LastSuccess.Before(time.Now().Add(-1*minBadDays*time.Hour*24)) &&
ka.Attempts >= maxFailures {
return true
}
return false
return false
}
peer/listener.go
+82 -64
View File
@@ -1,127 +1,145 @@
package peer
import (
. "github.com/tendermint/tendermint/common"
"sync/atomic"
"net"
. "github.com/tendermint/tendermint/common"
"net"
"sync/atomic"
)
const (
DEFAULT_PORT = 8001
DEFAULT_PORT = 8001
)
/* Listener */
type Listener interface {
Connections() <-chan *Connection
LocalAddress() *NetAddress
Stop()
Connections() <-chan *Connection
LocalAddress() *NetAddress
Stop()
}
/* DefaultListener */
type DefaultListener struct {
listener net.Listener
connections chan *Connection
stopped uint32
listener net.Listener
connections chan *Connection
stopped uint32
}
const (
DEFAULT_BUFFERED_CONNECTIONS = 10
DEFAULT_BUFFERED_CONNECTIONS = 10
)
func NewDefaultListener(protocol string, listenAddr string) Listener {
listener, err := net.Listen(protocol, listenAddr)
if err != nil { panic(err) }
listener, err := net.Listen(protocol, listenAddr)
if err != nil {
panic(err)
}
dl := &DefaultListener{
listener: listener,
connections: make(chan *Connection, DEFAULT_BUFFERED_CONNECTIONS),
}
dl := &DefaultListener{
listener: listener,
connections: make(chan *Connection, DEFAULT_BUFFERED_CONNECTIONS),
}
go dl.listenHandler()
go dl.listenHandler()
return dl
return dl
}
func (l *DefaultListener) listenHandler() {
for {
conn, err := l.listener.Accept()
for {
conn, err := l.listener.Accept()
if atomic.LoadUint32(&l.stopped) == 1 { return }
if atomic.LoadUint32(&l.stopped) == 1 {
return
}
// listener wasn't stopped,
// yet we encountered an error.
if err != nil { panic(err) }
// listener wasn't stopped,
// yet we encountered an error.
if err != nil {
panic(err)
}
c := NewConnection(conn)
l.connections <- c
}
c := NewConnection(conn)
l.connections <- c
}
// cleanup
close(l.connections)
for _ = range l.connections {
// drain
}
// cleanup
close(l.connections)
for _ = range l.connections {
// drain
}
}
func (l *DefaultListener) Connections() <-chan *Connection {
return l.connections
return l.connections
}
func (l *DefaultListener) LocalAddress() *NetAddress {
return GetLocalAddress()
return GetLocalAddress()
}
func (l *DefaultListener) Stop() {
if atomic.CompareAndSwapUint32(&l.stopped, 0, 1) {
l.listener.Close()
}
if atomic.CompareAndSwapUint32(&l.stopped, 0, 1) {
l.listener.Close()
}
}
/* local address helpers */
func GetLocalAddress() *NetAddress {
laddr := GetUPNPLocalAddress()
if laddr != nil { return laddr }
laddr := GetUPNPLocalAddress()
if laddr != nil {
return laddr
}
laddr = GetDefaultLocalAddress()
if laddr != nil { return laddr }
laddr = GetDefaultLocalAddress()
if laddr != nil {
return laddr
}
panic("Could not determine local address")
panic("Could not determine local address")
}
// UPNP external address discovery & port mapping
// TODO: more flexible internal & external ports
func GetUPNPLocalAddress() *NetAddress {
nat, err := Discover()
if err != nil { return nil }
nat, err := Discover()
if err != nil {
return nil
}
ext, err := nat.GetExternalAddress()
if err != nil { return nil }
ext, err := nat.GetExternalAddress()
if err != nil {
return nil
}
_, err = nat.AddPortMapping("tcp", DEFAULT_PORT, DEFAULT_PORT, "tendermint", 0)
if err != nil { return nil }
_, err = nat.AddPortMapping("tcp", DEFAULT_PORT, DEFAULT_PORT, "tendermint", 0)
if err != nil {
return nil
}
return NewNetAddressIPPort(ext, DEFAULT_PORT)
return NewNetAddressIPPort(ext, DEFAULT_PORT)
}
// Naive local IPv4 interface address detection
// TODO: use syscalls to get actual ourIP. http://pastebin.com/9exZG4rh
func GetDefaultLocalAddress() *NetAddress {
addrs, err := net.InterfaceAddrs()
if err != nil { Panicf("Unexpected error fetching interface addresses: %v", err) }
addrs, err := net.InterfaceAddrs()
if err != nil {
Panicf("Unexpected error fetching interface addresses: %v", err)
}
for _, a := range addrs {
ipnet, ok := a.(*net.IPNet)
if !ok { continue }
v4 := ipnet.IP.To4()
if v4 == nil || v4[0] == 127 { continue } // loopback
return NewNetAddressIPPort(ipnet.IP, DEFAULT_PORT)
}
return nil
for _, a := range addrs {
ipnet, ok := a.(*net.IPNet)
if !ok {
continue
}
v4 := ipnet.IP.To4()
if v4 == nil || v4[0] == 127 {
continue
} // loopback
return NewNetAddressIPPort(ipnet.IP, DEFAULT_PORT)
}
return nil
}
peer/log.go
+8 -6
View File
@@ -1,14 +1,14 @@
package peer
import (
"github.com/cihub/seelog"
"github.com/cihub/seelog"
)
var log seelog.LoggerInterface
func init() {
// TODO: replace with configuration file in the ~/.tendermint directory.
config := `
// TODO: replace with configuration file in the ~/.tendermint directory.
config := `
<seelog type="sync">
<outputs formatid="colored">
<console/>
@@ -19,7 +19,9 @@ func init() {
</formats>
</seelog>`
var err error
log, err = seelog.LoggerFromConfigAsBytes([]byte(config))
if err != nil { panic(err) }
var err error
log, err = seelog.LoggerFromConfigAsBytes([]byte(config))
if err != nil {
panic(err)
}
}
peer/msg.go
+30 -28
View File
@@ -1,59 +1,61 @@
package peer
import (
. "github.com/tendermint/tendermint/binary"
"io"
. "github.com/tendermint/tendermint/binary"
"io"
)
/* Packet */
type Packet struct {
Channel String
Bytes ByteSlice
// Hash
Channel String
Bytes ByteSlice
// Hash
}
func NewPacket(chName String, bytes ByteSlice) Packet {
return Packet{
Channel: chName,
Bytes: bytes,
}
return Packet{
Channel: chName,
Bytes: bytes,
}
}
func (p Packet) WriteTo(w io.Writer) (n int64, err error) {
n, err = WriteOnto(&p.Channel, w, n, err)
n, err = WriteOnto(&p.Bytes, w, n, err)
return
n, err = WriteOnto(&p.Channel, w, n, err)
n, err = WriteOnto(&p.Bytes, w, n, err)
return
}
func ReadPacketSafe(r io.Reader) (pkt Packet, err error) {
chName, err := ReadStringSafe(r)
if err != nil { return }
// TODO: packet length sanity check.
bytes, err := ReadByteSliceSafe(r)
if err != nil { return }
return NewPacket(chName, bytes), nil
chName, err := ReadStringSafe(r)
if err != nil {
return
}
// TODO: packet length sanity check.
bytes, err := ReadByteSliceSafe(r)
if err != nil {
return
}
return NewPacket(chName, bytes), nil
}
/* InboundPacket */
type InboundPacket struct {
Peer *Peer
Channel *Channel
Time Time
Packet
Peer *Peer
Channel *Channel
Time Time
Packet
}
/* NewFilterMsg */
type NewFilterMsg struct {
ChName String
Filter interface{} // todo
ChName String
Filter interface{} // todo
}
func (m *NewFilterMsg) WriteTo(w io.Writer) (int64, error) {
panic("TODO: implement")
return 0, nil // TODO
panic("TODO: implement")
return 0, nil // TODO
}
peer/netaddress.go
+123 -113
View File
@@ -5,150 +5,158 @@
package peer
import (
. "github.com/tendermint/tendermint/common"
. "github.com/tendermint/tendermint/binary"
"io"
"net"
"strconv"
. "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
"io"
"net"
"strconv"
)
/* NetAddress */
type NetAddress struct {
IP net.IP
Port UInt16
IP net.IP
Port UInt16
}
// TODO: socks proxies?
func NewNetAddress(addr net.Addr) *NetAddress {
tcpAddr, ok := addr.(*net.TCPAddr)
if !ok { Panicf("Only TCPAddrs are supported. Got: %v", addr) }
ip := tcpAddr.IP
port := UInt16(tcpAddr.Port)
return NewNetAddressIPPort(ip, port)
tcpAddr, ok := addr.(*net.TCPAddr)
if !ok {
Panicf("Only TCPAddrs are supported. Got: %v", addr)
}
ip := tcpAddr.IP
port := UInt16(tcpAddr.Port)
return NewNetAddressIPPort(ip, port)
}
func NewNetAddressString(addr string) *NetAddress {
host, portStr, err := net.SplitHostPort(addr)
if err != nil { panic(err) }
ip := net.ParseIP(host)
port, err := strconv.ParseUint(portStr, 10, 16)
if err != nil { panic(err) }
na := NewNetAddressIPPort(ip, UInt16(port))
return na
host, portStr, err := net.SplitHostPort(addr)
if err != nil {
panic(err)
}
ip := net.ParseIP(host)
port, err := strconv.ParseUint(portStr, 10, 16)
if err != nil {
panic(err)
}
na := NewNetAddressIPPort(ip, UInt16(port))
return na
}
func NewNetAddressIPPort(ip net.IP, port UInt16) *NetAddress {
na := NetAddress{
IP: ip,
Port: port,
}
return &na
na := NetAddress{
IP: ip,
Port: port,
}
return &na
}
func ReadNetAddress(r io.Reader) *NetAddress {
return &NetAddress{
IP: net.IP(ReadByteSlice(r)),
Port: ReadUInt16(r),
}
return &NetAddress{
IP: net.IP(ReadByteSlice(r)),
Port: ReadUInt16(r),
}
}
func (na *NetAddress) WriteTo(w io.Writer) (n int64, err error) {
n, err = WriteOnto(ByteSlice(na.IP.To16()), w, n, err)
n, err = WriteOnto(na.Port, w, n, err)
return
n, err = WriteOnto(ByteSlice(na.IP.To16()), w, n, err)
n, err = WriteOnto(na.Port, w, n, err)
return
}
func (na *NetAddress) Equals(other Binary) bool {
if o, ok := other.(*NetAddress); ok {
return na.String() == o.String()
} else {
return false
}
if o, ok := other.(*NetAddress); ok {
return na.String() == o.String()
} else {
return false
}
}
func (na *NetAddress) Less(other Binary) bool {
if o, ok := other.(*NetAddress); ok {
return na.String() < o.String()
} else {
panic("Cannot compare unequal types")
}
if o, ok := other.(*NetAddress); ok {
return na.String() < o.String()
} else {
panic("Cannot compare unequal types")
}
}
func (na *NetAddress) String() string {
port := strconv.FormatUint(uint64(na.Port), 10)
addr := net.JoinHostPort(na.IP.String(), port)
return addr
port := strconv.FormatUint(uint64(na.Port), 10)
addr := net.JoinHostPort(na.IP.String(), port)
return addr
}
func (na *NetAddress) Dial() (*Connection, error) {
conn, err := net.Dial("tcp", na.String())
if err != nil { return nil, err }
return NewConnection(conn), nil
conn, err := net.Dial("tcp", na.String())
if err != nil {
return nil, err
}
return NewConnection(conn), nil
}
func (na *NetAddress) Routable() bool {
// TODO(oga) bitcoind doesn't include RFC3849 here, but should we?
return na.Valid() && !(na.RFC1918() || na.RFC3927() || na.RFC4862() ||
na.RFC4193() || na.RFC4843() || na.Local())
// TODO(oga) bitcoind doesn't include RFC3849 here, but should we?
return na.Valid() && !(na.RFC1918() || na.RFC3927() || na.RFC4862() ||
na.RFC4193() || na.RFC4843() || na.Local())
}
// For IPv4 these are either a 0 or all bits set address. For IPv6 a zero
// address or one that matches the RFC3849 documentation address format.
func (na *NetAddress) Valid() bool {
return na.IP != nil && !(na.IP.IsUnspecified() || na.RFC3849() ||
na.IP.Equal(net.IPv4bcast))
return na.IP != nil && !(na.IP.IsUnspecified() || na.RFC3849() ||
na.IP.Equal(net.IPv4bcast))
}
func (na *NetAddress) Local() bool {
return na.IP.IsLoopback() || zero4.Contains(na.IP)
return na.IP.IsLoopback() || zero4.Contains(na.IP)
}
func (na *NetAddress) ReachabilityTo(o *NetAddress) int {
const (
Unreachable = 0
Default = iota
Teredo
Ipv6_weak
Ipv4
Ipv6_strong
Private
)
if !na.Routable() {
return Unreachable
} else if na.RFC4380() {
if !o.Routable() {
return Default
} else if o.RFC4380() {
return Teredo
} else if o.IP.To4() != nil {
return Ipv4
} else { // ipv6
return Ipv6_weak
}
} else if na.IP.To4() != nil {
if o.Routable() && o.IP.To4() != nil {
return Ipv4
}
return Default
} else /* ipv6 */ {
var tunnelled bool
// Is our v6 is tunnelled?
if o.RFC3964() || o.RFC6052() || o.RFC6145() {
tunnelled = true
}
if !o.Routable() {
return Default
} else if o.RFC4380() {
return Teredo
} else if o.IP.To4() != nil {
return Ipv4
} else if tunnelled {
// only prioritise ipv6 if we aren't tunnelling it.
return Ipv6_weak
}
return Ipv6_strong
}
const (
Unreachable = 0
Default = iota
Teredo
Ipv6_weak
Ipv4
Ipv6_strong
Private
)
if !na.Routable() {
return Unreachable
} else if na.RFC4380() {
if !o.Routable() {
return Default
} else if o.RFC4380() {
return Teredo
} else if o.IP.To4() != nil {
return Ipv4
} else { // ipv6
return Ipv6_weak
}
} else if na.IP.To4() != nil {
if o.Routable() && o.IP.To4() != nil {
return Ipv4
}
return Default
} else /* ipv6 */ {
var tunnelled bool
// Is our v6 is tunnelled?
if o.RFC3964() || o.RFC6052() || o.RFC6145() {
tunnelled = true
}
if !o.Routable() {
return Default
} else if o.RFC4380() {
return Teredo
} else if o.IP.To4() != nil {
return Ipv4
} else if tunnelled {
// only prioritise ipv6 if we aren't tunnelling it.
return Ipv6_weak
}
return Ipv6_strong
}
}
// RFC1918: IPv4 Private networks (10.0.0.0/8, 192.168.0.0/16, 172.16.0.0/12)
@@ -161,23 +169,25 @@ func (na *NetAddress) ReachabilityTo(o *NetAddress) int {
// RFC4862: IPv6 Autoconfig (FE80::/64)
// RFC6052: IPv6 well known prefix (64:FF9B::/96)
// RFC6145: IPv6 IPv4 translated address ::FFFF:0:0:0/96
var rfc1918_10 = net.IPNet{IP: net.ParseIP("10.0.0.0"), Mask: net.CIDRMask(8, 32)}
var rfc1918_192 = net.IPNet{IP: net.ParseIP("192.168.0.0"), Mask: net.CIDRMask(16, 32)}
var rfc1918_172 = net.IPNet{IP: net.ParseIP("172.16.0.0"), Mask: net.CIDRMask(12, 32)}
var rfc3849 = net.IPNet{IP: net.ParseIP("2001:0DB8::"), Mask: net.CIDRMask(32, 128)}
var rfc3927 = net.IPNet{IP: net.ParseIP("169.254.0.0"), Mask: net.CIDRMask(16, 32)}
var rfc3964 = net.IPNet{IP: net.ParseIP("2002::"), Mask: net.CIDRMask(16, 128)}
var rfc4193 = net.IPNet{IP: net.ParseIP("FC00::"), Mask: net.CIDRMask(7, 128)}
var rfc4380 = net.IPNet{IP: net.ParseIP("2001::"), Mask: net.CIDRMask(32, 128)}
var rfc4843 = net.IPNet{IP: net.ParseIP("2001:10::"), Mask: net.CIDRMask(28, 128)}
var rfc4862 = net.IPNet{IP: net.ParseIP("FE80::"), Mask: net.CIDRMask(64, 128)}
var rfc6052 = net.IPNet{IP: net.ParseIP("64:FF9B::"), Mask: net.CIDRMask(96, 128)}
var rfc6145 = net.IPNet{IP: net.ParseIP("::FFFF:0:0:0"), Mask: net.CIDRMask(96, 128)}
var zero4 = net.IPNet{IP: net.ParseIP("0.0.0.0"), Mask: net.CIDRMask(8, 32)}
var rfc1918_10 = net.IPNet{IP: net.ParseIP("10.0.0.0"), Mask: net.CIDRMask(8, 32)}
var rfc1918_192 = net.IPNet{IP: net.ParseIP("192.168.0.0"), Mask: net.CIDRMask(16, 32)}
var rfc1918_172 = net.IPNet{IP: net.ParseIP("172.16.0.0"), Mask: net.CIDRMask(12, 32)}
var rfc3849 = net.IPNet{IP: net.ParseIP("2001:0DB8::"), Mask: net.CIDRMask(32, 128)}
var rfc3927 = net.IPNet{IP: net.ParseIP("169.254.0.0"), Mask: net.CIDRMask(16, 32)}
var rfc3964 = net.IPNet{IP: net.ParseIP("2002::"), Mask: net.CIDRMask(16, 128)}
var rfc4193 = net.IPNet{IP: net.ParseIP("FC00::"), Mask: net.CIDRMask(7, 128)}
var rfc4380 = net.IPNet{IP: net.ParseIP("2001::"), Mask: net.CIDRMask(32, 128)}
var rfc4843 = net.IPNet{IP: net.ParseIP("2001:10::"), Mask: net.CIDRMask(28, 128)}
var rfc4862 = net.IPNet{IP: net.ParseIP("FE80::"), Mask: net.CIDRMask(64, 128)}
var rfc6052 = net.IPNet{IP: net.ParseIP("64:FF9B::"), Mask: net.CIDRMask(96, 128)}
var rfc6145 = net.IPNet{IP: net.ParseIP("::FFFF:0:0:0"), Mask: net.CIDRMask(96, 128)}
var zero4 = net.IPNet{IP: net.ParseIP("0.0.0.0"), Mask: net.CIDRMask(8, 32)}
func (na *NetAddress) RFC1918() bool { return rfc1918_10.Contains(na.IP) ||
rfc1918_192.Contains(na.IP) ||
rfc1918_172.Contains(na.IP) }
func (na *NetAddress) RFC1918() bool {
return rfc1918_10.Contains(na.IP) ||
rfc1918_192.Contains(na.IP) ||
rfc1918_172.Contains(na.IP)
}
func (na *NetAddress) RFC3849() bool { return rfc3849.Contains(na.IP) }
func (na *NetAddress) RFC3927() bool { return rfc3927.Contains(na.IP) }
func (na *NetAddress) RFC3964() bool { return rfc3964.Contains(na.IP) }
peer/peer.go
+109 -107
View File
@@ -1,172 +1,174 @@
package peer
import (
. "github.com/tendermint/tendermint/binary"
"sync/atomic"
"sync"
"io"
"time"
"fmt"
"fmt"
. "github.com/tendermint/tendermint/binary"
"io"
"sync"
"sync/atomic"
"time"
)
/* Peer */
type Peer struct {
outgoing bool
conn *Connection
channels map[String]*Channel
outgoing bool
conn *Connection
channels map[String]*Channel
mtx sync.Mutex
quit chan struct{}
stopped uint32
mtx sync.Mutex
quit chan struct{}
stopped uint32
}
func NewPeer(conn *Connection) *Peer {
return &Peer{
conn: conn,
quit: make(chan struct{}),
stopped: 0,
}
return &Peer{
conn: conn,
quit: make(chan struct{}),
stopped: 0,
}
}
func (p *Peer) Start(peerRecvQueues map[String]chan *InboundPacket ) {
log.Debugf("Starting %v", p)
p.conn.Start(p.channels)
for chName, _ := range p.channels {
go p.recvHandler(chName, peerRecvQueues[chName])
go p.sendHandler(chName)
}
func (p *Peer) Start(peerRecvQueues map[String]chan *InboundPacket) {
log.Debugf("Starting %v", p)
p.conn.Start(p.channels)
for chName, _ := range p.channels {
go p.recvHandler(chName, peerRecvQueues[chName])
go p.sendHandler(chName)
}
}
func (p *Peer) Stop() {
// lock
p.mtx.Lock()
if atomic.CompareAndSwapUint32(&p.stopped, 0, 1) {
log.Debugf("Stopping %v", p)
close(p.quit)
p.conn.Stop()
}
p.mtx.Unlock()
// unlock
// lock
p.mtx.Lock()
if atomic.CompareAndSwapUint32(&p.stopped, 0, 1) {
log.Debugf("Stopping %v", p)
close(p.quit)
p.conn.Stop()
}
p.mtx.Unlock()
// unlock
}
func (p *Peer) LocalAddress() *NetAddress {
return p.conn.LocalAddress()
return p.conn.LocalAddress()
}
func (p *Peer) RemoteAddress() *NetAddress {
return p.conn.RemoteAddress()
return p.conn.RemoteAddress()
}
func (p *Peer) Channel(chName String) *Channel {
return p.channels[chName]
return p.channels[chName]
}
// If the channel's queue is full, just return false.
// Later the sendHandler will send the pkt to the underlying connection.
func (p *Peer) TrySend(pkt Packet) bool {
channel := p.Channel(pkt.Channel)
sendQueue := channel.SendQueue()
channel := p.Channel(pkt.Channel)
sendQueue := channel.SendQueue()
// lock & defer
p.mtx.Lock(); defer p.mtx.Unlock()
if p.stopped == 1 { return false }
select {
case sendQueue <- pkt:
return true
default: // buffer full
return false
}
// unlock deferred
// lock & defer
p.mtx.Lock()
defer p.mtx.Unlock()
if p.stopped == 1 {
return false
}
select {
case sendQueue <- pkt:
return true
default: // buffer full
return false
}
// unlock deferred
}
func (p *Peer) WriteTo(w io.Writer) (n int64, err error) {
return p.RemoteAddress().WriteTo(w)
return p.RemoteAddress().WriteTo(w)
}
func (p *Peer) String() string {
return fmt.Sprintf("Peer{%v-%v,o:%v}", p.LocalAddress(), p.RemoteAddress(), p.outgoing)
return fmt.Sprintf("Peer{%v-%v,o:%v}", p.LocalAddress(), p.RemoteAddress(), p.outgoing)
}
func (p *Peer) recvHandler(chName String, inboundPacketQueue chan<- *InboundPacket) {
log.Tracef("%v recvHandler [%v]", p, chName)
channel := p.channels[chName]
recvQueue := channel.RecvQueue()
log.Tracef("%v recvHandler [%v]", p, chName)
channel := p.channels[chName]
recvQueue := channel.RecvQueue()
FOR_LOOP:
for {
select {
case <-p.quit:
break FOR_LOOP
case pkt := <-recvQueue:
// send to inboundPacketQueue
inboundPacket := &InboundPacket{
Peer: p,
Channel: channel,
Time: Time{time.Now()},
Packet: pkt,
}
select {
case <-p.quit:
break FOR_LOOP
case inboundPacketQueue <- inboundPacket:
continue
}
}
}
FOR_LOOP:
for {
select {
case <-p.quit:
break FOR_LOOP
case pkt := <-recvQueue:
// send to inboundPacketQueue
inboundPacket := &InboundPacket{
Peer: p,
Channel: channel,
Time: Time{time.Now()},
Packet: pkt,
}
select {
case <-p.quit:
break FOR_LOOP
case inboundPacketQueue <- inboundPacket:
continue
}
}
}
log.Tracef("%v recvHandler [%v] closed", p, chName)
// cleanup
// (none)
log.Tracef("%v recvHandler [%v] closed", p, chName)
// cleanup
// (none)
}
func (p *Peer) sendHandler(chName String) {
log.Tracef("%v sendHandler [%v]", p, chName)
chSendQueue := p.channels[chName].sendQueue
FOR_LOOP:
for {
select {
case <-p.quit:
break FOR_LOOP
case pkt := <-chSendQueue:
log.Tracef("Sending packet to peer chSendQueue")
// blocks until the connection is Stop'd,
// which happens when this peer is Stop'd.
p.conn.Send(pkt)
}
}
log.Tracef("%v sendHandler [%v]", p, chName)
chSendQueue := p.channels[chName].sendQueue
FOR_LOOP:
for {
select {
case <-p.quit:
break FOR_LOOP
case pkt := <-chSendQueue:
log.Tracef("Sending packet to peer chSendQueue")
// blocks until the connection is Stop'd,
// which happens when this peer is Stop'd.
p.conn.Send(pkt)
}
}
log.Tracef("%v sendHandler [%v] closed", p, chName)
// cleanup
// (none)
log.Tracef("%v sendHandler [%v] closed", p, chName)
// cleanup
// (none)
}
/* Channel */
type Channel struct {
name String
recvQueue chan Packet
sendQueue chan Packet
//stats Stats
name String
recvQueue chan Packet
sendQueue chan Packet
//stats Stats
}
func NewChannel(name String, bufferSize int) *Channel {
return &Channel{
name: name,
recvQueue: make(chan Packet, bufferSize),
sendQueue: make(chan Packet, bufferSize),
}
return &Channel{
name: name,
recvQueue: make(chan Packet, bufferSize),
sendQueue: make(chan Packet, bufferSize),
}
}
func (c *Channel) Name() String {
return c.name
return c.name
}
func (c *Channel) RecvQueue() <-chan Packet {
return c.recvQueue
return c.recvQueue
}
func (c *Channel) SendQueue() chan<- Packet {
return c.sendQueue
return c.sendQueue
}
peer/server.go
+18 -19
View File
@@ -1,39 +1,38 @@
package peer
import (
)
import ()
/* Server */
type Server struct {
listener Listener
client *Client
listener Listener
client *Client
}
func NewServer(protocol string, laddr string, c *Client) *Server {
l := NewDefaultListener(protocol, laddr)
s := &Server{
listener: l,
client: c,
}
go s.IncomingConnectionHandler()
return s
l := NewDefaultListener(protocol, laddr)
s := &Server{
listener: l,
client: c,
}
go s.IncomingConnectionHandler()
return s
}
func (s *Server) LocalAddress() *NetAddress {
return s.listener.LocalAddress()
return s.listener.LocalAddress()
}
// meant to run in a goroutine
func (s *Server) IncomingConnectionHandler() {
for conn := range s.listener.Connections() {
log.Infof("New connection found: %v", conn)
s.client.AddPeerWithConnection(conn, false)
}
for conn := range s.listener.Connections() {
log.Infof("New connection found: %v", conn)
s.client.AddPeerWithConnection(conn, false)
}
}
func (s *Server) Stop() {
log.Infof("Stopping server")
s.listener.Stop()
s.client.Stop()
log.Infof("Stopping server")
s.listener.Stop()
s.client.Stop()
}
peer/upnp.go
+279 -279
View File
@@ -7,370 +7,370 @@ package peer
//
import (
"bytes"
"encoding/xml"
"errors"
"io/ioutil"
"net"
"net/http"
"strconv"
"strings"
"time"
"bytes"
"encoding/xml"
"errors"
"io/ioutil"
"net"
"net/http"
"strconv"
"strings"
"time"
)
type upnpNAT struct {
serviceURL string
ourIP string
urnDomain string
serviceURL string
ourIP string
urnDomain string
}
// protocol is either "udp" or "tcp"
type NAT interface {
GetExternalAddress() (addr net.IP, err error)
AddPortMapping(protocol string, externalPort, internalPort int, description string, timeout int) (mappedExternalPort int, err error)
DeletePortMapping(protocol string, externalPort, internalPort int) (err error)
GetExternalAddress() (addr net.IP, err error)
AddPortMapping(protocol string, externalPort, internalPort int, description string, timeout int) (mappedExternalPort int, err error)
DeletePortMapping(protocol string, externalPort, internalPort int) (err error)
}
func Discover() (nat NAT, err error) {
ssdp, err := net.ResolveUDPAddr("udp4", "239.255.255.250:1900")
if err != nil {
return
}
conn, err := net.ListenPacket("udp4", ":0")
if err != nil {
return
}
socket := conn.(*net.UDPConn)
defer socket.Close()
ssdp, err := net.ResolveUDPAddr("udp4", "239.255.255.250:1900")
if err != nil {
return
}
conn, err := net.ListenPacket("udp4", ":0")
if err != nil {
return
}
socket := conn.(*net.UDPConn)
defer socket.Close()
err = socket.SetDeadline(time.Now().Add(3 * time.Second))
if err != nil {
return
}
err = socket.SetDeadline(time.Now().Add(3 * time.Second))
if err != nil {
return
}
st := "InternetGatewayDevice:1"
st := "InternetGatewayDevice:1"
buf := bytes.NewBufferString(
"M-SEARCH * HTTP/1.1\r\n" +
"HOST: 239.255.255.250:1900\r\n" +
"ST: ssdp:all\r\n" +
"MAN: \"ssdp:discover\"\r\n" +
"MX: 2\r\n\r\n")
message := buf.Bytes()
answerBytes := make([]byte, 1024)
for i := 0; i < 3; i++ {
_, err = socket.WriteToUDP(message, ssdp)
if err != nil {
return
}
var n int
n, _, err = socket.ReadFromUDP(answerBytes)
for {
n, _, err = socket.ReadFromUDP(answerBytes)
if err != nil {
break
}
answer := string(answerBytes[0:n])
if strings.Index(answer, st) < 0 {
continue
}
// HTTP header field names are case-insensitive.
// http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2
locString := "\r\nlocation:"
answer = strings.ToLower(answer)
locIndex := strings.Index(answer, locString)
if locIndex < 0 {
continue
}
loc := answer[locIndex+len(locString):]
endIndex := strings.Index(loc, "\r\n")
if endIndex < 0 {
continue
}
locURL := strings.TrimSpace(loc[0:endIndex])
var serviceURL, urnDomain string
serviceURL, urnDomain, err = getServiceURL(locURL)
if err != nil {
return
}
var ourIP net.IP
ourIP, err = localIPv4()
if err != nil {
return
}
nat = &upnpNAT{serviceURL: serviceURL, ourIP: ourIP.String(), urnDomain: urnDomain}
return
}
}
err = errors.New("UPnP port discovery failed.")
return
buf := bytes.NewBufferString(
"M-SEARCH * HTTP/1.1\r\n" +
"HOST: 239.255.255.250:1900\r\n" +
"ST: ssdp:all\r\n" +
"MAN: \"ssdp:discover\"\r\n" +
"MX: 2\r\n\r\n")
message := buf.Bytes()
answerBytes := make([]byte, 1024)
for i := 0; i < 3; i++ {
_, err = socket.WriteToUDP(message, ssdp)
if err != nil {
return
}
var n int
n, _, err = socket.ReadFromUDP(answerBytes)
for {
n, _, err = socket.ReadFromUDP(answerBytes)
if err != nil {
break
}
answer := string(answerBytes[0:n])
if strings.Index(answer, st) < 0 {
continue
}
// HTTP header field names are case-insensitive.
// http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2
locString := "\r\nlocation:"
answer = strings.ToLower(answer)
locIndex := strings.Index(answer, locString)
if locIndex < 0 {
continue
}
loc := answer[locIndex+len(locString):]
endIndex := strings.Index(loc, "\r\n")
if endIndex < 0 {
continue
}
locURL := strings.TrimSpace(loc[0:endIndex])
var serviceURL, urnDomain string
serviceURL, urnDomain, err = getServiceURL(locURL)
if err != nil {
return
}
var ourIP net.IP
ourIP, err = localIPv4()
if err != nil {
return
}
nat = &upnpNAT{serviceURL: serviceURL, ourIP: ourIP.String(), urnDomain: urnDomain}
return
}
}
err = errors.New("UPnP port discovery failed.")
return
}
type Envelope struct {
XMLName xml.Name `xml:"http://schemas.xmlsoap.org/soap/envelope/ Envelope"`
Soap *SoapBody
XMLName xml.Name `xml:"http://schemas.xmlsoap.org/soap/envelope/ Envelope"`
Soap *SoapBody
}
type SoapBody struct {
XMLName xml.Name `xml:"http://schemas.xmlsoap.org/soap/envelope/ Body"`
ExternalIP *ExternalIPAddressResponse
XMLName xml.Name `xml:"http://schemas.xmlsoap.org/soap/envelope/ Body"`
ExternalIP *ExternalIPAddressResponse
}
type ExternalIPAddressResponse struct {
XMLName xml.Name `xml:"GetExternalIPAddressResponse"`
IPAddress string `xml:"NewExternalIPAddress"`
XMLName xml.Name `xml:"GetExternalIPAddressResponse"`
IPAddress string `xml:"NewExternalIPAddress"`
}
type ExternalIPAddress struct {
XMLName xml.Name `xml:"NewExternalIPAddress"`
IP string
XMLName xml.Name `xml:"NewExternalIPAddress"`
IP string
}
type Service struct {
ServiceType string `xml:"serviceType"`
ControlURL string `xml:"controlURL"`
ServiceType string `xml:"serviceType"`
ControlURL string `xml:"controlURL"`
}
type DeviceList struct {
Device []Device `xml:"device"`
Device []Device `xml:"device"`
}
type ServiceList struct {
Service []Service `xml:"service"`
Service []Service `xml:"service"`
}
type Device struct {
XMLName xml.Name `xml:"device"`
DeviceType string `xml:"deviceType"`
DeviceList DeviceList `xml:"deviceList"`
ServiceList ServiceList `xml:"serviceList"`
XMLName xml.Name `xml:"device"`
DeviceType string `xml:"deviceType"`
DeviceList DeviceList `xml:"deviceList"`
ServiceList ServiceList `xml:"serviceList"`
}
type Root struct {
Device Device
Device Device
}
func getChildDevice(d *Device, deviceType string) *Device {
dl := d.DeviceList.Device
for i := 0; i < len(dl); i++ {
if strings.Index(dl[i].DeviceType, deviceType) >= 0 {
return &dl[i]
}
}
return nil
dl := d.DeviceList.Device
for i := 0; i < len(dl); i++ {
if strings.Index(dl[i].DeviceType, deviceType) >= 0 {
return &dl[i]
}
}
return nil
}
func getChildService(d *Device, serviceType string) *Service {
sl := d.ServiceList.Service
for i := 0; i < len(sl); i++ {
if strings.Index(sl[i].ServiceType, serviceType) >= 0 {
return &sl[i]
}
}
return nil
sl := d.ServiceList.Service
for i := 0; i < len(sl); i++ {
if strings.Index(sl[i].ServiceType, serviceType) >= 0 {
return &sl[i]
}
}
return nil
}
func localIPv4() (net.IP, error) {
tt, err := net.Interfaces()
if err != nil {
return nil, err
}
for _, t := range tt {
aa, err := t.Addrs()
if err != nil {
return nil, err
}
for _, a := range aa {
ipnet, ok := a.(*net.IPNet)
if !ok {
continue
}
v4 := ipnet.IP.To4()
if v4 == nil || v4[0] == 127 { // loopback address
continue
}
return v4, nil
}
}
return nil, errors.New("cannot find local IP address")
tt, err := net.Interfaces()
if err != nil {
return nil, err
}
for _, t := range tt {
aa, err := t.Addrs()
if err != nil {
return nil, err
}
for _, a := range aa {
ipnet, ok := a.(*net.IPNet)
if !ok {
continue
}
v4 := ipnet.IP.To4()
if v4 == nil || v4[0] == 127 { // loopback address
continue
}
return v4, nil
}
}
return nil, errors.New("cannot find local IP address")
}
func getServiceURL(rootURL string) (url, urnDomain string, err error) {
r, err := http.Get(rootURL)
if err != nil {
return
}
defer r.Body.Close()
if r.StatusCode >= 400 {
err = errors.New(string(r.StatusCode))
return
}
var root Root
err = xml.NewDecoder(r.Body).Decode(&root)
if err != nil {
return
}
a := &root.Device
if strings.Index(a.DeviceType, "InternetGatewayDevice:1") < 0 {
err = errors.New("No InternetGatewayDevice")
return
}
b := getChildDevice(a, "WANDevice:1")
if b == nil {
err = errors.New("No WANDevice")
return
}
c := getChildDevice(b, "WANConnectionDevice:1")
if c == nil {
err = errors.New("No WANConnectionDevice")
return
}
d := getChildService(c, "WANIPConnection:1")
if d == nil {
// Some routers don't follow the UPnP spec, and put WanIPConnection under WanDevice,
// instead of under WanConnectionDevice
d = getChildService(b, "WANIPConnection:1")
r, err := http.Get(rootURL)
if err != nil {
return
}
defer r.Body.Close()
if r.StatusCode >= 400 {
err = errors.New(string(r.StatusCode))
return
}
var root Root
err = xml.NewDecoder(r.Body).Decode(&root)
if err != nil {
return
}
a := &root.Device
if strings.Index(a.DeviceType, "InternetGatewayDevice:1") < 0 {
err = errors.New("No InternetGatewayDevice")
return
}
b := getChildDevice(a, "WANDevice:1")
if b == nil {
err = errors.New("No WANDevice")
return
}
c := getChildDevice(b, "WANConnectionDevice:1")
if c == nil {
err = errors.New("No WANConnectionDevice")
return
}
d := getChildService(c, "WANIPConnection:1")
if d == nil {
// Some routers don't follow the UPnP spec, and put WanIPConnection under WanDevice,
// instead of under WanConnectionDevice
d = getChildService(b, "WANIPConnection:1")
if d == nil {
err = errors.New("No WANIPConnection")
return
}
}
// Extract the domain name, which isn't always 'schemas-upnp-org'
urnDomain = strings.Split(d.ServiceType, ":")[1]
url = combineURL(rootURL, d.ControlURL)
return
if d == nil {
err = errors.New("No WANIPConnection")
return
}
}
// Extract the domain name, which isn't always 'schemas-upnp-org'
urnDomain = strings.Split(d.ServiceType, ":")[1]
url = combineURL(rootURL, d.ControlURL)
return
}
func combineURL(rootURL, subURL string) string {
protocolEnd := "://"
protoEndIndex := strings.Index(rootURL, protocolEnd)
a := rootURL[protoEndIndex+len(protocolEnd):]
rootIndex := strings.Index(a, "/")
return rootURL[0:protoEndIndex+len(protocolEnd)+rootIndex] + subURL
protocolEnd := "://"
protoEndIndex := strings.Index(rootURL, protocolEnd)
a := rootURL[protoEndIndex+len(protocolEnd):]
rootIndex := strings.Index(a, "/")
return rootURL[0:protoEndIndex+len(protocolEnd)+rootIndex] + subURL
}
func soapRequest(url, function, message, domain string) (r *http.Response, err error) {
fullMessage := "<?xml version=\"1.0\" ?>" +
"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\">\r\n" +
"<s:Body>" + message + "</s:Body></s:Envelope>"
fullMessage := "<?xml version=\"1.0\" ?>" +
"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\">\r\n" +
"<s:Body>" + message + "</s:Body></s:Envelope>"
req, err := http.NewRequest("POST", url, strings.NewReader(fullMessage))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "text/xml ; charset=\"utf-8\"")
req.Header.Set("User-Agent", "Darwin/10.0.0, UPnP/1.0, MiniUPnPc/1.3")
//req.Header.Set("Transfer-Encoding", "chunked")
req.Header.Set("SOAPAction", "\"urn:"+domain+":service:WANIPConnection:1#"+function+"\"")
req.Header.Set("Connection", "Close")
req.Header.Set("Cache-Control", "no-cache")
req.Header.Set("Pragma", "no-cache")
req, err := http.NewRequest("POST", url, strings.NewReader(fullMessage))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "text/xml ; charset=\"utf-8\"")
req.Header.Set("User-Agent", "Darwin/10.0.0, UPnP/1.0, MiniUPnPc/1.3")
//req.Header.Set("Transfer-Encoding", "chunked")
req.Header.Set("SOAPAction", "\"urn:"+domain+":service:WANIPConnection:1#"+function+"\"")
req.Header.Set("Connection", "Close")
req.Header.Set("Cache-Control", "no-cache")
req.Header.Set("Pragma", "no-cache")
// log.Stderr("soapRequest ", req)
// log.Stderr("soapRequest ", req)
r, err = http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
/*if r.Body != nil {
defer r.Body.Close()
}*/
r, err = http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
/*if r.Body != nil {
defer r.Body.Close()
}*/
if r.StatusCode >= 400 {
// log.Stderr(function, r.StatusCode)
err = errors.New("Error " + strconv.Itoa(r.StatusCode) + " for " + function)
r = nil
return
}
return
if r.StatusCode >= 400 {
// log.Stderr(function, r.StatusCode)
err = errors.New("Error " + strconv.Itoa(r.StatusCode) + " for " + function)
r = nil
return
}
return
}
type statusInfo struct {
externalIpAddress string
externalIpAddress string
}
func (n *upnpNAT) getExternalIPAddress() (info statusInfo, err error) {
message := "<u:GetExternalIPAddress xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"</u:GetExternalIPAddress>"
message := "<u:GetExternalIPAddress xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"</u:GetExternalIPAddress>"
var response *http.Response
response, err = soapRequest(n.serviceURL, "GetExternalIPAddress", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
var envelope Envelope
data, err := ioutil.ReadAll(response.Body)
reader := bytes.NewReader(data)
xml.NewDecoder(reader).Decode(&envelope)
var response *http.Response
response, err = soapRequest(n.serviceURL, "GetExternalIPAddress", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
var envelope Envelope
data, err := ioutil.ReadAll(response.Body)
reader := bytes.NewReader(data)
xml.NewDecoder(reader).Decode(&envelope)
info = statusInfo{envelope.Soap.ExternalIP.IPAddress}
info = statusInfo{envelope.Soap.ExternalIP.IPAddress}
if err != nil {
return
}
if err != nil {
return
}
return
return
}
func (n *upnpNAT) GetExternalAddress() (addr net.IP, err error) {
info, err := n.getExternalIPAddress()
if err != nil {
return
}
addr = net.ParseIP(info.externalIpAddress)
return
info, err := n.getExternalIPAddress()
if err != nil {
return
}
addr = net.ParseIP(info.externalIpAddress)
return
}
func (n *upnpNAT) AddPortMapping(protocol string, externalPort, internalPort int, description string, timeout int) (mappedExternalPort int, err error) {
// A single concatenation would break ARM compilation.
message := "<u:AddPortMapping xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"<NewRemoteHost></NewRemoteHost><NewExternalPort>" + strconv.Itoa(externalPort)
message += "</NewExternalPort><NewProtocol>" + protocol + "</NewProtocol>"
message += "<NewInternalPort>" + strconv.Itoa(internalPort) + "</NewInternalPort>" +
"<NewInternalClient>" + n.ourIP + "</NewInternalClient>" +
"<NewEnabled>1</NewEnabled><NewPortMappingDescription>"
message += description +
"</NewPortMappingDescription><NewLeaseDuration>" + strconv.Itoa(timeout) +
"</NewLeaseDuration></u:AddPortMapping>"
// A single concatenation would break ARM compilation.
message := "<u:AddPortMapping xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"<NewRemoteHost></NewRemoteHost><NewExternalPort>" + strconv.Itoa(externalPort)
message += "</NewExternalPort><NewProtocol>" + protocol + "</NewProtocol>"
message += "<NewInternalPort>" + strconv.Itoa(internalPort) + "</NewInternalPort>" +
"<NewInternalClient>" + n.ourIP + "</NewInternalClient>" +
"<NewEnabled>1</NewEnabled><NewPortMappingDescription>"
message += description +
"</NewPortMappingDescription><NewLeaseDuration>" + strconv.Itoa(timeout) +
"</NewLeaseDuration></u:AddPortMapping>"
var response *http.Response
response, err = soapRequest(n.serviceURL, "AddPortMapping", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
var response *http.Response
response, err = soapRequest(n.serviceURL, "AddPortMapping", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
// TODO: check response to see if the port was forwarded
// log.Println(message, response)
mappedExternalPort = externalPort
_ = response
return
// TODO: check response to see if the port was forwarded
// log.Println(message, response)
mappedExternalPort = externalPort
_ = response
return
}
func (n *upnpNAT) DeletePortMapping(protocol string, externalPort, internalPort int) (err error) {
message := "<u:DeletePortMapping xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"<NewRemoteHost></NewRemoteHost><NewExternalPort>" + strconv.Itoa(externalPort) +
"</NewExternalPort><NewProtocol>" + protocol + "</NewProtocol>" +
"</u:DeletePortMapping>"
message := "<u:DeletePortMapping xmlns:u=\"urn:" + n.urnDomain + ":service:WANIPConnection:1\">\r\n" +
"<NewRemoteHost></NewRemoteHost><NewExternalPort>" + strconv.Itoa(externalPort) +
"</NewExternalPort><NewProtocol>" + protocol + "</NewProtocol>" +
"</u:DeletePortMapping>"
var response *http.Response
response, err = soapRequest(n.serviceURL, "DeletePortMapping", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
var response *http.Response
response, err = soapRequest(n.serviceURL, "DeletePortMapping", message, n.urnDomain)
if response != nil {
defer response.Body.Close()
}
if err != nil {
return
}
// TODO: check response to see if the port was deleted
// log.Println(message, response)
_ = response
return
// TODO: check response to see if the port was deleted
// log.Println(message, response)
_ = response
return
}
peer/upnp_test.go
+28 -28
View File
@@ -1,8 +1,8 @@
package peer
import (
"testing"
"time"
"testing"
"time"
)
/*
@@ -11,38 +11,38 @@ TODO: set up or find a service to probe open ports.
*/
func TestUPNP(t *testing.T) {
t.Log("hello!")
t.Log("hello!")
nat, err := Discover()
if err != nil {
t.Fatalf("NAT upnp could not be discovered: %v", err)
}
nat, err := Discover()
if err != nil {
t.Fatalf("NAT upnp could not be discovered: %v", err)
}
t.Log("ourIP: ", nat.(*upnpNAT).ourIP)
t.Log("ourIP: ", nat.(*upnpNAT).ourIP)
ext, err := nat.GetExternalAddress()
if err != nil {
t.Fatalf("External address error: %v", err)
}
t.Logf("External address: %v", ext)
ext, err := nat.GetExternalAddress()
if err != nil {
t.Fatalf("External address error: %v", err)
}
t.Logf("External address: %v", ext)
port, err := nat.AddPortMapping("tcp", 8001, 8001, "testing", 0)
if err != nil {
t.Fatalf("Port mapping error: %v", err)
}
t.Logf("Port mapping mapped: %v", port)
port, err := nat.AddPortMapping("tcp", 8001, 8001, "testing", 0)
if err != nil {
t.Fatalf("Port mapping error: %v", err)
}
t.Logf("Port mapping mapped: %v", port)
// also run the listener, open for all remote addresses.
listener := NewDefaultListener("tcp", "0.0.0.0:8001")
// also run the listener, open for all remote addresses.
listener := NewDefaultListener("tcp", "0.0.0.0:8001")
// now sleep for 10 seconds
time.Sleep(10 * time.Second)
// now sleep for 10 seconds
time.Sleep(10 * time.Second)
err = nat.DeletePortMapping("tcp", 8001, 8001)
if err != nil {
t.Fatalf("Port mapping delete error: %v", err)
}
t.Logf("Port mapping deleted")
err = nat.DeletePortMapping("tcp", 8001, 8001)
if err != nil {
t.Fatalf("Port mapping delete error: %v", err)
}
t.Logf("Port mapping deleted")
listener.Stop()
listener.Stop()
}
peer/util.go
+1 -1
View File
@@ -1,7 +1,7 @@
package peer
import (
"crypto/sha256"
"crypto/sha256"
)
// DoubleSha256 calculates sha256(sha256(b)) and returns the resulting bytes.