Skip to content

Commit 9a42542

Browse files
gregns1gregns1
authored
CBG-4747: remove serialisation from caching process (#7885)
1 parent 1f6d9a6 commit 9a42542

File tree

3 files changed

+51
-47
lines changed

3 files changed

+51
-47
lines changed

‎channels/set.go‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ package channels
1111
import (
1212
"fmt"
1313
"sort"
14-
"strconv"
1514
"strings"
1615

1716
"github.com/couchbase/sync_gateway/base"
@@ -32,21 +31,21 @@ const AllChannelWildcard = "*" // wildcard for 'all channels'
3231

3332
// ID represents a single channel inside a collection
3433
type ID struct {
35-
Name string // name of channel
36-
CollectionID uint32 // collection it belongs to
37-
serialization string // private method for logging and matching inside changeWaiter notification
34+
Name string // name of channel
35+
CollectionID uint32 // collection it belongs to
3836
}
3937

38+
// String returns the channel name as a string, omitting the collection ID, so beware when using for logging purposes
39+
// that the collection name is logged from the context on the log line
4040
func (c ID) String() string {
41-
return c.serialization
41+
return c.Name
4242
}
4343

4444
// NewID returns a new ChannelID
4545
func NewID(channelName string, collectionID uint32) ID {
4646
return ID{
47-
Name: channelName,
48-
CollectionID: collectionID,
49-
serialization: strconv.FormatUint(uint64(collectionID), 10) + "." + base.UserDataPrefix + channelName + base.UserDataSuffix,
47+
Name: channelName,
48+
CollectionID: collectionID,
5049
}
5150
}
5251

@@ -196,7 +195,7 @@ func (redactorSet RedactorSet) GetRedactionString(shouldRedact bool) string {
196195
tmp := []byte("{")
197196
iterationCount := 0
198197
for setItem, _ := range redactorSet.set {
199-
tmp = append(tmp, redactorSet.redactorFunc(setItem).String()...)
198+
tmp = append(tmp, redactorSet.redactorFunc(setItem).Redact()...)
200199
iterationCount++
201200
if iterationCount != len(redactorSet.set) {
202201
tmp = append(tmp, ", "...)

‎channels/set_test.go‎

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -373,14 +373,14 @@ func TestSetString(t *testing.T) {
373373
NewID("B", 2): present{},
374374
NewID("C", 1): present{},
375375
},
376-
output: "{1.<ud>A</ud>, 1.<ud>C</ud>, 2.<ud>B</ud>}",
376+
output: "{A, B, C}",
377377
redactedOutput: []string{
378-
"{1.<ud>A</ud>, 1.<ud>C</ud>, 2.<ud>B</ud>}",
379-
"{1.<ud>A</ud>, 2.<ud>B</ud>, 1.<ud>C</ud>}",
380-
"{1.<ud>C</ud>, 1.<ud>A</ud>, 2.<ud>B</ud>}",
381-
"{1.<ud>C</ud>, 2.<ud>B</ud>, 1.<ud>A</ud>}",
382-
"{2.<ud>B</ud>, 1.<ud>A</ud>, 1.<ud>C</ud>}",
383-
"{2.<ud>B</ud>, 1.<ud>C</ud>, 1.<ud>A</ud>}",
378+
"{<ud>A</ud>, <ud>C</ud>, <ud>B</ud>}",
379+
"{<ud>A</ud>, <ud>B</ud>, <ud>C</ud>}",
380+
"{<ud>C</ud>, <ud>A</ud>, <ud>B</ud>}",
381+
"{<ud>C</ud>, <ud>B</ud>, <ud>A</ud>}",
382+
"{<ud>B</ud>, <ud>A</ud>, <ud>C</ud>}",
383+
"{<ud>B</ud>, <ud>C</ud>, <ud>A</ud>}",
384384
},
385385
},
386386
{
@@ -390,23 +390,23 @@ func TestSetString(t *testing.T) {
390390
NewID("B", 2): present{},
391391
NewID("C", 1): present{},
392392
},
393-
output: "{1.<ud>C</ud>, 2.<ud>A</ud>, 2.<ud>B</ud>}",
393+
output: "{A, B, C}",
394394
redactedOutput: []string{
395-
"{2.<ud>A</ud>, 1.<ud>C</ud>, 2.<ud>B</ud>}",
396-
"{2.<ud>A</ud>, 2.<ud>B</ud>, 1.<ud>C</ud>}",
397-
"{1.<ud>C</ud>, 2.<ud>A</ud>, 2.<ud>B</ud>}",
398-
"{1.<ud>C</ud>, 2.<ud>B</ud>, 2.<ud>A</ud>}",
399-
"{2.<ud>B</ud>, 2.<ud>A</ud>, 1.<ud>C</ud>}",
400-
"{2.<ud>B</ud>, 1.<ud>C</ud>, 2.<ud>A</ud>}",
395+
"{<ud>A</ud>, <ud>C</ud>, <ud>B</ud>}",
396+
"{<ud>A</ud>, <ud>B</ud>, <ud>C</ud>}",
397+
"{<ud>C</ud>, <ud>A</ud>, <ud>B</ud>}",
398+
"{<ud>C</ud>, <ud>B</ud>, <ud>A</ud>}",
399+
"{<ud>B</ud>, <ud>A</ud>, <ud>C</ud>}",
400+
"{<ud>B</ud>, <ud>C</ud>, <ud>A</ud>}",
401401
},
402402
},
403403
{
404404
name: "one collection",
405405
input: Set{
406406
NewID("A", 1): present{},
407407
},
408-
output: "{1.<ud>A</ud>}",
409-
redactedOutput: []string{"{1.<ud>A</ud>}"},
408+
output: "{A}",
409+
redactedOutput: []string{"{<ud>A</ud>}"},
410410
},
411411
}
412412
for _, test := range testCases {

‎db/change_listener.go‎

Lines changed: 27 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ type changeListener struct {
4141
FeedArgs sgbucket.FeedArguments // The Tap Args (backfill, etc)
4242
counter uint64 // Event counter; increments on every doc update
4343
_terminateCheckCounter uint64 // Termination Event counter; increments on every notifyCheckForTermination
44-
keyCounts map[string]uint64 // Latest count at which each doc key was updated
44+
keyCounts map[channels.ID]uint64 // Latest count at which each doc key was updated
4545
OnChangeCallback DocChangedFunc
4646
terminator chan bool // Signal to cause DCP feed to exit
4747
broadcastChangesDoneChan chan struct{} // Channel to signal that broadcast changes goroutine has terminated
@@ -53,13 +53,16 @@ type changeListener struct {
5353
// unusedSeqChannelID marks the unused sequence key for the channel cache. This is a marker that is global to all collections.
5454
var unusedSeqChannelID = channels.NewID(unusedSeqKey, unusedSeqCollectionID)
5555

56+
// principalDocCollectionIDForChannelID is the collection ID for construction of channel ID for principal documents (users, roles).
57+
const principalDocCollectionIDForChannelID = 0
58+
5659
type DocChangedFunc func(event sgbucket.FeedEvent, docType DocumentType)
5760

5861
func (listener *changeListener) Init(name string, groupID string, db *DatabaseContext) {
5962
listener.bucketName = name
6063
listener.counter = 1
6164
listener._terminateCheckCounter = 0
62-
listener.keyCounts = map[string]uint64{}
65+
listener.keyCounts = map[channels.ID]uint64{}
6366
listener.tapNotifier = sync.NewCond(&sync.Mutex{})
6467
listener.sgCfgPrefix = db.MetadataKeys.SGCfgPrefix(groupID)
6568
listener.metaKeys = db.MetadataKeys
@@ -181,7 +184,8 @@ func (listener *changeListener) ProcessFeedEvent(event sgbucket.FeedEvent) bool
181184
docType := listener.DocumentType(event.Key)
182185
if docType == DocTypeUser || docType == DocTypeRole {
183186
// defer to notify after callback completion
184-
defer listener.notifyKey(listener.ctx, string(event.Key))
187+
key := channels.NewID(string(event.Key), principalDocCollectionIDForChannelID)
188+
defer listener.notifyKey(listener.ctx, key)
185189
}
186190

187191
listener.OnDocChanged(event, docType)
@@ -251,7 +255,7 @@ func (listener *changeListener) Notify(ctx context.Context, keys channels.Set) {
251255
listener.tapNotifier.L.Lock()
252256
listener.counter++
253257
for key := range keys {
254-
listener.keyCounts[key.String()] = listener.counter
258+
listener.keyCounts[key] = listener.counter
255259
}
256260
base.DebugfCtx(ctx, base.KeyChanges, "Listener keys %q for %s have changed, count=%d",
257261
base.UD(keys), base.MD(listener.bucketName), listener.counter)
@@ -309,7 +313,7 @@ func tickerValForBroadcastSpeed(skippedSequencePresent bool) time.Duration {
309313
}
310314

311315
// Changes the counter, notifying waiting clients. Only use for a key update.
312-
func (listener *changeListener) notifyKey(ctx context.Context, key string) {
316+
func (listener *changeListener) notifyKey(ctx context.Context, key channels.ID) {
313317
listener.tapNotifier.L.Lock()
314318
listener.counter++
315319
listener.keyCounts[key] = listener.counter
@@ -340,7 +344,7 @@ func (listener *changeListener) NotifyCheckForTermination(ctx context.Context, k
340344
}
341345

342346
// Waits until either the counter, or terminateCheckCounter exceeds the given value. Returns the new counters.
343-
func (listener *changeListener) Wait(ctx context.Context, keys []string, counter uint64, terminateCheckCounter uint64) (uint64, uint64) {
347+
func (listener *changeListener) Wait(ctx context.Context, keys []channels.ID, counter uint64, terminateCheckCounter uint64) (uint64, uint64) {
344348
listener.tapNotifier.L.Lock()
345349
defer listener.tapNotifier.L.Unlock()
346350
base.DebugfCtx(ctx, base.KeyChanges, "No new changes to send to change listener. Waiting for %q's count to pass %d",
@@ -367,13 +371,13 @@ func (listener *changeListener) Wait(ctx context.Context, keys []string, counter
367371
}
368372

369373
// Returns the max value of the counter for all the given keys
370-
func (listener *changeListener) CurrentCount(keys []string) uint64 {
374+
func (listener *changeListener) CurrentCount(keys []channels.ID) uint64 {
371375
listener.tapNotifier.L.Lock()
372376
defer listener.tapNotifier.L.Unlock()
373377
return listener._currentCount(keys)
374378
}
375379

376-
func (listener *changeListener) _currentCount(keys []string) uint64 {
380+
func (listener *changeListener) _currentCount(keys []channels.ID) uint64 {
377381
var max uint64 = 0
378382
for _, key := range keys {
379383
if count := listener.keyCounts[key]; count > max {
@@ -389,23 +393,23 @@ func (listener *changeListener) _currentCount(keys []string) uint64 {
389393
// listener's counter to increment from the value at the last call.
390394
type ChangeWaiter struct {
391395
listener *changeListener
392-
keys []string
393-
userKeys []string
396+
keys []channels.ID
397+
userKeys []channels.ID
394398
lastCounter uint64
395399
lastTerminateCheckCounter uint64
396400
lastUserCount uint64
397401
trackUnusedSequences bool // track unused sequences in Wait functions
398402
}
399403

400404
// NewWaiter a new ChangeWaiter that will wait for changes for the given document keys, and will optionally track unused sequences.
401-
func (listener *changeListener) NewWaiter(keys []string, trackUnusedSequences bool) *ChangeWaiter {
405+
func (listener *changeListener) NewWaiter(keys []channels.ID, trackUnusedSequences bool) *ChangeWaiter {
402406
listener.tapNotifier.L.Lock()
403407
defer listener.tapNotifier.L.Unlock()
404408
return listener._newWaiter(keys, trackUnusedSequences)
405409
}
406410

407411
// _newWaiter a new ChangeWaiter that will wait for changes for the given document keys, and will optionally track unused sequences.
408-
func (listener *changeListener) _newWaiter(keys []string, trackUnusedSequences bool) *ChangeWaiter {
412+
func (listener *changeListener) _newWaiter(keys []channels.ID, trackUnusedSequences bool) *ChangeWaiter {
409413
return &ChangeWaiter{
410414
listener: listener,
411415
keys: keys,
@@ -417,15 +421,16 @@ func (listener *changeListener) _newWaiter(keys []string, trackUnusedSequences b
417421

418422
// NewWaiterWithChannels creates ChangeWaiter for a given channel and user, and will optionally track unused sequences.
419423
func (listener *changeListener) NewWaiterWithChannels(chans channels.Set, user auth.User, trackUnusedSequences bool) *ChangeWaiter {
420-
waitKeys := make([]string, 0, 5)
424+
waitKeys := make([]channels.ID, 0, 5)
421425
for channel := range chans {
422-
waitKeys = append(waitKeys, channel.String())
426+
waitKeys = append(waitKeys, channel)
423427
}
424-
var userKeys []string
428+
var userKeys []channels.ID
425429
if user != nil {
426-
userKeys = []string{listener.metaKeys.UserKey(user.Name())}
430+
usrID := channels.NewID(listener.metaKeys.UserKey(user.Name()), principalDocCollectionIDForChannelID)
431+
userKeys = []channels.ID{usrID}
427432
for role := range user.RoleNames() {
428-
userKeys = append(userKeys, listener.metaKeys.RoleKey(role))
433+
userKeys = append(userKeys, channels.NewID(listener.metaKeys.RoleKey(role), principalDocCollectionIDForChannelID))
429434
}
430435
waitKeys = append(waitKeys, userKeys...)
431436
}
@@ -480,12 +485,12 @@ func (waiter *ChangeWaiter) RefreshUserCount() bool {
480485
func (waiter *ChangeWaiter) UpdateChannels(collectionID uint32, timedSet channels.TimedSet) {
481486
// This capacity is not right can not accommodate channels without iteration.
482487
initialCapacity := len(waiter.userKeys)
483-
updatedKeys := make([]string, 0, initialCapacity)
488+
updatedKeys := make([]channels.ID, 0, initialCapacity)
484489
for channelName, _ := range timedSet {
485-
updatedKeys = append(updatedKeys, channels.NewID(channelName, collectionID).String())
490+
updatedKeys = append(updatedKeys, channels.NewID(channelName, collectionID))
486491
}
487492
if waiter.trackUnusedSequences {
488-
updatedKeys = append(updatedKeys, unusedSeqChannelID.String())
493+
updatedKeys = append(updatedKeys, unusedSeqChannelID)
489494
}
490495
if len(waiter.userKeys) > 0 {
491496
updatedKeys = append(updatedKeys, waiter.userKeys...)
@@ -505,9 +510,9 @@ func (waiter *ChangeWaiter) RefreshUserKeys(user auth.User, metaKeys *base.Metad
505510
if len(waiter.userKeys) == 1 && len(user.RoleNames()) == 0 {
506511
return
507512
}
508-
waiter.userKeys = []string{metaKeys.UserKey(user.Name())}
513+
waiter.userKeys = []channels.ID{channels.NewID(metaKeys.UserKey(user.Name()), principalDocCollectionIDForChannelID)}
509514
for role := range user.RoleNames() {
510-
waiter.userKeys = append(waiter.userKeys, metaKeys.RoleKey(role))
515+
waiter.userKeys = append(waiter.userKeys, channels.NewID(metaKeys.RoleKey(role), principalDocCollectionIDForChannelID))
511516
}
512517
waiter.lastUserCount = waiter.listener.CurrentCount(waiter.userKeys)
513518

0 commit comments

Comments
 (0)