Vihren Atlas

outbound_queue_factory.go ×8

Joint frontier

unlabeled · c_7432b9215b67

1 test · 6559 LOC · 253 files · introduces 1 test · 144 LOC · 7 files

Introduces — evidence that enters the hierarchy at this concept

Code
27 ranges144 lines · 7 files
Tests
1 test

Contains — complete concept membership

All code (extent)
1146 ranges6559 lines · 253 files · Browse complete extent
All tests (intent)
1 testBrowse 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.

1 test introduced at this concept.

Introduced code

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

7 files ranked by introduced lines: 144 introduced LOC across 27 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/outbound_queue_factory.go 81 introduced LOC · 8 ranges

Open complete file

66 }
67
68 > func (l groupLimiter) Concurrency() int { outbound_queue_factory.go
69 > // Ditto comment above.
70 > nsName := getNamespaceNameOrDefault(
71 > l.namespaceRegistry,
72 > l.key.NamespaceID,
73 > "",
74 > l.metricsHandler,
75 > )
76 > return l.concurrency(nsName, l.key.Destination)
77 > }
78
79 type outboundQueueFactory struct {
89 rateLimiterPool := collection.NewOnceMap(
90 func(key tasks.TaskGroupNamespaceIDAndDestination) quotas.RateLimiter {
91 > return quotas.NewDefaultOutgoingRateLimiter(func() float64 { outbound_queue_factory.go
92 > // This is intentionally not failing the function in case of error. The task
93 > // scheduler doesn't expect errors to happen, and modifying to handle errors
94 > // would make it unnecessarily complex. Also, in this case, if the namespace
95 > // registry fails to get the name, then the task itself will fail when it is
96 > // processed and tries to get the namespace name.
97 > nsName := getNamespaceNameOrDefault(
98 > params.NamespaceRegistry,
99 > key.NamespaceID,
100 > "",
101 > metricsHandler,
102 > )
103 > return params.Config.OutboundQueueHostSchedulerMaxTaskRPS(nsName, key.Destination)
104 > })
105 },
106 )
125 Logger: params.Logger,
126 KeyFn: func(e queues.Executable) tasks.TaskGroupNamespaceIDAndDestination {
127 > return grouper.KeyTyped(e.GetTask()) outbound_queue_factory.go
128 > },
129 > RunnableFactory: func(e queues.Executable) ctasks.Runnable {
130 > key := grouper.KeyTyped(e.GetTask())
131 > nsName := getNamespaceNameOrDefault(
132 > params.NamespaceRegistry,
133 > key.NamespaceID,
134 > key.NamespaceID,
135 > metricsHandler,
136 > )
137 > taggedMetricsHandler := metricsHandler.WithTags(
138 > metrics.NamespaceTag(nsName),
139 > metrics.DestinationTag(key.Destination),
140 > )
141 > return ctasks.NewRateLimitedTaskRunnableFromTask(
142 > ctasks.RunnableTask{
143 > Task: queues.NewCircuitBreakerExecutable(
144 > e,
145 > params.CircuitBreakerPool.Get(key),
146 > taggedMetricsHandler,
147 > ),
148 > },
149 > rateLimiterPool.Get(key),
150 > taggedMetricsHandler,
151 > )
152 > },
153 SchedulerFactory: func(
154 key tasks.TaskGroupNamespaceIDAndDestination,
155 > ) ctasks.RunnableScheduler { outbound_queue_factory.go
156 > nsName := getNamespaceNameOrDefault(
157 > params.NamespaceRegistry,
158 > key.NamespaceID,
159 > key.NamespaceID,
160 > metricsHandler,
161 > )
162 > return ctasks.NewDynamicWorkerPoolScheduler(
163 > groupLimiter{
164 > key: key,
165 > namespaceRegistry: params.NamespaceRegistry,
166 > metricsHandler: metricsHandler,
167 > bufferSize: params.Config.OutboundQueueGroupLimiterBufferSize,
168 > concurrency: params.Config.OutboundQueueGroupLimiterConcurrency,
169 > },
170 > metricsHandler.WithTags(
171 > metrics.NamespaceTag(nsName),
172 > metrics.DestinationTag(key.Destination),
173 > ),
174 > )
175 > },
176 },
177 ),
255
256 if f.ExecutorWrapper != nil {
257 > executor = f.ExecutorWrapper.Wrap(executor) outbound_queue_factory.go
258 > }
259
260 factory := queues.NewExecutableFactory(
354 def string,
355 metricsHandler metrics.Handler,
356 > ) string { outbound_queue_factory.go
357 > nsName, err := registry.GetNamespaceName(namespace.ID(namespaceID))
358 > if err != nil {
359 metrics.ReadNamespaceErrors.With(metricsHandler).
360 Record(1, metrics.ReasonTag(metrics.ReasonString(err.Error())))
361 return def
362 }
363 > return nsName.String() outbound_queue_factory.go
364 }
365
370 return func(taskSlice []tasks.Task) {
371 for _, t := range taskSlice {
372 > if ct, ok := t.(*tasks.ChasmTask); ok { outbound_queue_factory.go
373 > if rt, ok := registry.TaskByID(ct.Info.GetTypeId()); ok {
374 > ct.SetOutboundTaskGroup(rt.TaskGroup())
375 > }
376 }
377 }
go.temporal.io/server/service/history/queues/executable.go 29 introduced LOC · 8 ranges

Open complete file

543 func (e *executableImpl) HandleErr(err error) (retErr error) {
544 if err == nil {
545 > return nil executable.go
546 > }
547
548 defer func() {
775
776 // GetDestination returns the embedded task's destination if it exists. Defaults to an empty string.
777 > func (e *executableImpl) GetDestination() string { executable.go
778 > if t, ok := e.Task.(tasks.HasDestination); ok {
779 > return t.GetDestination()
780 > }
781 return ""
782 }
783
784 // OutboundTaskGroup returns the embedded task's outbound task group. Defaults to an empty string.
785 > func (e *executableImpl) OutboundTaskGroup() string { executable.go
786 > if t, ok := e.Task.(tasks.HasOutboundTaskGroup); ok {
787 > return t.OutboundTaskGroup()
788 > }
789 return ""
790 }
934 cb circuitbreaker.TwoStepCircuitBreaker,
935 metricsHandler metrics.Handler,
936 > ) *CircuitBreakerExecutable { executable.go
937 > return &CircuitBreakerExecutable{
938 > Executable: e,
939 > cb: cb,
940 >
941 > metricsHandler: metricsHandler,
942 > }
943 > }
944
945 // This is roughly the same implementation of the `gobreaker.CircuitBreaker.Execute` function,
946 // but checks if the error is `DestinationDownError` to report success, and unwrap it.
947 > func (e *CircuitBreakerExecutable) Execute() error { executable.go
948 > doneCb, err := e.cb.Allow()
949 > if err != nil {
950 metrics.CircuitBreakerExecutableBlocked.With(e.metricsHandler).Record(1)
951 // Return a resource exhausted error to ensure that this task is retried less aggressively
961 }
962
963 > defer func() { executable.go
964 > e := recover()
965 > if e != nil {
966 doneCb(false)
967 panic(e)
969 }()
970
971 > err = e.Executable.Execute() executable.go
972 > var destinationDownErr *queueserrors.DestinationDownError
973 > if errors.As(err, &destinationDownErr) {
974 err = destinationDownErr.Unwrap()
975 }
976
977 > doneCb(destinationDownErr == nil) executable.go
978 > return err
979 }
go.temporal.io/server/common/tasks/runnable_scheduler.go 18 introduced LOC · 5 ranges

Open complete file

26
27 // Run the embedded task, handling errors and aborting on context errors.
28 > func (a RunnableTask) Run(ctx context.Context) { runnable_scheduler.go
29 > if err := a.HandleErr(a.Execute()); err != nil {
30 if ctx.Err() != nil {
31 a.Abort()
51 limiter quotas.RateLimiter,
52 metricsHandler metrics.Handler,
53 > ) RateLimitedTaskRunnable { runnable_scheduler.go
54 > return RateLimitedTaskRunnable{
55 > Runnable: RunnableTask{task},
56 > Limiter: limiter,
57 >
58 > metricsHandler: metricsHandler,
59 > }
60 > }
61
62 // Run the embedded [Runnable], applying the rate limiter.
63 > func (r RateLimitedTaskRunnable) Run(ctx context.Context) { runnable_scheduler.go
64 > t0 := time.Now()
65 > if err := r.Limiter.Wait(ctx); err != nil {
66 r.Abort()
67 return
68 }
69
70 > metrics.RateLimitedTaskRunnableWaitTime.With(r.metricsHandler).Record(time.Since(t0)) runnable_scheduler.go
71 > r.Runnable.Run(ctx)
72 }
go.temporal.io/server/service/history/tasks/chasm_task.go 9 introduced LOC · 3 ranges

Open complete file

118 }
119
120 > func (t *ChasmTask) GetDestination() string { chasm_task.go
121 > return t.Destination
122 > }
123
124 > func (t *ChasmTask) OutboundTaskGroup() string { chasm_task.go
125 > return t.outboundTaskGroup
126 > }
127
128 > func (t *ChasmTask) SetOutboundTaskGroup(taskgroup string) { chasm_task.go
129 > t.outboundTaskGroup = taskgroup
130 > }
go.temporal.io/server/service/history/circuitbreakerpool/circuit_breaker_factory.go 3 introduced LOC · 1 range

Open complete file

10 }
11
12 > func (p *CircuitBreakerPool[K]) Get(key K) circuitbreaker.TwoStepCircuitBreaker { circuit_breaker_factory.go
13 > return p.m.Get(key)
14 > }
15
16 func NewCircuitBreakerPool[K comparable](
go.temporal.io/server/service/history/queues/metrics.go 2 introduced LOC · 1 range

Open complete file

239 case tasks.CategoryArchival:
240 return GetArchivalTaskTypeTagValue(task)
241 > case tasks.CategoryOutbound: metrics.go
242 > return GetOutboundTaskTypeTagValue(task, isActive, chasmRegistry)
243 default:
244 return task.GetType().String()
go.temporal.io/server/service/history/queues/scheduler_quotas.go 2 introduced LOC · 1 range

Open complete file

25 hostRateFnWithFallback := func() float64 {
26 if rate := hostRateFn(); rate > 0 {
27 > return rate scheduler_quotas.go
28 > }
29
30 return persistenceHostRateFn()

Direct hierarchy

Children