Merge pull request ceph#1344 from ceph/wip-7539

Wip 7539

Reviewed-by: Sage Weil <[email protected]>

src/common/TrackedOp.cc

@@ -118,10 +118,11 @@ void OpTracker::register_inflight_op(xlist<TrackedOp*>::item *i)
 
 void OpTracker::unregister_inflight_op(TrackedOp *i)
 {
-  i->request->clear_data();
+  // caller checks;
+  assert(tracking_enabled);
 
-  if (!tracking_enabled)
-    return;
+  i->request->clear_data();
+  i->request->clear_payload();
 
   Mutex::Locker locker(ops_in_flight_lock);
   assert(i->xitem.get_list() == &ops_in_flight);

src/messages/MOSDECSubOpWrite.h

@@ -60,6 +60,11 @@ class MOSDECSubOpWrite : public Message {
 	<< " " << op;
     out << ")";
   }
+
+  void clear_buffers() {
+    op.t = ObjectStore::Transaction();
+    op.log_entries.clear();
+  }
 };
 
 #endif

src/messages/MOSDOp.h

@@ -337,6 +337,9 @@ struct ceph_osd_request_head {
     OSDOp::split_osd_op_vector_in_data(ops, data);
   }
 
+  void clear_buffers() {
+    ops.clear();
+  }
 
   const char *get_type_name() const { return "osd_op"; }
   void print(ostream& out) const {

src/msg/Message.h

@@ -403,10 +403,13 @@ class Message : public RefCountedObject {
     payload.clear();
     middle.clear();
   }
+
+  virtual void clear_buffers() {}
   void clear_data() {
     if (byte_throttler)
       byte_throttler->put(data.length());
     data.clear();
+    clear_buffers(); // let subclass drop buffers as well
   }
 
   bool empty_payload() { return payload.length() == 0; }

src/osd/PG.cc

@@ -2631,6 +2631,12 @@ void PG::add_log_entry(pg_log_entry_t& e, bufferlist& log_bl)
   if (e.user_version > info.last_user_version)
     info.last_user_version = e.user_version;
 
+  /**
+   * Make sure we don't keep around more than we need to in the
+   * in-memory log
+   */
+  e.mod_desc.trim_bl();
+
   // log mutation
   pg_log.add(e);
   dout(10) << "add_log_entry " << e << dendl;

src/osd/ReplicatedPG.cc

@@ -8745,6 +8745,8 @@ void ReplicatedPG::mark_all_unfound_lost(int what)
   dirty_info = true;
   write_if_dirty(*t);
 
+  t->register_on_complete(new ObjectStore::C_DeleteTransaction(t));
+
   osd->store->queue_transaction(osr.get(), t, c, NULL, new C_OSD_OndiskWriteUnlockList(&c->obcs));
 
   // Send out the PG log to all replicas

src/osd/osd_types.h

@@ -1856,6 +1856,16 @@ class ObjectModDesc {
   bool empty() const {
     return can_local_rollback && (bl.length() == 0);
   }
+
+  /**
+   * Create fresh copy of bl bytes to avoid keeping large buffers around
+   * in the case that bl contains ptrs which point into a much larger
+   * message buffer
+   */
+  void trim_bl() {
+    if (bl.length() > 0)
+      bl.rebuild();
+  }
   void encode(bufferlist &bl) const;
   void decode(bufferlist::iterator &bl);
   void dump(Formatter *f) const;