librados: implement aio_stat

Implement aio stat and also export this functionality to the C API.

Signed-off-by: Filippos Giannakos <philipgian@grnet.gr>
Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

src/include/rados/librados.h

@@ -23,7 +23,7 @@ extern "C" {
 #endif
 
 #define LIBRADOS_VER_MAJOR 0
-#define LIBRADOS_VER_MINOR 48
+#define LIBRADOS_VER_MINOR 49
 #define LIBRADOS_VER_EXTRA 0
 
 #define LIBRADOS_VERSION(maj, min, extra) ((maj << 16) + (min << 8) + extra)
@@ -1444,6 +1444,20 @@ int rados_aio_read(rados_ioctx_t io, const char *oid,
  */
 int rados_aio_flush(rados_ioctx_t io);
 
+
+/**
+ * Asynchronously get object stats (size/mtime)
+ *
+ * @param io ioctx
+ * @param o object name
+ * @param psize where to store object size
+ * @param pmtime where to store modification time
+ * @returns 0 on success, negative error code on failure
+ */
+int rados_aio_stat(rados_ioctx_t io, const char *o,
+		   rados_completion_t completion,
+		   uint64_t *psize, time_t *pmtime);
+
 /** @} Asynchronous I/O */
 
 /**

src/include/rados/librados.hpp

@@ -478,9 +478,11 @@ namespace librados
      * other than CEPH_NOSNAP
      */
     int aio_remove(const std::string& oid, AioCompletion *c);
-    
+
     int aio_flush();
 
+    int aio_stat(const std::string& oid, AioCompletion *c, uint64_t *psize, time_t *pmtime);
+
     int aio_exec(const std::string& oid, AioCompletion *c, const char *cls, const char *method,
 	         bufferlist& inbl, bufferlist *outbl);
 

src/librados/IoCtxImpl.cc

@@ -851,6 +851,21 @@ int librados::IoCtxImpl::aio_remove(const object_t &oid, AioCompletionImpl *c)
   return 0;
 }
 
+
+int librados::IoCtxImpl::aio_stat(const object_t& oid, AioCompletionImpl *c,
+				  uint64_t *psize, time_t *pmtime)
+{
+  c->io = this;
+  C_aio_stat_Ack *onack = new C_aio_stat_Ack(c, pmtime);
+
+  Mutex::Locker l(*lock);
+  objecter->stat(oid, oloc,
+		 snap_seq, psize, &onack->mtime, 0,
+		 onack, &c->objver);
+
+  return 0;
+}
+
 int librados::IoCtxImpl::remove(const object_t& oid)
 {
   utime_t ut = ceph_clock_now(client->cct);
@@ -1564,6 +1579,33 @@ void librados::IoCtxImpl::C_aio_Ack::finish(int r)
   c->put_unlock();
 }
 
+///////////////////////////// C_aio_stat_Ack ////////////////////////////
+
+librados::IoCtxImpl::C_aio_stat_Ack::C_aio_stat_Ack(AioCompletionImpl *_c,
+						    time_t *pm)
+   : c(_c), pmtime(pm)
+{
+  c->get();
+}
+
+void librados::IoCtxImpl::C_aio_stat_Ack::finish(int r)
+{
+  c->lock.Lock();
+  c->rval = r;
+  c->ack = true;
+  c->cond.Signal();
+
+  if (r >= 0 && pmtime) {
+    *pmtime = mtime.sec();
+  }
+
+  if (c->callback_complete) {
+    c->io->client->finisher.queue(new C_AioComplete(c));
+  }
+
+  c->put_unlock();
+}
+
 /////////////////////// C_aio_sparse_read_Ack //////////////////////////
 
 librados::IoCtxImpl::C_aio_sparse_read_Ack::C_aio_sparse_read_Ack(AioCompletionImpl *_c,

src/librados/IoCtxImpl.h

@@ -145,6 +145,14 @@ struct librados::IoCtxImpl {
     void finish(int r);
   };
 
+  struct C_aio_stat_Ack : public Context {
+    librados::AioCompletionImpl *c;
+    time_t *pmtime;
+    utime_t mtime;
+    C_aio_stat_Ack(AioCompletionImpl *_c, time_t *pm);
+    void finish(int r);
+  };
+
   struct C_aio_sparse_read_Ack : public Context {
     AioCompletionImpl *c;
     bufferlist *data_bl;
@@ -177,6 +185,7 @@ struct librados::IoCtxImpl {
   int aio_remove(const object_t &oid, AioCompletionImpl *c);
   int aio_exec(const object_t& oid, AioCompletionImpl *c, const char *cls,
 	       const char *method, bufferlist& inbl, bufferlist *outbl);
+  int aio_stat(const object_t& oid, AioCompletionImpl *c, uint64_t *psize, time_t *pmtime);
 
   int pool_change_auid(unsigned long long auid);
   int pool_change_auid_async(unsigned long long auid, PoolAsyncCompletionImpl *c);

src/librados/librados.cc

@@ -978,6 +978,14 @@ int librados::IoCtx::aio_flush()
   return 0;
 }
 
+int librados::IoCtx::aio_stat(const std::string& oid, librados::AioCompletion *c,
+			      uint64_t *psize, time_t *pmtime)
+{
+  object_t obj(oid);
+  return io_ctx_impl->aio_stat(oid, c->pc, psize, pmtime);
+}
+
+
 int librados::IoCtx::watch(const string& oid, uint64_t ver, uint64_t *cookie,
 			   librados::WatchCtx *ctx)
 {
@@ -2173,6 +2181,16 @@ extern "C" int rados_aio_flush(rados_ioctx_t io)
   return 0;
 }
 
+extern "C" int rados_aio_stat(rados_ioctx_t io, const char *o, 
+			      rados_completion_t completion,
+			      uint64_t *psize, time_t *pmtime)
+{
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  object_t oid(o);
+  return ctx->aio_stat(oid, (librados::AioCompletionImpl*)completion, 
+		       psize, pmtime);
+}
+
 struct C_WatchCB : public librados::WatchCtx {
   rados_watchcb_t wcb;
   void *arg;