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/ceph_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 using namespace std;
  37
  38 class MDSTableIOContext : public MDSIOContextBase
  39 {
  40   protected:
  41     MDSTable *ida;
  42     MDSRank *get_mds() override {return ida->mds;}
  43   public:
  44     explicit MDSTableIOContext(MDSTable *ida_) : ida(ida_) {
  45       ceph_assert(ida != NULL);
  46     }
  47 };
  48
  49
  50 class C_IO_MT_Save : public MDSTableIOContext {
  51   version_t version;
  52 public:
  53   C_IO_MT_Save(MDSTable *i, version_t v) : MDSTableIOContext(i), version(v) {}
  54   void finish(int r) override {
  55     ida->save_2(r, version);
  56   }
  57   void print(ostream& out) const override {
  58     out << "table_save(" << ida->table_name << ")";
  59   }
  60 };
  61
  62 void MDSTable::save(MDSContext *onfinish, version_t v)
  63 {
  64   if (v > 0 && v <= committing_version) {
  65     dout(10) << "save v " << version << " - already saving "
  66              << committing_version << " >= needed " << v << dendl;
  67     if (onfinish)
  68       waitfor_save[v].push_back(onfinish);
  69     return;
  70   }
  71
  72   dout(10) << "save v " << version << dendl;
  73   ceph_assert(is_active());
  74
  75   bufferlist bl;
  76   encode(version, bl);
  77   encode_state(bl);
  78
  79   committing_version = version;
  80
  81   if (onfinish)
  82     waitfor_save[version].push_back(onfinish);
  83
  84   // write (async)
  85   SnapContext snapc;
  86   object_t oid = get_object_name();
  87   object_locator_t oloc(mds->get_metadata_pool());
  88   mds->objecter->write_full(oid, oloc,
  89                             snapc,
  90                             bl, ceph::real_clock::now(), 0,
  91                             new C_OnFinisher(new C_IO_MT_Save(this, version),
  92                                              mds->finisher));
  93 }
  94
  95 void MDSTable::save_2(int r, version_t v)
  96 {
  97   if (r < 0) {
  98     dout(1) << "save error " << r << " v " << v << dendl;
  99     mds->clog->error() << "failed to store table " << table_name << " object,"
 100                        << " errno " << r;
 101     mds->handle_write_error(r);
 102     return;
 103   }
 104
 105   dout(10) << "save_2 v " << v << dendl;
 106   committed_version = v;
 107
 108   MDSContext::vec ls;
 109   while (!waitfor_save.empty()) {
 110     auto it = waitfor_save.begin();
 111     if (it->first > v) break;
 112     auto& v = it->second;
 113     ls.insert(ls.end(), v.begin(), v.end());
 114     waitfor_save.erase(it);
 115   }
 116   finish_contexts(g_ceph_context, ls, 0);
 117 }
 118
 119
 120 void MDSTable::reset()
 121 {
 122   reset_state();
 123   projected_version = version;
 124   state = STATE_ACTIVE;
 125 }
 126
 127
 128
 129 // -----------------------
 130
 131 class C_IO_MT_Load : public MDSTableIOContext {
 132 public:
 133   Context *onfinish;
 134   bufferlist bl;
 135   C_IO_MT_Load(MDSTable *i, Context *o) : MDSTableIOContext(i), onfinish(o) {}
 136   void finish(int r) override {
 137     ida->load_2(r, bl, onfinish);
 138   }
 139   void print(ostream& out) const override {
 140     out << "table_load(" << ida->table_name << ")";
 141   }
 142 };
 143
 144 object_t MDSTable::get_object_name() const
 145 {
 146   char n[50];
 147   if (per_mds)
 148     snprintf(n, sizeof(n), "mds%d_%s", int(rank), table_name.c_str());
 149   else
 150     snprintf(n, sizeof(n), "mds_%s", table_name.c_str());
 151   return object_t(n);
 152 }
 153
 154 void MDSTable::load(MDSContext *onfinish)
 155 {
 156   dout(10) << "load" << dendl;
 157
 158   ceph_assert(is_undef());
 159   state = STATE_OPENING;
 160
 161   C_IO_MT_Load *c = new C_IO_MT_Load(this, onfinish);
 162   object_t oid = get_object_name();
 163   object_locator_t oloc(mds->get_metadata_pool());
 164   mds->objecter->read_full(oid, oloc, CEPH_NOSNAP, &c->bl, 0,
 165                            new C_OnFinisher(c, mds->finisher));
 166 }
 167
 168 void MDSTable::load_2(int r, bufferlist& bl, Context *onfinish)
 169 {
 170   ceph_assert(is_opening());
 171   state = STATE_ACTIVE;
 172   if (r == -CEPHFS_EBLOCKLISTED) {
 173     mds->respawn();
 174     return;
 175   }
 176   if (r < 0) {
 177     derr << "load_2 could not read table: " << r << dendl;
 178     mds->clog->error() << "error reading table object '" << get_object_name()
 179                        << "' " << r << " (" << cpp_strerror(r) << ")";
 180     mds->damaged();
 181     ceph_assert(r >= 0);  // Should be unreachable because damaged() calls respawn()
 182   }
 183
 184   dout(10) << "load_2 got " << bl.length() << " bytes" << dendl;
 185   auto p = bl.cbegin();
 186
 187   try {
 188     decode(version, p);
 189     projected_version = committed_version = version;
 190     dout(10) << "load_2 loaded v" << version << dendl;
 191     decode_state(p);
 192   } catch (buffer::error &e) {
 193     mds->clog->error() << "error decoding table object '" << get_object_name()
 194                        << "': " << e.what();
 195     mds->damaged();
 196     ceph_assert(r >= 0);  // Should be unreachable because damaged() calls respawn()
 197   }
 198
 199   if (onfinish) {
 200     onfinish->complete(0);
 201   }
 202 }