src/ceph/src/mds/MDSTable.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include "MDSTable.h"
  16
  17 #include "MDSRank.h"
  18 #include "MDLog.h"
  19
  20 #include "osdc/Filer.h"
  21
  22 #include "include/types.h"
  23
  24 #include "common/config.h"
  25 #include "common/errno.h"
  26 #include "common/Finisher.h"
  27
  28 #include "include/assert.h"
  29
  30
  31 #define dout_context g_ceph_context
  32 #define dout_subsys ceph_subsys_mds
  33 #undef dout_prefix
  34 #define dout_prefix *_dout << "mds." << rank << "." << table_name << ": "
  35
  36
  37 class MDSTableIOContext : public MDSIOContextBase
  38 {
  39   protected:
  40     MDSTable *ida;
  41     MDSRank *get_mds() override {return ida->mds;}
  42   public:
  43     explicit MDSTableIOContext(MDSTable *ida_) : ida(ida_) {
  44       assert(ida != NULL);
  45     }
  46 };
  47
  48
  49 class C_IO_MT_Save : public MDSTableIOContext {
  50   version_t version;
  51 public:
  52   C_IO_MT_Save(MDSTable *i, version_t v) : MDSTableIOContext(i), version(v) {}
  53   void finish(int r) override {
  54     ida->save_2(r, version);
  55   }
  56 };
  57
  58 void MDSTable::save(MDSInternalContextBase *onfinish, version_t v)
  59 {
  60   if (v > 0 && v <= committing_version) {
  61     dout(10) << "save v " << version << " - already saving "
  62              << committing_version << " >= needed " << v << dendl;
  63     waitfor_save[v].push_back(onfinish);
  64     return;
  65   }
  66
  67   dout(10) << "save v " << version << dendl;
  68   assert(is_active());
  69
  70   bufferlist bl;
  71   ::encode(version, bl);
  72   encode_state(bl);
  73
  74   committing_version = version;
  75
  76   if (onfinish)
  77     waitfor_save[version].push_back(onfinish);
  78
  79   // write (async)
  80   SnapContext snapc;
  81   object_t oid = get_object_name();
  82   object_locator_t oloc(mds->mdsmap->get_metadata_pool());
  83   mds->objecter->write_full(oid, oloc,
  84                             snapc,
  85                             bl, ceph::real_clock::now(), 0,
  86                             new C_OnFinisher(new C_IO_MT_Save(this, version),
  87                                              mds->finisher));
  88 }
  89
  90 void MDSTable::save_2(int r, version_t v)
  91 {
  92   if (r < 0) {
  93     dout(1) << "save error " << r << " v " << v << dendl;
  94     mds->clog->error() << "failed to store table " << table_name << " object,"
  95                        << " errno " << r;
  96     mds->handle_write_error(r);
  97     return;
  98   }
  99
 100   dout(10) << "save_2 v " << v << dendl;
 101   committed_version = v;
 102
 103   list<MDSInternalContextBase*> ls;
 104   while (!waitfor_save.empty()) {
 105     if (waitfor_save.begin()->first > v) break;
 106     ls.splice(ls.end(), waitfor_save.begin()->second);
 107     waitfor_save.erase(waitfor_save.begin());
 108   }
 109   finish_contexts(g_ceph_context, ls,0);
 110 }
 111
 112
 113 void MDSTable::reset()
 114 {
 115   reset_state();
 116   state = STATE_ACTIVE;
 117 }
 118
 119
 120
 121 // -----------------------
 122
 123 class C_IO_MT_Load : public MDSTableIOContext {
 124 public:
 125   Context *onfinish;
 126   bufferlist bl;
 127   C_IO_MT_Load(MDSTable *i, Context *o) : MDSTableIOContext(i), onfinish(o) {}
 128   void finish(int r) override {
 129     ida->load_2(r, bl, onfinish);
 130   }
 131 };
 132
 133 object_t MDSTable::get_object_name() const
 134 {
 135   char n[50];
 136   if (per_mds)
 137     snprintf(n, sizeof(n), "mds%d_%s", int(mds->get_nodeid()), table_name);
 138   else
 139     snprintf(n, sizeof(n), "mds_%s", table_name);
 140   return object_t(n);
 141 }
 142
 143 void MDSTable::load(MDSInternalContextBase *onfinish)
 144 {
 145   dout(10) << "load" << dendl;
 146
 147   assert(is_undef());
 148   state = STATE_OPENING;
 149
 150   C_IO_MT_Load *c = new C_IO_MT_Load(this, onfinish);
 151   object_t oid = get_object_name();
 152   object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 153   mds->objecter->read_full(oid, oloc, CEPH_NOSNAP, &c->bl, 0,
 154                            new C_OnFinisher(c, mds->finisher));
 155 }
 156
 157 void MDSTable::load_2(int r, bufferlist& bl, Context *onfinish)
 158 {
 159   assert(is_opening());
 160   state = STATE_ACTIVE;
 161   if (r == -EBLACKLISTED) {
 162     mds->respawn();
 163     return;
 164   }
 165   if (r < 0) {
 166     derr << "load_2 could not read table: " << r << dendl;
 167     mds->clog->error() << "error reading table object '" << get_object_name()
 168                        << "' " << r << " (" << cpp_strerror(r) << ")";
 169     mds->damaged();
 170     assert(r >= 0);  // Should be unreachable because damaged() calls respawn()
 171   }
 172
 173   dout(10) << "load_2 got " << bl.length() << " bytes" << dendl;
 174   bufferlist::iterator p = bl.begin();
 175
 176   try {
 177     ::decode(version, p);
 178     projected_version = committed_version = version;
 179     dout(10) << "load_2 loaded v" << version << dendl;
 180     decode_state(p);
 181   } catch (buffer::error &e) {
 182     mds->clog->error() << "error decoding table object '" << get_object_name()
 183                        << "': " << e.what();
 184     mds->damaged();
 185     assert(r >= 0);  // Should be unreachable because damaged() calls respawn()
 186   }
 187
 188   if (onfinish) {
 189     onfinish->complete(0);
 190   }
 191 }