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