Added OSD_OP_* values for WRLOCK and WRUNLOCK, and handling of lock and unlock cases...

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@759 29311d96-e01e-0410-9327-a35deaab8ce9

diff --git a/ceph/messages/MOSDOp.h b/ceph/messages/MOSDOp.h
index b78a6bb2f4741e890c4738b8750d4491ace990e4..5b0b74c70aafc2231fff9951f07abdd4a07156e4 100644 (file)
--- a/ceph/messages/MOSDOp.h
+++ b/ceph/messages/MOSDOp.h
@@ -36,6 +36,8 @@
 #define OSD_OP_DELETE     11
 #define OSD_OP_TRUNCATE   12
 #define OSD_OP_ZERO       13
+#define OSD_OP_WRLOCK     14
+#define OSD_OP_WRUNLOCK   15
 
 #define OSD_OP_IS_REP(x)  ((x) >= 30)
 
@@ -43,6 +45,8 @@
 #define OSD_OP_REP_WRITE    (100+OSD_OP_WRITE)     // replicated (partial object) write
 #define OSD_OP_REP_TRUNCATE (100+OSD_OP_TRUNCATE)  // replicated truncate
 #define OSD_OP_REP_DELETE   (100+OSD_OP_DELETE)
+#define OSD_OP_REP_WRLOCK   (100+OSD_OP_WRLOCK)
+#define OSD_OP_REP_WRUNLOCK (100+OSD_OP_WRUNLOCK)
 
 #define OSD_OP_REP_PULL     30   // whole object read
 #define OSD_OP_REP_PUSH     31   // whole object write

diff --git a/ceph/osd/OSD.cc b/ceph/osd/OSD.cc
index 1828a9f86a483ace7c8530b8e1de30a5773b7aa1..f482ccb61bc15db8fc26688c71373d3d11700ed4 100644 (file)
--- a/ceph/osd/OSD.cc
+++ b/ceph/osd/OSD.cc
@@ -1586,29 +1586,18 @@ void OSD::op_rep_modify(MOSDOp *op)
   
   dout(12) << "rep_modify in " << *pg << " o " << hex << oid << dec << " v " << op->get_version() << " (i have " << ov << ")" << endl;
   
-  int r = 0;
   Context *oncommit = 0;
   
   op->hack_blah = true;  // hack: make sure any 'commit' goes out _after_ our ack
-  
+
+  oncommit = new C_OSD_RepModifyCommit(this, op);
+  op_apply(op, op->get_version(), oncommit); 
+ 
   if (op->get_op() == OSD_OP_REP_WRITE) {
-       // write
-       assert(op->get_data().length() == op->get_length());
-       oncommit = new C_OSD_RepModifyCommit(this, op);
-       r = apply_write(op, op->get_version(), oncommit);
-       store->collection_add(pg->get_pgid(), oid);
-       
-       logger->inc("r_wr");
+       logger->inc("r_wr");
        logger->inc("r_wrb", op->get_length());
-  } else if (op->get_op() == OSD_OP_REP_DELETE) {
-       // delete
-       store->collection_remove(pg->get_pgid(), op->get_oid());
-       r = store->remove(oid);
-  } else if (op->get_op() == OSD_OP_REP_TRUNCATE) {
-       // truncate
-       r = store->truncate(oid, op->get_offset());
-  } else assert(0);
-  
+  }
+
   // update pg version too
   pg->info.last_update = op->get_version();
   if (pg->info.last_complete == ov)
@@ -1629,8 +1618,6 @@ void OSD::op_rep_modify(MOSDOp *op)
 }
 
 
-
-
 // =========================================================
 // OPS
 
@@ -2143,49 +2130,60 @@ void OSD::op_modify(MOSDOp *op)
   //messenger->send_message(reply, op->get_asker());
   
   // do it
-  int r;
-  if (op->get_op() == OSD_OP_WRITE) {
+  Context *oncommit = new C_OSD_WriteCommit(this, repop);
+  op_apply(op, nv, oncommit);
+
+  get_repop(repop);
+  assert(repop->waitfor_ack.count(0));
+  repop->waitfor_ack.erase(0);
+  put_repop(repop);
+
+  if (op->get_op() == OSD_OP_WRITE)
+       {
+         logger->inc("c_wr");
+         logger->inc("c_wrb", op->get_length());
+       }
+  
+  //unlock_object(oid);
+}
+
+
+
+void OSD::op_apply(MOSDOp *op, version_t version, Context* oncommit)
+{
+   object_t oid = op->get_oid();
+   pg_t pgid = op->get_pg();
+   int r;
+
+  if ((op->get_op() == OSD_OP_WRLOCK)||
+         (op->get_op() == OSD_OP_REP_WRLOCK)){
+       //lock object
+       r = store->setattr(oid, "wrlock", &op->get_source(), sizeof(msg_addr_t), oncommit);
+  }
+  else if ((op->get_op() == OSD_OP_WRUNLOCK) ||
+                  (op->get_op() == OSD_OP_REP_WRUNLOCK)) {     
+       //unlock object
+       r = store->rmattr(oid, "wrlock", oncommit);
+  }
+  else if ((op->get_op() == OSD_OP_WRITE) ||
+                  (op->get_op() == OSD_OP_REP_WRITE)){
        // write
        assert(op->get_data().length() == op->get_length());
-       Context *oncommit = new C_OSD_WriteCommit(this, repop);
-       r = apply_write(op, nv, oncommit);
+       r = apply_write(op, version, oncommit);
        
        // put new object in proper collection
        store->collection_add(pgid, oid);        // FIXME : be careful w/ locking
-       
-       get_repop(repop);
-       assert(repop->waitfor_ack.count(0));
-       repop->waitfor_ack.erase(0);
-       put_repop(repop);
-       
-       logger->inc("c_wr");
-       logger->inc("c_wrb", op->get_length());
   } 
-  else if (op->get_op() == OSD_OP_TRUNCATE) {
+  else if ((op->get_op() == OSD_OP_TRUNCATE)||
+                  (op->get_op() == OSD_OP_REP_TRUNCATE)) {
        // truncate
-       r = store->truncate(oid, op->get_offset());
-       get_repop(repop);
-       assert(repop->waitfor_ack.count(0));
-       assert(repop->waitfor_commit.count(0));
-       repop->waitfor_ack.erase(0);
-       repop->waitfor_commit.erase(0);
-       put_repop(repop);
+       r = store->truncate(oid, op->get_offset(), oncommit);
   }
-  else if (op->get_op() == OSD_OP_DELETE) {
+  else if ((op->get_op() == OSD_OP_DELETE)||
+                  (op->get_op() == OSD_OP_REP_DELETE)) {
        // delete
        store->collection_remove(pgid, oid);  // be careful with locking
-       r = store->remove(oid);
-       get_repop(repop);
-       assert(repop->waitfor_ack.count(0));
-       assert(repop->waitfor_commit.count(0));
-       repop->waitfor_ack.erase(0);
-       repop->waitfor_commit.erase(0);
-       put_repop(repop);
+       r = store->remove(oid, oncommit);
   }
   else assert(0);
-  
-  //unlock_object(oid);
 }
-
-
-

diff --git a/ceph/osd/OSD.h b/ceph/osd/OSD.h
index 765ee929a9d70ce2df757380b1bfca9b995c73cd..d4eebf190986ace626fa8d4e704a83b4c8928121 100644 (file)
--- a/ceph/osd/OSD.h
+++ b/ceph/osd/OSD.h
@@ -129,7 +129,7 @@ public:
 
   int apply_write(MOSDOp *op, version_t v,
                                  Context *oncommit = 0); 
-  
+  void op_apply(MOSDOp* op, version_t version, Context* oncommit = 0);
   
   
   friend class PG;

diff --git a/ceph/osd/ObjectStore.h b/ceph/osd/ObjectStore.h
index 2a16f39d3ea915e2b9f3ceb836bb3058df1004b1..dd23792a596ede4a487836bbee25736b8f8fa090 100644 (file)
--- a/ceph/osd/ObjectStore.h
+++ b/ceph/osd/ObjectStore.h
@@ -86,8 +86,10 @@ class ObjectStore {
                                          Context *onsafe=0) {return 0;} //= 0;
   virtual int getattr(object_t oid, const char *name,
                                          void *value, size_t size) {return 0;} //= 0;
+
   virtual int rmattr(object_t oid, const char *name,
                                         Context *onsafe=0) {return 0;}
+
   virtual int listattr(object_t oid, char *attrs, size_t size) {return 0;} //= 0;
   
   // collections