BufferedWriter.h

Go to the documentation of this file.

#pragma once

#include <chrono>
#include <memory>
#include <tuple>
#include <variant>
#include <vector>

#include "logdevice/include/Client.h"
#include "logdevice/include/types.h"

namespace facebook { namespace logdevice {

class BufferedWriterImpl;
class BufferedWriterAppendSink;

class BufferedWriter {
 public:
  using PayloadVariant = std::variant<std::string, PayloadGroup>;

  class AppendCallback {
   public:
    using Context = void*;
    using ContextSet = std::vector<std::pair<Context, PayloadVariant>>;

    virtual void onSuccess(logid_t /*log_id*/,
                           ContextSet /*contexts_and_payloads*/,
                           const DataRecordAttributes& /*attrs*/) {}
    virtual void onFailure(logid_t /*log_id*/,
                           ContextSet /*contexts_and_payloads*/,
                           Status /*status*/) {}

    enum class RetryDecision { ALLOW, DENY };

    virtual RetryDecision onRetry(logid_t /*log_id*/,
                                  const ContextSet& /*contexts_and_payloads*/,
                                  Status /*status*/) {
      return RetryDecision::ALLOW;
    }

    virtual ~AppendCallback() {}
  };
  struct Append {
    Append(logid_t log_id,
           std::string payload,
           AppendCallback::Context context,
           AppendAttributes attrs = AppendAttributes())
        : log_id(log_id),
          payload(std::move(payload)),
          context(context),
          attrs(std::move(attrs)) {}

    Append(logid_t log_id,
           PayloadGroup&& payload,
           AppendCallback::Context context,
           AppendAttributes attrs = AppendAttributes())
        : log_id(log_id),
          payload(std::move(payload)),
          context(context),
          attrs(std::move(attrs)) {}

    Append(const Append&) = delete;
    Append(Append&&) = default;
    Append& operator=(Append&&) = default;
    logid_t log_id;
    PayloadVariant payload;
    AppendCallback::Context context;
    AppendAttributes attrs;
  };
  struct LogOptions {
    // TODO: Remove it from this struct.
    using Compression = facebook::logdevice::Compression;

    LogOptions() {}

    // Flush buffered writes for a log when the oldest has been buffered this
    // long (negative for no trigger)
    std::chrono::milliseconds time_trigger{-1};

    // Flush buffered writes for a log as soon there are this many payload
    // bytes buffered (negative for no trigger)
    ssize_t size_trigger = -1;

    enum class Mode {
      // Write each batch independently (also applies to retries if
      // configured).
      //
      // This is the default mode which allows for highest throughput, but can
      // cause writes to get reordered.  For example, suppose two batches 1
      // and 2 get sent out, 1 fails and 2 succeeds.  After 1 is retried, the
      // contents of the log would be 21 (or 121 if the very first write
      // actually succeeded but we could not get confirmation).
      INDEPENDENT,

      // Only allow one batch at a time to be inflight to LogDevice servers.
      // This fixes the ordering issue in the INDEPENDENT mode.  It is
      // especially useful for stream processing cases where records in
      // LogDevice need to end up in the same order as in the input stream.
      //
      // The size and time triggers are relaxed (batches can get larger and
      // delayed while one is already in flight).  This mode possibly limits
      // throughput under certain conditions (extremely high throughput on a
      // single log and/or errors writing to LogDevice).
      ONE_AT_A_TIME,

      // Enables FIFO ordering for appends to a log from the same
      // BufferedWriter. STREAM mode does not guarantee exactly-once semantics
      // and hence records maybe duplicated. Precisely, for records r_1, r_2,
      // ... appended using BufferedWriter::append(), the first occurrence of
      // record r_i in the read stream is before the first occurrence of
      // r_{i+1}. In the event of a failure, we recover a continuous prefix
      // of records accepted by BufferedWriter::append().
      //
      // Callbacks to the BufferedWriter happen in order of accepted append
      // messages. LSN reported back via the callback is not necessarily the
      // first occurrence of the record and hence need not increase
      // monotonically. It is not recommended to start reading directly from the
      // LSN returned, since the reader may encounter gaps. We recommend reading
      // from the beginning and caching the largest sequence number encountered,
      // to consume records in FIFO order. We currently do not expose any
      // sequence number to readers and hence it must be encoded as part of the
      // payload to support exactly-once FIFO consumption.
      STREAM,
    };
    Mode mode = Mode::INDEPENDENT;
    // Max number of times to retry (0 for no retrying, negative for
    // unlimited).  You may also manually track retries and have onRetry()
    // return DENY to stop retrying a particular batch.
    int retry_count = 0;
    // Initial delay before retrying (negative for a default 2x the append
    // timeout).  Subsequent retries are made after successively larger delays
    // (exponential backoff with a factor of 2) up to retry_max_delay
    std::chrono::milliseconds retry_initial_delay{-1};
    // Max delay when retrying (negative for no limit)
    std::chrono::milliseconds retry_max_delay{60000};

    // Compression codec.
    Compression compression = Compression::LZ4;

    // If set to true, will destroy individual payloads immediately after they
    // are batched together. onSuccess(), onFailure() and onRetry() callbacks
    // will not contain payloads.
    bool destroy_payloads = false;

    // Returns "independent" or "one_at_a_time".
    static std::string modeToString(Mode mode);
    // Returns 0 on success, -1 on error.
    static int parseMode(const char* str, Mode* out_mode);

    // COMPAT. Use `compressionToString` and `parseCompression` from
    // facebook::logdevice namespace.
    static std::string compressionToString(Compression c);
    static int parseCompression(const char* str, Compression* out_c);
  };

  // NOTE: It is possible to have different options for each log but currently
  // this feature is not supported in the interface.
  struct Options : public LogOptions {
    Options() : LogOptions() {}

    // Approximate memory budget for buffered and inflight writes.  If an
    // append() call would exceed this limit, it fails fast with E::NOBUFS.
    //
    // Accounting is not completely accurate for performance reasons.  There
    // is internal overhead per batch and there may be pathological cases
    // where actual memory usage exceeds the limit.  However, in most cases it
    // should stay well under.
    //
    // Negative for no limit.
    int32_t memory_limit_mb = -1;
  };

  static std::unique_ptr<BufferedWriter> create(std::shared_ptr<Client> client,
                                                AppendCallback* callback,
                                                Options options = Options());

  int append(logid_t logid,
             std::string&& payload,
             AppendCallback::Context callback_context,
             AppendAttributes&& attrs = AppendAttributes());
  int append(logid_t logid,
             PayloadGroup&& payload_group,
             AppendCallback::Context callback_context,
             AppendAttributes&& attrs = AppendAttributes());

  std::vector<Status> append(std::vector<Append>&& appends);

  int flushAll();

  virtual ~BufferedWriter() {}

 private:
  BufferedWriter() {} // can be constructed by the factory only
  BufferedWriter(const BufferedWriter&) = delete;
  BufferedWriter& operator=(const BufferedWriter&) = delete;

  friend class BufferedWriterImpl;
  BufferedWriterImpl* impl(); // downcasts (this)
};
}} // namespace facebook::logdevice