Skip to content

Commit 570aafa

Browse files
CaideyipiCaideyipi
authored
[Bugfix] Reclaim RPC buffers by evicting idle clients (#18853) (#18882)
* [Bugfix] Reclaim idle RPC buffers when the shared pool is exhausted * [Bugfix] Evict idle RPC clients to release buffers (cherry picked from commit 1b290ae)
1 parent f6b122c commit 570aafa

2 files changed

Lines changed: 288 additions & 1 deletion

File tree

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/property/ClientPoolProperty.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,9 @@ public ClientPoolProperty<V> build() {
101101
poolConfig.setMaxIdlePerKey(maxIdleClientNumForEachNode);
102102
poolConfig.setTimeBetweenEvictionRuns(Duration.ofMillis(timeBetweenEvictionRuns));
103103
poolConfig.setMinEvictableIdleTime(Duration.ofMillis(minIdleTimeForClient));
104+
// -1 checks all idle clients across all nodes in each run. The default of 3 can leave
105+
// expired connections (and their RPC buffers) in a large pool for hours.
106+
poolConfig.setNumTestsPerEvictionRun(-1);
104107
poolConfig.setMaxWait(Duration.ofMillis(waitClientTimeoutMs));
105108
poolConfig.setTestOnReturn(true);
106109
poolConfig.setTestOnBorrow(true);
@@ -114,7 +117,7 @@ private DefaultProperty() {}
114117

115118
public static final long WAIT_CLIENT_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
116119
public static final long MIN_IDLE_TIME_FOR_CLIENT_MS = TimeUnit.MINUTES.toMillis(1);
117-
public static final long TIME_BETWEEN_EVICTION_RUNS_MS = TimeUnit.MINUTES.toMillis(1);
120+
public static final long TIME_BETWEEN_EVICTION_RUNS_MS = TimeUnit.SECONDS.toMillis(30);
118121
public static final int MAX_CLIENT_NUM_FOR_EACH_NODE = 1000;
119122
public static final int MAX_IDLE_CLIENT_NUM_FOR_EACH_NODE = 1000;
120123
}

‎iotdb-core/node-commons/src/test/java/org/apache/iotdb/rpc/ClientPoolIdleEvictionTest.java‎

Lines changed: 284 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,284 @@
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.rpc;
21+
22+
import org.apache.iotdb.commons.client.property.ClientPoolProperty;
23+
24+
import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
25+
import org.apache.commons.pool2.PooledObject;
26+
import org.apache.commons.pool2.impl.DefaultPooledObject;
27+
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
28+
import org.apache.thrift.protocol.TBinaryProtocol;
29+
import org.apache.thrift.protocol.TProtocol;
30+
import org.apache.thrift.server.TServer;
31+
import org.apache.thrift.server.TThreadPoolServer;
32+
import org.apache.thrift.transport.TMemoryBuffer;
33+
import org.apache.thrift.transport.TServerSocket;
34+
import org.apache.thrift.transport.TSocket;
35+
import org.apache.thrift.transport.TTransport;
36+
import org.apache.thrift.transport.TTransportFactory;
37+
import org.junit.Before;
38+
import org.junit.Test;
39+
import org.junit.runner.RunWith;
40+
import org.junit.runners.Parameterized;
41+
42+
import java.net.InetSocketAddress;
43+
import java.net.ServerSocket;
44+
import java.nio.ByteBuffer;
45+
import java.time.Duration;
46+
import java.util.ArrayList;
47+
import java.util.Arrays;
48+
import java.util.List;
49+
import java.util.concurrent.CopyOnWriteArrayList;
50+
import java.util.concurrent.ExecutorService;
51+
import java.util.concurrent.Executors;
52+
import java.util.concurrent.TimeUnit;
53+
54+
import static org.awaitility.Awaitility.await;
55+
import static org.junit.Assert.assertArrayEquals;
56+
import static org.junit.Assert.assertEquals;
57+
import static org.junit.Assert.assertFalse;
58+
import static org.junit.Assert.assertSame;
59+
import static org.junit.Assert.assertTrue;
60+
61+
@RunWith(Parameterized.class)
62+
public class ClientPoolIdleEvictionTest {
63+
64+
private static final int INITIAL_BUFFER_SIZE = 32;
65+
private static final byte[] PAYLOAD = new byte[4096];
66+
67+
private final boolean compression;
68+
69+
public ClientPoolIdleEvictionTest(boolean compression) {
70+
this.compression = compression;
71+
}
72+
73+
@Parameterized.Parameters(name = "compression={0}")
74+
public static Iterable<Object[]> parameters() {
75+
return Arrays.asList(new Object[][] {{false}, {true}});
76+
}
77+
78+
@Before
79+
public void setUp() {
80+
Arrays.fill(PAYLOAD, (byte) 13);
81+
}
82+
83+
@Test
84+
public void testOneRunEvictsAllExpiredClientsAcrossNodes() throws Exception {
85+
try (GenericKeyedObjectPool<Integer, TElasticFramedTransport> pool =
86+
createPool(100, -1, null)) {
87+
TElasticFramedTransport active = pool.borrowObject(0);
88+
try {
89+
List<TElasticFramedTransport> idleClients = new ArrayList<>();
90+
for (int i = 0; i < 10; i++) {
91+
idleClients.add(pool.borrowObject(i % 2));
92+
}
93+
for (int i = 0; i < idleClients.size(); i++) {
94+
pool.returnObject(i % 2, idleClients.get(i));
95+
}
96+
97+
assertEquals(10, pool.getNumIdle());
98+
99+
// Wait for the idle timeout with the background evictor disabled, then run one sweep.
100+
waitForIdleTimeout();
101+
pool.evict();
102+
103+
assertEquals(0, pool.getNumIdle());
104+
assertEquals(10, pool.getDestroyedCount());
105+
assertEquals(1, pool.getNumActive());
106+
107+
// Eviction must preserve a borrowed client's frame and allow a new client to connect.
108+
byte[] actual = new byte[PAYLOAD.length];
109+
active.readAll(actual, 0, actual.length);
110+
assertArrayEquals(PAYLOAD, actual);
111+
TElasticFramedTransport replacement = pool.borrowObject(2);
112+
pool.returnObject(2, replacement);
113+
} finally {
114+
pool.returnObject(0, active);
115+
}
116+
}
117+
}
118+
119+
@Test
120+
public void testBackgroundEvictionClosesIdleClients() throws Exception {
121+
try (GenericKeyedObjectPool<Integer, TElasticFramedTransport> pool =
122+
createPool(100, 20, null)) {
123+
TElasticFramedTransport client = pool.borrowObject(0);
124+
pool.returnObject(0, client);
125+
126+
await()
127+
.atMost(5, TimeUnit.SECONDS)
128+
.pollInterval(Duration.ofMillis(10))
129+
.until(() -> pool.getNumIdle() == 0 && pool.getDestroyedCount() == 1);
130+
assertEquals(1, pool.getDestroyedCount());
131+
}
132+
}
133+
134+
@Test
135+
public void testUnexpiredClientRemainsReusable() throws Exception {
136+
try (GenericKeyedObjectPool<Integer, TElasticFramedTransport> pool =
137+
createPool(TimeUnit.MINUTES.toMillis(1), -1, null)) {
138+
TElasticFramedTransport client = pool.borrowObject(0);
139+
pool.returnObject(0, client);
140+
141+
pool.evict();
142+
assertEquals(1, pool.getNumIdle());
143+
assertEquals(0, pool.getDestroyedCount());
144+
145+
TElasticFramedTransport reused = pool.borrowObject(0);
146+
try {
147+
assertSame(client, reused);
148+
byte[] actual = new byte[PAYLOAD.length];
149+
reused.readAll(actual, 0, actual.length);
150+
assertArrayEquals(PAYLOAD, actual);
151+
} finally {
152+
pool.returnObject(0, reused);
153+
}
154+
}
155+
}
156+
157+
@Test(timeout = 20000)
158+
public void testEvictionClosesServerConnections() throws Exception {
159+
try (ServerSocket socket = new ServerSocket()) {
160+
socket.bind(new InetSocketAddress("127.0.0.1", 0));
161+
ExecutorService workers = Executors.newFixedThreadPool(2);
162+
List<TTransport> serverConnections = new CopyOnWriteArrayList<>();
163+
TServer server =
164+
new TThreadPoolServer(
165+
new TThreadPoolServer.Args(new TServerSocket(socket))
166+
.executorService(workers)
167+
.transportFactory(
168+
new TTransportFactory() {
169+
@Override
170+
public TTransport getTransport(TTransport transport) {
171+
serverConnections.add(transport);
172+
return createTransport(transport);
173+
}
174+
})
175+
.protocolFactory(new TBinaryProtocol.Factory())
176+
.processor(
177+
(input, output) -> {
178+
output.writeBinary(input.readBinary());
179+
output.getTransport().flush();
180+
}));
181+
Thread serverThread = new Thread(server::serve, "client-pool-idle-eviction-test-server");
182+
serverThread.setDaemon(true);
183+
serverThread.start();
184+
185+
try (GenericKeyedObjectPool<Integer, TElasticFramedTransport> pool =
186+
createPool(100, -1, socket.getLocalPort())) {
187+
TElasticFramedTransport first = pool.borrowObject(0);
188+
try {
189+
echo(first, PAYLOAD);
190+
} finally {
191+
pool.returnObject(0, first);
192+
}
193+
assertFalse(serverConnections.isEmpty());
194+
195+
waitForIdleTimeout();
196+
pool.evict();
197+
assertFalse(first.isOpen());
198+
// The server must observe EOF and close its side of the connection as well.
199+
await()
200+
.atMost(5, TimeUnit.SECONDS)
201+
.pollInterval(Duration.ofMillis(10))
202+
.until(() -> serverConnections.stream().noneMatch(TTransport::isOpen));
203+
204+
TElasticFramedTransport replacement = pool.borrowObject(1);
205+
try {
206+
echo(replacement, new byte[] {1, 2, 3, 4});
207+
} finally {
208+
pool.returnObject(1, replacement);
209+
}
210+
} finally {
211+
server.stop();
212+
workers.shutdownNow();
213+
assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
214+
serverThread.join(5000);
215+
assertFalse(serverThread.isAlive());
216+
}
217+
}
218+
}
219+
220+
private static void echo(TElasticFramedTransport transport, byte[] payload) throws Exception {
221+
TProtocol protocol = new TBinaryProtocol(transport);
222+
protocol.writeBinary(ByteBuffer.wrap(payload));
223+
transport.flush();
224+
assertEquals(ByteBuffer.wrap(payload), protocol.readBinary());
225+
}
226+
227+
private static void waitForIdleTimeout() {
228+
long returnedAt = System.nanoTime();
229+
await()
230+
.atMost(5, TimeUnit.SECONDS)
231+
.pollInterval(Duration.ofMillis(10))
232+
.until(() -> System.nanoTime() - returnedAt > TimeUnit.MILLISECONDS.toNanos(200));
233+
}
234+
235+
private GenericKeyedObjectPool<Integer, TElasticFramedTransport> createPool(
236+
long idleTimeoutMs, long evictionIntervalMs, Integer serverPort) {
237+
return new GenericKeyedObjectPool<>(
238+
new BaseKeyedPooledObjectFactory<Integer, TElasticFramedTransport>() {
239+
@Override
240+
public TElasticFramedTransport create(Integer node) throws Exception {
241+
TElasticFramedTransport transport =
242+
createTransport(
243+
serverPort == null
244+
? new TMemoryBuffer(0)
245+
: new TSocket("127.0.0.1", serverPort, 5000));
246+
try {
247+
if (serverPort == null) {
248+
transport.write(PAYLOAD, 0, PAYLOAD.length);
249+
transport.flush();
250+
} else {
251+
transport.open();
252+
}
253+
return transport;
254+
} catch (Exception e) {
255+
transport.close();
256+
throw e;
257+
}
258+
}
259+
260+
@Override
261+
public PooledObject<TElasticFramedTransport> wrap(TElasticFramedTransport transport) {
262+
return new DefaultPooledObject<>(transport);
263+
}
264+
265+
@Override
266+
public void destroyObject(Integer node, PooledObject<TElasticFramedTransport> transport) {
267+
transport.getObject().close();
268+
}
269+
},
270+
new ClientPoolProperty.Builder<TElasticFramedTransport>()
271+
.setMaxClientNumForEachNode(16)
272+
.setMaxIdleClientNumForEachNode(16)
273+
.setMinIdleTimeForClient(idleTimeoutMs)
274+
.setTimeBetweenEvictionRuns(evictionIntervalMs)
275+
.build()
276+
.getConfig());
277+
}
278+
279+
private TElasticFramedTransport createTransport(TTransport underlying) {
280+
return compression
281+
? new TSnappyElasticFramedTransport(underlying, INITIAL_BUFFER_SIZE, 65536, true)
282+
: new TElasticFramedTransport(underlying, INITIAL_BUFFER_SIZE, 65536, true);
283+
}
284+
}

0 commit comments

Comments
 (0)