CircularBufferSTM.java

package assignment13.stm;

import scala.concurrent.stm.Ref;
import scala.concurrent.stm.TArray;
import scala.concurrent.stm.japi.STM;

import java.util.concurrent.Callable;

/**
 * This class implements a {@link assignment13.stm.CircularBuffer} using software-transactional memory (more
 * specifically using ScalaSTM [http://nbronson.github.io/scala-stm/]).
 */
public class CircularBufferSTM<E> implements CircularBuffer<E> {
	
	private final TArray.View<E> buffer;
	private final Ref.View<Integer> size;
	private final Ref.View<Integer> in;
	private final Ref.View<Integer> out;
	
    CircularBufferSTM(int capacity) {
    	
    	// we initialize the size with capacity + 1 because one element in the buffer is not usable
    	// in order to be able to distinguish between isFull and isEmpty
        buffer = STM.newTArray(capacity + 1);
        size = STM.newRef(capacity + 1);
        in = STM.newRef(0);
        out = STM.newRef(0);
    }

    public void put(final E item) {
        STM.atomic(new Runnable() {
        	public void run() {
        		if(isFull()) {
        			STM.retry();
        		}
        		
        		buffer.update(in.get(), item);
        		in.update((in.get() + 1) % size.get());
        	}
        });
    }

    public E take() {
        return STM.atomic(new Callable<E>() {
        	public E call() {
        		if(isEmpty()) {
        			STM.retry();
        		}
        		
        		E item = buffer.apply(out.get());
        		out.update((out.get() + 1) % size.get());
        		return item;
        	}
        });
    }

    public boolean isEmpty() {
        return STM.atomic(new Callable<Boolean>() {
        	public Boolean call() {
        		return in.get() == out.get();
        	}
        });
    }

    public boolean isFull() {
        return STM.atomic(new Callable<Boolean>() {
        	public Boolean call() {
        		return (in.get() + 1) % size.get() == out.get();
        	}
        });
    }
}