8

I have the following class. I use ConcurrentHashMap. I have many threads writing to the maps and a Timer that saves the data in the map every 5 minutes. I manage to achieve thread safety by using putIfAbsent() when I write entries in the map. However, when I read from it and then remove all entries by clear() method, I want no other thread writes to map while I’m in the process of reading the map contents and then removing them. Obviously my code is not threadsafe even with synchronized(lock){}, b/c the thread that owns the lock in saveEntries(), is not necessarily the same thread that writes into my maps in log() method! Unless I lock the whole code in log() with the same lock object!

I was wondering is there any other way to achieve thread safety w/o enforcing synchronizing by an external lock? Any help is greatly appreciated.

public class Logging {

private static Logging instance;    
private static final String vendor1 = "vendor1";
private static final String vendor2 = "vendor2";    
private static long delay = 5 * 60 * 1000;

private ConcurrentMap<String, Event> vendor1Calls = new ConcurrentHashMap<String, Event>();
private ConcurrentMap<String, Event> vendor2Calls = new ConcurrentHashMap<String, Event>();

private Timer timer;    
private final Object lock = new Object();

private Logging(){
    timer = new Timer();                
    timer.schedule(new TimerTask() {
        public void run() {
            try {
                saveEntries();
            } catch (Throwable t) {
                timer.cancel();
                timer.purge();
            }
        }       
    }, 0, delay);
}

public static synchronized Logging getInstance(){     
    if (instance == null){
        instance = new Logging();
    }
    return instance;
 }

public void log(){      
    ConcurrentMap<String, Event> map;
    String key = "";        

    if (vendor1.equalsIgnoreCase(engine)){
        map = vendor1Calls;
    }else if(vendor2.equalsIgnoreCase(engine)){  
        map = vendor2Calls;
    }else{
        return;
    }       


    key = service + "." + method;
// It would be the code if I use a regular HashMap instead of ConcurrentHashMap
    /*Event event = map.get(key);       

    // Map does not contain this service.method, create an Event for the first     time.
    if(event == null){
        event = new Event();            
        map.put(key, event);

        // Map already contains this key, just adjust the numbers.
    }else{
        // Modify the object fields
    }*/
    //}

    // Make it thread-safe using CHM
    Event newEvent = new Event();
    Event existingEvent= map.putIfAbsent(key, newEvent); 

    if(existingEvent!=null && existingEvent!=newEvent){
        // Modify the object fields
}       

private void saveEntries(){

    Map<String, List<Event>> engineCalls = null;
    try {           

        engineCalls = new HashMap<String, List<Event>>();
        List<Event> events = null;

// How can I achieve therad safety here w/o applying any lock?
        //synchronized(lock){
            if(!vendor1Calls.isEmpty()){
                events = new ArrayList<Event>();
                events.addAll(vendor1Calls.values());
                engineCalls.put(vendor1, events);
                vendor1Calls.clear();
            }
            if(!vendor2Calls.isEmpty()){
                events = new ArrayList<Event>();
                events.addAll(vendor2Calls.values());
                engineCalls.put(vendor2, events);
                vendor2Calls.clear();
            }
        //}

// logICalls() saves the events in the DB.          
        DBHandle.logCalls(engineCalls);
    } catch (Throwable t) {         
    } finally {
        if(engineCalls!=null){
            engineCalls.clear();
        }                       
    }   
}       

}

4
  • 2
    You accept the answer by clicking the grey checkmark to the left of an answer. Don't accept the answer if it doesn't solve your problem, though, peer pressure notwithstanding. Aug 24, 2012 at 19:56
  • Yes, I suggest you go to stackoverflow.com/users/1052348/bluesky?tab=questions, click every link, and click the checkmark next to the answer that best solved your problem. PS you can accept an answer you made yourself to your own question.
    – Kevin Jin
    Aug 24, 2012 at 20:06
  • crossposted here, and you can make it atomic using CHM features alone.
    – jtahlborn
    Aug 25, 2012 at 12:30
  • I know this is an old question, but I don't understand why you are using a map for something that seems inherently like a queue in the first place. You could add log entries to a concurrent queue, and then in save entries look at the size of the queue and remove that many from the front of the queue and write them. Because you just look at the size once, it doesn't matter if more are added while you're saving them; they will be left for next time. Sorry if I've misunderstood your purpose. Sep 26, 2013 at 22:22

5 Answers 5

3

However, when I read from it and then remove all entries by clear() method, I want no other thread writes to map while I’m in the process of reading the map contents and then removing them.

I think what you're trying to say is that you don't really care about strictly locking the maps. Instead, you only really care about the loss of any log entries between the vender1Calls.values() and vendor1Calls.clear(), correct?

In that is the case, I can imagine that you can replace

events.addAll(vendor1Calls.values());
vendor1Calls.clear();

with this in saveEntries:

for (Iterator<Event> iter = vendor1Calls.values().iterator(); iter.hasNext(); ) {
    Event e = iter.next();
    events.add(e);
    iter.remove();
}

That way, you only remove the Events that you added to the events List already. You can still write to the vendor1Calls maps while saveEntries() is still executing, but the iterator skips the values added.

18
  • It may actually iterate over values added in the meantime, but there are no guarantees. Aug 24, 2012 at 20:11
  • Thanks for the reply. How using Iterator makes it thread safe? How can I make sure that not 2 threads at the same time, iterating thru the values() of maps and while one performing: Event e = iter.next(); the other thread is not performing: iter.remove(); ? Also my goal was to lock the entire map that no other thread writes to it while I'm in the process of reading from map, b/c what would happen to the entries added to the map when I'm in the process of iterating thru them?
    – blueSky
    Aug 24, 2012 at 20:13
  • I assumed from the fact that since you used a timer to call saveEntries(), that saveEntries() is private, and that you have no other code, you will not have more than one thread iterating over either vendor1Calls or vendor2Calls and possibly call remove on them.
    – Kevin Jin
    Aug 24, 2012 at 20:45
  • The ones that are written while being iterated may or may not be saved ("Iterators and Enumerations return elements reflecting the state of the hash table at some point at or since the creation of the iterator/enumeration."), but there's one thing for certain: absolutely no entry will be lost, because those that are removed by calling iter.remove() in saveEntries() have already been added to the events ArrayList.
    – Kevin Jin
    Aug 24, 2012 at 20:50
  • keep in mind that iter.remove() removes the element that was returned by iter.next(). don't expect that something will be added in the spot occupied by your current position in the iterator and that iter.remove() will remove that by mistake.
    – Kevin Jin
    Aug 24, 2012 at 20:52
3

Without any external synchronization you cannot achieve this with a CHM. The Iterator views returned are weakly consistent which means the contents of the Map can change while you are actually iterating over it.

It appears you would need to use a Collections.synchronizedMap to get the functionality you are looking for.

Edit to make my point more clear:

To achieve this with a synchronizedMap You would first have to synchronize on the Map and then you can iterate or copy the contents into another map an then clear.

Map map = Collections.synchronizedMap(new HashMap());

public void work(){
  Map local = new HashMap();
  synchronized(map){
     local.putAll(map);
     map.clear();
  }
  //do work on local instance 
}

Instead of the local instance, as I mentioned, you can iterate + remove similar to @Kevin Jin's answer.

5
  • syncronizedMap's iterators are useless for the job as they fall apart in the face of concurrent changes to the map. On the other hand, there's nothing wrong with the CHM iterator catching any later entries, it might even be better since more entries will be saved and removed from the map. Aug 24, 2012 at 21:36
  • @Marko Topolnik I was speaking on terms of atomicity. I should have made it more obvious that he would have to synchronize on the Map itself. This obviously has the side effect of performance degradation but if he needs to have no threads updating while another thread iterates then clears there would be no other way to do it. That however may not be the case from his needs.
    – John Vint
    Aug 24, 2012 at 21:40
  • @JohnVint: since Iterator does not give me a snapshot, then is this possible that I end up with a loop that never ends! B/c as threads actually might add entries to the map much faster than I remove them! So is there anyway that I obtain a snapshot or make it atomic somehow w/o using locks?
    – blueSky
    Aug 24, 2012 at 22:03
  • In theory it can but in practice its unlikely. You can throttle byt getting the size before iterating and if a count exceeds you can break early. That being said an infinite loop isnt something i would worry about
    – John Vint
    Aug 24, 2012 at 22:16
  • Thanks everybody for the time and responses. Very helpful. I'll think about it... looks like making it atomic is not an option here. I will choose bet. synchronizedMap or just iterating thru. Thanks again.
    – blueSky
    Aug 24, 2012 at 22:32
0

An atomic version of this example is shown in this thread (using only the features in ConcurrentMap).

0
0

My best suggestion would be to use a ReadWriteLock but as you specifically state that you don't want to use any locks (btw ConcurrentHashMap will probably use them internally) you could try the following.

Use an AtomicReference to each of your maps and when the time comes to log their contents use getAndSet to replace the old map with a brand new empty one.

You now have a Map with exclusive use that you can iterate across and clear as much as you like. Unfortunately there is one small problem (that using a lock will get rid of) and that is if another thread is in the process of adding to the map at the time you swap it out with an empty one. You could perhaps add a delay at this point in the hope of waiting long enough for the other thread to finish what it was doing. Perhaps there is some internal functionality of a ConcurrentHashMap you can use to wait until everyone has finished with it.

0

The following code uses a persistent map from the functional java project. It uses more memory, but is (AFAIK :) safe to use by multiple threads. The only mutable value in the AtomicReference and it is updated with a compare-and-set. The map and event are immutable, and hence thread-safe. Also, instead of clearing the map, I replace the reference to it.

import fj.F;
import fj.Ord;
import fj.data.TreeMap;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

public class Logging
{
    // Event is immutable
    private static class Event
    {
        // updates are done by creating new values
        Event update(String key)
        {
            return new Event();
        }
    }

    // Refactored code pertaining to one vendor into a separate class.
    private static class EngineLogger
    {
        public final String vendor;
        private final AtomicReference<TreeMap<String, Event>> mapRef =
                new AtomicReference<TreeMap<String, Event>>(TreeMap.<String, Event>empty(Ord.stringOrd));

        private EngineLogger(String vendor)
        {
            this.vendor = vendor;
        }

        public void log(String service, String method)
        {
            final String key = service + "." + method;
            boolean updated = false;
            while (! updated)
            {
                // get the current value of the map
                TreeMap<String, Event> currMap = mapRef.get();

                // create an updated value of the map, which is the current map plus info about the new key
                TreeMap<String, Event> newMap = currMap.update(key, new F<Event, Event>()
                {
                    @Override
                    public Event f(Event event)
                    {
                        // Modify the object fields of event, if the map contains the key
                        return event.update(key);
                    }
                    // create a new event if the map does not contain the key
                }, new Event());

                // compare-and-set the new value in .. repeat until this succeeds
                updated = mapRef.compareAndSet(currMap, newMap);
            }
        }

        public List<Event> reset()
        {
            /* replace the reference with a new map */
            TreeMap<String, Event> oldMap = mapRef.getAndSet(TreeMap.<String, Event>empty(Ord.stringOrd));

            /* use the old map to generate the list */
            return new ArrayList<Event>(oldMap.toMutableMap().values());
        }
    }

    private static Logging instance;
    private static long delay = 5 * 60 * 1000;
    private final Timer timer;

    private final EngineLogger vendor1 = new EngineLogger("vendor1");
    private final EngineLogger vendor2 = new EngineLogger("vendor2");

    private Logging()
    {
        timer = new Timer();
        timer.schedule(new TimerTask()
        {
            public void run()
            {
                try
                {
                    saveEntries();
                }
                catch (Throwable t)
                {
                    timer.cancel();
                    timer.purge();
                }
            }
        }, 0, delay);
    }

    public static synchronized Logging getInstance()
    {
        if (instance == null)
        {
            instance = new Logging();
        }
        return instance;
    }

    public void log(String engine, String service, String method)
    {
        if (vendor1.vendor.equals(engine))
        {
            vendor1.log(service, method);
        }
        else if (vendor2.vendor.equals(engine))
        {
            vendor2.log(service, method);
        }
    }

    private void saveEntries()
    {
        Map<String, List<Event>> engineCalls = new HashMap<String, List<Event>>();
        engineCalls.put(vendor1.vendor, vendor1.reset());
        engineCalls.put(vendor2.vendor, vendor2.reset());
        DBHandle.logCalls(engineCalls);
    }
}
3
  • Although may not be suitable for my purpose b/c of memory usage, but very helpful. Thanks so much for sharing.
    – blueSky
    Aug 28, 2012 at 16:53
  • @blueSky Thinking more, I realize that this solution scales poorly. The CAS in log method will fail even when different threads modify Events for different keys. Aug 28, 2012 at 21:12
  • Thanks for the update. I was not able to use it b/c of memory issue. Its always great to know what's out there :)
    – blueSky
    Aug 28, 2012 at 21:46

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.