[VOL-1512] Set device ownership
This commit consists of the following:
1) Set device ownership per Core in a Core-pair such that only 1
Core actively process a device (i.e. handles all the requests for
that device) while the other Core in the pair passively watch for
updates on that device and will take over in case the owner Core
failed to process the transaction.
2) Cleanup the lock mechanisms to ensure we use a read lock when
needed instead of just a lock.
3) Update logical port additions such that ports are added only when
the device is enabled.
4) Update the port Ids for the logical ports.
5) Update some sarama client configs for performance - this is an
ongoing tune up.
6) Update the adapter request handler in the Core to send back an
ACK immediately to the adapter request instead of processing the
request fully and then sending an ACK. This reduces the latency
over kafka and therefore reduces the likelihood of timeouts.
Change-Id: I9149bf3ba6fbad38e3a29c76ea8dba2f9f731d29
diff --git a/rw_core/core/grpc_nbi_api_handler.go b/rw_core/core/grpc_nbi_api_handler.go
index d7834eb..1e5cc5b 100644
--- a/rw_core/core/grpc_nbi_api_handler.go
+++ b/rw_core/core/grpc_nbi_api_handler.go
@@ -25,6 +25,7 @@
"github.com/opencord/voltha-go/protos/common"
"github.com/opencord/voltha-go/protos/openflow_13"
"github.com/opencord/voltha-go/protos/voltha"
+ "github.com/opencord/voltha-go/rw_core/utils"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
@@ -40,14 +41,6 @@
)
-type deviceID struct {
- id string
-}
-
-type logicalDeviceID struct {
- id string
-}
-
type APIHandler struct {
deviceMgr *DeviceManager
logicalDeviceMgr *LogicalDeviceManager
@@ -61,20 +54,6 @@
core *Core
}
-//func NewAPIHandler(deviceMgr *DeviceManager, lDeviceMgr *LogicalDeviceManager, adapterMgr *AdapterManager, inCompetingMode bool, longRunningRequestTimeout int64, defaultRequestTimeout int64 ) *APIHandler {
-// handler := &APIHandler{
-// deviceMgr: deviceMgr,
-// logicalDeviceMgr: lDeviceMgr,
-// adapterMgr:adapterMgr,
-// coreInCompetingMode:inCompetingMode,
-// longRunningRequestTimeout:longRunningRequestTimeout,
-// defaultRequestTimeout:defaultRequestTimeout,
-// // TODO: Figure out what the 'hint' parameter to queue.New does
-// packetInQueue: queue.New(10),
-// }
-// return handler
-//}
-
func NewAPIHandler(core *Core) *APIHandler {
handler := &APIHandler{
deviceMgr: core.deviceMgr,
@@ -155,16 +134,16 @@
} else {
if id != nil {
// The id can either be a device Id or a logical device id.
- if dId, ok := id.(*deviceID); ok {
+ if dId, ok := id.(*utils.DeviceID); ok {
// Since this core has not processed this request, let's load the device, along with its extended
// family (parents and children) in memory. This will keep this core in-sync with its paired core as
// much as possible. The watch feature in the core model will ensure that the contents of those objects in
// memory are in sync.
time.Sleep(2 * time.Second)
- go handler.deviceMgr.load(dId.id)
- } else if ldId, ok := id.(*logicalDeviceID); ok {
+ go handler.deviceMgr.load(dId.Id)
+ } else if ldId, ok := id.(*utils.LogicalDeviceID); ok {
// This will load the logical device along with its children and grandchildren
- go handler.logicalDeviceMgr.load(ldId.id)
+ go handler.logicalDeviceMgr.load(ldId.Id)
}
}
return nil, errors.New("failed-to-seize-request")
@@ -185,11 +164,7 @@
owned := false
if id != nil {
- if devId, ok := id.(*deviceID); ok {
- owned = handler.core.deviceOwnership.OwnedByMe(devId.id)
- } else if lDevId, ok := id.(*logicalDeviceID); ok {
- owned = handler.core.deviceOwnership.OwnedByMe(lDevId.id)
- }
+ owned = handler.core.deviceOwnership.OwnedByMe(id)
}
if owned {
if txn.Acquired(timeout) {
@@ -264,7 +239,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &logicalDeviceID{id:id.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.LogicalDeviceID{Id:id.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -285,7 +260,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &logicalDeviceID{id:id.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.LogicalDeviceID{Id:id.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -307,7 +282,7 @@
if handler.competeForTransaction() {
if !handler.isOFControllerRequest(ctx) { // No need to acquire the transaction as request is sent to one core only
- if txn, err := handler.takeRequestOwnership(ctx, &logicalDeviceID{id:flow.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.LogicalDeviceID{Id:flow.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -330,7 +305,7 @@
if handler.competeForTransaction() {
if !handler.isOFControllerRequest(ctx) { // No need to acquire the transaction as request is sent to one core only
- if txn, err := handler.takeRequestOwnership(ctx, &logicalDeviceID{id:flow.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.LogicalDeviceID{Id:flow.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -432,7 +407,7 @@
return &voltha.Device{}, err
}
if d, ok := res.(*voltha.Device); ok {
- handler.core.deviceOwnership.OwnedByMe(d.Id)
+ handler.core.deviceOwnership.OwnedByMe(&utils.DeviceID{Id:d.Id})
return d, nil
}
}
@@ -453,7 +428,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &deviceID{id:id.Id}, handler.longRunningRequestTimeout); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.DeviceID{Id:id.Id}, handler.longRunningRequestTimeout); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -474,7 +449,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &deviceID{id:id.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.DeviceID{Id:id.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -495,7 +470,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &deviceID{id:id.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.DeviceID{Id:id.Id}); err != nil {
return new(empty.Empty), err
} else {
defer txn.Close()
@@ -538,7 +513,7 @@
}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &deviceID{id:img.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.DeviceID{Id:img.Id}); err != nil {
return &common.OperationResp{}, err
} else {
defer txn.Close()
@@ -629,7 +604,7 @@
failedresponse := &voltha.ImageDownload{DownloadState: voltha.ImageDownload_DOWNLOAD_UNKNOWN}
if handler.competeForTransaction() {
- if txn, err := handler.takeRequestOwnership(ctx, &deviceID{id:img.Id}); err != nil {
+ if txn, err := handler.takeRequestOwnership(ctx, &utils.DeviceID{Id:img.Id}); err != nil {
return failedresponse, err
} else {
defer txn.Close()