diff --git a/commons/src/main/java/org/archive/bdb/StoredQueue.java b/commons/src/main/java/org/archive/bdb/StoredQueue.java index 671f8b4b..636784e5 100644 --- a/commons/src/main/java/org/archive/bdb/StoredQueue.java +++ b/commons/src/main/java/org/archive/bdb/StoredQueue.java @@ -22,14 +22,9 @@ package org.archive.bdb; import java.io.Serializable; import java.util.AbstractQueue; import java.util.Iterator; -import java.util.LinkedList; import java.util.concurrent.atomic.AtomicLong; -import java.util.logging.Level; import java.util.logging.Logger; -import org.apache.commons.lang.StringUtils; - - import com.sleepycat.bind.EntryBinding; import com.sleepycat.bind.serial.SerialBinding; import com.sleepycat.bind.serial.StoredClassCatalog; @@ -45,17 +40,15 @@ import com.sleepycat.je.DatabaseException; * * @param */ -public class StoredQueue extends AbstractQueue -implements Serializable { - private static final long serialVersionUID = 3L; +public class StoredQueue extends AbstractQueue { + @SuppressWarnings("unused") private static final Logger logger = Logger.getLogger(StoredQueue.class.getName()); transient StoredSortedMap queueMap; // Long -> E transient Database queueDb; // Database AtomicLong tailIndex; // next spot for insert - AtomicLong headIndex; // next spot for read - transient E peekItem = null; + transient volatile E peekItem = null; /** * Create a StoredQueue backed by the given Database. @@ -70,7 +63,6 @@ implements Serializable { */ public StoredQueue(Database db, Class clsOrNull, StoredClassCatalog classCatalog) { tailIndex = new AtomicLong(0); - headIndex = new AtomicLong(0); hookupDatabase(db, clsOrNull, classCatalog); } @@ -99,70 +91,39 @@ implements Serializable { @Override public int size() { - synchronized(tailIndex) { - synchronized(headIndex) { - return (int)(tailIndex.get()-headIndex.get()); - } + return queueMap.size(); + } + + @Override + public boolean isEmpty() { + if(peekItem!=null) { + return false; } - + return queueMap.isEmpty(); } public boolean offer(E o) { - synchronized (tailIndex) { - queueMap.put(tailIndex.get(), o); - tailIndex.getAndIncrement(); - } + long targetIndex = tailIndex.getAndIncrement(); + queueMap.put(targetIndex, o); return true; } - public E peek() { - synchronized (headIndex) { - if(peekItem != null) { - return peekItem; + public synchronized E peek() { + if(peekItem == null) { + if(queueMap.isEmpty()) { + return null; } - E head = null; - while(head == null && headIndex.get() < tailIndex.get()) { - head = (E) queueMap.get(headIndex.get()); - if(head != null) { - peekItem = head; - return head; - } - // ERROR; should never be null with headIndex < tailIndex - logger.log(Level.SEVERE, - "unexpected empty index of StoredQueue(" - + queueDb.getDatabaseName()+"): " - + headIndex.get() + " (tailIndex: " - + tailIndex.get() +") ", //+StringUtils.join(rememberedOps, ","), - new Exception()); - headIndex.incrementAndGet(); - } - return head; + peekItem = queueMap.remove(queueMap.firstKey()); } + return peekItem; } - public E poll() { - synchronized (headIndex) { - E head = peek(); - if(head!=null) { - long hi = headIndex.getAndIncrement(); - queueMap.remove(hi); -// debugRemember(-hi); - peekItem = null; - return head; - } else { - return null; - } - } + public synchronized E poll() { + E head = peek(); + peekItem = null; + return head; } -// LinkedList rememberedOps = new LinkedList(); -// protected void debugRemember(long l) { -// rememberedOps.addFirst(l); -// if(rememberedOps.size()>20) { -// rememberedOps.removeLast(); -// } -// } - /** * A suitable DatabaseConfig for the Database backing a StoredQueue. * (However, it is not necessary to use these config options.) @@ -176,24 +137,6 @@ implements Serializable { return dbConfig; } - /** - * Save the state to a stream (that is, serialize it). - * - * @serialData The capacity is emitted (int), followed by all of - * its elements (each an Object) in the proper order, - * followed by a null - * @param s the stream - */ - private void writeObject(java.io.ObjectOutputStream s) - throws java.io.IOException { - try { - queueDb.sync(); - } catch (DatabaseException e) { - throw new RuntimeException(e); - } - s.defaultWriteObject(); - } - public void close() { try { queueDb.sync();