Reputation: 109

Thread crashing

I have a simple project I'm working to better my understanding of locking and threading in Java. Essentially I have a cache object that starts a cleanup thread to remove items past a given age. The test class "Tester" runs two additional threads, one to add items to cache and one to printout the contents of cache. For some reason when the cleanup thread modifies the HashMap embedded in Cache it ceases any further iterations. I've tried synchronizing the accessor/mutator methods as well as synchronizing around a LOCK object in Cache. Any ideas or help would be muy beueno. ;)

public class Cache 
{
    private HashMap<Object, ObjectWrapper> cachedObjects = new HashMap<>();
    private static Cache cache = null;
    private static int TIME_TO_KEEP = 60000; // 60 seconds
    private final static Object LOCK = new Object();

    public static Cache getInstance()
    {
        if (cache == null)
        {
            cache = new Cache();
        }
        return cache;
    }

    private Cache()
    {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

        Runnable task = () -> {
            synchronized(LOCK)
            {
                System.out.println("Cleanup");
                Set<Object> keys = cachedObjects.keySet();
                final long now = System.currentTimeMillis();
                keys.forEach(k -> {
                    try
                    {

                        {
                            ObjectWrapper ow = cachedObjects.get(k);
                            if(ow.getExpireTime() < now)
                            {
                                int size = cachedObjects.size();
                                cachedObjects.remove(k, ow);
                                System.out.println("DEL:" + k + ", from " + size + " to " + cachedObjects.size());
                            }
                        }
                    }
                    catch (Throwable t)
                    {
                        t.printStackTrace();
                    }
                });
            }
        };

        executor.scheduleWithFixedDelay(task, 5, 15, TimeUnit.SECONDS);
    }

    public void addObject(Object key, Object obj)
    {
        synchronized(LOCK)
        {
            ObjectWrapper ow = new ObjectWrapper(obj, System.currentTimeMillis() + TIME_TO_KEEP);
            cachedObjects.put(key, ow);
        }
    }

    public ObjectWrapper getObj(Object key)
    {
        synchronized(LOCK)
        {
            return cachedObjects.get(key);
        }
    }

    public Collection<ObjectWrapper> getValues()
    {
        synchronized(LOCK)
        {
            return cachedObjects.values();
        }
    }

    public Set<Object> getKeys()
    {
        synchronized(LOCK)
        {
            return cachedObjects.keySet();
        }
    }

    public static void main(String[] args)
    {
        Cache cache = Cache.getInstance();
    }
}

import java.util.*;
import java.util.concurrent.*;

public class Tester 
{
    private static Integer id = 0;
    private static Cache cache = Cache.getInstance();

    public static void main(String[] args)
    {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

        Runnable adder = () -> {
            System.out.println("Adding id:" + ++id);
            Object o = new Object();
            cache.addObject(id, o);
        };

        executor.scheduleWithFixedDelay(adder, 5, 10, TimeUnit.SECONDS);


        Runnable tester = () -> {
            long currTime = System.currentTimeMillis();
            System.out.println("Test: ");
            Set<Object> keys = cache.getKeys();
            keys.forEach(k -> {
                ObjectWrapper o = cache.getObj(k);
                System.out.println(k + ">" + currTime + "::" + o.getExpireTime());
            });
        };

        executor.scheduleWithFixedDelay(tester, 20, 10, TimeUnit.SECONDS);
    }
}

public class ObjectWrapper
{
    private Object obj;
    private long expireTime;

    public ObjectWrapper(Object obj, long expireTime)
    {
        this.obj = obj;
        this.expireTime = expireTime;
    }

    public Object getObj()
    {
        return obj;
    }

    public void setObj(Object obj)
    {
        this.obj = obj;
    }

    public long getExpireTime()
    {
        return expireTime;
    }

    public void setExpireTime(long expireTime)
    {
        this.expireTime = expireTime;
    }
}

Upvotes: 2

Answers (2)

Dylon

Reputation: 1750

You are deleting elements from the keySet while iterating over it, resulting a ConcurrentModificationException that is thrown the at the next iteration (outside the try-catch block). If you move the try-catch block outside the loop:

private Cache()
{
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

    Runnable task = () -> {
        synchronized(LOCK)
        {
            try
            {
                System.out.println("Cleanup");
                Set<Object> keys = cachedObjects.keySet();
                final long now = System.currentTimeMillis();
                keys.forEach(k -> {

                        {
                            ObjectWrapper ow = cachedObjects.get(k);
                            if(ow.getExpireTime() < now)
                            {
                                int size = cachedObjects.size();
                                cachedObjects.remove(k, ow);
                                System.out.println("DEL:" + k + ", from " + size + " to " + cachedObjects.size());
                            }
                        }
                });
            }
            catch (Throwable t)
            {
                t.printStackTrace();
            }
        }
    };

    executor.scheduleWithFixedDelay(task, 5, 15, TimeUnit.SECONDS);
}

You'll get the stacktrace:

java.util.ConcurrentModificationException
        at java.util.HashMap$KeySet.forEach(HashMap.java:935)
        at Cache.lambda$new$1(Cache.java:32)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

Instead, use an iterator over the keySet:

private Cache()
{
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

    Runnable task = () -> {
        synchronized(LOCK)
        {
            try
            {
                System.out.println("Cleanup");
                final long now = System.currentTimeMillis();
                final Iterator<Map.Entry<Object, ObjectWrapper>> iter = cachedObjects.entrySet().iterator();
                while (iter.hasNext()) {
                    final Map.Entry<Object, ObjectWrapper> entry = iter.next();
                    final ObjectWrapper ow = entry.getValue();
                    if(ow.getExpireTime() < now)
                    {
                        final Object k = entry.getKey();
                        int size = cachedObjects.size();
                        iter.remove();
                        System.out.println("DEL:" + k + ", from " + size + " to " + cachedObjects.size());
                    }
                }
            }
            catch (Throwable t)
            {
                t.printStackTrace();
            }
        }
    };

    executor.scheduleWithFixedDelay(task, 5, 15, TimeUnit.SECONDS);
}

Doing so yields the expected output:

Cleanup
Adding id:1
Adding id:2
Cleanup
Test:
1>1467228864763::1467228909763
2>1467228864763::1467228919764
Adding id:3
Test:
1>1467228874766::1467228909763
2>1467228874766::1467228919764
3>1467228874766::1467228929764
Cleanup
Adding id:4
Test:
1>1467228884766::1467228909763
2>1467228884766::1467228919764
3>1467228884766::1467228929764
4>1467228884766::1467228939764
Adding id:5
Cleanup
Test:
1>1467228894770::1467228909763
2>1467228894770::1467228919764
3>1467228894770::1467228929764
4>1467228894770::1467228939764
5>1467228894770::1467228949765
Adding id:6
Test:
1>1467228904771::1467228909763
2>1467228904771::1467228919764
3>1467228904771::1467228929764
4>1467228904771::1467228939764
5>1467228904771::1467228949765
6>1467228904771::1467228959765
Cleanup
Adding id:7
Test:
1>1467228914771::1467228909763
2>1467228914771::1467228919764
3>1467228914771::1467228929764
4>1467228914771::1467228939764
5>1467228914771::1467228949765
6>1467228914771::1467228959765
7>1467228914771::1467228969765
Adding id:8
Cleanup
DEL:1, from 8 to 7
DEL:2, from 7 to 6
Test:
3>1467228924772::1467228929764
4>1467228924772::1467228939764
5>1467228924772::1467228949765
6>1467228924772::1467228959765
7>1467228924772::1467228969765
8>1467228924772::1467228979765
Adding id:9
Test:
3>1467228934772::1467228929764
4>1467228934772::1467228939764
5>1467228934772::1467228949765
6>1467228934772::1467228959765
7>1467228934772::1467228969765
8>1467228934772::1467228979765
9>1467228934772::1467228989766
Cleanup
DEL:3, from 7 to 6
Adding id:10
Test:
4>1467228944773::1467228939764
5>1467228944773::1467228949765
6>1467228944773::1467228959765
7>1467228944773::1467228969765
8>1467228944773::1467228979765
9>1467228944773::1467228989766
10>1467228944773::1467228999766
Adding id:11
Cleanup
DEL:4, from 8 to 7
DEL:5, from 7 to 6
Test:
6>1467228954773::1467228959765
7>1467228954773::1467228969765
8>1467228954773::1467228979765
9>1467228954773::1467228989766
10>1467228954773::1467228999766
11>1467228954773::1467229009766
Adding id:12
Test:
6>1467228964774::1467228959765
7>1467228964774::1467228969765
8>1467228964774::1467228979765
9>1467228964774::1467228989766
10>1467228964774::1467228999766
11>1467228964774::1467229009766
12>1467228964774::1467229019767
Cleanup
DEL:6, from 7 to 6
Adding id:13

Upvotes: 1

Nicolas Filotto

Reputation: 44965

Consider using a ConcurrentHashMap which is a map implementation natively thread safe which is not the case of an HashMap.

Your mistake is mainly here:

public Collection<ObjectWrapper> getValues()
{
    synchronized(LOCK)
    {
        return cachedObjects.values();
    }
}

public Set<Object> getKeys()
{
    synchronized(LOCK)
    {
        return cachedObjects.keySet();
    }
}

Doing this is not enough as you share the keys and the values of your HashMap that are actually provided in non thread safe collections, such that you need to synchronize access when you iterate over their content or you could simply return a safe copy as next:

public Collection<ObjectWrapper> getValues()
{
    synchronized(LOCK)
    {
        return new ArrayList<>(cachedObjects.values());
    }
}

public Set<Object> getKeys()
{
    synchronized(LOCK)
    {
        return new HashSet<>(cachedObjects.keySet());
    }
}

You need also to make ObjectWrapper thread safe too as it is meant to be shared otherwise your code won't be still thread safe. The simplest approach for this is to make it immutable like this:

public class ObjectWrapper
{
    private final Object obj;
    private final long expireTime;

    public ObjectWrapper(Object obj, long expireTime)
    {
        this.obj = obj;
        this.expireTime = expireTime;
    }

    public Object getObj()
    {
        return obj;
    }

    public long getExpireTime()
    {
        return expireTime;
    }
}

Upvotes: 2

Thread crashing

Answers (2)

Related Questions