libceph: fix error handling in handle_reply()

[karo-tx-linux.git] / net / ceph / osd_client.c

diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 3a246a6cab473496e58275cbb367485acbbda4d8..3ee4a1da02cd0a8b5812635c45f511d51bae7727 100644 (file)
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -503,7 +503,9 @@ void osd_req_op_extent_init(struct ceph_osd_request *osd_req,
        struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which, opcode);
        size_t payload_len = 0;
 
-       BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE);
+       BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
+              opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
+              opcode != CEPH_OSD_OP_TRUNCATE);
 
        op->extent.offset = offset;
        op->extent.length = length;
@@ -631,6 +633,9 @@ static u64 osd_req_encode_op(struct ceph_osd_request *req,
                break;
        case CEPH_OSD_OP_READ:
        case CEPH_OSD_OP_WRITE:
+       case CEPH_OSD_OP_ZERO:
+       case CEPH_OSD_OP_DELETE:
+       case CEPH_OSD_OP_TRUNCATE:
                if (src->op == CEPH_OSD_OP_WRITE)
                        request_data_len = src->extent.length;
                dst->extent.offset = cpu_to_le64(src->extent.offset);
@@ -715,7 +720,9 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
        u64 object_base;
        int r;
 
-       BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE);
+       BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
+              opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
+              opcode != CEPH_OSD_OP_TRUNCATE);
 
        req = ceph_osdc_alloc_request(osdc, snapc, num_ops, use_mempool,
                                        GFP_NOFS);
@@ -733,12 +740,14 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
 
        object_size = le32_to_cpu(layout->fl_object_size);
        object_base = off - objoff;
-       if (truncate_size <= object_base) {
-               truncate_size = 0;
-       } else {
-               truncate_size -= object_base;
-               if (truncate_size > object_size)
-                       truncate_size = object_size;
+       if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
+               if (truncate_size <= object_base) {
+                       truncate_size = 0;
+               } else {
+                       truncate_size -= object_base;
+                       if (truncate_size > object_size)
+                               truncate_size = object_size;
+               }
        }
 
        osd_req_op_extent_init(req, 0, opcode, objoff, objlen,
@@ -1174,6 +1183,7 @@ static void __register_linger_request(struct ceph_osd_client *osdc,
                                    struct ceph_osd_request *req)
 {
        dout("__register_linger_request %p\n", req);
+       ceph_osdc_get_request(req);
        list_add_tail(&req->r_linger_item, &osdc->req_linger);
        if (req->r_osd)
                list_add_tail(&req->r_linger_osd,
@@ -1196,6 +1206,7 @@ static void __unregister_linger_request(struct ceph_osd_client *osdc,
                if (list_empty(&req->r_osd_item))
                        req->r_osd = NULL;
        }
+       ceph_osdc_put_request(req);
 }
 
 void ceph_osdc_unregister_linger_request(struct ceph_osd_client *osdc,
@@ -1203,9 +1214,8 @@ void ceph_osdc_unregister_linger_request(struct ceph_osd_client *osdc,
 {
        mutex_lock(&osdc->request_mutex);
        if (req->r_linger) {
-               __unregister_linger_request(osdc, req);
                req->r_linger = 0;
-               ceph_osdc_put_request(req);
+               __unregister_linger_request(osdc, req);
        }
        mutex_unlock(&osdc->request_mutex);
 }
@@ -1217,11 +1227,6 @@ void ceph_osdc_set_request_linger(struct ceph_osd_client *osdc,
        if (!req->r_linger) {
                dout("set_request_linger %p\n", req);
                req->r_linger = 1;
-               /*
-                * caller is now responsible for calling
-                * unregister_linger_request
-                */
-               ceph_osdc_get_request(req);
        }
 }
 EXPORT_SYMBOL(ceph_osdc_set_request_linger);
@@ -1339,10 +1344,6 @@ static void __send_request(struct ceph_osd_client *osdc,
 
        ceph_msg_get(req->r_request); /* send consumes a ref */
 
-       /* Mark the request unsafe if this is the first timet's being sent. */
-
-       if (!req->r_sent && req->r_unsafe_callback)
-               req->r_unsafe_callback(req, true);
        req->r_sent = req->r_osd->o_incarnation;
 
        ceph_con_send(&req->r_osd->o_con, req->r_request);
@@ -1433,8 +1434,6 @@ static void handle_osds_timeout(struct work_struct *work)
 
 static void complete_request(struct ceph_osd_request *req)
 {
-       if (req->r_unsafe_callback)
-               req->r_unsafe_callback(req, false);
        complete_all(&req->r_safe_completion);  /* fsync waiter */
 }
 
@@ -1496,14 +1495,14 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
        dout("handle_reply %p tid %llu req %p result %d\n", msg, tid,
             req, result);
 
-       ceph_decode_need(&p, end, 4, bad);
+       ceph_decode_need(&p, end, 4, bad_put);
        numops = ceph_decode_32(&p);
        if (numops > CEPH_OSD_MAX_OP)
                goto bad_put;
        if (numops != req->r_num_ops)
                goto bad_put;
        payload_len = 0;
-       ceph_decode_need(&p, end, numops * sizeof(struct ceph_osd_op), bad);
+       ceph_decode_need(&p, end, numops * sizeof(struct ceph_osd_op), bad_put);
        for (i = 0; i < numops; i++) {
                struct ceph_osd_op *op = p;
                int len;
@@ -1521,11 +1520,13 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
                goto bad_put;
        }
 
-       ceph_decode_need(&p, end, 4 + numops * 4, bad);
+       ceph_decode_need(&p, end, 4 + numops * 4, bad_put);
        retry_attempt = ceph_decode_32(&p);
        for (i = 0; i < numops; i++)
                req->r_reply_op_result[i] = ceph_decode_32(&p);
 
+       already_completed = req->r_got_reply;
+
        if (!req->r_got_reply) {
 
                req->r_result = result;
@@ -1556,19 +1557,23 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
            ((flags & CEPH_OSD_FLAG_WRITE) == 0))
                __unregister_request(osdc, req);
 
-       already_completed = req->r_completed;
-       req->r_completed = 1;
        mutex_unlock(&osdc->request_mutex);
-       if (already_completed)
-               goto done;
 
-       if (req->r_callback)
-               req->r_callback(req, msg);
-       else
-               complete_all(&req->r_completion);
+       if (!already_completed) {
+               if (req->r_unsafe_callback &&
+                   result >= 0 && !(flags & CEPH_OSD_FLAG_ONDISK))
+                       req->r_unsafe_callback(req, true);
+               if (req->r_callback)
+                       req->r_callback(req, msg);
+               else
+                       complete_all(&req->r_completion);
+       }
 
-       if (flags & CEPH_OSD_FLAG_ONDISK)
+       if (flags & CEPH_OSD_FLAG_ONDISK) {
+               if (req->r_unsafe_callback && already_completed)
+                       req->r_unsafe_callback(req, false);
                complete_request(req);
+       }
 
 done:
        dout("req=%p req->r_linger=%d\n", req, req->r_linger);
@@ -1633,8 +1638,10 @@ static void kick_requests(struct ceph_osd_client *osdc, int force_resend)
                        dout("%p tid %llu restart on osd%d\n",
                             req, req->r_tid,
                             req->r_osd ? req->r_osd->o_osd : -1);
+                       ceph_osdc_get_request(req);
                        __unregister_request(osdc, req);
                        __register_linger_request(osdc, req);
+                       ceph_osdc_put_request(req);
                        continue;
                }
 
@@ -2123,13 +2130,14 @@ int ceph_osdc_start_request(struct ceph_osd_client *osdc,
        __register_request(osdc, req);
        req->r_sent = 0;
        req->r_got_reply = 0;
-       req->r_completed = 0;
        rc = __map_request(osdc, req, 0);
        if (rc < 0) {
                if (nofail) {
                        dout("osdc_start_request failed map, "
                                " will retry %lld\n", req->r_tid);
                        rc = 0;
+               } else {
+                       __unregister_request(osdc, req);
                }
                goto out_unlock;
        }
@@ -2456,8 +2464,10 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
        ceph_msg_revoke_incoming(req->r_reply);
 
        if (front > req->r_reply->front.iov_len) {
-               pr_warning("get_reply front %d > preallocated %d\n",
-                          front, (int)req->r_reply->front.iov_len);
+               pr_warning("get_reply front %d > preallocated %d (%u#%llu)\n",
+                          front, (int)req->r_reply->front.iov_len,
+                          (unsigned int)con->peer_name.type,
+                          le64_to_cpu(con->peer_name.num));
                m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front, GFP_NOFS, false);
                if (!m)
                        goto out;