rados: define rdcall, wrcall on arbitrary class, method

diff --git a/src/include/rados.h b/src/include/rados.h
index ec09487892c1bf8187cba48ae7010a718c97c703..5141c2d9b83692170ce33d653f9dcf51b78aee68 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -168,6 +168,7 @@ struct ceph_eversion {
 #define CEPH_OSD_OP_TYPE_LOCK  0x0100
 #define CEPH_OSD_OP_TYPE_DATA  0x0200
 #define CEPH_OSD_OP_TYPE_ATTR  0x0300
+#define CEPH_OSD_OP_TYPE_EXEC  0x0400
 
 enum {
        /** data **/
@@ -219,8 +220,9 @@ enum {
        CEPH_OSD_OP_DNLOCK     = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_LOCK | 6,
 
        /** exec **/
-       CEPH_OSD_OP_LOAD_CLASS = CEPH_OSD_OP_MODE_EXEC | 1,
-       CEPH_OSD_OP_EXEC       = CEPH_OSD_OP_MODE_EXEC | 2,
+       CEPH_OSD_OP_RDCALL     = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_EXEC | 1,
+       CEPH_OSD_OP_LOAD_CLASS = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_EXEC | 2,
+       CEPH_OSD_OP_WRCALL     = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_EXEC | 1,
 };
 
 static inline int ceph_osd_op_type_lock(int op)
@@ -235,6 +237,10 @@ static inline int ceph_osd_op_type_attr(int op)
 {
        return (op & CEPH_OSD_OP_TYPE) == CEPH_OSD_OP_TYPE_ATTR;
 }
+static inline int ceph_osd_op_type_exec(int op)
+{
+       return (op & CEPH_OSD_OP_TYPE) == CEPH_OSD_OP_TYPE_EXEC;
+}
 
 static inline int ceph_osd_op_mode_subop(int op)
 {
@@ -289,7 +295,8 @@ static inline const char *ceph_osd_op_name(int op)
        case CEPH_OSD_OP_UPLOCK: return "uplock";
        case CEPH_OSD_OP_DNLOCK: return "dnlock";
 
-       case CEPH_OSD_OP_EXEC: return "exec";
+       case CEPH_OSD_OP_RDCALL: return "rdcall";
+       case CEPH_OSD_OP_WRCALL: return "wrcall";
 
        default: return "???";
        }
@@ -328,6 +335,11 @@ struct ceph_osd_op {
                        __le64 truncate_size;
                        __le32 truncate_seq;
                };
+               struct {
+                       __u8 class_len;
+                       __u8 method_len;
+                       __le32 indata_len;
+               } __attribute__ ((packed));
        };
 } __attribute__ ((packed));
 

diff --git a/src/librados.cc b/src/librados.cc
index 81b8746624410e2c708c05068b6c342421a5a5f6..e661993b0a8fcf7eadf5d6732ad1dfef71d7bd60 100644 (file)
--- a/src/librados.cc
+++ b/src/librados.cc
@@ -281,7 +281,7 @@ int RadosClient::write(object_t& oid, const char *buf, off_t off, size_t len)
 int RadosClient::exec(object_t& oid, const char *code, off_t data_off, size_t data_len, char *buf, size_t out_len)
 {
   SnapContext snapc;
-  bufferlist bl, obl;
+  bufferlist bl;
   utime_t ut = g_clock.now();
 
   Mutex lock("RadosClient::exec");
@@ -296,19 +296,20 @@ int RadosClient::exec(object_t& oid, const char *code, off_t data_off, size_t da
 
   lock.Lock();
 
-  objecter->exec(oid, layout,
-             data_off, data_len, CEPH_NOSNAP, bl, 0,
-             &obl, out_len,
-              oncommit);
+  ObjectRead rd;
+  bufferlist inbl, outbl;
+  inbl.append("input data", 10);
+  rd.rdcall("test", "foo", inbl);
+  objecter->read(oid, layout, rd, CEPH_NOSNAP, &outbl, 0, oncommit);
 
-  dout(0) << "after write call" << dendl;
+  dout(0) << "after rdcall got " << outbl.length() << " bytes" << dendl;
 
   exec_wait.Wait(lock);
 
   lock.Unlock();
 
   if (out_len)
-    memcpy(buf, obl.c_str(), out_len);
+    memcpy(buf, outbl.c_str(), out_len);
 
   return out_len;
 }

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 3467f81ce1726fb2ec35bd50edc0656317221ba0..a3aceb4a0af19877d8f684fc9f04c4b96c3a645d 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -739,11 +739,18 @@ void ReplicatedPG::op_read(MOSDOp *op)
       osd->logger->inc(l_osd_c_rdb, p->length);
       break;
 
-   case CEPH_OSD_OP_EXEC:
+   case CEPH_OSD_OP_RDCALL:
     {
-        dout(0) << "CEPH_OSD_OP_EXEC" << dendl;
+        dout(0) << "CEPH_OSD_OP_RDCALL" << dendl;
 
-       if (!osd->get_class("test", info.pgid, op))
+       string cname, method;
+       bp.copy(p->class_len, cname);
+       bp.copy(p->method_len, method);
+
+       bufferlist indata;
+       ::decode_nohead(p->indata_len, indata, bp);
+
+       if (!osd->get_class(cname, info.pgid, op))
          return;
 
 #if 0

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 0162110c41ec12238e2866158e18e0b1735cb579..6c4d511ff14a75de06c436f2ec316972ddde7032 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -59,6 +59,17 @@ struct ObjectOperation {
     ops[s].name_len = namelen;
     ops[s].value_len = valuelen;
   }
+  void add_call(int op, const char *cname, const char *method, bufferlist &indata) {
+    int s = ops.size();
+    ops.resize(s+1);
+    memset(&ops[s], 0, sizeof(ops[s]));
+    ops[s].op = op;
+    ops[s].class_len = strlen(cname);
+    ops[s].method_len = strlen(method);
+    data.append(cname, ops[s].class_len);
+    data.append(method, ops[s].method_len);
+    data.append(indata);
+  }
 };
 
 struct ObjectRead : public ObjectOperation {
@@ -73,11 +84,20 @@ struct ObjectRead : public ObjectOperation {
   void getxattrs() {
     add_xattr(CEPH_OSD_OP_GETXATTRS, 0, 0);
   }
+
+  void rdcall(const char *cname, const char *method, bufferlist &indata) {
+    add_call(CEPH_OSD_OP_RDCALL, cname, method, indata);
+  }
 };
 
 struct ObjectMutation : public ObjectOperation {
   utime_t mtime;
   
+  // exec
+  void wrcall(const char *cname, const char *method, bufferlist &indata) {
+    add_call(CEPH_OSD_OP_WRCALL, cname, method, indata);
+  }
+  
   // object data
   void write(__u64 off, __u64 len, bufferlist& bl) {
     add_data(CEPH_OSD_OP_WRITE, off, len);
@@ -318,22 +338,6 @@ class Objecter {
     return read_submit(rd);
   }
 
-  tid_t exec(object_t oid, ceph_object_layout ol,
-             __u64 data_off, size_t data_len,
-             snapid_t snap, bufferlist &bl, int flags,
-              bufferlist *pbl, size_t out_len,
-              Context *onfinish) {
-    vector<ceph_osd_op> ops(1);
-    memset(&ops[0], 0, sizeof(ops[0]));
-    ops[0].op = CEPH_OSD_OP_EXEC;
-    ops[0].offset = data_off;
-    ops[0].length = data_len;
-    ReadOp *rd = new ReadOp(oid, ol, ops, snap, flags, onfinish);
-    rd->bl = bl;
-    rd->pbl = pbl;
-    return read_submit(rd);
-  }
-
   tid_t modify(object_t oid, ceph_object_layout ol, vector<ceph_osd_op>& ops,
               const SnapContext& snapc, bufferlist &bl, utime_t mtime, int flags,
               Context *onack, Context *oncommit) {