[VOL-1553] Improve device graph performance
This update focussed on generating routes using the device graph.
It improves the performance by at least 5 times.
Change-Id: I79bdbca8ea3d134e87848e45140d07ee3831f12c
diff --git a/rw_core/graph/device_graph.go b/rw_core/graph/device_graph.go
index 0ed2748..376df16 100644
--- a/rw_core/graph/device_graph.go
+++ b/rw_core/graph/device_graph.go
@@ -28,7 +28,7 @@
)
func init() {
- log.AddPackage(log.JSON, log.DebugLevel, nil)
+ log.AddPackage(log.JSON, log.WarnLevel, nil)
}
type RouteHop struct {
@@ -42,91 +42,202 @@
Egress uint32
}
+type ofPortLinkToPath struct {
+ link OFPortLink
+ path []RouteHop
+}
+
type GetDeviceFunc func(id string) (*voltha.Device, error)
-func concatDeviceIdPortId(deviceId string, portId uint32) string {
- return fmt.Sprintf("%s:%d", deviceId, portId)
-}
-
-func splitIntoDeviceIdPortId(id string) (string, uint32, error) {
- result := strings.Split(id, ":")
- if len(result) != 2 {
- return "", 0, errors.New(fmt.Sprintf("invalid-id-%s", id))
- }
- if temp, err := strconv.ParseInt(result[1], 10, 32); err != nil {
- return "", 0, errors.New(fmt.Sprintf("invalid-id-%s-%s", id, err.Error()))
- } else {
- return result[0], uint32(temp), nil
- }
-}
-
type DeviceGraph struct {
- GGraph goraph.Graph
- getDevice GetDeviceFunc
- logicalPorts []*voltha.LogicalPort
- RootPorts map[uint32]uint32
- Routes map[OFPortLink][]RouteHop
- graphBuildLock sync.RWMutex
- boundaryPorts sync.Map
+ logicalDeviceId string
+ GGraph goraph.Graph
+ getDeviceFromModel GetDeviceFunc
+ logicalPorts []*voltha.LogicalPort
+ rootPortsString map[string]uint32
+ nonRootPortsString map[string]uint32
+ RootPorts map[uint32]uint32
+ rootPortsLock sync.RWMutex
+ Routes map[OFPortLink][]RouteHop
+ graphBuildLock sync.RWMutex
+ boundaryPorts map[string]uint32
+ boundaryPortsLock sync.RWMutex
+ cachedDevices map[string]*voltha.Device
+ cachedDevicesLock sync.RWMutex
+ devicesAdded map[string]string
+ portsAdded map[string]string
}
-func NewDeviceGraph(getDevice GetDeviceFunc) *DeviceGraph {
+func NewDeviceGraph(logicalDeviceId string, getDevice GetDeviceFunc) *DeviceGraph {
var dg DeviceGraph
+ dg.logicalDeviceId = logicalDeviceId
dg.GGraph = goraph.NewGraph()
- dg.getDevice = getDevice
+ dg.getDeviceFromModel = getDevice
dg.graphBuildLock = sync.RWMutex{}
+ dg.cachedDevicesLock = sync.RWMutex{}
+ dg.rootPortsLock = sync.RWMutex{}
+ dg.devicesAdded = make(map[string]string)
+ dg.portsAdded = make(map[string]string)
+ dg.rootPortsString = make(map[string]uint32)
+ dg.nonRootPortsString = make(map[string]uint32)
+ dg.RootPorts = make(map[uint32]uint32)
+ dg.boundaryPorts = make(map[string]uint32)
+ dg.Routes = make(map[OFPortLink][]RouteHop)
+ dg.cachedDevices = make(map[string]*voltha.Device)
+ log.Debug("new device graph created ...")
return &dg
}
+//IsRootPort returns true if the port is a root port on a logical device
+func (dg *DeviceGraph) IsRootPort(port uint32) bool {
+ dg.rootPortsLock.RLock()
+ defer dg.rootPortsLock.RUnlock()
+ _, exist := dg.RootPorts[port]
+ return exist
+}
+
+//GetDeviceNodeIds retrieves all the nodes in the device graph
+func (dg *DeviceGraph) GetDeviceNodeIds() map[string]string {
+ dg.graphBuildLock.RLock()
+ defer dg.graphBuildLock.RUnlock()
+ nodeIds := make(map[string]string)
+ nodesMap := dg.GGraph.GetNodes()
+ for id, node := range nodesMap {
+ if len(strings.Split(node.String(), ":")) != 2 { // not port node
+ nodeIds[id.String()] = id.String()
+ }
+ }
+ return nodeIds
+}
+
+//ComputeRoutes creates a device graph from the logical ports and then calculates all the routes
+//between the logical ports. This will clear up the graph and routes if there were any.
func (dg *DeviceGraph) ComputeRoutes(lps []*voltha.LogicalPort) {
if dg == nil {
return
}
dg.graphBuildLock.Lock()
defer dg.graphBuildLock.Unlock()
- dg.logicalPorts = lps
- // Set the root ports
- dg.RootPorts = make(map[uint32]uint32)
- for _, lp := range lps {
- if lp.RootPort {
- dg.RootPorts[lp.OfpPort.PortNo] = lp.OfpPort.PortNo
- }
- }
- // set the boundary ports
- dg.boundaryPorts.Range(func(key interface{}, value interface{}) bool {
- dg.boundaryPorts.Delete(key)
- return true
- })
- //dg.boundaryPorts = sync.Map{}
+ // Clear the graph
+ dg.reset()
+
+ dg.logicalPorts = lps
+
+ // Set the root, non-root ports and boundary ports
for _, lp := range lps {
- dg.boundaryPorts.Store(fmt.Sprintf("%s:%d", lp.DeviceId, lp.DevicePortNo), lp.OfpPort.PortNo)
+ portId := concatDeviceIdPortId(lp.DeviceId, lp.DevicePortNo)
+ if lp.RootPort {
+ dg.rootPortsString[portId] = lp.OfpPort.PortNo
+ dg.RootPorts[lp.OfpPort.PortNo] = lp.OfpPort.PortNo
+ } else {
+ dg.nonRootPortsString[portId] = lp.OfpPort.PortNo
+ }
+ dg.boundaryPorts[portId] = lp.OfpPort.PortNo
}
- dg.Routes = make(map[OFPortLink][]RouteHop)
// Build the graph
var device *voltha.Device
- devicesAdded := make(map[string]string)
- portsAdded := make(map[string]string)
for _, logicalPort := range dg.logicalPorts {
device, _ = dg.getDevice(logicalPort.DeviceId)
- dg.GGraph = dg.addDevice(device, dg.GGraph, &devicesAdded, &portsAdded, dg.boundaryPorts)
+ dg.GGraph = dg.addDevice(device, dg.GGraph, &dg.devicesAdded, &dg.portsAdded, dg.boundaryPorts)
}
+
dg.Routes = dg.buildRoutes()
}
+// AddPort adds a port to the graph. If the graph is empty it will just invoke ComputeRoutes function
+func (dg *DeviceGraph) AddPort(lp *voltha.LogicalPort) {
+ // If the graph does not exist invoke ComputeRoutes.
+ if len(dg.boundaryPorts) == 0 {
+ dg.ComputeRoutes([]*voltha.LogicalPort{lp})
+ return
+ }
+
+ dg.graphBuildLock.Lock()
+ defer dg.graphBuildLock.Unlock()
+
+ portId := concatDeviceIdPortId(lp.DeviceId, lp.DevicePortNo)
+
+ // If the port is already part of the boundary ports, do nothing
+ if dg.portExist(portId) {
+ fmt.Println("port exists")
+ return
+ }
+ // Add the device where this port is located to the device graph. If the device is already added then
+ // only the missing port will be added
+ device, _ := dg.getDevice(lp.DeviceId)
+ dg.GGraph = dg.addDevice(device, dg.GGraph, &dg.devicesAdded, &dg.portsAdded, dg.boundaryPorts)
+
+ if lp.RootPort {
+ // Compute the route from this root port to all non-root ports
+ dg.rootPortsString[portId] = lp.OfpPort.PortNo
+ dg.RootPorts[lp.OfpPort.PortNo] = lp.OfpPort.PortNo
+ dg.Routes = dg.buildPathsToAllNonRootPorts(lp)
+ } else {
+ // Compute the route from this port to all root ports
+ dg.nonRootPortsString[portId] = lp.OfpPort.PortNo
+ dg.Routes = dg.buildPathsToAllRootPorts(lp)
+ }
+
+ dg.Print()
+}
+
+func (dg *DeviceGraph) Print() error {
+ if level, err := log.GetPackageLogLevel(); err == nil && level == log.DebugLevel {
+ output := ""
+ routeNumber := 1
+ for k, v := range dg.Routes {
+ key := fmt.Sprintf("LP:%d->LP:%d", k.Ingress, k.Egress)
+ val := ""
+ for _, i := range v {
+ val += fmt.Sprintf("{%d->%s->%d},", i.Ingress, i.DeviceID, i.Egress)
+ }
+ val = val[:len(val)-1]
+ output += fmt.Sprintf("%d:{%s=>%s} ", routeNumber, key, fmt.Sprintf("[%s]", val))
+ routeNumber += 1
+ }
+ log.Debugw("graph_routes", log.Fields{"lDeviceId": dg.logicalDeviceId, "Routes": output})
+ }
+ return nil
+}
+
+//getDevice returns the device either from the local cache (default) or from the model.
+//TODO: Set a cache timeout such that we do not use invalid data. The full device lifecycle should also
+//be taken in consideration
+func (dg *DeviceGraph) getDevice(id string) (*voltha.Device, error) {
+ dg.cachedDevicesLock.RLock()
+ if d, exist := dg.cachedDevices[id]; exist {
+ dg.cachedDevicesLock.RUnlock()
+ //log.Debugw("getDevice - returned from cache", log.Fields{"deviceId": id})
+ return d, nil
+ }
+ dg.cachedDevicesLock.RUnlock()
+ // Not cached
+ if d, err := dg.getDeviceFromModel(id); err != nil {
+ log.Errorw("device-not-found", log.Fields{"deviceId": id, "error": err})
+ return nil, err
+ } else { // cache it
+ dg.cachedDevicesLock.Lock()
+ dg.cachedDevices[id] = d
+ dg.cachedDevicesLock.Unlock()
+ //log.Debugw("getDevice - returned from model", log.Fields{"deviceId": id})
+ return d, nil
+ }
+}
+
+// addDevice adds a device to a device graph and setup edges that represent the device connections to its peers
func (dg *DeviceGraph) addDevice(device *voltha.Device, g goraph.Graph, devicesAdded *map[string]string, portsAdded *map[string]string,
- boundaryPorts sync.Map) goraph.Graph {
+ boundaryPorts map[string]uint32) goraph.Graph {
if device == nil {
return g
}
- if _, exist := (*devicesAdded)[device.Id]; exist {
- return g
+ if _, exist := (*devicesAdded)[device.Id]; !exist {
+ g.AddNode(goraph.NewNode(device.Id))
+ (*devicesAdded)[device.Id] = device.Id
}
- g.AddNode(goraph.NewNode(device.Id))
- (*devicesAdded)[device.Id] = device.Id
var portId string
var peerPortId string
@@ -152,81 +263,201 @@
return g
}
-func (dg *DeviceGraph) IsRootPort(port uint32) bool {
- _, exist := dg.RootPorts[port]
+//portExist returns true if the port ID is already part of the boundary ports map.
+func (dg *DeviceGraph) portExist(id string) bool {
+ dg.boundaryPortsLock.RLock()
+ defer dg.boundaryPortsLock.RUnlock()
+ _, exist := dg.boundaryPorts[id]
return exist
}
-func (dg *DeviceGraph) buildRoutes() map[OFPortLink][]RouteHop {
- var pathIds []goraph.ID
- path := make([]RouteHop, 0)
- paths := make(map[OFPortLink][]RouteHop)
- var err error
- var hop RouteHop
-
- dg.boundaryPorts.Range(func(src, srcPort interface{}) bool {
- source := src.(string)
- sourcePort := srcPort.(uint32)
-
- dg.boundaryPorts.Range(func(dst, dstPort interface{}) bool {
- target := dst.(string)
- targetPort := dstPort.(uint32)
-
- if source == target {
- return true
+// buildPathsToAllRootPorts builds all the paths from the non-root logical port to all root ports
+// on the logical device
+func (dg *DeviceGraph) buildPathsToAllRootPorts(lp *voltha.LogicalPort) map[OFPortLink][]RouteHop {
+ paths := dg.Routes
+ source := concatDeviceIdPortId(lp.DeviceId, lp.DevicePortNo)
+ sourcePort := lp.OfpPort.PortNo
+ ch := make(chan *ofPortLinkToPath)
+ numBuildRequest := 0
+ for target, targetPort := range dg.rootPortsString {
+ go dg.buildRoute(source, target, sourcePort, targetPort, ch)
+ numBuildRequest += 1
+ }
+ responseReceived := 0
+forloop:
+ for {
+ if responseReceived == numBuildRequest {
+ break
+ }
+ select {
+ case res, ok := <-ch:
+ if !ok {
+ log.Debug("channel closed")
+ break forloop
}
- //Ignore NNI - NNI Routes
- if dg.IsRootPort(sourcePort) && dg.IsRootPort(targetPort) {
- return true
+ if res != nil && len(res.path) > 0 {
+ paths[res.link] = res.path
+ paths[OFPortLink{Ingress: res.link.Egress, Egress: res.link.Ingress}] = getReverseRoute(res.path)
}
-
- //Ignore UNI - UNI Routes
- if !dg.IsRootPort(sourcePort) && !dg.IsRootPort(targetPort) {
- return true
- }
-
- if pathIds, _, err = goraph.Dijkstra(dg.GGraph, goraph.StringID(source), goraph.StringID(target)); err != nil {
- log.Errorw("no-path", log.Fields{"source": source, "target": target, "error": err})
- return true
- }
- if len(pathIds)%3 != 0 {
- return true
- }
- var deviceId string
- var ingressPort uint32
- var egressPort uint32
- for i := 0; i < len(pathIds); i = i + 3 {
- if deviceId, ingressPort, err = splitIntoDeviceIdPortId(pathIds[i].String()); err != nil {
- log.Errorw("id-error", log.Fields{"source": source, "target": target, "error": err})
- break
- }
- if _, egressPort, err = splitIntoDeviceIdPortId(pathIds[i+2].String()); err != nil {
- log.Errorw("id-error", log.Fields{"source": source, "target": target, "error": err})
- break
- }
- hop = RouteHop{Ingress: ingressPort, DeviceID: deviceId, Egress: egressPort}
- path = append(path, hop)
- }
- tmp := make([]RouteHop, len(path))
- copy(tmp, path)
- path = nil
- paths[OFPortLink{Ingress: sourcePort, Egress: targetPort}] = tmp
- return true
- })
- return true
- })
+ }
+ responseReceived += 1
+ }
return paths
}
-func (dg *DeviceGraph) GetDeviceNodeIds() map[string]string {
- dg.graphBuildLock.RLock()
- defer dg.graphBuildLock.RUnlock()
- nodeIds := make(map[string]string)
- nodesMap := dg.GGraph.GetNodes()
- for id, node := range nodesMap {
- if len(strings.Split(node.String(), ":")) != 2 { // not port node
- nodeIds[id.String()] = id.String()
+// buildPathsToAllNonRootPorts builds all the paths from the root logical port to all non-root ports
+// on the logical device
+func (dg *DeviceGraph) buildPathsToAllNonRootPorts(lp *voltha.LogicalPort) map[OFPortLink][]RouteHop {
+ paths := dg.Routes
+ source := concatDeviceIdPortId(lp.DeviceId, lp.DevicePortNo)
+ sourcePort := lp.OfpPort.PortNo
+ ch := make(chan *ofPortLinkToPath)
+ numBuildRequest := 0
+ for target, targetPort := range dg.nonRootPortsString {
+ go dg.buildRoute(source, target, sourcePort, targetPort, ch)
+ numBuildRequest += 1
+ }
+ responseReceived := 0
+forloop:
+ for {
+ if responseReceived == numBuildRequest {
+ break
+ }
+ select {
+ case res, ok := <-ch:
+ if !ok {
+ log.Debug("channel closed")
+ break forloop
+ }
+ if res != nil && len(res.path) > 0 {
+ paths[res.link] = res.path
+ paths[OFPortLink{Ingress: res.link.Egress, Egress: res.link.Ingress}] = getReverseRoute(res.path)
+ }
+ }
+ responseReceived += 1
+ }
+ return paths
+}
+
+//buildRoute builds a route between a source and a target logical port
+func (dg *DeviceGraph) buildRoute(sourceId, targetId string, sourcePort, targetPort uint32, ch chan *ofPortLinkToPath) {
+ var pathIds []goraph.ID
+ path := make([]RouteHop, 0)
+ var err error
+ var hop RouteHop
+ var result *ofPortLinkToPath
+
+ if sourceId == targetId {
+ ch <- result
+ return
+ }
+ //Ignore Root - Root Routes
+ if dg.IsRootPort(sourcePort) && dg.IsRootPort(targetPort) {
+ ch <- result
+ return
+ }
+
+ //Ignore non-Root - non-Root Routes
+ if !dg.IsRootPort(sourcePort) && !dg.IsRootPort(targetPort) {
+ ch <- result
+ return
+ }
+
+ if pathIds, _, err = goraph.Dijkstra(dg.GGraph, goraph.StringID(sourceId), goraph.StringID(targetId)); err != nil {
+ log.Errorw("no-path", log.Fields{"sourceId": sourceId, "targetId": targetId, "error": err})
+ ch <- result
+ return
+ }
+ if len(pathIds)%3 != 0 {
+ ch <- result
+ return
+ }
+ var deviceId string
+ var ingressPort uint32
+ var egressPort uint32
+ for i := 0; i < len(pathIds); i = i + 3 {
+ if deviceId, ingressPort, err = splitIntoDeviceIdPortId(pathIds[i].String()); err != nil {
+ log.Errorw("id-error", log.Fields{"sourceId": sourceId, "targetId": targetId, "error": err})
+ break
+ }
+ if _, egressPort, err = splitIntoDeviceIdPortId(pathIds[i+2].String()); err != nil {
+ log.Errorw("id-error", log.Fields{"sourceId": sourceId, "targetId": targetId, "error": err})
+ break
+ }
+ hop = RouteHop{Ingress: ingressPort, DeviceID: deviceId, Egress: egressPort}
+ path = append(path, hop)
+ }
+ result = &ofPortLinkToPath{link: OFPortLink{Ingress: sourcePort, Egress: targetPort}, path: path}
+ ch <- result
+}
+
+//buildRoutes build all routes between all the ports on the logical device
+func (dg *DeviceGraph) buildRoutes() map[OFPortLink][]RouteHop {
+ paths := make(map[OFPortLink][]RouteHop)
+ ch := make(chan *ofPortLinkToPath)
+ numBuildRequest := 0
+ for source, sourcePort := range dg.boundaryPorts {
+ for target, targetPort := range dg.boundaryPorts {
+ go dg.buildRoute(source, target, sourcePort, targetPort, ch)
+ numBuildRequest += 1
}
}
- return nodeIds
+ responseReceived := 0
+forloop:
+ for {
+ if responseReceived == numBuildRequest {
+ break
+ }
+ select {
+ case res, ok := <-ch:
+ if !ok {
+ log.Debug("channel closed")
+ break forloop
+ }
+ if res != nil && len(res.path) > 0 {
+ paths[res.link] = res.path
+ }
+ }
+ responseReceived += 1
+ }
+ return paths
+}
+
+// reset cleans up the device graph
+func (dg *DeviceGraph) reset() {
+ dg.devicesAdded = make(map[string]string)
+ dg.portsAdded = make(map[string]string)
+ dg.rootPortsString = make(map[string]uint32)
+ dg.nonRootPortsString = make(map[string]uint32)
+ dg.RootPorts = make(map[uint32]uint32)
+ dg.boundaryPorts = make(map[string]uint32)
+ dg.Routes = make(map[OFPortLink][]RouteHop)
+ dg.cachedDevices = make(map[string]*voltha.Device)
+}
+
+//concatDeviceIdPortId formats a portid using the device id and the port number
+func concatDeviceIdPortId(deviceId string, portNo uint32) string {
+ return fmt.Sprintf("%s:%d", deviceId, portNo)
+}
+
+// splitIntoDeviceIdPortId extracts the device id and port number from the portId
+func splitIntoDeviceIdPortId(id string) (string, uint32, error) {
+ result := strings.Split(id, ":")
+ if len(result) != 2 {
+ return "", 0, errors.New(fmt.Sprintf("invalid-id-%s", id))
+ }
+ if temp, err := strconv.ParseInt(result[1], 10, 32); err != nil {
+ return "", 0, errors.New(fmt.Sprintf("invalid-id-%s-%s", id, err.Error()))
+ } else {
+ return result[0], uint32(temp), nil
+ }
+}
+
+//getReverseRoute returns the reverse of the route in param
+func getReverseRoute(route []RouteHop) []RouteHop {
+ reverse := make([]RouteHop, len(route))
+ for i, j := 0, len(route)-1; i < j; i, j = i+1, j-1 {
+ reverse[i], reverse[j] = route[j], route[i]
+ }
+ return reverse
}