Adding in concurrent logic to help prevent ConcurrentModificationException

transitclock/src/main/java/org/transitclock/avl/AvlClient.java

@@ -17,6 +17,7 @@
 package org.transitclock.avl;
 
 import java.util.HashMap;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -79,6 +80,7 @@ public AvlReport getAvlReport() {
 	public void run() {
 		// Put a try/catch around everything so that if unexpected exception 
 		// occurs an e-mail is sent and the avl client thread isn't killed.
+		ReentrantLock l = new ReentrantLock();
 		try {
 			// If the data is bad throw it out
 			String errorMsg = avlReport.validateData();
@@ -89,7 +91,7 @@ public void run() {
 			}
 
 			// See if should filter out report
-			synchronized (avlReports) {
+			l.lock();
 				AvlReport previousReportForVehicle =
 						avlReports.get(avlReport.getVehicleId());
 
@@ -165,7 +167,7 @@ public void run() {
 				// filter
 				// the next one
 				avlReports.put(avlReport.getVehicleId(), avlReport);
-			}
+
 
 			// Process the report
 			logger.info("Thread={} AvlClient processing AVL data {}", 
@@ -180,6 +182,9 @@ public void run() {
 //					"For agencyId={} Exception {} for avlReport={}.",
 //					AgencyConfig.getAgencyId(), e.getMessage(), avlReport, e);
 		}
+		finally{
+			l.unlock();
+		}
 	}
 
 }

transitclock/src/main/java/org/transitclock/core/AvlProcessor.java

@@ -42,6 +42,7 @@
 import org.transitclock.utils.Time;
 
 import java.util.*;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * This is a very important high-level class. It takes the AVL data and
@@ -1583,19 +1584,19 @@ public void cacheAvlReportWithoutProcessing(AvlReport avlReport) {
 		VehicleState vehicleState =
 				VehicleStateManager.getInstance().getVehicleState(
 						avlReport.getVehicleId());
-
+		ReentrantLock l = new ReentrantLock();
+		l.lock();
 		// Since modifying the VehicleState should synchronize in case another
 		// thread simultaneously processes data for the same vehicle. This
 		// would be extremely rare but need to be safe.
-		synchronized (vehicleState) {
-			// Update AVL report for cached VehicleState
-			vehicleState.setAvlReport(avlReport);
+		// Update AVL report for cached VehicleState
+		vehicleState.setAvlReport(avlReport);
 
-			// Let vehicle data cache know that the vehicle state was updated
-			// so that new IPC vehicle data will be created and cached and
-			// made available to the API.
-			VehicleDataCache.getInstance().updateVehicle(vehicleState);
-		}
+		// Let vehicle data cache know that the vehicle state was updated
+		// so that new IPC vehicle data will be created and cached and
+		// made available to the API.
+		VehicleDataCache.getInstance().updateVehicle(vehicleState);
+		l.unlock();
 	}
 
 	private boolean isCanceled(VehicleState vehicleState) {
@@ -1637,7 +1638,7 @@ private void logInvalidAssignment(VehicleState vehicleState) {
 	 *            The new AVL report to be processed
 	 */
 	public void processAvlReport(AvlReport avlReport) {
-		IntervalTimer timer = new IntervalTimer(); 
+		IntervalTimer timer = new IntervalTimer();
 
 		// Handle special case where want to not use assignment from AVL
 		// report, most likely because want to test automatic assignment

transitclock/src/main/java/org/transitclock/core/Indices.java

@@ -18,6 +18,7 @@
 
 import java.io.Serializable;
 import java.util.Objects;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.transitclock.applications.Core;
 import org.transitclock.db.structs.ArrivalDeparture;
@@ -196,6 +197,8 @@ public boolean isEarlierStopPathThan(Indices indices) {
 	 * @return The resulting Indices object
 	 */
 	public Indices increment(long epochTime) {
+		ReentrantLock l = new ReentrantLock();
+		l.lock();
 		++segmentIndex;
 		if (segmentIndex >= block.numSegments(tripIndex, stopPathIndex)) {
 			segmentIndex = 0;
@@ -217,6 +220,7 @@ public Indices increment(long epochTime) {
 				}
 			}
 		}
+		l.unlock();
 
 		return this;
 	}

transitclock/src/main/java/org/transitclock/utils/threading/NamedThread.java

@@ -19,6 +19,7 @@
 import java.util.Date;
 import java.util.HashMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -79,7 +80,8 @@ public void uncaughtException(Thread t, Throwable e) {
 	 * @return
 	 */
 	private static String threadNameWithCounter(String name) {
-		synchronized (threadNameCountMap) {
+		ReentrantLock l = new ReentrantLock();
+		l.lock();
 			Integer count = threadNameCountMap.get(name);
 			if (count == null) {
 				count = 1;				
@@ -88,16 +90,19 @@ private static String threadNameWithCounter(String name) {
 				count++;			
 			}		
 			threadNameCountMap.put(name, count);
+			l.unlock();
 			return name + "-" + count;
-		}
 	}
 
 	@Override
 	public void run() {
 		logger.debug("Created NamedThread {}", getName());
+		ReentrantLock l = new ReentrantLock();
 		try {
+			l.lock();
 			numAlive.incrementAndGet();
 			super.run();
+
 		} catch (Throwable t) {
 			// Log the problem but do so within a try/catch in case it is
 			// an OutOfMemoryError and need to exit even if get another
@@ -141,6 +146,7 @@ public void run() {
 				System.exit(-1);
 			}
 		} finally {
+			l.unlock();
 			numAlive.decrementAndGet();
 			logger.debug("Exiting NamedThread {}", getName());
 		}