#244

Committed 10 Apr 2024 08:19PM UTC coverage: 77.465% (-0.08%) from 77.549%

Build # #244

Build Type

push

github

Committed by

web-flow

Commit Message

Slot supplier interface & fixed-size implementation (#2014)

https://github.com/temporalio/proposals/blob/master/all-sdk/autotuning.md

Run Details

286 of 388 new or added lines in 25 files covered. (73.71%)

3 existing lines in 3 files now uncovered.

19116 of 24677 relevant lines covered (77.46%)

0.77 hits per line

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

88.71

/temporal-sdk/src/main/java/io/temporal/internal/worker/TrackingSlotSupplier.java

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.temporal.internal.worker;

import com.uber.m3.tally.NoopScope;
import com.uber.m3.tally.Scope;
import io.temporal.worker.MetricsType;
import io.temporal.worker.tuning.*;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Wraps a slot supplier and supplements it with additional tracking information that is useful to
 * provide to all implementations. This type is used internally rather than {@link SlotSupplier}
 * directly.
 *
 * @param <SI> The slot info type
 */
public class TrackingSlotSupplier<SI extends SlotInfo> {
  private final SlotSupplier<SI> inner;
  private final AtomicInteger issuedSlots = new AtomicInteger();
  private final Map<SlotPermit, SI> usedSlots = new ConcurrentHashMap<>();
  private Scope metricsScope;

  public TrackingSlotSupplier(SlotSupplier<SI> inner) {
    this.inner = inner;
    metricsScope = new NoopScope();
  }

  public SlotPermit reserveSlot(SlotReservationData dat) throws InterruptedException {
    SlotPermit p = inner.reserveSlot(createCtx(dat));
    issuedSlots.incrementAndGet();
    return p;
  }

  public Optional<SlotPermit> tryReserveSlot(SlotReservationData dat) {
    Optional<SlotPermit> p = inner.tryReserveSlot(createCtx(dat));
    if (p.isPresent()) {
      issuedSlots.incrementAndGet();
    }
    return p;
  }

  public void markSlotUsed(SI slotInfo, SlotPermit permit) {
    if (permit == null) {
      return;
    }
    inner.markSlotUsed(new SlotMarkUsedContextImpl(slotInfo, permit));
    usedSlots.put(permit, slotInfo);
    publishSlotsMetric();
  }

  public void releaseSlot(SlotReleaseReason reason, SlotPermit permit) {
    if (permit == null) {
      return;
    }
    SI slotInfo = usedSlots.get(permit);
    inner.releaseSlot(new SlotReleaseContextImpl(reason, permit, slotInfo));
    issuedSlots.decrementAndGet();
    usedSlots.remove(permit);
    publishSlotsMetric();
  }

  public int maximumSlots() {
    return inner.getMaximumSlots();
  }

  public int getIssuedSlots() {
    return issuedSlots.get();
  }

  public void setMetricsScope(Scope metricsScope) {
    this.metricsScope = metricsScope;
  }

  Map<SlotPermit, SI> getUsedSlots() {
    return usedSlots;
  }

  private void publishSlotsMetric() {
    this.metricsScope
        .gauge(MetricsType.WORKER_TASK_SLOTS_AVAILABLE)
        .update(maximumSlots() - usedSlots.size());
  }

  private SlotReserveContext<SI> createCtx(SlotReservationData dat) {
    return new SlotReserveContextImpl(
        dat.taskQueue,
        Collections.unmodifiableMap(usedSlots),
        dat.workerIdentity,
        dat.workerBuildId);
  }

  private class SlotReserveContextImpl implements SlotReserveContext<SI> {
    private final String taskQueue;
    private final Map<SlotPermit, SI> usedSlots;
    private final String workerIdentity;
    private final String workerBuildId;

    private SlotReserveContextImpl(
        String taskQueue,
        Map<SlotPermit, SI> usedSlots,
        String workerIdentity,
        String workerBuildId) {
      this.taskQueue = taskQueue;
      this.usedSlots = usedSlots;
      this.workerIdentity = workerIdentity;
      this.workerBuildId = workerBuildId;
    }

    @Override
    public String getTaskQueue() {
      return taskQueue;
    }

    @Override
    public Map<SlotPermit, SI> getUsedSlots() {
      return usedSlots;
    }

    @Override
    public String getWorkerIdentity() {
      return workerIdentity;
    }

    @Override
    public String getWorkerBuildId() {
      return workerBuildId;
    }
  }

  private class SlotMarkUsedContextImpl implements SlotMarkUsedContext<SI> {
    private final SI slotInfo;
    private final SlotPermit slotPermit;

    protected SlotMarkUsedContextImpl(SI slotInfo, SlotPermit slotPermit) {
      this.slotInfo = slotInfo;
      this.slotPermit = slotPermit;
    }

    @Override
    public SI getSlotInfo() {
      return slotInfo;
    }

    @Override
    public SlotPermit getSlotPermit() {
      return slotPermit;
    }
  }

  private class SlotReleaseContextImpl implements SlotReleaseContext<SI> {
    private final SlotPermit slotPermit;
    private final SlotReleaseReason reason;
    private final SI slotInfo;

    protected SlotReleaseContextImpl(SlotReleaseReason reason, SlotPermit slotPermit, SI slotInfo) {
      this.slotPermit = slotPermit;
      this.reason = reason;
      this.slotInfo = slotInfo;
    }

    @Override
    public SlotReleaseReason getSlotReleaseReason() {
      return reason;
    }

    @Override
    public SlotPermit getSlotPermit() {
      return slotPermit;
    }

    @Override
    public SI getSlotInfo() {
      return slotInfo;
    }
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.worker;
22
23	import com.uber.m3.tally.NoopScope;
24	import com.uber.m3.tally.Scope;
25	import io.temporal.worker.MetricsType;
26	import io.temporal.worker.tuning.*;
27	import java.util.Collections;
28	import java.util.Map;
29	import java.util.Optional;
30	import java.util.concurrent.ConcurrentHashMap;
31	import java.util.concurrent.atomic.AtomicInteger;
32
33	/**
34	* Wraps a slot supplier and supplements it with additional tracking information that is useful to
35	* provide to all implementations. This type is used internally rather than {@link SlotSupplier}
36	* directly.
37	*
38	* @param <SI> The slot info type
39	*/
40	public class TrackingSlotSupplier<SI extends SlotInfo> {
41	private final SlotSupplier<SI> inner;
42	private final AtomicInteger issuedSlots = new AtomicInteger();	1✔
43	private final Map<SlotPermit, SI> usedSlots = new ConcurrentHashMap<>();	1✔
44	private Scope metricsScope;
45
46	public TrackingSlotSupplier(SlotSupplier<SI> inner) {	1✔
47	this.inner = inner;	1✔
48	metricsScope = new NoopScope();	1✔
49	}	1✔
50
51	public SlotPermit reserveSlot(SlotReservationData dat) throws InterruptedException {
52	SlotPermit p = inner.reserveSlot(createCtx(dat));	1✔
53	issuedSlots.incrementAndGet();	1✔
54	return p;	1✔
55	}
56
57	public Optional<SlotPermit> tryReserveSlot(SlotReservationData dat) {
58	Optional<SlotPermit> p = inner.tryReserveSlot(createCtx(dat));	1✔
59	if (p.isPresent()) {	1✔
60	issuedSlots.incrementAndGet();	1✔
61	}
62	return p;	1✔
63	}
64
65	public void markSlotUsed(SI slotInfo, SlotPermit permit) {
66	if (permit == null) {	1✔
67	return;	1✔
68	}
69	inner.markSlotUsed(new SlotMarkUsedContextImpl(slotInfo, permit));	1✔
70	usedSlots.put(permit, slotInfo);	1✔
71	publishSlotsMetric();	1✔
72	}	1✔
73
74	public void releaseSlot(SlotReleaseReason reason, SlotPermit permit) {
75	if (permit == null) {	1✔
76	return;	1✔
77	}
78	SI slotInfo = usedSlots.get(permit);	1✔
79	inner.releaseSlot(new SlotReleaseContextImpl(reason, permit, slotInfo));	1✔
80	issuedSlots.decrementAndGet();	1✔
81	usedSlots.remove(permit);	1✔
82	publishSlotsMetric();	1✔
83	}	1✔
84
85	public int maximumSlots() {
86	return inner.getMaximumSlots();	1✔
87	}
88
89	public int getIssuedSlots() {
90	return issuedSlots.get();	1✔
91	}
92
93	public void setMetricsScope(Scope metricsScope) {
94	this.metricsScope = metricsScope;	1✔
95	}	1✔
96
97	Map<SlotPermit, SI> getUsedSlots() {
98	return usedSlots;	1✔
99	}
100
101	private void publishSlotsMetric() {
102	this.metricsScope	1✔
103	.gauge(MetricsType.WORKER_TASK_SLOTS_AVAILABLE)	1✔
104	.update(maximumSlots() - usedSlots.size());	1✔
105	}	1✔
106
107	private SlotReserveContext<SI> createCtx(SlotReservationData dat) {
108	return new SlotReserveContextImpl(	1✔
109	dat.taskQueue,
110	Collections.unmodifiableMap(usedSlots),	1✔
111	dat.workerIdentity,
112	dat.workerBuildId);
113	}
114
115	private class SlotReserveContextImpl implements SlotReserveContext<SI> {
116	private final String taskQueue;
117	private final Map<SlotPermit, SI> usedSlots;
118	private final String workerIdentity;
119	private final String workerBuildId;
120
121	private SlotReserveContextImpl(
122	String taskQueue,
123	Map<SlotPermit, SI> usedSlots,
124	String workerIdentity,
125	String workerBuildId) {	1✔
126	this.taskQueue = taskQueue;	1✔
127	this.usedSlots = usedSlots;	1✔
128	this.workerIdentity = workerIdentity;	1✔
129	this.workerBuildId = workerBuildId;	1✔
130	}	1✔
131
132	@Override
133	public String getTaskQueue() {
134	return taskQueue;	1✔
135	}
136
137	@Override
138	public Map<SlotPermit, SI> getUsedSlots() {
139	return usedSlots;	1✔
140	}
141
142	@Override
143	public String getWorkerIdentity() {
NEW 144	return workerIdentity;	×
145	}
146
147	@Override
148	public String getWorkerBuildId() {
NEW 149	return workerBuildId;	×
150	}
151	}
152
153	private class SlotMarkUsedContextImpl implements SlotMarkUsedContext<SI> {
154	private final SI slotInfo;
155	private final SlotPermit slotPermit;
156
157	protected SlotMarkUsedContextImpl(SI slotInfo, SlotPermit slotPermit) {	1✔
158	this.slotInfo = slotInfo;	1✔
159	this.slotPermit = slotPermit;	1✔
160	}	1✔
161
162	@Override
163	public SI getSlotInfo() {
NEW 164	return slotInfo;	×
165	}
166
167	@Override
168	public SlotPermit getSlotPermit() {
NEW 169	return slotPermit;	×
170	}
171	}
172
173	private class SlotReleaseContextImpl implements SlotReleaseContext<SI> {
174	private final SlotPermit slotPermit;
175	private final SlotReleaseReason reason;
176	private final SI slotInfo;
177
178	protected SlotReleaseContextImpl(SlotReleaseReason reason, SlotPermit slotPermit, SI slotInfo) {	1✔
179	this.slotPermit = slotPermit;	1✔
180	this.reason = reason;	1✔
181	this.slotInfo = slotInfo;	1✔
182	}	1✔
183
184	@Override
185	public SlotReleaseReason getSlotReleaseReason() {
NEW 186	return reason;	×
187	}
188
189	@Override
190	public SlotPermit getSlotPermit() {
NEW 191	return slotPermit;	×
192	}
193
194	@Override
195	public SI getSlotInfo() {
NEW 196	return slotInfo;	×
197	}
198	}
199	}

temporalio / sdk-java / #244

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