<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2015 Couchbase, Inc
 *
 *   Licensed 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 SRC_TAPCONNECTION_H_
#define SRC_TAPCONNECTION_H_ 1

#include "config.h"

#include <list>
#include <map>
#include <queue>
#include <set>
#include <string>
#include <vector>

#include "atomic.h"
#include "common.h"
#include "locks.h"
#include "mutex.h"
#include "statwriter.h"

// forward decl
class ConnHandler;
class EventuallyPersistentEngine;
class TapConnMap;
class TapProducer;
class BGFetchCallback;
class CompleteBackfillOperation;
class Dispatcher;
class Item;
class VBucketFilter;

struct TapStatBuilder;
struct TapAggStatBuilder;
struct PopulateEventsBody;



#define MAX_TAP_KEEP_ALIVE 3600
#define MAX_TAKEOVER_TAP_LOG_SIZE 10
#define MINIMUM_BACKFILL_RESIDENT_THRESHOLD 0.7
#define DEFAULT_BACKFILL_RESIDENT_THRESHOLD 0.9

/**
 * A tap event that represents a change to the state of a vbucket.
 *
 * The tap stream may include other events than data mutation events,
 * but the data structures in the TapProducer does only store a key
 * for the item to store. We don't want to add more data to those elements,
 * because that could potentially consume a lot of memory (the tap queue
 * may have a lot of elements).
 */
class VBucketEvent {
public:
    /**
     * Create a new instance of the VBucketEvent and initialize
     * its members.
     * @param ev Type of event
     * @param b The bucket this event belongs to
     * @param s The state change for this event
     */
    VBucketEvent(uint16_t ev, uint16_t b, vbucket_state_t s) :
        event(ev), vbucket(b), state(s) {}
    uint16_t event;
    uint16_t vbucket;
    vbucket_state_t state;
};

/**
 * Aggregator object to count all tap stats.
 */
struct ConnCounter {
    ConnCounter()
        : conn_queue(0), totalConns(0), totalProducers(0),
          conn_queueFill(0), conn_queueDrain(0), conn_totalBytes(0), conn_queueRemaining(0),
          conn_queueBackoff(0), conn_queueBackfillRemaining(0), conn_queueItemOnDisk(0),
          conn_totalBacklogSize(0)
    {}

    ConnCounter& operator+=(const ConnCounter& other) {
        conn_queue += other.conn_queue;
        totalConns += other.totalConns;
        totalProducers += other.totalProducers;
        conn_queueFill += other.conn_queueFill;
        conn_queueDrain += other.conn_queueDrain;
        conn_totalBytes += other.conn_totalBytes;
        conn_queueRemaining += other.conn_queueRemaining;
        conn_queueBackoff += other.conn_queueBackoff;
        conn_queueBackfillRemaining += other.conn_queueBackfillRemaining;
        conn_queueItemOnDisk += other.conn_queueItemOnDisk;
        conn_totalBacklogSize += other.conn_totalBacklogSize;

        return *this;
    }

    size_t      conn_queue;
    size_t      totalConns;
    size_t      totalProducers;

    size_t      conn_queueFill;
    size_t      conn_queueDrain;
    size_t      conn_totalBytes;
    size_t      conn_queueRemaining;
    size_t      conn_queueBackoff;
    size_t      conn_queueBackfillRemaining;
    size_t      conn_queueItemOnDisk;
    size_t      conn_totalBacklogSize;
};

/**
 * Represents an item that has been sent over tap, but may need to be
 * rolled back if acks fail.
 */
class TapLogElement {

public:

    TapLogElement(uint32_t seqno, const VBucketEvent &e)
        : seqno_(seqno), event_(e.event), vbucket_(e.vbucket),
          state_(e.state) { }

    TapLogElement(uint32_t seqno, const queued_item &qi)
    {
        seqno_ = seqno;
        event_ = TAP_MUTATION;
        vbucket_ = qi->getVBucketId();
        state_ = vbucket_state_active;
        item_ = qi;

        switch(item_->getOperation()) {
        case queue_op_set:
            event_ = TAP_MUTATION;
            break;
        case queue_op_del:
            event_ = TAP_DELETION;
            break;
        case queue_op_flush:
            event_ = TAP_FLUSH;
            break;
        case queue_op_checkpoint_start:
            event_ = TAP_CHECKPOINT_START;
            break;
        case queue_op_checkpoint_end:
            event_ = TAP_CHECKPOINT_END;
            break;
        default:
            break;
        }
    }

    uint32_t seqno_;
    uint16_t event_;
    uint16_t vbucket_;

    vbucket_state_t state_;
    queued_item item_;
};

enum conn_type_t {
    TAP_CONN, //!< TAP connnection
    DCP_CONN  //!< DCP connection
};

class ConnHandler : public RCValue {
public:
    ConnHandler(EventuallyPersistentEngine& engine, const void* c,
                const std::string& name);

    virtual ~ConnHandler() {}

    virtual ENGINE_ERROR_CODE addStream(uint32_t opaque, uint16_t vbucket,
                                        uint32_t flags);

    virtual ENGINE_ERROR_CODE closeStream(uint32_t opaque, uint16_t vbucket);

    virtual ENGINE_ERROR_CODE streamEnd(uint32_t opaque, uint16_t vbucket,
                                        uint32_t flags);

    virtual ENGINE_ERROR_CODE mutation(uint32_t opaque, const void* key,
                                       uint16_t nkey, const void* value,
                                       uint32_t nvalue, uint64_t cas,
                                       uint16_t vbucket, uint32_t flags,
                                       uint8_t datatype, uint32_t locktime,
                                       uint64_t bySeqno, uint64_t revSeqno,
                                       uint32_t exptime, uint8_t nru,
                                       const void* meta, uint16_t nmeta);

    virtual ENGINE_ERROR_CODE deletion(uint32_t opaque, const void* key,
                                       uint16_t nkey, uint64_t cas,
                                       uint16_t vbucket, uint64_t bySeqno,
                                       uint64_t revSeqno, const void* meta,
                                       uint16_t nmeta);

    virtual ENGINE_ERROR_CODE expiration(uint32_t opaque, const void* key,
                                         uint16_t nkey, uint64_t cas,
                                         uint16_t vbucket, uint64_t bySeqno,
                                         uint64_t revSeqno, const void* meta,
                                         uint16_t nmeta);

    virtual ENGINE_ERROR_CODE snapshotMarker(uint32_t opaque,
                                             uint16_t vbucket,
                                             uint64_t start_seqno,
                                             uint64_t end_seqno,
                                             uint32_t flags);

    virtual ENGINE_ERROR_CODE flushall(uint32_t opaque, uint16_t vbucket);

    virtual ENGINE_ERROR_CODE setVBucketState(uint32_t opaque, uint16_t vbucket,
                                              vbucket_state_t state);

    virtual ENGINE_ERROR_CODE getFailoverLog(uint32_t opaque, uint16_t vbucket,
                                             dcp_add_failover_log callback);

    virtual ENGINE_ERROR_CODE streamRequest(uint32_t flags,
                                            uint32_t opaque,
                                            uint16_t vbucket,
                                            uint64_t start_seqno,
                                            uint64_t end_seqno,
                                            uint64_t vbucket_uuid,
                                            uint64_t snapStartSeqno,
                                            uint64_t snapEndSeqno,
                                            uint64_t *rollback_seqno,
                                            dcp_add_failover_log callback);

    virtual ENGINE_ERROR_CODE noop(uint32_t opaque);

    virtual ENGINE_ERROR_CODE bufferAcknowledgement(uint32_t opaque,
                                                    uint16_t vbucket,
                                                    uint32_t buffer_bytes);

    virtual ENGINE_ERROR_CODE control(uint32_t opaque, const void* key,
                                      uint16_t nkey, const void* value,
                                      uint32_t nvalue);

    virtual ENGINE_ERROR_CODE step(struct dcp_message_producers* producers);

    virtual ENGINE_ERROR_CODE handleResponse(
                                        protocol_binary_response_header *resp);

    EventuallyPersistentEngine& engine() {
        return engine_;
    }

    const char* logHeader() {
        return logString.c_str();
    }

    void setLogHeader(const std::string &header) {
        logString = header;
    }

    void releaseReference(bool force = false);

    void setSupportAck(bool ack) {
        supportAck = ack;
    }

    bool supportsAck() const {
        return supportAck;
    }

    void setSupportCheckpointSync(bool checkpointSync) {
        supportCheckpointSync_ = checkpointSync;
    }

    bool supportsCheckpointSync() const {
        return supportCheckpointSync_;
    }

    virtual const char *getType() const = 0;

    template <typename T>
    void addStat(const char *nm, const T &val, ADD_STAT add_stat, const void *c) {
        std::stringstream tap;
        tap << name << ":" << nm;
        std::stringstream value;
        value << val;
        std::string n = tap.str();
        add_casted_stat(n.data(), value.str().data(), add_stat, c);
    }

    void addStat(const char *nm, bool val, ADD_STAT add_stat, const void *c) {
        addStat(nm, val ? "true" : "false", add_stat, c);
    }

    virtual void addStats(ADD_STAT add_stat, const void *c) {
        addStat("type", getType(), add_stat, c);
        addStat("created", created, add_stat, c);
        addStat("connected", connected, add_stat, c);
        addStat("pending_disconnect", disconnect, add_stat, c);
        addStat("supports_ack", supportAck, add_stat, c);
        addStat("reserved", reserved.load(), add_stat, c);

        if (numDisconnects > 0) {
            addStat("disconnects", numDisconnects.load(), add_stat, c);
        }
    }

    virtual void aggregateQueueStats(ConnCounter* stats_aggregator) {
        // Empty
    }

    virtual void processedEvent(uint16_t event, ENGINE_ERROR_CODE ret) {
        (void) event;
        (void) ret;
    }

    const std::string &getName() const {
        return name;
    }

    void setName(const std::string &n) {
        name.assign(n);
    }

    bool setReserved(bool r) {
        bool inverse = !r;
        return reserved.compare_exchange_strong(inverse, r);
    }

    bool isReserved() const {
        return reserved;
    }

    const void *getCookie() const {
        return cookie;
    }

    void setCookie(const void *c) {
        cookie = c;
    }

    void setExpiryTime(rel_time_t t) {
        expiryTime = t;
    }

    rel_time_t getExpiryTime() {
        return expiryTime;
    }

    void setLastWalkTime() {
        lastWalkTime.store(ep_current_time());
    }

    rel_time_t getLastWalkTime() {
        return lastWalkTime.load();
    }

    void setConnected(bool s) {
        if (!s) {
            ++numDisconnects;
        }
        connected = s;
    }

    bool isConnected() {
        return connected;
    }

    bool doDisconnect() {
        return disconnect;
    }

    virtual void setDisconnect(bool val) {
        disconnect = val;
    }

    static std::string getAnonName() {
        uint64_t nextConnId = counter_++;
        std::stringstream s;
        s << "eq_tapq:anon_";
        s << nextConnId;
        return s.str();
    }

    hrtime_t getConnectionToken() const {
        return connToken;
    }

protected:
    EventuallyPersistentEngine &engine_;
    EPStats &stats;
    bool supportCheckpointSync_;

private:

     //! The name for this connection
    std::string name;

    //! The string used to prefix all log messages for this connection
    std::string logString;

    //! The cookie representing this connection (provided by the memcached code)
    const void* cookie;

    //! Whether or not the connection is reserved in the memcached layer
    AtomicValue<bool> reserved;

    //! Connection token created at connection instantiation time
    hrtime_t connToken;

    //! Connection creation time
    rel_time_t created;

    //! The last time this connection's step function was called
    AtomicValue<rel_time_t> lastWalkTime;

    //! Should we disconnect as soon as possible?
    bool disconnect;

    //! Is this tap conenction connected?
    bool connected;

    //! Number of times this connection was disconnected
    AtomicValue<size_t> numDisconnects;

    //! when this tap conneciton expires.
    rel_time_t expiryTime;

    //! Whether or not this connection supports acking
    bool supportAck;

    //! A counter used to generate unique names
    static AtomicValue<uint64_t> counter_;
};

enum proto_checkpoint_state {
    backfill,
    checkpoint_start,
    checkpoint_end,
    checkpoint_end_synced<commit_msg>Address possible data races in ConnHandler context<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2015 Couchbase, Inc
 *
 *   Licensed 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 SRC_TAPCONNECTION_H_
#define SRC_TAPCONNECTION_H_ 1

#include "config.h"

#include <list>
#include <map>
#include <queue>
#include <set>
#include <string>
#include <vector>

#include "atomic.h"
#include "common.h"
#include "locks.h"
#include "mutex.h"
#include "statwriter.h"

// forward decl
class ConnHandler;
class EventuallyPersistentEngine;
class TapConnMap;
class TapProducer;
class BGFetchCallback;
class CompleteBackfillOperation;
class Dispatcher;
class Item;
class VBucketFilter;

struct TapStatBuilder;
struct TapAggStatBuilder;
struct PopulateEventsBody;



#define MAX_TAP_KEEP_ALIVE 3600
#define MAX_TAKEOVER_TAP_LOG_SIZE 10
#define MINIMUM_BACKFILL_RESIDENT_THRESHOLD 0.7
#define DEFAULT_BACKFILL_RESIDENT_THRESHOLD 0.9

/**
 * A tap event that represents a change to the state of a vbucket.
 *
 * The tap stream may include other events than data mutation events,
 * but the data structures in the TapProducer does only store a key
 * for the item to store. We don't want to add more data to those elements,
 * because that could potentially consume a lot of memory (the tap queue
 * may have a lot of elements).
 */
class VBucketEvent {
public:
    /**
     * Create a new instance of the VBucketEvent and initialize
     * its members.
     * @param ev Type of event
     * @param b The bucket this event belongs to
     * @param s The state change for this event
     */
    VBucketEvent(uint16_t ev, uint16_t b, vbucket_state_t s) :
        event(ev), vbucket(b), state(s) {}
    uint16_t event;
    uint16_t vbucket;
    vbucket_state_t state;
};

/**
 * Aggregator object to count all tap stats.
 */
struct ConnCounter {
    ConnCounter()
        : conn_queue(0), totalConns(0), totalProducers(0),
          conn_queueFill(0), conn_queueDrain(0), conn_totalBytes(0), conn_queueRemaining(0),
          conn_queueBackoff(0), conn_queueBackfillRemaining(0), conn_queueItemOnDisk(0),
          conn_totalBacklogSize(0)
    {}

    ConnCounter& operator+=(const ConnCounter& other) {
        conn_queue += other.conn_queue;
        totalConns += other.totalConns;
        totalProducers += other.totalProducers;
        conn_queueFill += other.conn_queueFill;
        conn_queueDrain += other.conn_queueDrain;
        conn_totalBytes += other.conn_totalBytes;
        conn_queueRemaining += other.conn_queueRemaining;
        conn_queueBackoff += other.conn_queueBackoff;
        conn_queueBackfillRemaining += other.conn_queueBackfillRemaining;
        conn_queueItemOnDisk += other.conn_queueItemOnDisk;
        conn_totalBacklogSize += other.conn_totalBacklogSize;

        return *this;
    }

    size_t      conn_queue;
    size_t      totalConns;
    size_t      totalProducers;

    size_t      conn_queueFill;
    size_t      conn_queueDrain;
    size_t      conn_totalBytes;
    size_t      conn_queueRemaining;
    size_t      conn_queueBackoff;
    size_t      conn_queueBackfillRemaining;
    size_t      conn_queueItemOnDisk;
    size_t      conn_totalBacklogSize;
};

/**
 * Represents an item that has been sent over tap, but may need to be
 * rolled back if acks fail.
 */
class TapLogElement {

public:

    TapLogElement(uint32_t seqno, const VBucketEvent &e)
        : seqno_(seqno), event_(e.event), vbucket_(e.vbucket),
          state_(e.state) { }

    TapLogElement(uint32_t seqno, const queued_item &qi)
    {
        seqno_ = seqno;
        event_ = TAP_MUTATION;
        vbucket_ = qi->getVBucketId();
        state_ = vbucket_state_active;
        item_ = qi;

        switch(item_->getOperation()) {
        case queue_op_set:
            event_ = TAP_MUTATION;
            break;
        case queue_op_del:
            event_ = TAP_DELETION;
            break;
        case queue_op_flush:
            event_ = TAP_FLUSH;
            break;
        case queue_op_checkpoint_start:
            event_ = TAP_CHECKPOINT_START;
            break;
        case queue_op_checkpoint_end:
            event_ = TAP_CHECKPOINT_END;
            break;
        default:
            break;
        }
    }

    uint32_t seqno_;
    uint16_t event_;
    uint16_t vbucket_;

    vbucket_state_t state_;
    queued_item item_;
};

enum conn_type_t {
    TAP_CONN, //!< TAP connnection
    DCP_CONN  //!< DCP connection
};

class ConnHandler : public RCValue {
public:
    ConnHandler(EventuallyPersistentEngine& engine, const void* c,
                const std::string& name);

    virtual ~ConnHandler() {}

    virtual ENGINE_ERROR_CODE addStream(uint32_t opaque, uint16_t vbucket,
                                        uint32_t flags);

    virtual ENGINE_ERROR_CODE closeStream(uint32_t opaque, uint16_t vbucket);

    virtual ENGINE_ERROR_CODE streamEnd(uint32_t opaque, uint16_t vbucket,
                                        uint32_t flags);

    virtual ENGINE_ERROR_CODE mutation(uint32_t opaque, const void* key,
                                       uint16_t nkey, const void* value,
                                       uint32_t nvalue, uint64_t cas,
                                       uint16_t vbucket, uint32_t flags,
                                       uint8_t datatype, uint32_t locktime,
                                       uint64_t bySeqno, uint64_t revSeqno,
                                       uint32_t exptime, uint8_t nru,
                                       const void* meta, uint16_t nmeta);

    virtual ENGINE_ERROR_CODE deletion(uint32_t opaque, const void* key,
                                       uint16_t nkey, uint64_t cas,
                                       uint16_t vbucket, uint64_t bySeqno,
                                       uint64_t revSeqno, const void* meta,
                                       uint16_t nmeta);

    virtual ENGINE_ERROR_CODE expiration(uint32_t opaque, const void* key,
                                         uint16_t nkey, uint64_t cas,
                                         uint16_t vbucket, uint64_t bySeqno,
                                         uint64_t revSeqno, const void* meta,
                                         uint16_t nmeta);

    virtual ENGINE_ERROR_CODE snapshotMarker(uint32_t opaque,
                                             uint16_t vbucket,
                                             uint64_t start_seqno,
                                             uint64_t end_seqno,
                                             uint32_t flags);

    virtual ENGINE_ERROR_CODE flushall(uint32_t opaque, uint16_t vbucket);

    virtual ENGINE_ERROR_CODE setVBucketState(uint32_t opaque, uint16_t vbucket,
                                              vbucket_state_t state);

    virtual ENGINE_ERROR_CODE getFailoverLog(uint32_t opaque, uint16_t vbucket,
                                             dcp_add_failover_log callback);

    virtual ENGINE_ERROR_CODE streamRequest(uint32_t flags,
                                            uint32_t opaque,
                                            uint16_t vbucket,
                                            uint64_t start_seqno,
                                            uint64_t end_seqno,
                                            uint64_t vbucket_uuid,
                                            uint64_t snapStartSeqno,
                                            uint64_t snapEndSeqno,
                                            uint64_t *rollback_seqno,
                                            dcp_add_failover_log callback);

    virtual ENGINE_ERROR_CODE noop(uint32_t opaque);

    virtual ENGINE_ERROR_CODE bufferAcknowledgement(uint32_t opaque,
                                                    uint16_t vbucket,
                                                    uint32_t buffer_bytes);

    virtual ENGINE_ERROR_CODE control(uint32_t opaque, const void* key,
                                      uint16_t nkey, const void* value,
                                      uint32_t nvalue);

    virtual ENGINE_ERROR_CODE step(struct dcp_message_producers* producers);

    virtual ENGINE_ERROR_CODE handleResponse(
                                        protocol_binary_response_header *resp);

    EventuallyPersistentEngine& engine() {
        return engine_;
    }

    const char* logHeader() {
        return logString.c_str();
    }

    void setLogHeader(const std::string &header) {
        logString = header;
    }

    void releaseReference(bool force = false);

    void setSupportAck(bool ack) {
        supportAck.store(ack);
    }

    bool supportsAck() const {
        return supportAck.load();
    }

    void setSupportCheckpointSync(bool checkpointSync) {
        supportCheckpointSync_ = checkpointSync;
    }

    bool supportsCheckpointSync() const {
        return supportCheckpointSync_;
    }

    virtual const char *getType() const = 0;

    template <typename T>
    void addStat(const char *nm, const T &val, ADD_STAT add_stat, const void *c) {
        std::stringstream tap;
        tap << name << ":" << nm;
        std::stringstream value;
        value << val;
        std::string n = tap.str();
        add_casted_stat(n.data(), value.str().data(), add_stat, c);
    }

    void addStat(const char *nm, bool val, ADD_STAT add_stat, const void *c) {
        addStat(nm, val ? "true" : "false", add_stat, c);
    }

    virtual void addStats(ADD_STAT add_stat, const void *c) {
        addStat("type", getType(), add_stat, c);
        addStat("created", created.load(), add_stat, c);
        addStat("connected", connected.load(), add_stat, c);
        addStat("pending_disconnect", disconnect.load(), add_stat, c);
        addStat("supports_ack", supportAck.load(), add_stat, c);
        addStat("reserved", reserved.load(), add_stat, c);

        if (numDisconnects > 0) {
            addStat("disconnects", numDisconnects.load(), add_stat, c);
        }
    }

    virtual void aggregateQueueStats(ConnCounter* stats_aggregator) {
        // Empty
    }

    virtual void processedEvent(uint16_t event, ENGINE_ERROR_CODE ret) {
        (void) event;
        (void) ret;
    }

    const std::string &getName() const {
        return name;
    }

    void setName(const std::string &n) {
        name.assign(n);
    }

    bool setReserved(bool r) {
        bool inverse = !r;
        return reserved.compare_exchange_strong(inverse, r);
    }

    bool isReserved() const {
        return reserved;
    }

    const void *getCookie() const {
        return cookie;
    }

    void setCookie(const void *c) {
        cookie = c;
    }

    void setExpiryTime(rel_time_t t) {
        expiryTime = t;
    }

    rel_time_t getExpiryTime() {
        return expiryTime;
    }

    void setLastWalkTime() {
        lastWalkTime.store(ep_current_time());
    }

    rel_time_t getLastWalkTime() {
        return lastWalkTime.load();
    }

    void setConnected(bool s) {
        if (!s) {
            ++numDisconnects;
        }
        connected.store(s);
    }

    bool isConnected() {
        return connected.load();
    }

    bool doDisconnect() {
        return disconnect.load();
    }

    virtual void setDisconnect(bool val) {
        disconnect.store(val);
    }

    static std::string getAnonName() {
        uint64_t nextConnId = counter_++;
        std::stringstream s;
        s << "eq_tapq:anon_";
        s << nextConnId;
        return s.str();
    }

    hrtime_t getConnectionToken() const {
        return connToken;
    }

protected:
    EventuallyPersistentEngine &engine_;
    EPStats &stats;
    bool supportCheckpointSync_;

private:

     //! The name for this connection
    std::string name;

    //! The string used to prefix all log messages for this connection
    std::string logString;

    //! The cookie representing this connection (provided by the memcached code)
    const void* cookie;

    //! Whether or not the connection is reserved in the memcached layer
    AtomicValue<bool> reserved;

    //! Connection token created at connection instantiation time
    hrtime_t connToken;

    //! Connection creation time
    AtomicValue<rel_time_t> created;

    //! The last time this connection's step function was called
    AtomicValue<rel_time_t> lastWalkTime;

    //! Should we disconnect as soon as possible?
    AtomicValue<bool> disconnect;

    //! Is this tap conenction connected?
    AtomicValue<bool> connected;

    //! Number of times this connection was disconnected
    AtomicValue<size_t> numDisconnects;

    //! when this tap conneciton expires.
    rel_time_t expiryTime;

    //! Whether or not this connection supports acking
    AtomicValue<bool> supportAck;

    //! A counter used to generate unique names
    static AtomicValue<uint64_t> counter_;
};

enum proto_checkpoint_state {
    backfill,
    checkpoint_start,
    checkpoint_end,
    checkpoint_end_synced
{'fraction_non_alphanumeric': 0.07432808155699722, 'fraction_numerical': 0.013345690454124188, 'mean_word_length': 1.9976664073785977, 'pattern_counts': {'":': 2, '<': 57, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 116, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '876535', 'n_tokens_mistral': 8000, 'n_tokens_neox': 7492, 'n_words': 2054}