Merge pull request #7455 from yuyuyu101/bufferlist-move

buffer: use move construct to append/push_back/push_front

Reviewed-by: Sage Weil <sage@redhat.com>

src/common/SloppyCRCMap.h

@@ -34,9 +34,7 @@ class SloppyCRCMap {
     //zero_crc = ceph_crc32c(0xffffffff, NULL, block_size);
     if (b) {
       bufferlist bl;
-      bufferptr bp(block_size);
-      bp.zero();
-      bl.append(bp);
+      bl.append_zero(block_size);
       zero_crc = bl.crc32c(crc_iv);
     } else {
       zero_crc = crc_iv;

src/common/buffer.cc

@@ -708,6 +708,11 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
       bdout << "ptr " << this << " get " << _raw << bendl;
     }
   }
+  buffer::ptr::ptr(ptr&& p) : _raw(p._raw), _off(p._off), _len(p._len)
+  {
+    p._raw = nullptr;
+    p._off = p._len = 0;
+  }
   buffer::ptr::ptr(const ptr& p, unsigned o, unsigned l)
     : _raw(p._raw), _off(p._off + o), _len(l)
   {
@@ -733,6 +738,21 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
     }
     return *this;
   }
+  buffer::ptr& buffer::ptr::operator= (ptr&& p)
+  {
+    release();
+    buffer::raw *raw = p._raw;
+    if (raw) {
+      _raw = raw;
+      _off = p._off;
+      _len = p._len;
+      p._raw = nullptr;
+      p._off = p._len = 0;
+    } else {
+      _off = _len = 0;
+    }
+    return *this;
+  }
 
   buffer::raw *buffer::ptr::clone()
   {
@@ -1583,6 +1603,12 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
       push_back(bp);
   }
 
+  void buffer::list::append(ptr&& bp)
+  {
+    if (bp.length())
+      push_back(std::move(bp));
+  }
+
   void buffer::list::append(const ptr& bp, unsigned off, unsigned len)
   {
     assert(len+off <= bp.length());
@@ -1597,8 +1623,7 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
       }
     }
     // add new item to list
-    ptr tempbp(bp, off, len);
-    push_back(tempbp);
+    push_back(ptr(bp, off, len));
   }
 
   void buffer::list::append(const list& bl)
@@ -1625,7 +1650,7 @@ static simple_spinlock_t buffer_debug_lock = SIMPLE_SPINLOCK_INITIALIZER;
   {
     ptr bp(len);
     bp.zero();
-    append(bp);
+    append(std::move(bp));
   }
 
 
@@ -1834,7 +1859,7 @@ void buffer::list::encode_base64(buffer::list& o)
   bufferptr bp(length() * 4 / 3 + 3);
   int l = ceph_armor(bp.c_str(), bp.c_str() + bp.length(), c_str(), c_str() + length());
   bp.set_length(l);
-  o.push_back(bp);
+  o.push_back(std::move(bp));
 }
 
 void buffer::list::decode_base64(buffer::list& e)
@@ -1849,7 +1874,7 @@ void buffer::list::decode_base64(buffer::list& e)
   }
   assert(l <= (int)bp.length());
   bp.set_length(l);
-  push_back(bp);
+  push_back(std::move(bp));
 }
 
 
@@ -1903,7 +1928,7 @@ ssize_t buffer::list::read_fd(int fd, size_t len)
   ssize_t ret = safe_read(fd, (void*)bp.c_str(), len);
   if (ret >= 0) {
     bp.set_length(ret);
-    append(bp);
+    append(std::move(bp));
   }
   return ret;
 }
@@ -1912,8 +1937,7 @@ int buffer::list::read_fd_zero_copy(int fd, size_t len)
 {
 #ifdef CEPH_HAVE_SPLICE
   try {
-    bufferptr bp = buffer::create_zero_copy(len, fd, NULL);
-    append(bp);
+    append(buffer::create_zero_copy(len, fd, NULL));
   } catch (buffer::error_code &e) {
     return e.code;
   } catch (buffer::malformed_input &e) {

src/erasure-code/ErasureCode.cc

@@ -93,12 +93,12 @@ int ErasureCode::encode_prepare(const bufferlist &raw,
 
     raw.copy((k - padded_chunks) * blocksize, remainder, buf.c_str());
     buf.zero(remainder, blocksize - remainder);
-    encoded[chunk_index(k-padded_chunks)].push_back(buf);
+    encoded[chunk_index(k-padded_chunks)].push_back(std::move(buf));
 
     for (unsigned int i = k - padded_chunks + 1; i < k; i++) {
       bufferptr buf(buffer::create_aligned(blocksize, SIMD_ALIGN));
       buf.zero();
-      encoded[chunk_index(i)].push_back(buf);
+      encoded[chunk_index(i)].push_back(std::move(buf));
     }
   }
   for (unsigned int i = k; i < k + m; i++) {

src/include/buffer.h

@@ -174,8 +174,10 @@ namespace buffer CEPH_BUFFER_API {
     ptr(unsigned l);
     ptr(const char *d, unsigned l);
     ptr(const ptr& p);
+    ptr(ptr&& p);
     ptr(const ptr& p, unsigned o, unsigned l);
     ptr& operator= (const ptr& p);
+    ptr& operator= (ptr&& p);
     ~ptr() {
       release();
     }
@@ -376,6 +378,16 @@ namespace buffer CEPH_BUFFER_API {
       return *this;
     }
 
+    list& operator= (list&& other) {
+      _buffers = std::move(other._buffers);
+      _len = other._len;
+      _memcopy_count = other._memcopy_count;
+      last_p = begin();
+      append_buffer.swap(other.append_buffer);
+      other.clear();
+      return *this;
+    }
+
     unsigned get_memcopy_count() const {return _memcopy_count; }
     const std::list<ptr>& buffers() const { return _buffers; }
     void swap(list& other);
@@ -417,19 +429,29 @@ namespace buffer CEPH_BUFFER_API {
       _buffers.push_front(bp);
       _len += bp.length();
     }
+    void push_front(ptr&& bp) {
+      if (bp.length() == 0)
+	return;
+      _len += bp.length();
+      _buffers.push_front(std::move(bp));
+    }
     void push_front(raw *r) {
-      ptr bp(r);
-      push_front(bp);
+      push_front(ptr(r));
     }
     void push_back(const ptr& bp) {
       if (bp.length() == 0)
 	return;
       _buffers.push_back(bp);
       _len += bp.length();
     }
+    void push_back(ptr&& bp) {
+      if (bp.length() == 0)
+	return;
+      _len += bp.length();
+      _buffers.push_back(std::move(bp));
+    }
     void push_back(raw *r) {
-      ptr bp(r);
-      push_back(bp);
+      push_back(ptr(r));
     }
 
     void zero();
@@ -500,6 +522,7 @@ namespace buffer CEPH_BUFFER_API {
       append(s.data(), s.length());
     }
     void append(const ptr& bp);
+    void append(ptr&& bp);
     void append(const ptr& bp, unsigned off, unsigned len);
     void append(const list& bl);
     void append(std::istream& in);

src/librbd/librbd.cc

@@ -968,7 +968,7 @@ namespace librbd {
     ImageCtx *ictx = (ImageCtx *)ctx;
     tracepoint(librbd, read_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, ofs, len);
     bufferptr ptr(len);
-    bl.push_back(ptr);
+    bl.push_back(std::move(ptr));
     int r = ictx->aio_work_queue->read(ofs, len, bl.c_str(), 0);
     tracepoint(librbd, read_exit, r);
     return r;
@@ -980,7 +980,7 @@ namespace librbd {
     tracepoint(librbd, read2_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(),
 		ictx->read_only, ofs, len, op_flags);
     bufferptr ptr(len);
-    bl.push_back(ptr);
+    bl.push_back(std::move(ptr));
     int r = ictx->aio_work_queue->read(ofs, len, bl.c_str(), op_flags);
     tracepoint(librbd, read_exit, r);
     return r;

src/msg/async/AsyncConnection.cc

@@ -161,19 +161,16 @@ static void alloc_aligned_buffer(bufferlist& data, unsigned len, unsigned off)
     // head
     unsigned head = 0;
     head = MIN(CEPH_PAGE_SIZE - (off & ~CEPH_PAGE_MASK), left);
-    bufferptr bp = buffer::create(head);
-    data.push_back(bp);
+    data.push_back(buffer::create(head));
     left -= head;
   }
   unsigned middle = left & CEPH_PAGE_MASK;
   if (middle > 0) {
-    bufferptr bp = buffer::create_page_aligned(middle);
-    data.push_back(bp);
+    data.push_back(buffer::create_page_aligned(middle));
     left -= middle;
   }
   if (left) {
-    bufferptr bp = buffer::create(left);
-    data.push_back(bp);
+    data.push_back(buffer::create(left));
   }
 }
 
@@ -727,10 +724,9 @@ void AsyncConnection::process()
           // read front
           unsigned front_len = current_header.front_len;
           if (front_len) {
-            if (!front.length()) {
-              bufferptr ptr = buffer::create(front_len);
-              front.push_back(ptr);
-            }
+            if (!front.length())
+              front.push_back(buffer::create(front_len));
+
             r = read_until(front_len, front.c_str());
             if (r < 0) {
               ldout(async_msgr->cct, 1) << __func__ << " read message front failed" << dendl;
@@ -750,10 +746,9 @@ void AsyncConnection::process()
           // read middle
           unsigned middle_len = current_header.middle_len;
           if (middle_len) {
-            if (!middle.length()) {
-              bufferptr ptr = buffer::create(middle_len);
-              middle.push_back(ptr);
-            }
+            if (!middle.length())
+              middle.push_back(buffer::create(middle_len));
+
             r = read_until(middle_len, middle.c_str());
             if (r < 0) {
               ldout(async_msgr->cct, 1) << __func__ << " read message middle failed" << dendl;

src/msg/simple/Pipe.cc

@@ -333,7 +333,7 @@ int Pipe::accept()
   }
   {
     bufferptr tp(sizeof(peer_addr));
-    addrbl.push_back(tp);
+    addrbl.push_back(std::move(tp));
   }
   if (tcp_read(addrbl.c_str(), addrbl.length()) < 0) {
     ldout(msgr->cct,10) << "accept couldn't read peer_addr" << dendl;
@@ -371,7 +371,7 @@ int Pipe::accept()
         ldout(msgr->cct,10) << "accept couldn't read connect authorizer" << dendl;
         goto fail_unlocked;
       }
-      authorizer.push_back(bp);
+      authorizer.push_back(std::move(bp));
       authorizer_reply.clear();
     }
 
@@ -948,7 +948,7 @@ int Pipe::connect()
     int wirelen = sizeof(__u32) * 2 + sizeof(ceph_sockaddr_storage);
     bufferptr p(wirelen * 2);
 #endif
-    addrbl.push_back(p);
+    addrbl.push_back(std::move(p));
   }
   if (tcp_read(addrbl.c_str(), addrbl.length()) < 0) {
     ldout(msgr->cct,2) << "connect couldn't read peer addrs, " << cpp_strerror(errno) << dendl;
@@ -1875,19 +1875,16 @@ static void alloc_aligned_buffer(bufferlist& data, unsigned len, unsigned off)
     // head
     unsigned head = 0;
     head = MIN(CEPH_PAGE_SIZE - (off & ~CEPH_PAGE_MASK), left);
-    bufferptr bp = buffer::create(head);
-    data.push_back(bp);
+    data.push_back(buffer::create(head));
     left -= head;
   }
   unsigned middle = left & CEPH_PAGE_MASK;
   if (middle > 0) {
-    bufferptr bp = buffer::create_page_aligned(middle);
-    data.push_back(bp);
+    data.push_back(buffer::create_page_aligned(middle));
     left -= middle;
   }
   if (left) {
-    bufferptr bp = buffer::create(left);
-    data.push_back(bp);
+    data.push_back(buffer::create(left));
   }
 }
 
@@ -1975,7 +1972,7 @@ int Pipe::read_message(Message **pm, AuthSessionHandler* auth_handler)
     bufferptr bp = buffer::create(front_len);
     if (tcp_read(bp.c_str(), front_len) < 0)
       goto out_dethrottle;
-    front.push_back(bp);
+    front.push_back(std::move(bp));
     ldout(msgr->cct,20) << "reader got front " << front.length() << dendl;
   }
 
@@ -1985,7 +1982,7 @@ int Pipe::read_message(Message **pm, AuthSessionHandler* auth_handler)
     bufferptr bp = buffer::create(middle_len);
     if (tcp_read(bp.c_str(), middle_len) < 0)
       goto out_dethrottle;
-    middle.push_back(bp);
+    middle.push_back(std::move(bp));
     ldout(msgr->cct,20) << "reader got middle " << middle.length() << dendl;
   }
 

src/os/FuseStore.cc

@@ -809,9 +809,7 @@ static int os_write(const char *path, const char *buf, size_t size,
     } else {
       final.claim_append(*pbl);
       size_t zlen = offset - final.length();
-      bufferptr z(zlen);
-      z.zero();
-      final.append(z);
+      final.append_zero(zlen);
     }
   }
   final.append(buf, size);

src/os/bluestore/BlueFS.cc

@@ -304,9 +304,7 @@ int BlueFS::_write_super()
   uint32_t crc = bl.crc32c(-1);
   ::encode(crc, bl);
   assert(bl.length() <= get_super_length());
-  bufferptr z(get_super_length() - bl.length());
-  z.zero();
-  bl.append(z);
+  bl.append_zero(get_super_length() - bl.length());
   bl.rebuild();
 
   IOContext ioc(NULL);
@@ -889,12 +887,8 @@ void BlueFS::_pad_bl(bufferlist& bl)
 {
   uint64_t partial = bl.length() % super.block_size;
   if (partial) {
-    bufferptr z(super.block_size - partial);
-    dout(10) << __func__ << " padding with " << z.length() << " zeros" << dendl;
-    z.zero();
-    bufferlist zbl;
-    zbl.append(z);
-    bl.append(z);
+    dout(10) << __func__ << " padding with " << super.block_size - partial << " zeros" << dendl;
+    bl.append_zero(super.block_size - partial);
   }
 }
 
@@ -1038,9 +1032,7 @@ int BlueFS::_flush_range(FileWriter *h, uint64_t offset, uint64_t length)
       dout(20) << __func__ << " caching tail of " << tail
 	       << " and padding block with zeros" << dendl;
       h->tail_block.substr_of(bl, bl.length() - tail, tail);
-      bufferptr z(super.block_size - tail);
-      z.zero();
-      t.append(z);
+      t.append_zero(super.block_size - tail);
     }
     bdev[p->bdev]->aio_write(p->offset + x_off, t, h->iocv[p->bdev], true);
     bloff += x_len;