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