server: properly handle peer configured later

- send adj-in in all the existing peers to a peer configured after gobgpd started.
- peer sends update message information (path) to another peer directly instead of via gobgpd (this improves the performance)

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

3 files changed, 155 insertions, 72 deletions

diff --git a/server/peer.go b/server/peer.go
index 6f18f4ec..b65edc11 100644
--- a/server/peer.go
+++ b/server/peer.go
@@ -27,6 +27,19 @@ import (
 	"time"
 )
 
+type peerMsgType int
+
+const (
+	_ peerMsgType = iota
+	PEER_MSG_PATH
+	PEER_MSG_PEER_DOWN
+)
+
+type peerMsg struct {
+	msgType peerMsgType
+	msgData interface{}
+}
+
 type Peer struct {
 	t              tomb.Tomb
 	globalConfig   config.GlobalType
@@ -34,8 +47,8 @@ type Peer struct {
 	acceptedConnCh chan *net.TCPConn
 	incoming       chan *fsmMsg
 	outgoing       chan *bgp.BGPMessage
-	inEventCh      chan *message
-	outEventCh     chan *message
+	serverMsgCh    chan *serverMsg
+	peerMsgCh      chan *peerMsg
 	fsm            *FSM
 	adjRib         *table.AdjRib
 	// peer and rib are always not one-to-one so should not be
@@ -45,19 +58,24 @@ type Peer struct {
 	rf       bgp.RouteFamily
 	capMap   map[bgp.BGPCapabilityCode]bgp.ParameterCapabilityInterface
 	peerInfo *table.PeerInfo
+	siblings map[string]*serverMsgDataPeer
 }
 
-func NewPeer(g config.GlobalType, peer config.NeighborType, outEventCh chan *message) *Peer {
+func NewPeer(g config.GlobalType, peer config.NeighborType, serverMsgCh chan *serverMsg, peerMsgCh chan *peerMsg, peerList []*serverMsgDataPeer) *Peer {
 	p := &Peer{
 		globalConfig:   g,
 		peerConfig:     peer,
 		acceptedConnCh: make(chan *net.TCPConn),
 		incoming:       make(chan *fsmMsg, 4096),
 		outgoing:       make(chan *bgp.BGPMessage, 4096),
-		inEventCh:      make(chan *message, 4096),
-		outEventCh:     outEventCh,
+		serverMsgCh:    serverMsgCh,
+		peerMsgCh:      peerMsgCh,
 		capMap:         make(map[bgp.BGPCapabilityCode]bgp.ParameterCapabilityInterface),
 	}
+	p.siblings = make(map[string]*serverMsgDataPeer)
+	for _, s := range peerList {
+		p.siblings[s.address.String()] = s
+	}
 	p.fsm = NewFSM(&g, &peer, p.acceptedConnCh, p.incoming, p.outgoing)
 	peer.BgpNeighborCommonState.State = uint32(bgp.BGP_FSM_IDLE)
 	if peer.NeighborAddress.To4() != nil {
@@ -70,6 +88,7 @@ func NewPeer(g config.GlobalType, peer config.NeighborType, outEventCh chan *mes
 		VersionNum: 1,
 		LocalID:    g.RouterId,
 		RF:         p.rf,
+		Address:    peer.NeighborAddress,
 	}
 	p.adjRib = table.NewAdjRib()
 	p.rib = table.NewTableManager()
@@ -108,7 +127,14 @@ func (peer *Peer) handleBGPmessage(m *bgp.BGPMessage) {
 			return
 		}
 		peer.adjRib.UpdateIn(pathList)
-		peer.sendToHub("", PEER_MSG_PATH, pathList)
+		pm := &peerMsg{
+			msgType: PEER_MSG_PATH,
+			msgData: pathList,
+		}
+		for _, s := range peer.siblings {
+			// TODO: check rf
+			s.peerMsgCh <- pm
+		}
 	}
 }
 
@@ -140,7 +166,7 @@ func (peer *Peer) handleREST(restReq *api.RestRequest) {
 	close(restReq.ResponseCh)
 }
 
-func (peer *Peer) handlePeermessage(m *message) {
+func (peer *Peer) handlePeerMsg(m *peerMsg) {
 	sendpath := func(pList []table.Path, wList []table.Path) {
 		pathList := append([]table.Path(nil), pList...)
 		pathList = append(pathList, wList...)
@@ -154,15 +180,46 @@ func (peer *Peer) handlePeermessage(m *message) {
 		peer.sendMessages(table.CreateUpdateMsgFromPaths(pathList))
 	}
 
-	switch m.event {
+	switch m.msgType {
 	case PEER_MSG_PATH:
-		pList, wList, _ := peer.rib.ProcessPaths(m.data.([]table.Path))
+		pList, wList, _ := peer.rib.ProcessPaths(m.msgData.([]table.Path))
 		sendpath(pList, wList)
-	case PEER_MSG_DOWN:
-		pList, wList, _ := peer.rib.DeletePathsforPeer(m.data.(*table.PeerInfo))
+	case PEER_MSG_PEER_DOWN:
+		pList, wList, _ := peer.rib.DeletePathsforPeer(m.msgData.(*table.PeerInfo))
 		sendpath(pList, wList)
-	case PEER_MSG_REST:
-		peer.handleREST(m.data.(*api.RestRequest))
+	}
+}
+
+func (peer *Peer) handleServerMsg(m *serverMsg) {
+	switch m.msgType {
+	case SRV_MSG_PEER_ADDED:
+		d := m.msgData.(*serverMsgDataPeer)
+		peer.siblings[d.address.String()] = d
+		pathList := peer.adjRib.GetInPathList(d.rf)
+		if len(pathList) == 0 {
+			return
+		}
+		pm := &peerMsg{
+			msgType: PEER_MSG_PATH,
+			msgData: pathList,
+		}
+		for _, s := range peer.siblings {
+			// TODO: check rf
+			s.peerMsgCh <- pm
+		}
+	case SRV_MSG_PEER_DELETED:
+		d := m.msgData.(*table.PeerInfo)
+		_, found := peer.siblings[d.Address.String()]
+		if found {
+			delete(peer.siblings, d.Address.String())
+			// TODO: do the same that PEER_MSG_PEER_DOWN handler
+		} else {
+			log.Warning("can not find peer: ", d.Address.String())
+		}
+	case SRV_MSG_API:
+		peer.handleREST(m.msgData.(*api.RestRequest))
+	default:
+		log.Fatal("unknown server msg type ", m.msgType)
 	}
 }
 
@@ -200,13 +257,21 @@ func (peer *Peer) loop() error {
 					}
 					if oldState == bgp.BGP_FSM_ESTABLISHED {
 						peer.fsm.peerConfig.BgpNeighborCommonState.Uptime = time.Time{}
-						peer.sendToHub("", PEER_MSG_DOWN, peer.peerInfo)
+						pm := &peerMsg{
+							msgType: PEER_MSG_PEER_DOWN,
+							msgData: peer.peerInfo,
+						}
+						for _, s := range peer.siblings {
+							s.peerMsgCh <- pm
+						}
 					}
 				case FSM_MSG_BGP_MESSAGE:
 					peer.handleBGPmessage(e.MsgData.(*bgp.BGPMessage))
 				}
-			case m := <-peer.inEventCh:
-				peer.handlePeermessage(m)
+			case m := <-peer.serverMsgCh:
+				peer.handleServerMsg(m)
+			case m := <-peer.peerMsgCh:
+				peer.handlePeerMsg(m)
 			}
 		}
 	}
@@ -221,19 +286,6 @@ func (peer *Peer) PassConn(conn *net.TCPConn) {
 	peer.acceptedConnCh <- conn
 }
 
-func (peer *Peer) SendMessage(msg *message) {
-	peer.inEventCh <- msg
-}
-
-func (peer *Peer) sendToHub(destination string, event int, data interface{}) {
-	peer.outEventCh <- &message{
-		src:   peer.peerConfig.NeighborAddress.String(),
-		dst:   destination,
-		event: event,
-		data:  data,
-	}
-}
-
 func (peer *Peer) MarshalJSON() ([]byte, error) {
 
 	f := peer.fsm
diff --git a/server/server.go b/server/server.go
index 99ea974c..3275e5dc 100644
--- a/server/server.go
+++ b/server/server.go
@@ -21,25 +21,38 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/osrg/gobgp/api"
 	"github.com/osrg/gobgp/config"
+	"github.com/osrg/gobgp/packet"
 	"net"
 	"os"
 	"strconv"
 	"strings"
 )
 
+type serverMsgType int
+
 const (
-	_ = iota
-	PEER_MSG_NEW
-	PEER_MSG_PATH
-	PEER_MSG_DOWN
-	PEER_MSG_REST //hacky, fix later
+	_ serverMsgType = iota
+	SRV_MSG_PEER_ADDED
+	SRV_MSG_PEER_DELETED
+	SRV_MSG_API
 )
 
-type message struct {
-	src   string
-	dst   string
-	event int
-	data  interface{}
+type serverMsg struct {
+	msgType serverMsgType
+	msgData interface{}
+}
+
+type serverMsgDataPeer struct {
+	peerMsgCh chan *peerMsg
+	address   net.IP
+	rf        bgp.RouteFamily
+}
+
+type peerMapInfo struct {
+	peer        *Peer
+	serverMsgCh chan *serverMsg
+	peerMsgCh   chan *peerMsg
+	peerMsgData *serverMsgDataPeer
 }
 
 type BgpServer struct {
@@ -49,7 +62,7 @@ type BgpServer struct {
 	deletedPeerCh chan config.NeighborType
 	RestReqCh     chan *api.RestRequest
 	listenPort    int
-	peerMap       map[string]*Peer
+	peerMap       map[string]peerMapInfo
 }
 
 func NewBgpServer(port int) *BgpServer {
@@ -111,8 +124,7 @@ func (server *BgpServer) Serve() {
 		return f
 	}
 
-	server.peerMap = make(map[string]*Peer)
-	broadcastCh := make(chan *message)
+	server.peerMap = make(map[string]peerMapInfo)
 	for {
 		select {
 		case conn := <-acceptCh:
@@ -123,10 +135,10 @@ func (server *BgpServer) Serve() {
 				idx := strings.LastIndex(addrPort, ":")
 				return addrPort[1 : idx-1]
 			}(conn.RemoteAddr().String())
-			peer, found := server.peerMap[remoteAddr]
+			info, found := server.peerMap[remoteAddr]
 			if found {
 				log.Info("accepted a new passive connection from ", remoteAddr)
-				peer.PassConn(conn)
+				info.peer.PassConn(conn)
 			} else {
 				log.Info("can't find configuration for a new passive connection from ", remoteAddr)
 				conn.Close()
@@ -135,29 +147,59 @@ func (server *BgpServer) Serve() {
 			addr := peer.NeighborAddress.String()
 			f := listenFile(peer.NeighborAddress)
 			SetTcpMD5SigSockopts(int(f.Fd()), addr, peer.AuthPassword)
-			p := NewPeer(server.bgpConfig.Global, peer, broadcastCh)
-			server.peerMap[peer.NeighborAddress.String()] = p
+			sch := make(chan *serverMsg, 8)
+			pch := make(chan *peerMsg, 4096)
+			l := make([]*serverMsgDataPeer, len(server.peerMap))
+			i := 0
+			for _, v := range server.peerMap {
+				l[i] = v.peerMsgData
+				i++
+			}
+			p := NewPeer(server.bgpConfig.Global, peer, sch, pch, l)
+			d := &serverMsgDataPeer{
+				address:   peer.NeighborAddress,
+				peerMsgCh: pch,
+				rf:        p.peerInfo.RF,
+			}
+			msg := &serverMsg{
+				msgType: SRV_MSG_PEER_ADDED,
+				msgData: d,
+			}
+			sendServerMsgToAll(server.peerMap, msg)
+			server.peerMap[peer.NeighborAddress.String()] = peerMapInfo{
+				peer:        p,
+				serverMsgCh: sch,
+				peerMsgData: d,
+			}
 		case peer := <-server.deletedPeerCh:
 			addr := peer.NeighborAddress.String()
 			f := listenFile(peer.NeighborAddress)
 			SetTcpMD5SigSockopts(int(f.Fd()), addr, "")
-			p, found := server.peerMap[addr]
+			info, found := server.peerMap[addr]
 			if found {
 				log.Info("Delete a peer configuration for ", addr)
-				p.Stop()
+				info.peer.Stop()
 				delete(server.peerMap, addr)
+				msg := &serverMsg{
+					msgType: SRV_MSG_PEER_DELETED,
+					msgData: info.peer.peerInfo,
+				}
+				sendServerMsgToAll(server.peerMap, msg)
 			} else {
 				log.Info("Can't delete a peer configuration for ", addr)
 			}
 		case restReq := <-server.RestReqCh:
 			server.handleRest(restReq)
-
-		case msg := <-broadcastCh:
-			server.broadcast(msg)
 		}
 	}
 }
 
+func sendServerMsgToAll(peerMap map[string]peerMapInfo, msg *serverMsg) {
+	for _, info := range peerMap {
+		info.serverMsgCh <- msg
+	}
+}
+
 func (server *BgpServer) SetGlobalType(g config.GlobalType) {
 	server.globalTypeCh <- g
 }
@@ -170,25 +212,13 @@ func (server *BgpServer) PeerDelete(peer config.NeighborType) {
 	server.deletedPeerCh <- peer
 }
 
-func (server *BgpServer) broadcast(msg *message) {
-	for key := range server.peerMap {
-		if key == msg.src {
-			continue
-		}
-		if msg.dst == "" || msg.dst == key {
-			peer := server.peerMap[key]
-			peer.SendMessage(msg)
-		}
-	}
-}
-
 func (server *BgpServer) handleRest(restReq *api.RestRequest) {
 	switch restReq.RequestType {
 	case api.REQ_NEIGHBORS:
 		result := &api.RestResponse{}
 		peerList := make([]*Peer, 0)
-		for _, peer := range server.peerMap {
-			peerList = append(peerList, peer)
+		for _, info := range server.peerMap {
+			peerList = append(peerList, info.peer)
 		}
 		j, _ := json.Marshal(peerList)
 		result.Data = j
@@ -199,9 +229,9 @@ func (server *BgpServer) handleRest(restReq *api.RestRequest) {
 
 		remoteAddr := restReq.RemoteAddr
 		result := &api.RestResponse{}
-		peer, found := server.peerMap[remoteAddr]
+		info, found := server.peerMap[remoteAddr]
 		if found {
-			j, _ := json.Marshal(peer)
+			j, _ := json.Marshal(info.peer)
 			result.Data = j
 		} else {
 			result.ResponseErr = fmt.Errorf("Neighbor that has %v does not exist.", remoteAddr)
@@ -211,13 +241,13 @@ func (server *BgpServer) handleRest(restReq *api.RestRequest) {
 	case api.REQ_LOCAL_RIB:
 		remoteAddr := restReq.RemoteAddr
 		result := &api.RestResponse{}
-		peer, found := server.peerMap[remoteAddr]
+		info, found := server.peerMap[remoteAddr]
 		if found {
-			msg := message{
-				event: PEER_MSG_REST,
-				data:  restReq,
+			msg := &serverMsg{
+				msgType: SRV_MSG_API,
+				msgData: restReq,
 			}
-			peer.SendMessage(&msg)
+			info.peer.serverMsgCh <- msg
 		} else {
 			result.ResponseErr = fmt.Errorf("Neighbor that has %v does not exist.", remoteAddr)
 			restReq.ResponseCh <- result
diff --git a/table/destination.go b/table/destination.go
index 06bab95c..73cae66c 100644
--- a/table/destination.go
+++ b/table/destination.go
@@ -46,6 +46,7 @@ type PeerInfo struct {
 	VersionNum int
 	LocalID    net.IP
 	RF         bgp.RouteFamily
+	Address    net.IP
 }
 
 type Destination interface {