Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 542 insertions, 0 deletions

diff --git a/src/osdc/Journaler.h b/src/osdc/Journaler.h
new file mode 100644
index 000000000..3e8f0f665
--- /dev/null
+++ b/src/osdc/Journaler.h
@@ -0,0 +1,542 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+/* Journaler
+ *
+ * This class stripes a serial log over objects on the store.  Four
+ * logical pointers:
+ *
+ *  write_pos - where we're writing new entries
+ *  unused_field - where we're reading old entires
+ *  expire_pos - what is deemed "old" by user
+ *  trimmed_pos - where we're expiring old items
+ *
+ *  trimmed_pos <= expire_pos <= unused_field <= write_pos.
+ *
+ * Often, unused_field <= write_pos (as with MDS log).  During
+ * recovery, write_pos is undefined until the end of the log is
+ * discovered.
+ *
+ * A "head" struct at the beginning of the log is used to store
+ * metadata at regular intervals.  The basic invariants include:
+ *
+ *   head.unused_field <= unused_field -- the head may "lag", since
+ *                                        it's updated lazily.
+ *   head.write_pos  <= write_pos
+ *   head.expire_pos <= expire_pos
+ *   head.trimmed_pos   <= trimmed_pos
+ *
+ * More significantly,
+ *
+ *   head.expire_pos >= trimmed_pos -- this ensures we can find the
+ *                                     "beginning" of the log as last
+ *                                     recorded, before it is trimmed.
+ *                                     trimming will block until a
+ *                                     sufficiently current expire_pos
+ *                                     is committed.
+ *
+ * To recover log state, we simply start at the last write_pos in the
+ * head, and probe the object sequence sizes until we read the end.
+ *
+ * Head struct is stored in the first object.  Actual journal starts
+ * after layout.period() bytes.
+ *
+ */
+
+#ifndef CEPH_JOURNALER_H
+#define CEPH_JOURNALER_H
+
+#include <list>
+#include <map>
+
+#include "Objecter.h"
+#include "Filer.h"
+
+#include "common/Timer.h"
+#include "common/Throttle.h"
+#include "include/common_fwd.h"
+
+class Context;
+class Finisher;
+class C_OnFinisher;
+
+typedef __u8 stream_format_t;
+
+// Legacy envelope is leading uint32_t size
+enum StreamFormat {
+    JOURNAL_FORMAT_LEGACY = 0,
+    JOURNAL_FORMAT_RESILIENT = 1,
+    // Insert new formats here, before COUNT
+    JOURNAL_FORMAT_COUNT
+};
+
+// Highest journal format version that we support
+#define JOURNAL_FORMAT_MAX (JOURNAL_FORMAT_COUNT - 1)
+
+// Legacy envelope is leading uint32_t size
+#define JOURNAL_ENVELOPE_LEGACY (sizeof(uint32_t))
+
+// Resilient envelope is leading uint64_t sentinel, uint32_t size,
+// trailing uint64_t start_ptr
+#define JOURNAL_ENVELOPE_RESILIENT (sizeof(uint32_t) + sizeof(uint64_t) + \
+				    sizeof(uint64_t))
+
+/**
+ * Represents a collection of entries serialized in a byte stream.
+ *
+ * Each entry consists of:
+ *  - a blob (used by the next level up as a serialized LogEvent)
+ *  - a uint64_t (used by the next level up as a pointer to the start
+ *    of the entry in the collection bytestream)
+ */
+class JournalStream
+{
+  stream_format_t format;
+
+  public:
+  JournalStream(stream_format_t format_) : format(format_) {}
+
+  void set_format(stream_format_t format_) {format = format_;}
+
+  bool readable(bufferlist &bl, uint64_t *need) const;
+  size_t read(bufferlist &from, bufferlist *to, uint64_t *start_ptr);
+  size_t write(bufferlist &entry, bufferlist *to, uint64_t const &start_ptr);
+  size_t get_envelope_size() const {
+     if (format >= JOURNAL_FORMAT_RESILIENT) {
+       return JOURNAL_ENVELOPE_RESILIENT;
+     } else {
+       return JOURNAL_ENVELOPE_LEGACY;
+     }
+  }
+
+  // A magic number for the start of journal entries, so that we can
+  // identify them in damaged journals.
+  static const uint64_t sentinel = 0x3141592653589793;
+};
+
+
+class Journaler {
+public:
+  // this goes at the head of the log "file".
+  class Header {
+    public:
+    uint64_t trimmed_pos;
+    uint64_t expire_pos;
+    uint64_t unused_field;
+    uint64_t write_pos;
+    string magic;
+    file_layout_t layout; //< The mapping from byte stream offsets
+			     //  to RADOS objects
+    stream_format_t stream_format; //< The encoding of LogEvents
+				   //  within the journal byte stream
+
+    Header(const char *m="") :
+      trimmed_pos(0), expire_pos(0), unused_field(0), write_pos(0), magic(m),
+      stream_format(-1) {
+    }
+
+    void encode(bufferlist &bl) const {
+      ENCODE_START(2, 2, bl);
+      encode(magic, bl);
+      encode(trimmed_pos, bl);
+      encode(expire_pos, bl);
+      encode(unused_field, bl);
+      encode(write_pos, bl);
+      encode(layout, bl, 0);  // encode in legacy format
+      encode(stream_format, bl);
+      ENCODE_FINISH(bl);
+    }
+    void decode(bufferlist::const_iterator &bl) {
+      DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
+      decode(magic, bl);
+      decode(trimmed_pos, bl);
+      decode(expire_pos, bl);
+      decode(unused_field, bl);
+      decode(write_pos, bl);
+      decode(layout, bl);
+      if (struct_v > 1) {
+	decode(stream_format, bl);
+      } else {
+	stream_format = JOURNAL_FORMAT_LEGACY;
+      }
+      DECODE_FINISH(bl);
+    }
+
+    void dump(Formatter *f) const {
+      f->open_object_section("journal_header");
+      {
+	f->dump_string("magic", magic);
+	f->dump_unsigned("write_pos", write_pos);
+	f->dump_unsigned("expire_pos", expire_pos);
+	f->dump_unsigned("trimmed_pos", trimmed_pos);
+	f->dump_unsigned("stream_format", stream_format);
+	f->dump_object("layout", layout);
+      }
+      f->close_section(); // journal_header
+    }
+
+    static void generate_test_instances(list<Header*> &ls)
+    {
+      ls.push_back(new Header());
+
+      ls.push_back(new Header());
+      ls.back()->trimmed_pos = 1;
+      ls.back()->expire_pos = 2;
+      ls.back()->unused_field = 3;
+      ls.back()->write_pos = 4;
+      ls.back()->magic = "magique";
+
+      ls.push_back(new Header());
+      ls.back()->stream_format = JOURNAL_FORMAT_RESILIENT;
+    }
+  };
+  WRITE_CLASS_ENCODER(Header)
+
+  uint32_t get_stream_format() const {
+    return stream_format;
+  }
+
+  Header last_committed;
+
+private:
+  // me
+  CephContext *cct;
+  std::mutex lock;
+  const std::string name;
+  typedef std::lock_guard<std::mutex> lock_guard;
+  typedef std::unique_lock<std::mutex> unique_lock;
+  Finisher *finisher;
+  Header last_written;
+  inodeno_t ino;
+  int64_t pg_pool;
+  bool readonly;
+  file_layout_t layout;
+  uint32_t stream_format;
+  JournalStream journal_stream;
+
+  const char *magic;
+  Objecter *objecter;
+  Filer filer;
+
+  PerfCounters *logger;
+  int logger_key_lat;
+
+  class C_DelayFlush;
+  C_DelayFlush *delay_flush_event;
+  /*
+   * Do a flush as a result of a C_DelayFlush context.
+   */
+  void _do_delayed_flush()
+  {
+    ceph_assert(delay_flush_event != NULL);
+    lock_guard l(lock);
+    delay_flush_event = NULL;
+    _do_flush();
+  }
+
+  // my state
+  static const int STATE_UNDEF = 0;
+  static const int STATE_READHEAD = 1;
+  static const int STATE_PROBING = 2;
+  static const int STATE_ACTIVE = 3;
+  static const int STATE_REREADHEAD = 4;
+  static const int STATE_REPROBING = 5;
+  static const int STATE_STOPPING = 6;
+
+  int state;
+  int error;
+
+  void _write_head(Context *oncommit=NULL);
+  void _wait_for_flush(Context *onsafe);
+  void _trim();
+
+  // header
+  ceph::real_time last_wrote_head;
+  void _finish_write_head(int r, Header &wrote, C_OnFinisher *oncommit);
+  class C_WriteHead;
+  friend class C_WriteHead;
+
+  void _reread_head(Context *onfinish);
+  void _set_layout(file_layout_t const *l);
+  list<Context*> waitfor_recover;
+  void _read_head(Context *on_finish, bufferlist *bl);
+  void _finish_read_head(int r, bufferlist& bl);
+  void _finish_reread_head(int r, bufferlist& bl, Context *finish);
+  void _probe(Context *finish, uint64_t *end);
+  void _finish_probe_end(int r, uint64_t end);
+  void _reprobe(C_OnFinisher *onfinish);
+  void _finish_reprobe(int r, uint64_t end, C_OnFinisher *onfinish);
+  void _finish_reread_head_and_probe(int r, C_OnFinisher *onfinish);
+  class C_ReadHead;
+  friend class C_ReadHead;
+  class C_ProbeEnd;
+  friend class C_ProbeEnd;
+  class C_RereadHead;
+  friend class C_RereadHead;
+  class C_ReProbe;
+  friend class C_ReProbe;
+  class C_RereadHeadProbe;
+  friend class C_RereadHeadProbe;
+
+  // writer
+  uint64_t prezeroing_pos;
+  uint64_t prezero_pos; ///< we zero journal space ahead of write_pos to
+			//   avoid problems with tail probing
+  uint64_t write_pos; ///< logical write position, where next entry
+		      //   will go
+  uint64_t flush_pos; ///< where we will flush. if
+		      ///  write_pos>flush_pos, we're buffering writes.
+  uint64_t safe_pos; ///< what has been committed safely to disk.
+
+  uint64_t next_safe_pos; /// start position of the first entry that isn't
+			  /// being fully flushed. If we don't flush any
+			  // partial entry, it's equal to flush_pos.
+
+  bufferlist write_buf; ///< write buffer.  flush_pos +
+			///  write_buf.length() == write_pos.
+
+  // protect write_buf from bufferlist _len overflow 
+  Throttle write_buf_throttle;
+
+  uint64_t waiting_for_zero_pos;
+  interval_set<uint64_t> pending_zero;  // non-contig bits we've zeroed
+  list<Context*> waitfor_prezero;
+
+  std::map<uint64_t, uint64_t> pending_safe; // flush_pos -> safe_pos
+  // when safe through given offset
+  std::map<uint64_t, std::list<Context*> > waitfor_safe;
+
+  void _flush(C_OnFinisher *onsafe);
+  void _do_flush(unsigned amount=0);
+  void _finish_flush(int r, uint64_t start, ceph::real_time stamp);
+  class C_Flush;
+  friend class C_Flush;
+
+  // reader
+  uint64_t read_pos;      // logical read position, where next entry starts.
+  uint64_t requested_pos; // what we've requested from OSD.
+  uint64_t received_pos;  // what we've received from OSD.
+  // read buffer.  unused_field + read_buf.length() == prefetch_pos.
+  bufferlist read_buf;
+
+  map<uint64_t,bufferlist> prefetch_buf;
+
+  uint64_t fetch_len;     // how much to read at a time
+  uint64_t temp_fetch_len;
+
+  // for wait_for_readable()
+  C_OnFinisher *on_readable;
+  C_OnFinisher *on_write_error;
+  bool called_write_error;
+
+  // read completion callback
+  void _finish_read(int r, uint64_t offset, uint64_t length, bufferlist &bl);
+  void _finish_retry_read(int r);
+  void _assimilate_prefetch();
+  void _issue_read(uint64_t len); // read some more
+  void _prefetch(); // maybe read ahead
+  class C_Read;
+  friend class C_Read;
+  class C_RetryRead;
+  friend class C_RetryRead;
+
+  // trimmer
+  uint64_t expire_pos;    // what we're allowed to trim to
+  uint64_t trimming_pos;      // what we've requested to trim through
+  uint64_t trimmed_pos;   // what has been trimmed
+
+  bool readable;
+
+  void _finish_trim(int r, uint64_t to);
+  class C_Trim;
+  friend class C_Trim;
+
+  void _issue_prezero();
+  void _finish_prezero(int r, uint64_t from, uint64_t len);
+  friend struct C_Journaler_Prezero;
+
+  // only init_headers when following or first reading off-disk
+  void init_headers(Header& h) {
+    ceph_assert(readonly ||
+	   state == STATE_READHEAD ||
+	   state == STATE_REREADHEAD);
+    last_written = last_committed = h;
+  }
+
+  /**
+   * handle a write error
+   *
+   * called when we get an objecter error on a write.
+   *
+   * @param r error code
+   */
+  void handle_write_error(int r);
+
+  bool _is_readable();
+
+  void _finish_erase(int data_result, C_OnFinisher *completion);
+  class C_EraseFinish;
+  friend class C_EraseFinish;
+
+  C_OnFinisher *wrap_finisher(Context *c);
+
+  uint32_t write_iohint; // the fadvise flags for write op, see
+			 // CEPH_OSD_OP_FADIVSE_*
+
+public:
+  Journaler(const std::string &name_, inodeno_t ino_, int64_t pool,
+      const char *mag, Objecter *obj, PerfCounters *l, int lkey, Finisher *f) :
+    last_committed(mag),
+    cct(obj->cct), name(name_), finisher(f), last_written(mag),
+    ino(ino_), pg_pool(pool), readonly(true),
+    stream_format(-1), journal_stream(-1),
+    magic(mag),
+    objecter(obj), filer(objecter, f), logger(l), logger_key_lat(lkey),
+    delay_flush_event(0),
+    state(STATE_UNDEF), error(0),
+    prezeroing_pos(0), prezero_pos(0), write_pos(0), flush_pos(0),
+    safe_pos(0), next_safe_pos(0),
+    write_buf_throttle(cct, "write_buf_throttle", UINT_MAX - (UINT_MAX >> 3)),
+    waiting_for_zero_pos(0),
+    read_pos(0), requested_pos(0), received_pos(0),
+    fetch_len(0), temp_fetch_len(0),
+    on_readable(0), on_write_error(NULL), called_write_error(false),
+    expire_pos(0), trimming_pos(0), trimmed_pos(0), readable(false),
+    write_iohint(0)
+  {
+  }
+
+  /* reset
+   *
+   * NOTE: we assume the caller knows/has ensured that any objects in
+   * our sequence do not exist.. e.g. after a MKFS.  this is _not_ an
+   * "erase" method.
+   */
+  void reset() {
+    lock_guard l(lock);
+    ceph_assert(state == STATE_ACTIVE);
+
+    readonly = true;
+    delay_flush_event = NULL;
+    state = STATE_UNDEF;
+    error = 0;
+    prezeroing_pos = 0;
+    prezero_pos = 0;
+    write_pos = 0;
+    flush_pos = 0;
+    safe_pos = 0;
+    next_safe_pos = 0;
+    read_pos = 0;
+    requested_pos = 0;
+    received_pos = 0;
+    fetch_len = 0;
+    ceph_assert(!on_readable);
+    expire_pos = 0;
+    trimming_pos = 0;
+    trimmed_pos = 0;
+    waiting_for_zero_pos = 0;
+  }
+
+  // Asynchronous operations
+  // =======================
+  void erase(Context *completion);
+  void create(file_layout_t *layout, stream_format_t const sf);
+  void recover(Context *onfinish);
+  void reread_head(Context *onfinish);
+  void reread_head_and_probe(Context *onfinish);
+  void write_head(Context *onsave=0);
+  void wait_for_flush(Context *onsafe = 0);
+  void flush(Context *onsafe = 0);
+  void wait_for_readable(Context *onfinish);
+  bool have_waiter() const;
+  void wait_for_prezero(Context *onfinish);
+
+  // Synchronous setters
+  // ===================
+  void set_layout(file_layout_t const *l);
+  void set_readonly();
+  void set_writeable();
+  void set_write_pos(uint64_t p) {
+    lock_guard l(lock);
+    prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos = p;
+  }
+  void set_read_pos(uint64_t p) {
+    lock_guard l(lock);
+    // we can't cope w/ in-progress read right now.
+    ceph_assert(requested_pos == received_pos);
+    read_pos = requested_pos = received_pos = p;
+    read_buf.clear();
+  }
+  uint64_t append_entry(bufferlist& bl);
+  void set_expire_pos(uint64_t ep) {
+      lock_guard l(lock);
+      expire_pos = ep;
+  }
+  void set_trimmed_pos(uint64_t p) {
+      lock_guard l(lock);
+      trimming_pos = trimmed_pos = p;
+  }
+
+  bool _write_head_needed();
+  bool write_head_needed() {
+    lock_guard l(lock);
+    return _write_head_needed();
+  }
+
+
+  void trim();
+  void trim_tail() {
+    lock_guard l(lock);
+
+    ceph_assert(!readonly);
+    _issue_prezero();
+  }
+
+  void set_write_error_handler(Context *c);
+
+  void set_write_iohint(uint32_t iohint_flags) {
+    write_iohint = iohint_flags;
+  }
+  /**
+   * Cause any ongoing waits to error out with -EAGAIN, set error
+   * to -EAGAIN.
+   */
+  void shutdown();
+public:
+
+  // Synchronous getters
+  // ===================
+  // TODO: need some locks on reads for true safety
+  uint64_t get_layout_period() const {
+    return layout.get_period();
+  }
+  file_layout_t& get_layout() { return layout; }
+  bool is_active() { return state == STATE_ACTIVE; }
+  bool is_stopping() { return state == STATE_STOPPING; }
+  int get_error() { return error; }
+  bool is_readonly() { return readonly; }
+  bool is_readable();
+  bool try_read_entry(bufferlist& bl);
+  uint64_t get_write_pos() const { return write_pos; }
+  uint64_t get_write_safe_pos() const { return safe_pos; }
+  uint64_t get_read_pos() const { return read_pos; }
+  uint64_t get_expire_pos() const { return expire_pos; }
+  uint64_t get_trimmed_pos() const { return trimmed_pos; }
+  size_t get_journal_envelope_size() const { 
+    return journal_stream.get_envelope_size(); 
+  }
+};
+WRITE_CLASS_ENCODER(Journaler::Header)
+
+#endif