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   return index.init();
  86 }
  87
  88 int IndexManager::build_index(coll_t c, const char *path, CollectionIndex **index) {
  89   if (upgrade) {
  90     // Need to check the collection generation
  91     int r;
  92     uint32_t version = 0;
  93     r = get_version(path, &version);
  94     if (r < 0)
  95       return r;
  96
  97     switch (version) {
  98     case CollectionIndex::FLAT_INDEX_TAG:
  99     case CollectionIndex::HASH_INDEX_TAG: // fall through
 100     case CollectionIndex::HASH_INDEX_TAG_2: // fall through
 101     case CollectionIndex::HOBJECT_WITH_POOL: {
 102       // Must be a HashIndex
 103       *index = new HashIndex(cct, c, path,
 104                              cct->_conf->filestore_merge_threshold,
 105                              cct->_conf->filestore_split_multiple, version);
 106       return 0;
 107     }
 108     default: ceph_abort();
 109     }
 110
 111   } else {
 112     // No need to check
 113     *index = new HashIndex(cct, c, path, cct->_conf->filestore_merge_threshold,
 114                            cct->_conf->filestore_split_multiple,
 115                            CollectionIndex::HOBJECT_WITH_POOL,
 116                            cct->_conf->filestore_index_retry_probability);
 117     return 0;
 118   }
 119 }
 120
 121 bool IndexManager::get_index_optimistic(coll_t c, Index *index) {
 122   RWLock::RLocker l(lock);
 123   ceph::unordered_map<coll_t, CollectionIndex* > ::iterator it = col_indices.find(c);
 124   if (it == col_indices.end())
 125     return false;
 126   index->index = it->second;
 127   return true;
 128 }
 129
 130 int IndexManager::get_index(coll_t c, const string& baseDir, Index *index) {
 131   if (get_index_optimistic(c, index))
 132     return 0;
 133   RWLock::WLocker l(lock);
 134   ceph::unordered_map<coll_t, CollectionIndex* > ::iterator it = col_indices.find(c);
 135   if (it == col_indices.end()) {
 136     char path[PATH_MAX];
 137     snprintf(path, sizeof(path), "%s/current/%s", baseDir.c_str(), c.to_str().c_str());
 138     CollectionIndex* colIndex = NULL;
 139     int r = build_index(c, path, &colIndex);
 140     if (r < 0)
 141       return r;
 142     col_indices[c] = colIndex;
 143     index->index = colIndex;
 144   } else {
 145     index->index = it->second;
 146   }
 147   return 0;
 148 }