cache/cache.cpp

   1 /*
   2  * 2012+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
   3  * All rights reserved.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  */
  15
  16 #include <iostream>
  17 #include <deque>
  18 #include <vector>
  19 #include <deque>
  20
  21 #include <boost/unordered_map.hpp>
  22 #include <boost/shared_array.hpp>
  23 #include <boost/shared_ptr.hpp>
  24 #include <boost/make_shared.hpp>
  25 #include <boost/thread.hpp>
  26 #include <boost/intrusive/list.hpp>
  27 #include <boost/intrusive/set.hpp>
  28
  29 #include "../library/elliptics.h"
  30
  31 #include "elliptics/packet.h"
  32 #include "elliptics/interface.h"
  33
  34 namespace ioremap { namespace cache {
  35
  36 struct key_t {
  37         key_t(const unsigned char *id) {
  38                 memcpy(this->id, id, DNET_ID_SIZE);
  39         }
  40
  41         unsigned char id[DNET_ID_SIZE];
  42 };
  43
  44 size_t hash(const unsigned char *id) {
  45         size_t num = DNET_ID_SIZE / sizeof(size_t);
  46
  47         size_t *ptr = (size_t *)id;
  48         size_t hash = 0x883eaf5a;
  49         for (size_t i = 0; i < num; ++i)
  50                 hash ^= ptr[i];
  51
  52         return hash;
  53 }
  54
  55 struct hash_t {
  56         std::size_t operator()(const key_t &key) const {
  57                 return ioremap::cache::hash(key.id);
  58         }
  59 };
  60
  61 struct equal_to {
  62         bool operator() (const key_t &x, const key_t &y) const {
  63                 return memcmp(x.id, y.id, DNET_ID_SIZE) == 0;
  64         }
  65 };
  66
  67 class raw_data_t {
  68         public:
  69                 raw_data_t(const char *data, size_t size) {
  70                         m_data.reserve(size);
  71                         m_data.insert(m_data.begin(), data, data + size);
  72                 }
  73
  74                 std::vector<char> &data(void) {
  75                         return m_data;
  76                 }
  77
  78                 size_t size(void) {
  79                         return m_data.size();
  80                 }
  81
  82         private:
  83                 std::vector<char> m_data;
  84 };
  85
  86 struct data_lru_tag_t;
  87 typedef boost::intrusive::list_base_hook<boost::intrusive::tag<data_lru_tag_t>,
  88                                          boost::intrusive::link_mode<boost::intrusive::safe_link>
  89                                         > lru_list_base_hook_t;
  90 struct data_set_tag_t;
  91 typedef boost::intrusive::set_base_hook<boost::intrusive::tag<data_set_tag_t>,
  92                                          boost::intrusive::link_mode<boost::intrusive::safe_link>
  93                                         > set_base_hook_t;
  94
  95 struct time_set_tag_t;
  96 typedef boost::intrusive::set_base_hook<boost::intrusive::tag<time_set_tag_t>,
  97                                          boost::intrusive::link_mode<boost::intrusive::safe_link>
  98                                         > time_set_base_hook_t;
  99
 100 class data_t : public lru_list_base_hook_t, public set_base_hook_t, public time_set_base_hook_t {
 101         public:
 102                 data_t(const unsigned char *id) : m_lifetime(0) {
 103                         memcpy(m_id.id, id, DNET_ID_SIZE);
 104                 }
 105
 106                 data_t(const unsigned char *id, size_t lifetime, const char *data, size_t size, bool remove_from_disk) :
 107                 m_lifetime(0), m_remove_from_disk(remove_from_disk) {
 108                         memcpy(m_id.id, id, DNET_ID_SIZE);
 109
 110                         if (lifetime)
 111                                 m_lifetime = lifetime + time(NULL);
 112
 113                         m_data.reset(new raw_data_t(data, size));
 114                 }
 115
 116                 ~data_t() {
 117                 }
 118
 119                 const struct dnet_raw_id &id(void) const {
 120                         return m_id;
 121                 }
 122
 123                 boost::shared_ptr<raw_data_t> data(void) const {
 124                         return m_data;
 125                 }
 126
 127                 size_t lifetime(void) const {
 128                         return m_lifetime;
 129                 }
 130
 131                 bool remove_from_disk() const {
 132                         return m_remove_from_disk;
 133                 }
 134
 135                 size_t size(void) const {
 136                         return m_data->size();
 137                 }
 138
 139                 friend bool operator< (const data_t &a, const data_t &b) {
 140                         return dnet_id_cmp_str(a.id().id, b.id().id) < 0;
 141                 }
 142
 143                 friend bool operator> (const data_t &a, const data_t &b) {
 144                         return dnet_id_cmp_str(a.id().id, b.id().id) > 0;
 145                 }
 146
 147                 friend bool operator== (const data_t &a, const data_t &b) {
 148                         return dnet_id_cmp_str(a.id().id, b.id().id) == 0;
 149                 }
 150
 151         private:
 152                 size_t m_lifetime;
 153                 bool m_remove_from_disk;
 154                 struct dnet_raw_id m_id;
 155                 boost::shared_ptr<raw_data_t> m_data;
 156 };
 157
 158 typedef boost::intrusive::list<data_t, boost::intrusive::base_hook<lru_list_base_hook_t> > lru_list_t;
 159 typedef boost::intrusive::set<data_t, boost::intrusive::base_hook<set_base_hook_t>,
 160                                           boost::intrusive::compare<std::less<data_t> >
 161                              > iset_t;
 162
 163 struct lifetime_less {
 164         bool operator() (const data_t &x, const data_t &y) const {
 165                 return x.lifetime() < y.lifetime();
 166         }
 167 };
 168
 169 typedef boost::intrusive::set<data_t, boost::intrusive::base_hook<time_set_base_hook_t>,
 170                                           boost::intrusive::compare<lifetime_less>
 171                              > life_set_t;
 172
 173 class cache_t {
 174         public:
 175                 cache_t(struct dnet_node *n) : m_need_exit(false), m_node(n), m_cache_size(0), m_max_cache_size(n->cache_size) {
 176                         m_lifecheck = boost::thread(boost::bind(&cache_t::life_check, this));
 177                 }
 178
 179                 ~cache_t() {
 180                         m_need_exit = true;
 181                         m_lifecheck.join();
 182
 183                         while (!m_lru.empty()) {
 184                                 data_t raw = m_lru.front();
 185                                 erase_element(&raw);
 186                         }
 187                 }
 188
 189                 void write(const unsigned char *id, size_t lifetime, const char *data, size_t size, bool remove_from_disk) {
 190                         boost::mutex::scoped_lock guard(m_lock);
 191
 192                         iset_t::iterator it = m_set.find(id);
 193                         if (it != m_set.end())
 194                                 erase_element(&(*it));
 195
 196                         if (size + m_cache_size > m_max_cache_size)
 197                                 resize(size * 2);
 198
 199                         /*
 200                          * nothing throws exception below this 'new' operator, so there is no try/catch block
 201                          */
 202                         data_t *raw = new data_t(id, lifetime, data, size, remove_from_disk);
 203
 204                         m_set.insert(*raw);
 205                         m_lru.push_back(*raw);
 206                         if (lifetime)
 207                                 m_lifeset.insert(*raw);
 208
 209                         m_cache_size += size;
 210                 }
 211
 212                 boost::shared_ptr<raw_data_t> read(const unsigned char *id) {
 213                         boost::mutex::scoped_lock guard(m_lock);
 214
 215                         iset_t::iterator it = m_set.find(id);
 216                         if (it == m_set.end())
 217                                 throw std::runtime_error("no record");
 218
 219                         m_lru.erase(m_lru.iterator_to(*it));
 220                         m_lru.push_back(*it);
 221                         return it->data();
 222                 }
 223
 224                 bool remove(const unsigned char *id) {
 225                         bool removed = false;
 226                         bool remove_from_disk = false;
 227
 228                         boost::mutex::scoped_lock guard(m_lock);
 229                         iset_t::iterator it = m_set.find(id);
 230                         if (it != m_set.end()) {
 231                                 remove_from_disk = it->remove_from_disk();
 232                                 erase_element(&(*it));
 233                                 removed = true;
 234                         }
 235
 236                         guard.unlock();
 237
 238                         if (remove_from_disk) {
 239                                 struct dnet_id raw;
 240
 241                                 dnet_setup_id(&raw, 0, (unsigned char *)id);
 242                                 raw.type = -1;
 243
 244                                 dnet_remove_local(m_node, &raw);
 245                         }
 246
 247                         return removed;
 248                 }
 249
 250         private:
 251                 bool m_need_exit;
 252                 struct dnet_node *m_node;
 253                 size_t m_cache_size, m_max_cache_size;
 254                 boost::mutex m_lock;
 255                 iset_t m_set;
 256                 lru_list_t m_lru;
 257                 life_set_t m_lifeset;
 258                 boost::thread m_lifecheck;
 259
 260                 void resize(size_t reserve) {
 261                         while (!m_lru.empty()) {
 262                                 data_t *raw = &m_lru.front();
 263
 264                                 erase_element(raw);
 265
 266
 267                                 /* break early if free space in cache more than requested reserve */
 268                                 if (m_max_cache_size - m_cache_size > reserve)
 269                                         break;
 270                         }
 271                 }
 272
 273                 void erase_element(data_t *obj) {
 274                         m_lru.erase(m_lru.iterator_to(*obj));
 275                         m_set.erase(m_set.iterator_to(*obj));
 276                         if (obj->lifetime())
 277                                 m_lifeset.erase(m_lifeset.iterator_to(*obj));
 278
 279                         m_cache_size -= obj->size();
 280
 281                         delete obj;
 282                 }
 283
 284                 void life_check(void) {
 285                         while (!m_need_exit) {
 286                                 std::deque<struct dnet_id> remove;
 287
 288                                 while (!m_need_exit && !m_lifeset.empty()) {
 289                                         size_t time = ::time(NULL);
 290
 291                                         boost::mutex::scoped_lock guard(m_lock);
 292
 293                                         if (m_lifeset.empty())
 294                                                 break;
 295
 296                                         life_set_t::iterator it = m_lifeset.begin();
 297                                         if (it->lifetime() > time)
 298                                                 break;
 299
 300                                         if (it->remove_from_disk()) {
 301                                                 struct dnet_id id;
 302
 303                                                 dnet_setup_id(&id, 0, (unsigned char *)it->id().id);
 304                                                 id.type = -1;
 305
 306                                                 remove.push_back(id);
 307                                         }
 308
 309                                         erase_element(&(*it));
 310                                 }
 311
 312                                 for (std::deque<struct dnet_id>::iterator it = remove.begin(); it != remove.end(); ++it) {
 313                                         dnet_remove_local(m_node, &(*it));
 314                                 }
 315
 316                                 sleep(1);
 317                         }
 318                 }
 319 };
 320
 321 }}
 322
 323 using namespace ioremap::cache;
 324
 325 int dnet_cmd_cache_io(struct dnet_net_state *st, struct dnet_cmd *cmd, struct dnet_io_attr *io, char *data)
 326 {
 327         struct dnet_node *n = st->n;
 328         int err = -ENOTSUP;
 329
 330         if (!n->cache)
 331                 return -ENOTSUP;
 332
 333         cache_t *cache = (cache_t *)n->cache;
 334
 335         try {
 336                 boost::shared_ptr<raw_data_t> d;
 337
 338                 switch (cmd->cmd) {
 339                         case DNET_CMD_WRITE:
 340                                 if (io->flags & DNET_IO_FLAGS_COMPARE_AND_SWAP) {
 341                                         d = cache->read(io->id);
 342
 343                                         dnet_id csum;
 344                                         dnet_transform(n, d->data().data(), d->data().size(), &csum);
 345
 346                                         if (!memcmp(csum.id, io->parent, DNET_ID_SIZE)) {
 347                                                 err = -EINVAL;
 348                                                 break;
 349                                         }
 350                                 }
 351
 352                                 cache->write(io->id, io->start, data, io->size, !!(io->flags & DNET_IO_FLAGS_CACHE_REMOVE_FROM_DISK));
 353                                 err = 0;
 354                                 break;
 355                         case DNET_CMD_READ:
 356                                 d = cache->read(io->id);
 357                                 if (io->offset + io->size > d->size()) {
 358                                         dnet_log_raw(n, DNET_LOG_ERROR, "%s: %s cache: invalid offset/size: "
 359                                                         "offset: %llu, size: %llu, cached-size: %zd\n",
 360                                                         dnet_dump_id(&cmd->id), dnet_cmd_string(cmd->cmd),
 361                                                         (unsigned long long)io->offset, (unsigned long long)io->size,
 362                                                         d->size());
 363                                         err = -EINVAL;
 364                                         break;
 365                                 }
 366
 367                                 io->size = d->size();
 368                                 err = dnet_send_read_data(st, cmd, io, (char *)d->data().data() + io->offset, -1, io->offset, 0);
 369                                 break;
 370                         case DNET_CMD_DEL:
 371                                 err = -ENOENT;
 372                                 if (cache->remove(cmd->id.id))
 373                                         err = 0;
 374                                 break;
 375                 }
 376         } catch (const std::exception &e) {
 377                 dnet_log_raw(n, DNET_LOG_ERROR, "%s: %s cache operation failed: %s\n",
 378                                 dnet_dump_id(&cmd->id), dnet_cmd_string(cmd->cmd), e.what());
 379                 err = -ENOENT;
 380         }
 381
 382         return err;
 383 }
 384
 385 int dnet_cache_init(struct dnet_node *n)
 386 {
 387         if (!n->cache_size)
 388                 return 0;
 389
 390         try {
 391                 n->cache = (void *)(new cache_t(n));
 392         } catch (const std::exception &e) {
 393                 dnet_log_raw(n, DNET_LOG_ERROR, "Could not create cache: %s\n", e.what());
 394                 return -ENOMEM;
 395         }
 396
 397         return 0;
 398 }
 399
 400 void dnet_cache_cleanup(struct dnet_node *n)
 401 {
 402         if (n->cache)
 403                 delete (cache_t *)n->cache;
 404 }