31 #define STAT_L(e) pthread_mutex_lock(&e->stats_mutex);  argument
32 #define STAT_UL(e) pthread_mutex_unlock(&e->stats_mutex);  argument
33 #define STAT_INCR(e, stat, amount) { \  argument
34     pthread_mutex_lock(&e->stats_mutex); \
35     e->stats.stat += amount; \
36     pthread_mutex_unlock(&e->stats_mutex); \
39 #define STAT_DECR(e, stat, amount) { \  argument
40     pthread_mutex_lock(&e->stats_mutex); \
41     e->stats.stat -= amount; \
42     pthread_mutex_unlock(&e->stats_mutex); \
82     store_engine *e;  member
142 static store_io_thread *_get_io_thread(store_engine *e) {  in _get_io_thread()  argument
145     pthread_mutex_lock(&e->mutex);  in _get_io_thread()
149     for (int x = 0; x < e->io_threadcount; x++) {  in _get_io_thread()
150         if (e->io_threads[x].depth == 0) {  in _get_io_thread()
153         } else if (e->io_threads[x].depth < low) {  in _get_io_thread()
155             low = e->io_threads[x].depth;  in _get_io_thread()
158     pthread_mutex_unlock(&e->mutex);  in _get_io_thread()
160     return &e->io_threads[tid];  in _get_io_thread()
163 static uint64_t _next_version(store_engine *e) {  in _next_version()  argument
164     return e->version++;  in _next_version()
167 static void _free_page(store_engine *e, store_page *p);
173     store_engine *e = (store_engine *)ptr;  in extstore_get_stats()  local
174     STAT_L(e);  in extstore_get_stats()
175     memcpy(st, &e->stats, sizeof(struct extstore_stats));  in extstore_get_stats()
176     STAT_UL(e);  in extstore_get_stats()
179     pthread_mutex_lock(&e->mutex);  in extstore_get_stats()
180     st->pages_free = e->page_free;  in extstore_get_stats()
181     st->pages_used = e->page_count - e->page_free;  in extstore_get_stats()
182     pthread_mutex_unlock(&e->mutex);  in extstore_get_stats()
184     for (int x = 0; x < e->io_threadcount; x++) {  in extstore_get_stats()
185         pthread_mutex_lock(&e->io_threads[x].mutex);  in extstore_get_stats()
186         st->io_queue += e->io_threads[x].depth;  in extstore_get_stats()
187         pthread_mutex_unlock(&e->io_threads[x].mutex);  in extstore_get_stats()
191     st->bytes_fragmented = st->pages_used * e->page_size -  in extstore_get_stats()
196     store_engine *e = (store_engine *)ptr;  in extstore_get_page_data()  local
197     pthread_mutex_lock(&e->maint.mutex);  in extstore_get_page_data()
200     for (int i = 0; i < e->page_count; i++) {  in extstore_get_page_data()
201         store_page *p = &e->pages[i];  in extstore_get_page_data()
218             _free_page(e, p);  in extstore_get_page_data()
223     pthread_mutex_unlock(&e->maint.mutex);  in extstore_get_page_data()
284     store_engine *e = calloc(1, sizeof(store_engine));  in extstore_init()  local
285     if (e == NULL) {  in extstore_init()
290     e->page_size = cf->page_size;  in extstore_init()
299             free(e);  in extstore_init()
310             free(e);  in extstore_init()
315             free(e);  in extstore_init()
325         free(e);  in extstore_init()
328     e->page_count = temp_page_count;  in extstore_init()
330     e->pages = calloc(e->page_count, sizeof(store_page));  in extstore_init()
331     if (e->pages == NULL) {  in extstore_init()
334         free(e);  in extstore_init()
340     for (i = 0; i < e->page_count; i++) {  in extstore_init()
354         pthread_mutex_init(&e->pages[i].mutex, NULL);  in extstore_init()
355         e->pages[i].id = i;  in extstore_init()
356         e->pages[i].fd = f->fd;  in extstore_init()
357         e->pages[i].free_bucket = f->free_bucket;  in extstore_init()
358         e->pages[i].offset = f->offset;  in extstore_init()
359         e->pages[i].free = true;  in extstore_init()
360         f->offset += e->page_size;  in extstore_init()
364     e->free_page_buckets = calloc(cf->page_buckets, sizeof(store_page *));  in extstore_init()
365     e->free_page_bucketcount = cf->page_buckets;  in extstore_init()
367     e->page_free = e->page_count;  in extstore_init()
368     for (i = e->page_count-1; i >= 0; i--) {  in extstore_init()
369         int fb = e->pages[i].free_bucket;  in extstore_init()
370         e->pages[i].next = e->free_page_buckets[fb];  in extstore_init()
371         e->free_page_buckets[fb] = &e->pages[i];  in extstore_init()
375     e->version = 1;  in extstore_init()
378     e->stats.page_data =  in extstore_init()
379         calloc(e->page_count, sizeof(struct extstore_page_data));  in extstore_init()
380     e->stats.page_count = e->page_count;  in extstore_init()
381     e->stats.page_size = e->page_size;  in extstore_init()
384     e->page_buckets = calloc(cf->page_buckets, sizeof(store_page *));  in extstore_init()
385     e->page_bucketcount = cf->page_buckets;  in extstore_init()
393         w->next = e->wbuf_stack;  in extstore_init()
394         e->wbuf_stack = w;  in extstore_init()
395         io->next = e->io_stack;  in extstore_init()
396         e->io_stack = io;  in extstore_init()
399     pthread_mutex_init(&e->mutex, NULL);  in extstore_init()
400     pthread_mutex_init(&e->stats_mutex, NULL);  in extstore_init()
401     pthread_mutex_init(&e->maint.mutex, NULL);  in extstore_init()
403     e->io_depth = cf->io_depth;  in extstore_init()
406     e->io_threads = calloc(cf->io_threadcount, sizeof(store_io_thread));  in extstore_init()
408         pthread_mutex_init(&e->io_threads[i].mutex, NULL);  in extstore_init()
409         pthread_cond_init(&e->io_threads[i].cond, NULL);  in extstore_init()
410         e->io_threads[i].e = e;  in extstore_init()
412         pthread_create(&thread, NULL, extstore_io_thread, &e->io_threads[i]);  in extstore_init()
415     e->io_threadcount = cf->io_threadcount;  in extstore_init()
418     e->bg_thread = calloc(1, sizeof(store_io_thread));  in extstore_init()
419     e->bg_thread->e = e;  in extstore_init()
420     pthread_mutex_init(&e->bg_thread->mutex, NULL);  in extstore_init()
421     pthread_cond_init(&e->bg_thread->cond, NULL);  in extstore_init()
422     pthread_create(&thread, NULL, extstore_io_thread, e->bg_thread);  in extstore_init()
425     return (void *)e;  in extstore_init()
429 static void _evict_page(store_engine *e, unsigned int bucket,  in _evict_page()  argument
432     st.page_data = calloc(e->page_count, sizeof(struct extstore_page_data));  in _evict_page()
433     extstore_get_page_data(e, &st);  in _evict_page()
439     for (int i = 0; i < e->page_count; i++) {  in _evict_page()
461         extstore_evict_page(e, low_page, low_version);  in _evict_page()
466 static store_page *_allocate_page(store_engine *e, unsigned int bucket,  in _allocate_page()  argument
469     assert(!e->page_buckets[bucket] || e->page_buckets[bucket]->allocated == e->page_size);  in _allocate_page()
471     if (e->free_page_buckets[free_bucket] != NULL) {  in _allocate_page()
472         assert(e->page_free > 0);  in _allocate_page()
473         tmp = e->free_page_buckets[free_bucket];  in _allocate_page()
474         e->free_page_buckets[free_bucket] = tmp->next;  in _allocate_page()
475     } else if (e->free_page_buckets[0] != NULL) {  in _allocate_page()
477         assert(e->page_free > 0);  in _allocate_page()
478         tmp = e->free_page_buckets[0];  in _allocate_page()
479         e->free_page_buckets[0] = tmp->next;  in _allocate_page()
482         tmp->next = e->page_buckets[bucket];  in _allocate_page()
483         e->page_buckets[bucket] = tmp;  in _allocate_page()
487         tmp->version = _next_version(e);  in _allocate_page()
489         e->page_free--;  in _allocate_page()
490         STAT_INCR(e, page_allocs, 1);  in _allocate_page()
494         E_DEBUG("EXTSTORE: got page %u [free:%u]\n", tmp->id, e->page_free);  in _allocate_page()
499 static void _allocate_wbuf(store_engine *e, store_page *p) {  in _allocate_wbuf()  argument
502     pthread_mutex_lock(&e->mutex);  in _allocate_wbuf()
503     if (e->wbuf_stack) {  in _allocate_wbuf()
504         wbuf = e->wbuf_stack;  in _allocate_wbuf()
505         e->wbuf_stack = wbuf->next;  in _allocate_wbuf()
508     pthread_mutex_unlock(&e->mutex);  in _allocate_wbuf()
528     store_engine *e = (store_engine *)ep;  in _wbuf_cb()  local
529     store_page *p = &e->pages[io->page_id];  in _wbuf_cb()
541     if (p->written == e->page_size)  in _wbuf_cb()
545     pthread_mutex_lock(&e->mutex);  in _wbuf_cb()
546     w->next = e->wbuf_stack;  in _wbuf_cb()
547     e->wbuf_stack = w;  in _wbuf_cb()
549     io->next = e->io_stack;  in _wbuf_cb()
550     e->io_stack = io;  in _wbuf_cb()
551     pthread_mutex_unlock(&e->mutex);  in _wbuf_cb()
558 static void _submit_wbuf(store_engine *e, store_page *p) {  in _submit_wbuf()  argument
560     pthread_mutex_lock(&e->mutex);  in _submit_wbuf()
561     obj_io *io = e->io_stack;  in _submit_wbuf()
562     e->io_stack = io->next;  in _submit_wbuf()
563     pthread_mutex_unlock(&e->mutex);  in _submit_wbuf()
578     extstore_submit_bg(e, io);  in _submit_wbuf()
592     store_engine *e = (store_engine *)ptr;  in extstore_write_request()  local
595     if (bucket >= e->page_bucketcount)  in extstore_write_request()
598     pthread_mutex_lock(&e->mutex);  in extstore_write_request()
599     p = e->page_buckets[bucket];  in extstore_write_request()
601         p = _allocate_page(e, bucket, free_bucket);  in extstore_write_request()
603     pthread_mutex_unlock(&e->mutex);  in extstore_write_request()
605         _evict_page(e, bucket, free_bucket);  in extstore_write_request()
614             ((!p->wbuf || p->wbuf->full) && p->allocated >= e->page_size)) {  in extstore_write_request()
616         pthread_mutex_lock(&e->mutex);  in extstore_write_request()
617         store_page *temp_p = _allocate_page(e, bucket, free_bucket);  in extstore_write_request()
618         pthread_mutex_unlock(&e->mutex);  in extstore_write_request()
620             _evict_page(e, bucket, free_bucket);  in extstore_write_request()
627         _submit_wbuf(e, p);  in extstore_write_request()
631     if (!p->wbuf && p->allocated < e->page_size) {  in extstore_write_request()
632         _allocate_wbuf(e, p);  in extstore_write_request()
652     store_engine *e = (store_engine *)ptr;  in extstore_write()  local
653     store_page *p = &e->pages[io->page_id];  in extstore_write()
661     STAT_L(e);  in extstore_write()
662     e->stats.bytes_written += io->len;  in extstore_write()
663     e->stats.bytes_used += io->len;  in extstore_write()
664     e->stats.objects_written++;  in extstore_write()
665     e->stats.objects_used++;  in extstore_write()
666     STAT_UL(e);  in extstore_write()
709     store_engine *e = (store_engine *)ptr;  in extstore_submit()  local
710     store_io_thread *t = _get_io_thread(e);  in extstore_submit()
715     store_engine *e = (store_engine *)ptr;  in extstore_submit_bg()  local
716     store_io_thread *t = e->bg_thread;  in extstore_submit_bg()
725     store_engine *e = (store_engine *)ptr;  in extstore_delete()  local
727     store_page *p = &e->pages[page_id];  in extstore_delete()
743         STAT_L(e);  in extstore_delete()
744         e->stats.bytes_used -= bytes;  in extstore_delete()
745         e->stats.objects_used -= count;  in extstore_delete()
746         STAT_UL(e);  in extstore_delete()
749             _free_page(e, p);  in extstore_delete()
759     store_engine *e = (store_engine *)ptr;  in extstore_check()  local
760     store_page *p = &e->pages[page_id];  in extstore_check()
772     store_engine *e = (store_engine *)ptr;  in extstore_close_page()  local
773     store_page *p = &e->pages[page_id];  in extstore_close_page()
779             _free_page(e, p);  in extstore_close_page()
787     store_engine *e = (store_engine *)ptr;  in extstore_evict_page()  local
788     store_page *p = &e->pages[page_id];  in extstore_evict_page()
796         STAT_L(e);  in extstore_evict_page()
797         e->stats.page_evictions++;  in extstore_evict_page()
798         e->stats.objects_evicted += p->obj_count;  in extstore_evict_page()
799         e->stats.bytes_evicted += p->bytes_used;  in extstore_evict_page()
800         STAT_UL(e);  in extstore_evict_page()
802             _free_page(e, p);  in extstore_evict_page()
840     store_engine *e = me->e;  in extstore_io_thread()  local
857             for (i = 1; i < e->io_depth; i++) {  in extstore_io_thread()
878             store_page *p = &e->pages[cur_io->page_id];  in extstore_io_thread()
891                         STAT_L(e);  in extstore_io_thread()
892                         e->stats.bytes_read += cur_io->len;  in extstore_io_thread()
893                         e->stats.objects_read++;  in extstore_io_thread()
894                         STAT_UL(e);  in extstore_io_thread()
937             cur_io->cb(e, cur_io, ret);  in extstore_io_thread()
951 static void _free_page(store_engine *e, store_page *p) {  in _free_page()  argument
955     STAT_L(e);  in _free_page()
956     e->stats.objects_used -= p->obj_count;  in _free_page()
957     e->stats.bytes_used -= p->bytes_used;  in _free_page()
958     e->stats.page_reclaims++;  in _free_page()
959     STAT_UL(e);  in _free_page()
960     pthread_mutex_lock(&e->mutex);  in _free_page()
962     tmp = e->page_buckets[p->bucket];  in _free_page()
968                 e->page_buckets[p->bucket] = tmp->next;  in _free_page()
987     p->next = e->free_page_buckets[p->free_bucket];  in _free_page()
988     e->free_page_buckets[p->free_bucket] = p;  in _free_page()
989     e->page_free++;  in _free_page()
990     E_DEBUG("EXTSTORE: pages free %u\n", e->page_free);  in _free_page()
991     pthread_mutex_unlock(&e->mutex);  in _free_page()