From: John Spray <john.spray@redhat.com>
Date: Fri, 8 Aug 2014 00:49:26 +0000 (+0100)
Subject: osdc: Add lock to Filer::Probe
X-Git-Tag: v0.86~213^2~30
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=13e6c5f2de38f4c80854c3186d6f658eeebcf5e1;p=ceph.git

osdc: Add lock to Filer::Probe

This is necessary now that Objecter can call back
from multiple OSD op completions in parallel: otherwise
we get multiple threads trying to update
the same Probe object.

Signed-off-by: John Spray <john.spray@redhat.com>
---

diff --git a/src/osdc/Filer.cc b/src/osdc/Filer.cc
index c15f438b64b0..09749b02b1f5 100644
--- a/src/osdc/Filer.cc
+++ b/src/osdc/Filer.cc
@@ -45,11 +45,21 @@ public:
       assert(size == 0);
     }
 
-    // TODO: handle this error.
-    if (r != 0)
-      probe->err = r;
+    bool probe_complete;
+    {
+      Mutex::Locker l(probe->lock);
 
-    filer->_probed(probe, oid, size, mtime);
+      // TODO: handle this error.
+      if (r != 0) {
+        probe->err = r;
+      }
+
+      probe_complete = filer->_probed(probe, oid, size, mtime);
+    }
+    if (probe_complete) {
+      probe->onfinish->complete(probe->err);
+      delete probe;
+    }
   }  
 };
 
@@ -87,13 +97,20 @@ int Filer::probe(inodeno_t ino,
     probe->probing_off -= probe->probing_len;
   }
   
-  _probe(probe);
+  // Take lock before starting any I/Os, to protect us from concurrent calls
+  // back into C_Probe from OSD op completions from different OSDs
+  {
+    Mutex::Locker l(probe->lock);
+    _probe(probe);
+  }
   return 0;
 }
 
 
 void Filer::_probe(Probe *probe)
 {
+  assert(probe->lock.is_locked_by_me());
+
   ldout(cct, 10) << "_probe " << hex << probe->ino << dec 
 	   << " " << probe->probing_off << "~" << probe->probing_len 
 	   << dendl;
@@ -115,8 +132,13 @@ void Filer::_probe(Probe *probe)
   }
 }
 
-void Filer::_probed(Probe *probe, const object_t& oid, uint64_t size, utime_t mtime)
+/**
+ * @return true if probe is complete and Probe object may be freed.
+ */
+bool Filer::_probed(Probe *probe, const object_t& oid, uint64_t size, utime_t mtime)
 {
+  assert(probe->lock.is_locked_by_me());
+
   ldout(cct, 10) << "_probed " << probe->ino << " object " << oid
 	   << " has size " << size << " mtime " << mtime << dendl;
 
@@ -128,12 +150,10 @@ void Filer::_probed(Probe *probe, const object_t& oid, uint64_t size, utime_t mt
   probe->ops.erase(oid);
 
   if (!probe->ops.empty()) 
-    return;  // waiting for more!
+    return false;  // waiting for more!
 
   if (probe->err) { // we hit an error, propagate back up
-    probe->onfinish->complete(probe->err);
-    delete probe;
-    return;
+    return true;
   }
 
   // analyze!
@@ -206,7 +226,7 @@ void Filer::_probed(Probe *probe, const object_t& oid, uint64_t size, utime_t mt
       probe->probing_off -= period;
     }
     _probe(probe);
-    return;
+    return false;
   }
 
   if (probe->pmtime) {
@@ -214,9 +234,8 @@ void Filer::_probed(Probe *probe, const object_t& oid, uint64_t size, utime_t mt
     *probe->pmtime = probe->max_mtime;
   }
 
-  // done!  finish and clean up.
-  probe->onfinish->complete(probe->err);
-  delete probe;
+  // done!
+  return true;
 }
 
 
diff --git a/src/osdc/Filer.h b/src/osdc/Filer.h
index 607dc7b30d51..1164d95873a0 100644
--- a/src/osdc/Filer.h
+++ b/src/osdc/Filer.h
@@ -46,6 +46,7 @@ class Filer {
   
   // probes
   struct Probe {
+    Mutex lock;
     inodeno_t ino;
     ceph_file_layout layout;
     snapid_t snapid;
@@ -72,7 +73,7 @@ class Filer {
 
     Probe(inodeno_t i, ceph_file_layout &l, snapid_t sn,
 	  uint64_t f, uint64_t *e, utime_t *m, int fl, bool fw, Context *c) : 
-      ino(i), layout(l), snapid(sn),
+      lock("Filer::Probe"), ino(i), layout(l), snapid(sn),
       psize(e), pmtime(m), flags(fl), fwd(fw), onfinish(c),
       probing_off(f), probing_len(0),
       err(0), found_size(false) {}
@@ -81,7 +82,7 @@ class Filer {
   class C_Probe;
 
   void _probe(Probe *p);
-  void _probed(Probe *p, const object_t& oid, uint64_t size, utime_t mtime);
+  bool _probed(Probe *p, const object_t& oid, uint64_t size, utime_t mtime);
 
  public:
   Filer(const Filer& other);