greg/wterl
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Cache session/[{cursor, config}] for reuse and spawn threads when needed. #9

Merged
gburd merged 30 commits from gsb-ctx-cache into master 2013-07-03 12:31:15 +00:00
Conversation 0 Commits 30 Files changed 18 +18 -456
6 changed files with 18 additions and 456 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit bc0f5dbfc7 - Show all commits

3
c_src/async_nif.h
View file

@ -44,7 +44,6 @@ struct async_nif_req_entry {
void *args;
void (*fn_work)(ErlNifEnv*, ERL_NIF_TERM, ErlNifPid*, unsigned int, void *);
void (*fn_post)(void *);
uint64_t submitted;
STAILQ_ENTRY(async_nif_req_entry) entries;
};
@ -109,7 +108,6 @@ struct async_nif_state {
enif_make_atom(env, "shutdown")); \
} \
req = async_nif_reuse_req(async_nif); \
req->submitted = ts(ns); \
if (!req) { \
return enif_make_tuple2(env, enif_make_atom(env, "error"), \
enif_make_atom(env, "eagain")); \
@ -415,7 +413,6 @@ async_nif_worker_fn(void *arg)
/* Clean up req for reuse. */
req->ref = 0;
req->submitted = 0;
req->fn_work = 0;
req->fn_post = 0;
enif_free(req->args);

123
c_src/duration.h
View file

@ -1,123 +0,0 @@
/*
* Copyright (C) 2013, all rights reserved by Gregory Burd <greg@burd.me>
*
* This Source Code Form is subject to the terms of the Mozilla Public License,
* version 2 (MPLv2). If a copy of the MPL was not distributed with this file,
* you can obtain one at: http://mozilla.org/MPL/2.0/
*
* NOTES:
* - on some platforms this will require -lrt
*/
#include <stdio.h>
#include <stdint.h>
#include <time.h>
#include <sys/timeb.h>
#ifdef __MACH__
#include <mach/clock.h>
#include <mach/mach.h>
#endif
static inline void current_utc_time(struct timespec *ts)
{
#ifdef __MACH__ // OS X does not have clock_gettime, use clock_get_time
clock_serv_t cclock;
mach_timespec_t mts;
host_get_clock_service(mach_host_self(), CALENDAR_CLOCK, &cclock);
clock_get_time(cclock, &mts);
mach_port_deallocate(mach_task_self(), cclock);
ts->tv_sec = mts.tv_sec;
ts->tv_nsec = mts.tv_nsec;
#else
clock_gettime(CLOCK_REALTIME, ts);
#endif
}
typedef enum { ns = 0, mcs, ms, s } time_scale;
struct scale_time {
const char *abbreviation;
const char *name;
uint64_t mul, div, overhead, ticks_per;
};
static const struct scale_time scale[] = {
{ "ns", "nanosecond", 1000000000LL, 1LL, 10, 2300000000000LL },
{ "μs", "microsecond", 1000000LL, 1000LL, 10, 2300000000LL },
{ "ms", "millisecond", 1000LL, 1000000LL, 10, 2300000LL },
{ "sec", "second", 1LL, 1000000000LL, 10, 2300LL } };
static uint64_t ts(time_scale unit)
{
struct timespec ts;
current_utc_time(&ts);
return (((uint64_t)ts.tv_sec * scale[unit].mul) +
((uint64_t)ts.tv_nsec / scale[unit].div));
}
#if defined(__i386__) || defined(__x86_64__)
/**
* cpu_clock_ticks()
*
* A measure provided by Intel x86 CPUs which provides the number of cycles
* (aka "ticks") executed as a counter using the RDTSC instruction.
*/
static inline uint64_t cpu_clock_ticks()
{
uint32_t lo, hi;
__asm__ __volatile__ (
"XORL %%eax, %%eax\n" /* Flush the pipeline */
"CPUID\n"
"RDTSC\n" /* Get RDTSC counter in edx:eax */
: "=a" (lo), "=d" (hi)
:
: "%ebx", "%ecx" );
return (uint64_t)hi << 32 | lo;
}
#endif
#if 0
/**
* cpu_clock_ticks()
*
* An approximation of elapsed [ns, mcs, ms, s] from CPU clock ticks.
*/
static uint64_t elapsed_cpu_clock_ticks(uint64_t start, time_scale unit)
{
return (cpu_clock_ticks() - start - scale[unit].overhead) * scale[unit].ticks_per;
}
#endif
typedef struct {
uint64_t then;
time_scale unit;
} duration_t;
static inline uint64_t elapsed(duration_t *d)
{
uint64_t now = ts(d->unit);
uint64_t elapsed = now - d->then;
d->then = now;
return elapsed;
}
#define DURATION(name, resolution) duration_t name = \
{ts(resolution), resolution}
#define ELAPSED_DURING(result, resolution, block) \
do { \
DURATION(__x, resolution); \
do block while(0); \
*result = elapsed(&__x); \
} while(0);
#define CYCLES_DURING(result, block) \
do { \
uint64_t __begin = cpu_clock_ticks(); \
do block while(0); \
*result = cpu_clock_ticks() - __begin; \
} while(0);

11
c_src/queue.h
View file

@ -33,6 +33,17 @@
#ifndef _DB_QUEUE_H_
#define _DB_QUEUE_H_
#ifndef __offsetof
#define __offsetof(st, m) \
((size_t) ( (char *)&((st *)0)->m - (char *)0 ))
#endif
#ifndef __containerof
#define __containerof(ptr, type, member) ({ \
const typeof( ((type *)0)->member ) *__mptr = (ptr); \
(type *)( (char *)__mptr - __offsetof(type,member) );})
#endif
#if defined(__cplusplus)
extern "C" {
#endif

240
c_src/stats.c
View file

@ -1,240 +0,0 @@
/*
* stats:
*
* Copyright (c) 2012 Basho Technologies, Inc. All Rights Reserved.
* Author: Gregory Burd <greg@basho.com> <greg@burd.me>
*
* This file is provided to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain
* a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#include <inttypes.h>
#include "erl_nif.h"
#include "erl_driver.h"
#include "common.h"
#include "duration.h"
#include "stats.h"
/**
* Calculate the log2 of 64bit unsigned integers.
*/
#ifdef __GCC__
#define LOG2(X) ((unsigned) ((8 * (sizeof(uint64_t) - 1)) - __builtin_clzll((X))))
#else
static unsigned int __log2_64(uint64_t x) {
static const int tab64[64] = {
63, 0, 58, 1, 59, 47, 53, 2,
60, 39, 48, 27, 54, 33, 42, 3,
61, 51, 37, 40, 49, 18, 28, 20,
55, 30, 34, 11, 43, 14, 22, 4,
62, 57, 46, 52, 38, 26, 32, 41,
50, 36, 17, 19, 29, 10, 13, 21,
56, 45, 25, 31, 35, 16, 9, 12,
44, 24, 15, 8, 23, 7, 6, 5};
if (x == 0) return 0;
uint64_t v = x;
v |= v >> 1;
v |= v >> 2;
v |= v >> 4;
v |= v >> 8;
v |= v >> 16;
v |= v >> 32;
return tab64[((uint64_t)((v - (v >> 1)) * 0x07EDD5E59A4E28C2)) >> 58];
}
#define LOG2(X) __log2_64(X)
#endif
double
__stat_mean(struct stat *s)
{
uint32_t t, h;
double mean;
if (!s)
return 0.0;
t = s->h;
h = (s->h + 1) % s->num_samples;
mean = 0;
while (h != t) {
mean += s->samples[h];
h = (h + 1) % s->num_samples;
}
if (mean > 0)
mean /= (s->n < s->num_samples ? (double)s->n : (double)s->num_samples);
return mean;
}
double
__stat_mean_log2(struct stat *s)
{
uint32_t i;
double mean = 0.0;
if (!s)
return 0.0;
for (i = 0; i < 64; i++)
mean += (s->histogram[i] * i);
if (mean > 0)
mean /= (s->n < s->num_samples ? s->n : s->num_samples);
return mean;
}
uint64_t
__stat_tick(struct stat *s)
{
uint64_t t;
if (!s)
return 0.0;
t = ts(s->d.unit);
s->d.then = t;
return t;
}
void
__stat_reset(struct stat *s)
{
if (!s)
return;
s->h = 0;
s->d.unit = ns;
s->d.then = 0;
memset(s->histogram, 0, sizeof(uint64_t) * 64);
memset(s->samples, 0, sizeof(uint64_t) * s->num_samples);
}
void
__stat_add(struct stat *s, uint64_t elapsed)
{
uint32_t i;
if (s->n == s->num_samples) {
s->mean = (s->mean + __stat_mean(s)) / 2.0;
if (s->n >= 4294967295) {
__stat_reset(s);
}
}
i = s->h;
s->h = (s->h + 1) % s->num_samples;
s->samples[i] = elapsed;
if (elapsed != 0 && elapsed < s->min)
s->min = elapsed;
if (elapsed > s->max)
s->max = elapsed;
s->histogram[LOG2(elapsed)]++;
s->n++;
}
void
__stat_tock(struct stat *s)
{
if (s)
__stat_add(s, ts(s->d.unit));
}
void
__stat_print_histogram(struct stat *s, const char *mod)
{
uint8_t logs[64];
uint8_t i, j, max_log = 0;
double m;
if (!s)
return;
m = (s->mean + __stat_mean(s) / 2.0);
fprintf(stderr, "%s:async_nif request latency histogram:\n", mod);
for (i = 0; i < 64; i++) {
logs[i] = LOG2(s->histogram[i]);
if (logs[i] > max_log)
max_log = logs[i];
}
for (i = max_log; i > 0; i--) {
if (!(i % 10))
fprintf(stderr, "2^%2d ", i);
else
fprintf(stderr, " ");
for(j = 0; j < 64; j++)
fprintf(stderr, logs[j] >= i ? "•" : " ");
fprintf(stderr, "\n");
}
if (max_log == 0) {
fprintf(stderr, "[empty]\n");
} else {
fprintf(stderr, " ns μs ms s ks\n");
fprintf(stderr, "min: ");
if (s->min < 1000)
fprintf(stderr, "%llu (ns)", PRIuint64(s->min));
else if (s->min < 1000000)
fprintf(stderr, "%.2f (μs)", s->min / 1000.0);
else if (s->min < 1000000000)
fprintf(stderr, "%.2f (ms)", s->min / 1000000.0);
else if (s->min < 1000000000000)
fprintf(stderr, "%.2f (s)", s->min / 1000000000.0);
fprintf(stderr, " max: ");
if (s->max < 1000)
fprintf(stderr, "%llu (ns)", PRIuint64(s->max));
else if (s->max < 1000000)
fprintf(stderr, "%.2f (μs)", s->max / 1000.0);
else if (s->max < 1000000000)
fprintf(stderr, "%.2f (ms)", s->max / 1000000.0);
else if (s->max < 1000000000000)
fprintf(stderr, "%.2f (s)", s->max / 1000000000.0);
fprintf(stderr, " mean: ");
if (m < 1000)
fprintf(stderr, "%.2f (ns)", m);
else if (m < 1000000)
fprintf(stderr, "%.2f (μs)", m / 1000.0);
else if (m < 1000000000)
fprintf(stderr, "%.2f (ms)", m / 1000000.0);
else if (m < 1000000000000)
fprintf(stderr, "%.2f (s)", m / 1000000000.0);
fprintf(stderr, "\n");
}
fflush(stderr);
}
void
__stat_free(struct stat *s)
{
if (!s)
return;
enif_free(s->samples);
enif_free(s);
}
struct stat *
__stat_init(uint32_t n)
{
struct stat *s = enif_alloc(sizeof(struct stat) + (sizeof(uint64_t) * n));
if (!s)
return NULL;
memset(s, 0, sizeof(struct stat) + (sizeof(uint64_t) * n));
s->min = ~0;
s->max = 0;
s->mean = 0.0;
s->h = 0;
s->num_samples = n;
s->d.unit = ns;
return s;
}

55
c_src/stats.h
View file

@ -1,55 +0,0 @@
/*
* stats:
*
* Copyright (c) 2012 Basho Technologies, Inc. All Rights Reserved.
* Author: Gregory Burd <greg@basho.com> <greg@burd.me>
*
* This file is provided to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain
* a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#ifndef __STATS_H__
#define __STATS_H__
#if defined(__cplusplus)
extern "C" {
#endif
#define STAT_DEF(name) struct stat *name ## _stat;
struct stat {
duration_t d;
uint32_t h, n, num_samples;
uint64_t min, max;
double mean;
uint64_t histogram[64];
uint64_t samples[];
};
extern double __stat_mean(struct stat *s);
extern double __stat_mean_log2(struct stat *s);
extern uint64_t __stat_tick(struct stat *s);
extern void __stat_add(struct stat *s, uint64_t d);
extern void __stat_reset(struct stat *s);
extern void __stat_tock(struct stat *s);
extern void __stat_print_histogram(struct stat *s, const char *mod);
extern void __stat_free(struct stat *s);
extern struct stat *__stat_init(uint32_t n);
#if defined(__cplusplus)
}
#endif
#endif // __STATS_H__

42
c_src/wterl.c
View file

@ -29,8 +29,6 @@
#include "wiredtiger.h"
#include "common.h"
#include "duration.h"
#include "stats.h"
#include "async_nif.h"
#include "queue.h"
#include "cas.h"
@ -44,7 +42,6 @@ typedef char Uri[128];
struct wterl_ctx {
STAILQ_ENTRY(wterl_ctx) entries;
uint64_t tstamp;
uint64_t sig;
size_t sig_len;
WT_SESSION *session;
@ -64,8 +61,6 @@ typedef struct wterl_conn {
ErlNifMutex *cache_mutex;
uint32_t cache_size;
struct wterl_ctx *mru_ctx[ASYNC_NIF_MAX_WORKERS];
uint64_t histogram[64];
uint64_t histogram_count;
} WterlConnHandle;
typedef struct {
@ -193,47 +188,27 @@ static inline uint32_t __log2(uint64_t x) {
static int
__ctx_cache_evict(WterlConnHandle *conn_handle)
{
uint32_t mean, log, num_evicted, i;
uint64_t now, elapsed;
struct wterl_ctx *c, *n;
uint32_t mean, num_evicted;
struct wterl_ctx *c;
#ifndef DEBUG
if (conn_handle->cache_size < MAX_CACHE_SIZE)
return 0;
#endif
now = ts(ns);
mean = conn_handle->cache_size / 2;
if (mean < 2) return 0;
// Find the mean of the recorded times that items stayed in cache.
mean = 0;
for (i = 0; i < 64; i++)
mean += (conn_handle->histogram[i] * i);
if (mean > 0)
mean /= conn_handle->histogram_count;
// Clear out the histogram and hit/misses
memset(conn_handle->histogram, 0, sizeof(uint64_t) * 64);
conn_handle->histogram_count = 0;
/*
* Evict anything older than the mean time in queue by removing those
* items from the lists stored in the tree.
*/
num_evicted = 0;
c = STAILQ_FIRST(&conn_handle->cache);
while (c != NULL) {
n = STAILQ_NEXT(c, entries);
elapsed = c->tstamp - now;
log = __log2(elapsed);
if (log > mean) {
while (mean--) {
c = STAILQ_LAST(&conn_handle->cache, wterl_ctx, entries);
if (c) {
STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries);
DPRINTF("evicting: %llu", PRIuint64(c->sig));
if (c->session)
c->session->close(c->session, NULL);
enif_free(c);
num_evicted++;
}
c = n;
}
conn_handle->cache_size -= num_evicted;
return num_evicted;
@ -259,8 +234,6 @@ __ctx_cache_find(WterlConnHandle *conn_handle, const uint64_t sig)
if (c->sig == sig) { // TODO: hash collisions *will* lead to SEGVs
// cache hit:
STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries);
conn_handle->histogram[__log2(ts(ns) - c->tstamp)]++;
conn_handle->histogram_count++;
conn_handle->cache_size -= 1;
break;
}
@ -282,7 +255,6 @@ __ctx_cache_add(WterlConnHandle *conn_handle, struct wterl_ctx *c)
{
enif_mutex_lock(conn_handle->cache_mutex);
__ctx_cache_evict(conn_handle);
c->tstamp = ts(ns);
STAILQ_INSERT_TAIL(&conn_handle->cache, c, entries);
conn_handle->cache_size += 1;
#ifdef DEBUG