[ceph.git] / ceph / src / rgw / driver / rados / rgw_d3n_datacache.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab ft=cpp

#pragma once

#include "rgw_rados.h"
#include <curl/curl.h>

#include "rgw_common.h"

#include <unistd.h>
#include <signal.h>
#include "include/Context.h"
#include "include/lru.h"
#include "rgw_d3n_cacherequest.h"


/*D3nDataCache*/
struct D3nDataCache;


struct D3nChunkDataInfo : public LRUObject {
	CephContext *cct;
	uint64_t size;
	time_t access_time;
	std::string address;
	std::string oid;
	bool complete;
	struct D3nChunkDataInfo* lru_prev;
	struct D3nChunkDataInfo* lru_next;

	D3nChunkDataInfo(): size(0) {}

	void set_ctx(CephContext *_cct) {
		cct = _cct;
	}

	void dump(Formatter *f) const;
	static void generate_test_instances(std::list<D3nChunkDataInfo*>& o);
};

struct D3nCacheAioWriteRequest {
	std::string oid;
	void *data;
	int fd;
	struct aiocb *cb;
	D3nDataCache *priv_data;
	CephContext *cct;

	D3nCacheAioWriteRequest(CephContext *_cct) : cct(_cct) {}
	int d3n_libaio_prepare_write_op(bufferlist& bl, unsigned int len, std::string oid, std::string cache_location);

  ~D3nCacheAioWriteRequest() {
    ::close(fd);
		cb->aio_buf = nullptr;
		free(data);
		data = nullptr;
		delete(cb);
  }
};

struct D3nDataCache {

private:
  std::unordered_map<std::string, D3nChunkDataInfo*> d3n_cache_map;
  std::set<std::string> d3n_outstanding_write_list;
  std::mutex d3n_cache_lock;
  std::mutex d3n_eviction_lock;

  CephContext *cct;
  enum class _io_type {
    SYNC_IO = 1,
    ASYNC_IO = 2,
    SEND_FILE = 3
  } io_type;
  enum class _eviction_policy {
    LRU=0, RANDOM=1
  } eviction_policy;

  struct sigaction action;
  uint64_t free_data_cache_size = 0;
  uint64_t outstanding_write_size = 0;
  struct D3nChunkDataInfo* head;
  struct D3nChunkDataInfo* tail;

private:
  void add_io();

public:
  D3nDataCache();
  ~D3nDataCache() {
    while (lru_eviction() > 0);
  }

  std::string cache_location;

  bool get(const std::string& oid, const off_t len);
  void put(bufferlist& bl, unsigned int len, std::string& obj_key);
  int d3n_io_write(bufferlist& bl, unsigned int len, std::string oid);
  int d3n_libaio_create_write_request(bufferlist& bl, unsigned int len, std::string oid);
  void d3n_libaio_write_completion_cb(D3nCacheAioWriteRequest* c);
  size_t random_eviction();
  size_t lru_eviction();

  void init(CephContext *_cct);

  void lru_insert_head(struct D3nChunkDataInfo* o) {
    lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
    o->lru_next = head;
    o->lru_prev = nullptr;
    if (head) {
      head->lru_prev = o;
    } else {
      tail = o;
    }
    head = o;
  }

  void lru_insert_tail(struct D3nChunkDataInfo* o) {
    lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
    o->lru_next = nullptr;
    o->lru_prev = tail;
    if (tail) {
      tail->lru_next = o;
    } else {
      head = o;
    }
    tail = o;
  }

  void lru_remove(struct D3nChunkDataInfo* o) {
    lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
    if (o->lru_next)
      o->lru_next->lru_prev = o->lru_prev;
    else
      tail = o->lru_prev;
    if (o->lru_prev)
      o->lru_prev->lru_next = o->lru_next;
    else
      head = o->lru_next;
    o->lru_next = o->lru_prev = nullptr;
  }
};


template <class T>
class D3nRGWDataCache : public T {

public:
  D3nRGWDataCache() {}

  int init_rados() override {
    int ret;
    ret = T::init_rados();
    if (ret < 0)
      return ret;

    return 0;
  }

  int get_obj_iterate_cb(const DoutPrefixProvider *dpp, const rgw_raw_obj& read_obj, off_t obj_ofs,
                         off_t read_ofs, off_t len, bool is_head_obj,
                         RGWObjState *astate, void *arg) override;
};

template<typename T>
int D3nRGWDataCache<T>::get_obj_iterate_cb(const DoutPrefixProvider *dpp, const rgw_raw_obj& read_obj, off_t obj_ofs,
                                 off_t read_ofs, off_t len, bool is_head_obj,
                                 RGWObjState *astate, void *arg) {
  lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache::" << __func__ << "(): is head object : " << is_head_obj << dendl;
  librados::ObjectReadOperation op;
  struct get_obj_data* d = static_cast<struct get_obj_data*>(arg);
  std::string oid, key;

  if (is_head_obj) {
    // only when reading from the head object do we need to do the atomic test
    int r = T::append_atomic_test(dpp, astate, op);
    if (r < 0)
      return r;

    if (astate &&
        obj_ofs < astate->data.length()) {
      unsigned chunk_len = std::min((uint64_t)astate->data.length() - obj_ofs, (uint64_t)len);

      r = d->client_cb->handle_data(astate->data, obj_ofs, chunk_len);
      if (r < 0)
        return r;

      len -= chunk_len;
      d->offset += chunk_len;
      read_ofs += chunk_len;
      obj_ofs += chunk_len;
      if (!len)
        return 0;
    }

    auto obj = d->rgwrados->svc.rados->obj(read_obj);
    r = obj.open(dpp);
    if (r < 0) {
      lsubdout(g_ceph_context, rgw, 4) << "failed to open rados context for " << read_obj << dendl;
      return r;
    }

    ldpp_dout(dpp, 20) << "D3nDataCache::" << __func__ << "(): oid=" << read_obj.oid << " obj-ofs=" << obj_ofs << " read_ofs=" << read_ofs << " len=" << len << dendl;
    op.read(read_ofs, len, nullptr, nullptr);

    const uint64_t cost = len;
    const uint64_t id = obj_ofs; // use logical object offset for sorting replies

    auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
    return d->flush(std::move(completed));
  } else {
    ldpp_dout(dpp, 20) << "D3nDataCache::" << __func__ << "(): oid=" << read_obj.oid << ", is_head_obj=" << is_head_obj << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << ", len=" << len << dendl;
    int r;

    op.read(read_ofs, len, nullptr, nullptr);

    const uint64_t cost = len;
    const uint64_t id = obj_ofs; // use logical object offset for sorting replies
    oid = read_obj.oid;

    auto obj = d->rgwrados->svc.rados->obj(read_obj);
    r = obj.open(dpp);
    if (r < 0) {
      lsubdout(g_ceph_context, rgw, 0) << "D3nDataCache: Error: failed to open rados context for " << read_obj << ", r=" << r << dendl;
      return r;
    }

    const bool is_compressed = (astate->attrset.find(RGW_ATTR_COMPRESSION) != astate->attrset.end());
    const bool is_encrypted = (astate->attrset.find(RGW_ATTR_CRYPT_MODE) != astate->attrset.end());
    if (read_ofs != 0 || astate->size != astate->accounted_size || is_compressed || is_encrypted) {
      d->d3n_bypass_cache_write = true;
      lsubdout(g_ceph_context, rgw, 5) << "D3nDataCache: " << __func__ << "(): Note - bypassing datacache: oid=" << read_obj.oid << ", read_ofs!=0 = " << read_ofs << ", size=" << astate->size << " != accounted_size=" << astate->accounted_size << ", is_compressed=" << is_compressed << ", is_encrypted=" << is_encrypted  << dendl;
      auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
      r = d->flush(std::move(completed));
      return r;
    }

    if (d->rgwrados->d3n_data_cache->get(oid, len)) {
      // Read From Cache
      ldpp_dout(dpp, 20) << "D3nDataCache: " << __func__ << "(): READ FROM CACHE: oid=" << read_obj.oid << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << ", len=" << len << dendl;
      auto completed = d->aio->get(obj, rgw::Aio::d3n_cache_op(dpp, d->yield, read_ofs, len, d->rgwrados->d3n_data_cache->cache_location), cost, id);
      r = d->flush(std::move(completed));
      if (r < 0) {
        lsubdout(g_ceph_context, rgw, 0) << "D3nDataCache: " << __func__ << "(): Error: failed to drain/flush, r= " << r << dendl;
      }
      return r;
    } else {
      // Write To Cache
      ldpp_dout(dpp, 20) << "D3nDataCache: " << __func__ << "(): WRITE TO CACHE: oid=" << read_obj.oid << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << " len=" << len << dendl;
      auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
      return d->flush(std::move(completed));
    }
  }
  lsubdout(g_ceph_context, rgw, 1) << "D3nDataCache: " << __func__ << "(): Warning: Check head object cache handling flow, oid=" << read_obj.oid << dendl;

  return 0;
}
Commit	Line	Data
20effc67 TL	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab ft=cpp
	3
1e59de90	4	#pragma once
20effc67 TL	5
	6	#include "rgw_rados.h"
	7	#include <curl/curl.h>
	8
	9	#include "rgw_common.h"
	10
	11	#include <unistd.h>
	12	#include <signal.h>
	13	#include "include/Context.h"
	14	#include "include/lru.h"
	15	#include "rgw_d3n_cacherequest.h"
	16
	17
	18	/D3nDataCache/
	19	struct D3nDataCache;
	20
	21
	22	struct D3nChunkDataInfo : public LRUObject {
	23	CephContext *cct;
	24	uint64_t size;
	25	time_t access_time;
	26	std::string address;
	27	std::string oid;
	28	bool complete;
	29	struct D3nChunkDataInfo* lru_prev;
	30	struct D3nChunkDataInfo* lru_next;
	31
	32	D3nChunkDataInfo(): size(0) {}
	33
	34	void set_ctx(CephContext *_cct) {
	35	cct = _cct;
	36	}
	37
	38	void dump(Formatter *f) const;
	39	static void generate_test_instances(std::list<D3nChunkDataInfo*>& o);
	40	};
	41
	42	struct D3nCacheAioWriteRequest {
	43	std::string oid;
	44	void *data;
	45	int fd;
	46	struct aiocb *cb;
	47	D3nDataCache *priv_data;
	48	CephContext *cct;
	49
	50	D3nCacheAioWriteRequest(CephContext *_cct) : cct(_cct) {}
05a536ef	51	int d3n_libaio_prepare_write_op(bufferlist& bl, unsigned int len, std::string oid, std::string cache_location);
20effc67 TL	52
	53	~D3nCacheAioWriteRequest() {
	54	::close(fd);
	55	cb->aio_buf = nullptr;
	56	free(data);
	57	data = nullptr;
	58	delete(cb);
	59	}
	60	};
	61
	62	struct D3nDataCache {
	63
	64	private:
	65	std::unordered_map<std::string, D3nChunkDataInfo*> d3n_cache_map;
	66	std::set<std::string> d3n_outstanding_write_list;
	67	std::mutex d3n_cache_lock;
	68	std::mutex d3n_eviction_lock;
	69
	70	CephContext *cct;
	71	enum class _io_type {
	72	SYNC_IO = 1,
	73	ASYNC_IO = 2,
	74	SEND_FILE = 3
	75	} io_type;
	76	enum class _eviction_policy {
	77	LRU=0, RANDOM=1
	78	} eviction_policy;
	79
	80	struct sigaction action;
	81	uint64_t free_data_cache_size = 0;
	82	uint64_t outstanding_write_size = 0;
	83	struct D3nChunkDataInfo* head;
	84	struct D3nChunkDataInfo* tail;
	85
	86	private:
	87	void add_io();
	88
	89	public:
	90	D3nDataCache();
	91	~D3nDataCache() {
	92	while (lru_eviction() > 0);
	93	}
	94
	95	std::string cache_location;
	96
	97	bool get(const std::string& oid, const off_t len);
	98	void put(bufferlist& bl, unsigned int len, std::string& obj_key);
	99	int d3n_io_write(bufferlist& bl, unsigned int len, std::string oid);
	100	int d3n_libaio_create_write_request(bufferlist& bl, unsigned int len, std::string oid);
	101	void d3n_libaio_write_completion_cb(D3nCacheAioWriteRequest* c);
	102	size_t random_eviction();
	103	size_t lru_eviction();
	104
	105	void init(CephContext *_cct);
	106
	107	void lru_insert_head(struct D3nChunkDataInfo* o) {
	108	lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
	109	o->lru_next = head;
	110	o->lru_prev = nullptr;
	111	if (head) {
	112	head->lru_prev = o;
	113	} else {
	114	tail = o;
	115	}
116	head = o;
117	}
118
119	void lru_insert_tail(struct D3nChunkDataInfo* o) {
120	lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
121	o->lru_next = nullptr;
122	o->lru_prev = tail;
123	if (tail) {
124	tail->lru_next = o;
125	} else {
126	head = o;
127	}
128	tail = o;
129	}
130
131	void lru_remove(struct D3nChunkDataInfo* o) {
132	lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache: " << __func__ << "()" << dendl;
133	if (o->lru_next)
134	o->lru_next->lru_prev = o->lru_prev;
135	else
136	tail = o->lru_prev;
137	if (o->lru_prev)
138	o->lru_prev->lru_next = o->lru_next;
139	else
140	head = o->lru_next;
141	o->lru_next = o->lru_prev = nullptr;
142	}
143	};
144
145
146	template <class T>
147	class D3nRGWDataCache : public T {
148
149	public:
150	D3nRGWDataCache() {}
151
152	int init_rados() override {
153	int ret;
154	ret = T::init_rados();
155	if (ret < 0)
156	return ret;
157
158	return 0;
159	}
160
161	int get_obj_iterate_cb(const DoutPrefixProvider *dpp, const rgw_raw_obj& read_obj, off_t obj_ofs,
162	off_t read_ofs, off_t len, bool is_head_obj,
163	RGWObjState astate, void arg) override;
164	};
165
166	template<typename T>
167	int D3nRGWDataCache<T>::get_obj_iterate_cb(const DoutPrefixProvider *dpp, const rgw_raw_obj& read_obj, off_t obj_ofs,
168	off_t read_ofs, off_t len, bool is_head_obj,
169	RGWObjState astate, void arg) {
170	lsubdout(g_ceph_context, rgw_datacache, 30) << "D3nDataCache::" << __func__ << "(): is head object : " << is_head_obj << dendl;
171	librados::ObjectReadOperation op;
172	struct get_obj_data* d = static_cast<struct get_obj_data*>(arg);
173	std::string oid, key;
174
175	if (is_head_obj) {
176	// only when reading from the head object do we need to do the atomic test
177	int r = T::append_atomic_test(dpp, astate, op);
178	if (r < 0)
179	return r;
180
181	if (astate &&
182	obj_ofs < astate->data.length()) {
183	unsigned chunk_len = std::min((uint64_t)astate->data.length() - obj_ofs, (uint64_t)len);
184
185	r = d->client_cb->handle_data(astate->data, obj_ofs, chunk_len);
186	if (r < 0)
187	return r;
188
189	len -= chunk_len;
190	d->offset += chunk_len;
191	read_ofs += chunk_len;
192	obj_ofs += chunk_len;
193	if (!len)
194	return 0;
195	}
196
197	auto obj = d->rgwrados->svc.rados->obj(read_obj);
198	r = obj.open(dpp);
199	if (r < 0) {
200	lsubdout(g_ceph_context, rgw, 4) << "failed to open rados context for " << read_obj << dendl;
201	return r;
202	}
203
204	ldpp_dout(dpp, 20) << "D3nDataCache::" << __func__ << "(): oid=" << read_obj.oid << " obj-ofs=" << obj_ofs << " read_ofs=" << read_ofs << " len=" << len << dendl;
205	op.read(read_ofs, len, nullptr, nullptr);
206
207	const uint64_t cost = len;
208	const uint64_t id = obj_ofs; // use logical object offset for sorting replies
209
210	auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
211	return d->flush(std::move(completed));
212	} else {
213	ldpp_dout(dpp, 20) << "D3nDataCache::" << __func__ << "(): oid=" << read_obj.oid << ", is_head_obj=" << is_head_obj << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << ", len=" << len << dendl;
214	int r;
215
216	op.read(read_ofs, len, nullptr, nullptr);
217
218	const uint64_t cost = len;
219	const uint64_t id = obj_ofs; // use logical object offset for sorting replies
220	oid = read_obj.oid;
221
222	auto obj = d->rgwrados->svc.rados->obj(read_obj);
223	r = obj.open(dpp);
224	if (r < 0) {
225	lsubdout(g_ceph_context, rgw, 0) << "D3nDataCache: Error: failed to open rados context for " << read_obj << ", r=" << r << dendl;
226	return r;
227	}
228
229	const bool is_compressed = (astate->attrset.find(RGW_ATTR_COMPRESSION) != astate->attrset.end());
230	const bool is_encrypted = (astate->attrset.find(RGW_ATTR_CRYPT_MODE) != astate->attrset.end());
231	if (read_ofs != 0 \|\| astate->size != astate->accounted_size \|\| is_compressed \|\| is_encrypted) {
232	d->d3n_bypass_cache_write = true;
233	lsubdout(g_ceph_context, rgw, 5) << "D3nDataCache: " << __func__ << "(): Note - bypassing datacache: oid=" << read_obj.oid << ", read_ofs!=0 = " << read_ofs << ", size=" << astate->size << " != accounted_size=" << astate->accounted_size << ", is_compressed=" << is_compressed << ", is_encrypted=" << is_encrypted << dendl;
234	auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
235	r = d->flush(std::move(completed));
236	return r;
237	}
238
239	if (d->rgwrados->d3n_data_cache->get(oid, len)) {
240	// Read From Cache
241	ldpp_dout(dpp, 20) << "D3nDataCache: " << __func__ << "(): READ FROM CACHE: oid=" << read_obj.oid << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << ", len=" << len << dendl;
242	auto completed = d->aio->get(obj, rgw::Aio::d3n_cache_op(dpp, d->yield, read_ofs, len, d->rgwrados->d3n_data_cache->cache_location), cost, id);
243	r = d->flush(std::move(completed));
244	if (r < 0) {
245	lsubdout(g_ceph_context, rgw, 0) << "D3nDataCache: " << __func__ << "(): Error: failed to drain/flush, r= " << r << dendl;
246	}
247	return r;
248	} else {
249	// Write To Cache
250	ldpp_dout(dpp, 20) << "D3nDataCache: " << __func__ << "(): WRITE TO CACHE: oid=" << read_obj.oid << ", obj-ofs=" << obj_ofs << ", read_ofs=" << read_ofs << " len=" << len << dendl;
251	auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
252	return d->flush(std::move(completed));
253	}
254	}
255	lsubdout(g_ceph_context, rgw, 1) << "D3nDataCache: " << __func__ << "(): Warning: Check head object cache handling flow, oid=" << read_obj.oid << dendl;
256
257	return 0;
258	}
259