package voldemort.store.pausable;

import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.log4j.Logger;

import voldemort.VoldemortException;
import voldemort.annotations.jmx.JmxOperation;
import voldemort.store.StorageEngine;
import voldemort.store.StoreCapabilityType;
import voldemort.store.memory.InMemoryStorageEngine;
import voldemort.utils.ClosableIterator;
import voldemort.utils.Pair;
import voldemort.versioning.Version;
import voldemort.versioning.Versioned;

/**
 * A storage engine that can be paused via JMX to simulate a failure for
 * testing. While paused all operations on the store will block indefinitely.
 * 
 * @author jay
 * 
 * @param <K> The type of the key
 * @param <V> The type of the value
 */
public class PausableStorageEngine<K, V> implements StorageEngine<K, V> {

    private static final Logger logger = Logger.getLogger(PausableStorageEngine.class);

    private final InMemoryStorageEngine<K, V> inner;
    private final ReadWriteLock lock;

    public PausableStorageEngine(InMemoryStorageEngine<K, V> inner) {
        super();
        this.inner = inner;
        this.lock = new ReentrantReadWriteLock();
    }

    public void close() throws VoldemortException {
        inner.close();
    }

    public boolean delete(K key, Version version) {
        try {
            lock.readLock().lock();
            return inner.delete(key);
        } finally {
            lock.readLock().unlock();
        }
    }

    public List<Versioned<V>> get(K key) {
        try {
            lock.readLock().lock();
            return inner.get(key);
        } finally {
            lock.readLock().unlock();
        }
    }

    public Map<K, List<Versioned<V>>> getAll(Iterable<K> keys) {
        try {
            lock.readLock().lock();
            return inner.getAll(keys);
        } finally {
            lock.readLock().unlock();
        }
    }

    public void put(K key, Versioned<V> value) {
        try {
            lock.readLock().lock();
            inner.put(key, value);
        } finally {
            lock.readLock().unlock();
        }
    }

    public ClosableIterator<Pair<K, Versioned<V>>> entries() {
        return inner.entries();
    }

    public Object getCapability(StoreCapabilityType capability) {
        return inner.getCapability(capability);
    }

    public String getName() {
        return inner.getName();
    }

    @JmxOperation(description = "Pause all operations on the storage engine.")
    public void pause() {
        logger.info("Pausing store '" + getName() + "'.");
        this.lock.writeLock().lock();
    }

    @JmxOperation(description = "Unpause the storage engine.")
    public void unpause() {
        logger.info("Unpausing store '" + getName() + "'.");
        this.lock.writeLock().unlock();
    }

}

@@ -76,115 +76,8 @@
 Map;
-%0Aimport java.util.concurrent.locks.ReadWriteLock;%0Aimport java.util.concurrent.locks.ReentrantReadWriteLock;
 %0A%0Aim
@@ -520,16 +520,8 @@
 sed 
-via JMX 
 to s
@@ -541,19 +541,16 @@
 lure for
-%0A *
  testing
@@ -556,16 +556,19 @@
 g. While
+%0A *
  paused 
@@ -619,16 +619,82 @@
 initely.
+ The methods to%0A * pause and unpause are also exposed through JMX.
 %0A * %0A * 
@@ -1022,26 +1022,76 @@
 nal 
-ReadWriteLock lock
+Object condition = new Object();%0A    private volatile boolean paused
 ;%0A%0A 
@@ -1208,58 +1208,8 @@
 er;%0A
-        this.lock = new ReentrantReadWriteLock();%0A
     
@@ -1357,46 +1357,24 @@
     
-try %7B%0A            lock.readLock().lock
+blockIfNecessary
 ();%0A
@@ -1373,36 +1373,32 @@
 sary();%0A        
-    
 return inner.del
@@ -1415,61 +1415,82 @@
     
+%7D%0A%0A
     
-%7D finally %7B%0A            lock.readLock().unlock();
+private void blockIfNecessary() %7B%0A        synchronized(condition) %7B
 %0A   
@@ -1498,56 +1498,24 @@
     
-%7D%0A
     
-%7D%0A%0A    public List%3CVersioned%3CV%3E%3E get(K key
+while(paused
 ) %7B%0A
@@ -1514,32 +1514,40 @@
 used) %7B%0A        
+  
+      
 try %7B%0A          
@@ -1552,124 +1552,209 @@
     
-lock.readLock().lock();%0A            return inner.get(key);%0A        %7D finally %7B%0A            lock.readLock().unlock();
+        condition.wait();%0A                %7D catch(InterruptedException e) %7B%0A                    throw new VoldemortException(%22Pausable store interrupted while paused.%22);%0A                %7D%0A            %7D
 %0A   
@@ -1781,15 +1781,8 @@
 blic
- Map%3CK,
  Lis
@@ -1800,33 +1800,18 @@
 %3CV%3E%3E
-%3E
  get
-All(Iterable%3CK%3E
+(K
  key
-s
 ) %7B%0A
@@ -1822,54 +1822,28 @@
     
-try %7B%0A            lock.readLock().lock();%0A    
+blockIfNecessary();%0A
     
@@ -1866,295 +1866,264 @@
 .get
-All
 (key
-s
 );%0A    
+%7D%0A%0A
     
-%7D finally %7B%0A            lock.readLock().unlock();%0A        %7D%0A    %7D%0A%0A    public void put(K key, Versioned%3CV%3E value) %7B%0A        try %7B%0A            lock.readLock().lock();%0A            inner.put(key, value);%0A        %7D finally %7B%0A            lock.readLock().unlock();%0A        %7D
+public Map%3CK, List%3CVersioned%3CV%3E%3E%3E getAll(Iterable%3CK%3E keys) %7B%0A        blockIfNecessary();%0A        return inner.getAll(keys);%0A    %7D%0A%0A    public void put(K key, Versioned%3CV%3E value) %7B%0A        blockIfNecessary();%0A        inner.put(key, value);
 %0A   
@@ -2594,36 +2594,21 @@
     
-this.lock.writeLock().lock()
+paused = true
 ;%0A  
@@ -2776,39 +2776,102 @@
     
-this.lock.writeLock().unlock();
+paused = false;%0A        synchronized(condition) %7B%0A            condition.notifyAll();%0A        %7D
 %0A   

Fix Issue 139: pause in PausableStorageEngine has no effect if other methods called from same thread.