silo/masstree/kvthread.cc

   1 /* Masstree
   2  * Eddie Kohler, Yandong Mao, Robert Morris
   3  * Copyright (c) 2012-2014 President and Fellows of Harvard College
   4  * Copyright (c) 2012-2014 Massachusetts Institute of Technology
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, subject to the conditions
   9  * listed in the Masstree LICENSE file. These conditions include: you must
  10  * preserve this copyright notice, and you cannot mention the copyright
  11  * holders in advertising related to the Software without their permission.
  12  * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
  13  * notice is a summary of the Masstree LICENSE file; the license in that file
  14  * is legally binding.
  15  */
  16 #include "kvthread.hh"
  17 #include <string.h>
  18 #include <stdio.h>
  19 #include <stdlib.h>
  20 #include <new>
  21 #include <sys/mman.h>
  22 #if HAVE_SUPERPAGE && !NOSUPERPAGE
  23 #include <sys/types.h>
  24 #include <dirent.h>
  25 #endif
  26
  27 threadinfo *threadinfo::allthreads;
  28 pthread_key_t threadinfo::key;
  29 #if ENABLE_ASSERTIONS
  30 int threadinfo::no_pool_value;
  31 #endif
  32
  33 #if HAVE_MEMDEBUG
  34 void memdebug::landmark(char* buf, size_t size) const {
  35     if (this->magic != magic_value && this->magic != magic_free_value)
  36         snprintf(buf, size, "???");
  37     else if (this->file)
  38         snprintf(buf, size, "%s:%d", this->file, this->line);
  39     else if (this->line)
  40         snprintf(buf, size, "%d", this->line);
  41     else
  42         snprintf(buf, size, "0");
  43 }
  44
  45 void
  46 memdebug::hard_free_checks(const memdebug *m, size_t size, int freetype,
  47                            int after_rcu, const char *op) {
  48     char buf[256];
  49     m->landmark(buf, sizeof(buf));
  50     if (m->magic == magic_free_value)
  51         fprintf(stderr, "%s(%p): double free, was @%s\n",
  52                 op, m + 1, buf);
  53     else if (m->magic != magic_value)
  54         fprintf(stderr, "%s(%p): freeing unallocated pointer (%x)\n",
  55                 op, m + 1, m->magic);
  56     assert(m->magic == magic_value);
  57     if (freetype && m->freetype != freetype)
  58         fprintf(stderr, "%s(%p): expected type %x, saw %x, "
  59                 "allocated %s\n", op, m + 1, freetype, m->freetype, buf);
  60     if (!after_rcu && m->size != size)
  61         fprintf(stderr, "%s(%p): expected size %lu, saw %lu, "
  62                 "allocated %s\n", op, m + 1,
  63                 (unsigned long) size, (unsigned long) m->size, buf);
  64     if (m->after_rcu != after_rcu)
  65         fprintf(stderr, "%s(%p): double free after rcu, allocated @%s\n",
  66                 op, m + 1, buf);
  67     if (freetype)
  68         assert(m->freetype == freetype);
  69     if (!after_rcu)
  70         assert(m->size == size);
  71     assert(m->after_rcu == after_rcu);
  72 }
  73
  74 void
  75 memdebug::hard_assert_use(const void *ptr, memtag tag1, memtag tag2) {
  76     const memdebug *m = reinterpret_cast<const memdebug *>(ptr) - 1;
  77     char tagbuf[40], buf[256];
  78     m->landmark(buf, sizeof(buf));
  79     if (tag2 == (memtag) -1)
  80         sprintf(buf, "%x", tag1);
  81     else
  82         sprintf(buf, "%x/%x", tag1, tag2);
  83     if (m->magic == magic_free_value)
  84         fprintf(stderr, "%p: use tag %s after free, allocated %s\n",
  85                 m + 1, tagbuf, buf);
  86     else if (m->magic != magic_value)
  87         fprintf(stderr, "%p: pointer is unallocated, not tag %s\n",
  88                 m + 1, tagbuf);
  89     assert(m->magic == magic_value);
  90     if (tag1 != 0 && (m->freetype >> 8) != tag1 && (m->freetype >> 8) != tag2)
  91         fprintf(stderr, "%p: expected tag %s, got tag %x, allocated %s\n",
  92                 m + 1, tagbuf, m->freetype >> 8, buf);
  93     if (tag1 != 0)
  94         assert((m->freetype >> 8) == tag1 || (m->freetype >> 8) == tag2);
  95 }
  96 #endif
  97
  98 threadinfo *threadinfo::make(int purpose, int index) {
  99     static int threads_initialized;
 100
 101     threadinfo *ti = (threadinfo *) malloc(8192);
 102     memset(ti, 0, sizeof(*ti));
 103     ti->next_ = allthreads;
 104     ti->purpose_ = purpose;
 105     ti->index_ = index;
 106     ti->allthreads = ti;
 107     ti->ts_ = 2;
 108     void *limbo_space = ti->allocate(sizeof(limbo_group), memtag_limbo);
 109     ti->mark(tc_limbo_slots, limbo_group::capacity);
 110     ti->limbo_head_ = ti->limbo_tail_ = new(limbo_space) limbo_group;
 111
 112     if (!threads_initialized) {
 113 #if ENABLE_ASSERTIONS
 114         const char* s = getenv("_");
 115         no_pool_value = s && strstr(s, "valgrind") != 0;
 116 #endif
 117         threads_initialized = 1;
 118     }
 119
 120     return ti;
 121 }
 122
 123 void threadinfo::refill_rcu()
 124 {
 125     if (limbo_head_ == limbo_tail_ && !limbo_tail_->next_
 126         && limbo_tail_->head_ == limbo_tail_->tail_)
 127         limbo_tail_->head_ = limbo_tail_->tail_ = 0;
 128     else if (!limbo_tail_->next_) {
 129         void *limbo_space = allocate(sizeof(limbo_group), memtag_limbo);
 130         mark(tc_limbo_slots, limbo_group::capacity);
 131         limbo_tail_->next_ = new(limbo_space) limbo_group;
 132         limbo_tail_ = limbo_tail_->next_;
 133     } else
 134         limbo_tail_ = limbo_tail_->next_;
 135 }
 136
 137 void threadinfo::hard_rcu_quiesce()
 138 {
 139     uint64_t min_epoch = gc_epoch_;
 140     for (threadinfo *ti = allthreads; ti; ti = ti->next()) {
 141         prefetch((const void *) ti->next());
 142         uint64_t epoch = ti->gc_epoch_;
 143         if (epoch && (int64_t) (epoch - min_epoch) < 0)
 144             min_epoch = epoch;
 145     }
 146
 147     limbo_group *lg = limbo_head_;
 148     limbo_element *lb = &lg->e_[lg->head_];
 149     limbo_element *le = &lg->e_[lg->tail_];
 150
 151     if (lb != le && (int64_t) (lb->epoch_ - min_epoch) < 0) {
 152         while (1) {
 153             free_rcu(lb->ptr_, lb->freetype_);
 154             mark(tc_gc);
 155
 156             ++lb;
 157
 158             if (lb == le && lg == limbo_tail_) {
 159                 lg->head_ = lg->tail_;
 160                 break;
 161             } else if (lb == le) {
 162                 assert(lg->tail_ == lg->capacity && lg->next_);
 163                 lg->head_ = lg->tail_ = 0;
 164                 lg = lg->next_;
 165                 lb = &lg->e_[lg->head_];
 166                 le = &lg->e_[lg->tail_];
 167             } else if (lb->epoch_ < min_epoch) {
 168                 lg->head_ = lb - lg->e_;
 169                 break;
 170             }
 171         }
 172
 173         if (lg != limbo_head_) {
 174             // shift nodes in [limbo_head_, limbo_tail_) to be after
 175             // limbo_tail_
 176             limbo_group *old_head = limbo_head_;
 177             limbo_head_ = lg;
 178             limbo_group **last = &limbo_tail_->next_;
 179             while (*last)
 180                 last = &(*last)->next_;
 181             *last = old_head;
 182             while (*last != lg)
 183                 last = &(*last)->next_;
 184             *last = 0;
 185         }
 186     }
 187
 188     limbo_epoch_ = (lb == le ? 0 : lb->epoch_);
 189 }
 190
 191 void threadinfo::report_rcu(void *ptr) const
 192 {
 193     for (limbo_group *lg = limbo_head_; lg; lg = lg->next_) {
 194         int status = 0;
 195         for (int i = 0; i < lg->capacity; ++i) {
 196             if (i == lg->head_)
 197                 status = 1;
 198             if (i == lg->tail_)
 199                 status = 0;
 200             if (lg->e_[i].ptr_ == ptr)
 201                 fprintf(stderr, "thread %d: rcu %p@%d: %s as %x @%" PRIu64 "\n",
 202                         index_, lg, i, status ? "waiting" : "freed",
 203                         lg->e_[i].freetype_, lg->e_[i].epoch_);
 204         }
 205     }
 206 }
 207
 208 void threadinfo::report_rcu_all(void *ptr)
 209 {
 210     for (threadinfo *ti = allthreads; ti; ti = ti->next())
 211         ti->report_rcu(ptr);
 212 }
 213
 214
 215 #if HAVE_SUPERPAGE && !NOSUPERPAGE
 216 static size_t read_superpage_size() {
 217     if (DIR* d = opendir("/sys/kernel/mm/hugepages")) {
 218         size_t n = (size_t) -1;
 219         while (struct dirent* de = readdir(d))
 220             if (de->d_type == DT_DIR
 221                 && strncmp(de->d_name, "hugepages-", 10) == 0
 222                 && de->d_name[10] >= '0' && de->d_name[10] <= '9') {
 223                 size_t x = strtol(&de->d_name[10], 0, 10) << 10;
 224                 n = (x < n ? x : n);
 225             }
 226         closedir(d);
 227         return n;
 228     } else
 229         return 2 << 20;
 230 }
 231
 232 static size_t superpage_size = 0;
 233 #endif
 234
 235 static void initialize_pool(void* pool, size_t sz, size_t unit) {
 236     char* p = reinterpret_cast<char*>(pool);
 237     void** nextptr = reinterpret_cast<void**>(p);
 238     for (size_t off = unit; off + unit <= sz; off += unit) {
 239         *nextptr = p + off;
 240         nextptr = reinterpret_cast<void**>(p + off);
 241     }
 242     *nextptr = 0;
 243 }
 244
 245 void threadinfo::refill_pool(int nl) {
 246     assert(!pool_[nl - 1]);
 247
 248     if (!use_pool()) {
 249         pool_[nl - 1] = malloc(nl * CACHE_LINE_SIZE);
 250         if (pool_[nl - 1])
 251             *reinterpret_cast<void**>(pool_[nl - 1]) = 0;
 252         return;
 253     }
 254
 255     void* pool = 0;
 256     size_t pool_size = 0;
 257     int r;
 258
 259 #if HAVE_SUPERPAGE && !NOSUPERPAGE
 260     if (!superpage_size)
 261         superpage_size = read_superpage_size();
 262     if (superpage_size != (size_t) -1) {
 263         pool_size = superpage_size;
 264 # if MADV_HUGEPAGE
 265         if ((r = posix_memalign(&pool, pool_size, pool_size)) != 0) {
 266             fprintf(stderr, "posix_memalign superpage: %s\n", strerror(r));
 267             pool = 0;
 268             superpage_size = (size_t) -1;
 269         } else if (madvise(pool, pool_size, MADV_HUGEPAGE) != 0) {
 270             perror("madvise superpage");
 271             superpage_size = (size_t) -1;
 272         }
 273 # elif MAP_HUGETLB
 274         pool = mmap(0, pool_size, PROT_READ | PROT_WRITE,
 275                     MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB, -1, 0);
 276         if (pool == MAP_FAILED) {
 277             perror("mmap superpage");
 278             pool = 0;
 279             superpage_size = (size_t) -1;
 280         }
 281 # else
 282         superpage_size = (size_t) -1;
 283 # endif
 284     }
 285 #endif
 286
 287     if (!pool) {
 288         pool_size = 2 << 20;
 289         if ((r = posix_memalign(&pool, CACHE_LINE_SIZE, pool_size)) != 0) {
 290             fprintf(stderr, "posix_memalign: %s\n", strerror(r));
 291             abort();
 292         }
 293     }
 294
 295     initialize_pool(pool, pool_size, nl * CACHE_LINE_SIZE);
 296     pool_[nl - 1] = pool;
 297 }
 298
 299 void threadinfo::run() {
 300     threadid_ = pthread_self();
 301     pthread_setspecific(key, this);
 302 }
 303
 304 void* threadinfo::thread_trampoline(void* argument) {
 305     threadinfo* ti = static_cast<threadinfo*>(argument);
 306     ti->run();
 307     return ti->thread_func_(ti);
 308 }
 309
 310 int threadinfo::run(void* (*thread_func)(threadinfo*), void* thread_data) {
 311     assert(!thread_func_ && !threadid_);
 312     thread_func_ = thread_func;
 313     thread_data_ = thread_data;
 314     return pthread_create(&threadid_, 0, thread_trampoline, this);
 315 }