libceph: introduce ceph_osdc_abort_requests()

This will be used by the filesystem for "umount -f".

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index b73dd7ebe585a523f32c5cf552b5d786b33e8dd2..874c31c01f80f228d471a4997cc184acac62d561 100644 (file)
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -347,6 +347,7 @@ struct ceph_osd_client {
        struct rb_root         linger_map_checks;
        atomic_t               num_requests;
        atomic_t               num_homeless;
+       int                    abort_err;
        struct delayed_work    timeout_work;
        struct delayed_work    osds_timeout_work;
 #ifdef CONFIG_DEBUG_FS
@@ -378,6 +379,7 @@ extern void ceph_osdc_handle_reply(struct ceph_osd_client *osdc,
 extern void ceph_osdc_handle_map(struct ceph_osd_client *osdc,
                                 struct ceph_msg *msg);
 void ceph_osdc_update_epoch_barrier(struct ceph_osd_client *osdc, u32 eb);
+void ceph_osdc_abort_requests(struct ceph_osd_client *osdc, int err);
 
 extern void osd_req_op_init(struct ceph_osd_request *osd_req,
                            unsigned int which, u16 opcode, u32 flags);

diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 08b5fc1f90ccfff373b54d0bc0bc8a649326dc48..a7e090d2c9571e2b8e85c24793511c0046997b83 100644 (file)
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -1058,6 +1058,38 @@ EXPORT_SYMBOL(ceph_osdc_new_request);
 DEFINE_RB_FUNCS(request, struct ceph_osd_request, r_tid, r_node)
 DEFINE_RB_FUNCS(request_mc, struct ceph_osd_request, r_tid, r_mc_node)
 
+/*
+ * Call @fn on each OSD request as long as @fn returns 0.
+ */
+static void for_each_request(struct ceph_osd_client *osdc,
+                       int (*fn)(struct ceph_osd_request *req, void *arg),
+                       void *arg)
+{
+       struct rb_node *n, *p;
+
+       for (n = rb_first(&osdc->osds); n; n = rb_next(n)) {
+               struct ceph_osd *osd = rb_entry(n, struct ceph_osd, o_node);
+
+               for (p = rb_first(&osd->o_requests); p; ) {
+                       struct ceph_osd_request *req =
+                           rb_entry(p, struct ceph_osd_request, r_node);
+
+                       p = rb_next(p);
+                       if (fn(req, arg))
+                               return;
+               }
+       }
+
+       for (p = rb_first(&osdc->homeless_osd.o_requests); p; ) {
+               struct ceph_osd_request *req =
+                   rb_entry(p, struct ceph_osd_request, r_node);
+
+               p = rb_next(p);
+               if (fn(req, arg))
+                       return;
+       }
+}
+
 static bool osd_homeless(struct ceph_osd *osd)
 {
        return osd->o_osd == CEPH_HOMELESS_OSD;
@@ -2165,9 +2197,9 @@ static void __submit_request(struct ceph_osd_request *req, bool wrlocked)
        struct ceph_osd_client *osdc = req->r_osdc;
        struct ceph_osd *osd;
        enum calc_target_result ct_res;
+       int err = 0;
        bool need_send = false;
        bool promoted = false;
-       bool need_abort = false;
 
        WARN_ON(req->r_tid);
        dout("%s req %p wrlocked %d\n", __func__, req, wrlocked);
@@ -2183,7 +2215,10 @@ again:
                goto promote;
        }
 
-       if (osdc->osdmap->epoch < osdc->epoch_barrier) {
+       if (osdc->abort_err) {
+               dout("req %p abort_err %d\n", req, osdc->abort_err);
+               err = osdc->abort_err;
+       } else if (osdc->osdmap->epoch < osdc->epoch_barrier) {
                dout("req %p epoch %u barrier %u\n", req, osdc->osdmap->epoch,
                     osdc->epoch_barrier);
                req->r_t.paused = true;
@@ -2208,7 +2243,7 @@ again:
                req->r_t.paused = true;
                maybe_request_map(osdc);
                if (req->r_abort_on_full)
-                       need_abort = true;
+                       err = -ENOSPC;
        } else if (!osd_homeless(osd)) {
                need_send = true;
        } else {
@@ -2225,8 +2260,8 @@ again:
        link_request(osd, req);
        if (need_send)
                send_request(req);
-       else if (need_abort)
-               complete_request(req, -ENOSPC);
+       else if (err)
+               complete_request(req, err);
        mutex_unlock(&osd->lock);
 
        if (ct_res == CALC_TARGET_POOL_DNE)
@@ -2340,6 +2375,28 @@ static void abort_request(struct ceph_osd_request *req, int err)
        complete_request(req, err);
 }
 
+static int abort_fn(struct ceph_osd_request *req, void *arg)
+{
+       int err = *(int *)arg;
+
+       abort_request(req, err);
+       return 0; /* continue iteration */
+}
+
+/*
+ * Abort all in-flight requests with @err and arrange for all future
+ * requests to be failed immediately.
+ */
+void ceph_osdc_abort_requests(struct ceph_osd_client *osdc, int err)
+{
+       dout("%s osdc %p err %d\n", __func__, osdc, err);
+       down_write(&osdc->lock);
+       for_each_request(osdc, abort_fn, &err);
+       osdc->abort_err = err;
+       up_write(&osdc->lock);
+}
+EXPORT_SYMBOL(ceph_osdc_abort_requests);
+
 static void update_epoch_barrier(struct ceph_osd_client *osdc, u32 eb)
 {
        if (likely(eb > osdc->epoch_barrier)) {