Skip to content

Commit 9bfb5d1

Browse files
committed
Fix tag index memory accounting
1 parent 1d893f1 commit 9bfb5d1

2 files changed

Lines changed: 228 additions & 50 deletions

File tree

  • iotdb-core/datanode/src

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/tag/TagManager.java‎

Lines changed: 59 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -166,34 +166,33 @@ public void addIndex(String tagKey, String tagValue, IMeasurementMNode<?> measur
166166
return;
167167
}
168168

169-
int tagIndexOldSize = tagIndex.size();
170-
Map<String, Set<IMeasurementMNode<?>>> tagValueMap =
171-
tagIndex.computeIfAbsent(tagKey, k -> new ConcurrentHashMap<>());
172-
int tagIndexNewSize = tagIndex.size();
173-
174-
int tagValueMapOldSize = tagValueMap.size();
175-
Set<IMeasurementMNode<?>> measurementsSet =
176-
tagValueMap.computeIfAbsent(tagValue, v -> Collections.synchronizedSet(new HashSet<>()));
177-
int tagValueMapNewSize = tagValueMap.size();
178-
179-
int measurementsSetOldSize = measurementsSet.size();
180-
measurementsSet.add(measurementMNode);
181-
int measurementsSetNewSize = measurementsSet.size();
182-
183-
long memorySize = 0;
184-
if (tagIndexNewSize - tagIndexOldSize == 1) {
185-
// the last 4 is the memory occupied by the size of tagvaluemap
186-
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
187-
}
188-
if (tagValueMapNewSize - tagValueMapOldSize == 1) {
189-
// the last 4 is the memory occupied by the size of measurementsSet
190-
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
191-
}
192-
if (measurementsSetNewSize - measurementsSetOldSize == 1) {
193-
// 8 is the memory occupied by the length of the IMeasurementMNode
194-
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
195-
}
196-
requestMemory(memorySize);
169+
tagIndex.compute(
170+
tagKey,
171+
(key, tagValueMap) -> {
172+
long memorySize = 0;
173+
if (tagValueMap == null) {
174+
tagValueMap = new ConcurrentHashMap<>();
175+
// the last 4 is the memory occupied by the size of tagvaluemap
176+
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
177+
}
178+
179+
Set<IMeasurementMNode<?>> measurementsSet = tagValueMap.get(tagValue);
180+
if (measurementsSet == null) {
181+
measurementsSet = ConcurrentHashMap.newKeySet();
182+
tagValueMap.put(tagValue, measurementsSet);
183+
// the last 4 is the memory occupied by the size of measurementsSet
184+
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
185+
}
186+
187+
if (measurementsSet.add(measurementMNode)) {
188+
// 8 is the memory occupied by the length of the IMeasurementMNode
189+
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
190+
}
191+
if (memorySize > 0) {
192+
requestMemory(memorySize);
193+
}
194+
return tagValueMap;
195+
});
197196
}
198197

199198
public void addIndex(Map<String, String> tagsMap, IMeasurementMNode<?> measurementMNode) {
@@ -208,32 +207,42 @@ public void removeIndex(String tagKey, String tagValue, IMeasurementMNode<?> mea
208207
if (tagKey == null || tagValue == null || measurementMNode == null) {
209208
return;
210209
}
211-
// init memory size
212-
long memorySize = 0;
213-
if (tagIndex.get(tagKey).get(tagValue).remove(measurementMNode)) {
214-
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
215-
}
216-
if (tagIndex.get(tagKey).get(tagValue).isEmpty()) {
217-
if (tagIndex.get(tagKey).remove(tagValue) != null) {
218-
// the last 4 is the memory occupied by the size of IMeasurementMNodeSet
219-
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
220-
}
221-
}
222-
if (tagIndex.get(tagKey).isEmpty()) {
223-
if (tagIndex.remove(tagKey) != null) {
224-
// the last 4 is the memory occupied by the size of tagValueMap
225-
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
226-
}
227-
}
228-
releaseMemory(memorySize);
210+
tagIndex.computeIfPresent(
211+
tagKey,
212+
(key, tagValueMap) -> {
213+
long memorySize = 0;
214+
Set<IMeasurementMNode<?>> measurementsSet = tagValueMap.get(tagValue);
215+
if (measurementsSet == null) {
216+
return tagValueMap;
217+
}
218+
219+
if (measurementsSet.remove(measurementMNode)) {
220+
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
221+
}
222+
if (measurementsSet.isEmpty()) {
223+
if (tagValueMap.remove(tagValue, measurementsSet)) {
224+
// the last 4 is the memory occupied by the size of IMeasurementMNodeSet
225+
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
226+
}
227+
}
228+
if (tagValueMap.isEmpty()) {
229+
// the last 4 is the memory occupied by the size of tagValueMap
230+
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
231+
if (memorySize > 0) {
232+
releaseMemory(memorySize);
233+
}
234+
return null;
235+
}
236+
if (memorySize > 0) {
237+
releaseMemory(memorySize);
238+
}
239+
return tagValueMap;
240+
});
229241
}
230242

231243
private List<IMeasurementMNode<?>> getMatchedTimeseriesInIndex(TagFilter tagFilter) {
232-
if (!tagIndex.containsKey(tagFilter.getKey())) {
233-
return Collections.emptyList();
234-
}
235244
Map<String, Set<IMeasurementMNode<?>>> value2Node = tagIndex.get(tagFilter.getKey());
236-
if (value2Node.isEmpty()) {
245+
if (value2Node == null || value2Node.isEmpty()) {
237246
return Collections.emptyList();
238247
}
239248

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.iotdb.db.schemaengine.schemaregion.tag;
20+
21+
import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode;
22+
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
23+
import org.apache.iotdb.db.schemaengine.rescon.MemSchemaEngineStatistics;
24+
import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
25+
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
26+
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader;
27+
28+
import org.apache.tsfile.enums.TSDataType;
29+
import org.apache.tsfile.external.commons.io.FileUtils;
30+
import org.apache.tsfile.file.metadata.enums.CompressionType;
31+
import org.apache.tsfile.file.metadata.enums.TSEncoding;
32+
import org.apache.tsfile.utils.RamUsageEstimator;
33+
import org.apache.tsfile.write.schema.MeasurementSchema;
34+
import org.junit.After;
35+
import org.junit.Assert;
36+
import org.junit.Test;
37+
38+
import java.io.File;
39+
import java.nio.file.Files;
40+
import java.util.ArrayList;
41+
import java.util.List;
42+
import java.util.concurrent.CountDownLatch;
43+
import java.util.concurrent.ExecutorService;
44+
import java.util.concurrent.Executors;
45+
import java.util.concurrent.Future;
46+
import java.util.concurrent.TimeUnit;
47+
48+
public class TagManagerTest {
49+
50+
private File tempDir;
51+
private MemSchemaRegionStatistics regionStatistics;
52+
private TagManager tagManager;
53+
54+
@After
55+
public void tearDown() throws Exception {
56+
if (tagManager != null) {
57+
tagManager.clear();
58+
}
59+
if (regionStatistics != null) {
60+
regionStatistics.clear();
61+
}
62+
if (tempDir != null) {
63+
FileUtils.deleteDirectory(tempDir);
64+
}
65+
}
66+
67+
@Test
68+
public void removeIndexIgnoresMissingEntriesAndReleasesOnlyExistingMemory() throws Exception {
69+
initTagManager();
70+
final IMeasurementMNode<?> node = newMeasurementMNode("s0");
71+
72+
tagManager.removeIndex("missingKey", "missingValue", node);
73+
Assert.assertEquals(0, regionStatistics.getRegionMemoryUsage());
74+
75+
tagManager.addIndex("key", "value", node);
76+
final long expectedMemory = indexMemory("key", "value", 1);
77+
Assert.assertEquals(expectedMemory, regionStatistics.getRegionMemoryUsage());
78+
79+
tagManager.removeIndex("key", "missingValue", node);
80+
Assert.assertEquals(expectedMemory, regionStatistics.getRegionMemoryUsage());
81+
82+
tagManager.removeIndex("key", "value", newMeasurementMNode("other"));
83+
Assert.assertEquals(expectedMemory, regionStatistics.getRegionMemoryUsage());
84+
85+
tagManager.removeIndex("key", "value", node);
86+
Assert.assertEquals(0, regionStatistics.getRegionMemoryUsage());
87+
88+
tagManager.removeIndex("key", "value", node);
89+
Assert.assertEquals(0, regionStatistics.getRegionMemoryUsage());
90+
}
91+
92+
@Test
93+
public void concurrentAddIndexRequestsMemoryForActualInsertionsOnly() throws Exception {
94+
initTagManager();
95+
final String tagKey = "key";
96+
final String tagValue = "value";
97+
final int measurementCount = 128;
98+
final List<IMeasurementMNode<?>> nodes = new ArrayList<>();
99+
for (int i = 0; i < measurementCount; i++) {
100+
nodes.add(newMeasurementMNode("s" + i));
101+
}
102+
103+
final int workerCount = 16;
104+
final ExecutorService executorService = Executors.newFixedThreadPool(workerCount);
105+
final CountDownLatch readyLatch = new CountDownLatch(workerCount);
106+
final CountDownLatch startLatch = new CountDownLatch(1);
107+
final List<Future<?>> futures = new ArrayList<>();
108+
for (final IMeasurementMNode<?> node : nodes) {
109+
futures.add(
110+
executorService.submit(
111+
() -> {
112+
readyLatch.countDown();
113+
startLatch.await();
114+
tagManager.addIndex(tagKey, tagValue, node);
115+
return null;
116+
}));
117+
}
118+
119+
try {
120+
Assert.assertTrue(readyLatch.await(10, TimeUnit.SECONDS));
121+
startLatch.countDown();
122+
for (final Future<?> future : futures) {
123+
future.get(10, TimeUnit.SECONDS);
124+
}
125+
} finally {
126+
executorService.shutdownNow();
127+
}
128+
Assert.assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
129+
130+
final long expectedMemory = indexMemory(tagKey, tagValue, measurementCount);
131+
Assert.assertEquals(expectedMemory, regionStatistics.getRegionMemoryUsage());
132+
133+
for (final IMeasurementMNode<?> node : nodes) {
134+
tagManager.addIndex(tagKey, tagValue, node);
135+
}
136+
Assert.assertEquals(expectedMemory, regionStatistics.getRegionMemoryUsage());
137+
138+
for (final IMeasurementMNode<?> node : nodes) {
139+
tagManager.removeIndex(tagKey, tagValue, node);
140+
}
141+
Assert.assertEquals(0, regionStatistics.getRegionMemoryUsage());
142+
}
143+
144+
private void initTagManager() throws Exception {
145+
tempDir = Files.createTempDirectory("tag-manager").toFile();
146+
regionStatistics = new MemSchemaRegionStatistics(0, new MemSchemaEngineStatistics());
147+
tagManager = new TagManager(tempDir.getAbsolutePath(), regionStatistics);
148+
}
149+
150+
private static IMeasurementMNode<IMemMNode> newMeasurementMNode(final String measurement) {
151+
final IMNodeFactory<IMemMNode> nodeFactory =
152+
MNodeFactoryLoader.getInstance().getMemMNodeIMNodeFactory();
153+
return nodeFactory.createMeasurementMNode(
154+
null,
155+
measurement,
156+
new MeasurementSchema(
157+
measurement, TSDataType.INT64, TSEncoding.PLAIN, CompressionType.SNAPPY),
158+
null);
159+
}
160+
161+
private static long indexMemory(
162+
final String tagKey, final String tagValue, final int measurementCount) {
163+
return RamUsageEstimator.sizeOf(tagKey)
164+
+ 4
165+
+ RamUsageEstimator.sizeOf(tagValue)
166+
+ 4
167+
+ (RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4) * measurementCount;
168+
}
169+
}

0 commit comments

Comments
 (0)