I am creating an interface for the simultaneous execution of methods, abstracting the details of synchronization (to swap for a distributed implementation, when necessary). I created a single jvm implementation that allows you to use strings as a mutex, storing them on a map to ensure that one link is used, even if lines with different links are passed. concurrency seems to be working fine, however, I was surprised to see that the test shows that the link count never decreases. I suggested that using WeakValues () would be sufficient to prevent memory leaks, but it doesn't seem to be that way. Can someone point out what might cause this leak?
public class SynchronousMethodExecutorSynchronizedImpl implements ISynchronousMethodExecutor {
final Map<String, String> mutexMap = new MapMaker()
.weakValues()
.makeComputingMap(
new Function<String, String>() {
@Override
public String apply(String id) {
return id;
}
});
@Override
public Object doSynchronousMethod(String domain, String id, ISynchronousMethod synchronousMethod) {
synchronized(mutexMap.get(domain + "." + id))
{
return synchronousMethod.execute();
}
}
}
, :
public class SynchronousMethodExecutorSynchronizedImplTest extends TestCase {
int counter;
SynchronousMethodExecutorSynchronizedImpl methodExecutor;
@Override
public void before() throws Exception {
super.before();
methodExecutor = new SynchronousMethodExecutorSynchronizedImpl();
}
@Test
public void concurrentExecute() throws InterruptedException {
assertEquals(0, counter);
for(int i=0; i<1000; i++)
getConcurrentExecutorThread().start();
Thread.sleep(1000);
assertEquals(1, methodExecutor.mutexMap.size());
try
{
final List<long[]> infiniteList = new LinkedList<long[]>();
for(long i = Long.MIN_VALUE; i < Long.MAX_VALUE; i++)
infiniteList.add(new long[102400]);
fail("An OutOfMemoryError should be thrown");
}
catch(OutOfMemoryError e)
{
}
assertEquals(2000, counter);
assertEquals(0, methodExecutor.mutexMap.size());
}
private ISynchronousMethod method = new ISynchronousMethod() {
@Override
public Object execute() {
counter++;
return null;
}
};
private Thread getConcurrentExecutorThread() {
return new Thread() {
@Override
public void run() {
methodExecutor.doSynchronousMethod("TEST", "1", method);
try
{
Thread.sleep(500);
}
catch (InterruptedException e)
{
}
methodExecutor.doSynchronousMethod("TEST", new String("1"), method);
}
};
}
}
- , :
assertEquals (0, methodExecutor.mutexMap.size());