Skip to content

Commit b650fd4

Browse files
torcolvintorcolvin
authored
CBG-4441 switch cache to use int64 from time.Time (#7526)
1 parent 2090d8e commit b650fd4

File tree

6 files changed

+134
-117
lines changed

6 files changed

+134
-117
lines changed

‎channels/log_entry.go

Lines changed: 75 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -28,20 +28,19 @@ const (
2828

2929
)
3030

31+
// LogEntry stores information about a change to a document in a cache.
3132
type LogEntry struct {
32-
Sequence uint64 // Sequence number
33-
EndSequence uint64 // End sequence on range of sequences that have been released by the sequence allocator (0 if entry is single sequence)
34-
DocID string // Document ID
35-
RevID string // Revision ID
36-
Flags uint8 // Deleted/Removed/Hidden flags
37-
TimeSaved time.Time // Time doc revision was saved (just used for perf metrics)
38-
TimeReceived time.Time // Time received from tap feed
39-
Channels ChannelMap // Channels this entry is in or was removed from
40-
Skipped bool // Late arriving entry
41-
PrevSequence uint64 // Sequence of previous active revision
42-
IsPrincipal bool // Whether the log-entry is a tracking entry for a principal doc
43-
CollectionID uint32 // Collection ID
44-
UnusedSequence bool // Whether the log-entry is a tracking entry for a unused sequence(s)
33+
Channels ChannelMap // Channels this entry is in or was removed from
34+
DocID string // Document ID
35+
RevID string // Revision ID
36+
Sequence uint64 // Sequence number
37+
EndSequence uint64 // End sequence on range of sequences that have been released by the sequence allocator (0 if entry is single sequence)
38+
TimeReceived FeedTimestamp // Time received from tap feed
39+
CollectionID uint32 // Collection ID
40+
Flags uint8 // Deleted/Removed/Hidden flags
41+
Skipped bool // Late arriving entry
42+
IsPrincipal bool // Whether the log-entry is a tracking entry for a principal doc
43+
UnusedSequence bool // Whether the log-entry is a tracking entry for a unused sequence(s)
4544
}
4645

4746
func (l LogEntry) String() string {
@@ -54,6 +53,36 @@ func (l LogEntry) String() string {
5453
)
5554
}
5655

56+
// IsRemoved returns true if the entry represents a channel removal.
57+
func (l *LogEntry) IsRemoved() bool {
58+
return l.Flags&Removed != 0
59+
}
60+
61+
// IsDeleted returns true if the entry represents a document deletion.
62+
func (l *LogEntry) IsDeleted() bool {
63+
return l.Flags&Deleted != 0
64+
}
65+
66+
// IsActive returns false if the entry is either a removal or a delete.
67+
func (l *LogEntry) IsActive() bool {
68+
return !l.IsRemoved() && !l.IsDeleted()
69+
}
70+
71+
// SetRemoved marks the entry as a channel removal.
72+
func (l *LogEntry) SetRemoved() {
73+
l.Flags |= Removed
74+
}
75+
76+
// SetDeleted marks the entry as a document deletion.
77+
func (l *LogEntry) SetDeleted() {
78+
l.Flags |= Deleted
79+
}
80+
81+
// IsUnusedRange returns true if the entry represents an unused sequence document with more than one sequence.
82+
func (l *LogEntry) IsUnusedRange() bool {
83+
return l.UnusedSequence && l.EndSequence > 0
84+
}
85+
5786
type ChannelMap map[string]*ChannelRemoval
5887
type ChannelRemoval struct {
5988
Seq uint64 `json:"seq,omitempty"`
@@ -82,3 +111,36 @@ func (channelMap ChannelMap) KeySet() []string {
82111
}
83112
return result
84113
}
114+
115+
// FeedTimestamp is a timestamp struct used by DCP. This avoids a conversion from time.Time, while reducing the size from 24 bytes to 8 bytes while having type safety. The time is always assumed to be in local time.
116+
type FeedTimestamp int64
117+
118+
// NewFeedTimestampFromNow creates a new FeedTimestamp from the current time.
119+
func NewFeedTimestampFromNow() FeedTimestamp {
120+
return FeedTimestamp(time.Now().UnixNano())
121+
}
122+
123+
// NewFeedTimestamp creates a new FeedTimestamp from a specific time.Time.
124+
func NewFeedTimestamp(t *time.Time) FeedTimestamp {
125+
return FeedTimestamp(t.UnixNano())
126+
}
127+
128+
// Since returns the nanoseconds that have passed since this timestamp. This function can overflow.
129+
func (t FeedTimestamp) Since() int64 {
130+
return time.Now().UnixNano() - int64(t)
131+
}
132+
133+
// OlderThan returns true if the timestamp is older than the given duration.
134+
func (t FeedTimestamp) OlderThan(duration time.Duration) bool {
135+
return t.Since() > int64(duration)
136+
}
137+
138+
// OlderOrEqual returns true if the timestamp is older or equal to the given duration.
139+
func (t FeedTimestamp) OlderOrEqual(duration time.Duration) bool {
140+
return t.Since() >= int64(duration)
141+
}
142+
143+
// After returns true if the timestamp is after the given time.
144+
func (t FeedTimestamp) After(other time.Time) bool {
145+
return int64(t) > other.UnixNano()
146+
}

‎db/change_cache.go

Lines changed: 19 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -98,36 +98,7 @@ func (c *changeCache) updateStats(ctx context.Context) {
9898
c.db.DbStats.Cache().SkippedSeqCap.Set(skippedSequenceListStats.ListCapacityStat)
9999
}
100100

101-
type LogEntry channels.LogEntry
102-
103-
func (l LogEntry) String() string {
104-
return channels.LogEntry(l).String()
105-
}
106-
107-
func (entry *LogEntry) IsRemoved() bool {
108-
return entry.Flags&channels.Removed != 0
109-
}
110-
111-
func (entry *LogEntry) IsDeleted() bool {
112-
return entry.Flags&channels.Deleted != 0
113-
}
114-
115-
// Returns false if the entry is either a removal or a delete
116-
func (entry *LogEntry) IsActive() bool {
117-
return !entry.IsRemoved() && !entry.IsDeleted()
118-
}
119-
120-
func (entry *LogEntry) SetRemoved() {
121-
entry.Flags |= channels.Removed
122-
}
123-
124-
func (entry *LogEntry) SetDeleted() {
125-
entry.Flags |= channels.Deleted
126-
}
127-
128-
func (entry *LogEntry) IsUnusedRange() bool {
129-
return entry.UnusedSequence && entry.EndSequence > 0
130-
}
101+
type LogEntry = channels.LogEntry
131102

132103
type LogEntries []*LogEntry
133104

@@ -332,19 +303,20 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent) {
332303
docJSON := event.Value
333304
changedChannelsCombined := channels.Set{}
334305

306+
timeReceived := channels.NewFeedTimestamp(&event.TimeReceived)
335307
// ** This method does not directly access any state of c, so it doesn't lock.
336308
// Is this a user/role doc for this database?
337309
if strings.HasPrefix(docID, c.metaKeys.UserKeyPrefix()) {
338-
c.processPrincipalDoc(ctx, docID, docJSON, true, event.TimeReceived)
310+
c.processPrincipalDoc(ctx, docID, docJSON, true, timeReceived)
339311
return
340312
} else if strings.HasPrefix(docID, c.metaKeys.RoleKeyPrefix()) {
341-
c.processPrincipalDoc(ctx, docID, docJSON, false, event.TimeReceived)
313+
c.processPrincipalDoc(ctx, docID, docJSON, false, timeReceived)
342314
return
343315
}
344316

345317
// Is this an unused sequence notification?
346318
if strings.HasPrefix(docID, c.metaKeys.UnusedSeqPrefix()) {
347-
c.processUnusedSequence(ctx, docID, event.TimeReceived)
319+
c.processUnusedSequence(ctx, docID, timeReceived)
348320
return
349321
}
350322
if strings.HasPrefix(docID, c.metaKeys.UnusedSeqRangePrefix()) {
@@ -454,7 +426,7 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent) {
454426
base.InfofCtx(ctx, base.KeyCache, "Received unused #%d in unused_sequences property for (%q / %q)", seq, base.UD(docID), syncData.CurrentRev)
455427
change := &LogEntry{
456428
Sequence: seq,
457-
TimeReceived: event.TimeReceived,
429+
TimeReceived: timeReceived,
458430
CollectionID: event.CollectionID,
459431
}
460432
changedChannels := c.processEntry(ctx, change)
@@ -479,7 +451,7 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent) {
479451
base.InfofCtx(ctx, base.KeyCache, "Received deduplicated #%d in recent_sequences property for (%q / %q)", seq, base.UD(docID), syncData.CurrentRev)
480452
change := &LogEntry{
481453
Sequence: seq,
482-
TimeReceived: event.TimeReceived,
454+
TimeReceived: timeReceived,
483455
CollectionID: event.CollectionID,
484456
}
485457

@@ -506,8 +478,7 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent) {
506478
DocID: docID,
507479
RevID: syncData.CurrentRev,
508480
Flags: syncData.Flags,
509-
TimeReceived: event.TimeReceived,
510-
TimeSaved: syncData.TimeSaved,
481+
TimeReceived: timeReceived,
511482
Channels: syncData.Channels,
512483
CollectionID: event.CollectionID,
513484
}
@@ -549,7 +520,7 @@ func (c *changeCache) unmarshalCachePrincipal(docJSON []byte) (cachePrincipal, e
549520
}
550521

551522
// Process unused sequence notification. Extracts sequence from docID and sends to cache for buffering
552-
func (c *changeCache) processUnusedSequence(ctx context.Context, docID string, timeReceived time.Time) {
523+
func (c *changeCache) processUnusedSequence(ctx context.Context, docID string, timeReceived channels.FeedTimestamp) {
553524
sequenceStr := strings.TrimPrefix(docID, c.metaKeys.UnusedSeqPrefix())
554525
sequence, err := strconv.ParseUint(sequenceStr, 10, 64)
555526
if err != nil {
@@ -560,7 +531,7 @@ func (c *changeCache) processUnusedSequence(ctx context.Context, docID string, t
560531

561532
}
562533

563-
func (c *changeCache) releaseUnusedSequence(ctx context.Context, sequence uint64, timeReceived time.Time) {
534+
func (c *changeCache) releaseUnusedSequence(ctx context.Context, sequence uint64, timeReceived channels.FeedTimestamp) {
564535
change := &LogEntry{
565536
Sequence: sequence,
566537
TimeReceived: timeReceived,
@@ -583,7 +554,7 @@ func (c *changeCache) releaseUnusedSequence(ctx context.Context, sequence uint64
583554

584555
// releaseUnusedSequenceRange will handle unused sequence range arriving over DCP. It will batch remove from skipped or
585556
// push a range to pending sequences, or both.
586-
func (c *changeCache) releaseUnusedSequenceRange(ctx context.Context, fromSequence uint64, toSequence uint64, timeReceived time.Time) {
557+
func (c *changeCache) releaseUnusedSequenceRange(ctx context.Context, fromSequence uint64, toSequence uint64, timeReceived channels.FeedTimestamp) {
587558

588559
base.InfofCtx(ctx, base.KeyCache, "Received #%d-#%d (unused sequence range)", fromSequence, toSequence)
589560

@@ -613,7 +584,7 @@ func (c *changeCache) releaseUnusedSequenceRange(ctx context.Context, fromSequen
613584
}
614585

615586
// processUnusedRange handles pushing unused range to pending or skipped lists
616-
func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSequence uint64, allChangedChannels channels.Set, timeReceived time.Time) channels.Set {
587+
func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSequence uint64, allChangedChannels channels.Set, timeReceived channels.FeedTimestamp) channels.Set {
617588
c.lock.Lock()
618589
defer c.lock.Unlock()
619590

@@ -622,7 +593,7 @@ func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSe
622593
c.skippedSeqs.processUnusedSequenceRangeAtSkipped(ctx, fromSequence, toSequence)
623594
} else if fromSequence >= c.nextSequence {
624595
// whole range to pending
625-
c._pushRangeToPending(ctx, fromSequence, toSequence, timeReceived)
596+
c._pushRangeToPending(fromSequence, toSequence, timeReceived)
626597
// unblock any pending sequences we can after new range(s) have been pushed to pending
627598
changedChannels := c._addPendingLogs(ctx)
628599
allChangedChannels = allChangedChannels.Update(changedChannels)
@@ -640,7 +611,7 @@ func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSe
640611
}
641612

642613
// _pushRangeToPending will push an unused sequence range to pendingLogs
643-
func (c *changeCache) _pushRangeToPending(ctx context.Context, startSeq, endSeq uint64, timeReceived time.Time) {
614+
func (c *changeCache) _pushRangeToPending(startSeq, endSeq uint64, timeReceived channels.FeedTimestamp) {
644615

645616
entry := &LogEntry{
646617
TimeReceived: timeReceived,
@@ -672,10 +643,10 @@ func (c *changeCache) processUnusedSequenceRange(ctx context.Context, docID stri
672643
return
673644
}
674645

675-
c.releaseUnusedSequenceRange(ctx, fromSequence, toSequence, time.Now())
646+
c.releaseUnusedSequenceRange(ctx, fromSequence, toSequence, channels.NewFeedTimestampFromNow())
676647
}
677648

678-
func (c *changeCache) processPrincipalDoc(ctx context.Context, docID string, docJSON []byte, isUser bool, timeReceived time.Time) {
649+
func (c *changeCache) processPrincipalDoc(ctx context.Context, docID string, docJSON []byte, isUser bool, timeReceived channels.FeedTimestamp) {
679650

680651
// Currently the cache isn't really doing much with user docs; mostly it needs to know about
681652
// them because they have sequence numbers, so without them the sequence of sequences would
@@ -818,9 +789,9 @@ func (c *changeCache) _addToCache(ctx context.Context, change *LogEntry) channel
818789
base.DebugfCtx(ctx, base.KeyChanges, " #%d ==> channels %v", change.Sequence, base.UD(updatedChannels))
819790
}
820791

821-
if !change.TimeReceived.IsZero() {
792+
if change.TimeReceived != 0 {
822793
c.db.DbStats.Database().DCPCachingCount.Add(1)
823-
c.db.DbStats.Database().DCPCachingTime.Add(time.Since(change.TimeReceived).Nanoseconds())
794+
c.db.DbStats.Database().DCPCachingTime.Add(change.TimeReceived.Since())
824795
}
825796

826797
return updatedChannels
@@ -849,7 +820,7 @@ func (c *changeCache) _addPendingLogs(ctx context.Context) channels.Set {
849820
if oldestPending.IsUnusedRange() && oldestPending.EndSequence >= c.nextSequence {
850821
c.nextSequence = oldestPending.EndSequence + 1
851822
}
852-
} else if len(c.pendingLogs) > c.options.CachePendingSeqMaxNum || time.Since(c.pendingLogs[0].TimeReceived) >= c.options.CachePendingSeqMaxWait {
823+
} else if len(c.pendingLogs) > c.options.CachePendingSeqMaxNum || c.pendingLogs[0].TimeReceived.OlderOrEqual(c.options.CachePendingSeqMaxWait) {
853824
// Skip all sequences up to the oldest Pending
854825
c.PushSkipped(ctx, c.nextSequence, oldestPending.Sequence-1)
855826
c.nextSequence = oldestPending.Sequence

0 commit comments

Comments
 (0)