test/ping/probing/mpconn.go

package probing

import (
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
	"trial/ping/probing/icmp"

	"git.wecise.com/wecise/common/logger"
)

var connm sync.Mutex
var conns = map[string]*mpingconn{}

func MPConn(ipv4 bool, protocol string) *mpingconn {
	key := protocol
	if ipv4 {
		key += "4"
	} else {
		key += "6"
	}
	connm.Lock()
	mp := conns[key]
	if mp == nil {
		mp = newMPConn(ipv4, protocol)
		conns[key] = mp
	}
	connm.Unlock()
	return mp
}

type mpingconn struct {
	*icmp.MPacketConn
	mutex      sync.Mutex
	pingidinfo map[int]*mpinfo
}

type mpinfo struct {
	host        string
	ipaddr      *net.IPAddr
	id          int
	lastseq     int
	size        int
	timeout     time.Duration
	seqpkt      map[int]*icmp.Packet
	OnSend      func(*icmp.Packet)
	OnRecv      func(*icmp.Packet)
	OnRecvDup   func(*icmp.Packet)
	OnRecvError func(error)
}

var pingid int32
var ETIMEDOUT error = fmt.Errorf("timeout")

func newPingInfo(host string, ipaddr *net.IPAddr, size int, timeout time.Duration) *mpinfo {
	return &mpinfo{
		host:    host,
		ipaddr:  ipaddr,
		id:      int(atomic.AddInt32(&pingid, 1)),
		seqpkt:  make(map[int]*icmp.Packet),
		size:    size,
		timeout: timeout,
	}
}

func newMPConn(ipv4 bool, protocol string) *mpingconn {
	mpconn := &mpingconn{
		MPacketConn: &icmp.MPacketConn{
			IPV4:     ipv4,
			Protocol: protocol,
			Source:   "",
			Backlog:  10,
			TTL:      64,
		},
		pingidinfo: make(map[int]*mpinfo),
	}
	mpconn.MPacketConn.OnRecvPacket = mpconn.OnRecvPacket
	mpconn.MPacketConn.OnRecvError = mpconn.OnRecvError
	return mpconn
}

func (p *mpingconn) Listen() error {
	err := p.MPacketConn.Listen()
	if err != nil {
		return err
	}
	return nil
}

func (p *mpingconn) Close() error {
	err := p.MPacketConn.Close()
	if err != nil {
		return err
	}
	return nil
}

func (p *mpingconn) OnRecvPacket(recvpkt *icmp.Packet) {

	// fmt.Println("recv", recvpkt)

	p.mutex.Lock()
	defer p.mutex.Unlock()
	pinfo := p.pingidinfo[recvpkt.ID]
	if pinfo == nil {
		return
	}
	_, inflight := pinfo.seqpkt[recvpkt.Seq]
	if inflight {
		// remove it from the list of sequences we're waiting for so we don't get duplicates.
		delete(pinfo.seqpkt, recvpkt.Seq)
		if pinfo.OnRecv != nil {
			go pinfo.OnRecv(recvpkt)
		}
		return
	}
	if pinfo.OnRecvDup != nil {
		go pinfo.OnRecvDup(recvpkt)
	}
}

func (p *mpingconn) OnRecvError(err error) {
	logger.Error(err)
}

func (p *mpingconn) Ping(pinfo *mpinfo) error {
	p.mutex.Lock()
	if _, has := p.pingidinfo[pinfo.id]; !has {
		p.pingidinfo[pinfo.id] = pinfo
	}
	seq := pinfo.lastseq
	pinfo.lastseq++
	outpkt := &icmp.Packet{
		IPAddr: pinfo.ipaddr,
		ID:     pinfo.id,
		Seq:    seq,
		Nbytes: pinfo.size,
	}
	pinfo.seqpkt[seq] = outpkt
	p.mutex.Unlock()

	err := p.SendPacket(outpkt)
	if err != nil {
		return err
	}
	if pinfo.OnSend != nil {
		pinfo.OnSend(outpkt)
	}

	// fmt.Println("sent", outpkt)

	return nil
}