ceph/src/os/filestore/IndexManager.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 "include/unordered_map.h"
  16
  17 #if defined(__FreeBSD__)
  18 #include <sys/param.h>
  19 #endif
  20
  21 #include <errno.h>
  22
  23 #include "common/Cond.h"
  24 #include "common/config.h"
  25 #include "common/debug.h"
  26 #include "include/buffer.h"
  27
  28 #include "IndexManager.h"
  29 #include "HashIndex.h"
  30 #include "CollectionIndex.h"
  31
  32 #include "chain_xattr.h"
  33
  34 using std::string;
  35
  36 using ceph::bufferlist;
  37 using ceph::bufferptr;
  38 using ceph::decode;
  39 using ceph::encode;
  40
  41 static int set_version(const char *path, uint32_t version) {
  42   bufferlist bl;
  43   encode(version, bl);
  44   return chain_setxattr<true, true>(
  45     path, "user.cephos.collection_version", bl.c_str(),
  46     bl.length());
  47 }
  48
  49 static int get_version(const char *path, uint32_t *version) {
  50   bufferptr bp(PATH_MAX);
  51   int r = chain_getxattr(path, "user.cephos.collection_version",
  52                       bp.c_str(), bp.length());
  53   if (r < 0) {
  54     if (r != -ENOENT) {
  55       *version = 0;
  56       return 0;
  57     } else {
  58       return r;
  59     }
  60   }
  61   bp.set_length(r);
  62   bufferlist bl;
  63   bl.push_back(bp);
  64   auto i = bl.cbegin();
  65   decode(*version, i);
  66   return 0;
  67 }
  68
  69 IndexManager::~IndexManager() {
  70
  71   for (ceph::unordered_map<coll_t, CollectionIndex* > ::iterator it = col_indices.begin();
  72        it != col_indices.end(); ++it) {
  73
  74     delete it->second;
  75     it->second = NULL;
  76   }
  77   col_indices.clear();
  78 }
  79
  80
  81 int IndexManager::init_index(coll_t c, const char *path, uint32_t version) {
  82   std::unique_lock l{lock};
  83   int r = set_version(path, version);
  84   if (r < 0)
  85     return r;
  86   HashIndex index(cct, c, path, cct->_conf->filestore_merge_threshold,
  87                   cct->_conf->filestore_split_multiple,
  88                   version,
  89                   cct->_conf->filestore_index_retry_probability);
  90   r = index.init();
  91   if (r < 0)
  92     return r;
  93   return index.read_settings();
  94 }
  95
  96 int IndexManager::build_index(coll_t c, const char *path, CollectionIndex **index) {
  97   if (upgrade) {
  98     // Need to check the collection generation
  99     int r;
 100     uint32_t version = 0;
 101     r = get_version(path, &version);
 102     if (r < 0)
 103       return r;
 104
 105     switch (version) {
 106     case CollectionIndex::FLAT_INDEX_TAG:
 107     case CollectionIndex::HASH_INDEX_TAG: // fall through
 108     case CollectionIndex::HASH_INDEX_TAG_2: // fall through
 109     case CollectionIndex::HOBJECT_WITH_POOL: {
 110       // Must be a HashIndex
 111       *index = new HashIndex(cct, c, path,
 112                              cct->_conf->filestore_merge_threshold,
 113                              cct->_conf->filestore_split_multiple,
 114                              version);
 115       return (*index)->read_settings();
 116     }
 117     default: ceph_abort();
 118     }
 119
 120   } else {
 121     // No need to check
 122     *index = new HashIndex(cct, c, path, cct->_conf->filestore_merge_threshold,
 123                            cct->_conf->filestore_split_multiple,
 124                            CollectionIndex::HOBJECT_WITH_POOL,
 125                            cct->_conf->filestore_index_retry_probability);
 126     return (*index)->read_settings();
 127   }
 128 }
 129
 130 bool IndexManager::get_index_optimistic(coll_t c, Index *index) {
 131   std::shared_lock l{lock};
 132   ceph::unordered_map<coll_t, CollectionIndex* > ::iterator it = col_indices.find(c);
 133   if (it == col_indices.end())
 134     return false;
 135   index->index = it->second;
 136   return true;
 137 }
 138
 139 int IndexManager::get_index(coll_t c, const string& baseDir, Index *index) {
 140   if (get_index_optimistic(c, index))
 141     return 0;
 142   std::unique_lock l{lock};
 143   ceph::unordered_map<coll_t, CollectionIndex* > ::iterator it = col_indices.find(c);
 144   if (it == col_indices.end()) {
 145     char path[PATH_MAX];
 146     snprintf(path, sizeof(path), "%s/current/%s", baseDir.c_str(), c.to_str().c_str());
 147     CollectionIndex* colIndex = NULL;
 148     int r = build_index(c, path, &colIndex);
 149     if (r < 0)
 150       return r;
 151     col_indices[c] = colIndex;
 152     index->index = colIndex;
 153   } else {
 154     index->index = it->second;
 155   }
 156   return 0;
 157 }