Skip to content

Commit 8cb4067

Browse files
CaideyipiCaideyipi
committed
Fix queue invariant checks
1 parent eb4409d commit 8cb4067

11 files changed

Lines changed: 399 additions & 0 deletions

File tree

‎iotdb-core/calc-commons/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,11 @@
9999
<groupId>at.yawk.lz4</groupId>
100100
<artifactId>lz4-java</artifactId>
101101
</dependency>
102+
<dependency>
103+
<groupId>junit</groupId>
104+
<artifactId>junit</artifactId>
105+
<scope>test</scope>
106+
</dependency>
102107
</dependencies>
103108
<build>
104109
<plugins>

‎iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingQueue.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@ public abstract class IndexedBlockingQueue<E extends IDIndexedAccessible> {
4242

4343
public static final String TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG =
4444
"The system can't allow more queries.";
45+
public static final String SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG =
46+
"The queue has already contained the same ID element.";
4547

4648
protected final int capacity;
4749
protected final E queryHolder;
@@ -57,6 +59,7 @@ public abstract class IndexedBlockingQueue<E extends IDIndexedAccessible> {
5759
* @throws IllegalArgumentException if maxCapacity <= 0.
5860
*/
5961
protected IndexedBlockingQueue(int maxCapacity, E queryHolder) {
62+
Preconditions.checkArgument(maxCapacity > 0, "maxCapacity must be greater than 0.");
6063
this.capacity = maxCapacity;
6164
this.queryHolder = queryHolder;
6265
}
@@ -93,6 +96,7 @@ public synchronized void push(E element) {
9396
throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
9497
}
9598
Preconditions.checkState(size < capacity, TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG);
99+
Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
96100
pushToQueue(element);
97101
size++;
98102
this.notifyAll();

‎iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingReserveQueue.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ public synchronized void push(E element) {
5757
throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
5858
}
5959
Preconditions.checkState(size + reservedSize < capacity, TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG);
60+
Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
6061
pushToQueue(element);
6162
size++;
6263
this.notifyAll();
@@ -67,6 +68,8 @@ public synchronized void repush(E element) {
6768
if (element == null) {
6869
throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
6970
}
71+
Preconditions.checkState(reservedSize > 0, "No reserved space is available.");
72+
Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
7073
pushToQueue(element);
7174
reservedSize--;
7275
size++;
@@ -77,6 +80,7 @@ public synchronized void repush(E element) {
7780
* For task that is not in readyQueue when it's cleared, it won't be added into the queue again.
7881
*/
7982
public synchronized void decreaseReservedSize() {
83+
Preconditions.checkState(reservedSize > 0, "No reserved space is available.");
8084
this.reservedSize--;
8185
}
8286
}

‎iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingQueueTest.java‎

Lines changed: 226 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,226 @@
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+
20+
package org.apache.iotdb.calc.execution.schedule.queue;
21+
22+
import org.junit.Assert;
23+
import org.junit.Test;
24+
25+
import java.util.ArrayDeque;
26+
import java.util.HashMap;
27+
import java.util.Map;
28+
import java.util.Queue;
29+
30+
public class IndexedBlockingQueueTest {
31+
32+
@Test
33+
public void testRejectNonPositiveCapacity() {
34+
Assert.assertThrows(IllegalArgumentException.class, () -> new SimpleQueue(0));
35+
Assert.assertThrows(IllegalArgumentException.class, () -> new ReserveQueue(-1));
36+
}
37+
38+
@Test
39+
public void testRejectDuplicateIdPush() throws InterruptedException {
40+
SimpleQueue queue = new SimpleQueue(2);
41+
Element first = new Element(1);
42+
Element duplicate = new Element(1);
43+
44+
queue.push(first);
45+
Assert.assertThrows(IllegalStateException.class, () -> queue.push(duplicate));
46+
Assert.assertEquals(1, queue.size());
47+
Assert.assertSame(first, queue.poll());
48+
Assert.assertTrue(queue.isEmpty());
49+
}
50+
51+
@Test
52+
public void testRejectRepushWithoutReservedSpace() {
53+
ReserveQueue queue = new ReserveQueue(1);
54+
55+
Assert.assertThrows(IllegalStateException.class, () -> queue.repush(new Element(1)));
56+
}
57+
58+
@Test
59+
public void testRejectDecreaseWithoutReservedSpace() {
60+
ReserveQueue queue = new ReserveQueue(1);
61+
62+
Assert.assertThrows(IllegalStateException.class, queue::decreaseReservedSize);
63+
}
64+
65+
@Test
66+
public void testReservedSpaceCanOnlyBeReleasedOnce() throws InterruptedException {
67+
ReserveQueue queue = new ReserveQueue(1);
68+
Element element = new Element(1);
69+
queue.push(element);
70+
71+
Assert.assertSame(element, queue.poll());
72+
queue.decreaseReservedSize();
73+
74+
Assert.assertThrows(IllegalStateException.class, queue::decreaseReservedSize);
75+
}
76+
77+
private static class SimpleQueue extends IndexedBlockingQueue<Element> {
78+
private final Queue<Element> elements = new ArrayDeque<>();
79+
private final Map<ID, Element> keyedElements = new HashMap<>();
80+
81+
private SimpleQueue(int capacity) {
82+
super(capacity, new Element(0));
83+
}
84+
85+
@Override
86+
protected Element remove(Element element) {
87+
Element removed = keyedElements.remove(element.getDriverTaskId());
88+
if (removed != null) {
89+
elements.remove(removed);
90+
}
91+
return removed;
92+
}
93+
94+
@Override
95+
protected Element get(Element element) {
96+
return keyedElements.get(element.getDriverTaskId());
97+
}
98+
99+
@Override
100+
public boolean isEmpty() {
101+
return elements.isEmpty();
102+
}
103+
104+
@Override
105+
protected Element pollFirst() {
106+
Element first = elements.remove();
107+
keyedElements.remove(first.getDriverTaskId());
108+
return first;
109+
}
110+
111+
@Override
112+
protected void pushToQueue(Element element) {
113+
elements.add(element);
114+
keyedElements.put(element.getDriverTaskId(), element);
115+
}
116+
117+
@Override
118+
protected boolean contains(Element element) {
119+
return keyedElements.containsKey(element.getDriverTaskId());
120+
}
121+
122+
@Override
123+
protected void clearAllElements() {
124+
elements.clear();
125+
keyedElements.clear();
126+
}
127+
}
128+
129+
private static class ReserveQueue extends IndexedBlockingReserveQueue<Element> {
130+
private final Queue<Element> elements = new ArrayDeque<>();
131+
private final Map<ID, Element> keyedElements = new HashMap<>();
132+
133+
private ReserveQueue(int capacity) {
134+
super(capacity, new Element(0));
135+
}
136+
137+
@Override
138+
protected Element remove(Element element) {
139+
Element removed = keyedElements.remove(element.getDriverTaskId());
140+
if (removed != null) {
141+
elements.remove(removed);
142+
}
143+
return removed;
144+
}
145+
146+
@Override
147+
protected Element get(Element element) {
148+
return keyedElements.get(element.getDriverTaskId());
149+
}
150+
151+
@Override
152+
public boolean isEmpty() {
153+
return elements.isEmpty();
154+
}
155+
156+
@Override
157+
protected Element pollFirst() {
158+
Element first = elements.remove();
159+
keyedElements.remove(first.getDriverTaskId());
160+
return first;
161+
}
162+
163+
@Override
164+
protected void pushToQueue(Element element) {
165+
elements.add(element);
166+
keyedElements.put(element.getDriverTaskId(), element);
167+
}
168+
169+
@Override
170+
protected boolean contains(Element element) {
171+
return keyedElements.containsKey(element.getDriverTaskId());
172+
}
173+
174+
@Override
175+
protected void clearAllElements() {
176+
elements.clear();
177+
keyedElements.clear();
178+
}
179+
}
180+
181+
private static class Element implements IDIndexedAccessible {
182+
private ElementId id;
183+
184+
private Element(int id) {
185+
this.id = new ElementId(id);
186+
}
187+
188+
@Override
189+
public ID getDriverTaskId() {
190+
return id;
191+
}
192+
193+
@Override
194+
public void setId(ID id) {
195+
this.id = (ElementId) id;
196+
}
197+
198+
@Override
199+
public boolean equals(Object obj) {
200+
return obj instanceof Element && ((Element) obj).id.equals(id);
201+
}
202+
203+
@Override
204+
public int hashCode() {
205+
return id.hashCode();
206+
}
207+
}
208+
209+
private static class ElementId implements ID {
210+
private final int id;
211+
212+
private ElementId(int id) {
213+
this.id = id;
214+
}
215+
216+
@Override
217+
public boolean equals(Object obj) {
218+
return obj instanceof ElementId && ((ElementId) obj).id == id;
219+
}
220+
221+
@Override
222+
public int hashCode() {
223+
return Integer.hashCode(id);
224+
}
225+
}
226+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/concurrent/FiniteSemaphore.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,9 @@ public class FiniteSemaphore {
3232
private int permit;
3333

3434
public FiniteSemaphore(int capacity, int permit) {
35+
if (capacity < 0 || permit < 0) {
36+
throw new IllegalArgumentException("Capacity and initial permits should be non-negative.");
37+
}
3538
if (capacity < permit) {
3639
throw new IllegalArgumentException(DataNodeMiscMessages.CAPACITY_LARGER_THAN_INITIAL_PERMITS);
3740
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,9 @@ public class FixedPriorityBlockingQueue<T> {
4343
protected List<PollLastHook> pollLastHookList = new CopyOnWriteArrayList<>();
4444

4545
public FixedPriorityBlockingQueue(int maxSize, Comparator<T> comparator) {
46+
if (maxSize <= 0) {
47+
throw new IllegalArgumentException("maxSize must be greater than 0.");
48+
}
4649
this.maxSize = maxSize;
4750
this.comparator = comparator;
4851
this.queue = MinMaxPriorityQueue.orderedBy(comparator).maximumSize(maxSize + 1).create();

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L1PriorityQueueTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,30 @@ public void testPushExceedCapacity() {
8686
}
8787
}
8888

89+
@Test
90+
public void testPushDuplicateId() throws InterruptedException {
91+
IndexedBlockingQueue<QueueElement> queue =
92+
new L1PriorityQueue<>(
93+
10,
94+
(o1, o2) -> {
95+
if (o1.equals(o2)) {
96+
return 0;
97+
}
98+
return Integer.compare(o1.getValue(), o2.getValue());
99+
},
100+
new QueueElement(new QueueElement.QueueElementID(0), 0));
101+
QueueElement.QueueElementID id = new QueueElement.QueueElementID(1);
102+
QueueElement e1 = new QueueElement(id, 1);
103+
QueueElement e2 = new QueueElement(id, 2);
104+
queue.push(e1);
105+
106+
Assert.assertThrows(IllegalStateException.class, () -> queue.push(e2));
107+
Assert.assertEquals(1, queue.size());
108+
Assert.assertEquals(e1, queue.poll());
109+
Assert.assertEquals(0, queue.size());
110+
Assert.assertTrue(queue.isEmpty());
111+
}
112+
89113
@Test
90114
public void testPushAndPoll() throws InterruptedException {
91115
IndexedBlockingQueue<QueueElement> queue =

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L2PriorityQueueTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,30 @@ public void testPushExceedCapacity() {
9191
}
9292
}
9393

94+
@Test
95+
public void testPushDuplicateId() throws InterruptedException {
96+
IndexedBlockingQueue<QueueElement> queue =
97+
new L2PriorityQueue<>(
98+
10,
99+
(o1, o2) -> {
100+
if (o1.equals(o2)) {
101+
return 0;
102+
}
103+
return Integer.compare(o1.getValue(), o2.getValue());
104+
},
105+
new QueueElement(new QueueElement.QueueElementID(0), 0));
106+
QueueElement.QueueElementID id = new QueueElement.QueueElementID(1);
107+
QueueElement e1 = new QueueElement(id, 1);
108+
QueueElement e2 = new QueueElement(id, 2);
109+
queue.push(e1);
110+
111+
Assert.assertThrows(IllegalStateException.class, () -> queue.push(e2));
112+
Assert.assertEquals(1, queue.size());
113+
Assert.assertEquals(e1, queue.poll());
114+
Assert.assertEquals(0, queue.size());
115+
Assert.assertTrue(queue.isEmpty());
116+
}
117+
94118
@Test
95119
public void testPushAndPoll() throws InterruptedException {
96120
IndexedBlockingQueue<QueueElement> queue =

0 commit comments

Comments
 (0)