Vihren Atlas

matching_engine.go ×6

Code frontier

unlabeled · c_ede814d347de

9 tests · 5319 LOC · 189 files · introduces 0 tests · 44 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
12 ranges44 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1066 ranges5319 lines · 189 files · Browse complete extent
All tests (intent)
9 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

4 files ranked by introduced lines: 44 introduced LOC across 12 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/matching_engine.go 24 introduced LOC · 6 ranges

Open complete file

2074 operationCreatedTombstones := false
2075 switch req.GetOperation().(type) {
2076 > case *matchingservice.UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_: matching_engine.go
2077 > // Only apply the limit when request is initiated by a user.
2078 > updateOptions.TaskQueueLimitPerBuildId = e.config.TaskQueueLimitPerBuildId(ns.Name().String())
2079 case *matchingservice.UpdateWorkerBuildIdCompatibilityRequest_RemoveBuildIds_:
2080 updateOptions.KnownVersion = req.GetRemoveBuildIds().GetKnownUserDataVersion()
2082
2083 _, err = pm.GetUserDataManager().UpdateUserData(ctx, updateOptions, func(data *persistencespb.TaskQueueUserData) (*persistencespb.TaskQueueUserData, bool, error) {
2084 > clk := data.GetClock() matching_engine.go
2085 > if clk == nil {
2086 tmp := hlc.Zero(e.clusterMeta.GetClusterID())
2087 clk = tmp
2088 }
2089 > updatedClock := hlc.Next(clk, e.timeSource) matching_engine.go
2090 > var versioningData *persistencespb.VersioningData
2091 > switch req.GetOperation().(type) {
2092 > case *matchingservice.UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_:
2093 > var err error
2094 > versioningData, err = UpdateVersionSets(
2095 > updatedClock,
2096 > data.GetVersioningData(),
2097 > req.GetApplyPublicRequest().GetRequest(),
2098 > e.config.VersionCompatibleSetLimitPerQueue(ns.Name().String()),
2099 > e.config.VersionBuildIdLimitPerQueue(ns.Name().String()),
2100 > )
2101 > if err != nil {
2102 return nil, false, err
2103 }
2124 }
2125 // Avoid mutation
2126 > ret := common.CloneProto(data) matching_engine.go
2127 > ret.Clock = updatedClock
2128 > ret.VersioningData = versioningData
2129 > return ret, true, nil
2130 })
2131 if err != nil {
2134
2135 // Only clear tombstones after they have been replicated.
2136 > if operationCreatedTombstones { matching_engine.go
2137 opts := UserDataUpdateOptions{Source: "UpdateWorkerBuildIdCompatibility/clear-tombstones"}
2138 _, err = pm.GetUserDataManager().UpdateUserData(ctx, opts, func(data *persistencespb.TaskQueueUserData) (*persistencespb.TaskQueueUserData, bool, error) {
2148 }
2149 }
2150 > return &matchingservice.UpdateWorkerBuildIdCompatibilityResponse{}, nil matching_engine.go
2151 }
2152
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 9 introduced LOC · 2 ranges

Open complete file

2321 }
2322
2323 > func (x *UpdateWorkerBuildIdCompatibilityRequest) GetApplyPublicRequest() *UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest { request_response.pb.go
2324 > if x != nil {
2325 > if x, ok := x.Operation.(*UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_); ok {
2326 > return x.ApplyPublicRequest
2327 > }
2328 }
2329 return nil
5948 }
5949
5950 > func (x *UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest) GetRequest() *v1.UpdateWorkerBuildIdCompatibilityRequest { request_response.pb.go
5951 > if x != nil {
5952 > return x.Request
5953 > }
5954 return nil
5955 }
go.temporal.io/server/service/matching/user_data_manager.go 9 introduced LOC · 2 ranges

Open complete file

542 added, removed := GetBuildIdDeltas(preUpdateData.GetVersioningData(), updatedUserData.GetVersioningData())
543 if options.TaskQueueLimitPerBuildId > 0 && len(added) > 0 {
544 > // We iterate here but in practice there should only be a single build Id added when the limit is enforced. user_data_manager.go
545 > // We do not enforce the limit when applying replication events.
546 > for _, buildId := range added {
547 > numTaskQueues, err := m.store.CountTaskQueuesByBuildId(ctx, &persistence.CountTaskQueuesByBuildIdRequest{
548 > NamespaceID: m.partition.NamespaceId(),
549 > BuildID: buildId,
550 > })
551 > if err != nil {
552 return nil, false, err
553 }
554 > if numTaskQueues >= options.TaskQueueLimitPerBuildId { user_data_manager.go
555 return nil, false, serviceerror.NewFailedPreconditionf("Exceeded max task queues allowed to be mapped to a single build ID: %d", options.TaskQueueLimitPerBuildId)
556 }
go.temporal.io/server/service/matching/version_sets.go 2 introduced LOC · 2 ranges

Open complete file

58 return serviceerror.NewFailedPreconditionf("update would exceed number of build IDs permitted in namespace dynamic config (%v/%v)", numBuildIds, maxBuildIds)
59 }
60 > return nil version_sets.go
61 }
62
169 func UpdateVersionSets(clock *hlc.Clock, data *persistencespb.VersioningData, req *workflowservice.UpdateWorkerBuildIdCompatibilityRequest, maxSets, maxBuildIds int) (*persistencespb.VersioningData, error) {
170 if data == nil {
171 > data = &persistencespb.VersioningData{VersionSets: make([]*persistencespb.CompatibleVersionSet, 0)} version_sets.go
172 } else {
173 data = common.CloneProto(data)

Direct hierarchy

Parents

Children