[VOL-2688] Improve core model performance
This commit addresses the low-hanging performance hogs in the
core model. In particular, the following changes are made:
1) Remove proto message comparision when it's possible. The proto
message deep comparison is quite expensive.
2) Since the Core already has a lock on the device/logicaldevice/
adapters/etc before invoking the model proxy then there is no
need for the latter to create an additional lock on these artifacts
duting an update
3) The model creates a watch on every artifacts it adds to the KV
store. Since in the next Voltha release we will not be using Voltha
Core in pairs then there is no point in keeping these watches (these
is only 1 Core that will ever update an artifact in the next
deployment). This update removes these watch.
4) Additional unit tests has been created, mostly around flows, in an
attempt to exercise both the core and the model further.
Change-Id: Ieaf1f6b9b05c56e819600bc55b46a05f73b8efcf
diff --git a/db/model/node.go b/db/model/node.go
index bec07a5..152bf29 100644
--- a/db/model/node.go
+++ b/db/model/node.go
@@ -598,29 +598,23 @@
return nil
}
- // TODO: validate that this actually works
- //if n.hasChildren(data) {
- // return nil
- //}
-
if n.GetProxy() != nil {
log.Debug("invoking proxy PreUpdate Callbacks")
n.GetProxy().InvokeCallbacks(ctx, PreUpdate, false, branch.GetLatest(), data)
}
- if branch.GetLatest().GetData().(proto.Message).String() != data.(proto.Message).String() {
- if strict {
- // TODO: checkAccessViolations(data, Branch.GetLatest.data)
- log.Debugf("checking access violations")
- }
-
- rev := branch.GetLatest().UpdateData(ctx, data, branch)
- changes := []ChangeTuple{{PostUpdate, branch.GetLatest().GetData(), rev.GetData()}}
- n.makeLatest(branch, rev, changes)
-
- return rev
+ if strict {
+ // TODO: checkAccessViolations(data, Branch.GetLatest.data)
+ log.Warn("access-violations-not-supported")
}
- return branch.GetLatest()
+
+ // The way the model is used, this function is only invoked upon data change. Therefore, to also
+ // avoid a deep proto.message comparison (expensive), just create a new branch regardless
+ rev := branch.GetLatest().UpdateData(ctx, data, branch)
+ changes := []ChangeTuple{{PostUpdate, branch.GetLatest().GetData(), rev.GetData()}}
+ n.makeLatest(branch, rev, changes)
+
+ return rev
}
// Add inserts a new node at the specified path with the provided data
@@ -691,7 +685,6 @@
updatedRev := rev.UpdateChildren(ctx, name, children, branch)
changes := []ChangeTuple{{PostAdd, nil, childRev.GetData()}}
- childRev.SetupWatch(ctx, childRev.GetName())
n.makeLatest(branch, updatedRev, changes)
diff --git a/db/model/non_persisted_revision.go b/db/model/non_persisted_revision.go
index 347be0d..20aeec3 100644
--- a/db/model/non_persisted_revision.go
+++ b/db/model/non_persisted_revision.go
@@ -384,29 +384,10 @@
// Does the existing list contain a child with that name?
if nameExists {
- // Check if the data has changed or not
- if existingChildren[nameIndex].GetData().(proto.Message).String() != newChild.GetData().(proto.Message).String() {
- log.Debugw("replacing-existing-child", log.Fields{
- "old-hash": existingChildren[nameIndex].GetHash(),
- "old-data": existingChildren[nameIndex].GetData(),
- "new-hash": newChild.GetHash(),
- "new-data": newChild.GetData(),
- })
-
- // replace entry
- newChild.getNode().SetRoot(existingChildren[nameIndex].getNode().GetRoot())
- updatedChildren = append(updatedChildren, newChild)
- } else {
- log.Debugw("keeping-existing-child", log.Fields{
- "old-hash": existingChildren[nameIndex].GetHash(),
- "old-data": existingChildren[nameIndex].GetData(),
- "new-hash": newChild.GetHash(),
- "new-data": newChild.GetData(),
- })
-
- // keep existing entry
- updatedChildren = append(updatedChildren, existingChildren[nameIndex])
- }
+ // This function is invoked only when the data has actually changed (current Core usage). Therefore,
+ // we need to avoid an expensive deep proto.message comparison and treat the data as an update
+ newChild.getNode().SetRoot(existingChildren[nameIndex].getNode().GetRoot())
+ updatedChildren = append(updatedChildren, newChild)
} else {
log.Debugw("adding-unknown-child", log.Fields{
"hash": newChild.GetHash(),
diff --git a/db/model/persisted_revision.go b/db/model/persisted_revision.go
index 3637e9a..15e438c 100644
--- a/db/model/persisted_revision.go
+++ b/db/model/persisted_revision.go
@@ -25,7 +25,6 @@
"sync"
"github.com/golang/protobuf/proto"
- "github.com/google/uuid"
"github.com/opencord/voltha-lib-go/v3/pkg/db"
"github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
"github.com/opencord/voltha-lib-go/v3/pkg/log"
@@ -42,21 +41,6 @@
versionMutex sync.RWMutex
Version int64
isStored bool
- isWatched bool
-}
-
-type watchCache struct {
- Cache sync.Map
-}
-
-var watchCacheInstance *watchCache
-var watchCacheOne sync.Once
-
-func watches() *watchCache {
- watchCacheOne.Do(func() {
- watchCacheInstance = &watchCache{Cache: sync.Map{}}
- })
- return watchCacheInstance
}
// NewPersistedRevision creates a new instance of a PersistentRevision structure
@@ -119,154 +103,6 @@
}
}
-// SetupWatch -
-func (pr *PersistedRevision) SetupWatch(ctx context.Context, key string) {
- if key == "" {
- log.Debugw("ignoring-watch", log.Fields{"key": key, "revision-hash": pr.GetHash()})
- return
- }
-
- if _, exists := watches().Cache.LoadOrStore(key+"-"+pr.GetHash(), struct{}{}); exists {
- return
- }
-
- if pr.events == nil {
- pr.events = make(chan *kvstore.Event)
-
- log.Debugw("setting-watch-channel", log.Fields{"key": key, "revision-hash": pr.GetHash()})
-
- pr.SetName(key)
- pr.events = pr.kvStore.CreateWatch(ctx, key, false)
- }
-
- if !pr.isWatched {
- pr.isWatched = true
-
- log.Debugw("setting-watch-routine", log.Fields{"key": key, "revision-hash": pr.GetHash()})
-
- // Start watching
- go pr.startWatching(ctx)
- }
-}
-
-func (pr *PersistedRevision) startWatching(ctx context.Context) {
- log.Debugw("starting-watch", log.Fields{"key": pr.GetHash(), "watch": pr.GetName()})
-
-StopWatchLoop:
- for {
- latestRev := pr.GetBranch().GetLatest()
- event, ok := <-pr.events
- if !ok {
- log.Errorw("event-channel-failure: stopping watch loop", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
- break StopWatchLoop
- }
- log.Debugw("received-event", log.Fields{"type": event.EventType, "watch": latestRev.GetName()})
-
- switch event.EventType {
- case kvstore.DELETE:
- log.Debugw("delete-from-memory", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
-
- // Remove reference from cache
- getRevCache().Delete(latestRev.GetName())
-
- // Remove reference from parent
- parent := pr.GetBranch().Node.GetRoot()
- parent.GetBranch(NONE).Latest.ChildDropByName(latestRev.GetName())
-
- break StopWatchLoop
-
- case kvstore.PUT:
- log.Debugw("update-in-memory", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
- if latestRev.getVersion() >= event.Version {
- log.Debugw("skipping-matching-or-older-revision", log.Fields{
- "watch": latestRev.GetName(),
- "watch-version": event.Version,
- "latest-version": latestRev.getVersion(),
- })
- continue
- } else {
- log.Debugw("watch-revision-is-newer", log.Fields{
- "watch": latestRev.GetName(),
- "watch-version": event.Version,
- "latest-version": latestRev.getVersion(),
- })
- }
-
- data := reflect.New(reflect.TypeOf(latestRev.GetData()).Elem())
-
- if err := proto.Unmarshal(event.Value.([]byte), data.Interface().(proto.Message)); err != nil {
- log.Errorw("failed-to-unmarshal-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "error": err})
- } else {
- log.Debugw("un-marshaled-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "data": data.Interface()})
-
- var pathLock string
-
- // The watch reported new persistence data.
- // Construct an object that will be used to update the memory
- blobs := make(map[string]*kvstore.KVPair)
- key, _ := kvstore.ToString(event.Key)
- blobs[key] = &kvstore.KVPair{
- Key: key,
- Value: event.Value,
- Session: "",
- Lease: 0,
- Version: event.Version,
- }
-
- if latestRev.getNode().GetProxy() != nil {
- //
- // If a proxy exists for this revision, use it to lock access to the path
- // and prevent simultaneous updates to the object in memory
- //
-
- //If the proxy already has a request in progress, then there is no need to process the watch
- if latestRev.getNode().GetProxy().GetOperation() != ProxyNone {
- log.Debugw("operation-in-progress", log.Fields{
- "key": latestRev.GetHash(),
- "path": latestRev.getNode().GetProxy().getFullPath(),
- "operation": latestRev.getNode().GetProxy().operation.String(),
- })
- continue
- }
-
- pathLock, _ = latestRev.getNode().GetProxy().parseForControlledPath(latestRev.getNode().GetProxy().getFullPath())
-
- // Reserve the path to prevent others to modify while we reload from persistence
- if _, err = latestRev.getNode().GetProxy().getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("Unable to acquire a key and set it to a given value", log.Fields{"error": err})
- }
- latestRev.getNode().GetProxy().SetOperation(ProxyWatch)
-
- // Load changes and apply to memory
- if _, err = latestRev.LoadFromPersistence(ctx, latestRev.GetName(), "", blobs); err != nil {
- log.Errorw("Unable to refresh the memory by adding missing entries", log.Fields{"error": err})
- }
-
- // Release path
- if err = latestRev.getNode().GetProxy().getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_"); err != nil {
- log.Errorw("Unable to release reservation for a specific key", log.Fields{"error": err})
- }
- } else {
- // This block should be reached only if coming from a non-proxied request
- log.Debugw("revision-with-no-proxy", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
-
- // Load changes and apply to memory
- if _, err = latestRev.LoadFromPersistence(ctx, latestRev.GetName(), "", blobs); err != nil {
- log.Errorw("Unable to refresh the memory by adding missing entries", log.Fields{"error": err})
- }
- }
- }
-
- default:
- log.Debugw("unhandled-event", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "type": event.EventType})
- }
- }
-
- watches().Cache.Delete(pr.GetName() + "-" + pr.GetHash())
-
- log.Debugw("exiting-watch", log.Fields{"key": pr.GetHash(), "watch": pr.GetName()})
-}
-
// UpdateData modifies the information in the data model and saves it in the persistent storage
func (pr *PersistedRevision) UpdateData(ctx context.Context, data interface{}, branch *Branch) Revision {
log.Debugw("updating-persisted-data", log.Fields{"hash": pr.GetHash()})
@@ -274,12 +110,11 @@
newNPR := pr.Revision.UpdateData(ctx, data, branch)
newPR := &PersistedRevision{
- Revision: newNPR,
- Compress: pr.Compress,
- kvStore: pr.kvStore,
- events: pr.events,
- Version: pr.getVersion(),
- isWatched: pr.isWatched,
+ Revision: newNPR,
+ Compress: pr.Compress,
+ kvStore: pr.kvStore,
+ events: pr.events,
+ Version: pr.getVersion(),
}
if newPR.GetHash() != pr.GetHash() {
@@ -300,12 +135,11 @@
newNPR := pr.Revision.UpdateChildren(ctx, name, children, branch)
newPR := &PersistedRevision{
- Revision: newNPR,
- Compress: pr.Compress,
- kvStore: pr.kvStore,
- events: pr.events,
- Version: pr.getVersion(),
- isWatched: pr.isWatched,
+ Revision: newNPR,
+ Compress: pr.Compress,
+ kvStore: pr.kvStore,
+ events: pr.events,
+ Version: pr.getVersion(),
}
if newPR.GetHash() != pr.GetHash() {
@@ -325,12 +159,11 @@
newNPR := pr.Revision.UpdateAllChildren(ctx, children, branch)
newPR := &PersistedRevision{
- Revision: newNPR,
- Compress: pr.Compress,
- kvStore: pr.kvStore,
- events: pr.events,
- Version: pr.getVersion(),
- isWatched: pr.isWatched,
+ Revision: newNPR,
+ Compress: pr.Compress,
+ kvStore: pr.kvStore,
+ events: pr.events,
+ Version: pr.getVersion(),
}
if newPR.GetHash() != pr.GetHash() {
@@ -357,11 +190,6 @@
pr.mutex.Lock()
defer pr.mutex.Unlock()
if pr.kvStore != nil && txid == "" {
- if pr.isWatched {
- pr.kvStore.DeleteWatch(pr.GetName(), pr.events)
- pr.isWatched = false
- }
-
if err := pr.kvStore.Delete(ctx, pr.GetName()); err != nil {
log.Errorw("failed-to-remove-revision", log.Fields{"hash": pr.GetHash(), "error": err.Error()})
} else {
@@ -412,7 +240,6 @@
updatedChildRev := childRev.UpdateData(ctx, data, childRev.GetBranch())
updatedChildRev.getNode().SetProxy(childRev.getNode().GetProxy())
- updatedChildRev.SetupWatch(ctx, updatedChildRev.GetName())
updatedChildRev.SetLastUpdate()
updatedChildRev.(*PersistedRevision).setVersion(version)
@@ -482,7 +309,6 @@
// We need to start watching this entry for future changes
childRev.SetName(typeName + "/" + keyValue)
- childRev.SetupWatch(ctx, childRev.GetName())
childRev.(*PersistedRevision).setVersion(version)
// Add entry to cache
diff --git a/db/model/proxy.go b/db/model/proxy.go
index 3ffc9ff..303bc4e 100644
--- a/db/model/proxy.go
+++ b/db/model/proxy.go
@@ -26,7 +26,6 @@
"strings"
"sync"
- "github.com/google/uuid"
"github.com/opencord/voltha-lib-go/v3/pkg/log"
)
@@ -191,29 +190,6 @@
p.operation = operation
}
-// parseForControlledPath verifies if a proxy path matches a pattern
-// for locations that need to be access controlled.
-func (p *Proxy) parseForControlledPath(path string) (pathLock string, controlled bool) {
- // TODO: Add other path prefixes that may need control
- if strings.HasPrefix(path, "/devices") ||
- strings.HasPrefix(path, "/logical_devices") ||
- strings.HasPrefix(path, "/adapters") {
-
- split := strings.SplitN(path, "/", -1)
- switch len(split) {
- case 2:
- controlled = false
- pathLock = ""
- case 3:
- fallthrough
- default:
- pathLock = fmt.Sprintf("%s/%s", split[1], split[2])
- controlled = true
- }
- }
- return pathLock, controlled
-}
-
// List will retrieve information from the data model at the specified path location
// A list operation will force access to persistence storage
func (p *Proxy) List(ctx context.Context, path string, depth int, deep bool, txid string) (interface{}, error) {
@@ -224,17 +200,13 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyList)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-list", log.Fields{
- "path": path,
- "effective": effectivePath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "operation": p.GetOperation(),
})
return p.getRoot().List(ctx, path, "", depth, deep, txid)
}
@@ -248,17 +220,13 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyGet)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-get", log.Fields{
- "path": path,
- "effective": effectivePath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "operation": p.GetOperation(),
})
return p.getRoot().Get(ctx, path, "", depth, deep, txid)
@@ -280,33 +248,16 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyUpdate)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-update", log.Fields{
- "path": path,
- "effective": effectivePath,
- "full": fullPath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "full": fullPath,
+ "operation": p.GetOperation(),
})
- if p.getRoot().KvStore != nil {
- if _, err := p.getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("unable-to-acquire-key-from-kvstore", log.Fields{"error": err})
- return nil, err
- }
- defer func() {
- err := p.getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_")
- if err != nil {
- log.Errorw("Unable to release reservation for key", log.Fields{"error": err})
- }
- }()
- }
-
result := p.getRoot().Update(ctx, fullPath, data, strict, txid, nil)
if result != nil {
@@ -334,33 +285,16 @@
effectivePath = p.getFullPath() + path + "/" + id
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyAdd)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-add-with-id", log.Fields{
- "path": path,
- "effective": effectivePath,
- "full": fullPath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "full": fullPath,
+ "operation": p.GetOperation(),
})
- if p.getRoot().KvStore != nil {
- if _, err := p.getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("unable-to-acquire-key-from-kvstore", log.Fields{"error": err})
- return nil, err
- }
- defer func() {
- err := p.getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_")
- if err != nil {
- log.Errorw("Unable to release reservation for key", log.Fields{"error": err})
- }
- }()
- }
-
result := p.getRoot().Add(ctx, fullPath, data, txid, nil)
if result != nil {
@@ -386,33 +320,16 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyAdd)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-add", log.Fields{
- "path": path,
- "effective": effectivePath,
- "full": fullPath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "full": fullPath,
+ "operation": p.GetOperation(),
})
- if p.getRoot().KvStore != nil {
- if _, err := p.getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("unable-to-acquire-key-from-kvstore", log.Fields{"error": err})
- return nil, err
- }
- defer func() {
- err := p.getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_")
- if err != nil {
- log.Errorw("Unable to release reservation for key", log.Fields{"error": err})
- }
- }()
- }
-
result := p.getRoot().Add(ctx, fullPath, data, txid, nil)
if result != nil {
@@ -438,33 +355,16 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyRemove)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-remove", log.Fields{
- "path": path,
- "effective": effectivePath,
- "full": fullPath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "full": fullPath,
+ "operation": p.GetOperation(),
})
- if p.getRoot().KvStore != nil {
- if _, err := p.getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("unable-to-acquire-key-from-kvstore", log.Fields{"error": err})
- return nil, err
- }
- defer func() {
- err := p.getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_")
- if err != nil {
- log.Errorw("Unable to release reservation for key", log.Fields{"error": err})
- }
- }()
- }
-
result := p.getRoot().Remove(ctx, fullPath, txid, nil)
if result != nil {
@@ -491,32 +391,16 @@
effectivePath = p.getFullPath() + path
}
- pathLock, controlled := p.parseForControlledPath(effectivePath)
-
p.SetOperation(ProxyCreate)
defer p.SetOperation(ProxyNone)
log.Debugw("proxy-create", log.Fields{
- "path": path,
- "effective": effectivePath,
- "full": fullPath,
- "pathLock": pathLock,
- "controlled": controlled,
- "operation": p.GetOperation(),
+ "path": path,
+ "effective": effectivePath,
+ "full": fullPath,
+ "operation": p.GetOperation(),
})
- if p.getRoot().KvStore != nil {
- if _, err := p.getRoot().KvStore.Client.Reserve(ctx, pathLock+"_", uuid.New().String(), ReservationTTL); err != nil {
- log.Errorw("unable-to-acquire-key-from-kvstore", log.Fields{"error": err})
- return nil, err
- }
- defer func() {
- err := p.getRoot().KvStore.Client.ReleaseReservation(ctx, pathLock+"_")
- if err != nil {
- log.Errorw("Unable to release reservation for key", log.Fields{"error": err})
- }
- }()
- }
return p.getRoot().CreateProxy(ctx, fullPath, exclusive)
}