Merge pull request #117 from subutai-io/dev

Dev

instance.go

@@ -8,8 +8,25 @@ import (
 	ptp "github.com/subutai-io/p2p/lib"
 	"os"
 	"runtime"
+	"time"
 )
 
+var InstanceLock bool = false
+
+func WaitLock() {
+	for InstanceLock {
+		time.Sleep(100 * time.Microsecond)
+	}
+}
+
+func Lock() {
+	InstanceLock = true
+}
+
+func Unlock() {
+	InstanceLock = false
+}
+
 type RunArgs struct {
 	IP      string
 	Mac     string
@@ -150,6 +167,8 @@ func (p *Procedures) SetLog(args *NameValueArg, resp *Response) error {
 }
 
 func (p *Procedures) AddKey(args *RunArgs, resp *Response) error {
+	WaitLock()
+	Lock()
 	resp.ExitCode = 0
 	if args.Hash == "" {
 		resp.ExitCode = 1
@@ -170,19 +189,35 @@ func (p *Procedures) AddKey(args *RunArgs, resp *Response) error {
 		newKey = Instances[args.Hash].PTP.Crypter.EnrichKeyValues(newKey, args.Key, args.TTL)
 		Instances[args.Hash].PTP.Crypter.Keys = append(Instances[args.Hash].PTP.Crypter.Keys, newKey)
 	}
+	Unlock()
 	return nil
 }
 
 func (p *Procedures) Execute(args *Args, resp *Response) error {
-	ptp.Log(ptp.INFO, "Received %v", args)
 	resp.ExitCode = 0
-	resp.Output = "Command executed"
+	resp.Output = ""
 	return nil
 }
 
 func (p *Procedures) Run(args *RunArgs, resp *Response) error {
+	WaitLock()
+	Lock()
 	resp.ExitCode = 0
 	resp.Output = "Running new P2P instance for " + args.Hash + "\n"
+	defer Unlock()
+
+	// Validate if interface name is unique
+	if args.Dev != "" {
+		for _, inst := range Instances {
+			if inst.PTP.DeviceName == args.Dev {
+				resp.ExitCode = 1
+				resp.Output = "Device name is already in use"
+				Unlock()
+				return errors.New(resp.Output)
+			}
+		}
+	}
+
 	var exists bool
 	_, exists = Instances[args.Hash]
 	if !exists {
@@ -204,22 +239,29 @@ func (p *Procedures) Run(args *RunArgs, resp *Response) error {
 		Instances[args.Hash] = newInst
 		ptpInstance := ptp.StartP2PInstance(args.IP, args.Mac, args.Dev, "", args.Hash, args.Dht, args.Keyfile, args.Key, args.TTL, "", args.Fwd, args.Port)
 		if ptpInstance == nil {
+			delete(Instances, args.Hash)
 			resp.Output = resp.Output + "Failed to create P2P Instance"
+			Unlock()
 			return errors.New("Failed to create P2P Instance")
 		}
 		newInst.PTP = ptpInstance
 		Instances[args.Hash] = newInst
 		go ptpInstance.Run()
 		if SaveFile != "" {
+			resp.Output = resp.Output + "Saving instance into file"
 			SaveInstances(SaveFile)
 		}
 	} else {
 		resp.Output = resp.Output + "Hash already in use\n"
 	}
+	Unlock()
 	return nil
 }
 
 func (p *Procedures) Stop(args *StopArgs, resp *Response) error {
+	WaitLock()
+	Lock()
+	defer Unlock()
 	resp.ExitCode = 0
 	var exists bool
 	_, exists = Instances[args.Hash]
@@ -231,6 +273,7 @@ func (p *Procedures) Stop(args *StopArgs, resp *Response) error {
 		Instances[args.Hash].PTP.StopInstance()
 		delete(Instances, args.Hash)
 	}
+	Unlock()
 	return nil
 }
 
@@ -255,7 +298,11 @@ func (p *Procedures) Show(args *Args, resp *Response) error {
 			resp.Output = "No instances was found"
 		}
 		for key, inst := range Instances {
-			resp.Output = resp.Output + "\t" + inst.PTP.Mac + "\t" + inst.PTP.IP + "\t" + key
+			if inst.PTP != nil {
+				resp.Output = resp.Output + "\t" + inst.PTP.Mac + "\t" + inst.PTP.IP + "\t" + key
+			} else {
+				resp.Output = resp.Output + "\tUnknown\tUnknown\t" + key
+			}
 			resp.Output = resp.Output + "\n"
 		}
 	}

lib/dht.go

@@ -491,9 +491,7 @@ func (dht *DHTClient) Initialize(config *DHTClient, ips []net.IP) *DHTClient {
 		}
 	}
 	if connected == 0 {
-		Log(WARNING, "Failed to connect to DHT. Retrying in 5 seconds")
-		time.Sleep(5 * time.Second)
-		return dht.Initialize(config, ips)
+		return nil
 	} else {
 		return dht
 	}

lib/net.go

@@ -108,6 +108,17 @@ func CreatePingP2PMessage() *P2PMessage {
 	return msg
 }
 
+func CreateXpingP2PMessage(t uint16, hw string) *P2PMessage {
+	msg := new(P2PMessage)
+	msg.Header = new(P2PMessageHeader)
+	msg.Header.Magic = MAGIC_COOKIE
+	msg.Header.Type = uint16(MT_XPING)
+	msg.Header.NetProto = t
+	msg.Header.Length = uint16(len(hw))
+	msg.Data = []byte(hw)
+	return msg
+}
+
 func CreateIntroP2PMessage(c Crypto, data string, netProto uint16) *P2PMessage {
 	msg := new(P2PMessage)
 	msg.Header = new(P2PMessageHeader)

lib/p2p.go

@@ -50,6 +50,7 @@ type NetworkPeer struct {
 	Retries      int              // Number of introduction retries
 	Ready        bool             // Set to true when peer is ready to communicate with p2p network
 	State        PeerState        // State of a peer
+	FailedPings  int              // Number of missed pings
 }
 
 // Creates TUN/TAP Interface and configures it with provided IP tool
@@ -127,21 +128,24 @@ func (p *PTPCloud) ListenInterface() {
 	Log(INFO, "Shutting down interface listener")
 }
 
-// This method will generate device name if none were specified at startup
-func (p *PTPCloud) GenerateDeviceName(i int) string {
-	var devName string = GetDeviceBase() + fmt.Sprintf("%d", i)
+func (p *PTPCloud) IsDeviceExists(name string) bool {
 	inf, err := net.Interfaces()
 	if err != nil {
 		Log(ERROR, "Failed to retrieve list of network interfaces")
-		return ""
+		return true
 	}
-	var exist bool = false
 	for _, i := range inf {
-		if i.Name == devName {
-			exist = true
+		if i.Name == name {
+			return true
 		}
 	}
-	if exist {
+	return false
+}
+
+// This method will generate device name if none were specified at startup
+func (p *PTPCloud) GenerateDeviceName(i int) string {
+	var devName string = GetDeviceBase() + fmt.Sprintf("%d", i)
+	if p.IsDeviceExists(devName) {
 		return p.GenerateDeviceName(i + 1)
 	} else {
 		return devName
@@ -253,6 +257,10 @@ func StartP2PInstance(argIp, argMac, argDev, argDirect, argHash, argDht, argKeyf
 			return nil
 		}
 	}
+	if p.IsDeviceExists(argDev) {
+		Log(ERROR, "Interface is already in use. Can't create duplicate")
+		return nil
+	}
 
 	if argKeyfile != "" {
 		p.Crypter.ReadKeysFromFile(argKeyfile)
@@ -279,6 +287,7 @@ func StartP2PInstance(argIp, argMac, argDev, argDirect, argHash, argDht, argKeyf
 	p.MessageHandlers = make(map[uint16]MessageHandler)
 	p.MessageHandlers[MT_NENC] = p.HandleNotEncryptedMessage
 	p.MessageHandlers[MT_PING] = p.HandlePingMessage
+	p.MessageHandlers[MT_XPING] = p.HandleXpingMessage
 	p.MessageHandlers[MT_ENC] = p.HandleMessage
 	p.MessageHandlers[MT_INTRO] = p.HandleIntroMessage
 	p.MessageHandlers[MT_INTRO_REQ] = p.HandleIntroRequestMessage
@@ -306,11 +315,19 @@ func StartP2PInstance(argIp, argMac, argDev, argDirect, argHash, argDht, argKeyf
 		config.Routers = argDht
 	}
 	p.Dht = dhtClient.Initialize(config, p.LocalIPs)
+	for p.Dht == nil {
+		Log(WARNING, "Failed to connect to DHT. Retrying in 5 seconds")
+		time.Sleep(5 * time.Second)
+		p.LocalIPs = p.LocalIPs[:0]
+		p.FindNetworkAddresses()
+		p.Dht = dhtClient.Initialize(config, p.LocalIPs)
+	}
 	// Wait for ID
 	for len(p.Dht.ID) < 32 {
 		time.Sleep(100 * time.Millisecond)
 	}
 	Log(INFO, "ID assigned. Continue")
+	var retries int = 0
 	if argIp == "dhcp" {
 		Log(INFO, "Requesting IP")
 		p.Dht.RequestIP()
@@ -319,6 +336,11 @@ func StartP2PInstance(argIp, argMac, argDev, argDirect, argHash, argDht, argKeyf
 			Log(INFO, "No IP were received. Requesting again")
 			p.Dht.RequestIP()
 			time.Sleep(3 * time.Second)
+			retries++
+			if retries >= 10 {
+				Log(ERROR, "Failed to retrieve IP from network after 10 retries")
+				return nil
+			}
 		}
 		m := p.Dht.Network.Mask
 		mask := fmt.Sprintf("%d.%d.%d.%d", m[0], m[1], m[2], m[3])
@@ -353,6 +375,7 @@ func StartP2PInstance(argIp, argMac, argDev, argDirect, argHash, argDht, argKeyf
 }
 
 func (p *PTPCloud) Run() {
+	go p.PingPeers()
 	for {
 		if p.Shutdown {
 			// TODO: Do it more safely
@@ -809,6 +832,27 @@ func (p *PTPCloud) HandlePingMessage(msg *P2PMessage, src_addr *net.UDPAddr) {
 	p.UDPSocket.SendMessage(msg, src_addr)
 }
 
+func (p *PTPCloud) HandleXpingMessage(msg *P2PMessage, src_addr *net.UDPAddr) {
+	if msg.Header.NetProto == 1 {
+		Log(DEBUG, "PING REQUEST")
+		// Send response
+		rsp := CreateXpingP2PMessage(2, p.HardwareAddr.String())
+		addr, err := net.ParseMAC(string(msg.Data))
+		if err != nil {
+			return
+		}
+		p.SendTo(addr, rsp)
+	} else {
+		Log(DEBUG, "PING RESPONSE")
+		for i, peer := range p.NetworkPeers {
+			if peer.PeerHW.String() == string(msg.Data) {
+				peer.FailedPings = 0
+				p.NetworkPeers[i] = peer
+			}
+		}
+	}
+}
+
 func (p *PTPCloud) IsPeerReady(id string) bool {
 	for _, peer := range p.NetworkPeers {
 		if peer.ID == id {
@@ -917,6 +961,45 @@ func (p *PTPCloud) StopInstance() {
 	msg := CreateTestP2PMessage(p.Crypter, "STOP", 1)
 	addr, _ := net.ResolveUDPAddr("udp", fmt.Sprintf("127.0.0.1:%d", p.Dht.P2PPort))
 	p.UDPSocket.SendMessage(msg, addr)
+	var ipIt int = 200
+	for p.IsDeviceExists(p.DeviceName) {
+		time.Sleep(1 * time.Second)
+		ip := p.Dht.Network.IP
+		target := fmt.Sprintf("%d.%d.%d.%d:99", ip[0], ip[1], ip[2], ipIt)
+		Log(INFO, "Dialing %s", target)
+		_, err := net.DialTimeout("tcp", target, 2*time.Second)
+		if err != nil {
+			Log(INFO, "ERROR: %v", err)
+		}
+		ipIt++
+		if ipIt == 255 {
+			break
+		}
+	}
 	time.Sleep(3 * time.Second)
 	p.ReadyToStop = true
 }
+
+func (p *PTPCloud) PingPeers() {
+	for {
+		if p.Shutdown {
+			break
+		}
+		msg := CreateXpingP2PMessage(1, p.HardwareAddr.String())
+		Log(DEBUG, "Sending pings")
+		for i, peer := range p.NetworkPeers {
+			if peer.Unknown {
+				continue
+			}
+			if peer.FailedPings > 2 {
+				Log(INFO, "Peer timeout. Removing")
+				delete(p.NetworkPeers, i)
+				continue
+			}
+			p.SendTo(peer.PeerHW, msg)
+			peer.FailedPings++
+			p.NetworkPeers[i] = peer
+		}
+		time.Sleep(5 * time.Second)
+	}
+}

lib/tuntap_linux.go

@@ -8,7 +8,8 @@ import (
 )
 
 const (
-	CONFIG_DIR string = "/usr/local/etc"
+	CONFIG_DIR  string = "/usr/local/etc"
+	DEFAULT_MTU string = "1600"
 )
 
 func openDevice(ifPattern string) (*os.File, error) {
@@ -43,6 +44,11 @@ func ConfigureInterface(dev *Interface, ip, mac, device, tool string) error {
 		return err
 	}
 
+	err = SetMTU(dev, device, tool, DEFAULT_MTU)
+	if err != nil {
+		return err
+	}
+
 	// Configure new device
 	err = SetIp(ip, device, tool)
 	if err != nil {
@@ -56,6 +62,16 @@ func ConfigureInterface(dev *Interface, ip, mac, device, tool string) error {
 	return nil
 }
 
+func SetMTU(dev *Interface, device, tool, mtu string) error {
+	setmtu := exec.Command(tool, "link", "set", "dev", device, "mtu", mtu)
+	err := setmtu.Run()
+	if err != nil {
+		Log(ERROR, "Failed to set MTU on device %s: %v", device, err)
+		return err
+	}
+	return nil
+}
+
 func LinkUp(device, tool string) error {
 	linkup := exec.Command(tool, "link", "set", "dev", device, "up")
 	err := linkup.Run()

lib/vars.go

@@ -25,6 +25,7 @@ const (
 	MT_TEST               = 6        // Packet tests established connection
 	MT_PROXY              = 7        // Information about proxy (forwarder)
 	MT_BAD_TUN            = 8        // Notifies about dead tunnel
+	MT_XPING              = 9        // Internal crosspeer ping message
 )
 
 // List of commands used in DHT