[RMA] Feature Optimize ObjectFlusher test.
MODIFY
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/storage/impl/ObjectFlusher.java
BackEnd/Core/sailfish-core/src/main/java/com/exactpro/sf/storage/impl/ObjectFlusher.java
@@ -40,8 +40,10 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
private final Logger logger = LoggerFactory.getLogger(getClass().getName() + "@" + Integer.toHexString(hashCode()));
private static final Logger USER_EVENTS_LOG = CommonLoggers.USER_EVENTS_LOGGER;
+ private static final long DEFAULT_FLUSH_TASK_TIMEOUT = 1000;
private static final long JOIN_TIMEOUT = 2000;
+
private final Lock monitor = new ReentrantLock(true);
private final ReadWriteLock sourceMonitor = new ReentrantReadWriteLock();
private final Condition needFlush = monitor.newCondition();
@@ -49,6 +51,7 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
private final int bufferSize;
private final long maxStorageQueueSize;
+ private final long flushTaskTimeout;
private long storageProviderQueueSize;
private long lastThrottleNotification;
@@ -104,10 +107,15 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
}
public ObjectFlusher(IFlushProvider<T> provider, int bufferSize, long maxStorageQueueSize) {
+ this(provider, bufferSize, maxStorageQueueSize, DEFAULT_FLUSH_TASK_TIMEOUT);
+ }
+
+ public ObjectFlusher(IFlushProvider<T> provider, int bufferSize, long maxStorageQueueSize, long flushTaskTimeout) {
this.provider = Objects.requireNonNull(provider, "provider cannot be null");
this.bufferSize = bufferSize;
this.objects = new ArrayList<>(bufferSize + 1);
this.maxStorageQueueSize = maxStorageQueueSize;
+ this.flushTaskTimeout = flushTaskTimeout;
}
@Override
@@ -118,7 +126,7 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
throw new EPSCommonException("Cannot start flusher. Flusher is already started");
}
- flushTask = new FlushTask();
+ flushTask = new FlushTask(flushTaskTimeout);
flushThread = new Thread(flushTask, logger.getName());
flushThread.setDaemon(true);
flushThread.start();
@@ -230,7 +238,11 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
}
public class FlushTask implements Runnable {
- private static final long TIMEOUT = 1000;
+ private final long timeout;
+
+ public FlushTask(long timeout) {
+ this.timeout = timeout;
+ }
@Override
public void run() {
@@ -258,7 +270,7 @@ public class ObjectFlusher<T extends IMeasurable> implements IObjectFlusher<T> {
logger.debug("monitor locked");
if (wait) {
- needFlush.await(TIMEOUT, TimeUnit.MILLISECONDS);
+ needFlush.await(timeout, TimeUnit.MILLISECONDS);
}
temp = objects;

MODIFY
BackEnd/Core/sailfish-core/src/test/java/com/exactpro/sf/TestObjectFlusher.java
BackEnd/Core/sailfish-core/src/test/java/com/exactpro/sf/TestObjectFlusher.java
package com.exactpro.sf;
+import java.io.Closeable;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.junit.Assert;
import org.junit.Test;
-
-import java.io.Closeable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.exactpro.sf.storage.IMeasurable;
import com.exactpro.sf.storage.impl.ObjectFlusher;
public class TestObjectFlusher {
- private static final long timeToRun = 1000 * 10;
+ private static final int COUNT_ELEMENTS = 32;
+ private static final int SIZE_OF_OBJECT = 1024 * 1024;
+ private static final int BUFFER_SIZE = COUNT_ELEMENTS * 2;
+ private static final int ADD_ELEMENTS_FACTOR = 64;
+ private static final long TASK_TIMEOUT = 5;
+
+ private static final Logger logger = LoggerFactory.getLogger(TestObjectFlusher.class);
@Test
public void testOOM() throws Exception {
- ObjectFlusher<OmNomNom> objectFlusher = new ObjectFlusher<>(o -> {
- System.out.println("try to flush objects " + o.size());
- Assert.assertEquals(32, o.size());//check that limit is not exceeded
- System.out.println("objects flushed");
- }, 32*2, 32*1024*1024); //flush thereshold is 64 but limit lower (only 32 objects)
-
- try (Closeable closeable = objectFlusher::stop) {
+ AtomicInteger consumerCounter = new AtomicInteger();
+ ObjectFlusher<OmNomNom> objectFlusher = new ObjectFlusher<>(
+ list -> consumerCounter.addAndGet(list.size()),
+ COUNT_ELEMENTS * 2,
+ COUNT_ELEMENTS * SIZE_OF_OBJECT,
+ TASK_TIMEOUT
+ );
- long startTime = System.currentTimeMillis();
objectFlusher.start();
- while (System.currentTimeMillis() - startTime <= timeToRun) {
+
+ try (Closeable closeable = objectFlusher::stop) {
+ for (int i = 0; i < BUFFER_SIZE * ADD_ELEMENTS_FACTOR; i++) {
objectFlusher.add(new OmNomNom());
}
}
+ logger.debug("Counting elements: %d. Max elements: %d", consumerCounter.get(), BUFFER_SIZE * ADD_ELEMENTS_FACTOR);
+ System.out.println(consumerCounter.get());
+ Assert.assertTrue(consumerCounter.get() < BUFFER_SIZE * ADD_ELEMENTS_FACTOR);
}
private class OmNomNom implements IMeasurable {
- private byte[] eatRam = new byte[1024*1024];
-
@Override
public long getSize() {
- return eatRam.length;
- }
-
- public byte[] getEatRam() {
- return eatRam;
- }
-
- public void setEatRam(byte[] eatRam) {
- this.eatRam = eatRam;
+ return SIZE_OF_OBJECT;
}
}

[RMA] Feature #49122. Optimize ObjectFlusher test.

Change-Id: I0ddd7e06e53c06d9fba4d8d4940b09656c8876cc