/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * 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 "core/conf.h"
#include "core/htrace.h"
#include "core/htracer.h"
#include "core/span.h"
#include "receiver/hrpc.h"
#include "receiver/receiver.h"
#include "test/test.h"
#include "util/cmp.h"
#include "util/cmp_util.h"
#include "util/log.h"
#include "util/string.h"
#include "util/time.h"

#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

/**
 * @file htraced.c
 *
 * The htraced span receiver which implements sending spans on to the htraced
 * daemon.
 *
 * We send spans via the HRPC protocol.  HRPC consists of a simple fixed-size
 * header specifying a magic number, the length, and a message type, followed by
 * data in the msgpack serialization format.  The message type will determine
 * the type of msgpack data.  Messages bodies are sent as maps, essentially
 * making all fields optional and allowing the protocol to evolve.  See hrpc.c
 * and rpc.go for the implementation of HRPC.
 *
 * Spans are serialized immediately when they are added to the buffer.  This is
 * one of the advantages of msgpack-- it has a good streaming interface.  We do
 * not need to keep around the span objects after htraced_rcv_add_span.
 *
 * The htraced receiver keeps two equally sized buffers around internally.
 * While we are writing spans to one buffer, we can be sending the data from the
 * other buffer over the wire.  The intention here is to avoid copies as much as
 * possible.  In general, what we send over the wire is exactly what is in the
 * buffer, except that we have to add a short "prequel" to it containing the
 * other WriteSpansReq fields.
 *
 * Note that we may change the serialization in the future if we discover better
 * alternatives.  Sending spans over HTTP as JSON will always be supported
 * as a fallback.
 */

/**
 * The maximum length of the message we will send to the server.
 * This must be the same or shorter than MAX_HRPC_BODY_LENGTH in rpc.go.
 */
#define MAX_HRPC_LEN (32ULL * 1024ULL * 1024ULL)

/**
 * The maximum length of the prequel in a WriteSpans message.
 */
#define MAX_WRITESPANS_PREQUEL_LEN 1024

/**
 * The maximum length of the span data in a WriteSpans message.
 */
#define MAX_SPAN_DATA_LEN (MAX_HRPC_LEN - MAX_WRITESPANS_PREQUEL_LEN)

/**
 * The minimum total buffer size to allow.
 *
 * This should allow at least a few spans to be buffered.
 */
#define HTRACED_MIN_BUFFER_SIZE (4ULL * 1024ULL * 1024ULL)

/**
 * The maximum total buffer size to allow.
 */
#define HTRACED_MAX_BUFFER_SIZE (2ULL * MAX_SPAN_DATA_LEN)

/**
 * The minimum number of milliseconds to allow for flush_interval_ms.
 */
#define HTRACED_FLUSH_INTERVAL_MS_MIN 30000LL

/**
 * The maximum number of milliseconds to allow for flush_interval_ms.
 * This is mainly to avoid overflow.
 */
#define HTRACED_FLUSH_INTERVAL_MS_MAX 86400000LL

/**
 * The minimum number of milliseconds to allow for tcp write timeouts.
 */
#define HTRACED_WRITE_TIMEO_MS_MIN 50LL

/**
 * The minimum number of milliseconds to allow for tcp read timeouts.
 */
#define HTRACED_READ_TIMEO_MS_MIN 50LL

/**
 * The maximum number of times we will try to add a span to the circular buffer
 * before giving up.
 */
#define HTRACED_MAX_ADD_TRIES 3

/**
 * The maximum number of times to try to send some spans to the htraced daemon
 * before giving up.
 */
#define HTRACED_MAX_SEND_TRIES 3

/**
 * The number of milliseconds to sleep after failing to send some spans to the
 * htraced daemon.
 */
#define HTRACED_SEND_RETRY_SLEEP_MS 5000

/**
 * The number of buffers used by htraced.
 */
#define HTRACED_NUM_BUFS 2

/**
 * An HTraced send buffer.
 */
struct htraced_sbuf {
    /**
     * Current offset within the buffer.
     */
    uint64_t off;

    /**
     * Length of the buffer.
     */
    uint64_t len;

    /**
     * The number of spans in the buffer.
     */
    uint64_t num_spans;

    /**
     * The buffer data.  This field actually has size 'len,' not size 1.
     */
    char buf[1];
};

/*
 * A span receiver that writes spans to htraced.
 */
struct htraced_rcv {
    struct htrace_rcv base;

    /**
     * Nonzero if the receiver should shut down.
     */
    int shutdown;

    /**
     * The htracer object associated with this receciver.
     */
    struct htracer *tracer;

    /**
     * Buffered span data becomes eligible to be sent even if there isn't much
     * in the buffer after this timeout elapses.
     */
    uint64_t flush_interval_ms;

    /**
     * The maximum number of bytes we will buffer before waking the sending
     * thread.  We may sometimes send slightly more than this amount if the
     * thread takes a while to wake up.
     */
    uint64_t send_threshold;

    /**
     * The HRPC client.
     */
    struct hrpc_client *hcli;

    /**
     * The monotonic-clock time at which we last did a send operation.
     */
    uint64_t last_send_ms;

    /**
     * The index of the active buffer.
     */
    int active_buf;

    /**
     * The two send buffers.
     */
    struct htraced_sbuf *sbuf[HTRACED_NUM_BUFS];

    /**
     * Lock protecting the buffers from concurrent writes.
     */
    pthread_mutex_t lock;

    /**
     * Condition variable used to wake up the background thread.
     */
    pthread_cond_t bg_cond;

    /**
     * Condition variable used to wake up flushing threads.
     */
    pthread_cond_t flush_cond;

    /**
     * Background transmitter thread.
     */
    pthread_t xmit_thread;
};

void* run_htraced_xmit_manager(void *data);
static int should_xmit(struct htraced_rcv *rcv, uint64_t now);
static void htraced_xmit(struct htraced_rcv *rcv, uint64_t now);

static int htraced_sbufs_empty(struct htraced_rcv *rcv)
{
    int i;
    for (i = 0; i < HTRACED_NUM_BUFS; i++) {
        if (rcv->sbuf[i]->off) {
            return 0;
        }
    }
    return 1;
}

static struct htraced_sbuf *htraced_sbuf_alloc(uint64_t len)
{
    struct htraced_sbuf *sbuf;

    // The final field of the htraced_sbuf structure is declared as having size
    // 1, but really it has size 'len'.  This avoids a pointer dereference when
    // accessing data in the sbuf.
    sbuf = malloc(offsetof(struct htraced_sbuf, buf) + len);
    if (!sbuf) {
        return NULL;
    }
    sbuf->off = 0;
    sbuf->len = len;
    sbuf->num_spans = 0;
    return sbuf;
}

static void htraced_sbuf_free(struct htraced_sbuf *sbuf)
{
    free(sbuf);
}

static uint64_t htraced_sbuf_remaining(const struct htraced_sbuf *sbuf)
{
    return sbuf->len - sbuf->off;
}

static uint64_t htraced_get_bounded_u64(struct htrace_log *lg,
                const struct htrace_conf *cnf, const char *prop,
                uint64_t min, uint64_t max)
{
    uint64_t val = htrace_conf_get_u64(lg, cnf, prop);
    if (val < min) {
        htrace_log(lg, "htraced_rcv_create: can't set %s to %"PRId64
                   ".  Using minimum value of %"PRId64 " instead.\n",
                   prop, val, min);
        return min;
    } else if (val > max) {
        htrace_log(lg, "htraced_rcv_create: can't set %s to %"PRId64
                   ".  Using maximum value of %"PRId64 " instead.\n",
                   prop, val, max);
        return max;
    }
    return val;
}

static double htraced_get_bounded_double(struct htrace_log *lg,
                const struct htrace_conf *cnf, const char *prop,
                double min, double max)
{
    double val = htrace_conf_get_double(lg, cnf, prop);
    if (val < min) {
        htrace_log(lg, "htraced_rcv_create: can't set %s to %g"
                   ".  Using minimum value of %g instead.\n",
                   prop, val, min);
        return min;
    } else if (val > max) {
        htrace_log(lg, "htraced_rcv_create: can't set %s to %g"
                   ".  Using maximum value of %g instead.\n",
                   prop, val, max);
        return max;
    }
    return val;
}

static struct htrace_rcv *htraced_rcv_create(struct htracer *tracer,
                                             const struct htrace_conf *conf)
{
    struct htraced_rcv *rcv;
    const char *endpoint;
    int i, ret;
    uint64_t write_timeo_ms, read_timeo_ms, buf_len;
    double send_fraction;

    endpoint = htrace_conf_get(conf, HTRACED_ADDRESS_KEY);
    if (!endpoint) {
        htrace_log(tracer->lg, "htraced_rcv_create: no value found for %s. "
                   "You must set this configuration key to the "
                   "hostname:port identifying the htraced server.\n",
                   HTRACED_ADDRESS_KEY);
        goto error;
    }
    rcv = calloc(1, sizeof(*rcv));
    if (!rcv) {
        htrace_log(tracer->lg, "htraced_rcv_create: OOM while "
                   "allocating htraced_rcv.\n");
        goto error;
    }
    rcv->base.ty = &g_htraced_rcv_ty;
    rcv->shutdown = 0;
    rcv->tracer = tracer;

    rcv->flush_interval_ms = htraced_get_bounded_u64(tracer->lg, conf,
                HTRACED_FLUSH_INTERVAL_MS_KEY, HTRACED_FLUSH_INTERVAL_MS_MIN,
                HTRACED_FLUSH_INTERVAL_MS_MAX);
    write_timeo_ms = htraced_get_bounded_u64(tracer->lg, conf,
                HTRACED_WRITE_TIMEO_MS_KEY, HTRACED_WRITE_TIMEO_MS_MIN,
                0x7fffffffffffffffULL);
    read_timeo_ms = htraced_get_bounded_u64(tracer->lg, conf,
                HTRACED_READ_TIMEO_MS_KEY, HTRACED_READ_TIMEO_MS_MIN,
                0x7fffffffffffffffULL);
    rcv->hcli = hrpc_client_alloc(tracer->lg, write_timeo_ms,
                                  read_timeo_ms, endpoint);
    if (!rcv->hcli) {
        goto error_free_rcv;
    }
    buf_len = htraced_get_bounded_u64(tracer->lg, conf,
                HTRACED_BUFFER_SIZE_KEY, HTRACED_MIN_BUFFER_SIZE,
                HTRACED_MAX_BUFFER_SIZE) / 2;
    for (i = 0; i < HTRACED_NUM_BUFS; i++) {
        rcv->sbuf[i] = htraced_sbuf_alloc(buf_len);
        if (!rcv->sbuf[i]) {
            htrace_log(tracer->lg, "htraced_rcv_create: htraced_sbuf_alloc("
                       "buf_len=%"PRId64") failed: OOM.\n", buf_len);
            goto error_free_bufs;
        }
    }
    send_fraction = htraced_get_bounded_double(tracer->lg, conf,
                HTRACED_BUFFER_SEND_TRIGGER_FRACTION, 0.1, 1.0);
    rcv->send_threshold = buf_len * send_fraction;
    if (rcv->send_threshold > buf_len) {
        rcv->send_threshold = buf_len;
    }
    rcv->last_send_ms = monotonic_now_ms(tracer->lg);
    ret = pthread_mutex_init(&rcv->lock, NULL);
    if (ret) {
        htrace_log(tracer->lg, "htraced_rcv_create: pthread_mutex_init "
                   "error %d: %s\n", ret, terror(ret));
        goto error_free_bufs;
    }
    ret = pthread_cond_init(&rcv->bg_cond, NULL);
    if (ret) {
        htrace_log(tracer->lg, "htraced_rcv_create: pthread_cond_init("
                   "bg_cond) error %d: %s\n", ret, terror(ret));
        goto error_free_lock;
    }
    ret = pthread_cond_init(&rcv->flush_cond, NULL);
    if (ret) {
        htrace_log(tracer->lg, "htraced_rcv_create: pthread_cond_init("
                   "flush_cond) error %d: %s\n", ret, terror(ret));
        goto error_free_bg_cond;
    }
    ret = pthread_create(&rcv->xmit_thread, NULL, run_htraced_xmit_manager, rcv);
    if (ret) {
        htrace_log(tracer->lg, "htraced_rcv_create: failed to create xmit thread: "
                   "error %d: %s\n", ret, terror(ret));
        goto error_free_flush_cond;
    }
    htrace_log(tracer->lg, "Initialized htraced receiver for %s"
                ", flush_interval_ms=%" PRId64 ", send_threshold=%" PRId64
                ", write_timeo_ms=%" PRId64 ", read_timeo_ms=%" PRId64
                ", buf_len=%" PRId64 ".\n", hrpc_client_get_endpoint(rcv->hcli),
                rcv->flush_interval_ms, rcv->send_threshold,
                write_timeo_ms, read_timeo_ms, buf_len);
    return (struct htrace_rcv*)rcv;

error_free_flush_cond:
    pthread_cond_destroy(&rcv->flush_cond);
error_free_bg_cond:
    pthread_cond_destroy(&rcv->bg_cond);
error_free_lock:
    pthread_mutex_destroy(&rcv->lock);
error_free_bufs:
    for (i = 0; i < HTRACED_NUM_BUFS; i++) {
        htraced_sbuf_free(rcv->sbuf[i]);
    }
    hrpc_client_free(rcv->hcli);
error_free_rcv:
    free(rcv);
error:
    return NULL;
}

void* run_htraced_xmit_manager(void *data)
{
    struct htraced_rcv *rcv = data;
    struct htrace_log *lg = rcv->tracer->lg;
    uint64_t now, wakeup;
    struct timespec wakeup_ts;
    int ret;

    pthread_mutex_lock(&rcv->lock);
    while (1) {
        now = monotonic_now_ms(lg);
        while (should_xmit(rcv, now)) {
            htraced_xmit(rcv, now);
        }
        if (rcv->shutdown) {
            while (!htraced_sbufs_empty(rcv)) {
                htraced_xmit(rcv, now);
            }
            break;
        }
        // Wait for one of a few things to happen:
        // * Shutdown
        // * The wakeup timer to elapse, leading us to check if we should send
        //      because of send_timeo_ms.
        // * A writer to signal that we should wake up because enough bytes are
        //      buffered.
        wakeup = now + (rcv->flush_interval_ms / 2);
        ms_to_timespec(wakeup, &wakeup_ts);
        ret = pthread_cond_timedwait(&rcv->bg_cond, &rcv->lock, &wakeup_ts);
        if ((ret != 0) && (ret != ETIMEDOUT)) {
            htrace_log(lg, "run_htraced_xmit_manager: pthread_cond_timedwait "
                       "error: %d (%s)\n", ret, terror(ret));
        }
    }
    pthread_mutex_unlock(&rcv->lock);
    htrace_log(lg, "run_htraced_xmit_manager: shutting down the transmission "
               "manager thread.\n");
    return NULL;
}

/**
 * Determine if the xmit manager should send.
 * This function must be called with the lock held.
 *
 * @param rcv           The htraced receiver.
 * @param now           The current time in milliseconds.
 *
 * @return              nonzero if we should send now.
 */
static int should_xmit(struct htraced_rcv *rcv, uint64_t now)
{
    uint64_t off = rcv->sbuf[rcv->active_buf]->off;

    if (off > rcv->send_threshold) {
        // We have buffered a lot of bytes, so let's send.
        return 1;
    }
    if (now - rcv->last_send_ms > rcv->flush_interval_ms) {
        // It's been too long since the last transmission, so let's send.
        if (off > 0) {
            return 1;
        }
    }
    return 0; // Let's wait.
}

#define DEFAULT_TRID_STR        "DefaultTrid"
#define DEFAULT_TRID_STR_LEN    (sizeof(DEFAULT_TRID_STR) - 1)
#define NUM_SPANS_STR               "NumSpans"
#define NUM_SPANS_STR_LEN           (sizeof(NUM_SPANS_STR) - 1)

/**
 * Write the prequel to the WriteSpans message.
 */
static int add_writespans_prequel(struct htraced_rcv *rcv,
                                  struct htraced_sbuf *sbuf, uint8_t *prequel)
{
    struct cmp_bcopy_ctx bctx;
    struct cmp_ctx_s *ctx =  (struct cmp_ctx_s *)&bctx;
    cmp_bcopy_ctx_init(&bctx, prequel, MAX_WRITESPANS_PREQUEL_LEN);
    if (!cmp_write_fixmap(ctx, 2)) {
        return -1;
    }
    if (!cmp_write_fixstr(ctx, DEFAULT_TRID_STR, DEFAULT_TRID_STR_LEN)) {
        return -1;
    }
    if (!cmp_write_str(ctx, rcv->tracer->trid, strlen(rcv->tracer->trid))) {
        return -1;
    }
    if (!cmp_write_fixstr(ctx, NUM_SPANS_STR, NUM_SPANS_STR_LEN)) {
        return -1;
    }
    if (!cmp_write_uint(ctx, sbuf->num_spans)) {
        return -1;
    }
    return bctx.off;
}

/**
 * Send all the spans which we have buffered.
 *
 * @param rcv           The htraced receiver.
 * @param sbuf          The span buffer to send.
 *
 * @return              1 on success; 0 otherwise.
 */
static int htraced_xmit_impl(struct htraced_rcv *rcv, struct htraced_sbuf *sbuf)
{
    struct htrace_log *lg = rcv->tracer->lg;
    uint8_t prequel[MAX_WRITESPANS_PREQUEL_LEN];
    int prequel_len, ret;
    char *err = NULL, *resp = NULL;
    size_t resp_len = 0;

    prequel_len = add_writespans_prequel(rcv, sbuf, prequel);
    if (prequel_len < 0) {
        htrace_log(lg, "htrace_xmit_impl: add_writespans_prequel failed.\n");
        ret = 0;
        goto done;
    }
    ret = hrpc_client_call(rcv->hcli, METHOD_ID_WRITE_SPANS,
                    prequel, prequel_len, sbuf->buf, sbuf->off,
                    &err, (void**)&resp, &resp_len);
    if (!ret) {
        htrace_log(lg, "htrace_xmit_impl: hrpc_client_call failed.\n");
        goto done;
    } else if (err) {
        htrace_log(lg, "htrace_xmit_impl: server returned error: %s\n", err);
        ret = 0;
        goto done;
    }
    ret = 1;
done:
    free(err);
    free(resp);
    return ret;
}

static void htraced_xmit(struct htraced_rcv *rcv, uint64_t now)
{
    int tries = 0;
    struct htraced_sbuf *sbuf;

    // Flip to the other buffer.
    sbuf = rcv->sbuf[rcv->active_buf];
    rcv->active_buf = !rcv->active_buf;

    // Release the lock while doing network I/O, so that we don't block threads
    // adding spans.
    pthread_mutex_unlock(&rcv->lock);
    while (1) {
        int retry, success = htraced_xmit_impl(rcv, sbuf);
        if (success) {
            break;
        }
        tries++;
        retry = (tries < HTRACED_MAX_SEND_TRIES);
        htrace_log(rcv->tracer->lg, "htraced_xmit(%s) failed on try %d.  %s\n",
                   hrpc_client_get_endpoint(rcv->hcli), tries,
                   (retry ? "Retrying after a delay." : "Giving up."));
        if (!retry) {
            break;
        }
    }
    sbuf->off = 0;
    sbuf->num_spans = 0;
    pthread_mutex_lock(&rcv->lock);
    rcv->last_send_ms = now;
    pthread_cond_broadcast(&rcv->flush_cond);
}

static void htraced_rcv_add_span(struct htrace_rcv *r,
                                 struct htrace_span *span)
{
    int tries, retry;
    uint64_t rem, off;
    struct htraced_rcv *rcv = (struct htraced_rcv *)r;
    struct htraced_sbuf *sbuf;
    struct htrace_log *lg = rcv->tracer->lg;
    struct cmp_counter_ctx cctx;
    struct cmp_bcopy_ctx bctx;
    uint64_t msgpack_len;

    // Determine the length of the span when serialized to msgpack.
    cmp_counter_ctx_init(&cctx);
    if (!span_write_msgpack(span, (cmp_ctx_t*)&cctx)) {
        htrace_log(lg, "htraced_rcv_add_span: span_write_msgpack failed.\n");
        return;
    }
    msgpack_len = cctx.count;

    // Try to get enough space in the current buffer.
    tries = 0;
    do {
        pthread_mutex_lock(&rcv->lock);
        sbuf = rcv->sbuf[rcv->active_buf];
        rem = htraced_sbuf_remaining(sbuf);
        if (rem < msgpack_len) {
            pthread_cond_signal(&rcv->bg_cond);
            pthread_mutex_unlock(&rcv->lock);
            tries++;
            retry = tries < HTRACED_MAX_ADD_TRIES;
            htrace_log(lg, "htraced_rcv_add_span: not enough space in the "
                           "current buffer.  Have %" PRId64 ", need %"
                           PRId64 ".  %s...\n", rem, msgpack_len,
                           (retry ? "Retrying" : "Giving up"));
            if (retry) {
                pthread_yield();
                continue;
            }
            pthread_mutex_unlock(&rcv->lock);
            return;
        }
    } while (0);
    // OK, now we have the lock, and we know that there is enough space in the
    // current buffer.
    off = sbuf->off;
    cmp_bcopy_ctx_init(&bctx, sbuf->buf + off, msgpack_len);
    bctx.base.write = cmp_bcopy_write_nocheck_fn;
    span_write_msgpack(span, (cmp_ctx_t*)&bctx);
    off += msgpack_len;
    sbuf->off = off;
    sbuf->num_spans++;
    if (off > rcv->send_threshold) {
        pthread_cond_signal(&rcv->bg_cond);
    }
    pthread_mutex_unlock(&rcv->lock);
}

static void htraced_rcv_flush(struct htrace_rcv *r)
{
    struct htraced_rcv *rcv = (struct htraced_rcv *)r;
    uint64_t now;

    // Note: This assumes that we only flush one buffer at once, and
    // that we flush buffers in order.  If we revisit those assumptions we'll
    // need to change this.
    // The SpanReceiver flush is only used for testing anyway.
    pthread_mutex_lock(&rcv->lock);
    now = monotonic_now_ms(rcv->tracer->lg);
    while (1) {
        if (rcv->last_send_ms >= now) {
            break;
        }
        if (htraced_sbufs_empty(rcv)) {
            break;
        }
        rcv->last_send_ms = 0;
        pthread_cond_wait(&rcv->flush_cond, &rcv->lock);
    }
    pthread_mutex_unlock(&rcv->lock);
}

static void htraced_rcv_free(struct htrace_rcv *r)
{
    struct htraced_rcv *rcv = (struct htraced_rcv *)r;
    struct htrace_log *lg;
    int i, ret;

    if (!rcv) {
        return;
    }
    lg = rcv->tracer->lg;
    htrace_log(lg, "Shutting down htraced receiver for %s\n",
               hrpc_client_get_endpoint(rcv->hcli));
    pthread_mutex_lock(&rcv->lock);
    rcv->shutdown = 1;
    pthread_cond_signal(&rcv->bg_cond);
    pthread_mutex_unlock(&rcv->lock);
    ret = pthread_join(rcv->xmit_thread, NULL);
    if (ret) {
        htrace_log(lg, "htraced_rcv_free: pthread_join "
                   "error %d: %s\n", ret, terror(ret));
    }
    for (i = 0; i < HTRACED_NUM_BUFS; i++) {
        htraced_sbuf_free(rcv->sbuf[i]);
    }
    hrpc_client_free(rcv->hcli);
    ret = pthread_mutex_destroy(&rcv->lock);
    if (ret) {
        htrace_log(lg, "htraced_rcv_free: pthread_mutex_destroy "
                   "error %d: %s\n", ret, terror(ret));
    }
    ret = pthread_cond_destroy(&rcv->bg_cond);
    if (ret) {
        htrace_log(lg, "htraced_rcv_free: pthread_cond_destroy(bg_cond) "
                   "error %d: %s\n", ret, terror(ret));
    }
    ret = pthread_cond_destroy(&rcv->flush_cond);
    if (ret) {
        htrace_log(lg, "htraced_rcv_free: pthread_cond_destroy(flush_cond) "
                   "error %d: %s\n", ret, terror(ret));
    }
    free(rcv);
}

const struct htrace_rcv_ty g_htraced_rcv_ty = {
    "htraced",
    htraced_rcv_create,
    htraced_rcv_add_span,
    htraced_rcv_flush,
    htraced_rcv_free,
};

// vim:ts=4:sw=4:et