erasure-code: enforce chunk size alignment (giant)

src/common/buffer.cc

@@ -1116,11 +1116,17 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
   }
 
 void buffer::list::rebuild_aligned(unsigned align)
+{
+  rebuild_aligned_size_and_memory(align, align);
+}
+
+void buffer::list::rebuild_aligned_size_and_memory(unsigned align_size,
+						   unsigned align_memory)
 {
   std::list<ptr>::iterator p = _buffers.begin();
   while (p != _buffers.end()) {
     // keep anything that's already align and sized aligned
-    if (p->is_aligned(align) && p->is_n_align_sized(align)) {
+    if (p->is_aligned(align_memory) && p->is_n_align_sized(align_size)) {
       /*cout << " segment " << (void*)p->c_str()
 	     << " offset " << ((unsigned long)p->c_str() & (align - 1))
 	     << " length " << p->length()
@@ -1144,11 +1150,11 @@ void buffer::list::rebuild_aligned(unsigned align)
       unaligned.push_back(*p);
       _buffers.erase(p++);
     } while (p != _buffers.end() &&
-	     (!p->is_aligned(align) ||
-	      !p->is_n_align_sized(align) ||
-	      (offset & (align-1))));
-    if (!(unaligned.is_contiguous() && unaligned._buffers.front().is_aligned(align))) {
-      ptr nb(buffer::create_aligned(unaligned._len, align));
+	     (!p->is_aligned(align_memory) ||
+	      !p->is_n_align_sized(align_size) ||
+	      (offset % align_size)));
+    if (!(unaligned.is_contiguous() && unaligned._buffers.front().is_aligned(align_memory))) {
+      ptr nb(buffer::create_aligned(unaligned._len, align_memory));
       unaligned.rebuild(nb);
     }
     _buffers.insert(p, unaligned._buffers.front());

src/erasure-code/ErasureCode.cc

@@ -66,21 +66,14 @@ int ErasureCode::encode_prepare(const bufferlist &raw,
   unsigned int k = get_data_chunk_count();
   unsigned int m = get_chunk_count() - k;
   unsigned blocksize = get_chunk_size(raw.length());
-  unsigned pad_len = blocksize * k - raw.length();
   unsigned padded_chunks = k - raw.length() / blocksize;
   bufferlist prepared = raw;
 
-  if (!prepared.is_aligned(SIMD_ALIGN)) {
-    // splice padded chunks off to make the rebuild faster
-    if (padded_chunks)
-      prepared.splice((k - padded_chunks) * blocksize,
-                      padded_chunks * blocksize - pad_len);
-    prepared.rebuild_aligned(SIMD_ALIGN);
-  }
-
   for (unsigned int i = 0; i < k - padded_chunks; i++) {
     bufferlist &chunk = encoded[chunk_index(i)];
     chunk.substr_of(prepared, i * blocksize, blocksize);
+    chunk.rebuild_aligned_size_and_memory(blocksize, SIMD_ALIGN);
+    assert(chunk.is_contiguous());
   }
   if (padded_chunks) {
     unsigned remainder = raw.length() - (k - padded_chunks) * blocksize;

src/include/buffer.h

@@ -184,7 +184,7 @@ class buffer {
     bool is_page_aligned() const { return is_aligned(CEPH_PAGE_SIZE); }
     bool is_n_align_sized(unsigned align) const
     {
-      return (length() & (align-1)) == 0;
+      return (length() % align) == 0;
     }
     bool is_n_page_sized() const { return is_n_align_sized(CEPH_PAGE_SIZE); }
 
@@ -393,6 +393,8 @@ class buffer {
     void rebuild();
     void rebuild(ptr& nb);
     void rebuild_aligned(unsigned align);
+    void rebuild_aligned_size_and_memory(unsigned align_size,
+					 unsigned align_memory);
     void rebuild_page_aligned();
 
     // sort-of-like-assignment-op

src/test/bufferlist.cc

@@ -1212,6 +1212,52 @@ TEST(BufferList, is_n_page_sized) {
   }
 }
 
+TEST(BufferList, rebuild_aligned_size_and_memory) {
+  const unsigned SIMD_ALIGN = 32;
+  const unsigned BUFFER_SIZE = 67;
+
+  bufferlist bl;
+  // These two must be concatenated into one memory + size aligned
+  // bufferptr
+  {
+    bufferptr ptr(buffer::create_aligned(2, SIMD_ALIGN));
+    ptr.set_offset(1);
+    ptr.set_length(1);
+    bl.append(ptr);
+  }
+  {
+    bufferptr ptr(buffer::create_aligned(BUFFER_SIZE - 1, SIMD_ALIGN));
+    bl.append(ptr);
+  }
+  // This one must be left alone
+  {
+    bufferptr ptr(buffer::create_aligned(BUFFER_SIZE, SIMD_ALIGN));
+    bl.append(ptr);
+  }
+  // These two must be concatenated into one memory + size aligned
+  // bufferptr
+  {
+    bufferptr ptr(buffer::create_aligned(2, SIMD_ALIGN));
+    ptr.set_offset(1);
+    ptr.set_length(1);
+    bl.append(ptr);
+  }
+  {
+    bufferptr ptr(buffer::create_aligned(BUFFER_SIZE - 1, SIMD_ALIGN));
+    bl.append(ptr);
+  }
+  EXPECT_FALSE(bl.is_aligned(SIMD_ALIGN));
+  EXPECT_FALSE(bl.is_n_align_sized(BUFFER_SIZE));
+  EXPECT_EQ(BUFFER_SIZE * 3, bl.length());
+  EXPECT_FALSE(bl.buffers().front().is_aligned(SIMD_ALIGN));
+  EXPECT_FALSE(bl.buffers().front().is_n_align_sized(BUFFER_SIZE));
+  EXPECT_EQ(5U, bl.buffers().size());
+  bl.rebuild_aligned_size_and_memory(BUFFER_SIZE, SIMD_ALIGN);
+  EXPECT_TRUE(bl.is_aligned(SIMD_ALIGN));
+  EXPECT_TRUE(bl.is_n_align_sized(BUFFER_SIZE));
+  EXPECT_EQ(3U, bl.buffers().size());
+}
+
 TEST(BufferList, is_zero) {
   {
     bufferlist bl;

src/test/erasure-code/TestErasureCode.cc

@@ -111,6 +111,42 @@ TEST(ErasureCodeTest, encode_memory_align)
   ASSERT_NE(encoded[1][chunk_size / 2], 'X');
 }
 
+TEST(ErasureCodeTest, encode_misaligned_non_contiguous)
+{
+  int k = 3;
+  int m = 1;
+  unsigned chunk_size = ErasureCode::SIMD_ALIGN * 7;
+  ErasureCodeTest erasure_code(k, m, chunk_size);
+
+  set<int> want_to_encode;
+  for (unsigned int i = 0; i < erasure_code.get_chunk_count(); i++)
+    want_to_encode.insert(i);
+  string data(chunk_size, 'X');
+  // create a non contiguous bufferlist where the frist and the second
+  // bufferptr are not size aligned although they are memory aligned
+  bufferlist in;
+  {
+    bufferptr ptr(buffer::create_aligned(data.length() - 1, ErasureCode::SIMD_ALIGN));
+    in.append(ptr);
+  }
+  {
+    bufferptr ptr(buffer::create_aligned(data.length() + 1, ErasureCode::SIMD_ALIGN));
+    in.append(ptr);
+  }
+  map<int, bufferlist> encoded;
+
+  ASSERT_FALSE(in.is_contiguous());
+  ASSERT_TRUE(in.buffers().front().is_aligned(ErasureCode::SIMD_ALIGN));
+  ASSERT_FALSE(in.buffers().front().is_n_align_sized(chunk_size));
+  ASSERT_TRUE(in.buffers().back().is_aligned(ErasureCode::SIMD_ALIGN));
+  ASSERT_FALSE(in.buffers().back().is_n_align_sized(chunk_size));
+  ASSERT_EQ(0, erasure_code.encode(want_to_encode, in, &encoded));
+  for (unsigned int i = 0; i < erasure_code.get_chunk_count(); i++) {
+    ASSERT_TRUE(encoded[i].is_aligned(ErasureCode::SIMD_ALIGN));
+    ASSERT_TRUE(encoded[i].is_n_align_sized(chunk_size));
+  }
+}
+
 int main(int argc, char **argv)
 {
   vector<const char*> args;