[ceph.git] / ceph / src / test / libradosstriper / striping.cc

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

#include "include/compat.h"
#include "include/types.h"
#include "include/rados/librados.h"
#include "include/rados/librados.hpp"
#include "include/radosstriper/libradosstriper.h"
#include "include/radosstriper/libradosstriper.hpp"
#include "include/ceph_fs.h"
#include "common/backport14.h"
#include "test/librados/test.h"
#include "test/libradosstriper/TestCase.h"

#include <string>
#include <errno.h>
using namespace librados;
using namespace libradosstriper;

class StriperTestRT : public StriperTestParam {
public:
  StriperTestRT() : StriperTestParam() {}
protected:
  char* getObjName(const std::string& soid, uint64_t nb)
  {
    char name[soid.size()+18];
    sprintf(name, "%s.%016llx", soid.c_str(), (long long unsigned int)nb);
    return strdup(name);
  }
  
  void checkObjectFromRados(const std::string& soid, bufferlist &bl,
                            uint64_t exp_stripe_unit, uint64_t exp_stripe_count,
                            uint64_t exp_object_size, size_t size)
  {
    checkObjectFromRados(soid, bl, exp_stripe_unit, exp_stripe_count, exp_object_size, size, size);
  }
      
  void checkObjectFromRados(const std::string& soid, bufferlist &bl,
                            uint64_t exp_stripe_unit, uint64_t exp_stripe_count,
                            uint64_t exp_object_size, size_t size,
                            size_t actual_size_if_sparse)
  {
    // checking first object's rados xattrs
    bufferlist xattrbl;
    char* firstOid = getObjName(soid, 0);
    ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.stripe_unit", xattrbl));
    std::string s_xattr(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
    uint64_t stripe_unit = strtoll(s_xattr.c_str(), NULL, 10);
    ASSERT_LT((unsigned)0, stripe_unit);
    ASSERT_EQ(stripe_unit, exp_stripe_unit);
    xattrbl.clear();
    ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.stripe_count", xattrbl));
    s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
    uint64_t stripe_count = strtoll(s_xattr.c_str(), NULL, 10);
    ASSERT_LT(0U, stripe_count);
    ASSERT_EQ(stripe_count, exp_stripe_count);
    xattrbl.clear();
    ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.object_size", xattrbl));
    s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
    uint64_t object_size = strtoll(s_xattr.c_str(), NULL, 10);
    ASSERT_EQ(object_size, exp_object_size);
    xattrbl.clear();
    ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.size", xattrbl));
    s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
    uint64_t xa_size = strtoll(s_xattr.c_str(), NULL, 10);
    ASSERT_EQ(xa_size, size);
    // checking object content from rados point of view
    // we will go stripe by stripe, read the content of each of them and
    // check with expectations
    uint64_t stripe_per_object = object_size / stripe_unit;
    uint64_t stripe_per_objectset = stripe_per_object * stripe_count;
    uint64_t nb_stripes_in_object = (size+stripe_unit-1)/stripe_unit;
    for (uint64_t stripe_nb = 0;
         stripe_nb < nb_stripes_in_object;
         stripe_nb++) {
      // find out where this stripe is stored
      uint64_t objectset = stripe_nb / stripe_per_objectset;
      uint64_t stripe_in_object_set = stripe_nb % stripe_per_objectset;
      uint64_t object_in_set = stripe_in_object_set % stripe_count;
      uint64_t stripe_in_object = stripe_in_object_set / stripe_count;
      uint64_t object_nb = objectset * stripe_count + object_in_set;
      uint64_t start = stripe_in_object * stripe_unit;
      uint64_t len = stripe_unit;
      if (stripe_nb == nb_stripes_in_object-1 and size % stripe_unit != 0) {
        len = size % stripe_unit;
      }
      // handle case of sparse object (can only be sparse at the end in our tests)
      if (actual_size_if_sparse < size and
          ((actual_size_if_sparse+stripe_unit-1)/stripe_unit)-1 == stripe_nb) {
        len = actual_size_if_sparse % stripe_unit;
        if (0 == len) len = stripe_unit;
      }
      bufferlist stripe_data;
      // check object content
      char* oid = getObjName(soid, object_nb);
      int rc = ioctx.read(oid, stripe_data, len, start);
      if (actual_size_if_sparse < size and
          (actual_size_if_sparse+stripe_unit-1)/stripe_unit <= stripe_nb) {
        // sparse object case : the stripe does not exist, but the rados object may
        uint64_t object_start = (object_in_set + objectset*stripe_per_objectset) * stripe_unit;
        if (actual_size_if_sparse <= object_start) {
          ASSERT_EQ(rc, -ENOENT);
        } else {
          ASSERT_EQ(rc, 0);
        }
      } else {
        ASSERT_EQ((uint64_t)rc, len);
        bufferlist original_data;
        original_data.substr_of(bl, stripe_nb*stripe_unit, len);
        ASSERT_EQ(0, memcmp(original_data.c_str(), stripe_data.c_str(), len));
      }
      free(oid);
    }
    // checking rados object sizes; we go object by object
    uint64_t nb_full_object_sets = nb_stripes_in_object / stripe_per_objectset;
    uint64_t nb_extra_objects = nb_stripes_in_object % stripe_per_objectset;
    if (nb_extra_objects > stripe_count) nb_extra_objects = stripe_count;
    uint64_t nb_objects = nb_full_object_sets * stripe_count + nb_extra_objects;
    for (uint64_t object_nb = 0; object_nb < nb_objects; object_nb++) {
      uint64_t rados_size;
      time_t mtime;
      char* oid = getObjName(soid, object_nb);
      uint64_t nb_full_object_set = object_nb / stripe_count;
      uint64_t object_index_in_set = object_nb % stripe_count;
      uint64_t object_start_stripe = nb_full_object_set * stripe_per_objectset + object_index_in_set;
      uint64_t object_start_off = object_start_stripe * stripe_unit;
      if (actual_size_if_sparse < size and actual_size_if_sparse <= object_start_off) {
        ASSERT_EQ(-ENOENT, ioctx.stat(oid, &rados_size, &mtime));
      } else {
        ASSERT_EQ(0, ioctx.stat(oid, &rados_size, &mtime));
        uint64_t offset;
        uint64_t stripe_size = stripe_count * stripe_unit;
        uint64_t set_size = stripe_count * object_size;
        uint64_t len = 0;
        for (offset = object_start_off;
             (offset < (object_start_off) + set_size) && (offset < actual_size_if_sparse);
             offset += stripe_size) {
          if (offset + stripe_unit > actual_size_if_sparse) {
            len += actual_size_if_sparse-offset;
          } else {
            len += stripe_unit;
          }
        }
        ASSERT_EQ(len, rados_size);
      }
      free(oid);
    }
    // check we do not have an extra object behind
    uint64_t rados_size;
    time_t mtime;
    char* oid = getObjName(soid, nb_objects);
    ASSERT_EQ(-ENOENT, ioctx.stat(oid, &rados_size, &mtime));
    free(oid);
    free(firstOid);
  }
};
  
TEST_P(StriperTestRT, StripedRoundtrip) {
  // get striping parameters and apply them
  TestData testData = GetParam();
  ASSERT_EQ(0, striper.set_object_layout_stripe_unit(testData.stripe_unit));
  ASSERT_EQ(0, striper.set_object_layout_stripe_count(testData.stripe_count));
  ASSERT_EQ(0, striper.set_object_layout_object_size(testData.object_size));
  std::ostringstream oss;
  oss << "StripedRoundtrip_" << testData.stripe_unit << "_"
      << testData.stripe_count << "_" << testData.object_size
      << "_" << testData.size;
  std::string soid = oss.str();
  // writing striped data
  std::unique_ptr<char[]> buf1;
  bufferlist bl1;
  {
    SCOPED_TRACE("Writing initial object"); 
    buf1 = ceph::make_unique<char[]>(testData.size);
    for (unsigned int i = 0; i < testData.size; i++) buf1[i] = 13*((unsigned char)i);
    bl1.append(buf1.get(), testData.size);
    ASSERT_EQ(0, striper.write(soid, bl1, testData.size, 0));
    // checking object state from Rados point of view
    ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
                                                 testData.stripe_count, testData.object_size,
                                                 testData.size));
  }
  // adding more data to object and checking again
  std::unique_ptr<char[]> buf2;
  bufferlist bl2;
  {
    SCOPED_TRACE("Testing append");
    buf2 = ceph::make_unique<char[]>(testData.size);
    for (unsigned int i = 0; i < testData.size; i++) buf2[i] = 17*((unsigned char)i);
    bl2.append(buf2.get(), testData.size);
    ASSERT_EQ(0, striper.append(soid, bl2, testData.size));
    bl1.append(buf2.get(), testData.size);
    ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
                                                 testData.stripe_count, testData.object_size,
                                                 testData.size*2));
  }
  // truncating to half original size and checking again
  {
    SCOPED_TRACE("Testing trunc to truncate object");
    ASSERT_EQ(0, striper.trunc(soid, testData.size/2));
    ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
                                                 testData.stripe_count, testData.object_size,
                                                 testData.size/2));
  }
  // truncating back to original size and checking again (especially for 0s)
  {
    SCOPED_TRACE("Testing trunc to extend object with 0s");
    ASSERT_EQ(0, striper.trunc(soid, testData.size));
    bufferlist bl3;
    bl3.substr_of(bl1, 0, testData.size/2);
    bl3.append_zero(testData.size - testData.size/2);
    ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl3, testData.stripe_unit,
                                                 testData.stripe_count, testData.object_size,
                                                 testData.size, testData.size/2));
  }
  {
    SCOPED_TRACE("Testing write_full");
    // using write_full and checking again
    ASSERT_EQ(0, striper.write_full(soid, bl2));
    checkObjectFromRados(soid, bl2, testData.stripe_unit,
                         testData.stripe_count, testData.object_size,
                         testData.size);
  }
  {
    SCOPED_TRACE("Testing standard remove");
    // call remove
    ASSERT_EQ(0, striper.remove(soid));
    // check that the removal was successful
    uint64_t size;
    time_t mtime;   
    for (uint64_t object_nb = 0;
         object_nb < testData.size*2/testData.object_size + testData.stripe_count;
         object_nb++) {
      char* oid = getObjName(soid, object_nb);
      ASSERT_EQ(-ENOENT, ioctx.stat(oid, &size, &mtime));
      free(oid);
    }
  }
  {
    SCOPED_TRACE("Testing remove when no object size");
    // recreate object
    ASSERT_EQ(0, striper.write(soid, bl1, testData.size*2, 0));
    // remove the object size attribute from the striped object
    char* firstOid = getObjName(soid, 0);
    ASSERT_EQ(0, ioctx.rmxattr(firstOid, "striper.size"));
    free(firstOid);
    // check that stat fails
    uint64_t size;
    time_t mtime;   
    ASSERT_EQ(-ENODATA, striper.stat(soid, &size, &mtime));
    // call remove
    ASSERT_EQ(0, striper.remove(soid));
    // check that the removal was successful
    for (uint64_t object_nb = 0;
         object_nb < testData.size*2/testData.object_size + testData.stripe_count;
         object_nb++) {
      char* oid = getObjName(soid, object_nb);
      ASSERT_EQ(-ENOENT, ioctx.stat(oid, &size, &mtime));
      free(oid);
    }
  }
}

const TestData simple_stripe_schemes[] = {
  // stripe_unit,        stripe_count, object_size,            size
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   2},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 8*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 15*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 25*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 5,            3*CEPH_MIN_STRIPE_UNIT, 45*CEPH_MIN_STRIPE_UNIT+100},
  {262144,               5,            262144,                 2},
  {262144,               5,            262144,                 262144},
  {262144,               5,            262144,                 262144-1},
  {262144,               5,            262144,                 2*262144},
  {262144,               5,            262144,                 12*262144},
  {262144,               5,            262144,                 2*262144-1},
  {262144,               5,            262144,                 12*262144-1},
  {262144,               5,            262144,                 2*262144+100},
  {262144,               5,            262144,                 12*262144+100},
  {262144,               5,            3*262144,               2*262144+100},
  {262144,               5,            3*262144,               8*262144+100},
  {262144,               5,            3*262144,               12*262144+100},
  {262144,               5,            3*262144,               15*262144+100},
  {262144,               5,            3*262144,               25*262144+100},
  {262144,               5,            3*262144,               45*262144+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   2},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 8*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 15*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 25*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 1,            3*CEPH_MIN_STRIPE_UNIT, 45*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   2},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT-1},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           CEPH_MIN_STRIPE_UNIT,   12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 8*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 15*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 25*CEPH_MIN_STRIPE_UNIT+100},
  {CEPH_MIN_STRIPE_UNIT, 50,           3*CEPH_MIN_STRIPE_UNIT, 45*CEPH_MIN_STRIPE_UNIT+100}
};

INSTANTIATE_TEST_CASE_P(SimpleStriping,
                        StriperTestRT,
                        ::testing::ValuesIn(simple_stripe_schemes));
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3
31f18b77	4	#include "include/compat.h"
7c673cae FG	5	#include "include/types.h"
	6	#include "include/rados/librados.h"
	7	#include "include/rados/librados.hpp"
	8	#include "include/radosstriper/libradosstriper.h"
	9	#include "include/radosstriper/libradosstriper.hpp"
	10	#include "include/ceph_fs.h"
224ce89b	11	#include "common/backport14.h"
7c673cae FG	12	#include "test/librados/test.h"
	13	#include "test/libradosstriper/TestCase.h"
	14
	15	#include <string>
	16	#include <errno.h>
	17	using namespace librados;
	18	using namespace libradosstriper;
	19
	20	class StriperTestRT : public StriperTestParam {
	21	public:
	22	StriperTestRT() : StriperTestParam() {}
	23	protected:
	24	char* getObjName(const std::string& soid, uint64_t nb)
	25	{
	26	char name[soid.size()+18];
	27	sprintf(name, "%s.%016llx", soid.c_str(), (long long unsigned int)nb);
	28	return strdup(name);
	29	}
	30
	31	void checkObjectFromRados(const std::string& soid, bufferlist &bl,
	32	uint64_t exp_stripe_unit, uint64_t exp_stripe_count,
	33	uint64_t exp_object_size, size_t size)
	34	{
	35	checkObjectFromRados(soid, bl, exp_stripe_unit, exp_stripe_count, exp_object_size, size, size);
	36	}
	37
	38	void checkObjectFromRados(const std::string& soid, bufferlist &bl,
	39	uint64_t exp_stripe_unit, uint64_t exp_stripe_count,
	40	uint64_t exp_object_size, size_t size,
	41	size_t actual_size_if_sparse)
	42	{
	43	// checking first object's rados xattrs
	44	bufferlist xattrbl;
	45	char* firstOid = getObjName(soid, 0);
	46	ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.stripe_unit", xattrbl));
	47	std::string s_xattr(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
	48	uint64_t stripe_unit = strtoll(s_xattr.c_str(), NULL, 10);
	49	ASSERT_LT((unsigned)0, stripe_unit);
	50	ASSERT_EQ(stripe_unit, exp_stripe_unit);
	51	xattrbl.clear();
	52	ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.stripe_count", xattrbl));
	53	s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
	54	uint64_t stripe_count = strtoll(s_xattr.c_str(), NULL, 10);
	55	ASSERT_LT(0U, stripe_count);
	56	ASSERT_EQ(stripe_count, exp_stripe_count);
	57	xattrbl.clear();
	58	ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.layout.object_size", xattrbl));
	59	s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
	60	uint64_t object_size = strtoll(s_xattr.c_str(), NULL, 10);
	61	ASSERT_EQ(object_size, exp_object_size);
	62	xattrbl.clear();
	63	ASSERT_LT(0, ioctx.getxattr(firstOid, "striper.size", xattrbl));
	64	s_xattr = std::string(xattrbl.c_str(), xattrbl.length()); // adds 0 byte at the end
	65	uint64_t xa_size = strtoll(s_xattr.c_str(), NULL, 10);
	66	ASSERT_EQ(xa_size, size);
	67	// checking object content from rados point of view
	68	// we will go stripe by stripe, read the content of each of them and
	69	// check with expectations
	70	uint64_t stripe_per_object = object_size / stripe_unit;
	71	uint64_t stripe_per_objectset = stripe_per_object * stripe_count;
	72	uint64_t nb_stripes_in_object = (size+stripe_unit-1)/stripe_unit;
	73	for (uint64_t stripe_nb = 0;
	74	stripe_nb < nb_stripes_in_object;
	75	stripe_nb++) {
76	// find out where this stripe is stored
77	uint64_t objectset = stripe_nb / stripe_per_objectset;
78	uint64_t stripe_in_object_set = stripe_nb % stripe_per_objectset;
79	uint64_t object_in_set = stripe_in_object_set % stripe_count;
80	uint64_t stripe_in_object = stripe_in_object_set / stripe_count;
81	uint64_t object_nb = objectset * stripe_count + object_in_set;
82	uint64_t start = stripe_in_object * stripe_unit;
83	uint64_t len = stripe_unit;
84	if (stripe_nb == nb_stripes_in_object-1 and size % stripe_unit != 0) {
85	len = size % stripe_unit;
86	}
87	// handle case of sparse object (can only be sparse at the end in our tests)
88	if (actual_size_if_sparse < size and
89	((actual_size_if_sparse+stripe_unit-1)/stripe_unit)-1 == stripe_nb) {
90	len = actual_size_if_sparse % stripe_unit;
91	if (0 == len) len = stripe_unit;
92	}
93	bufferlist stripe_data;
94	// check object content
95	char* oid = getObjName(soid, object_nb);
96	int rc = ioctx.read(oid, stripe_data, len, start);
97	if (actual_size_if_sparse < size and
98	(actual_size_if_sparse+stripe_unit-1)/stripe_unit <= stripe_nb) {
99	// sparse object case : the stripe does not exist, but the rados object may
100	uint64_t object_start = (object_in_set + objectsetstripe_per_objectset) stripe_unit;
101	if (actual_size_if_sparse <= object_start) {
102	ASSERT_EQ(rc, -ENOENT);
103	} else {
104	ASSERT_EQ(rc, 0);
105	}
106	} else {
107	ASSERT_EQ((uint64_t)rc, len);
108	bufferlist original_data;
109	original_data.substr_of(bl, stripe_nb*stripe_unit, len);
110	ASSERT_EQ(0, memcmp(original_data.c_str(), stripe_data.c_str(), len));
111	}
112	free(oid);
113	}
114	// checking rados object sizes; we go object by object
115	uint64_t nb_full_object_sets = nb_stripes_in_object / stripe_per_objectset;
116	uint64_t nb_extra_objects = nb_stripes_in_object % stripe_per_objectset;
117	if (nb_extra_objects > stripe_count) nb_extra_objects = stripe_count;
118	uint64_t nb_objects = nb_full_object_sets * stripe_count + nb_extra_objects;
119	for (uint64_t object_nb = 0; object_nb < nb_objects; object_nb++) {
120	uint64_t rados_size;
121	time_t mtime;
122	char* oid = getObjName(soid, object_nb);
123	uint64_t nb_full_object_set = object_nb / stripe_count;
124	uint64_t object_index_in_set = object_nb % stripe_count;
125	uint64_t object_start_stripe = nb_full_object_set * stripe_per_objectset + object_index_in_set;
126	uint64_t object_start_off = object_start_stripe * stripe_unit;
127	if (actual_size_if_sparse < size and actual_size_if_sparse <= object_start_off) {
128	ASSERT_EQ(-ENOENT, ioctx.stat(oid, &rados_size, &mtime));
129	} else {
130	ASSERT_EQ(0, ioctx.stat(oid, &rados_size, &mtime));
131	uint64_t offset;
132	uint64_t stripe_size = stripe_count * stripe_unit;
133	uint64_t set_size = stripe_count * object_size;
134	uint64_t len = 0;
135	for (offset = object_start_off;
136	(offset < (object_start_off) + set_size) && (offset < actual_size_if_sparse);
137	offset += stripe_size) {
138	if (offset + stripe_unit > actual_size_if_sparse) {
139	len += actual_size_if_sparse-offset;
140	} else {
141	len += stripe_unit;
142	}
143	}
144	ASSERT_EQ(len, rados_size);
145	}
146	free(oid);
147	}
148	// check we do not have an extra object behind
149	uint64_t rados_size;
150	time_t mtime;
151	char* oid = getObjName(soid, nb_objects);
152	ASSERT_EQ(-ENOENT, ioctx.stat(oid, &rados_size, &mtime));
153	free(oid);
154	free(firstOid);
155	}
156	};
157
158	TEST_P(StriperTestRT, StripedRoundtrip) {
159	// get striping parameters and apply them
160	TestData testData = GetParam();
161	ASSERT_EQ(0, striper.set_object_layout_stripe_unit(testData.stripe_unit));
162	ASSERT_EQ(0, striper.set_object_layout_stripe_count(testData.stripe_count));
163	ASSERT_EQ(0, striper.set_object_layout_object_size(testData.object_size));
164	std::ostringstream oss;
165	oss << "StripedRoundtrip_" << testData.stripe_unit << "_"
166	<< testData.stripe_count << "_" << testData.object_size
167	<< "_" << testData.size;
168	std::string soid = oss.str();
169	// writing striped data
224ce89b	170	std::unique_ptr<char[]> buf1;
7c673cae FG	171	bufferlist bl1;
	172	{
	173	SCOPED_TRACE("Writing initial object");
224ce89b	174	buf1 = ceph::make_unique<char[]>(testData.size);
7c673cae	175	for (unsigned int i = 0; i < testData.size; i++) buf1[i] = 13*((unsigned char)i);
224ce89b	176	bl1.append(buf1.get(), testData.size);
7c673cae FG	177	ASSERT_EQ(0, striper.write(soid, bl1, testData.size, 0));
	178	// checking object state from Rados point of view
	179	ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
	180	testData.stripe_count, testData.object_size,
	181	testData.size));
	182	}
	183	// adding more data to object and checking again
224ce89b	184	std::unique_ptr<char[]> buf2;
7c673cae FG	185	bufferlist bl2;
	186	{
	187	SCOPED_TRACE("Testing append");
224ce89b	188	buf2 = ceph::make_unique<char[]>(testData.size);
7c673cae	189	for (unsigned int i = 0; i < testData.size; i++) buf2[i] = 17*((unsigned char)i);
224ce89b	190	bl2.append(buf2.get(), testData.size);
7c673cae	191	ASSERT_EQ(0, striper.append(soid, bl2, testData.size));
224ce89b	192	bl1.append(buf2.get(), testData.size);
7c673cae FG	193	ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
	194	testData.stripe_count, testData.object_size,
	195	testData.size*2));
	196	}
	197	// truncating to half original size and checking again
	198	{
	199	SCOPED_TRACE("Testing trunc to truncate object");
	200	ASSERT_EQ(0, striper.trunc(soid, testData.size/2));
	201	ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl1, testData.stripe_unit,
	202	testData.stripe_count, testData.object_size,
	203	testData.size/2));
	204	}
	205	// truncating back to original size and checking again (especially for 0s)
	206	{
	207	SCOPED_TRACE("Testing trunc to extend object with 0s");
	208	ASSERT_EQ(0, striper.trunc(soid, testData.size));
	209	bufferlist bl3;
	210	bl3.substr_of(bl1, 0, testData.size/2);
	211	bl3.append_zero(testData.size - testData.size/2);
	212	ASSERT_NO_FATAL_FAILURE(checkObjectFromRados(soid, bl3, testData.stripe_unit,
	213	testData.stripe_count, testData.object_size,
	214	testData.size, testData.size/2));
	215	}
	216	{
	217	SCOPED_TRACE("Testing write_full");
	218	// using write_full and checking again
	219	ASSERT_EQ(0, striper.write_full(soid, bl2));
	220	checkObjectFromRados(soid, bl2, testData.stripe_unit,
	221	testData.stripe_count, testData.object_size,
	222	testData.size);
	223	}
	224	{
	225	SCOPED_TRACE("Testing standard remove");
	226	// call remove
	227	ASSERT_EQ(0, striper.remove(soid));
	228	// check that the removal was successful
	229	uint64_t size;
	230	time_t mtime;
	231	for (uint64_t object_nb = 0;
	232	object_nb < testData.size*2/testData.object_size + testData.stripe_count;
	233	object_nb++) {
	234	char* oid = getObjName(soid, object_nb);
	235	ASSERT_EQ(-ENOENT, ioctx.stat(oid, &size, &mtime));
	236	free(oid);
	237	}
	238	}
	239	{
	240	SCOPED_TRACE("Testing remove when no object size");
	241	// recreate object
	242	ASSERT_EQ(0, striper.write(soid, bl1, testData.size*2, 0));
	243	// remove the object size attribute from the striped object
	244	char* firstOid = getObjName(soid, 0);
	245	ASSERT_EQ(0, ioctx.rmxattr(firstOid, "striper.size"));
	246	free(firstOid);
	247	// check that stat fails
	248	uint64_t size;
	249	time_t mtime;
	250	ASSERT_EQ(-ENODATA, striper.stat(soid, &size, &mtime));
	251	// call remove
	252	ASSERT_EQ(0, striper.remove(soid));
	253	// check that the removal was successful
	254	for (uint64_t object_nb = 0;
	255	object_nb < testData.size*2/testData.object_size + testData.stripe_count;
	256	object_nb++) {
257	char* oid = getObjName(soid, object_nb);
258	ASSERT_EQ(-ENOENT, ioctx.stat(oid, &size, &mtime));
259	free(oid);
260	}
261	}
7c673cae FG	262	}
	263
	264	const TestData simple_stripe_schemes[] = {
	265	// stripe_unit, stripe_count, object_size, size
	266	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 2},
	267	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT},
	268	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT-1},
	269	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT},
	270	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT},
	271	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT-1},
	272	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT-1},
	273	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
	274	{CEPH_MIN_STRIPE_UNIT, 5, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
	275	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 2CEPH_MIN_STRIPE_UNIT+100},
	276	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 8CEPH_MIN_STRIPE_UNIT+100},
	277	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 12CEPH_MIN_STRIPE_UNIT+100},
	278	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 15CEPH_MIN_STRIPE_UNIT+100},
	279	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 25CEPH_MIN_STRIPE_UNIT+100},
	280	{CEPH_MIN_STRIPE_UNIT, 5, 3CEPH_MIN_STRIPE_UNIT, 45CEPH_MIN_STRIPE_UNIT+100},
	281	{262144, 5, 262144, 2},
	282	{262144, 5, 262144, 262144},
	283	{262144, 5, 262144, 262144-1},
	284	{262144, 5, 262144, 2*262144},
	285	{262144, 5, 262144, 12*262144},
	286	{262144, 5, 262144, 2*262144-1},
	287	{262144, 5, 262144, 12*262144-1},
	288	{262144, 5, 262144, 2*262144+100},
	289	{262144, 5, 262144, 12*262144+100},
	290	{262144, 5, 3262144, 2262144+100},
	291	{262144, 5, 3262144, 8262144+100},
	292	{262144, 5, 3262144, 12262144+100},
	293	{262144, 5, 3262144, 15262144+100},
	294	{262144, 5, 3262144, 25262144+100},
	295	{262144, 5, 3262144, 45262144+100},
	296	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 2},
	297	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT},
	298	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT-1},
	299	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT},
	300	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT},
	301	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT-1},
	302	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT-1},
	303	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
	304	{CEPH_MIN_STRIPE_UNIT, 1, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
	305	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 2CEPH_MIN_STRIPE_UNIT+100},
	306	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 8CEPH_MIN_STRIPE_UNIT+100},
	307	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 12CEPH_MIN_STRIPE_UNIT+100},
	308	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 15CEPH_MIN_STRIPE_UNIT+100},
	309	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 25CEPH_MIN_STRIPE_UNIT+100},
	310	{CEPH_MIN_STRIPE_UNIT, 1, 3CEPH_MIN_STRIPE_UNIT, 45CEPH_MIN_STRIPE_UNIT+100},
	311	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 2},
	312	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT},
	313	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, CEPH_MIN_STRIPE_UNIT-1},
	314	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT},
	315	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT},
	316	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT-1},
	317	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT-1},
	318	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 2*CEPH_MIN_STRIPE_UNIT+100},
	319	{CEPH_MIN_STRIPE_UNIT, 50, CEPH_MIN_STRIPE_UNIT, 12*CEPH_MIN_STRIPE_UNIT+100},
	320	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 2CEPH_MIN_STRIPE_UNIT+100},
	321	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 8CEPH_MIN_STRIPE_UNIT+100},
	322	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 12CEPH_MIN_STRIPE_UNIT+100},
	323	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 15CEPH_MIN_STRIPE_UNIT+100},
	324	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 25CEPH_MIN_STRIPE_UNIT+100},
	325	{CEPH_MIN_STRIPE_UNIT, 50, 3CEPH_MIN_STRIPE_UNIT, 45CEPH_MIN_STRIPE_UNIT+100}
326	};
327
328	INSTANTIATE_TEST_CASE_P(SimpleStriping,
329	StriperTestRT,
330	::testing::ValuesIn(simple_stripe_schemes));