Merge pull request #157 from Athou/parallel

Parallel updates

src/main/java/com/commafeed/backend/LockMap.java

@@ -0,0 +1,69 @@
+package com.commafeed.backend;
+
+import java.lang.ref.WeakReference;
+import java.util.WeakHashMap;
+
+// A map that creates and stores lock objects for arbitrary keys values.
+// Lock objects which are no longer referenced are automatically released during garbage collection.
+// Author: Christian d'Heureuse, www.source-code.biz
+// Based on IdMutexProvider by McDowell, http://illegalargumentexception.blogspot.ch/2008/04/java-synchronizing-on-transient-id.html
+// See also http://stackoverflow.com/questions/5639870/simple-java-name-based-locks
+public class LockMap<KEY> {
+
+	private WeakHashMap<KeyWrapper<KEY>, WeakReference<KeyWrapper<KEY>>> map;
+
+	public LockMap() {
+		map = new WeakHashMap<KeyWrapper<KEY>, WeakReference<KeyWrapper<KEY>>>();
+	}
+
+	// Returns a lock object for the specified key.
+	public synchronized Object get(KEY key) {
+		if (key == null) {
+			throw new NullPointerException();
+		}
+		KeyWrapper<KEY> newKeyWrapper = new KeyWrapper<KEY>(key);
+		WeakReference<KeyWrapper<KEY>> ref = map.get(newKeyWrapper);
+		KeyWrapper<KEY> oldKeyWrapper = (ref == null) ? null : ref.get();
+		if (oldKeyWrapper != null) {
+			return oldKeyWrapper;
+		}
+		map.put(newKeyWrapper,
+				new WeakReference<KeyWrapper<KEY>>(newKeyWrapper));
+		return newKeyWrapper;
+	}
+
+	// Returns the number of used entries in the map.
+	public synchronized int size() {
+		return map.size();
+	}
+
+	// KeyWrapper wraps a key value and is used in three ways:
+	// - as the key for the internal WeakHashMap
+	// - as the value for the internal WeakHashMap, additionally wrapped in a
+	// WeakReference
+	// - as the lock object associated to the key
+	private static class KeyWrapper<KEY> {
+		private KEY key;
+		private int hashCode;
+
+		public KeyWrapper(KEY key) {
+			this.key = key;
+			hashCode = key.hashCode();
+		}
+
+		public boolean equals(Object obj) {
+			if (obj == this) {
+				return true;
+			}
+			if (obj instanceof KeyWrapper) {
+				return ((KeyWrapper<?>) obj).key.equals(key);
+			}
+			return false;
+		}
+
+		public int hashCode() {
+			return hashCode;
+		}
+	}
+
+} // end class LockMap

src/main/java/com/commafeed/backend/MetricsBean.java

@@ -33,13 +33,17 @@ public class MetricsBean {
 		thisHour.feedsRefreshed++;
 	}
 
-	public void feedUpdated(int entriesCount, int statusesCount) {
+	public void feedUpdated() {
 		thisHour.feedsUpdated++;
 		thisMinute.feedsUpdated++;
 
-		thisHour.entriesInserted += entriesCount;
+	}
-		thisMinute.entriesInserted += entriesCount;
+
-		
+	public void entryUpdated(int statusesCount) {
+
+		thisHour.entriesInserted++;
+		thisMinute.entriesInserted++;
+
 		thisHour.statusesInserted += statusesCount;
 		thisMinute.statusesInserted += statusesCount;
 	}

src/main/java/com/commafeed/backend/dao/FeedEntryDAO.java

@@ -5,6 +5,7 @@ import java.util.List;
 import javax.ejb.Stateless;
 import javax.persistence.TypedQuery;
 import javax.persistence.criteria.CriteriaQuery;
+import javax.persistence.criteria.JoinType;
 import javax.persistence.criteria.Root;
 import javax.persistence.criteria.SetJoin;
 
@@ -14,30 +15,19 @@ import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
 import com.commafeed.backend.model.FeedEntry_;
 import com.commafeed.backend.model.Feed_;
-import com.google.api.client.util.Lists;
-import com.uaihebert.model.EasyCriteria;
 
 @Stateless
 public class FeedEntryDAO extends GenericDAO<FeedEntry> {
 
-	public List<FeedEntry> findByGuids(List<String> guids) {
+	public List<FeedEntry> findByGuid(String guid) {
-		List<String> hashes = Lists.newArrayList();
+		String hash = DigestUtils.sha1Hex(guid);
-		for (String guid : guids) {
-			hashes.add(DigestUtils.sha1Hex(guid));
-		}
 
-		EasyCriteria<FeedEntry> criteria = createCriteria();
+		CriteriaQuery<FeedEntry> query = builder.createQuery(getType());
-		criteria.setDistinctTrue();
+		Root<FeedEntry> root = query.from(getType());
-		criteria.andStringIn(FeedEntry_.guidHash.getName(), hashes);
+		query.where(builder.equal(root.get(FeedEntry_.guidHash), hash));
-		criteria.leftJoinFetch(FeedEntry_.feeds.getName());
+		root.fetch(FeedEntry_.feeds, JoinType.LEFT);
-
+		TypedQuery<FeedEntry> q = em.createQuery(query);
-		List<FeedEntry> list = Lists.newArrayList();
+		return q.getResultList();
-		for (FeedEntry entry : criteria.getResultList()) {
-			if (guids.contains(entry.getGuid())) {
-				list.add(entry);
-			}
-		}
-		return list;
 	}
 
 	public List<FeedEntry> findByFeed(Feed feed, int offset, int limit) {

src/main/java/com/commafeed/backend/feeds/FeedRefreshTaskGiver.java

@@ -6,10 +6,8 @@ import java.util.List;
 import java.util.Queue;
 
 import javax.annotation.PostConstruct;
-import javax.ejb.Lock;
-import javax.ejb.LockType;
-import javax.ejb.Singleton;
 import javax.inject.Inject;
+import javax.inject.Singleton;
 
 import org.apache.commons.lang3.time.DateUtils;
 
@@ -34,7 +32,8 @@ public class FeedRefreshTaskGiver {
 	private int backgroundThreads;
 
 	private Queue<Feed> addQueue = Queues.newConcurrentLinkedQueue();
-	private Queue<Feed> queue = Queues.newConcurrentLinkedQueue();
+	private Queue<Feed> takeQueue = Queues.newConcurrentLinkedQueue();
+	private Queue<Feed> giveBackQueue = Queues.newConcurrentLinkedQueue();
 
 	@PostConstruct
 	public void init() {
@@ -42,7 +41,6 @@ public class FeedRefreshTaskGiver {
 				.getBackgroundThreads();
 	}
 
-	@Lock(LockType.WRITE)
 	public void add(Feed feed) {
 		Date now = Calendar.getInstance().getTime();
 		boolean heavyLoad = applicationSettingsService.get().isHeavyLoad();
@@ -52,31 +50,40 @@ public class FeedRefreshTaskGiver {
 			feed.setEtagHeader(null);
 			feed.setLastModifiedHeader(null);
 		}
-
 		addQueue.add(feed);
 	}
 
-	@Lock(LockType.WRITE)
+	public synchronized Feed take() {
-	public Feed take() {
+		Feed feed = takeQueue.poll();
-		Feed feed = queue.poll();
 		if (feed == null) {
 			int count = Math.min(100, 5 * backgroundThreads);
 			List<Feed> feeds = feedDAO.findNextUpdatable(count);
 
-			int addQueueSize = queue.size();
+			int size = addQueue.size();
-			for (int i = 0; i < addQueueSize; i++) {
+			for (int i = 0; i < size; i++) {
 				feeds.add(addQueue.poll());
 			}
 
 			for (Feed f : feeds) {
-				queue.add(f);
+				takeQueue.add(f);
 				f.setLastUpdated(Calendar.getInstance().getTime());
 			}
+
+			size = giveBackQueue.size();
+			for (int i = 0; i < size; i++) {
+				feeds.add(giveBackQueue.poll());
+			}
+
 			feedDAO.update(feeds);
-			feed = queue.poll();
+
+			feed = takeQueue.poll();
 		}
 		metricsBean.feedRefreshed();
 		return feed;
 	}
 
+	public void giveBack(Feed feed) {
+		giveBackQueue.add(feed);
+	}
+
 }

src/main/java/com/commafeed/backend/feeds/FeedRefreshUpdater.java

@@ -1,69 +1,70 @@
 package com.commafeed.backend.feeds;
 
 import java.util.Collection;
+import java.util.List;
 
-import javax.ejb.Asynchronous;
-import javax.ejb.Stateless;
 import javax.inject.Inject;
+import javax.inject.Singleton;
 
-import org.apache.commons.collections.CollectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.commafeed.backend.LockMap;
 import com.commafeed.backend.dao.FeedDAO;
+import com.commafeed.backend.dao.FeedSubscriptionDAO;
 import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
-import com.commafeed.backend.model.FeedEntryContent;
 import com.commafeed.backend.model.FeedPushInfo;
+import com.commafeed.backend.model.FeedSubscription;
 import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
 import com.commafeed.backend.services.ApplicationSettingsService;
 import com.commafeed.backend.services.FeedUpdateService;
 
-@Stateless
+@Singleton
 public class FeedRefreshUpdater {
 
 	protected static Logger log = LoggerFactory
 			.getLogger(FeedRefreshUpdater.class);
 
+	private static LockMap<String> lockMap = new LockMap<String>();
+
 	@Inject
 	FeedUpdateService feedUpdateService;
 
 	@Inject
 	SubscriptionHandler handler;
 
+	@Inject
+	FeedRefreshTaskGiver taskGiver;
+
 	@Inject
 	FeedDAO feedDAO;
 
 	@Inject
 	ApplicationSettingsService applicationSettingsService;
 
-	@Asynchronous
+	@Inject
-	public void updateEntries(Feed feed, Collection<FeedEntry> entries) {
+	FeedSubscriptionDAO feedSubscriptionDAO;
-		if (CollectionUtils.isNotEmpty(entries)) {
+
+	public void updateFeed(Feed feed, Collection<FeedEntry> entries) {
+		if (entries != null) {
+			List<FeedSubscription> subscriptions = feedSubscriptionDAO
+					.findByFeed(feed);
 			for (FeedEntry entry : entries) {
-				handleEntry(feed, entry);
+				updateEntry(feed, entry, subscriptions);
 			}
-			feedUpdateService.updateEntries(feed, entries);
 		}
-		feedDAO.update(feed);
+
 		if (applicationSettingsService.get().isPubsubhubbub()) {
 			handlePubSub(feed);
 		}
+		taskGiver.giveBack(feed);
 	}
 
-	private void handleEntry(Feed feed, FeedEntry entry) {
+	private void updateEntry(Feed feed, FeedEntry entry,
-		String baseUri = feed.getLink();
+			List<FeedSubscription> subscriptions) {
-		FeedEntryContent content = entry.getContent();
+		synchronized (lockMap.get(entry.getGuid())) {
-
+			feedUpdateService.updateEntry(feed, entry, subscriptions);
-		content.setContent(FeedUtils.handleContent(content.getContent(),
-				baseUri));
-		String title = FeedUtils.handleContent(content.getTitle(), baseUri);
-		if (title != null) {
-			content.setTitle(title.substring(0, Math.min(2048, title.length())));
-		}
-		String author = entry.getAuthor();
-		if (author != null) {
-			entry.setAuthor(author.substring(0, Math.min(128, author.length())));
 		}
 	}
 

src/main/java/com/commafeed/backend/feeds/FeedRefreshWorker.java

@@ -5,12 +5,6 @@ import java.util.Collection;
 import java.util.Date;
 
 import javax.inject.Inject;
-import javax.jms.JMSException;
-import javax.transaction.HeuristicMixedException;
-import javax.transaction.HeuristicRollbackException;
-import javax.transaction.NotSupportedException;
-import javax.transaction.RollbackException;
-import javax.transaction.SystemException;
 
 import org.apache.commons.lang.mutable.MutableBoolean;
 import org.apache.commons.lang3.StringUtils;
@@ -22,6 +16,7 @@ import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
 import com.commafeed.backend.model.FeedPushInfo;
 import com.commafeed.backend.services.ApplicationSettingsService;
+import com.commafeed.backend.services.FeedPushInfoService;
 import com.sun.syndication.io.FeedException;
 
 public class FeedRefreshWorker {
@@ -41,6 +36,9 @@ public class FeedRefreshWorker {
 	@Inject
 	ApplicationSettingsService applicationSettingsService;
 
+	@Inject
+	FeedPushInfoService feedPushInfoService;
+
 	public void start(MutableBoolean running, String threadName) {
 		log.info("{} starting", threadName);
 
@@ -77,10 +75,7 @@ public class FeedRefreshWorker {
 		}
 	}
 
-	private void update(Feed feed) throws NotSupportedException,
+	private void update(Feed feed) {
-			SystemException, SecurityException, IllegalStateException,
-			RollbackException, HeuristicMixedException,
-			HeuristicRollbackException, JMSException {
 
 		FetchedFeed fetchedFeed = null;
 		Collection<FeedEntry> entries = null;
@@ -130,7 +125,7 @@ public class FeedRefreshWorker {
 		feed.setMessage(message);
 		feed.setDisabledUntil(disabledUntil);
 
-		feedRefreshUpdater.updateEntries(feed, entries);
+		feedRefreshUpdater.updateFeed(feed, entries);
 
 	}
 
@@ -141,7 +136,7 @@ public class FeedRefreshWorker {
 			log.debug("feed {} has pubsub info: {}", feed.getUrl(), topic);
 			FeedPushInfo info = feed.getPushInfo();
 			if (info == null) {
-				info = new FeedPushInfo();
+				info = feedPushInfoService.findOrCreate(feed, hub, topic);
 			}
 			if (!StringUtils.equals(hub, info.getHub())
 					|| !StringUtils.equals(topic, info.getTopic())) {

src/main/java/com/commafeed/backend/services/FeedPushInfoService.java

@@ -0,0 +1,40 @@
+package com.commafeed.backend.services;
+
+import java.util.List;
+
+import javax.ejb.Lock;
+import javax.ejb.LockType;
+import javax.ejb.Singleton;
+import javax.inject.Inject;
+
+import com.commafeed.backend.dao.FeedPushInfoDAO;
+import com.commafeed.backend.model.Feed;
+import com.commafeed.backend.model.FeedPushInfo;
+import com.commafeed.backend.model.FeedPushInfo_;
+
+@Singleton
+public class FeedPushInfoService {
+
+	@Inject
+	FeedPushInfoDAO feedPushInfoDAO;
+
+	@Lock(LockType.WRITE)
+	public FeedPushInfo findOrCreate(Feed feed, String hub, String topic) {
+		FeedPushInfo info = null;
+
+		List<FeedPushInfo> infos = feedPushInfoDAO.findByField(
+				FeedPushInfo_.feed, feed);
+		if (infos.isEmpty()) {
+			info = new FeedPushInfo();
+			info.setFeed(feed);
+			info.setHub(hub);
+			info.setTopic(topic);
+			info.setActive(false);
+			feedPushInfoDAO.save(info);
+		} else {
+			info = infos.get(0);
+		}
+		return info;
+	}
+
+}

src/main/java/com/commafeed/backend/services/FeedUpdateService.java

@@ -1,15 +1,10 @@
 package com.commafeed.backend.services;
 
-import java.util.ArrayList;
 import java.util.Calendar;
-import java.util.Collection;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
+import java.util.Set;
 
-import javax.ejb.AccessTimeout;
+import javax.ejb.Stateless;
-import javax.ejb.Lock;
-import javax.ejb.LockType;
-import javax.ejb.Singleton;
 import javax.inject.Inject;
 
 import org.apache.commons.lang.ObjectUtils;
@@ -19,13 +14,15 @@ import com.commafeed.backend.MetricsBean;
 import com.commafeed.backend.dao.FeedEntryDAO;
 import com.commafeed.backend.dao.FeedEntryStatusDAO;
 import com.commafeed.backend.dao.FeedSubscriptionDAO;
+import com.commafeed.backend.feeds.FeedUtils;
 import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
+import com.commafeed.backend.model.FeedEntryContent;
 import com.commafeed.backend.model.FeedEntryStatus;
 import com.commafeed.backend.model.FeedSubscription;
 import com.google.common.collect.Lists;
 
-@Singleton
+@Stateless
 public class FeedUpdateService {
 
 	@Inject
@@ -40,74 +37,78 @@ public class FeedUpdateService {
 	@Inject
 	MetricsBean metricsBean;
 
-	@Lock(LockType.WRITE)
+	public void updateEntry(Feed feed, FeedEntry entry,
-	@AccessTimeout(value = 5, unit = TimeUnit.MINUTES)
+			List<FeedSubscription> subscriptions) {
-	public void updateEntries(Feed feed, Collection<FeedEntry> entries) {
 
-		List<FeedEntry> existingEntries = getExistingEntries(entries);
+		FeedEntry foundEntry = findEntry(
-		List<FeedSubscription> subscriptions = feedSubscriptionDAO
+				feedEntryDAO.findByGuid(entry.getGuid()), entry);
-				.findByFeed(feed);
 
-		List<FeedEntry> entryUpdateList = Lists.newArrayList();
+		FeedEntry update = null;
-		List<FeedEntryStatus> statusUpdateList = Lists.newArrayList();
+		if (foundEntry == null) {
-		for (FeedEntry entry : entries) {
+			handleEntry(feed, entry);
+			entry.setInserted(Calendar.getInstance().getTime());
+			entry.getFeeds().add(feed);
 
-			FeedEntry foundEntry = findEntry(existingEntries, entry);
+			update = entry;
+		} else {
 
-			if (foundEntry == null) {
+			if (!findFeed(foundEntry.getFeeds(), feed)) {
-				entry.setInserted(Calendar.getInstance().getTime());
+				foundEntry.getFeeds().add(feed);
-				entry.getFeeds().add(feed);
+				update = foundEntry;
-				entryUpdateList.add(entry);
-			} else {
-				boolean foundFeed = false;
-				for (Feed existingFeed : foundEntry.getFeeds()) {
-					if (ObjectUtils.equals(existingFeed.getId(), feed.getId())) {
-						foundFeed = true;
-						break;
-					}
-				}
-
-				if (!foundFeed) {
-					foundEntry.getFeeds().add(feed);
-					entryUpdateList.add(foundEntry);
-				}
 			}
 		}
-		for (FeedEntry entry : entryUpdateList) {
+
+		if (update != null) {
+			List<FeedEntryStatus> statusUpdateList = Lists.newArrayList();
 			for (FeedSubscription sub : subscriptions) {
 				FeedEntryStatus status = new FeedEntryStatus();
-				status.setEntry(entry);
+				status.setEntry(update);
 				status.setSubscription(sub);
 				statusUpdateList.add(status);
 			}
+			feedEntryDAO.saveOrUpdate(update);
+			feedEntryStatusDAO.saveOrUpdate(statusUpdateList);
+			metricsBean.entryUpdated(statusUpdateList.size());
 		}
-
-		feedEntryDAO.saveOrUpdate(entryUpdateList);
-		feedEntryStatusDAO.saveOrUpdate(statusUpdateList);
-		metricsBean
-				.feedUpdated(entryUpdateList.size(), statusUpdateList.size());
 	}
 
 	private FeedEntry findEntry(List<FeedEntry> existingEntries, FeedEntry entry) {
-		FeedEntry foundEntry = null;
+		FeedEntry found = null;
-		for (FeedEntry existingEntry : existingEntries) {
+		for (FeedEntry existing : existingEntries) {
-			if (StringUtils.equals(entry.getGuid(), existingEntry.getGuid())
+			if (StringUtils.equals(entry.getGuid(), existing.getGuid())
-					&& StringUtils.equals(entry.getUrl(),
+					&& StringUtils.equals(entry.getUrl(), existing.getUrl())) {
-							existingEntry.getUrl())) {
+				found = existing;
-				foundEntry = existingEntry;
 				break;
 			}
 		}
-		return foundEntry;
+		return found;
 	}
 
-	private List<FeedEntry> getExistingEntries(Collection<FeedEntry> entries) {
+	private boolean findFeed(Set<Feed> feeds, Feed feed) {
-		List<String> guids = Lists.newArrayList();
+		boolean found = false;
-		for (FeedEntry entry : entries) {
+		for (Feed existingFeed : feeds) {
-			guids.add(entry.getGuid());
+			if (ObjectUtils.equals(existingFeed.getId(), feed.getId())) {
+				found = true;
+				break;
+			}
 		}
-		List<FeedEntry> existingEntries = guids.isEmpty() ? new ArrayList<FeedEntry>()
+		return found;
-				: feedEntryDAO.findByGuids(guids);
-		return existingEntries;
 	}
+
+	private void handleEntry(Feed feed, FeedEntry entry) {
+		String baseUri = feed.getLink();
+		FeedEntryContent content = entry.getContent();
+
+		content.setContent(FeedUtils.handleContent(content.getContent(),
+				baseUri));
+		String title = FeedUtils.handleContent(content.getTitle(), baseUri);
+		if (title != null) {
+			content.setTitle(title.substring(0, Math.min(2048, title.length())));
+		}
+		String author = entry.getAuthor();
+		if (author != null) {
+			entry.setAuthor(author.substring(0, Math.min(128, author.length())));
+		}
+	}
+
 }

src/main/tomee/conf/tomee.xml

@@ -1,5 +1,14 @@
 <?xml version="1.0"?>
 <tomee>
+
+	<Resource id="My DataSource" type="DataSource">
+	    jdbcDriver = org.hsqldb.jdbcDriver
+	    jdbcUrl = jdbc:hsqldb:file:data/hsqldb;hsqldb.tx=mvcc
+	    UserName = sa
+	    Password = 
+	    MaxActive = 50
+	</Resource>
+
 <!-- 
 	<Resource id="MySQL" type="DataSource">
 		JdbcDriver com.mysql.jdbc.Driver