package simpledb.buffer;

import simpledb.file.*;
import java.util.*;

/**
 * Manages the pinning and unpinning of buffers to blocks.
 * @author Edward Sciore
 *
 */
class BasicBufferMgr {
	private Map<Block,Buffer>allocated = new HashMap<Block,Buffer>();
	private List<Buffer>     unpinned  = new LinkedList<Buffer>();
	private int numAvailable;

	/**
	 * Creates a buffer manager having the specified number
	 * of buffer slots.
	 * @param numbuffs the number of buffer slots to allocate
	 */
	BasicBufferMgr(int numbuffs) {
		numAvailable = numbuffs;
		for (int i=0; i<numbuffs; i++)
			unpinned.add(new Buffer(i));
	}

	/**
	 * Flushes the dirty buffers modified by the specified transaction.
	 * @param txnum the transaction's id number
	 */
	synchronized void flushAll(int txnum) {
		for (Buffer buff : allocated.values())
			if (buff.isModifiedBy(txnum))
				buff.flush();
	}

	/**
	 * Pins a buffer to the specified block.
	 * If there is already a buffer assigned to that block
	 * then that buffer is used;
	 * otherwise, an unpinned buffer from the pool is chosen.
	 * Returns a null value if there are no available buffers.
	 * @param blk a reference to a disk block
	 * @return the pinned buffer
	 */
	synchronized Buffer pin(Block blk) {
		Buffer buff = findExistingBuffer(blk);
		if (buff == null) {
			buff = chooseUnpinnedBuffer();
			if (buff == null)
				return null;
			buff.assignToBlock(blk);
			allocated.put(blk, buff);
		}
		if (!buff.isPinned())
			numAvailable--;
		buff.pin();
		return buff;
	}

	/**
	 * Allocates a new block in the specified file, and
	 * pins a buffer to it.
	 * Returns null (without allocating the block) if
	 * there are no available buffers.
	 * @param filename the name of the file
	 * @param fmtr a pageformatter object, used to format the new block
	 * @return the pinned buffer
	 */
	synchronized Buffer pinNew(String filename, PageFormatter fmtr) {
		Buffer buff = chooseUnpinnedBuffer();
		if (buff == null)
			return null;
		buff.assignToNew(filename, fmtr);
		allocated.put(buff.block(), buff);
		numAvailable--;
		buff.pin();
		return buff;
	}

	/**
	 * Unpins the specified buffer.
	 * @param buff the buffer to be unpinned
	 */
	synchronized void unpin(Buffer buff) {
		buff.unpin();
		if (!buff.isPinned()) {
			unpinned.add(buff);
			numAvailable++;
		}
	}

	/**
	 * Returns the number of available (i.e. unpinned) buffers.
	 * @return the number of available buffers
	 */
	int available() {
		return numAvailable;
	}

	private Buffer findExistingBuffer(Block blk) {
		return allocated.get(blk);
	}

	private Buffer chooseUnpinnedBuffer() {
		if (unpinned.isEmpty())
			return null;
		Buffer buff = unpinned.remove(0);
		allocated.remove(buff.block());
		return buff;
	}

	public String toString() {
		String result = "";
		for (Buffer buff : allocated.values())
			result += buff.toString() + "\n";
		return result + "============";
	}
}