Merge pull request ceph#6489 from xiexingguo/xxg-wip-13715

librados: fix pool alignment API overflow issue

Reviewed-by: Kefu Chai <[email protected]>
Reviewed-by: David Zafman <[email protected]>

src/include/rados/librados.h

@@ -809,8 +809,33 @@ CEPH_RADOS_API int rados_ioctx_pool_set_auid(rados_ioctx_t io, uint64_t auid);
  */
 CEPH_RADOS_API int rados_ioctx_pool_get_auid(rados_ioctx_t io, uint64_t *auid);
 
-CEPH_RADOS_API int rados_ioctx_pool_requires_alignment(rados_ioctx_t io);
-CEPH_RADOS_API uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io);
+/* deprecated, use rados_ioctx_pool_requires_alignment2 instead */
+CEPH_RADOS_API int rados_ioctx_pool_requires_alignment(rados_ioctx_t io)
+  __attribute__((deprecated));
+
+/**
+ * Test whether the specified pool requires alignment or not.
+ *
+ * @param io pool to query
+ * @param requires 1 if alignment is supported, 0 if not. 
+ * @returns 0 on success, negative error code on failure
+ */
+CEPH_RADOS_API int rados_ioctx_pool_requires_alignment2(rados_ioctx_t io,
+  int *requires);
+
+/* deprecated, use rados_ioctx_pool_required_alignment2 instead */
+CEPH_RADOS_API uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io)
+  __attribute__((deprecated));
+
+/**
+ * Get the alignment flavor of a pool
+ *
+ * @param io pool to query
+ * @param alignment where to store the alignment flavor
+ * @returns 0 on success, negative error code on failure
+ */
+CEPH_RADOS_API int rados_ioctx_pool_required_alignment2(rados_ioctx_t io,
+  uint64_t *alignment);
 
 /**
  * Get the pool id of the io context

src/include/rados/librados.hpp

@@ -647,7 +647,9 @@ namespace librados
     std::string get_pool_name();
 
     bool pool_requires_alignment();
+    int pool_requires_alignment2(bool * requires);
     uint64_t pool_required_alignment();
+    int pool_required_alignment2(uint64_t * alignment);
 
     // create an object
     int create(const std::string& oid, bool exclusive);

src/librados/RadosClient.cc

@@ -94,30 +94,70 @@ int64_t librados::RadosClient::lookup_pool(const char *name)
 
 bool librados::RadosClient::pool_requires_alignment(int64_t pool_id)
 {
+  bool requires;
+  int r = pool_requires_alignment2(pool_id, &requires);
+  if (r < 0) {
+    // Cast answer to false, this is a little bit problematic
+    // since we really don't know the answer yet, say.
+    return false;
+  }
+
+  return requires;
+}
+
+// a safer version of pool_requires_alignment
+int librados::RadosClient::pool_requires_alignment2(int64_t pool_id,
+	bool *requires)
+{
+  if (!requires)
+    return -EINVAL;
+
   int r = wait_for_osdmap();
   if (r < 0) {
     return r;
   }
 
   const OSDMap *osdmap = objecter->get_osdmap_read();
-  bool ret = osdmap->have_pg_pool(pool_id) &&
-    osdmap->get_pg_pool(pool_id)->requires_aligned_append();
+  if (!osdmap->have_pg_pool(pool_id)) { 
+    objecter->put_osdmap_read();
+    return -ENOENT;
+  }
+  *requires = osdmap->get_pg_pool(pool_id)->requires_aligned_append();
   objecter->put_osdmap_read();
-  return ret;
+  return 0;
 }
 
 uint64_t librados::RadosClient::pool_required_alignment(int64_t pool_id)
 {
+  uint64_t alignment;
+  int r = pool_required_alignment2(pool_id, &alignment);
+  if (r < 0) {
+    return 0;
+  }
+
+  return alignment;
+}
+
+// a safer version of pool_required_alignment
+int librados::RadosClient::pool_required_alignment2(int64_t pool_id,
+	uint64_t *alignment)
+{
+  if (!alignment)
+    return -EINVAL;
+
   int r = wait_for_osdmap();
   if (r < 0) {
     return r;
   }
 
   const OSDMap *osdmap = objecter->get_osdmap_read();
-  uint64_t ret = osdmap->have_pg_pool(pool_id) ?
-    osdmap->get_pg_pool(pool_id)->required_alignment() : 0;
+  if (!osdmap->have_pg_pool(pool_id)) {
+    objecter->put_osdmap_read();
+    return -ENOENT;
+  }
+  *alignment = osdmap->get_pg_pool(pool_id)->required_alignment();
   objecter->put_osdmap_read();
-  return ret;
+  return 0;
 }
 
 int librados::RadosClient::pool_get_auid(uint64_t pool_id, unsigned long long *auid)

src/librados/RadosClient.h

@@ -93,7 +93,9 @@ class librados::RadosClient : public Dispatcher
   int get_fsid(std::string *s);
   int64_t lookup_pool(const char *name);
   bool pool_requires_alignment(int64_t pool_id);
+  int pool_requires_alignment2(int64_t pool_id, bool *requires);
   uint64_t pool_required_alignment(int64_t pool_id);
+  int pool_required_alignment2(int64_t pool_id, uint64_t *alignment);
   int pool_get_auid(uint64_t pool_id, unsigned long long *auid);
   int pool_get_name(uint64_t pool_id, std::string *auid);
 

src/librados/librados.cc

@@ -1086,11 +1086,21 @@ bool librados::IoCtx::pool_requires_alignment()
   return io_ctx_impl->client->pool_requires_alignment(get_id());
 }
 
+int librados::IoCtx::pool_requires_alignment2(bool *requires)
+{
+  return io_ctx_impl->client->pool_requires_alignment2(get_id(), requires);
+}
+
 uint64_t librados::IoCtx::pool_required_alignment()
 {
   return io_ctx_impl->client->pool_required_alignment(get_id());
 }
 
+int librados::IoCtx::pool_required_alignment2(uint64_t *alignment)
+{
+  return io_ctx_impl->client->pool_required_alignment2(get_id(), alignment);
+}
+
 std::string librados::IoCtx::get_pool_name()
 {
   std::string s;
@@ -3143,6 +3153,18 @@ extern "C" int rados_ioctx_pool_requires_alignment(rados_ioctx_t io)
   return retval;
 }
 
+extern "C" int rados_ioctx_pool_requires_alignment2(rados_ioctx_t io,
+	int *requires)
+{
+  tracepoint(librados, rados_ioctx_pool_requires_alignment_enter2, io);
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  int retval = ctx->client->pool_requires_alignment2(ctx->get_id(), 
+  	(bool *)requires);
+  tracepoint(librados, rados_ioctx_pool_requires_alignment_exit2, retval, 
+  	*requires);
+  return retval;
+}
+
 extern "C" uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io)
 {
   tracepoint(librados, rados_ioctx_pool_required_alignment_enter, io);
@@ -3152,6 +3174,18 @@ extern "C" uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io)
   return retval;
 }
 
+extern "C" int rados_ioctx_pool_required_alignment2(rados_ioctx_t io,
+	uint64_t *alignment)
+{
+  tracepoint(librados, rados_ioctx_pool_required_alignment_enter2, io);
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  int retval = ctx->client->pool_required_alignment2(ctx->get_id(),
+  	alignment);
+  tracepoint(librados, rados_ioctx_pool_required_alignment_exit2, retval, 
+  	*alignment);
+  return retval;
+}
+
 extern "C" void rados_ioctx_locator_set_key(rados_ioctx_t io, const char *key)
 {
   tracepoint(librados, rados_ioctx_locator_set_key_enter, io, key);

src/rgw/rgw_rados.cc

@@ -1401,7 +1401,27 @@ int RGWRados::get_required_alignment(rgw_bucket& bucket, uint64_t *alignment)
     return r;
   }
 
-  *alignment = ioctx.pool_required_alignment();
+  bool requires;
+  r = ioctx.pool_requires_alignment2(&requires);
+  if (r < 0) {
+    ldout(cct, 0) << "ERROR: ioctx.pool_requires_alignment2() returned " 
+      << r << dendl;
+    return r;
+  }
+
+  if (!requires) {
+    *alignment = 0;
+    return 0;
+  }
+
+  uint64_t align;
+  r = ioctx.pool_required_alignment2(&align);
+  if (r < 0) {
+    ldout(cct, 0) << "ERROR: ioctx.pool_required_alignment2() returned " 
+      << r << dendl;
+    return r;
+  }
+  *alignment = align;
   return 0;
 }
 

src/tools/rados/rados.cc

@@ -1450,14 +1450,31 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
       goto out;
     }
 
-    // align op_size
-    if (io_ctx.pool_requires_alignment()) {
-      const uint64_t align = io_ctx.pool_required_alignment();
-      const uint64_t prev_op_size = op_size;
-      op_size = uint64_t((op_size + align - 1) / align) * align;
-      // Warn: if user specified and it was rounded
-      if (prev_op_size != default_op_size && prev_op_size != op_size)
-	cerr << "INFO: op_size has been rounded to " << op_size << std::endl;
+   // align op_size
+   {
+      bool requires;
+      ret = io_ctx.pool_requires_alignment2(&requires);
+      if (ret < 0) {
+        cerr << "error checking pool alignment requirement"
+          << cpp_strerror(ret) << std::endl;
+        goto out;	
+      }
+
+      if (requires) {
+        uint64_t align = 0;
+        ret = io_ctx.pool_required_alignment2(&align);
+        if (ret < 0) {
+          cerr << "error getting pool alignment"
+            << cpp_strerror(ret) << std::endl;
+          goto out;	
+        }
+
+        const uint64_t prev_op_size = op_size;
+        op_size = uint64_t((op_size + align - 1) / align) * align;
+        // Warn: if user specified and it was rounded
+        if (prev_op_size != default_op_size && prev_op_size != op_size)
+          cerr << "INFO: op_size has been rounded to " << op_size << std::endl;
+      }
     }
 
     // create striper interface

src/tracing/librados.tp

@@ -1091,6 +1091,24 @@ TRACEPOINT_EVENT(librados, rados_ioctx_pool_requires_alignment_exit,
     )
 )
 
+TRACEPOINT_EVENT(librados, rados_ioctx_pool_requires_alignment_enter2,
+    TP_ARGS(
+        rados_ioctx_t, ioctx),
+    TP_FIELDS(
+        ctf_integer_hex(rados_ioctx_t, ioctx, ioctx)
+    )
+)
+
+TRACEPOINT_EVENT(librados, rados_ioctx_pool_requires_alignment_exit2,
+    TP_ARGS(
+        int, retval,
+        int, requires),
+    TP_FIELDS(
+        ctf_integer(int, retval, retval)
+        ctf_integer(int, requires, requires)
+    )
+)
+
 TRACEPOINT_EVENT(librados, rados_ioctx_pool_required_alignment_enter,
     TP_ARGS(
         rados_ioctx_t, ioctx),
@@ -1107,6 +1125,24 @@ TRACEPOINT_EVENT(librados, rados_ioctx_pool_required_alignment_exit,
     )
 )
 
+TRACEPOINT_EVENT(librados, rados_ioctx_pool_required_alignment_enter2,
+    TP_ARGS(
+        rados_ioctx_t, ioctx),
+    TP_FIELDS(
+        ctf_integer_hex(rados_ioctx_t, ioctx, ioctx)
+    )
+)
+
+TRACEPOINT_EVENT(librados, rados_ioctx_pool_required_alignment_exit2,
+    TP_ARGS(
+        int, retval,
+        uint64_t, alignment),
+    TP_FIELDS(
+        ctf_integer(int, retval, retval)
+        ctf_integer(uint64_t, alignment, alignment)
+    )
+)
+
 TRACEPOINT_EVENT(librados, rados_ioctx_locator_set_key_enter,
     TP_ARGS(
         rados_ioctx_t, ioctx,