ceph: handle epoch barriers in cap messages

Have the client store and update the osdc epoch_barrier when a cap
message comes in with one.

When sending cap messages, send the epoch barrier as well. This allows
clients to inform servers that their released caps may not be used until
a particular OSD map epoch.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
Reviewed-by: "Yan, Zheng” <zyan@redhat.com>
Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

1 files changed, 16 insertions, 5 deletions

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 60185434162a..a3ebb632294e 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -1015,6 +1015,7 @@ static int send_cap_msg(struct cap_msg_args *arg)
         void *p;
         size_t extra_len;
         struct timespec zerotime = {0};
+        struct ceph_osd_client *osdc = &arg->session->s_mdsc->fsc->client->osdc;
 
         dout("send_cap_msg %s %llx %llx caps %s wanted %s dirty %s"
              " seq %u/%u tid %llu/%llu mseq %u follows %lld size %llu/%llu"
@@ -1076,8 +1077,12 @@ static int send_cap_msg(struct cap_msg_args *arg)
         ceph_encode_64(&p, arg->inline_data ? 0 : CEPH_INLINE_NONE);
         /* inline data size */
         ceph_encode_32(&p, 0);
-        /* osd_epoch_barrier (version 5) */
-        ceph_encode_32(&p, 0);
+        /*
+         * osd_epoch_barrier (version 5)
+         * The epoch_barrier is protected osdc->lock, so READ_ONCE here in
+         * case it was recently changed
+         */
+        ceph_encode_32(&p, READ_ONCE(osdc->epoch_barrier));
         /* oldest_flush_tid (version 6) */
         ceph_encode_64(&p, arg->oldest_flush_tid);
 
@@ -3633,13 +3638,19 @@ void ceph_handle_caps(struct ceph_mds_session *session,
                 p += inline_len;
         }
 
+        if (le16_to_cpu(msg->hdr.version) >= 5) {
+                struct ceph_osd_client  *osdc = &mdsc->fsc->client->osdc;
+                u32                     epoch_barrier;
+
+                ceph_decode_32_safe(&p, end, epoch_barrier, bad);
+                ceph_osdc_update_epoch_barrier(osdc, epoch_barrier);
+        }
+
         if (le16_to_cpu(msg->hdr.version) >= 8) {
                 u64 flush_tid;
                 u32 caller_uid, caller_gid;
-                u32 osd_epoch_barrier;
                 u32 pool_ns_len;
-                /* version >= 5 */
-                ceph_decode_32_safe(&p, end, osd_epoch_barrier, bad);
+
                 /* version >= 6 */
                 ceph_decode_64_safe(&p, end, flush_tid, bad);
                 /* version >= 7 */