integralitymanager.cpp

GiPS是一个面向数据密集型应用的分布式文件系统

#pragma once
#include "IntegralityManager.h"
#include <set>
#include <iostream>

using namespace std;
using namespace Cluster;
using namespace ClientMaster;

#define REPLICA_NUM 2

IntegralityManager::IntegralityManager(BlockTable* blockTable, Cluster::LoadTable* loadTable, Cluster::ClerkMap* clerkMap, ClerkFileList* clerkFileList) :
	m_blockTable(blockTable), m_loadTable(loadTable), m_clerkMap(clerkMap), m_clerkFileList(clerkFileList)
{
	this->m_loadBalancer = new GLoadBalancer(m_loadTable);
}

void
IntegralityManager::maintain(vector<string>* badClerks)
{
	cout << "begin maintain integrality" << endl;
	cout << "bad clerk num: " << badClerks->size() << endl;
	set<PCFBlock>* fileSet = new set<PCFBlock>();
	// remove bad clerk from BlockTable
	for (vector<string>::iterator cit = badClerks->begin(); cit != badClerks->end(); cit++) {
		ClerkFileList::iterator cfit = m_clerkFileList->find(*cit);
		if (cfit != m_clerkFileList->end())	{
			cout << "Now handle clerk: " << *cit << endl;
			SubFileList* sflist = cfit->second;
			for (SubFileList::iterator sfit = sflist->begin(); sfit != sflist->end(); sfit++) {
				PCFBlock pcfb = *sfit;
				BlockTable::iterator bit = m_blockTable->find(pcfb->fid);
				if (bit != m_blockTable->end())	{
					FileObject* fobj = bit->second;
					FileBlock fb = (*fobj)[pcfb->blockid-1];
					for(desIPList::iterator ipit = fb.IPs.begin(); ipit != fb.IPs.end(); ipit++) {
						int ans = ipit->compare(cfit->first);
						cout<<"answer: "<<ans<<endl;
						if(ans == 0){
							cout << "Erase " << *ipit << endl;
							fb.IPs.erase(ipit);
							cout << "BLock ID: " << fb.indexID << endl;
							cout << "Index ID: " << fb.blockID << endl;
							cout << "Left: " << fb.IPs.size() << endl;
							break;
						}
					}

					if (fb.IPs.size() < REPLICA_NUM) {
						fileSet->insert(*sfit);
						cout << fb.indexID << " needs backup" << endl;
					}

					(*fobj)[pcfb->blockid-1] = fb;
				}
			}
			m_clerkFileList->erase(cfit);
		}
	}

	// back up those blocks which has less replica than the standard
	for (set<PCFBlock>::iterator sit = fileSet->begin(); sit != fileSet->end(); sit++) {
		PCFBlock pcfb = *sit;

		BlockTable::iterator bit = m_blockTable->find(pcfb->fid);
		if (bit != m_blockTable->end()) {
			cout << "Block ID: " << pcfb->blockid << endl;
			cout << "File ID: " << pcfb->fid << endl;
			FileObject fobj = *(bit->second);
			FileBlock fb = fobj[pcfb->blockid-1];

			desIPList* iplist = m_loadBalancer->chooseClerk(REPLICA_NUM-fb.IPs.size(), fb.IPs);
			cout << iplist->size() << endl;

			ReplicaInfo info;
			info.replicaID = fb.indexID;

			for (int i = 0; i < fb.IPs.size(); i++)
				cout << "Avalable: " << fb.IPs[i] << endl;

			cout << "source: " << fb.IPs[0] << endl;
			ClerkMap::iterator mit = m_clerkMap->find(fb.IPs[0]);
			if (mit != m_clerkMap->end()) {
				cout << "DO !!!" << endl;
				for (desIPList::iterator dit = iplist->begin(); dit != iplist->end(); dit++) {
					info.clerkIP = *dit;
					cout << *dit << " is doing backup" << endl;
					if (mit->second->replicaCopy(info)) {
						fb.IPs.push_back(info.clerkIP);
						for (LoadTable::iterator lit = m_loadTable->begin(); lit != m_loadTable->end(); lit++) {
							if (!(*lit)->clerkIP.compare(info.clerkIP)) {
								(*lit)->blockNum++;
								break;
							}
						}
					}
					else
						cout << "Integrality Error at " << info.clerkIP << endl;
				}
			}
		}
	}

	delete fileSet;
	cout << "Maintain finished" << endl;
}