/* Filename: Bucket.h
 *
 * Programmer: Papaloukopoulos K. George and S.Sioutas
 *
 * Date: June 23, 2006
 * Updated: October 1, 2006
 *
 * This header file implements the data structure of buckets, which 
 * is a set of two layers lists. This data structure is the implementation
 * of the leaves of the uper level of the btree.
 */

#ifndef __BUCKET_H__
#define __BUCKET_H__

#include <math.h>
#include "DLLNode.h"
#include "BTree.h"

extern int N;
int operation;

#define FUSE	1
#define SPLIT	2
#define	TRANS	3

/* Given:   Nothing.
 *           
 * Task:    Defines the prototype class of Bucket.
 *	    Also defines the methods and the data members of 
 *	    this class.	
 * Return:  The implicity Bucket object.
 */
template <class T>
class Bucket {

	public:
		Bucket (long loc);
		Bucket ( );
		long insert (T key);
		long remove (T key);
		DLLNode<T> * search (T key);
		Bucket<T> * rebalancing (btr::BTree<T> * bt);
		double criticality (int N);
		//double prevCriticality ();
		DLLNode<T> * getBucketHead ();
        	void display();
		long getLocation();

	private:
		long location;			//location of the bucket into the disk
		long prevBucketLoc;		//location of the previous bucket into the file 
		long nextBucketLoc;		//location of the next bucket into the file
		long listHeadLoc;		//location of the first layer list into the file
		T bucketRepKey;

	public:
		int bucketSize;			//the number of keys which are stored inside the bucket
		double prevCritic;
		long ctritListNodeLoc;  	//node location of third layer list of criticality
		int l1;				//index for first layer list of criticality
		int l2;				//index for second layer list of criticality
	private:	
		Bucket<T> * split();					//create a new bucket
		void fuse(Bucket<T> * dest, Bucket<T> * with);		//fuse two buckets
		void transfer(Bucket<T> * dest, Bucket<T> * from);	//transfer some blocks from one bucket to a sibling bucket
};	


/* Given:   The offset where the bucket will be stored into the disk.
 *           
 * Task:    Is the constructor of this class.Allocates memory,
 *	    construct the two layer lists and initialize the 
 *	    variables of this class.	
 * Return:  The implicity Bucket object.
 */
template <class T>
Bucket<T>::Bucket(long loc) {

	DLLNode<T> * listHeadNode = new DLLNode<T>();
	LLNode<T> * node = new LLNode<T>();

	location = loc;  							//allocate memory for the bucket 
	listHeadLoc = D->getFreeDiskBlock(sizeof(DLLNode<T>));			//allocate memory for the dll node of the bucket

	listHeadNode->list[0] = D->getFreeDiskBlock(sizeof(LLNode<T>));		//allocate memory for the ll node of the ddl node
    	listHeadNode->location = listHeadLoc;
	D->write((char *)listHeadNode, sizeof(DLLNode<T>), listHeadLoc);
	
    	node->location = listHeadNode->list[0];
	D->write((char *)node, sizeof(LLNode<T>), listHeadNode->list[0]);

	bucketSize = 0;
	prevBucketLoc = nextBucketLoc = -1;

	prevCritic = 0;//criticality(N); 
	
	D->write((char *)this, sizeof(Bucket<T>), this->location);
	
	//free(listHeadNode);
	//free(node);
    	/*cout<<"llnSize = "<<sizeof(LLNode<T>)<<endl;
    	cout<<"dllnSize = "<<sizeof(DLLNode<T>)<<endl;
    	cout<<"bucketSize = "<<sizeof(Bucket<T>)<<endl;
    	cout<<"BUCKET: location = "<<location<<endl;
    	cout<<"BUCKET: listHeadLoc = "<<listHeadLoc<<endl;

    	cout<<"BUCKET: listHeadNode = "<<listHeadNode->list[0]<<endl;
	*/
}

/* Given:   Nothing.
 *           
 * Task:    This is the second constructor of this class.
 * 	    Initializes the variables of the class without 	
 *	    allocate any memory in the disk...Only use is for
 *	    reading already stored buckets. 	
 * Return:  The implicity Bucket object.
 */
template <class T>
Bucket<T>::Bucket() {

	bucketSize = 0;
	prevBucketLoc = -1;
	nextBucketLoc = -1;
}

/* Given:   
 *           
 * Task:    
 * Return:  
 */
template <class T>
long Bucket<T>::getLocation() {

	return location;
}


template <class T>
void Bucket<T>::display() {
	
	int i,j,t = 0;

    	LLNode<T> * node = new LLNode<T>();
	DLLNode<T> * listHeadNode = new DLLNode<T>();
    	D->read((char *)listHeadNode, sizeof(DLLNode<T>), listHeadLoc);

	while(true) {

    		for (i = 0; i < listHeadNode->nodeSize; i++) {
        
        		D->read((char *)node, sizeof(LLNode<T>), listHeadNode->list[i]);
			
			t += node->listSize;
			cout<<"["<<listHeadNode->keys[i]<<"]->";
			
			while (true) {
        			for (j = 0; j < node->nodeSize; j++)
            				cout<<" "<<node->keys[j];
				if (node->nextNodeLoc != -1){ 
					D->read((char *)node, sizeof(LLNode<T>), node->nextNodeLoc);
					cout<<" // ";
				}
				else
					break;
			}
			cout<<endl;
    		}
		cout<<"-- "<<t<<" --"<<endl;
		t = 0;
		//cout<<"DLL "<<listHeadNode->location<<" "<<listHeadNode->nextNodeLoc<<endl;
		if (listHeadNode->nextNodeLoc != -1) {
    			D->read((char *)listHeadNode, sizeof(DLLNode<T>), listHeadNode->nextNodeLoc);
			cout<<"DLL "<<endl;
		}
		else
			break;
	}
	//free(listHeadNode);
	//free(node);
}


/* Given:   The key that we want to insert into the data structure.
 *           
 * Task:    Search into the two layer list to find the coresponding 
 * 	    block where will be inserted the key. If the second layer
 * 	    list has size bigger than log2(N), then split it.
 * 	
 * Return:  Nothing.
 */
template <class T>
long Bucket<T>::insert (T key) {

	long flag = 0;
	LLNode<T> * newHead = new LLNode<T>();

	DLLNode<T> * dlln = search (key);			//this is the corresponding node of the 1st layer
	
	if (dlln->nodeSize == 0) {
		dlln->keys[0] = key;
		dlln->nodeSize++;
		D->write((char *)dlln, sizeof(DLLNode<T>), dlln->location);
	}

	LLNode<T> * head = dlln->findLList (key); 		//this is the head of the 2nd layer list
	//cout<<"HeadListLoc="<<head->location<<endl;
	if ((flag = head->insert(key)) != -1) {		//the key is already inserted
		head->listSize++;
		bucketSize++;
	}

	if (head->listSize >= log2(N) && head->listSize != head->nodeSize) {
		//cout<<"SPLIT LIST"<<endl;
		newHead = head->splitList (head);
		dlln->insert(newHead);
		
	}
	D->write((char *)head, sizeof(LLNode<T>), head->location);
	D->write((char *)this, sizeof(Bucket<T>), location);
	
	//free(newHead);
	
	return flag;
}

/* Given:   The key that we want to delete from the data structure.
 *           
 * Task:    Search into the two layer list to find the coresponding 
 * 	    block where the key lie. If the second layer
 * 	    list has size smaller than log2(N), then fuse or transfer 
 * 	    with/from a sibling list.
 * Return:  Nothing.
 */
template <class T>
long Bucket<T>::remove (T key) {

	int i = 0, j = 0, k = 0;
	long flag = 0;
	LLNode<T> * lln = new LLNode<T>();

	DLLNode<T> * dlln = search (key);			//this is the corresponding node of the 1st layer
	LLNode<T> * head = dlln->findLList (key); 		//this is the head of the list, 2nd layer

	//head->listSize--;
	//head->remove (key);
	//bucketSize--;
    	//cout<<"BUCKET:remove  dllnKey = "<<dlln->keys<<endl;
	
	if ((flag = head->remove (key)) != -1) {
		head->listSize--;
		bucketSize--;
	
		if (head->listSize < log2(N)/2) {
			LLNode<T> * prevHeadList = new LLNode<T>();
			LLNode<T> * nextHeadList = new LLNode<T>();

			j = dlln->nodeSize;
			for (i = 0; i < dlln->nodeSize; i++)
				if (key < dlln->keys[i]) {
					j = i;
					break;
				}
			//
        		//find the previous and next lists of the cuurent list
			//
			if (j > 1 && j < dlln->nodeSize) {	//the corresponding key is in dlln->list[j-1]
				D->read ((char *)prevHeadList, sizeof(LLNode<T>), dlln->list[j-2]);
				//k = j - 2; 
			}
			else if (j == dlln->nodeSize) {	//the corresponding key is in dlln->list[j-1]
				D->read ((char *)prevHeadList, sizeof(LLNode<T>), dlln->list[j-1]);
				//k = j - 1;
			}
			
			if (j < MAX_KEYS)	//then we can read the next list
				D->read ((char *)nextHeadList, sizeof(LLNode<T>), dlln->list[j]);
			//
			//check out if we can fuse or tranferm any list
			//
        		if (j > 1 && (prevHeadList->listSize + head->listSize) <= log2(N)) {
			
				cout<<"BUCKET:remove... fuseList with prevHeadList "<<endl;
				head->fuseList (prevHeadList, head);
				dlln->remove(head);
				//head = prevHeadList;	//now the key which we removed is in the prevHeadList
			}
        		else if (j < MAX_KEYS && (nextHeadList->listSize + head->listSize) <= log2(N)) {   
			
				cout<<"BUCKET:remove... fuseList with nextHeadList "<<endl;
				head->fuseList (head, nextHeadList);
				dlln->remove(nextHeadList);
			}
			else if (j > 1 && prevHeadList->listSize >= log2(N)/2 && prevHeadList->listSize > MAX_KEYS) {
				//the last condition means that we have at least two nodes into this list
				cout<<"BUCKET:remove... transferList from prevHeadList "<<endl;
				//cout<<"2a).Head->keys[0]="<<head->keys[0]<<endl;
				head->transferList (head, prevHeadList);
				dlln->keys[j-1] = head->keys[0];
				dlln->list[j-1] = head->location;
				//cout<<"2b).Head->keys[0]="<<head->keys[0]<<endl;
				D->write((char *)dlln, sizeof(DLLNode<T>), dlln->location);
			}
			else if (j < MAX_KEYS && nextHeadList->listSize >= log2(N)/2 && nextHeadList->listSize > MAX_KEYS) {
				//the last condition means that we have at least two nodes into this list
				cout<<"BUCKET:remove... transferList from nextHeadList "<<endl;
				//cout<<"2a).nextHeadList->keys[0]="<<nextHeadList->keys[0]<<endl;
				head->transferList (head, nextHeadList);
				dlln->keys[j] = nextHeadList->keys[0];
				dlln->list[j] = nextHeadList->location;
				//cout<<"2b).nextHeadList->keys[0]="<<nextHeadList->keys[0]<<endl;
				D->write((char *)dlln, sizeof(DLLNode<T>), dlln->location);
			}
			//free(prevHeadList);
			//free(nextHeadList);
		}
		//D->write((char *)head, sizeof(LLNode<T>), head->location);
		D->write((char *)this, sizeof(Bucket<T>), location);
	}

	//free(lln);

	return flag;
}

/* Given:   A key.
 *           
 * Task:    Search into the first layer list to find the coresponding
 * 	    node of this layer list where the key is lie.
 * Return:  A double linked list node of the first layer list.
 */
template <class T>
DLLNode<T> * Bucket<T>::search (T key) {

	DLLNode<T> * dlln = getBucketHead ();

    	//cout<<"dllLoc = "<<dlln->location<<" ListHeadLoc = "<<listHeadLoc<<endl;

	return dlln->search (key);		 //the corresponding head of the list, 2nd layer
}

/* Given:   Nothing.
 *           
 * Task:    Return the first node of the first layer list.
 *
 * Return:  A double linked list node of the first layer list.
 */
template <class T>
DLLNode<T> * Bucket<T>::getBucketHead() {

	DLLNode<T> * node = new DLLNode<T>();
	D->read ((char *)node, sizeof(DLLNode<T>), listHeadLoc);

	//cout<<"getBucketHead="<<node->location<<" listHeadLoc="<<listHeadLoc<<endl;

	return node;
}

/* Given:   Nothing.
 *           
 * Task:    Defines the prototype classes of BTree and BTNode.
 *	    Also difines the methods and the data members of 
 *	    those classes.	
 * Return:  The implicity BTree object.
 */
template <class T>
double Bucket<T>::criticality (int N) {
	
	double alpha = 1.2;

	if (N == 0)
		N = 1;
	return max(0.7*pow(log2(N),2)-bucketSize, bucketSize-1.8*pow(log2(N),2))/(alpha*log2(N)); 
}

/* Given:   Nothing.
 *           
 * Task:    Defines the prototype classes of BTree and BTNode.
 *	    Also difines the methods and the data members of 
 *	    those classes.	
 * Return:  The implicity BTree object.
 *
template <class T>
double Bucket<T>::prevCriticality () {

	return prevCritic;
}
*/
/* Given:   The btree.
 *           
 * Task:    Check the most critical bucket and if it is critical
 * 	    perform the the appropriate rebalancing operation for this bucket.
 * 	    This operation will affect the btree.
 * Return:  Nothing.
 */
template <class T>
Bucket<T> * Bucket<T>::rebalancing (btr::BTree<T> * bt) {

	int i = 0;
	double fullness = bucketSize/(pow(log2(N),2));
	DLLNode<T> * dllHead = getBucketHead();
	Bucket<T> * leftSiblingBucket = new Bucket<T>();
	Bucket<T> * rightSiblingBucket = new Bucket<T>();

	cout<<"BUCKET: bucketSize = "<<bucketSize<<" -- N = "<<N<<"  cur = "<<location<<"  prev = "<<prevBucketLoc<<"  next = "<<nextBucketLoc<<endl;
	cout<<"BUCKET: rebalancing... fullness = "<<fullness<<endl;
	if (fullness > 1.8) {
		Bucket<T> * newBucket = split();
		bt->insert(newBucket->bucketRepKey, newBucket->location);
		bt->display();
		
		operation = SPLIT;
		return newBucket;
	}
	else if (fullness < 0.7) {
		if (prevBucketLoc != -1) {
			D->read((char *)leftSiblingBucket, sizeof(Bucket<T>), prevBucketLoc);
			if (leftSiblingBucket->bucketSize/(pow(log2(N),2)) >= 1) {
				T tmp = (this->getBucketHead())->keys[0];
				transfer(this, leftSiblingBucket);
				bt->replaceKey(tmp, (this->getBucketHead())->keys[0]);
				operation = TRANS;
				cout<<"BUCKET:rebalancing... transfer from leftSibling "<<endl;
			}
			else {
				fuse(leftSiblingBucket, this);
				bt->remove(this->bucketRepKey);
				operation = FUSE;
				cout<<"BUCKET:rebalancing... fuse with leftSibling "<<endl;
			}
				
			return leftSiblingBucket;
		}
		else if (nextBucketLoc != -1) {
			D->read((char *)rightSiblingBucket, sizeof(Bucket<T>), nextBucketLoc);
			if (rightSiblingBucket->bucketSize/(pow(log2(N),2)) >= 1) {
				T tmp = rightSiblingBucket->getBucketHead()->keys[0];
				transfer(this, rightSiblingBucket);
				bt->replaceKey(tmp, rightSiblingBucket->getBucketHead()->keys[0]);
				operation = TRANS;
				cout<<"BUCKET:rebalancing... transfer from rightSibling prevKey = "<<tmp<<"  newKey = "<<rightSiblingBucket->getBucketHead()->keys[0]<<endl;
				bt->display();
			}
			else {
				cout<<"FUSE BUCKET: right->bucketRepKey="<<rightSiblingBucket->getBucketHead()->keys[0]<<" RSL="<<rightSiblingBucket->location<<endl;
				fuse(this, rightSiblingBucket);
				cout<<"FUSE BUCKET: right->bucketRepKey="<<rightSiblingBucket->bucketRepKey<<" RSL="<<rightSiblingBucket->location<<endl;
				bt->remove(rightSiblingBucket->bucketRepKey);
				operation = FUSE;
				cout<<"BUCKET:rebalancing... fuse with rightSibling "<<endl;
				bt->display();
			}

			return rightSiblingBucket;
		}
	}

	return NULL;
}

/* Given:   Nothing.
 *           
 * Task:    The bucket has too many keys and needs to be splited.	
 * Return:  Nothing.
 */
template <class T>
Bucket<T> * Bucket<T>::split( ) {

	int i = 0, sum = 0;
	bool flag = false;
	DLLNode<T> * splitNode = new DLLNode<T>();
	DLLNode<T> * newDLLHeadNode = new DLLNode<T>();
	DLLNode<T> * prevDLLNode = new DLLNode<T>();
	LLNode<T> * listHead  = new LLNode<T>();
	Bucket<T> * newBucket = new Bucket<T>();
	Bucket<T> * dummy = new Bucket<T>();

	//find the corresponding DLLNode which split the bucket to 2 equal size buckets
	splitNode = getBucketHead();
	
	//while (sum < bucketSize/2 && splitNode->nextNodeLoc != -1) {
	while (sum/(pow(log2(N),2)) < 0.8 && splitNode->nextNodeLoc != -1) {
		D->read ((char *)listHead, sizeof(LLNode<T>), splitNode->list[i]);
		sum += listHead->listSize;
		i++;

		if (i == splitNode->nodeSize && splitNode->nextNodeLoc != -1) {
			D->read ((char *)splitNode, sizeof(DLLNode<T>), splitNode->nextNodeLoc);
			i = 0;
		}
	}
	
	while (i != 0 && i <= splitNode->nodeSize-1) {
		flag = true;
		D->read ((char *)listHead, sizeof(LLNode<T>), splitNode->list[i]);
		sum += listHead->listSize;
		i++;
	}

	newBucket->location = D->getFreeDiskBlock(sizeof(Bucket<T>));
	newBucket->bucketSize = bucketSize-sum;
	bucketSize = sum;
	newBucket->prevBucketLoc = location;
	newBucket->nextBucketLoc = nextBucketLoc;
	
	
	if (nextBucketLoc != -1) {
		D->read ((char *)dummy, sizeof(Bucket<T>), nextBucketLoc);
		dummy->prevBucketLoc = newBucket->location;
		D->write ((char *)dummy, sizeof(Bucket<T>), dummy->location);
		cout<<"1"<<endl;
	}
	nextBucketLoc = newBucket->location;

	if (splitNode->nextNodeLoc != -1 && flag == true) {	//it has been changed!!!
		newBucket->listHeadLoc = splitNode->nextNodeLoc;
		splitNode->nextNodeLoc = -1;
		D->write ((char *)splitNode, sizeof(DLLNode<T>), splitNode->location);
		cout<<"2"<<endl;
        	D->read ((char *)newDLLHeadNode, sizeof(DLLNode<T>), newBucket->listHeadLoc);
        	newDLLHeadNode->prevNodeLoc = -1;
        	D->write ((char *)newDLLHeadNode, sizeof(DLLNode<T>), newBucket->listHeadLoc);
    	}
	else {
		newBucket->listHeadLoc = splitNode->location; 	
        	D->read ((char *)prevDLLNode, sizeof(DLLNode<T>), splitNode->prevNodeLoc);
        	prevDLLNode->nextNodeLoc = -1;
        	D->write ((char *)prevDLLNode, sizeof(DLLNode<T>), prevDLLNode->location);
		cout<<"3"<<endl;
		splitNode->prevNodeLoc = -1;
		D->write ((char *)splitNode, sizeof(DLLNode<T>), splitNode->location);
    	}

	newBucket->bucketRepKey = (newBucket->getBucketHead())->keys[0];
	
	D->write((char *)newBucket, sizeof(Bucket<T>), newBucket->location);
	D->write((char *)this, sizeof(Bucket<T>), location);

	//cout<<"[BUCKET] newBucket->listHeadLoc="<<newBucket->listHeadLoc<<" splitBucket->listHeadLoc="<<listHeadLoc<<endl;

	//free(splitNode);
	//free(newDLLHeadNode);
	//free(prevDLLNode);
	//free(listHead);
	//free(dummy);

	return newBucket;
}


/* Given:   The bucket where the blocks will be tranfered and the bucket
 * 	    where the blocks will be removed.
 *           
 * Task:    Transfer some blocks between two buckets.
 *
 * Return:  Nothing.
 */
template <class T>
void Bucket<T>::transfer(Bucket<T> * dest, Bucket<T> * from) {
	
	bool flag = false;
	int i = 0, sum;
	long newDLLHeadLoc = 0;

	double fullness = dest->bucketSize/pow(log2(N),2);

	LLNode<T> * listHead = new LLNode<T>();
	DLLNode<T> * tail = new DLLNode<T>();
	DLLNode<T> * prevFromNode = new DLLNode<T>();

	DLLNode<T> * destNode = dest->getBucketHead();
	DLLNode<T> * fromNode = from->getBucketHead();

	if (fromNode->nextNodeLoc == -1 && fromNode->nodeSize >= MAX_KEYS/2)	//this bucket has only one node in the first layer list
		fromNode->split(fromNode);
	
	if (destNode->keys[0] < fromNode->keys[0]) {		
		
		//cout<<"BUCKET:transfer... destKey[0] = "<<destNode->keys[0]<<"  fromKey[0] = "<<fromNode->keys[0]<<endl;
		D->read((char *)tail, sizeof(DLLNode<T>), destNode->location);
		while (tail->nextNodeLoc != -1)
			D->read((char *)tail, sizeof(DLLNode<T>), tail->nextNodeLoc);

		tail->nextNodeLoc = fromNode->location;
		fromNode->prevNodeLoc = tail->location;
		
		sum = dest->bucketSize;
		while (sum/pow(log2(N),2) < 0.8) {
			D->read ((char *)listHead, sizeof(LLNode<T>), fromNode->list[i]);
			sum += listHead->listSize;
			i++;

			if (i == fromNode->nodeSize && fromNode->nextNodeLoc != -1) {
				//cout<<"Second dlln..."<<endl;
				D->read ((char *)fromNode, sizeof(DLLNode<T>), fromNode->nextNodeLoc);
				i = 0;
			}
		}
	
		while (i != 0 && i <= fromNode->nodeSize-1) {
			
			flag = true;
			D->read ((char *)listHead, sizeof(LLNode<T>), fromNode->list[i]);
			sum += listHead->listSize;
			i++;
		}
		//cout<<"Sum = "<<sum<<endl;
		from->bucketSize -= (sum - dest->bucketSize);
		dest->bucketSize = sum;

		if (flag == true) {

			from->listHeadLoc = fromNode->nextNodeLoc;
			fromNode->nextNodeLoc = -1;
			D->write ((char *)fromNode, sizeof(DLLNode<T>), fromNode->location);	
			D->read ((char *)fromNode, sizeof(DLLNode<T>), from->listHeadLoc);
			fromNode->prevNodeLoc = -1;
			D->write ((char *)fromNode, sizeof(DLLNode<T>), fromNode->location);
		}
		else {

			from->listHeadLoc = fromNode->location;
			D->read ((char *)prevFromNode, sizeof(DLLNode<T>), fromNode->prevNodeLoc);	
			prevFromNode->nextNodeLoc = -1;
			D->write ((char *)prevFromNode, sizeof(DLLNode<T>), prevFromNode->location);	
			fromNode->prevNodeLoc = -1;
			D->write ((char *)fromNode, sizeof(DLLNode<T>), fromNode->location);
		}
		
		from->bucketRepKey = fromNode->keys[0];
		D->write ((char *)tail, sizeof(DLLNode<T>), tail->location);	
	}
	else {							
		D->read((char *)tail, sizeof(DLLNode<T>), fromNode->location);
		while (tail->nextNodeLoc != -1)
			D->read((char *)tail, sizeof(DLLNode<T>), tail->nextNodeLoc);

		tail->nextNodeLoc = destNode->location;
		destNode->prevNodeLoc = tail->location;
	
		sum = dest->bucketSize;
		i = tail->nodeSize - 1;
		while (sum/pow(log2(N),2) < 0.8) {
			D->read ((char *)listHead, sizeof(LLNode<T>), tail->list[i]);
			sum += listHead->listSize;
			i--;

			if (i == -1 && tail->prevNodeLoc != -1) {
				D->read ((char *)tail, sizeof(DLLNode<T>), tail->prevNodeLoc);
				i = tail->nodeSize-1;
			}
		}
	
		while (i > 0 && i != tail->nodeSize-1) {
			flag = true;
			D->read ((char *)listHead, sizeof(LLNode<T>), tail->list[i]);
			sum += listHead->listSize;
			i--;
		}

		from->bucketSize -= (sum - dest->bucketSize);
		dest->bucketSize = sum;

		if (flag == true) {
			dest->listHeadLoc = tail->location;
			
			D->read ((char *)fromNode, sizeof(DLLNode<T>), tail->prevNodeLoc);
			fromNode->nextNodeLoc = -1;
			D->write ((char *)fromNode, sizeof(DLLNode<T>), fromNode->location);
			tail->prevNodeLoc = -1;
			D->write ((char *)tail, sizeof(DLLNode<T>), dest->listHeadLoc);
		}
		else {
			dest->listHeadLoc = tail->nextNodeLoc;
			D->read ((char *)fromNode, sizeof(DLLNode<T>), tail->nextNodeLoc);
			fromNode->prevNodeLoc = -1;
			D->write ((char *)fromNode, sizeof(DLLNode<T>), fromNode->location);
			tail->nextNodeLoc = -1;
			D->write ((char *)tail, sizeof(DLLNode<T>), dest->listHeadLoc);
		}

		dest->bucketRepKey = destNode->keys[0];		
		D->write ((char *)destNode, sizeof(DLLNode<T>), destNode->location);
	}
	D->write ((char *)dest, sizeof(Bucket<T>), dest->location);
	D->write ((char *)from, sizeof(Bucket<T>), from->location);
}


/* Given:   The buckets which will be fused.
 *           
 * Task:    Fuse the buckets.
 *
 * Return:  Nothing.
 */
template <class T>
void Bucket<T>::fuse(Bucket<T> * dest, Bucket<T> * with) {

	DLLNode<T> * dummy = new DLLNode<T>();
	DLLNode<T> * w_head = with->getBucketHead();
	DLLNode<T> * d_head = dest->getBucketHead();

	dummy = d_head;
	while (dummy->nextNodeLoc != -1)
		D->read ((char *)dummy, sizeof(DLLNode<T>), dummy->nextNodeLoc);

	dummy->nextNodeLoc = w_head->location;
	w_head->prevNodeLoc = dummy->location;
	
	dest->bucketSize += with->bucketSize;
	dest->nextBucketLoc = with->nextBucketLoc;
    
    	if (with->nextBucketLoc != -1) {
        	Bucket<T> * buc = new Bucket<T>();
        	D->read ((char *)buc, sizeof(Bucket<T>), with->nextBucketLoc);
        	buc->prevBucketLoc = dest->location;
        	D->write ((char *)buc, sizeof(Bucket<T>), buc->location);
		//cout<<"######## buc="<<dummy->location<<" fusenode="<<w_head->location<<" bucket="<<dest->location<<endl;
    	}

	D->write ((char *)dummy, sizeof(DLLNode<T>), dummy->location);
   	D->write ((char *)w_head, sizeof(DLLNode<T>), w_head->location);
	D->write ((char *)dest, sizeof(Bucket<T>), dest->location);
}

#endif
