[VOL-2164] Update rw-core to use the Async Kafka API
This commit consists of the following:
1. Process per-device requests in the Core in the order they are
received. If there are lots of requests on a given device then
there will be some latencies introduced due to ordering. With
recent changes in the model along with keeping the request lock
to a minimal then these latencies are reduced. Testing did not
show and noticeable latencies.
2) Keep the request lock from the moment a request started
processing to the moment that request is sent to kafka (when
applicable). Adapter responses are received and processed
asynchronously. Therefore, an adapter can takes all the time it
needs to process a transaction. The Core still has a context
with timeout (configurable) to cater for cases where the adapter
does not return a response.
3) Adapter requests are processed to completion before sending a
reponse back to the adapter. Previously, in some cases, a
separate go routine was created to process the request and a
successful response is sent to the adapter. Now if the request
fails then the adapter will receive an error. The adapter
requests for a given device are therefore processed in the
order they are received.
4) Some changes are made when retrieving a handler to execute
a device state transition. This was necessary as there was some
transition overlap found.
Update after multiple reviews.
Change-Id: I55a189efec1549a662f2d71e18e6eca9015a3a17
diff --git a/rw_core/core/device_state_transitions.go b/rw_core/core/device_state_transitions.go
index 9b4acf3..69dfd78 100644
--- a/rw_core/core/device_state_transitions.go
+++ b/rw_core/core/device_state_transitions.go
@@ -32,6 +32,35 @@
any DeviceType = 2
)
+type MatchResult uint8
+
+const (
+ noMatch MatchResult = iota // current state has not match in the transition table
+ currWildcardMatch // current state matches the wildcard *_UNKNOWN state in the transition table
+ currStateOnlyMatch // current state matches the current state and previous state matches the wildcard in the transition table
+ currPrevStateMatch // both current and previous states match in the transition table
+)
+
+// match is used to keep the current match states
+type match struct {
+ admin, oper, conn MatchResult
+}
+
+// toInt returns an integer representing the matching level of the match (the larger the number the better)
+func (m *match) toInt() int {
+ return int(m.admin<<4 | m.oper<<2 | m.conn)
+}
+
+// isExactMatch returns true if match is an exact match
+func (m *match) isExactMatch() bool {
+ return m.admin == currPrevStateMatch && m.oper == currPrevStateMatch && m.conn == currPrevStateMatch
+}
+
+// isBetterMatch returns true if newMatch is a worse match
+func (m *match) isBetterMatch(newMatch *match) bool {
+ return m.toInt() > newMatch.toInt()
+}
+
// DeviceState has admin, operational and connection status of device
type DeviceState struct {
Admin voltha.AdminState_Types
@@ -39,8 +68,8 @@
Operational voltha.OperStatus_Types
}
-// TransitionHandler function type which takes device as input parameter
-type TransitionHandler func(context.Context, *voltha.Device) error
+// TransitionHandler function type which takes the current and previous device info as input parameter
+type TransitionHandler func(ctx context.Context, curr *voltha.Device, prev *voltha.Device) error
// Transition represent transition related attributes
type Transition struct {
@@ -61,7 +90,8 @@
var transitionMap TransitionMap
transitionMap.dMgr = dMgr
transitionMap.transitions = make([]Transition, 0)
- transitionMap.transitions = append(transitionMap.transitions,
+ transitionMap.transitions = append(
+ transitionMap.transitions,
Transition{
deviceType: parent,
previousState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVATING},
@@ -71,12 +101,24 @@
Transition{
deviceType: child,
previousState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_DISCOVERED},
+ currentState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVATING},
+ handlers: []TransitionHandler{}})
+ transitionMap.transitions = append(transitionMap.transitions,
+ Transition{
+ deviceType: child,
+ previousState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_DISCOVERED},
currentState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVE},
handlers: []TransitionHandler{dMgr.SetupUNILogicalPorts}})
transitionMap.transitions = append(transitionMap.transitions,
Transition{
deviceType: child,
previousState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVATING},
+ currentState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_DISCOVERED},
+ handlers: []TransitionHandler{}})
+ transitionMap.transitions = append(transitionMap.transitions,
+ Transition{
+ deviceType: child,
+ previousState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVATING},
currentState: DeviceState{Admin: voltha.AdminState_ENABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_ACTIVE},
handlers: []TransitionHandler{dMgr.SetupUNILogicalPorts}})
transitionMap.transitions = append(transitionMap.transitions,
@@ -88,13 +130,13 @@
transitionMap.transitions = append(transitionMap.transitions,
Transition{
deviceType: parent,
- previousState: DeviceState{Admin: voltha.AdminState_DISABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
+ previousState: DeviceState{Admin: voltha.AdminState_UNKNOWN, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
currentState: DeviceState{Admin: voltha.AdminState_DELETED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
handlers: []TransitionHandler{dMgr.DisableAllChildDevices, dMgr.DeleteAllUNILogicalPorts, dMgr.DeleteAllChildDevices, dMgr.DeleteLogicalDevice, dMgr.RunPostDeviceDelete}})
transitionMap.transitions = append(transitionMap.transitions,
Transition{
deviceType: child,
- previousState: DeviceState{Admin: voltha.AdminState_DISABLED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
+ previousState: DeviceState{Admin: voltha.AdminState_UNKNOWN, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
currentState: DeviceState{Admin: voltha.AdminState_DELETED, Connection: voltha.ConnectStatus_UNKNOWN, Operational: voltha.OperStatus_UNKNOWN},
handlers: []TransitionHandler{dMgr.ChildDeviceLost, dMgr.DeleteLogicalPorts, dMgr.RunPostDeviceDelete}})
transitionMap.transitions = append(transitionMap.transitions,
@@ -178,37 +220,57 @@
}
// isMatched matches a state transition. It returns whether there is a match and if there is whether it is an exact match
-func getHandler(previous *DeviceState, current *DeviceState, transition *Transition) ([]TransitionHandler, bool) {
-
+func getHandler(previous *DeviceState, current *DeviceState, transition *Transition) ([]TransitionHandler, *match) {
+ m := &match{}
// Do we have an exact match?
if *previous == transition.previousState && *current == transition.currentState {
- return transition.handlers, true
+ return transition.handlers, &match{admin: currPrevStateMatch, oper: currPrevStateMatch, conn: currPrevStateMatch}
}
- // Admin states must match
- if previous.Admin != transition.previousState.Admin || current.Admin != transition.currentState.Admin {
- return nil, false
+ // Do we have Admin state match?
+ if current.Admin == transition.currentState.Admin && transition.currentState.Admin != voltha.AdminState_UNKNOWN {
+ if previous.Admin == transition.previousState.Admin {
+ m.admin = currPrevStateMatch
+ } else if transition.previousState.Admin == voltha.AdminState_UNKNOWN {
+ m.admin = currStateOnlyMatch
+ }
+ } else if current.Admin == transition.currentState.Admin && transition.currentState.Admin == voltha.AdminState_UNKNOWN {
+ if previous.Admin == transition.previousState.Admin || transition.previousState.Admin == voltha.AdminState_UNKNOWN {
+ m.admin = currWildcardMatch
+ }
+ }
+ if m.admin == noMatch {
+ // invalid transition - need to match on current admin state
+ return nil, m
}
- // If the admin state changed then prioritize it first
- if previous.Admin != current.Admin {
- if previous.Admin == transition.previousState.Admin && current.Admin == transition.currentState.Admin {
- return transition.handlers, false
+ // Do we have an operational state match?
+ if current.Operational == transition.currentState.Operational && transition.previousState.Operational != voltha.OperStatus_UNKNOWN {
+ if previous.Operational == transition.previousState.Operational || transition.previousState.Operational == voltha.OperStatus_UNKNOWN {
+ m.oper = currPrevStateMatch
+ } else {
+ m.oper = currStateOnlyMatch
+ }
+ } else if current.Operational == transition.currentState.Operational && transition.previousState.Operational == voltha.OperStatus_UNKNOWN {
+ if previous.Operational == transition.previousState.Operational || transition.previousState.Operational == voltha.OperStatus_UNKNOWN {
+ m.oper = currWildcardMatch
}
}
- // If the operational state changed then prioritize it in second position
- if previous.Operational != current.Operational {
- if previous.Operational == transition.previousState.Operational && current.Operational == transition.currentState.Operational {
- return transition.handlers, false
+
+ // Do we have an connection state match?
+ if current.Connection == transition.currentState.Connection && transition.previousState.Connection != voltha.ConnectStatus_UNKNOWN {
+ if previous.Connection == transition.previousState.Connection || transition.previousState.Connection == voltha.ConnectStatus_UNKNOWN {
+ m.conn = currPrevStateMatch
+ } else {
+ m.conn = currStateOnlyMatch
+ }
+ } else if current.Connection == transition.currentState.Connection && transition.previousState.Connection == voltha.ConnectStatus_UNKNOWN {
+ if previous.Connection == transition.previousState.Connection || transition.previousState.Connection == voltha.ConnectStatus_UNKNOWN {
+ m.conn = currWildcardMatch
}
}
- // If the connection state changed then prioritize it in third position
- if previous.Connection != current.Connection {
- if previous.Connection == transition.previousState.Connection && current.Connection == transition.currentState.Connection {
- return transition.handlers, false
- }
- }
- return nil, false
+
+ return transition.handlers, m
}
// GetTransitionHandler returns transition handler
@@ -216,6 +278,12 @@
//1. Get the previous and current set of states
pState := getDeviceStates(pDevice)
cState := getDeviceStates(cDevice)
+
+ // Do nothing is there are no states change
+ if *pState == *cState {
+ return nil
+ }
+
//log.Infow("DeviceType", log.Fields{"device": pDevice})
deviceType := parent
if !pDevice.Root {
@@ -227,22 +295,20 @@
//2. Go over transition array to get the right transition
var currentMatch []TransitionHandler
var tempHandler []TransitionHandler
- var exactStateMatch bool
- var stateMatchFound bool
+ var m *match
+ bestMatch := &match{}
for _, aTransition := range tMap.transitions {
// consider transition only if it matches deviceType or is a wild card - any
if aTransition.deviceType != deviceType && aTransition.deviceType != any {
continue
}
- tempHandler, exactStateMatch = getHandler(pState, cState, &aTransition)
+ tempHandler, m = getHandler(pState, cState, &aTransition)
if tempHandler != nil {
- if exactStateMatch && aTransition.deviceType == deviceType {
+ if m.isExactMatch() && aTransition.deviceType == deviceType {
return tempHandler
- } else if exactStateMatch {
+ } else if m.isExactMatch() || m.isBetterMatch(bestMatch) {
currentMatch = tempHandler
- stateMatchFound = true
- } else if currentMatch == nil && !stateMatchFound {
- currentMatch = tempHandler
+ bestMatch = m
}
}
}