blog review 第十五期

template <typename T>
struct TimestampValuePair {
  unsigned long long timestamp{0};
  T value;

  TimestampValuePair<T>() {
    timestamp = 0;
    value = T();
  }

  // need this because of static cast
  TimestampValuePair<T>(const unsigned long long& a) {
    timestamp = 0;
    value = T();
  }

  TimestampValuePair<T>(const unsigned long long& ts, const T& v) {
    timestamp = ts;
    value = v;
  }
  unsigned size() { return value.size() + sizeof(unsigned long long); }
};

template <typename T>
class LWWPairLattice : public Lattice<TimestampValuePair<T>> {
 protected:
  void do_merge(const TimestampValuePair<T>& p) {
    if (p.timestamp >= this->element.timestamp) {
      this->element.timestamp = p.timestamp;
      this->element.value = p.value;
    }
  }

 public:
  LWWPairLattice() : Lattice<TimestampValuePair<T>>(TimestampValuePair<T>()) {}
  LWWPairLattice(const TimestampValuePair<T>& p) :
      Lattice<TimestampValuePair<T>>(p) {}
  MaxLattice<unsigned> size() { return {this->element.size()}; }
};

  if (kSelfTier == Tier::MEMORY) {
    MemoryLWWKVS *lww_kvs = new MemoryLWWKVS();
    lww_serializer = new MemoryLWWSerializer(lww_kvs);

    MemorySetKVS *set_kvs = new MemorySetKVS();
    set_serializer = new MemorySetSerializer(set_kvs);

    MemoryOrderedSetKVS *ordered_set_kvs = new MemoryOrderedSetKVS();
    ordered_set_serializer = new MemoryOrderedSetSerializer(ordered_set_kvs);

    MemorySingleKeyCausalKVS *causal_kvs = new MemorySingleKeyCausalKVS();
    sk_causal_serializer = new MemorySingleKeyCausalSerializer(causal_kvs);

    MemoryMultiKeyCausalKVS *multi_key_causal_kvs =
        new MemoryMultiKeyCausalKVS();
    mk_causal_serializer =
        new MemoryMultiKeyCausalSerializer(multi_key_causal_kvs);

    MemoryPriorityKVS *priority_kvs = new MemoryPriorityKVS();
    priority_serializer = new MemoryPrioritySerializer(priority_kvs);
  } else if (kSelfTier == Tier::DISK) {
    lww_serializer = new DiskLWWSerializer(thread_id);
    set_serializer = new DiskSetSerializer(thread_id);
    ordered_set_serializer = new DiskOrderedSetSerializer(thread_id);
    sk_causal_serializer = new DiskSingleKeyCausalSerializer(thread_id);
    mk_causal_serializer = new DiskMultiKeyCausalSerializer(thread_id);
    priority_serializer = new DiskPrioritySerializer(thread_id);
  } else {

  string get(const Key &key, AnnaError &error) {
    string res;
    LWWValue value;

    // open a new filestream for reading in a binary
    string fname = ebs_root_ + "ebs_" + std::to_string(tid_) + "/" + key;
    std::fstream input(fname, std::ios::in | std::ios::binary);

    if (!input) {
      error = AnnaError::KEY_DNE;
    } else if (!value.ParseFromIstream(&input)) {
      std::cerr << "Failed to parse payload." << std::endl;
      error = AnnaError::KEY_DNE;
    } else {
      if (value.value() == "") {
        error = AnnaError::KEY_DNE;
      } else {
        value.SerializeToString(&res);
      }
    }
    return res;
  }

  unsigned put(const Key &key, const string &serialized) {
    LWWValue input_value;
    input_value.ParseFromString(serialized);

    LWWValue original_value;

    string fname = ebs_root_ + "ebs_" + std::to_string(tid_) + "/" + key;
    std::fstream input(fname, std::ios::in | std::ios::binary);

    if (!input) { // in this case, this key has never been seen before, so we
                  // attempt to create a new file for it

      // ios::trunc means that we overwrite the existing file
      std::fstream output(fname,
                          std::ios::out | std::ios::trunc | std::ios::binary);
      if (!input_value.SerializeToOstream(&output)) {
        std::cerr << "Failed to write payload." << std::endl;
      }
      return output.tellp();
    } else if (!original_value.ParseFromIstream(
                   &input)) { // if we have seen the key before, attempt to
                              // parse what was there before
      std::cerr << "Failed to parse payload." << std::endl;
      return 0;
    } else {
      if (input_value.timestamp() >= original_value.timestamp()) {
        std::fstream output(fname,
                            std::ios::out | std::ios::trunc | std::ios::binary);
        if (!input_value.SerializeToOstream(&output)) {
          std::cerr << "Failed to write payload" << std::endl;
        }
        return output.tellp();
      } else {
        return input.tellp();
      }
    }
  }

#include <linux/aio_abi.h>
int io_setup(unsigned nr_events, aio_context_t *ctxp);
int io_destroy(aio_context_t ctx);
int io_submit(aio_context_t ctx, long nr, struct iocb **iocbpp);
int io_cancel(aio_context_t ctx, struct iocb *, struct io_event *result);
int io_getevents(aio_context_t ctx, long min_nr, long nr,
    struct io_event *events, struct timespec *timeout);

/* read() from /dev/aio returns these structures. */
struct io_event {
        __u64           data;           /* the data field from the iocb */
        __u64           obj;            /* what iocb this event came from */
        __s64           res;            /* result code for this event */
        __s64           res2;           /* secondary result */
};

/*
 * we always use a 64bit off_t when communicating
 * with userland.  its up to libraries to do the
 * proper padding and aio_error abstraction
 */

struct iocb {
        /* these are internal to the kernel/libc. */
        __u64   aio_data;       /* data to be returned in event's data */
        __u32   PADDED(aio_key, aio_reserved1);
                                /* the kernel sets aio_key to the req # */

        /* common fields */
        __u16   aio_lio_opcode; /* see IOCB_CMD_ above */
        __s16   aio_reqprio;
        __u32   aio_fildes;

        __u64   aio_buf;
        __u64   aio_nbytes;
        __s64   aio_offset;

        /* extra parameters */
        __u64   aio_reserved2;  /* TODO: use this for a (struct sigevent *) */

        /* flags for the "struct iocb" */
        __u32   aio_flags;

        /*
         * if the IOCB_FLAG_RESFD flag of "aio_flags" is set, this is an
         * eventfd to signal AIO readiness to
         */
        __u32   aio_resfd;
}; /* 64 bytes */

enum {
        IOCB_CMD_PREAD = 0, //pread
        IOCB_CMD_PWRITE = 1, //pwrite
        IOCB_CMD_FSYNC = 2,  // fsync
        IOCB_CMD_FDSYNC = 3, //fdatasync 不同步meta，fsync同步所有
        /* These two are experimental.
         * IOCB_CMD_PREADX = 4,
         * IOCB_CMD_POLL = 5,
         */
        IOCB_CMD_NOOP = 6,
        IOCB_CMD_PREADV = 7,//preadv
        IOCB_CMD_PWRITEV = 8,//pwritev
};