<commit_before>
#include "include/types.h"
#include "common/Clock.h"

#include "msg/Messenger.h"

#include "osd/OSDCluster.h"
#include "osd/Filer.h"

#include "MDS.h"
#include "MDCache.h"
#include "MDStore.h"
#include "MDLog.h"
#include "MDCluster.h"
#include "MDBalancer.h"
#include "IdAllocator.h"
#include "AnchorTable.h"
#include "OSDMonitor.h"

#include "include/filepath.h"

#include "common/Logger.h"
#include "common/LogType.h"

#include "messages/MPing.h"
#include "messages/MPingAck.h"
#include "messages/MGenericMessage.h"

#include "messages/MClientMount.h"
#include "messages/MClientMountAck.h"
#include "messages/MClientRequest.h"
#include "messages/MClientReply.h"

#include "messages/MLock.h"
#include "messages/MInodeWriterClosed.h"

#include "messages/MInodeLink.h"

#include "events/EInodeUpdate.h"

#include <errno.h>
#include <fcntl.h>

#include <list>
#include <iostream>
using namespace std;


LogType mds_logtype;

#include "include/config.h"
#undef dout
#define  dout(l)    if (l<=g_conf.debug) cout << "mds" << whoami << " "
#define  dout3(l,mds)    if (l<=g_conf.debug) cout << "mds" << mds->get_nodeid() << " "



ostream& operator<<(ostream& out, MDS& mds)
{
  out << "mds" << mds.get_nodeid() << " ";
}

void C_MDS_RetryMessage::redelegate(MDS *mds, int newmds)
{
  // forward message to new mds
  dout3(5,mds) << "redelegating context " << this << " by forwarding message " << m << " to mds" << newmds << endl;

  mds->messenger->send_message(m,
							   newmds, m->get_dest_port(),
							   MDS_PORT_MAIN);  // mostly meaningless
}



// extern 
//MDS *g_mds;


// cons/des
MDS::MDS(MDCluster *mdc, int whoami, Messenger *m) {
  this->whoami = whoami;
  mdcluster = mdc;

  messenger = m;
  messenger->set_dispatcher(this);

  mdcache = new MDCache(this);
  mdstore = new MDStore(this);
  mdlog = new MDLog(this);
  balancer = new MDBalancer(this);
  anchormgr = new AnchorTable(this);
  osdmonitor = new OSDMonitor(this);


  // <HACK set up OSDCluster from g_conf>
  osdcluster = new OSDCluster();
  OSDGroup osdg;
  osdg.num_osds = g_conf.num_osd;
  for (int i=0; i<osdg.num_osds; i++) osdg.osds.push_back(i);
  osdg.weight = 100;
  osdcluster->add_group(osdg);
  // </HACK>

  filer = new Filer(messenger, osdcluster);

  mdlog->set_max_events(g_conf.mds_log_max_len);

  shutting_down = false;
  shut_down = false;

  stat_ops = 0;
  last_heartbeat = 0;

  // log
  string name;
  name = "mds";
  int w = MSG_ADDR_NUM(whoami);
  if (w >= 1000) name += ('0' + ((w/1000)%10));
  if (w >= 100) name += ('0' + ((w/100)%10));
  if (w >= 10) name += ('0' + ((w/10)%10));
  name += ('0' + ((w/1)%10));

  logger = new Logger(name, (LogType*)&mds_logtype);

  // alloc
  idalloc = new IdAllocator(this);

}

MDS::~MDS() {
  if (mdcache) { delete mdcache; mdcache = NULL; }
  if (mdstore) { delete mdstore; mdstore = NULL; }
  if (mdlog) { delete mdlog; mdlog = NULL; }
  if (balancer) { delete balancer; balancer = NULL; }

  if (logger) { delete logger; logger = 0; }

  if (messenger) { delete messenger; messenger = NULL; }

  if (idalloc) { delete idalloc; idalloc = NULL; }
}


int MDS::init()
{
}


int MDS::shutdown_start()
{
  dout(1) << "shutdown_start" << endl;
  for (int i=0; i<mdcluster->get_num_mds(); i++) {
	if (i == whoami) continue;
	dout(1) << "sending MShutdownStart to mds" << i << endl;
	messenger->send_message(new MGenericMessage(MSG_MDS_SHUTDOWNSTART),
							i, MDS_PORT_MAIN,
							MDS_PORT_MAIN);
  }

  handle_shutdown_start(NULL);
}


void MDS::handle_shutdown_start(Message *m)
{
  dout(1) << " handle_shutdown_start" << endl;

  // set flag
  shutting_down = true;

  mdcache->shutdown_start();
  
  // save anchor table
  anchormgr->save(0);  // FIXME FIXME

  // flush log
  mdlog->set_max_events(0);
  mdlog->trim(NULL);

  if (m) delete m;
}

void MDS::handle_shutdown_finish(Message *m)
{
  int mds = whoami;
  if (m) 
	mds = m->get_source();
						 
  dout(2) << "handle_shutdown_finish from " << mds << endl;
  did_shut_down.insert(mds);
  dout(2) << " shut down so far: " << did_shut_down << endl;
  
  if (did_shut_down.size() == mdcluster->get_num_mds()) {
	// MDS's all shut down!

	// shut down osd's
	for (int i=0; i<g_conf.num_osd; i++) {
	  messenger->send_message(new MGenericMessage(MSG_SHUTDOWN),
							  MSG_ADDR_OSD(i), 0, 0);
	}

	// shut myself down.
	shutting_down = false;

	shutdown_final();
  }

  // done
  delete m;
}



int MDS::shutdown_final()
{
  // shut down cache
  mdcache->shutdown();
  
  // shut down messenger
  messenger->shutdown();

  return 0;
}



mds_load_t MDS::get_load()
{
  timepair_t now = g_clock.gettimepair();
  mds_load_t l;
  if (mdcache->get_root()) {
	l.root_pop = mdcache->get_root()->popularity[MDS_POP_ANYDOM].get(now);
  } else
	l.root_pop = 0;
  l.req_rate = stat_req.get(now);
  l.rd_rate = stat_read.get(now);
  l.wr_rate = stat_write.get(now);
  return l;
}


class C_MDS_Unpause : public Context {
public:
  MDS *mds;
  C_MDS_Unpause(MDS *mds) {
	this->mds = mds;
  }
  void finish(int r) {
	mds->mds_paused = false;
	mds->queue_finished(mds->waiting_for_unpause);
  }
};


void MDS::proc_message(Message *m) 
{
  switch (m->get_type()) {
	// OSD ===============
  case MSG_OSD_READREPLY:
	filer->handle_osd_read_reply((MOSDReadReply*)m);
	return;
  case MSG_OSD_WRITEREPLY:
	filer->handle_osd_write_reply((MOSDWriteReply*)m);
	return;
  case MSG_OSD_OPREPLY:
	filer->handle_osd_op_reply((MOSDOpReply*)m);
	return;

	// MDS
  case MSG_MDS_SHUTDOWNSTART:
	handle_shutdown_start(m);
	return;

  case MSG_MDS_SHUTDOWNFINISH:
	handle_shutdown_finish(m);
	return;

  case MSG_PING:
	handle_ping((MPing*)m);
	return;
  }


  // paused?
  if (mds_paused) {
	dout(3) << "paused" << endl;
	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	return;
  }


  switch (m->get_type()) {
  case MSG_CLIENT_MOUNT:
	handle_client_mount((MClientMount*)m);
	return;
  case MSG_CLIENT_UNMOUNT:
	handle_client_unmount(m);
	return;
  }
 

  // need idalloc
  if (!idalloc->is_open()) {
	dout(3) << "idalloc not open yet" << endl;

	mds_paused = true;
	idalloc->load(new C_MDS_Unpause(this));
	
	// defer
	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	return;
  }


  switch (m->get_type()) {
  case MSG_CLIENT_REQUEST:
	handle_client_request((MClientRequest*)m);
	return;
  }

  dout(1) << " main unknown message " << m->get_type() << endl;
  assert(0);
}


void MDS::dispatch(Message *m)
{
  mds_lock.Lock();

  my_dispatch(m);

  mds_lock.Unlock();
}

void MDS::my_dispatch(Message *m)
{

  switch (m->get_dest_port()) {
	
  case MDS_PORT_STORE:
	mdstore->proc_message(m);
	break;

  case MDS_PORT_ANCHORMGR:
	anchormgr->proc_message(m);
	break;
	
  case MDS_PORT_OSDMON:
	osdmonitor->proc_message(m);
	break;

  case MDS_PORT_CACHE:
	mdcache->proc_message(m);
	break;

	/*
  case MSG_PORT_MDLOG:
	mymds->logger->proc_message(m);
	break;
	*/
	
  case MDS_PORT_BALANCER:
	balancer->proc_message(m);
	break;
	

  case MDS_PORT_MAIN:
  case MDS_PORT_SERVER:
	proc_message(m);
	break;

  default:
	dout(1) << "MDS dispatch unknown message port" << m->get_dest_port() << endl;
	assert(0);
  }


  // HACK FOR NOW
  /*
  static bool did_heartbeat_hack = false;
  if (!shutting_down && !shut_down &&
	  false && 
	  !did_heartbeat_hack) {
	osdmonitor->initiate_heartbeat();
	did_heartbeat_hack = true;
  }
  */
	

  // finish any triggered contexts
  if (finished_queue.size()) {
	dout(7) << "mds has " << finished_queue.size() << " queued contexts" << endl;
	list<Context*> ls;
	ls.splice(ls.begin(), finished_queue);
	assert(finished_queue.empty());
	finish_contexts(ls);
  }

  // balance?
  if (true && whoami == 0 &&
	  stat_ops >= last_heartbeat + g_conf.mds_bal_interval) {
	last_heartbeat = stat_ops;
	balancer->send_heartbeat();
  }

  if (whoami == 0) {
	static bool didit = false;
	
	// 7 to 1
	CInode *in = mdcache->get_inode(1001);
	if (in && in->is_dir() && !didit) {
	  CDir *dir = in->get_or_open_dir(this);
	  if (dir->is_auth()) {
		dout(1) << "FORCING EXPORT" << endl;
		mdcache->export_dir(dir,1);
		didit = true;
	  }
	}
  }

  // shut down?
  if (shutting_down && !shut_down) {
	if (mdcache->shutdown_pass()) {
	  shutting_down = false;
	  shut_down = true;
	  if (whoami) shutdown_final();
	}
  }

}


void MDS::handle_client_mount(MClientMount *m)
{
  // mkfs?  (sorta hack!)
  if (int cmd = m->get_mkfs()) {
	dout(3) << "MKFS flag is set" << endl;
	if (mdcache->get_root()) {
	  dout(3) << "   root inode is already open" << endl;
	} else {
	  dout(3) << "   root inode isn't open yet, inventing a fresh filesystem" << endl;
	  
	  mdcache->open_root(0);
	  CInode *root = mdcache->get_root();
	  assert(root);

	  // force empty root dir
	  CDir *dir = root->dir;
	  dir->mark_complete();
	  dir->mark_dirty();

	  if (cmd == MDS_MKFS_FULL) {
		// wipe osds too
		dout(3) << "wiping osds too" << endl;
		mds_paused = true;
		filer->mkfs(new C_MDS_Unpause(this));
	  	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	  	return;
	  }
	  
	}
  }

  int n = MSG_ADDR_NUM(m->get_source());
  dout(3) << "mount by client" << n << endl;
  mounted_clients.insert(n);

  assert(whoami == 0);  // mds0 mounts/unmounts



  // ack<commit_msg>*** empty log message ***<commit_after>
#include "include/types.h"
#include "common/Clock.h"

#include "msg/Messenger.h"

#include "osd/OSDCluster.h"
#include "osd/Filer.h"

#include "MDS.h"
#include "MDCache.h"
#include "MDStore.h"
#include "MDLog.h"
#include "MDCluster.h"
#include "MDBalancer.h"
#include "IdAllocator.h"
#include "AnchorTable.h"
#include "OSDMonitor.h"

#include "include/filepath.h"

#include "common/Logger.h"
#include "common/LogType.h"

#include "messages/MPing.h"
#include "messages/MPingAck.h"
#include "messages/MGenericMessage.h"

#include "messages/MClientMount.h"
#include "messages/MClientMountAck.h"
#include "messages/MClientRequest.h"
#include "messages/MClientReply.h"

#include "messages/MLock.h"
#include "messages/MInodeWriterClosed.h"

#include "messages/MInodeLink.h"

#include "events/EInodeUpdate.h"

#include <errno.h>
#include <fcntl.h>

#include <list>
#include <iostream>
using namespace std;


LogType mds_logtype;

#include "include/config.h"
#undef dout
#define  dout(l)    if (l<=g_conf.debug) cout << "mds" << whoami << " "
#define  dout3(l,mds)    if (l<=g_conf.debug) cout << "mds" << mds->get_nodeid() << " "



ostream& operator<<(ostream& out, MDS& mds)
{
  out << "mds" << mds.get_nodeid() << " ";
}

void C_MDS_RetryMessage::redelegate(MDS *mds, int newmds)
{
  // forward message to new mds
  dout3(5,mds) << "redelegating context " << this << " by forwarding message " << m << " to mds" << newmds << endl;

  mds->messenger->send_message(m,
							   newmds, m->get_dest_port(),
							   MDS_PORT_MAIN);  // mostly meaningless
}



// extern 
//MDS *g_mds;


// cons/des
MDS::MDS(MDCluster *mdc, int whoami, Messenger *m) {
  this->whoami = whoami;
  mdcluster = mdc;

  messenger = m;
  messenger->set_dispatcher(this);

  mdcache = new MDCache(this);
  mdstore = new MDStore(this);
  mdlog = new MDLog(this);
  balancer = new MDBalancer(this);
  anchormgr = new AnchorTable(this);
  osdmonitor = new OSDMonitor(this);


  // <HACK set up OSDCluster from g_conf>
  osdcluster = new OSDCluster();
  OSDGroup osdg;
  osdg.num_osds = g_conf.num_osd;
  for (int i=0; i<osdg.num_osds; i++) osdg.osds.push_back(i);
  osdg.weight = 100;
  osdcluster->add_group(osdg);
  // </HACK>

  filer = new Filer(messenger, osdcluster);

  mdlog->set_max_events(g_conf.mds_log_max_len);

  shutting_down = false;
  shut_down = false;

  stat_ops = 0;
  last_heartbeat = 0;

  // log
  string name;
  name = "mds";
  int w = MSG_ADDR_NUM(whoami);
  if (w >= 1000) name += ('0' + ((w/1000)%10));
  if (w >= 100) name += ('0' + ((w/100)%10));
  if (w >= 10) name += ('0' + ((w/10)%10));
  name += ('0' + ((w/1)%10));

  logger = new Logger(name, (LogType*)&mds_logtype);

  // alloc
  idalloc = new IdAllocator(this);

}

MDS::~MDS() {
  if (mdcache) { delete mdcache; mdcache = NULL; }
  if (mdstore) { delete mdstore; mdstore = NULL; }
  if (mdlog) { delete mdlog; mdlog = NULL; }
  if (balancer) { delete balancer; balancer = NULL; }

  if (logger) { delete logger; logger = 0; }

  if (messenger) { delete messenger; messenger = NULL; }

  if (idalloc) { delete idalloc; idalloc = NULL; }
}


int MDS::init()
{
}


int MDS::shutdown_start()
{
  dout(1) << "shutdown_start" << endl;
  for (int i=0; i<mdcluster->get_num_mds(); i++) {
	if (i == whoami) continue;
	dout(1) << "sending MShutdownStart to mds" << i << endl;
	messenger->send_message(new MGenericMessage(MSG_MDS_SHUTDOWNSTART),
							i, MDS_PORT_MAIN,
							MDS_PORT_MAIN);
  }

  handle_shutdown_start(NULL);
}


void MDS::handle_shutdown_start(Message *m)
{
  dout(1) << " handle_shutdown_start" << endl;

  // set flag
  shutting_down = true;

  mdcache->shutdown_start();
  
  // save anchor table
  anchormgr->save(0);  // FIXME FIXME

  // flush log
  mdlog->set_max_events(0);
  mdlog->trim(NULL);

  if (m) delete m;
}

void MDS::handle_shutdown_finish(Message *m)
{
  int mds = whoami;
  if (m) 
	mds = m->get_source();
						 
  dout(2) << "handle_shutdown_finish from " << mds << endl;
  did_shut_down.insert(mds);
  dout(2) << " shut down so far: " << did_shut_down << endl;
  
  if (did_shut_down.size() == mdcluster->get_num_mds()) {
	// MDS's all shut down!

	// shut down osd's
	for (int i=0; i<g_conf.num_osd; i++) {
	  messenger->send_message(new MGenericMessage(MSG_SHUTDOWN),
							  MSG_ADDR_OSD(i), 0, 0);
	}

	// shut myself down.
	shutting_down = false;

	shutdown_final();
  }

  // done
  delete m;
}



int MDS::shutdown_final()
{
  // shut down cache
  mdcache->shutdown();
  
  // shut down messenger
  messenger->shutdown();

  return 0;
}



mds_load_t MDS::get_load()
{
  timepair_t now = g_clock.gettimepair();
  mds_load_t l;
  if (mdcache->get_root()) {
	l.root_pop = mdcache->get_root()->popularity[MDS_POP_ANYDOM].get(now);
  } else
	l.root_pop = 0;
  l.req_rate = stat_req.get(now);
  l.rd_rate = stat_read.get(now);
  l.wr_rate = stat_write.get(now);
  return l;
}


class C_MDS_Unpause : public Context {
public:
  MDS *mds;
  C_MDS_Unpause(MDS *mds) {
	this->mds = mds;
  }
  void finish(int r) {
	mds->mds_paused = false;
	mds->queue_finished(mds->waiting_for_unpause);
  }
};


void MDS::proc_message(Message *m) 
{
  switch (m->get_type()) {
	// OSD ===============
  case MSG_OSD_READREPLY:
	filer->handle_osd_read_reply((MOSDReadReply*)m);
	return;
  case MSG_OSD_WRITEREPLY:
	filer->handle_osd_write_reply((MOSDWriteReply*)m);
	return;
  case MSG_OSD_OPREPLY:
	filer->handle_osd_op_reply((MOSDOpReply*)m);
	return;

	// MDS
  case MSG_MDS_SHUTDOWNSTART:
	handle_shutdown_start(m);
	return;

  case MSG_MDS_SHUTDOWNFINISH:
	handle_shutdown_finish(m);
	return;

  case MSG_PING:
	handle_ping((MPing*)m);
	return;
  }


  // paused?
  if (mds_paused) {
	dout(3) << "paused" << endl;
	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	return;
  }


  switch (m->get_type()) {
  case MSG_CLIENT_MOUNT:
	handle_client_mount((MClientMount*)m);
	return;
  case MSG_CLIENT_UNMOUNT:
	handle_client_unmount(m);
	return;
  }
 

  // need idalloc
  if (!idalloc->is_open()) {
	dout(3) << "idalloc not open yet" << endl;

	mds_paused = true;
	idalloc->load(new C_MDS_Unpause(this));
	
	// defer
	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	return;
  }


  switch (m->get_type()) {
  case MSG_CLIENT_REQUEST:
	handle_client_request((MClientRequest*)m);
	return;
  }

  dout(1) << " main unknown message " << m->get_type() << endl;
  assert(0);
}


void MDS::dispatch(Message *m)
{
  mds_lock.Lock();

  my_dispatch(m);

  mds_lock.Unlock();
}

void MDS::my_dispatch(Message *m)
{

  switch (m->get_dest_port()) {
	
  case MDS_PORT_STORE:
	mdstore->proc_message(m);
	break;

  case MDS_PORT_ANCHORMGR:
	anchormgr->proc_message(m);
	break;
	
  case MDS_PORT_OSDMON:
	osdmonitor->proc_message(m);
	break;

  case MDS_PORT_CACHE:
	mdcache->proc_message(m);
	break;

	/*
  case MSG_PORT_MDLOG:
	mymds->logger->proc_message(m);
	break;
	*/
	
  case MDS_PORT_BALANCER:
	balancer->proc_message(m);
	break;
	

  case MDS_PORT_MAIN:
  case MDS_PORT_SERVER:
	proc_message(m);
	break;

  default:
	dout(1) << "MDS dispatch unknown message port" << m->get_dest_port() << endl;
	assert(0);
  }


  // HACK FOR NOW
  /*
  static bool did_heartbeat_hack = false;
  if (!shutting_down && !shut_down &&
	  false && 
	  !did_heartbeat_hack) {
	osdmonitor->initiate_heartbeat();
	did_heartbeat_hack = true;
  }
  */
	

  // finish any triggered contexts
  if (finished_queue.size()) {
	dout(7) << "mds has " << finished_queue.size() << " queued contexts" << endl;
	list<Context*> ls;
	ls.splice(ls.begin(), finished_queue);
	assert(finished_queue.empty());
	finish_contexts(ls);
  }

  // balance?
  if (true && whoami == 0 &&
	  stat_ops >= last_heartbeat + g_conf.mds_bal_interval) {
	last_heartbeat = stat_ops;
	balancer->send_heartbeat();
  }

  if (whoami == 0) {
	static bool didit = false;
	
	// 7 to 1
	CInode *in = mdcache->get_inode(1001);
	if (in && in->is_dir() && !didit) {
	  CDir *dir = in->get_or_open_dir(this);
	  if (dir->is_auth()) {
		dout(1) << "FORCING EXPORT" << endl;
		mdcache->export_dir(dir,1);
		didit = true;
	  }
	}
  }

  // shut down?
  if (shutting_down && !shut_down) {
	if (mdcache->shutdown_pass()) {
	  shutting_down = false;
	  shut_down = true;
	  if (whoami) shutdown_final();
	}
  }

}


void MDS::handle_client_mount(MClientMount *m)
{
  // mkfs?  (sorta hack!)
  if (int cmd = m->get_mkfs()) {
	dout(3) << "MKFS flag is set" << endl;
	if (mdcache->get_root()) {
	  dout(3) << "   root inode is already open" << endl;
	} else {
	  dout(3) << "   root inode isn't open yet, inventing a fresh filesystem" << endl;
	  
	  mdcache->open_root(0);
	  CInode *root = mdcache->get_root();
	  assert(root);

	  // force empty root dir
	  CDir *dir = root->dir;
	  dir->mark_complete();
	  dir->mark_dirty();

	  if (cmd == MDS_MKFS_FULL) {
		// wipe osds too
		dout(3) << "wiping osds too" << endl;
		mds_paused = true;
		filer->mkfs(new C_MDS_Unpause(this));
	  	waiting_for_unpause.push_back(new C_MDS_RetryMessage(this, m));
	  	return;
	  } else {
		// fake out idalloc (reset, pretend loaded)
		mds->idalloc->reset();
	  }
	  
	}
  }

  int n = MSG_ADDR_NUM(m->get_source());
  dout(3) << "mount by client" << n << endl;
  mounted_clients.insert(n);

  assert(whoami == 0);  // mds0 mounts/unmounts



  // ack
{'fraction_non_alphanumeric': 0.16408317580340265, 'fraction_numerical': 0.008533621388063732, 'mean_word_length': 3.763570877283252, 'pattern_counts': {'":': 0, '<': 255, '<?xml version=': 0, '>': 149, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5062422', 'n_tokens_mistral': 8104, 'n_tokens_neox': 7658, 'n_words': 1489}