|
Packit |
4e8bc4 |
/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#include <stdlib.h>
|
|
Packit |
4e8bc4 |
#include <stdio.h>
|
|
Packit |
4e8bc4 |
#include <string.h>
|
|
Packit |
4e8bc4 |
#include <errno.h>
|
|
Packit |
4e8bc4 |
#include <poll.h>
|
|
Packit |
4e8bc4 |
#include <ctype.h>
|
|
Packit |
4e8bc4 |
#include <stdarg.h>
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#if defined(__sun)
|
|
Packit |
4e8bc4 |
#include <atomic.h>
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#include "memcached.h"
|
|
Packit |
4e8bc4 |
#include "bipbuffer.h"
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#ifdef LOGGER_DEBUG
|
|
Packit |
4e8bc4 |
#define L_DEBUG(...) \
|
|
Packit |
4e8bc4 |
do { \
|
|
Packit |
4e8bc4 |
fprintf(stderr, __VA_ARGS__); \
|
|
Packit |
4e8bc4 |
} while (0)
|
|
Packit |
4e8bc4 |
#else
|
|
Packit |
4e8bc4 |
#define L_DEBUG(...)
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* TODO: put this in a struct and ditch the global vars. */
|
|
Packit |
4e8bc4 |
static logger *logger_stack_head = NULL;
|
|
Packit |
4e8bc4 |
static logger *logger_stack_tail = NULL;
|
|
Packit |
4e8bc4 |
static unsigned int logger_count = 0;
|
|
Packit |
4e8bc4 |
static volatile int do_run_logger_thread = 1;
|
|
Packit |
4e8bc4 |
static pthread_t logger_tid;
|
|
Packit |
4e8bc4 |
pthread_mutex_t logger_stack_lock = PTHREAD_MUTEX_INITIALIZER;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_key_t logger_key;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#if !defined(HAVE_GCC_64ATOMICS) && !defined(__sun)
|
|
Packit |
4e8bc4 |
pthread_mutex_t logger_atomics_mutex = PTHREAD_MUTEX_INITIALIZER;
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#define WATCHER_LIMIT 20
|
|
Packit |
4e8bc4 |
logger_watcher *watchers[20];
|
|
Packit |
4e8bc4 |
struct pollfd watchers_pollfds[20];
|
|
Packit |
4e8bc4 |
int watcher_count = 0;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Should this go somewhere else? */
|
|
Packit |
4e8bc4 |
static const entry_details default_entries[] = {
|
|
Packit |
4e8bc4 |
[LOGGER_ASCII_CMD] = {LOGGER_TEXT_ENTRY, 512, LOG_RAWCMDS, "<%d %s"},
|
|
Packit |
4e8bc4 |
[LOGGER_EVICTION] = {LOGGER_EVICTION_ENTRY, 512, LOG_EVICTIONS, NULL},
|
|
Packit |
4e8bc4 |
[LOGGER_ITEM_GET] = {LOGGER_ITEM_GET_ENTRY, 512, LOG_FETCHERS, NULL},
|
|
Packit |
4e8bc4 |
[LOGGER_ITEM_STORE] = {LOGGER_ITEM_STORE_ENTRY, 512, LOG_MUTATIONS, NULL},
|
|
Packit |
4e8bc4 |
[LOGGER_CRAWLER_STATUS] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=lru_crawler crawler=%d lru=%s low_mark=%llu next_reclaims=%llu since_run=%u next_run=%d elapsed=%u examined=%llu reclaimed=%llu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_SLAB_MOVE] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=slab_move src=%d dst=%d"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
#ifdef EXTSTORE
|
|
Packit |
4e8bc4 |
[LOGGER_EXTSTORE_WRITE] = {LOGGER_EXT_WRITE_ENTRY, 512, LOG_EVICTIONS, NULL},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_START] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_start id=%lu version=%llu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_ABORT] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_abort id=%lu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_READ_START] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_read_start id=%lu offset=%llu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_READ_END] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_read_end id=%lu offset=%llu rescues=%lu lost=%lu skipped=%lu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_END] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_end id=%lu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
[LOGGER_COMPACT_FRAGINFO] = {LOGGER_TEXT_ENTRY, 512, LOG_SYSEVENTS,
|
|
Packit |
4e8bc4 |
"type=compact_fraginfo ratio=%.2f bytes=%lu"
|
|
Packit |
4e8bc4 |
},
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
};
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#define WATCHER_ALL -1
|
|
Packit |
4e8bc4 |
static int logger_thread_poll_watchers(int force_poll, int watcher);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/*************************
|
|
Packit |
4e8bc4 |
* Util functions shared between bg thread and workers
|
|
Packit |
4e8bc4 |
*************************/
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Logger GID's can be used by watchers to put logs back into strict order
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static uint64_t logger_gid = 0;
|
|
Packit |
4e8bc4 |
uint64_t logger_get_gid(void) {
|
|
Packit |
4e8bc4 |
#ifdef HAVE_GCC_64ATOMICS
|
|
Packit |
4e8bc4 |
return __sync_add_and_fetch(&logger_gid, 1);
|
|
Packit |
4e8bc4 |
#elif defined(__sun)
|
|
Packit |
4e8bc4 |
return atomic_inc_64_nv(&logger_gid);
|
|
Packit |
4e8bc4 |
#else
|
|
Packit |
4e8bc4 |
mutex_lock(&logger_atomics_mutex);
|
|
Packit |
4e8bc4 |
uint64_t res = ++logger_gid;
|
|
Packit |
4e8bc4 |
mutex_unlock(&logger_atomics_mutex);
|
|
Packit |
4e8bc4 |
return res;
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
void logger_set_gid(uint64_t gid) {
|
|
Packit |
4e8bc4 |
#ifdef HAVE_GCC_64ATOMICS
|
|
Packit |
4e8bc4 |
__sync_add_and_fetch(&logger_gid, gid);
|
|
Packit |
4e8bc4 |
#elif defined(__sun)
|
|
Packit |
4e8bc4 |
atomic_add_64(&logger_gid);
|
|
Packit |
4e8bc4 |
#else
|
|
Packit |
4e8bc4 |
mutex_lock(&logger_atomics_mutex);
|
|
Packit |
4e8bc4 |
logger_gid = gid;
|
|
Packit |
4e8bc4 |
mutex_unlock(&logger_atomics_mutex);
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* TODO: genericize lists. would be nice to import queue.h if the impact is
|
|
Packit |
4e8bc4 |
* studied... otherwise can just write a local one.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
/* Add to the list of threads with a logger object */
|
|
Packit |
4e8bc4 |
static void logger_link_q(logger *l) {
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
assert(l != logger_stack_head);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
l->prev = 0;
|
|
Packit |
4e8bc4 |
l->next = logger_stack_head;
|
|
Packit |
4e8bc4 |
if (l->next) l->next->prev = l;
|
|
Packit |
4e8bc4 |
logger_stack_head = l;
|
|
Packit |
4e8bc4 |
if (logger_stack_tail == 0) logger_stack_tail = l;
|
|
Packit |
4e8bc4 |
logger_count++;
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Remove from the list of threads with a logger object */
|
|
Packit |
4e8bc4 |
/*static void logger_unlink_q(logger *l) {
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
if (logger_stack_head == l) {
|
|
Packit |
4e8bc4 |
assert(l->prev == 0);
|
|
Packit |
4e8bc4 |
logger_stack_head = l->next;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
if (logger_stack_tail == l) {
|
|
Packit |
4e8bc4 |
assert(l->next == 0);
|
|
Packit |
4e8bc4 |
logger_stack_tail = l->prev;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
assert(l->next != l);
|
|
Packit |
4e8bc4 |
assert(l->prev != l);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (l->next) l->next->prev = l->prev;
|
|
Packit |
4e8bc4 |
if (l->prev) l->prev->next = l->next;
|
|
Packit |
4e8bc4 |
logger_count--;
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return;
|
|
Packit |
4e8bc4 |
}*/
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Called with logger stack locked.
|
|
Packit |
4e8bc4 |
* Iterates over every watcher collecting enabled flags.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static void logger_set_flags(void) {
|
|
Packit |
4e8bc4 |
logger *l = NULL;
|
|
Packit |
4e8bc4 |
int x = 0;
|
|
Packit |
4e8bc4 |
uint16_t f = 0; /* logger eflags */
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
for (x = 0; x < WATCHER_LIMIT; x++) {
|
|
Packit |
4e8bc4 |
logger_watcher *w = watchers[x];
|
|
Packit |
4e8bc4 |
if (w == NULL)
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
f |= w->eflags;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
for (l = logger_stack_head; l != NULL; l=l->next) {
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&l->mutex);
|
|
Packit |
4e8bc4 |
l->eflags = f;
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
return;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/*************************
|
|
Packit |
4e8bc4 |
* Logger background thread functions. Aggregates per-worker buffers and
|
|
Packit |
4e8bc4 |
* writes to any watchers.
|
|
Packit |
4e8bc4 |
*************************/
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#define LOGGER_PARSE_SCRATCH 4096
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static int _logger_thread_parse_ise(logentry *e, char *scratch) {
|
|
Packit |
4e8bc4 |
int total;
|
|
Packit |
4e8bc4 |
const char *cmd = "na";
|
|
Packit |
4e8bc4 |
char keybuf[KEY_MAX_URI_ENCODED_LENGTH];
|
|
Packit |
4e8bc4 |
struct logentry_item_store *le = (struct logentry_item_store *) e->data;
|
|
Packit |
4e8bc4 |
const char * const status_map[] = {
|
|
Packit |
4e8bc4 |
"not_stored", "stored", "exists", "not_found", "too_large", "no_memory" };
|
|
Packit |
4e8bc4 |
const char * const cmd_map[] = {
|
|
Packit |
4e8bc4 |
"null", "add", "set", "replace", "append", "prepend", "cas" };
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (le->cmd <= 6)
|
|
Packit |
4e8bc4 |
cmd = cmd_map[le->cmd];
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
uriencode(le->key, keybuf, le->nkey, KEY_MAX_URI_ENCODED_LENGTH);
|
|
Packit |
4e8bc4 |
total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
|
|
Packit |
4e8bc4 |
"ts=%d.%d gid=%llu type=item_store key=%s status=%s cmd=%s ttl=%u clsid=%u cfd=%d\n",
|
|
Packit |
4e8bc4 |
(int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
|
|
Packit |
4e8bc4 |
keybuf, status_map[le->status], cmd, le->ttl, le->clsid, le->sfd);
|
|
Packit |
4e8bc4 |
return total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static int _logger_thread_parse_ige(logentry *e, char *scratch) {
|
|
Packit |
4e8bc4 |
int total;
|
|
Packit |
4e8bc4 |
struct logentry_item_get *le = (struct logentry_item_get *) e->data;
|
|
Packit |
4e8bc4 |
char keybuf[KEY_MAX_URI_ENCODED_LENGTH];
|
|
Packit |
4e8bc4 |
const char * const was_found_map[] = {
|
|
Packit |
4e8bc4 |
"not_found", "found", "flushed", "expired" };
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
uriencode(le->key, keybuf, le->nkey, KEY_MAX_URI_ENCODED_LENGTH);
|
|
Packit |
4e8bc4 |
total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
|
|
Packit |
4e8bc4 |
"ts=%d.%d gid=%llu type=item_get key=%s status=%s clsid=%u cfd=%d\n",
|
|
Packit |
4e8bc4 |
(int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
|
|
Packit |
4e8bc4 |
keybuf, was_found_map[le->was_found], le->clsid, le->sfd);
|
|
Packit |
4e8bc4 |
return total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static int _logger_thread_parse_ee(logentry *e, char *scratch) {
|
|
Packit |
4e8bc4 |
int total;
|
|
Packit |
4e8bc4 |
char keybuf[KEY_MAX_URI_ENCODED_LENGTH];
|
|
Packit |
4e8bc4 |
struct logentry_eviction *le = (struct logentry_eviction *) e->data;
|
|
Packit |
4e8bc4 |
uriencode(le->key, keybuf, le->nkey, KEY_MAX_URI_ENCODED_LENGTH);
|
|
Packit |
4e8bc4 |
total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
|
|
Packit |
4e8bc4 |
"ts=%d.%d gid=%llu type=eviction key=%s fetch=%s ttl=%lld la=%d clsid=%u\n",
|
|
Packit |
4e8bc4 |
(int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
|
|
Packit |
4e8bc4 |
keybuf, (le->it_flags & ITEM_FETCHED) ? "yes" : "no",
|
|
Packit |
4e8bc4 |
(long long int)le->exptime, le->latime, le->clsid);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
return total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
#ifdef EXTSTORE
|
|
Packit |
4e8bc4 |
static int _logger_thread_parse_extw(logentry *e, char *scratch) {
|
|
Packit |
4e8bc4 |
int total;
|
|
Packit |
4e8bc4 |
char keybuf[KEY_MAX_URI_ENCODED_LENGTH];
|
|
Packit |
4e8bc4 |
struct logentry_ext_write *le = (struct logentry_ext_write *) e->data;
|
|
Packit |
4e8bc4 |
uriencode(le->key, keybuf, le->nkey, KEY_MAX_URI_ENCODED_LENGTH);
|
|
Packit |
4e8bc4 |
total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
|
|
Packit |
4e8bc4 |
"ts=%d.%d gid=%llu type=extwrite key=%s fetch=%s ttl=%lld la=%d clsid=%u bucket=%u\n",
|
|
Packit |
4e8bc4 |
(int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
|
|
Packit |
4e8bc4 |
keybuf, (le->it_flags & ITEM_FETCHED) ? "yes" : "no",
|
|
Packit |
4e8bc4 |
(long long int)le->exptime, le->latime, le->clsid, le->bucket);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
return total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
/* Completes rendering of log line. */
|
|
Packit |
4e8bc4 |
static enum logger_parse_entry_ret logger_thread_parse_entry(logentry *e, struct logger_stats *ls,
|
|
Packit |
4e8bc4 |
char *scratch, int *scratch_len) {
|
|
Packit |
4e8bc4 |
int total = 0;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
switch (e->event) {
|
|
Packit |
4e8bc4 |
case LOGGER_TEXT_ENTRY:
|
|
Packit |
4e8bc4 |
total = snprintf(scratch, LOGGER_PARSE_SCRATCH, "ts=%d.%d gid=%llu %s\n",
|
|
Packit |
4e8bc4 |
(int)e->tv.tv_sec, (int)e->tv.tv_usec,
|
|
Packit |
4e8bc4 |
(unsigned long long) e->gid, (char *) e->data);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
case LOGGER_EVICTION_ENTRY:
|
|
Packit |
4e8bc4 |
total = _logger_thread_parse_ee(e, scratch);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
#ifdef EXTSTORE
|
|
Packit |
4e8bc4 |
case LOGGER_EXT_WRITE_ENTRY:
|
|
Packit |
4e8bc4 |
total = _logger_thread_parse_extw(e, scratch);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
case LOGGER_ITEM_GET_ENTRY:
|
|
Packit |
4e8bc4 |
total = _logger_thread_parse_ige(e, scratch);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
case LOGGER_ITEM_STORE_ENTRY:
|
|
Packit |
4e8bc4 |
total = _logger_thread_parse_ise(e, scratch);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (total >= LOGGER_PARSE_SCRATCH || total <= 0) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Failed to flatten log entry!\n");
|
|
Packit |
4e8bc4 |
return LOGGER_PARSE_ENTRY_FAILED;
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
*scratch_len = total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
return LOGGER_PARSE_ENTRY_OK;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Writes flattened entry to available watchers */
|
|
Packit |
4e8bc4 |
static void logger_thread_write_entry(logentry *e, struct logger_stats *ls,
|
|
Packit |
4e8bc4 |
char *scratch, int scratch_len) {
|
|
Packit |
4e8bc4 |
int x, total;
|
|
Packit |
4e8bc4 |
/* Write the line into available watchers with matching flags */
|
|
Packit |
4e8bc4 |
for (x = 0; x < WATCHER_LIMIT; x++) {
|
|
Packit |
4e8bc4 |
logger_watcher *w = watchers[x];
|
|
Packit |
4e8bc4 |
char *skip_scr = NULL;
|
|
Packit |
4e8bc4 |
if (w == NULL || (e->eflags & w->eflags) == 0)
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Avoid poll()'ing constantly when buffer is full by resetting a
|
|
Packit |
4e8bc4 |
* flag periodically.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
while (!w->failed_flush &&
|
|
Packit |
4e8bc4 |
(skip_scr = (char *) bipbuf_request(w->buf, scratch_len + 128)) == NULL) {
|
|
Packit |
4e8bc4 |
if (logger_thread_poll_watchers(0, x) <= 0) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Watcher had no free space for line of size (%d)\n", scratch_len + 128);
|
|
Packit |
4e8bc4 |
w->failed_flush = true;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (w->failed_flush) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Fast skipped for watcher [%d] due to failed_flush\n", w->sfd);
|
|
Packit |
4e8bc4 |
w->skipped++;
|
|
Packit |
4e8bc4 |
ls->watcher_skipped++;
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (w->skipped > 0) {
|
|
Packit |
4e8bc4 |
total = snprintf(skip_scr, 128, "skipped=%llu\n", (unsigned long long) w->skipped);
|
|
Packit |
4e8bc4 |
if (total >= 128 || total <= 0) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Failed to flatten skipped message into watcher [%d]\n", w->sfd);
|
|
Packit |
4e8bc4 |
w->skipped++;
|
|
Packit |
4e8bc4 |
ls->watcher_skipped++;
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
bipbuf_push(w->buf, total);
|
|
Packit |
4e8bc4 |
w->skipped = 0;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
/* Can't fail because bipbuf_request succeeded. */
|
|
Packit |
4e8bc4 |
bipbuf_offer(w->buf, (unsigned char *) scratch, scratch_len);
|
|
Packit |
4e8bc4 |
ls->watcher_sent++;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Called with logger stack locked.
|
|
Packit |
4e8bc4 |
* Releases every chunk associated with a watcher and closes the connection.
|
|
Packit |
4e8bc4 |
* We can't presently send a connection back to the worker for further
|
|
Packit |
4e8bc4 |
* processing.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static void logger_thread_close_watcher(logger_watcher *w) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Closing dead watcher\n");
|
|
Packit |
4e8bc4 |
watchers[w->id] = NULL;
|
|
Packit |
4e8bc4 |
sidethread_conn_close(w->c);
|
|
Packit |
4e8bc4 |
watcher_count--;
|
|
Packit |
4e8bc4 |
bipbuf_free(w->buf);
|
|
Packit |
4e8bc4 |
free(w);
|
|
Packit |
4e8bc4 |
logger_set_flags();
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Reads a particular worker thread's available bipbuf bytes. Parses each log
|
|
Packit |
4e8bc4 |
* entry into the watcher buffers.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static int logger_thread_read(logger *l, struct logger_stats *ls) {
|
|
Packit |
4e8bc4 |
unsigned int size;
|
|
Packit |
4e8bc4 |
unsigned int pos = 0;
|
|
Packit |
4e8bc4 |
unsigned char *data;
|
|
Packit |
4e8bc4 |
char scratch[LOGGER_PARSE_SCRATCH];
|
|
Packit |
4e8bc4 |
logentry *e;
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&l->mutex);
|
|
Packit |
4e8bc4 |
data = bipbuf_peek_all(l->buf, &size);
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (data == NULL) {
|
|
Packit |
4e8bc4 |
return 0;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Got %d bytes from bipbuffer\n", size);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* parse buffer */
|
|
Packit |
4e8bc4 |
while (pos < size && watcher_count > 0) {
|
|
Packit |
4e8bc4 |
enum logger_parse_entry_ret ret;
|
|
Packit |
4e8bc4 |
int scratch_len = 0;
|
|
Packit |
4e8bc4 |
e = (logentry *) (data + pos);
|
|
Packit |
4e8bc4 |
ret = logger_thread_parse_entry(e, ls, scratch, &scratch_len);
|
|
Packit |
4e8bc4 |
if (ret != LOGGER_PARSE_ENTRY_OK) {
|
|
Packit |
4e8bc4 |
/* TODO: stats counter */
|
|
Packit |
4e8bc4 |
fprintf(stderr, "LOGGER: Failed to parse log entry\n");
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
logger_thread_write_entry(e, ls, scratch, scratch_len);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
pos += sizeof(logentry) + e->size + e->pad;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
assert(pos <= size);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&l->mutex);
|
|
Packit |
4e8bc4 |
data = bipbuf_poll(l->buf, size);
|
|
Packit |
4e8bc4 |
ls->worker_written += l->written;
|
|
Packit |
4e8bc4 |
ls->worker_dropped += l->dropped;
|
|
Packit |
4e8bc4 |
l->written = 0;
|
|
Packit |
4e8bc4 |
l->dropped = 0;
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
if (data == NULL) {
|
|
Packit |
4e8bc4 |
fprintf(stderr, "LOGGER: unexpectedly couldn't advance buf pointer\n");
|
|
Packit |
4e8bc4 |
assert(0);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
return size; /* maybe the count of objects iterated? */
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Since the event loop code isn't reusable without a refactor, and we have a
|
|
Packit |
4e8bc4 |
* limited number of potential watchers, we run our own poll loop.
|
|
Packit |
4e8bc4 |
* This calls poll() unnecessarily during write flushes, should be possible to
|
|
Packit |
4e8bc4 |
* micro-optimize later.
|
|
Packit |
4e8bc4 |
*
|
|
Packit |
4e8bc4 |
* This flushes buffers attached to watchers, iterating through the bytes set
|
|
Packit |
4e8bc4 |
* to each worker. Also checks for readability in case client connection was
|
|
Packit |
4e8bc4 |
* closed.
|
|
Packit |
4e8bc4 |
*
|
|
Packit |
4e8bc4 |
* Allows a specific watcher to be flushed (if buf full)
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static int logger_thread_poll_watchers(int force_poll, int watcher) {
|
|
Packit |
4e8bc4 |
int x;
|
|
Packit |
4e8bc4 |
int nfd = 0;
|
|
Packit |
4e8bc4 |
unsigned char *data;
|
|
Packit |
4e8bc4 |
unsigned int data_size = 0;
|
|
Packit |
4e8bc4 |
int flushed = 0;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
for (x = 0; x < WATCHER_LIMIT; x++) {
|
|
Packit |
4e8bc4 |
logger_watcher *w = watchers[x];
|
|
Packit |
4e8bc4 |
if (w == NULL || (watcher != WATCHER_ALL && x != watcher))
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
data = bipbuf_peek_all(w->buf, &data_size);
|
|
Packit |
4e8bc4 |
if (data != NULL) {
|
|
Packit |
4e8bc4 |
watchers_pollfds[nfd].fd = w->sfd;
|
|
Packit |
4e8bc4 |
watchers_pollfds[nfd].events = POLLOUT;
|
|
Packit |
4e8bc4 |
nfd++;
|
|
Packit |
4e8bc4 |
} else if (force_poll) {
|
|
Packit |
4e8bc4 |
watchers_pollfds[nfd].fd = w->sfd;
|
|
Packit |
4e8bc4 |
watchers_pollfds[nfd].events = POLLIN;
|
|
Packit |
4e8bc4 |
nfd++;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
/* This gets set after a call to poll, and should be used to gate on
|
|
Packit |
4e8bc4 |
* calling poll again.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
w->failed_flush = false;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (nfd == 0)
|
|
Packit |
4e8bc4 |
return 0;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
//L_DEBUG("LOGGER: calling poll() [data_size: %d]\n", data_size);
|
|
Packit |
4e8bc4 |
int ret = poll(watchers_pollfds, nfd, 0);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (ret < 0) {
|
|
Packit |
4e8bc4 |
perror("something failed with logger thread watcher fd polling");
|
|
Packit |
4e8bc4 |
return -1;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
nfd = 0;
|
|
Packit |
4e8bc4 |
for (x = 0; x < WATCHER_LIMIT; x++) {
|
|
Packit |
4e8bc4 |
logger_watcher *w = watchers[x];
|
|
Packit |
4e8bc4 |
if (w == NULL)
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
data_size = 0;
|
|
Packit |
4e8bc4 |
/* Early detection of a disconnect. Otherwise we have to wait until
|
|
Packit |
4e8bc4 |
* the next write
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
if (watchers_pollfds[nfd].revents & POLLIN) {
|
|
Packit |
4e8bc4 |
char buf[1];
|
|
Packit |
4e8bc4 |
int res = ((conn*)w->c)->read(w->c, buf, 1);
|
|
Packit |
4e8bc4 |
if (res == 0 || (res == -1 && (errno != EAGAIN && errno != EWOULDBLOCK))) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: watcher closed remotely\n");
|
|
Packit |
4e8bc4 |
logger_thread_close_watcher(w);
|
|
Packit |
4e8bc4 |
nfd++;
|
|
Packit |
4e8bc4 |
continue;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
if ((data = bipbuf_peek_all(w->buf, &data_size)) != NULL) {
|
|
Packit |
4e8bc4 |
if (watchers_pollfds[nfd].revents & (POLLHUP|POLLERR)) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: watcher closed during poll() call\n");
|
|
Packit |
4e8bc4 |
logger_thread_close_watcher(w);
|
|
Packit |
4e8bc4 |
} else if (watchers_pollfds[nfd].revents & POLLOUT) {
|
|
Packit |
4e8bc4 |
int total = 0;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* We can write a bit. */
|
|
Packit |
4e8bc4 |
switch (w->t) {
|
|
Packit |
4e8bc4 |
case LOGGER_WATCHER_STDERR:
|
|
Packit |
4e8bc4 |
total = fwrite(data, 1, data_size, stderr);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
case LOGGER_WATCHER_CLIENT:
|
|
Packit |
4e8bc4 |
total = ((conn*)w->c)->write(w->c, data, data_size);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: poll() wrote %d to %d (data_size: %d) (bipbuf_used: %d)\n", total, w->sfd,
|
|
Packit |
4e8bc4 |
data_size, bipbuf_used(w->buf));
|
|
Packit |
4e8bc4 |
if (total == -1) {
|
|
Packit |
4e8bc4 |
if (errno != EAGAIN && errno != EWOULDBLOCK) {
|
|
Packit |
4e8bc4 |
logger_thread_close_watcher(w);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: watcher hit EAGAIN\n");
|
|
Packit |
4e8bc4 |
} else if (total == 0) {
|
|
Packit |
4e8bc4 |
logger_thread_close_watcher(w);
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
bipbuf_poll(w->buf, total);
|
|
Packit |
4e8bc4 |
flushed += total;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
nfd++;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
return flushed;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static void logger_thread_sum_stats(struct logger_stats *ls) {
|
|
Packit |
4e8bc4 |
STATS_LOCK();
|
|
Packit |
4e8bc4 |
stats.log_worker_dropped += ls->worker_dropped;
|
|
Packit |
4e8bc4 |
stats.log_worker_written += ls->worker_written;
|
|
Packit |
4e8bc4 |
stats.log_watcher_skipped += ls->watcher_skipped;
|
|
Packit |
4e8bc4 |
stats.log_watcher_sent += ls->watcher_sent;
|
|
Packit |
4e8bc4 |
STATS_UNLOCK();
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#define MAX_LOGGER_SLEEP 1000000
|
|
Packit |
4e8bc4 |
#define MIN_LOGGER_SLEEP 1000
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Primary logger thread routine */
|
|
Packit |
4e8bc4 |
static void *logger_thread(void *arg) {
|
|
Packit |
4e8bc4 |
useconds_t to_sleep = MIN_LOGGER_SLEEP;
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Starting logger thread\n");
|
|
Packit |
4e8bc4 |
// TODO: If we ever have item references in the logger code, will need to
|
|
Packit |
4e8bc4 |
// ensure everything is dequeued before stopping the thread.
|
|
Packit |
4e8bc4 |
while (do_run_logger_thread) {
|
|
Packit |
4e8bc4 |
int found_logs = 0;
|
|
Packit |
4e8bc4 |
logger *l;
|
|
Packit |
4e8bc4 |
struct logger_stats ls;
|
|
Packit |
4e8bc4 |
memset(&ls, 0, sizeof(struct logger_stats));
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* only sleep if we're *above* the minimum */
|
|
Packit |
4e8bc4 |
if (to_sleep > MIN_LOGGER_SLEEP)
|
|
Packit |
4e8bc4 |
usleep(to_sleep);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Call function to iterate each logger. */
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
for (l = logger_stack_head; l != NULL; l=l->next) {
|
|
Packit |
4e8bc4 |
/* lock logger, call function to manipulate it */
|
|
Packit |
4e8bc4 |
found_logs += logger_thread_read(l, &ls);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
logger_thread_poll_watchers(1, WATCHER_ALL);
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* TODO: abstract into a function and share with lru_crawler */
|
|
Packit |
4e8bc4 |
if (!found_logs) {
|
|
Packit |
4e8bc4 |
if (to_sleep < MAX_LOGGER_SLEEP)
|
|
Packit |
4e8bc4 |
to_sleep += to_sleep / 8;
|
|
Packit |
4e8bc4 |
if (to_sleep > MAX_LOGGER_SLEEP)
|
|
Packit |
4e8bc4 |
to_sleep = MAX_LOGGER_SLEEP;
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
to_sleep /= 2;
|
|
Packit |
4e8bc4 |
if (to_sleep < MIN_LOGGER_SLEEP)
|
|
Packit |
4e8bc4 |
to_sleep = MIN_LOGGER_SLEEP;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
logger_thread_sum_stats(&ls);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
return NULL;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static int start_logger_thread(void) {
|
|
Packit |
4e8bc4 |
int ret;
|
|
Packit |
4e8bc4 |
do_run_logger_thread = 1;
|
|
Packit |
4e8bc4 |
if ((ret = pthread_create(&logger_tid, NULL,
|
|
Packit |
4e8bc4 |
logger_thread, NULL)) != 0) {
|
|
Packit |
4e8bc4 |
fprintf(stderr, "Can't start logger thread: %s\n", strerror(ret));
|
|
Packit |
4e8bc4 |
return -1;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
return 0;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static int stop_logger_thread(void) {
|
|
Packit |
4e8bc4 |
do_run_logger_thread = 0;
|
|
Packit |
4e8bc4 |
pthread_join(logger_tid, NULL);
|
|
Packit |
4e8bc4 |
return 0;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/*************************
|
|
Packit |
4e8bc4 |
* Public functions for submitting logs and starting loggers from workers.
|
|
Packit |
4e8bc4 |
*************************/
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Global logger thread start/init */
|
|
Packit |
4e8bc4 |
void logger_init(void) {
|
|
Packit |
4e8bc4 |
/* TODO: auto destructor when threads exit */
|
|
Packit |
4e8bc4 |
/* TODO: error handling */
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* init stack for iterating loggers */
|
|
Packit |
4e8bc4 |
logger_stack_head = 0;
|
|
Packit |
4e8bc4 |
logger_stack_tail = 0;
|
|
Packit |
4e8bc4 |
pthread_key_create(&logger_key, NULL);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (start_logger_thread() != 0) {
|
|
Packit |
4e8bc4 |
abort();
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* This is what adding a STDERR watcher looks like. should replace old
|
|
Packit |
4e8bc4 |
* "verbose" settings. */
|
|
Packit |
4e8bc4 |
//logger_add_watcher(NULL, 0);
|
|
Packit |
4e8bc4 |
return;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
void logger_stop(void) {
|
|
Packit |
4e8bc4 |
stop_logger_thread();
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* called *from* the thread using a logger.
|
|
Packit |
4e8bc4 |
* initializes the per-thread bipbuf, links it into the list of loggers
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
logger *logger_create(void) {
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Creating and linking new logger instance\n");
|
|
Packit |
4e8bc4 |
logger *l = calloc(1, sizeof(logger));
|
|
Packit |
4e8bc4 |
if (l == NULL) {
|
|
Packit |
4e8bc4 |
return NULL;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
l->buf = bipbuf_new(settings.logger_buf_size);
|
|
Packit |
4e8bc4 |
if (l->buf == NULL) {
|
|
Packit |
4e8bc4 |
free(l);
|
|
Packit |
4e8bc4 |
return NULL;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
l->entry_map = default_entries;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_mutex_init(&l->mutex, NULL);
|
|
Packit |
4e8bc4 |
pthread_setspecific(logger_key, l);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* add to list of loggers */
|
|
Packit |
4e8bc4 |
logger_link_q(l);
|
|
Packit |
4e8bc4 |
return l;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* helpers for logger_log */
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static void _logger_log_evictions(logentry *e, item *it) {
|
|
Packit |
4e8bc4 |
struct logentry_eviction *le = (struct logentry_eviction *) e->data;
|
|
Packit |
4e8bc4 |
le->exptime = (it->exptime > 0) ? (long long int)(it->exptime - current_time) : (long long int) -1;
|
|
Packit |
4e8bc4 |
le->latime = current_time - it->time;
|
|
Packit |
4e8bc4 |
le->it_flags = it->it_flags;
|
|
Packit |
4e8bc4 |
le->nkey = it->nkey;
|
|
Packit |
4e8bc4 |
le->clsid = ITEM_clsid(it);
|
|
Packit |
4e8bc4 |
memcpy(le->key, ITEM_key(it), it->nkey);
|
|
Packit |
4e8bc4 |
e->size = sizeof(struct logentry_eviction) + le->nkey;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
#ifdef EXTSTORE
|
|
Packit |
4e8bc4 |
/* TODO: When more logging endpoints are done and the extstore API has matured
|
|
Packit |
4e8bc4 |
* more, this could be merged with above and print different types of
|
|
Packit |
4e8bc4 |
* expulsion events.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static void _logger_log_ext_write(logentry *e, item *it, uint8_t bucket) {
|
|
Packit |
4e8bc4 |
struct logentry_ext_write *le = (struct logentry_ext_write *) e->data;
|
|
Packit |
4e8bc4 |
le->exptime = (it->exptime > 0) ? (long long int)(it->exptime - current_time) : (long long int) -1;
|
|
Packit |
4e8bc4 |
le->latime = current_time - it->time;
|
|
Packit |
4e8bc4 |
le->it_flags = it->it_flags;
|
|
Packit |
4e8bc4 |
le->nkey = it->nkey;
|
|
Packit |
4e8bc4 |
le->clsid = ITEM_clsid(it);
|
|
Packit |
4e8bc4 |
le->bucket = bucket;
|
|
Packit |
4e8bc4 |
memcpy(le->key, ITEM_key(it), it->nkey);
|
|
Packit |
4e8bc4 |
e->size = sizeof(struct logentry_ext_write) + le->nkey;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
/* 0 == nf, 1 == found. 2 == flushed. 3 == expired.
|
|
Packit |
4e8bc4 |
* might be useful to store/print the flags an item has?
|
|
Packit |
4e8bc4 |
* could also collapse this and above code into an "item status" struct. wait
|
|
Packit |
4e8bc4 |
* for more endpoints to be written before making it generic, though.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
static void _logger_log_item_get(logentry *e, const int was_found, const char *key,
|
|
Packit |
4e8bc4 |
const int nkey, const uint8_t clsid, const int sfd) {
|
|
Packit |
4e8bc4 |
struct logentry_item_get *le = (struct logentry_item_get *) e->data;
|
|
Packit |
4e8bc4 |
le->was_found = was_found;
|
|
Packit |
4e8bc4 |
le->nkey = nkey;
|
|
Packit |
4e8bc4 |
le->clsid = clsid;
|
|
Packit |
4e8bc4 |
memcpy(le->key, key, nkey);
|
|
Packit |
4e8bc4 |
le->sfd = sfd;
|
|
Packit |
4e8bc4 |
e->size = sizeof(struct logentry_item_get) + nkey;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
static void _logger_log_item_store(logentry *e, const enum store_item_type status,
|
|
Packit |
4e8bc4 |
const int comm, char *key, const int nkey, rel_time_t ttl, const uint8_t clsid, int sfd) {
|
|
Packit |
4e8bc4 |
struct logentry_item_store *le = (struct logentry_item_store *) e->data;
|
|
Packit |
4e8bc4 |
le->status = status;
|
|
Packit |
4e8bc4 |
le->cmd = comm;
|
|
Packit |
4e8bc4 |
le->nkey = nkey;
|
|
Packit |
4e8bc4 |
le->clsid = clsid;
|
|
Packit |
4e8bc4 |
if (ttl != 0) {
|
|
Packit |
4e8bc4 |
le->ttl = ttl - current_time;
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
le->ttl = 0;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
memcpy(le->key, key, nkey);
|
|
Packit |
4e8bc4 |
le->sfd = sfd;
|
|
Packit |
4e8bc4 |
e->size = sizeof(struct logentry_item_store) + nkey;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Public function for logging an entry.
|
|
Packit |
4e8bc4 |
* Tries to encapsulate as much of the formatting as possible to simplify the
|
|
Packit |
4e8bc4 |
* caller's code.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
enum logger_ret_type logger_log(logger *l, const enum log_entry_type event, const void *entry, ...) {
|
|
Packit |
4e8bc4 |
bipbuf_t *buf = l->buf;
|
|
Packit |
4e8bc4 |
bool nospace = false;
|
|
Packit |
4e8bc4 |
va_list ap;
|
|
Packit |
4e8bc4 |
int total = 0;
|
|
Packit |
4e8bc4 |
logentry *e;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
const entry_details *d = &l->entry_map[event];
|
|
Packit |
4e8bc4 |
int reqlen = d->reqlen;
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&l->mutex);
|
|
Packit |
4e8bc4 |
/* Request a maximum length of data to write to */
|
|
Packit |
4e8bc4 |
e = (logentry *) bipbuf_request(buf, (sizeof(logentry) + reqlen));
|
|
Packit |
4e8bc4 |
if (e == NULL) {
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
l->dropped++;
|
|
Packit |
4e8bc4 |
return LOGGER_RET_NOSPACE;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
e->event = d->subtype;
|
|
Packit |
4e8bc4 |
e->pad = 0;
|
|
Packit |
4e8bc4 |
e->gid = logger_get_gid();
|
|
Packit |
4e8bc4 |
/* TODO: Could pass this down as an argument now that we're using
|
|
Packit |
4e8bc4 |
* LOGGER_LOG() macro.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
e->eflags = d->eflags;
|
|
Packit |
4e8bc4 |
/* Noting time isn't optional. A feature may be added to avoid rendering
|
|
Packit |
4e8bc4 |
* time and/or gid to a logger.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
gettimeofday(&e->tv, NULL);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
switch (d->subtype) {
|
|
Packit |
4e8bc4 |
case LOGGER_TEXT_ENTRY:
|
|
Packit |
4e8bc4 |
va_start(ap, entry);
|
|
Packit |
4e8bc4 |
total = vsnprintf((char *) e->data, reqlen, d->format, ap);
|
|
Packit |
4e8bc4 |
va_end(ap);
|
|
Packit |
4e8bc4 |
if (total >= reqlen || total <= 0) {
|
|
Packit |
4e8bc4 |
fprintf(stderr, "LOGGER: Failed to vsnprintf a text entry: (total) %d\n", total);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
e->size = total + 1; /* null byte */
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
case LOGGER_EVICTION_ENTRY:
|
|
Packit |
4e8bc4 |
_logger_log_evictions(e, (item *)entry);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
#ifdef EXTSTORE
|
|
Packit |
4e8bc4 |
case LOGGER_EXT_WRITE_ENTRY:
|
|
Packit |
4e8bc4 |
va_start(ap, entry);
|
|
Packit |
4e8bc4 |
int ew_bucket = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
va_end(ap);
|
|
Packit |
4e8bc4 |
_logger_log_ext_write(e, (item *)entry, ew_bucket);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
case LOGGER_ITEM_GET_ENTRY:
|
|
Packit |
4e8bc4 |
va_start(ap, entry);
|
|
Packit |
4e8bc4 |
int was_found = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
char *key = va_arg(ap, char *);
|
|
Packit |
4e8bc4 |
size_t nkey = va_arg(ap, size_t);
|
|
Packit |
4e8bc4 |
uint8_t gclsid = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
int gsfd = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
_logger_log_item_get(e, was_found, key, nkey, gclsid, gsfd);
|
|
Packit |
4e8bc4 |
va_end(ap);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
case LOGGER_ITEM_STORE_ENTRY:
|
|
Packit |
4e8bc4 |
va_start(ap, entry);
|
|
Packit |
4e8bc4 |
enum store_item_type status = va_arg(ap, enum store_item_type);
|
|
Packit |
4e8bc4 |
int comm = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
char *skey = va_arg(ap, char *);
|
|
Packit |
4e8bc4 |
size_t snkey = va_arg(ap, size_t);
|
|
Packit |
4e8bc4 |
rel_time_t sttl = va_arg(ap, rel_time_t);
|
|
Packit |
4e8bc4 |
uint8_t sclsid = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
int ssfd = va_arg(ap, int);
|
|
Packit |
4e8bc4 |
_logger_log_item_store(e, status, comm, skey, snkey, sttl, sclsid, ssfd);
|
|
Packit |
4e8bc4 |
va_end(ap);
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
#ifdef NEED_ALIGN
|
|
Packit |
4e8bc4 |
/* Need to ensure *next* request is aligned. */
|
|
Packit |
4e8bc4 |
if (sizeof(logentry) + e->size % 8 != 0) {
|
|
Packit |
4e8bc4 |
e->pad = 8 - (sizeof(logentry) + e->size % 8);
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
#endif
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Push pointer forward by the actual amount required */
|
|
Packit |
4e8bc4 |
if (bipbuf_push(buf, (sizeof(logentry) + e->size + e->pad)) == 0) {
|
|
Packit |
4e8bc4 |
fprintf(stderr, "LOGGER: Failed to bipbuf push a text entry\n");
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
return LOGGER_RET_ERR;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
l->written++;
|
|
Packit |
4e8bc4 |
L_DEBUG("LOGGER: Requested %d bytes, wrote %lu bytes\n", reqlen,
|
|
Packit |
4e8bc4 |
(sizeof(logentry) + e->size));
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&l->mutex);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
if (nospace) {
|
|
Packit |
4e8bc4 |
return LOGGER_RET_NOSPACE;
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
return LOGGER_RET_OK;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
/* Passes a client connection socket from a primary worker thread to the
|
|
Packit |
4e8bc4 |
* logger thread. Caller *must* event_del() the client before handing it over.
|
|
Packit |
4e8bc4 |
* Presently there's no way to hand the client back to the worker thread.
|
|
Packit |
4e8bc4 |
*/
|
|
Packit |
4e8bc4 |
enum logger_add_watcher_ret logger_add_watcher(void *c, const int sfd, uint16_t f) {
|
|
Packit |
4e8bc4 |
int x;
|
|
Packit |
4e8bc4 |
logger_watcher *w = NULL;
|
|
Packit |
4e8bc4 |
pthread_mutex_lock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
if (watcher_count >= WATCHER_LIMIT) {
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return LOGGER_ADD_WATCHER_TOO_MANY;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
for (x = 0; x < WATCHER_LIMIT-1; x++) {
|
|
Packit |
4e8bc4 |
if (watchers[x] == NULL)
|
|
Packit |
4e8bc4 |
break;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
w = calloc(1, sizeof(logger_watcher));
|
|
Packit |
4e8bc4 |
if (w == NULL) {
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return LOGGER_ADD_WATCHER_FAILED;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
w->c = c;
|
|
Packit |
4e8bc4 |
w->sfd = sfd;
|
|
Packit |
4e8bc4 |
if (sfd == 0 && c == NULL) {
|
|
Packit |
4e8bc4 |
w->t = LOGGER_WATCHER_STDERR;
|
|
Packit |
4e8bc4 |
} else {
|
|
Packit |
4e8bc4 |
w->t = LOGGER_WATCHER_CLIENT;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
w->id = x;
|
|
Packit |
4e8bc4 |
w->eflags = f;
|
|
Packit |
4e8bc4 |
w->buf = bipbuf_new(settings.logger_watcher_buf_size);
|
|
Packit |
4e8bc4 |
if (w->buf == NULL) {
|
|
Packit |
4e8bc4 |
free(w);
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return LOGGER_ADD_WATCHER_FAILED;
|
|
Packit |
4e8bc4 |
}
|
|
Packit |
4e8bc4 |
bipbuf_offer(w->buf, (unsigned char *) "OK\r\n", 4);
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
watchers[x] = w;
|
|
Packit |
4e8bc4 |
watcher_count++;
|
|
Packit |
4e8bc4 |
/* Update what flags the global logs will watch */
|
|
Packit |
4e8bc4 |
logger_set_flags();
|
|
Packit |
4e8bc4 |
|
|
Packit |
4e8bc4 |
pthread_mutex_unlock(&logger_stack_lock);
|
|
Packit |
4e8bc4 |
return LOGGER_ADD_WATCHER_OK;
|
|
Packit |
4e8bc4 |
}
|