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