Merge tag 'ceph-for-4.8-rc1' of git://github.com/ceph/ceph-client

Pull Ceph updates from Ilya Dryomov:
 "The highlights are:

   - RADOS namespace support in libceph and CephFS (Zheng Yan and
     myself).  The stopgaps added in 4.5 to deny access to inodes in
     namespaces are removed and CEPH_FEATURE_FS_FILE_LAYOUT_V2 feature
     bit is now fully supported

   - A large rework of the MDS cap flushing code (Zheng Yan)

   - Handle some of ->d_revalidate() in RCU mode (Jeff Layton).  We were
     overly pessimistic before, bailing at the first sight of LOOKUP_RCU

  On top of that we've got a few CephFS bug fixes, a couple of cleanups
  and Arnd's workaround for a weird genksyms issue"

* tag 'ceph-for-4.8-rc1' of git://github.com/ceph/ceph-client: (34 commits)
  ceph: fix symbol versioning for ceph_monc_do_statfs
  ceph: Correctly return NXIO errors from ceph_llseek
  ceph: Mark the file cache as unreclaimable
  ceph: optimize cap flush waiting
  ceph: cleanup ceph_flush_snaps()
  ceph: kick cap flushes before sending other cap message
  ceph: introduce an inode flag to indicates if snapflush is needed
  ceph: avoid sending duplicated cap flush message
  ceph: unify cap flush and snapcap flush
  ceph: use list instead of rbtree to track cap flushes
  ceph: update types of some local varibles
  ceph: include 'follows' of pending snapflush in cap reconnect message
  ceph: update cap reconnect message to version 3
  ceph: mount non-default filesystem by name
  libceph: fsmap.user subscription support
  ceph: handle LOOKUP_RCU in ceph_d_revalidate
  ceph: allow dentry_lease_is_valid to work under RCU walk
  ceph: clear d_fsinfo pointer under d_lock
  ceph: remove ceph_mdsc_lease_release
  ceph: don't use ->d_time
  ...

9 files changed, 245 insertions, 24 deletions

diff --git a/net/ceph/Makefile b/net/ceph/Makefile
index 958d9856912c..84cbed630c4b 100644
--- a/net/ceph/Makefile
+++ b/net/ceph/Makefile
@@ -11,5 +11,5 @@ libceph-y := ceph_common.o messenger.o msgpool.o buffer.o pagelist.o \
         crypto.o armor.o \
         auth_x.o \
         ceph_fs.o ceph_strings.o ceph_hash.o \
-        pagevec.o snapshot.o
+        pagevec.o snapshot.o string_table.o
 
diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 55d2bfee16d7..bddfcf6f09c2 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -747,6 +747,8 @@ out:
 static void __exit exit_ceph_lib(void)
 {
         dout("exit_ceph_lib\n");
+        WARN_ON(!ceph_strings_empty());
+
         ceph_osdc_cleanup();
         ceph_msgr_exit();
         ceph_crypto_shutdown();
diff --git a/net/ceph/ceph_fs.c b/net/ceph/ceph_fs.c
index 41466ccb972a..7d54e944de5e 100644
--- a/net/ceph/ceph_fs.c
+++ b/net/ceph/ceph_fs.c
@@ -9,9 +9,9 @@
  */
 int ceph_file_layout_is_valid(const struct ceph_file_layout *layout)
 {
-        __u32 su = le32_to_cpu(layout->fl_stripe_unit);
-        __u32 sc = le32_to_cpu(layout->fl_stripe_count);
-        __u32 os = le32_to_cpu(layout->fl_object_size);
+        __u32 su = layout->stripe_unit;
+        __u32 sc = layout->stripe_count;
+        __u32 os = layout->object_size;
 
         /* stripe unit, object size must be non-zero, 64k increment */
         if (!su || (su & (CEPH_MIN_STRIPE_UNIT-1)))
@@ -27,6 +27,30 @@ int ceph_file_layout_is_valid(const struct ceph_file_layout *layout)
         return 1;
 }
 
+void ceph_file_layout_from_legacy(struct ceph_file_layout *fl,
+                                  struct ceph_file_layout_legacy *legacy)
+{
+        fl->stripe_unit = le32_to_cpu(legacy->fl_stripe_unit);
+        fl->stripe_count = le32_to_cpu(legacy->fl_stripe_count);
+        fl->object_size = le32_to_cpu(legacy->fl_object_size);
+        fl->pool_id = le32_to_cpu(legacy->fl_pg_pool);
+        if (fl->pool_id == 0)
+                fl->pool_id = -1;
+}
+EXPORT_SYMBOL(ceph_file_layout_from_legacy);
+
+void ceph_file_layout_to_legacy(struct ceph_file_layout *fl,
+                                struct ceph_file_layout_legacy *legacy)
+{
+        legacy->fl_stripe_unit = cpu_to_le32(fl->stripe_unit);
+        legacy->fl_stripe_count = cpu_to_le32(fl->stripe_count);
+        legacy->fl_object_size = cpu_to_le32(fl->object_size);
+        if (fl->pool_id >= 0)
+                legacy->fl_pg_pool = cpu_to_le32(fl->pool_id);
+        else
+                legacy->fl_pg_pool = 0;
+}
+EXPORT_SYMBOL(ceph_file_layout_to_legacy);
 
 int ceph_flags_to_mode(int flags)
 {
diff --git a/net/ceph/debugfs.c b/net/ceph/debugfs.c
index e77b04ca7802..c62b2b029a6e 100644
--- a/net/ceph/debugfs.c
+++ b/net/ceph/debugfs.c
@@ -156,8 +156,16 @@ static void dump_target(struct seq_file *s, struct ceph_osd_request_target *t)
         seq_printf(s, "]/%d\t[", t->up.primary);
         for (i = 0; i < t->acting.size; i++)
                 seq_printf(s, "%s%d", (!i ? "" : ","), t->acting.osds[i]);
-        seq_printf(s, "]/%d\t%*pE\t0x%x", t->acting.primary,
-                   t->target_oid.name_len, t->target_oid.name, t->flags);
+        seq_printf(s, "]/%d\t", t->acting.primary);
+        if (t->target_oloc.pool_ns) {
+                seq_printf(s, "%*pE/%*pE\t0x%x",
+                        (int)t->target_oloc.pool_ns->len,
+                        t->target_oloc.pool_ns->str,
+                        t->target_oid.name_len, t->target_oid.name, t->flags);
+        } else {
+                seq_printf(s, "%*pE\t0x%x", t->target_oid.name_len,
+                        t->target_oid.name, t->flags);
+        }
         if (t->paused)
                 seq_puts(s, "\tP");
 }
diff --git a/net/ceph/mon_client.c b/net/ceph/mon_client.c
index 37c38a7fb5c5..c83326c5ba58 100644
--- a/net/ceph/mon_client.c
+++ b/net/ceph/mon_client.c
@@ -227,9 +227,10 @@ static void __schedule_delayed(struct ceph_mon_client *monc)
 }
 
 const char *ceph_sub_str[] = {
-        [CEPH_SUB_MDSMAP] = "mdsmap",
         [CEPH_SUB_MONMAP] = "monmap",
         [CEPH_SUB_OSDMAP] = "osdmap",
+        [CEPH_SUB_FSMAP]  = "fsmap.user",
+        [CEPH_SUB_MDSMAP] = "mdsmap",
 };
 
 /*
@@ -1193,6 +1194,7 @@ static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
         case CEPH_MSG_MON_MAP:
         case CEPH_MSG_MDS_MAP:
         case CEPH_MSG_OSD_MAP:
+        case CEPH_MSG_FS_MAP_USER:
                 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
                 if (!m)
                         return NULL;    /* ENOMEM--return skip == 0 */
diff --git a/net/ceph/msgpool.c b/net/ceph/msgpool.c
index ddec1c10ac80..aaed59a47b1d 100644
--- a/net/ceph/msgpool.c
+++ b/net/ceph/msgpool.c
@@ -5,6 +5,7 @@
 #include <linux/types.h>
 #include <linux/vmalloc.h>
 
+#include <linux/ceph/messenger.h>
 #include <linux/ceph/msgpool.h>
 
 static void *msgpool_alloc(gfp_t gfp_mask, void *arg)
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 89469592076c..b5ec09612ff7 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -387,7 +387,9 @@ static void target_copy(struct ceph_osd_request_target *dest,
 static void target_destroy(struct ceph_osd_request_target *t)
 {
         ceph_oid_destroy(&t->base_oid);
+        ceph_oloc_destroy(&t->base_oloc);
         ceph_oid_destroy(&t->target_oid);
+        ceph_oloc_destroy(&t->target_oloc);
 }
 
 /*
@@ -533,6 +535,11 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 }
 EXPORT_SYMBOL(ceph_osdc_alloc_request);
 
+static int ceph_oloc_encoding_size(struct ceph_object_locator *oloc)
+{
+        return 8 + 4 + 4 + 4 + (oloc->pool_ns ? oloc->pool_ns->len : 0);
+}
+
 int ceph_osdc_alloc_messages(struct ceph_osd_request *req, gfp_t gfp)
 {
         struct ceph_osd_client *osdc = req->r_osdc;
@@ -540,11 +547,13 @@ int ceph_osdc_alloc_messages(struct ceph_osd_request *req, gfp_t gfp)
         int msg_size;
 
         WARN_ON(ceph_oid_empty(&req->r_base_oid));
+        WARN_ON(ceph_oloc_empty(&req->r_base_oloc));
 
         /* create request message */
         msg_size = 4 + 4 + 4; /* client_inc, osdmap_epoch, flags */
         msg_size += 4 + 4 + 4 + 8; /* mtime, reassert_version */
-        msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
+        msg_size += CEPH_ENCODING_START_BLK_LEN +
+                        ceph_oloc_encoding_size(&req->r_base_oloc); /* oloc */
         msg_size += 1 + 8 + 4 + 4; /* pgid */
         msg_size += 4 + req->r_base_oid.name_len; /* oid */
         msg_size += 2 + req->r_num_ops * sizeof(struct ceph_osd_op);
@@ -932,7 +941,7 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
         if (opcode == CEPH_OSD_OP_CREATE || opcode == CEPH_OSD_OP_DELETE) {
                 osd_req_op_init(req, which, opcode, 0);
         } else {
-                u32 object_size = le32_to_cpu(layout->fl_object_size);
+                u32 object_size = layout->object_size;
                 u32 object_base = off - objoff;
                 if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
                         if (truncate_size <= object_base) {
@@ -948,7 +957,8 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
         }
 
         req->r_flags = flags;
-        req->r_base_oloc.pool = ceph_file_layout_pg_pool(*layout);
+        req->r_base_oloc.pool = layout->pool_id;
+        req->r_base_oloc.pool_ns = ceph_try_get_string(layout->pool_ns);
         ceph_oid_printf(&req->r_base_oid, "%llx.%08llx", vino.ino, objnum);
 
         req->r_snapid = vino.snap;
@@ -1489,12 +1499,16 @@ static void encode_request(struct ceph_osd_request *req, struct ceph_msg *msg)
         p += sizeof(req->r_replay_version);
 
         /* oloc */
-        ceph_encode_8(&p, 4);
-        ceph_encode_8(&p, 4);
-        ceph_encode_32(&p, 8 + 4 + 4);
+        ceph_start_encoding(&p, 5, 4,
+                            ceph_oloc_encoding_size(&req->r_t.target_oloc));
         ceph_encode_64(&p, req->r_t.target_oloc.pool);
         ceph_encode_32(&p, -1); /* preferred */
         ceph_encode_32(&p, 0); /* key len */
+        if (req->r_t.target_oloc.pool_ns)
+                ceph_encode_string(&p, end, req->r_t.target_oloc.pool_ns->str,
+                                   req->r_t.target_oloc.pool_ns->len);
+        else
+                ceph_encode_32(&p, 0);
 
         /* pgid */
         ceph_encode_8(&p, 1);
@@ -2594,9 +2608,22 @@ static int ceph_oloc_decode(void **p, void *end,
         }
 
         if (struct_v >= 5) {
+                bool changed = false;
+
                 len = ceph_decode_32(p);
                 if (len > 0) {
-                        pr_warn("ceph_object_locator::nspace is set\n");
+                        ceph_decode_need(p, end, len, e_inval);
+                        if (!oloc->pool_ns ||
+                            ceph_compare_string(oloc->pool_ns, *p, len))
+                                changed = true;
+                        *p += len;
+                } else {
+                        if (oloc->pool_ns)
+                                changed = true;
+                }
+                if (changed) {
+                        /* redirect changes namespace */
+                        pr_warn("ceph_object_locator::nspace is changed\n");
                         goto e_inval;
                 }
         }
@@ -2806,7 +2833,9 @@ static void handle_reply(struct ceph_osd *osd, struct ceph_msg *msg)
                 goto out_unlock_session;
         }
 
+        m.redirect.oloc.pool_ns = req->r_t.target_oloc.pool_ns;
         ret = decode_MOSDOpReply(msg, &m);
+        m.redirect.oloc.pool_ns = NULL;
         if (ret) {
                 pr_err("failed to decode MOSDOpReply for tid %llu: %d\n",
                        req->r_tid, ret);
@@ -2835,7 +2864,11 @@ static void handle_reply(struct ceph_osd *osd, struct ceph_msg *msg)
                 unlink_request(osd, req);
                 mutex_unlock(&osd->lock);
 
-                ceph_oloc_copy(&req->r_t.target_oloc, &m.redirect.oloc);
+                /*
+                 * Not ceph_oloc_copy() - changing pool_ns is not
+                 * supported.
+                 */
+                req->r_t.target_oloc.pool = m.redirect.oloc.pool;
                 req->r_flags |= CEPH_OSD_FLAG_REDIRECTED;
                 req->r_tid = 0;
                 __submit_request(req, false);
diff --git a/net/ceph/osdmap.c b/net/ceph/osdmap.c
index 7e480bf75bcf..d2436880b305 100644
--- a/net/ceph/osdmap.c
+++ b/net/ceph/osdmap.c
@@ -1510,6 +1510,24 @@ bad:
         return ERR_PTR(err);
 }
 
+void ceph_oloc_copy(struct ceph_object_locator *dest,
+                    const struct ceph_object_locator *src)
+{
+        WARN_ON(!ceph_oloc_empty(dest));
+        WARN_ON(dest->pool_ns); /* empty() only covers ->pool */
+
+        dest->pool = src->pool;
+        if (src->pool_ns)
+                dest->pool_ns = ceph_get_string(src->pool_ns);
+}
+EXPORT_SYMBOL(ceph_oloc_copy);
+
+void ceph_oloc_destroy(struct ceph_object_locator *oloc)
+{
+        ceph_put_string(oloc->pool_ns);
+}
+EXPORT_SYMBOL(ceph_oloc_destroy);
+
 void ceph_oid_copy(struct ceph_object_id *dest,
                    const struct ceph_object_id *src)
 {
@@ -1770,9 +1788,9 @@ int ceph_calc_file_object_mapping(struct ceph_file_layout *layout,
                                    u64 *ono,
                                    u64 *oxoff, u64 *oxlen)
 {
-        u32 osize = le32_to_cpu(layout->fl_object_size);
-        u32 su = le32_to_cpu(layout->fl_stripe_unit);
-        u32 sc = le32_to_cpu(layout->fl_stripe_count);
+        u32 osize = layout->object_size;
+        u32 su = layout->stripe_unit;
+        u32 sc = layout->stripe_count;
         u32 bl, stripeno, stripepos, objsetno;
         u32 su_per_object;
         u64 t, su_offset;
@@ -1844,12 +1862,34 @@ int ceph_object_locator_to_pg(struct ceph_osdmap *osdmap,
         if (!pi)
                 return -ENOENT;
 
-        raw_pgid->pool = oloc->pool;
-        raw_pgid->seed = ceph_str_hash(pi->object_hash, oid->name,
-                                       oid->name_len);
-
-        dout("%s %s -> raw_pgid %llu.%x\n", __func__, oid->name,
-             raw_pgid->pool, raw_pgid->seed);
+        if (!oloc->pool_ns) {
+                raw_pgid->pool = oloc->pool;
+                raw_pgid->seed = ceph_str_hash(pi->object_hash, oid->name,
+                                             oid->name_len);
+                dout("%s %s -> raw_pgid %llu.%x\n", __func__, oid->name,
+                     raw_pgid->pool, raw_pgid->seed);
+        } else {
+                char stack_buf[256];
+                char *buf = stack_buf;
+                int nsl = oloc->pool_ns->len;
+                size_t total = nsl + 1 + oid->name_len;
+
+                if (total > sizeof(stack_buf)) {
+                        buf = kmalloc(total, GFP_NOIO);
+                        if (!buf)
+                                return -ENOMEM;
+                }
+                memcpy(buf, oloc->pool_ns->str, nsl);
+                buf[nsl] = '\037';
+                memcpy(buf + nsl + 1, oid->name, oid->name_len);
+                raw_pgid->pool = oloc->pool;
+                raw_pgid->seed = ceph_str_hash(pi->object_hash, buf, total);
+                if (buf != stack_buf)
+                        kfree(buf);
+                dout("%s %s ns %.*s -> raw_pgid %llu.%x\n", __func__,
+                     oid->name, nsl, oloc->pool_ns->str,
+                     raw_pgid->pool, raw_pgid->seed);
+        }
         return 0;
 }
 EXPORT_SYMBOL(ceph_object_locator_to_pg);
diff --git a/net/ceph/string_table.c b/net/ceph/string_table.c
new file mode 100644
index 000000000000..ca53c8319209
--- /dev/null
+++ b/net/ceph/string_table.c
@@ -0,0 +1,111 @@
+#include <linux/slab.h>
+#include <linux/gfp.h>
+#include <linux/string.h>
+#include <linux/spinlock.h>
+#include <linux/ceph/string_table.h>
+
+static DEFINE_SPINLOCK(string_tree_lock);
+static struct rb_root string_tree = RB_ROOT;
+
+struct ceph_string *ceph_find_or_create_string(const char* str, size_t len)
+{
+        struct ceph_string *cs, *exist;
+        struct rb_node **p, *parent;
+        int ret;
+
+        exist = NULL;
+        spin_lock(&string_tree_lock);
+        p = &string_tree.rb_node;
+        while (*p) {
+                exist = rb_entry(*p, struct ceph_string, node);
+                ret = ceph_compare_string(exist, str, len);
+                if (ret > 0)
+                        p = &(*p)->rb_left;
+                else if (ret < 0)
+                        p = &(*p)->rb_right;
+                else
+                        break;
+                exist = NULL;
+        }
+        if (exist && !kref_get_unless_zero(&exist->kref)) {
+                rb_erase(&exist->node, &string_tree);
+                RB_CLEAR_NODE(&exist->node);
+                exist = NULL;
+        }
+        spin_unlock(&string_tree_lock);
+        if (exist)
+                return exist;
+
+        cs = kmalloc(sizeof(*cs) + len + 1, GFP_NOFS);
+        if (!cs)
+                return NULL;
+
+        kref_init(&cs->kref);
+        cs->len = len;
+        memcpy(cs->str, str, len);
+        cs->str[len] = 0;
+
+retry:
+        exist = NULL;
+        parent = NULL;
+        p = &string_tree.rb_node;
+        spin_lock(&string_tree_lock);
+        while (*p) {
+                parent = *p;
+                exist = rb_entry(*p, struct ceph_string, node);
+                ret = ceph_compare_string(exist, str, len);
+                if (ret > 0)
+                        p = &(*p)->rb_left;
+                else if (ret < 0)
+                        p = &(*p)->rb_right;
+                else
+                        break;
+                exist = NULL;
+        }
+        ret = 0;
+        if (!exist) {
+                rb_link_node(&cs->node, parent, p);
+                rb_insert_color(&cs->node, &string_tree);
+        } else if (!kref_get_unless_zero(&exist->kref)) {
+                rb_erase(&exist->node, &string_tree);
+                RB_CLEAR_NODE(&exist->node);
+                ret = -EAGAIN;
+        }
+        spin_unlock(&string_tree_lock);
+        if (ret == -EAGAIN)
+                goto retry;
+
+        if (exist) {
+                kfree(cs);
+                cs = exist;
+        }
+
+        return cs;
+}
+EXPORT_SYMBOL(ceph_find_or_create_string);
+
+static void ceph_free_string(struct rcu_head *head)
+{
+        struct ceph_string *cs = container_of(head, struct ceph_string, rcu);
+        kfree(cs);
+}
+
+void ceph_release_string(struct kref *ref)
+{
+        struct ceph_string *cs = container_of(ref, struct ceph_string, kref);
+
+        spin_lock(&string_tree_lock);
+        if (!RB_EMPTY_NODE(&cs->node)) {
+                rb_erase(&cs->node, &string_tree);
+                RB_CLEAR_NODE(&cs->node);
+        }
+        spin_unlock(&string_tree_lock);
+
+        call_rcu(&cs->rcu, ceph_free_string);
+}
+EXPORT_SYMBOL(ceph_release_string);
+
+bool ceph_strings_empty(void)
+{
+        return RB_EMPTY_ROOT(&string_tree);
+}