stuff

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@828 29311d96-e01e-0410-9327-a35deaab8ce9

diff --git a/ceph/common/Logger.cc b/ceph/common/Logger.cc
index e2c7111f86001dc478e658df6123b1054f8024d9..5523a215b8ee81a08875b958c9a092e2caef9b89 100644 (file)
--- a/ceph/common/Logger.cc
+++ b/ceph/common/Logger.cc
@@ -142,6 +142,8 @@ void Logger::flush(bool force)
   }
 
   utime_t fromstart = g_clock.recent_now();
+  if (fromstart < start) 
+       start = fromstart;
   fromstart -= start;
 
   while (force ||

diff --git a/ceph/common/Logger.h b/ceph/common/Logger.h
index 9288a521fff6057cf97b168f8b1aab21f3a9ab37..e1ae8bfb3b28e5fc86d88adaa8f4084f58af7c80 100644 (file)
--- a/ceph/common/Logger.h
+++ b/ceph/common/Logger.h
@@ -58,6 +58,7 @@ class Logger {
   Logger(string fn, LogType *type);
   ~Logger();
 
+  void set_start(const utime_t& a) { start = a; }
   utime_t& get_start() { return start; }
 
   long inc(const char *s, long v = 1);
@@ -66,7 +67,7 @@ class Logger {
 
   double fset(const char *s, double v);
   double finc(const char *s, double v);
-  
+
   void flush(bool force = false);
 };
 

diff --git a/ceph/config.cc b/ceph/config.cc
index 63edcfa9dd55d6cf0ae7f8d9e15672b5c1115ab3..cc342a300f415fa55da4e9569f6a6da7d8d22ab9 100644 (file)
--- a/ceph/config.cc
+++ b/ceph/config.cc
@@ -104,7 +104,7 @@ md_config_t g_conf = {
 
   // --- mon ---
   mon_tick_interval: 5,
-  mon_osd_down_out_interval: 20,  // seconds
+  mon_osd_down_out_interval: 15,  // seconds
 
   // --- client ---
   client_cache_size: 300,
@@ -188,6 +188,7 @@ md_config_t g_conf = {
   osd_age_time: 0,
   osd_heartbeat_interval: 10,
   osd_replay_window: 15,
+  osd_max_pull: 2,
   
   // --- fakestore ---
   fakestore_fake_sync: 2,    // 2 seconds
@@ -199,8 +200,8 @@ md_config_t g_conf = {
 
   // --- ebofs ---
   ebofs: 1,
-  ebofs_commit_ms:      10000,      // 0 = no forced commit timeout (for debugging/tracing)
-  ebofs_idle_commit_ms: 100,//100,        // 0 = no idle detection.  use this -or- bdev_idle_kick_after_ms
+  ebofs_commit_ms:      2000,       // 0 = no forced commit timeout (for debugging/tracing)
+  ebofs_idle_commit_ms: 100,        // 0 = no idle detection.  use this -or- bdev_idle_kick_after_ms
   ebofs_oc_size:        10000,      // onode cache
   ebofs_cc_size:        10000,      // cnode cache
   ebofs_bc_size:        (350 *256), // 4k blocks, *256 for MB
@@ -589,6 +590,8 @@ void parse_config_options(vector<char*>& args)
          g_conf.osd_max_rep = atoi(args[++i]);
        else if (strcmp(args[i], "--osd_maxthreads") == 0) 
          g_conf.osd_maxthreads = atoi(args[++i]);
+       else if (strcmp(args[i], "--osd_max_pull") == 0) 
+         g_conf.osd_max_pull = atoi(args[++i]);
 
 
        else if (strcmp(args[i], "--bdev_lock") == 0) 

diff --git a/ceph/config.h b/ceph/config.h
index 159e8310550f8c7c5935ff030dc5d8281d6debe3..e4022f2d07212a2bdc0ad16b8e9bef630b91d28a 100644 (file)
--- a/ceph/config.h
+++ b/ceph/config.h
@@ -160,6 +160,7 @@ struct md_config_t {
   int   osd_age_time;
   int   osd_heartbeat_interval;
   int   osd_replay_window;
+  int   osd_max_pull;
 
   int   fakestore_fake_sync;
   bool  fakestore_fsync;

diff --git a/ceph/crush/Bucket.h b/ceph/crush/Bucket.h
index 833d34818851cedb03cdb4266973203a08c9f74a..28bdd15b6c6b165b45591d46d1ec4315e852ee4b 100644 (file)
--- a/ceph/crush/Bucket.h
+++ b/ceph/crush/Bucket.h
@@ -169,6 +169,7 @@ namespace crush {
          if (items.empty())
                item_weight = w;
          items.push_back(item);
+         weight += item_weight;
          make_primes();
        }
 

diff --git a/ceph/jobs/rados/map_dist b/ceph/jobs/rados/map_dist
index 09cc41b14adec8e45febec3944d497bcedaa51ea..5df679708d42a7d821b4c45c3adfedb2d07911e2 100644 (file)
--- a/ceph/jobs/rados/map_dist
+++ b/ceph/jobs/rados/map_dist
@@ -4,30 +4,25 @@
 {
        'sleep' => 3,
 
+       'osdbits' => [6,8,10],
+       'pgperbits' => [4,6,8],
+
        'nummds' => 1,
-       'numosd' => [16,
-                                32,
-                                64,
-                                128,
-#                               256,
-#                               1024,
-#                               4096,
-#                               16384,
-#                               65536,
-#                               256*256
-                                ],
+
+       '_dep' => [ 'numosd' => '1 << $osdbits',
+                               'pg_bits' => '$pgperbits + $osdbits',
+                               'n' => '10 + $numosd / 32'],
        'numclient' => 0,
-       'n' => 6,
 
-       'fs' => ['ebofs'],
+       'fs' => 'ebofs',
 
        'start' => 30,
        'end' => 180,
        'kill_after' => 180,
        
-       'osd_pg_bits' => [10, 14, 18],
+       'osd_pg_bits' => [18],
 
-       'custom' => '--bdev_lock 0 --ms_stripe_osds',
+       'custom' => '--bdev_lock 0 --ms_stripe_osds --fake_osdmap_updates 1 --osd_maxthreads 0',
        #'custom' => '--tcp_skip_rank0',
 
        'comb' => {

diff --git a/ceph/mon/Monitor.cc b/ceph/mon/Monitor.cc
index 476c3d861426dd2790b00554d9f1851e678792a6..f13ded62d5b680e742e13d7159fdb3ed4d882321 100644 (file)
--- a/ceph/mon/Monitor.cc
+++ b/ceph/mon/Monitor.cc
@@ -104,6 +104,7 @@ void Monitor::init()
 
   // <HACK set up OSDMap from g_conf>
   osdmap = new OSDMap();
+  osdmap->ctime = g_clock.now();
   osdmap->set_pg_bits(g_conf.osd_pg_bits);
 
   // start at epoch 0 until all osds boot
@@ -113,17 +114,63 @@ void Monitor::init()
 
   //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.num_osd >= 12) {
+       int ndom = g_conf.osd_max_rep;
+       UniformBucket *domain[ndom];
+       int domid[ndom];
+       for (int i=0; i<ndom; i++) {
+         domain[i] = new UniformBucket(1, 0);
+         domid[i] = osdmap->crush.add_bucket(domain[i]);
+       }
+
+       // add osds
+       int nper = ((g_conf.num_osd - 1) / ndom) + 1;
+       cerr << ndom << " failure domains, " << nper << " osds each" << endl;
+       int i = 0;
+       for (int dom=0; dom<ndom; dom++) {
+         for (int j=0; j<nper; j++) {
+               osdmap->osds.insert(i);
+               domain[dom]->add_item(i, 1.0);
+               //cerr << "osd" << i << " in domain " << dom << endl;
+               i++;
+               if (i == g_conf.num_osd) break;
+         }
+       }
+
+       // root
+       Bucket *root = new ListBucket(2);
+       for (int i=0; i<ndom; i++) {
+         //cerr << "dom " << i << " w " << domain[i]->get_weight() << endl;
+         root->add_item(domid[i], domain[i]->get_weight());
+       }
+       int nroot = osdmap->crush.add_bucket(root);     
+
+       // rules
+       for (int i=1; i<=ndom; i++) {
+         osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_TAKE, nroot));
+         osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_CHOOSE, i, 1));
+         osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_CHOOSE, 1, 0));      
+         osdmap->crush.rules[i].steps.push_back(RuleStep(CRUSH_RULE_EMIT));
+       }
+
+       // test
+       vector<int> out;
+       osdmap->pg_to_osds(0x40200000110ULL, out);
+
+  } else {
+       // one bucket
+       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.0);
+       }
+       
+       for (int i=1; i<=g_conf.osd_max_rep; 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) {
@@ -371,6 +418,7 @@ void Monitor::accept_pending()
   dout(-10) << "accept_pending " << osdmap->get_epoch() << " -> " << pending.epoch << endl;
 
   // accept pending into a new map!
+  pending.ctime = g_clock.now();
   pending.encode( inc_maps[ pending.epoch ] );
   
   // advance!
@@ -515,7 +563,12 @@ void Monitor::tick()
                 i++) {
          pending_out.erase(*i);
          pending.new_out.push_back( *i );
+       }
+       if (!mark_out.empty()) {
          accept_pending();
+
+         // hrmpf.  bcast map for now.  FIXME FIXME.
+         bcast_latest_osd_map_osd();
        }
        
        // next!

diff --git a/ceph/osd/OSD.cc b/ceph/osd/OSD.cc
index 6cd5eb34671c268af69a79d590519887d67de563..c50a99035515209e908f8af828bc1c27acff9300 100644 (file)
--- a/ceph/osd/OSD.cc
+++ b/ceph/osd/OSD.cc
@@ -113,8 +113,8 @@ OSD::OSD(int id, Messenger *m, char *dev)
   boot_epoch = 0;
 
   last_tid = 0;
+  num_pulling = 0;
 
-  max_recovery_ops = 5;
 
   pending_ops = 0;
   waiting_for_no_ops = false;
@@ -248,7 +248,6 @@ int OSD::init()
        osd_logtype.add_inc("r_wr");
        osd_logtype.add_inc("r_wrb");
        
-       osd_logtype.add_inc("rlsum");
        osd_logtype.add_inc("rlnum");
 
        osd_logtype.add_set("numpg");
@@ -1050,6 +1049,9 @@ void OSD::advance_map(ObjectStore::Transaction& t)
        dout(1) << "mkfs" << endl;
        assert(osdmap->get_epoch() == 1);
 
+       //cerr << "osdmap " << osdmap->get_ctime() << " logger start " << logger->get_start() << endl;
+       logger->set_start( osdmap->get_ctime() );
+
        ps_t maxps = 1LL << osdmap->get_pg_bits();
        
        // create PGs
@@ -1130,10 +1132,7 @@ void OSD::advance_map(ObjectStore::Transaction& t)
          // did primary change?
          if (oldprimary != primary) {
                pg->info.same_primary_since = osdmap->get_epoch();
-
-               // forget about where missing items are, or anything we're pulling
-               pg->missing.loc.clear();
-               pg->objects_pulling.clear();
+               pg->cancel_recovery();
          }
          
          if (role != oldrole) {
@@ -1984,6 +1983,7 @@ void OSD::pull(PG *pg, object_t oid, eversion_t v)
   
   // take note
   assert(pg->objects_pulling.count(oid) == 0);
+  num_pulling++;
   pg->objects_pulling[oid] = v;
 }
 
@@ -1994,8 +1994,6 @@ void OSD::pull(PG *pg, object_t oid, eversion_t v)
  */
 void OSD::op_rep_pull(MOSDOp *op, PG *pg)
 {
-  long got = 0;
-  
   const object_t oid = op->get_oid();
 
   dout(7) << "rep_pull on " << hex << oid << dec << " v >= " << op->get_version() << endl;
@@ -2028,6 +2026,9 @@ void OSD::op_rep_pull(MOSDOp *op, PG *pg)
                  << " in " << *pg
                  << endl;
   assert(v >= op->get_version());
+
+  logger->inc("r_pull");
+  logger->inc("r_pullb", bl.length());
   
   // reply
   MOSDOpReply *reply = new MOSDOpReply(op, 0, osdmap->get_epoch(), true); 
@@ -2041,9 +2042,6 @@ void OSD::op_rep_pull(MOSDOp *op, PG *pg)
   messenger->send_message(reply, op->get_asker());
   
   delete op;
-
-  logger->inc("r_pull");
-  logger->inc("r_pullb", got);
 }
 
 
@@ -2086,6 +2084,7 @@ void OSD::op_rep_pull_reply(MOSDOpReply *op)
   t.collection_add(pgid, oid);
 
   // close out pull op.
+  num_pulling--;
   pg->objects_pulling.erase(oid);
   pg->missing.got(oid, v);
 
@@ -2237,6 +2236,9 @@ void OSD::op_rep_modify(MOSDOp *op, PG *pg)
        prepare_op_transaction(t, op, nv, pg);
        oncommit = new C_OSD_RepModifyCommit(this, op,
                                                                                 pg->info.last_complete);
+
+       logger->inc("r_wr");
+       logger->inc("r_wrb", op->get_length());
   }
 
   // go

diff --git a/ceph/osd/OSD.h b/ceph/osd/OSD.h
index 7bff2e909d45245892ccf96fba42759def0e27a6..32959d74046164ee85e7d31d4909cf526b1b60ec 100644 (file)
--- a/ceph/osd/OSD.h
+++ b/ceph/osd/OSD.h
@@ -99,8 +99,6 @@ public:
 
   class Logger      *logger;
 
-  int max_recovery_ops;
-
   // local store
   char dev_path[100];
   class ObjectStore *store;
@@ -218,6 +216,7 @@ public:
 
 
   tid_t               last_tid;
+  int                 num_pulling;
 
   hash_map<pg_t, list<Message*> >        waiting_for_pg;
 

diff --git a/ceph/osd/OSDMap.h b/ceph/osd/OSDMap.h
index 12c4c5196391f3ba8d9c49ba3cb7ea2448ecfb33..fa5b19ec51c8189a87a8741f490862a193ba3876 100644 (file)
--- a/ceph/osd/OSDMap.h
+++ b/ceph/osd/OSDMap.h
@@ -120,6 +120,8 @@ private:
   int get_pg_bits() const { return pg_bits; }
   void set_pg_bits(int b) { pg_bits = b; }
 
+  const utime_t& get_ctime() const { return ctime; }
+
   bool is_mkfs() const { return epoch == 1; }
   //void set_mkfs() { assert(epoch == 1); }
 

diff --git a/ceph/osd/PG.cc b/ceph/osd/PG.cc
index 22075eb2ebd9f301ec0d61d22e1e992a6c25672c..7dad430a13df6b8a97aea6be5789e988935331c8 100644 (file)
--- a/ceph/osd/PG.cc
+++ b/ceph/osd/PG.cc
@@ -74,11 +74,11 @@ void PG::merge_log(Log &olog, Missing &omissing, int fromosd)
   dout(10) << "merge_log " << olog << " from osd" << fromosd
                   << " into " << log << endl;
 
-  cout << "log" << endl;
+  /*cout << "log" << endl;
   log.print(cout);
   cout << "olog" << endl;
   olog.print(cout);
-
+  */
   if (log.empty() ||
          (olog.bottom > log.top && olog.backlog)) { // e.g. log=(0,20] olog=(40,50]+backlog) 
        // i'm missing everything after old log.top.
@@ -183,7 +183,7 @@ void PG::merge_log(Log &olog, Missing &omissing, int fromosd)
   }
   
   dout(10) << "merge_log result " << log << " " << missing << endl;
-  log.print(cout);
+  //log.print(cout);
 
   // found items?
   for (map<object_t,eversion_t>::iterator p = missing.missing.begin();
@@ -783,13 +783,30 @@ void PG::clean_up_local(ObjectStore::Transaction& t)
 }
 
 
+
+void PG::cancel_recovery()
+{
+  // forget about where missing items are, or anything we're pulling
+  missing.loc.clear();
+  osd->num_pulling -= objects_pulling.size();
+  objects_pulling.clear();
+}
+
 /**
  * do one recovery op.
  * return true if done, false if nothing left to do.
  */
 bool PG::do_recovery()
 {
-  dout(10) << "do_recovery" << endl;
+  dout(-10) << "do_recovery pulling " << objects_pulling.size() << " in pg, "
+                  << osd->num_pulling << "/" << g_conf.osd_max_pull << " total"
+                  << endl;
+
+  // can we slow down on this PG?
+  if (osd->num_pulling >= g_conf.osd_max_pull && !objects_pulling.empty()) {
+       dout(-10) << "do_recovery already pulling max, waiting" << endl;
+       return true;
+  }
 
   // look at log!
   Log::Entry *latest = 0;

diff --git a/ceph/osd/PG.h b/ceph/osd/PG.h
index 94fbeeefd2645bc55b8317c8dc11b7a39bab6042..b7dcff9f46ffeacdbf5e1bce863f7f835ab7b6bf 100644 (file)
--- a/ceph/osd/PG.h
+++ b/ceph/osd/PG.h
@@ -458,6 +458,7 @@ public:
 
   void activate(ObjectStore::Transaction& t);
 
+  void cancel_recovery();
   bool do_recovery();
 
   void clean_replicas();