ceph/src/os/ObjectStore.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 #include <ctype.h>
  15 #include <sstream>
  16 #include "include/memory.h"
  17 #include "ObjectStore.h"
  18 #include "common/Formatter.h"
  19 #include "common/safe_io.h"
  20
  21 #include "filestore/FileStore.h"
  22 #include "memstore/MemStore.h"
  23 #if defined(HAVE_LIBAIO)
  24 #include "bluestore/BlueStore.h"
  25 #endif
  26 #include "kstore/KStore.h"
  27
  28 void decode_str_str_map_to_bl(bufferlist::iterator& p,
  29                               bufferlist *out)
  30 {
  31   bufferlist::iterator start = p;
  32   __u32 n;
  33   ::decode(n, p);
  34   unsigned len = 4;
  35   while (n--) {
  36     __u32 l;
  37     ::decode(l, p);
  38     p.advance(l);
  39     len += 4 + l;
  40     ::decode(l, p);
  41     p.advance(l);
  42     len += 4 + l;
  43   }
  44   start.copy(len, *out);
  45 }
  46
  47 void decode_str_set_to_bl(bufferlist::iterator& p,
  48                           bufferlist *out)
  49 {
  50   bufferlist::iterator start = p;
  51   __u32 n;
  52   ::decode(n, p);
  53   unsigned len = 4;
  54   while (n--) {
  55     __u32 l;
  56     ::decode(l, p);
  57     p.advance(l);
  58     len += 4 + l;
  59   }
  60   start.copy(len, *out);
  61 }
  62
  63 ObjectStore *ObjectStore::create(CephContext *cct,
  64                                  const string& type,
  65                                  const string& data,
  66                                  const string& journal,
  67                                  osflagbits_t flags)
  68 {
  69   if (type == "filestore") {
  70     return new FileStore(cct, data, journal, flags);
  71   }
  72   if (type == "memstore") {
  73     return new MemStore(cct, data);
  74   }
  75 #if defined(HAVE_LIBAIO)
  76   if (type == "bluestore" &&
  77       cct->check_experimental_feature_enabled("bluestore")) {
  78     return new BlueStore(cct, data);
  79   }
  80   if (type == "random" &&
  81       cct->check_experimental_feature_enabled("bluestore")) {
  82     if (rand() % 2) {
  83       return new FileStore(cct, data, journal, flags);
  84     } else {
  85       return new BlueStore(cct, data);
  86     }
  87   }
  88 #else
  89   if (type == "random") {
  90     return new FileStore(cct, data, journal, flags);
  91   }
  92 #endif
  93   if (type == "kstore" &&
  94       cct->check_experimental_feature_enabled("kstore")) {
  95     return new KStore(cct, data);
  96   }
  97   return NULL;
  98 }
  99
 100 int ObjectStore::probe_block_device_fsid(
 101   CephContext *cct,
 102   const string& path,
 103   uuid_d *fsid)
 104 {
 105   int r;
 106
 107 #if defined(HAVE_LIBAIO)
 108   // first try bluestore -- it has a crc on its header and will fail
 109   // reliably.
 110   r = BlueStore::get_block_device_fsid(cct, path, fsid);
 111   if (r == 0) {
 112     lgeneric_dout(cct, 0) << __func__ << " " << path << " is bluestore, "
 113                           << *fsid << dendl;
 114     return r;
 115   }
 116 #endif
 117
 118   // okay, try FileStore (journal).
 119   r = FileStore::get_block_device_fsid(cct, path, fsid);
 120   if (r == 0) {
 121     lgeneric_dout(cct, 0) << __func__ << " " << path << " is filestore, "
 122                           << *fsid << dendl;
 123     return r;
 124   }
 125
 126   return -EINVAL;
 127 }
 128
 129 int ObjectStore::write_meta(const std::string& key,
 130                             const std::string& value)
 131 {
 132   string v = value;
 133   v += "\n";
 134   int r = safe_write_file(path.c_str(), key.c_str(),
 135                           v.c_str(), v.length());
 136   if (r < 0)
 137     return r;
 138   return 0;
 139 }
 140
 141 int ObjectStore::read_meta(const std::string& key,
 142                            std::string *value)
 143 {
 144   char buf[4096];
 145   int r = safe_read_file(path.c_str(), key.c_str(),
 146                          buf, sizeof(buf));
 147   if (r <= 0)
 148     return r;
 149   // drop trailing newlines
 150   while (r && isspace(buf[r-1])) {
 151     --r;
 152   }
 153   *value = string(buf, r);
 154   return 0;
 155 }
 156
 157
 158
 159
 160 ostream& operator<<(ostream& out, const ObjectStore::Sequencer& s)
 161 {
 162   return out << "osr(" << s.get_name() << " " << &s << ")";
 163 }
 164
 165 ostream& operator<<(ostream& out, const ObjectStore::Transaction& tx) {
 166
 167   return out << "Transaction(" << &tx << ")";
 168 }
 169
 170 unsigned ObjectStore::apply_transactions(Sequencer *osr,
 171                                          vector<Transaction>& tls,
 172                                          Context *ondisk)
 173 {
 174   // use op pool
 175   Cond my_cond;
 176   Mutex my_lock("ObjectStore::apply_transaction::my_lock");
 177   int r = 0;
 178   bool done;
 179   C_SafeCond *onreadable = new C_SafeCond(&my_lock, &my_cond, &done, &r);
 180
 181   queue_transactions(osr, tls, onreadable, ondisk);
 182
 183   my_lock.Lock();
 184   while (!done)
 185     my_cond.Wait(my_lock);
 186   my_lock.Unlock();
 187   return r;
 188 }
 189
 190 int ObjectStore::queue_transactions(
 191   Sequencer *osr,
 192   vector<Transaction>& tls,
 193   Context *onreadable,
 194   Context *oncommit,
 195   Context *onreadable_sync,
 196   Context *oncomplete,
 197   TrackedOpRef op = TrackedOpRef())
 198 {
 199   RunOnDeleteRef _complete (std::make_shared<RunOnDelete>(oncomplete));
 200   Context *_onreadable = new Wrapper<RunOnDeleteRef>(
 201     onreadable, _complete);
 202   Context *_oncommit = new Wrapper<RunOnDeleteRef>(
 203     oncommit, _complete);
 204   return queue_transactions(osr, tls, _onreadable, _oncommit,
 205                             onreadable_sync, op);
 206 }