[jira] [Commented] (IGNITE-7918) Huge memory leak when data streamer used together with local cache

["Zbyszek B (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/IGNITE-7918?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16395086#comment-16395086 ] 

Zbyszek B commented on IGNITE-7918:
-----------------------------------

{code:java}
package leak;

import org.apache.ignite.*;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.binary.BinaryObjectBuilder;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;

import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;


public class Demo {

private final static Ignite ignite = createIgnite();

private final static AtomicLong producedCount = new AtomicLong();
private final static AtomicLong consumedCount = new AtomicLong();
private final static int numOfLocalFields = 10;
private static BinaryObject localPrototype = createLocalPrototype(); // prototype of local cache entity with all fields set to null

private final static int numOfGlobalFields = 5;
private final static List<String> globalFields = globalFields();
private static BinaryObject globalPrototype = createGlobalPrototype(); // prototype of cluster-wide cache entity with all fields set to null
private static final IgniteCache<String, BinaryObject> globalCache = getOrCreateGlobalCache();

private final static String ID = "id";

private final static boolean useLocalCacheToCauseMemLeak = true;

private final static BlockingQueue<Object> queue = new LinkedBlockingQueue<>(10);

public static void main(String[] args) throws Exception {

CompletableFuture.runAsync(Demo::runProducer);
CompletableFuture.runAsync(Demo::runConsumer);

BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
do {
String input = br.readLine();
if ("q".equals(input)) {
System.out.println("Exit!");
System.exit(0);
} else {
String timestamp = new SimpleDateFormat("HH:mm:ss").format(new Date());
System.out.println(String.format("[%s] Queue size: %d", timestamp, queue.size()));
System.out.println(String.format("[%s] Local caches produced: %d", timestamp, producedCount.longValue()));
System.out.println(String.format("[%s] Local caches consumed %d", timestamp, consumedCount.longValue()));
System.out.println(String.format("[%s] Global cache size: %d", timestamp, globalCache.size(CachePeekMode.ALL)));
}
} while (true);
}

private static void runProducer() {
do {
long cacheId = producedCount.incrementAndGet();
try {
BinaryObject bObj = createLocalObject(1);
if (useLocalCacheToCauseMemLeak) {
IgniteCache<String, BinaryObject> cache = createLocalCache(cacheId);
cache.put(bObj.field(ID), bObj);
queue.put(cache);
} else {
Map<String, BinaryObject> cache = new HashMap<>();
cache.put(bObj.field(ID), bObj);
queue.put(cache);
}
} catch (Exception e) {
System.out.println(e.toString());
System.exit(1);
}
} while (true);
}


private static List<?> getSQLSelectRow(BinaryObject localObject) {
List<?> res = new ArrayList<>();
for (String f : globalFields) {
res.add(localObject.field(f));
}
return res;
}


@SuppressWarnings("unchecked")
private static void runConsumer() {
do {
try {
final Object taken = queue.take();
final List<List<?>> globalSQLRows = new ArrayList<>(); // Note: In practice rows are generated by executing SqlFieldsQuery against local cache; here omitted for sake of simplicity.
if (taken instanceof IgniteCache) {
final IgniteCache<String, BinaryObject> cache = IgniteCache.class.cast(taken);
cache.forEach(e -> {
globalSQLRows.add(getSQLSelectRow(e.getValue()));
});
cache.destroy();
} else if (taken instanceof Map) {
final Map<String, BinaryObject> cache = Map.class.cast(taken);
cache.forEach((k, v) -> {
globalSQLRows.add(getSQLSelectRow(v));
});
}

consumedCount.incrementAndGet();

final IgniteDataStreamer<String, BinaryObject> globalStreamer = createGlobalStreamer(globalCache);
for (List<?> row : globalSQLRows) {
final BinaryObjectBuilder globalBuilder = globalPrototype.toBuilder();
for (int i = 0; i < globalFields.size(); i++) {
globalBuilder.setField(globalFields.get(i), row.get(i));
}
final BinaryObject gObj = globalBuilder.build();
globalStreamer.addData(gObj.field(ID), gObj);
}
globalStreamer.flush();
globalStreamer.close();

} catch (Exception e) {
System.out.println(e.toString());
System.exit(1);
}
} while (true);
}

private static Ignite createIgnite() {
IgniteConfiguration iCfg = new IgniteConfiguration();
TcpCommunicationSpi tcpCommunication = new TcpCommunicationSpi();
tcpCommunication.setMessageQueueLimit(1024); // to get rid of the warning
iCfg.setCommunicationSpi(tcpCommunication);
String workDirectory = System.getProperty("user.home") + File.separator + "ignite";
iCfg.setWorkDirectory(workDirectory);
System.out.println();
System.out.println(String.format(">>> Starting Ignite on %s; work directory %s ...", "MyLeakingNode", workDirectory));
System.out.println();
final Ignite ignite = Ignition.start(iCfg);
ClusterNode localNode = ignite.cluster().localNode();
System.out.println();
System.out.println(String.format(">>> Ignite started on %s (%s) successfully!", "MyLeakingNode", localNode.id()));
System.out.println();
return ignite;
}

private static IgniteCache<String, BinaryObject> createLocalCache(long id) {
final String cacheName = "LocalCache" + id;
final CacheConfiguration<String, BinaryObject> cCfg = new CacheConfiguration<>();
cCfg.setName(cacheName);
cCfg.setStoreKeepBinary(true);
cCfg.setCacheMode(CacheMode.LOCAL);
cCfg.setOnheapCacheEnabled(false);
cCfg.setCopyOnRead(false);
cCfg.setBackups(0);
cCfg.setWriteBehindEnabled(false);
cCfg.setReadThrough(false);
cCfg.setReadFromBackup(false);

final QueryEntity queryEntity = new QueryEntity(String.class.getTypeName(), "LocalEntity");
for (String field : localFields()) {
queryEntity.addQueryField(field, String.class.getTypeName(), field);
}
cCfg.setQueryEntities(Collections.singletonList( queryEntity));

ignite.destroyCache(cacheName); // local cache is not really local - reference can be kept by other nodes if restart during the load happens
return ignite.createCache(cCfg).withKeepBinary();
}


private static List<String> localFields() {
return Stream.concat(Stream.of(ID), IntStream.rangeClosed(1, numOfLocalFields).boxed().map(e -> String.format("field%s", e))).collect(Collectors.toList());
}

private static BinaryObject createLocalPrototype() {
BinaryObjectBuilder builder = ignite.binary().builder("LocalEntity");
for (String field : localFields()) {
builder.setField(field, null, String.class);
}
return builder.build();
}

private static BinaryObject createLocalObject(long id) {
BinaryObjectBuilder res = localPrototype.toBuilder();
for (String field : localFields()) {
if (ID.equals(field)) {
res.setField(field, String.valueOf(id), String.class);
} else {
res.setField(field, id + "@" + field, String.class);
}
}
return res.build();
}

private static IgniteCache<String, BinaryObject> getOrCreateGlobalCache() {
final String cacheName = "GlobalCache";
final CacheConfiguration<String, BinaryObject> cCfg = new CacheConfiguration<>();
cCfg.setName(cacheName);
cCfg.setStoreKeepBinary(true);
cCfg.setCacheMode(CacheMode.PARTITIONED);
cCfg.setOnheapCacheEnabled(false);
cCfg.setCopyOnRead(false);
cCfg.setBackups(0);
cCfg.setWriteBehindEnabled(false);
cCfg.setReadThrough(false);
cCfg.setQueryEntities(Collections.singletonList(new QueryEntity(String.class.getTypeName(), cacheName))); // must be defined otherwise indexing Spi remove method is not called
return ignite.getOrCreateCache(cCfg).withKeepBinary();
}

private static IgniteDataStreamer<String, BinaryObject> createGlobalStreamer(IgniteCache<String, BinaryObject> cache) {
IgniteDataStreamer<String, BinaryObject> streamer = ignite.dataStreamer(cache.getName());
streamer.allowOverwrite(true);
streamer.skipStore(true);
streamer.keepBinary(true);
return streamer;
}

private static List<String> globalFields() {
return Stream.concat(Stream.of(ID), IntStream.rangeClosed(1, numOfGlobalFields).boxed().map(e -> String.format("field%s", e))).collect(Collectors.toList());
}

private static BinaryObject createGlobalPrototype() {
BinaryObjectBuilder builder = ignite.binary().builder("GlobalEntity");
for (String field : globalFields) {
builder.setField(field, null, String.class);
}
return builder.build();
}


}


{code}

> Huge memory leak when data streamer used together with local cache
> ------------------------------------------------------------------
>
>                 Key: IGNITE-7918
>                 URL: https://issues.apache.org/jira/browse/IGNITE-7918
>             Project: Ignite
>          Issue Type: Bug
>          Components: cache
>    Affects Versions: 2.3
>            Reporter: Zbyszek B
>            Priority: Blocker
>
> Dear Igniters,
> We observe huge memory leak when data streamer used together with local cache.
> In the attached demo producer produces local cache with single binary object and passes this to queue. Consumer picks up the cache from the queue, constructs different binary object from it, adds it to global partitioned cache and destroys local cache.
> This design causes a significant leak - the whole heap it takes within minutes (no matter if this is 4G or 24G).
>  
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)