couchbase / sync_gateway / 402

Committed 12 May 2025 04:42PM UTC coverage: 64.073% (-0.03%) from 64.101%

Build # 402

Build Type

push

jenkins

Committed by

web-flow

Commit Message

fix processEntry mode on caching tool (#7515)

Run Details

0 of 26 new or added lines in 2 files covered. (0.0%)

11 existing lines in 2 files now uncovered.

36691 of 57264 relevant lines covered (64.07%)

0.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/tools/cache_perf_tool/processEntry.go

// Copyright 2025-Present Couchbase, Inc.
//
// Use of this software is governed by the Business Source License included
// in the file licenses/BSL-Couchbase.txt.  As of the Change Date specified
// in that file, in accordance with the Business Source License, use of this
// software will be governed by the Apache License, Version 2.0, included in
// the file licenses/APL2.txt.

package main

import (
        "context"
        "strconv"
        "testing"
        "time"

        "github.com/couchbase/sync_gateway/channels"
        "github.com/couchbase/sync_gateway/db"
)

type processEntryGen struct {
        seqAlloc       *syncSeqMock
        delays         []time.Duration
        dbCtx          *db.DatabaseContext
        t              *testing.T
        numNodes       int
        batchSize      int
        numChansPerDoc int
        totalChans     int
}

func (p *processEntryGen) spawnDocCreationGoroutine(ctx context.Context) {
        for i := 0; i < p.numNodes; i++ {
                // create new sgw node abstraction
                sgNode := &sgwNode{nodeID: i, seqAlloc: newSequenceAllocator(p.batchSize, p.seqAlloc)}
                // delay list should be same length as num sgw nodes
                go p.nodeWrites(ctx, sgNode, p.delays[i])
        }
}

func (p *processEntryGen) nodeWrites(ctx context.Context, node *sgwNode, delay time.Duration) {
        docCount := uint64(0)
        numGoroutines.Add(1)
        defer numGoroutines.Add(-1)
        // create map of configured channels, initialised channels are names test-x where x is integer between 0 and total
        // number of system channels. Thus to assign to channels we can increment through channels initialised
        chanCountZeroWait := 0
        chanCountWait := 0
        var chanMap channels.ChannelMap
        if delay.Nanoseconds() == 0 {
                // mutate as fast as possible
                for {
                        chanMap = make(channels.ChannelMap, p.numChansPerDoc)
                        for range p.numChansPerDoc {
                                if chanCountZeroWait == p.totalChans {
                                        // when count gets to total number configured channels we rest index to 0
                                        chanCountZeroWait = 0
                                }
                                chanName := "test-" + strconv.Itoa(chanCountZeroWait)
                                chanMap[chanName] = nil
                                chanCountZeroWait++
                        }
                        select {
                        case <-ctx.Done():
                                return
                        default:
                                timeStamp := time.Now()
                                sgwSeqno := node.seqAlloc.nextSeq()
                                docCount++
                                logEntry := &db.LogEntry{
                                        Sequence:     sgwSeqno,
                                        DocID:        "key-" + strconv.FormatUint(docCount, 10) + "-" + strconv.FormatUint(sgwSeqno, 10),
                                        RevID:        "1-abc",
                                        Flags:        0,
                                        TimeReceived: timeStamp,
                                        TimeSaved:    timeStamp,
                                        Channels:     chanMap,
                                        CollectionID: 0,
                                }
                                p.dbCtx.CallProcessEntry(p.t, ctx, logEntry)
                        }
                }
        }

        ticker := time.NewTicker(delay)
        defer ticker.Stop()
        for {
                chanMap = make(channels.ChannelMap, p.numChansPerDoc)
                for range p.numChansPerDoc {
                        if chanCountWait == p.totalChans {
                                // when count gets to total number configured channels we rest index to 0
                                chanCountWait = 0
                        }
                        chanName := "test-" + strconv.Itoa(chanCountWait)
                        chanMap[chanName] = nil
                        chanCountWait++
                }
                select {
                case <-ctx.Done():
                        return
                case <-ticker.C:
                        timeStamp := time.Now()
                        docCount++
                        sgwSeqno := node.seqAlloc.nextSeq()
                        logEntry := &db.LogEntry{
                                Sequence:     sgwSeqno,
                                DocID:        "key-" + strconv.FormatUint(docCount, 10) + "-" + strconv.FormatUint(sgwSeqno, 10),
                                RevID:        "1-abc",
                                Flags:        0,
                                TimeReceived: timeStamp,
                                TimeSaved:    timeStamp,
                                Channels:     chanMap,
                                CollectionID: 0,
                        }
                        p.dbCtx.CallProcessEntry(p.t, ctx, logEntry)
                }
        }
}

1	// Copyright 2025-Present Couchbase, Inc.
2	//
3	// Use of this software is governed by the Business Source License included
4	// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified
5	// in that file, in accordance with the Business Source License, use of this
6	// software will be governed by the Apache License, Version 2.0, included in
7	// the file licenses/APL2.txt.
8
9	package main
10
11	import (
12	"context"
13	"strconv"
14	"testing"
15	"time"
16
17	"github.com/couchbase/sync_gateway/channels"
18	"github.com/couchbase/sync_gateway/db"
19	)
20
21	type processEntryGen struct {
22	seqAlloc *syncSeqMock
23	delays []time.Duration
24	dbCtx *db.DatabaseContext
25	t *testing.T
26	numNodes int
27	batchSize int
28	numChansPerDoc int
29	totalChans int
30	}
31
32	func (p *processEntryGen) spawnDocCreationGoroutine(ctx context.Context) {	×
33	for i := 0; i < p.numNodes; i++ {	×
34	// create new sgw node abstraction	×
35	sgNode := &sgwNode{nodeID: i, seqAlloc: newSequenceAllocator(p.batchSize, p.seqAlloc)}	×
36	// delay list should be same length as num sgw nodes	×
37	go p.nodeWrites(ctx, sgNode, p.delays[i])	×
38	}	×
39	}
40
41	func (p processEntryGen) nodeWrites(ctx context.Context, node sgwNode, delay time.Duration) {	×
42	docCount := uint64(0)	×
43	numGoroutines.Add(1)	×
44	defer numGoroutines.Add(-1)	×
NEW 45	// create map of configured channels, initialised channels are names test-x where x is integer between 0 and total	×
NEW 46	// number of system channels. Thus to assign to channels we can increment through channels initialised	×
NEW 47	chanCountZeroWait := 0	×
NEW 48	chanCountWait := 0	×
NEW 49	var chanMap channels.ChannelMap	×
50	if delay.Nanoseconds() == 0 {	×
51	// mutate as fast as possible	×
52	for {	×
NEW 53	chanMap = make(channels.ChannelMap, p.numChansPerDoc)	×
NEW 54	for range p.numChansPerDoc {	×
NEW 55	if chanCountZeroWait == p.totalChans {	×
NEW 56	// when count gets to total number configured channels we rest index to 0	×
NEW 57	chanCountZeroWait = 0	×
NEW 58	}	×
NEW 59	chanName := "test-" + strconv.Itoa(chanCountZeroWait)	×
NEW 60	chanMap[chanName] = nil	×
NEW 61	chanCountZeroWait++	×
62	}
63	select {	×
64	case <-ctx.Done():	×
65	return	×
66	default:	×
NEW 67	timeStamp := time.Now()	×
68	sgwSeqno := node.seqAlloc.nextSeq()	×
69	docCount++	×
70	logEntry := &db.LogEntry{	×
71	Sequence: sgwSeqno,	×
72	DocID: "key-" + strconv.FormatUint(docCount, 10) + "-" + strconv.FormatUint(sgwSeqno, 10),	×
73	RevID: "1-abc",	×
74	Flags: 0,	×
75	TimeReceived: timeStamp,	×
76	TimeSaved: timeStamp,	×
77	Channels: chanMap,	×
78	CollectionID: 0,	×
79	}	×
80	p.dbCtx.CallProcessEntry(p.t, ctx, logEntry)	×
81	}
82	}
83	}
84
85	ticker := time.NewTicker(delay)	×
86	defer ticker.Stop()	×
87	for {	×
NEW 88	chanMap = make(channels.ChannelMap, p.numChansPerDoc)	×
NEW 89	for range p.numChansPerDoc {	×
NEW 90	if chanCountWait == p.totalChans {	×
NEW 91	// when count gets to total number configured channels we rest index to 0	×
NEW 92	chanCountWait = 0	×
NEW 93	}	×
NEW 94	chanName := "test-" + strconv.Itoa(chanCountWait)	×
NEW 95	chanMap[chanName] = nil	×
NEW 96	chanCountWait++	×
97	}
98	select {	×
99	case <-ctx.Done():	×
100	return	×
101	case <-ticker.C:	×
NEW 102	timeStamp := time.Now()	×
103	docCount++	×
104	sgwSeqno := node.seqAlloc.nextSeq()	×
105	logEntry := &db.LogEntry{	×
106	Sequence: sgwSeqno,	×
107	DocID: "key-" + strconv.FormatUint(docCount, 10) + "-" + strconv.FormatUint(sgwSeqno, 10),	×
108	RevID: "1-abc",	×
109	Flags: 0,	×
110	TimeReceived: timeStamp,	×
111	TimeSaved: timeStamp,	×
112	Channels: chanMap,	×
113	CollectionID: 0,	×
114	}	×
115	p.dbCtx.CallProcessEntry(p.t, ctx, logEntry)	×
116	}
117	}
118	}

couchbase / sync_gateway / 402

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous