<commit_before>// -*- mode:C++; tab-width:4; c-basic-offset:2; indent-tabs-mode:t -*- 
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 <NAME> <<EMAIL>>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software 
 * Foundation.  See file COPYING.
 * 
 */

// TODO: missing run() method, which creates the two main timers, refreshTimer and readTimer

#include "Monitor.h"

#include "osd/OSDMap.h"

#include "msg/Message.h"
#include "msg/Messenger.h"

#include "messages/MPing.h"
#include "messages/MPingAck.h"
#include "messages/MOSDFailure.h"
#include "messages/MOSDMap.h"
#include "messages/MOSDGetMap.h"
#include "messages/MOSDBoot.h"
#include "messages/MOSDIn.h"
#include "messages/MOSDOut.h"

#include "common/Timer.h"
#include "common/Clock.h"

#include "config.h"
#undef dout
#define  dout(l) if (l<=g_conf.debug || l<=g_conf.debug_mon) cout << "mon" << whoami << " e" << (osdmap ? osdmap->get_epoch():0) << " "
#define  derr(l) if (l<=g_conf.debug || l<=g_conf.debug_mon) cerr << "mon" << whoami << " e" << (osdmap ? osdmap->get_epoch():0) << " "


class C_Mon_Tick : public Context {
  Monitor *mon;
public:
  C_Mon_Tick(Monitor *m) : mon(m) {}
  void finish(int r) {
	mon->tick();
  }
};

class C_Mon_FakeOSDFailure : public Context {
  Monitor *mon;
  int osd;
  bool down;
public:
  C_Mon_FakeOSDFailure(Monitor *m, int o, bool d) : mon(m), osd(o), down(d) {}
  void finish(int r) {
	mon->fake_osd_failure(osd,down);
  }
};


void Monitor::fake_osdmap_update()
{
  dout(1) << "fake_osdmap_update" << endl;
  accept_pending();

  // tell a random osd
  send_incremental_map(osdmap->get_epoch()-1,                    // ick! FIXME
					   MSG_ADDR_OSD(rand() % g_conf.num_osd));
}


void Monitor::fake_reorg() 
{
  int r = rand() % g_conf.num_osd;
  
  if (osdmap->is_out(r)) {
	dout(1) << "fake_reorg marking osd" << r << " in" << endl;
	pending.new_in.push_back(r);
  } else {
	dout(1) << "fake_reorg marking osd" << r << " out" << endl;
	pending.new_out.push_back(r);
  }

  accept_pending();
  
  // tell him!
  send_incremental_map(osdmap->get_epoch()-1, MSG_ADDR_OSD(r));
  
  // do it again?
  /*
  if (g_conf.num_osd - d > 4 &&
	  g_conf.num_osd - d > g_conf.num_osd/2)
	g_timer.add_event_after(g_conf.fake_osdmap_expand,
							new C_Mon_Faker(this));
  */
}


void Monitor::init()
{
  dout(1) << "init" << endl;


  // <HACK set up OSDMap from g_conf>
  osdmap = new OSDMap();
  osdmap->set_pg_bits(g_conf.osd_pg_bits);

  // start at epoch 0 until all osds boot
  //osdmap->inc_epoch();  // = 1
  //assert(osdmap->get_epoch() == 1);


  //if (g_conf.mkfs) osdmap->set_mkfs();

  Bucket *b = new UniformBucket(1, 0);
  int root = osdmap->crush.add_bucket(b);
  for (int i=0; i<g_conf.num_osd; i++) {
	osdmap->osds.insert(i);
	b->add_item(i, 1);
  }
  
  for (int i=1; i<5; i++) {
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_TAKE, root));
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_CHOOSE, i, 0));
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_EMIT));
  }

  if (g_conf.mds_local_osd) {
	// add mds osds, but don't put them in the crush mapping func
	for (int i=0; i<g_conf.num_mds; i++) 
	  osdmap->osds.insert(i+10000);
  }

  //osdmap->encode(maps[osdmap->get_epoch()]); // 1
  //pending.epoch = osdmap->get_epoch()+1;     // 2
  // </HACK>


  
  if (whoami == 0) {
	// fake osd failures
	for (map<int,float>::iterator i = g_fake_osd_down.begin();
		 i != g_fake_osd_down.end();
		 i++) {
	  dout(0) << "will fake osd" << i->first << " DOWN after " << i->second << endl;
	  g_timer.add_event_after(i->second, new C_Mon_FakeOSDFailure(this, i->first, 1));
	}
	for (map<int,float>::iterator i = g_fake_osd_out.begin();
		 i != g_fake_osd_out.end();
		 i++) {
	  dout(0) << "will fake osd" << i->first << " OUT after " << i->second << endl;
	  g_timer.add_event_after(i->second, new C_Mon_FakeOSDFailure(this, i->first, 0));
	}
  }

  
  // i'm ready!
  messenger->set_dispatcher(this);
  
  // start ticker
  g_timer.add_event_after(g_conf.mon_tick_interval, new C_Mon_Tick(this));
}


void Monitor::dispatch(Message *m)
{
  lock.Lock();
  {
	switch (m->get_type()) {
	case MSG_FAILURE:
	  handle_osd_failure((MOSDFailure*)m);
	  break;
	  
	case MSG_PING_ACK:
	  handle_ping_ack((MPingAck*)m);
	  break;
	  
	case MSG_OSD_GETMAP:
	  handle_osd_getmap((MOSDGetMap*)m);
	  break;

	case MSG_OSD_BOOT:
	  handle_osd_boot((MOSDBoot*)m);<commit_msg>*** empty log message ***<commit_after>// -*- mode:C++; tab-width:4; c-basic-offset:2; indent-tabs-mode:t -*- 
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 <NAME> <<EMAIL>>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software 
 * Foundation.  See file COPYING.
 * 
 */

// TODO: missing run() method, which creates the two main timers, refreshTimer and readTimer

#include "Monitor.h"

#include "osd/OSDMap.h"

#include "msg/Message.h"
#include "msg/Messenger.h"

#include "messages/MPing.h"
#include "messages/MPingAck.h"
#include "messages/MOSDFailure.h"
#include "messages/MOSDMap.h"
#include "messages/MOSDGetMap.h"
#include "messages/MOSDBoot.h"
#include "messages/MOSDIn.h"
#include "messages/MOSDOut.h"

#include "common/Timer.h"
#include "common/Clock.h"

#include "config.h"
#undef dout
#define  dout(l) if (l<=g_conf.debug || l<=g_conf.debug_mon) cout << "mon" << whoami << " e" << (osdmap ? osdmap->get_epoch():0) << " "
#define  derr(l) if (l<=g_conf.debug || l<=g_conf.debug_mon) cerr << "mon" << whoami << " e" << (osdmap ? osdmap->get_epoch():0) << " "


class C_Mon_Tick : public Context {
  Monitor *mon;
public:
  C_Mon_Tick(Monitor *m) : mon(m) {}
  void finish(int r) {
	mon->tick();
  }
};

class C_Mon_FakeOSDFailure : public Context {
  Monitor *mon;
  int osd;
  bool down;
public:
  C_Mon_FakeOSDFailure(Monitor *m, int o, bool d) : mon(m), osd(o), down(d) {}
  void finish(int r) {
	mon->fake_osd_failure(osd,down);
  }
};


void Monitor::fake_osdmap_update()
{
  dout(1) << "fake_osdmap_update" << endl;
  accept_pending();

  // tell a random osd
  send_incremental_map(osdmap->get_epoch()-1,                    // ick! FIXME
					   MSG_ADDR_OSD(rand() % g_conf.num_osd));
}


void Monitor::fake_reorg() 
{
  int r = rand() % g_conf.num_osd;
  
  if (osdmap->is_out(r)) {
	dout(1) << "fake_reorg marking osd" << r << " in" << endl;
	pending.new_in.push_back(r);
  } else {
	dout(1) << "fake_reorg marking osd" << r << " out" << endl;
	pending.new_out.push_back(r);
  }

  accept_pending();
  
  // tell him!
  send_incremental_map(osdmap->get_epoch()-1, MSG_ADDR_OSD(r));
  
  // do it again?
  /*
  if (g_conf.num_osd - d > 4 &&
	  g_conf.num_osd - d > g_conf.num_osd/2)
	g_timer.add_event_after(g_conf.fake_osdmap_expand,
							new C_Mon_Faker(this));
  */
}


void Monitor::init()
{
  dout(1) << "init" << endl;


  // <HACK set up OSDMap from g_conf>
  osdmap = new OSDMap();
  osdmap->set_pg_bits(g_conf.osd_pg_bits);

  // start at epoch 0 until all osds boot
  //osdmap->inc_epoch();  // = 1
  //assert(osdmap->get_epoch() == 1);


  //if (g_conf.mkfs) osdmap->set_mkfs();

  Bucket *b = new UniformBucket(1, 0);
  int root = osdmap->crush.add_bucket(b);
  for (int i=0; i<g_conf.num_osd; i++) {
	osdmap->osds.insert(i);
	b->add_item(i, 1);
  }
  
  for (int i=1; i<5; i++) {
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_TAKE, root));
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_CHOOSE, i, 0));
	osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_EMIT));
  }

  if (g_conf.mds_local_osd) {
	// add mds osds, but don't put them in the crush mapping func
	for (int i=0; i<g_conf.num_mds; i++) 
	  osdmap->osds.insert(i+10000);
  }

  //osdmap->encode(maps[osdmap->get_epoch()]); // 1
  //pending.epoch = osdmap->get_epoch()+1;     // 2
  // </HACK>


  
  if (whoami == 0) {
	// fake osd failures
	for (map<int,float>::iterator i = g_fake_osd_down.begin();
		 i != g_fake_osd_down.end();
		 i++) {
	  dout(0) << "will fake osd" << i->first << " DOWN after " << i->second << endl;
	  g_timer.add_event_after(i->second, new C_Mon_FakeOSDFailure(this, i->first, 1));
	}
	for (map<int,float>::iterator i = g_fake_osd_out.begin();
		 i != g_fake_osd_out.end();
		 i++) {
	  dout(0) << "will fake osd" << i->first << " OUT after " << i->second << endl;
	  g_timer.add_event_after(i->second, new C_Mon_FakeOSDFailure(this, i->first, 0));
	}
  }

  
  // i'm ready!
  messenger->set_dispatcher(this);
  
  // start ticker
  g_timer.add_event_after(g_conf.mon_tick_interval, new C_Mon_Tick(this));
}


void Monitor::dispatch(Message *m)
{
  lock.Lock();
  {
	switch (m->get_type()) {
	case MSG_OSD_FAILURE:
	  handle_osd_failure((MOSDFailure*)m);
	  break;
	  
	case MSG_PING_ACK:
	  handle_ping_ack((MPingAck*)m);
	  break;
	  
	case MSG_OSD_GETMAP:
	  handle_osd_getmap((MOSDGetMap*)m);
	  break;

	case MSG_OSD_BOOT:
	  handle_osd_boot((MOSDBoot*)m);
{'fraction_non_alphanumeric': 0.17666374781085814, 'fraction_numerical': 0.010070052539404553, 'mean_word_length': 3.6880451513596717, 'pattern_counts': {'":': 0, '<': 159, '<?xml version=': 0, '>': 83, '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': '2927704', 'n_tokens_mistral': 4022, 'n_tokens_neox': 3841, 'n_words': 745}