tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

test: fixed class DefaultTbQueueRequestTemplate

pull/4753/head
Sergey Matvienko 5 years ago
parent
9666986156
commit
28235732c6
2 changed files with 73 additions and 54 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 90
      common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java
  2. 37
      common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java

90
common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java
View File

@ -36,10 +36,12 @@ import javax.annotation.Nullable;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response extends TbQueueMsg> extends AbstractTbQueueTemplate
@ -48,16 +50,15 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
private final TbQueueAdmin queueAdmin;
private final TbQueueProducer<Request> requestTemplate;
private final TbQueueConsumer<Response> responseTemplate;
final ConcurrentMap<UUID, DefaultTbQueueRequestTemplate.ResponseMetaData<Response>> pendingRequests;
final ConcurrentHashMap<UUID, DefaultTbQueueRequestTemplate.ResponseMetaData<Response>> pendingRequests = new ConcurrentHashMap<>();
final boolean internalExecutor;
final ExecutorService executor;
final long maxRequestTimeout;
final long maxRequestTimeoutNs;
final long maxPendingRequests;
final long pollInterval;
volatile long tickTs = 0L;
volatile long tickSize = 0L;
volatile boolean stopped = false;
long nextCleanupMs = 0L;
long nextCleanupNs = 0L;
private final Lock cleanerLock = new ReentrantLock();
private MessagesStats messagesStats;
@ -72,8 +73,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
this.queueAdmin = queueAdmin;
this.requestTemplate = requestTemplate;
this.responseTemplate = responseTemplate;
this.pendingRequests = new ConcurrentHashMap<>();
this.maxRequestTimeout = maxRequestTimeout;
this.maxRequestTimeoutNs = TimeUnit.MILLISECONDS.toNanos(maxRequestTimeout);
this.maxPendingRequests = maxPendingRequests;
this.pollInterval = pollInterval;
this.internalExecutor = (executor == null);
@ -88,7 +88,6 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
public void init() {
queueAdmin.createTopicIfNotExists(responseTemplate.getTopic());
requestTemplate.init();
tickTs = getCurrentTime();
responseTemplate.subscribe();
executor.submit(this::mainLoop);
}
@ -105,7 +104,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
}
void fetchAndProcessResponses() {
final int pendingRequestsCount = pendingRequests.size();
final long pendingRequestsCount = pendingRequests.mappingCount();
log.info("Starting template pool topic {}, for pendingRequests {}", responseTemplate.getTopic(), pendingRequestsCount);
List<Response> responses = doPoll(); //poll js responses
//if (responses.size() > 0) {
@ -113,25 +112,36 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
//}
responses.forEach(this::processResponse); //this can take a long time
responseTemplate.commit();
tickTs = getCurrentTime();
tickSize = pendingRequests.size();
if (nextCleanupMs < tickTs) {
//cleanup;
pendingRequests.forEach((key, value) -> {
if (value.expTime < tickTs) {
ResponseMetaData<Response> staleRequest = pendingRequests.remove(key);
if (staleRequest != null) {
setTimeoutException(key, staleRequest, tickTs);
tryCleanStaleRequests();
}
private boolean tryCleanStaleRequests() {
if (!cleanerLock.tryLock()) {
return false;
}
try {
log.trace("tryCleanStaleRequest...");
final long currentNs = getCurrentClockNs();
if (nextCleanupNs < currentNs) {
pendingRequests.forEach((key, value) -> {
if (value.expTime < currentNs) {
ResponseMetaData<Response> staleRequest = pendingRequests.remove(key);
if (staleRequest != null) {
setTimeoutException(key, staleRequest, currentNs);
}
}
}
});
setupNextCleanup();
});
setupNextCleanup();
}
} finally {
cleanerLock.unlock();
}
return true;
}
void setupNextCleanup() {
nextCleanupMs = tickTs + maxRequestTimeout;
log.info("setupNextCleanup {}", nextCleanupMs);
nextCleanupNs = getCurrentClockNs() + maxRequestTimeoutNs;
log.info("setupNextCleanup {}", nextCleanupNs);
}
List<Response> doPoll() {
@ -146,11 +156,11 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
}
}
void setTimeoutException(UUID key, ResponseMetaData<Response> staleRequest, long tickTs) {
if (tickTs >= staleRequest.getSubmitTime() + staleRequest.getTimeout()) {
log.info("Request timeout detected, tickTs [{}], {}, key [{}]", tickTs, staleRequest, key);
void setTimeoutException(UUID key, ResponseMetaData<Response> staleRequest, long currentNs) {
if (currentNs >= staleRequest.getSubmitTime() + staleRequest.getTimeout()) {
log.info("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key);
} else {
log.error("Request timeout detected, tickTs [{}], {}, key [{}]", tickTs, staleRequest, key);
log.error("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key);
}
staleRequest.future.setException(new TimeoutException());
@ -197,23 +207,31 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
@Override
public ListenableFuture<Response> send(Request request) {
if (tickSize > maxPendingRequests) {
if (pendingRequests.mappingCount() >= maxPendingRequests) {
log.warn("Pending request map is full [{}]! Consider to increase maxPendingRequests or increase processing performance", maxPendingRequests);
return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!"));
}
UUID requestId = UUID.randomUUID();
request.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId));
request.getHeaders().put(RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic()));
long currentTime = getCurrentTime();
request.getHeaders().put(REQUEST_TIME, longToBytes(currentTime));
request.getHeaders().put(REQUEST_TIME, longToBytes(getCurrentTimeMs()));
long currentClockNs = getCurrentClockNs();
SettableFuture<Response> future = SettableFuture.create();
ResponseMetaData<Response> responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future, currentTime, maxRequestTimeout);
log.info("pending {}", responseMetaData);
pendingRequests.putIfAbsent(requestId, responseMetaData);
ResponseMetaData<Response> responseMetaData = new ResponseMetaData<>(currentClockNs + maxRequestTimeoutNs, future, currentClockNs, maxRequestTimeoutNs);
log.info("pending {}", responseMetaData); //TODO trace
if (pendingRequests.putIfAbsent(requestId, responseMetaData) != null) {
log.warn("Pending request already exists [{}]!", maxPendingRequests);
return Futures.immediateFailedFuture(new RuntimeException("Pending request already exists !" + requestId));
}
sendToRequestTemplate(request, requestId, future, responseMetaData);
return future;
}
long getCurrentTime() {
long getCurrentClockNs() {
return System.nanoTime(); //MONOTONIC clock instead wall clock
}
long getCurrentTimeMs() { //Wall clock to send Ts to the an external service
return System.currentTimeMillis();
}
@ -261,8 +279,8 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
return "ResponseMetaData{" +
"submitTime=" + submitTime +
", calculatedExpTime=" + (submitTime + timeout) +
", expTime=" + expTime +
", deltaMs=" + (expTime - submitTime) +
", expTime=" + expTime +
", future=" + future +
'}';
}

37
common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java
View File

@ -29,6 +29,7 @@
* OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART.
*/
package org.thingsboard.server.queue.common;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
import org.junit.Before;
@ -52,9 +53,9 @@ import java.util.concurrent.atomic.AtomicLong;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willDoNothing;
@ -86,7 +87,7 @@ public class DefaultTbQueueRequestTemplateTest {
ExecutorService executor;
String topic = "js-responses-tb-node-0";
long maxRequestTimeout = 10;
long maxPendingRequests = 32;
long maxPendingRequests = 1000;
long pollInterval = 5;
DefaultTbQueueRequestTemplate inst;
@ -109,12 +110,12 @@ public class DefaultTbQueueRequestTemplateTest {
@Test
public void givenInstance_whenVerifyInitialParameters_thenOK() {
assertEquals(maxPendingRequests, inst.maxPendingRequests);
assertEquals(maxRequestTimeout, inst.maxRequestTimeout);
assertEquals(pollInterval, inst.pollInterval);
assertEquals(executorMock, inst.executor);
assertFalse(inst.stopped);
assertFalse(inst.internalExecutor);
assertThat(inst.maxPendingRequests, equalTo(maxPendingRequests));
assertThat(inst.maxRequestTimeoutNs, equalTo(TimeUnit.MILLISECONDS.toNanos(maxRequestTimeout)));
assertThat(inst.pollInterval, equalTo(pollInterval));
assertThat(inst.executor, is(executorMock));
assertThat(inst.stopped, is(false));
assertThat(inst.internalExecutor, is(false));
}
@Test
@ -122,8 +123,8 @@ public class DefaultTbQueueRequestTemplateTest {
willDoNothing().given(inst).mainLoop();
inst.init();
assertNotEquals(0, inst.tickTs);
assertEquals(0, inst.nextCleanupMs);
//assertNotEquals(0, inst.tickTs);
assertEquals(0, inst.nextCleanupNs);
verify(queueAdmin, times(1)).createTopicIfNotExists(topic);
verify(requestTemplate, times(1)).init();
verify(responseTemplate, times(1)).subscribe();
@ -168,7 +169,7 @@ public class DefaultTbQueueRequestTemplateTest {
for (int i = 0; i < msgCount; i++) {
inst.send(getRequestMsgMock());
}
assertEquals(msgCount, inst.pendingRequests.size());
assertEquals(msgCount, inst.pendingRequests.mappingCount());
verify(inst, times(msgCount)).sendToRequestTemplate(any(), any(), any(), any());
}
@ -176,15 +177,14 @@ public class DefaultTbQueueRequestTemplateTest {
public void givenMessagesOverMaxPendingRequests_whenSend_thenImmediateFailedFutureForTheOfRequests() {
willDoNothing().given(inst).sendToRequestTemplate(any(), any(), any(), any());
inst.init();
assertEquals(0, inst.tickSize);
int msgOverflowCount = 10;
for (int i = 0; i < inst.maxPendingRequests; i++) {
assertFalse(inst.send(getRequestMsgMock()).isDone()); //SettableFuture future - pending only
}
for (int i = 0; i < msgOverflowCount; i++) {
assertFalse("max pending requests overflow", inst.send(getRequestMsgMock()).isDone()); //overflow, immediate failed future
assertTrue("max pending requests overflow", inst.send(getRequestMsgMock()).isDone()); //overflow, immediate failed future
}
assertThat(inst.pendingRequests.size(), equalTo(inst.maxPendingRequests));
assertThat(inst.pendingRequests.mappingCount(), equalTo(inst.maxPendingRequests));
verify(inst, times((int) inst.maxPendingRequests)).sendToRequestTemplate(any(), any(), any(), any());
}
@ -195,17 +195,18 @@ public class DefaultTbQueueRequestTemplateTest {
willAnswer(x -> {
log.info("currentTime={}", currentTime.get());
return currentTime.get();
}).given(inst).getCurrentTime();
}).given(inst).getCurrentClockNs();
inst.init();
inst.setupNextCleanup();
willReturn(Collections.emptyList()).given(inst).doPoll();
willDoNothing().given(inst).processResponse(any());
//when
for (int i = 0; i <= inst.maxRequestTimeout*2; i++) {
currentTime.incrementAndGet();
long stepNs = TimeUnit.MILLISECONDS.toNanos(1);
for (long i = 0; i <= inst.maxRequestTimeoutNs * 2; i = i + stepNs) {
currentTime.addAndGet(stepNs);
assertFalse(inst.send(getRequestMsgMock()).isDone()); //SettableFuture future - pending only
if (i % (inst.maxRequestTimeout * 3 / 2) == 0) {
if (i % (inst.maxRequestTimeoutNs * 3 / 2) == 0) {
inst.fetchAndProcessResponses();
}
}

Write Preview
Loading…
Cancel
Save