9 files changed, 153 insertions, 228 deletions

diff --git a/net/ceph/Kconfig b/net/ceph/Kconfig
index e50cc69ae8ca..f8cceb99e732 100644
--- a/net/ceph/Kconfig
+++ b/net/ceph/Kconfig
@@ -3,6 +3,7 @@ config CEPH_LIB
         depends on INET
         select LIBCRC32C
         select CRYPTO_AES
+        select CRYPTO_CBC
         select CRYPTO
         select KEYS
         default n
diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 1675021d8c12..58fbfe134f93 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -293,17 +293,20 @@ static int get_secret(struct ceph_crypto_key *dst, const char *name) {
                 key_err = PTR_ERR(ukey);
                 switch (key_err) {
                 case -ENOKEY:
-                        pr_warning("ceph: Mount failed due to key not found: %s\n", name);
+                        pr_warn("ceph: Mount failed due to key not found: %s\n",
+                                name);
                         break;
                 case -EKEYEXPIRED:
-                        pr_warning("ceph: Mount failed due to expired key: %s\n", name);
+                        pr_warn("ceph: Mount failed due to expired key: %s\n",
+                                name);
                         break;
                 case -EKEYREVOKED:
-                        pr_warning("ceph: Mount failed due to revoked key: %s\n", name);
+                        pr_warn("ceph: Mount failed due to revoked key: %s\n",
+                                name);
                         break;
                 default:
-                        pr_warning("ceph: Mount failed due to unknown key error"
-                               " %d: %s\n", key_err, name);
+                        pr_warn("ceph: Mount failed due to unknown key error %d: %s\n",
+                                key_err, name);
                 }
                 err = -EPERM;
                 goto out;
@@ -433,7 +436,7 @@ ceph_parse_options(char *options, const char *dev_name,
 
                         /* misc */
                 case Opt_osdtimeout:
-                        pr_warning("ignoring deprecated osdtimeout option\n");
+                        pr_warn("ignoring deprecated osdtimeout option\n");
                         break;
                 case Opt_osdkeepalivetimeout:
                         opt->osd_keepalive_timeout = intval;
diff --git a/net/ceph/ceph_strings.c b/net/ceph/ceph_strings.c
index 1348df96fe15..30560202f57b 100644
--- a/net/ceph/ceph_strings.c
+++ b/net/ceph/ceph_strings.c
@@ -19,77 +19,12 @@ const char *ceph_entity_type_name(int type)
 const char *ceph_osd_op_name(int op)
 {
         switch (op) {
-        case CEPH_OSD_OP_READ: return "read";
-        case CEPH_OSD_OP_STAT: return "stat";
-        case CEPH_OSD_OP_MAPEXT: return "mapext";
-        case CEPH_OSD_OP_SPARSE_READ: return "sparse-read";
-        case CEPH_OSD_OP_NOTIFY: return "notify";
-        case CEPH_OSD_OP_NOTIFY_ACK: return "notify-ack";
-        case CEPH_OSD_OP_ASSERT_VER: return "assert-version";
-
-        case CEPH_OSD_OP_MASKTRUNC: return "masktrunc";
-
-        case CEPH_OSD_OP_CREATE: return "create";
-        case CEPH_OSD_OP_WRITE: return "write";
-        case CEPH_OSD_OP_DELETE: return "delete";
-        case CEPH_OSD_OP_TRUNCATE: return "truncate";
-        case CEPH_OSD_OP_ZERO: return "zero";
-        case CEPH_OSD_OP_WRITEFULL: return "writefull";
-        case CEPH_OSD_OP_ROLLBACK: return "rollback";
-
-        case CEPH_OSD_OP_APPEND: return "append";
-        case CEPH_OSD_OP_STARTSYNC: return "startsync";
-        case CEPH_OSD_OP_SETTRUNC: return "settrunc";
-        case CEPH_OSD_OP_TRIMTRUNC: return "trimtrunc";
-
-        case CEPH_OSD_OP_TMAPUP: return "tmapup";
-        case CEPH_OSD_OP_TMAPGET: return "tmapget";
-        case CEPH_OSD_OP_TMAPPUT: return "tmapput";
-        case CEPH_OSD_OP_WATCH: return "watch";
-
-        case CEPH_OSD_OP_CLONERANGE: return "clonerange";
-        case CEPH_OSD_OP_ASSERT_SRC_VERSION: return "assert-src-version";
-        case CEPH_OSD_OP_SRC_CMPXATTR: return "src-cmpxattr";
-
-        case CEPH_OSD_OP_GETXATTR: return "getxattr";
-        case CEPH_OSD_OP_GETXATTRS: return "getxattrs";
-        case CEPH_OSD_OP_SETXATTR: return "setxattr";
-        case CEPH_OSD_OP_SETXATTRS: return "setxattrs";
-        case CEPH_OSD_OP_RESETXATTRS: return "resetxattrs";
-        case CEPH_OSD_OP_RMXATTR: return "rmxattr";
-        case CEPH_OSD_OP_CMPXATTR: return "cmpxattr";
-
-        case CEPH_OSD_OP_PULL: return "pull";
-        case CEPH_OSD_OP_PUSH: return "push";
-        case CEPH_OSD_OP_BALANCEREADS: return "balance-reads";
-        case CEPH_OSD_OP_UNBALANCEREADS: return "unbalance-reads";
-        case CEPH_OSD_OP_SCRUB: return "scrub";
-        case CEPH_OSD_OP_SCRUB_RESERVE: return "scrub-reserve";
-        case CEPH_OSD_OP_SCRUB_UNRESERVE: return "scrub-unreserve";
-        case CEPH_OSD_OP_SCRUB_STOP: return "scrub-stop";
-        case CEPH_OSD_OP_SCRUB_MAP: return "scrub-map";
-
-        case CEPH_OSD_OP_WRLOCK: return "wrlock";
-        case CEPH_OSD_OP_WRUNLOCK: return "wrunlock";
-        case CEPH_OSD_OP_RDLOCK: return "rdlock";
-        case CEPH_OSD_OP_RDUNLOCK: return "rdunlock";
-        case CEPH_OSD_OP_UPLOCK: return "uplock";
-        case CEPH_OSD_OP_DNLOCK: return "dnlock";
-
-        case CEPH_OSD_OP_CALL: return "call";
-
-        case CEPH_OSD_OP_PGLS: return "pgls";
-        case CEPH_OSD_OP_PGLS_FILTER: return "pgls-filter";
-        case CEPH_OSD_OP_OMAPGETKEYS: return "omap-get-keys";
-        case CEPH_OSD_OP_OMAPGETVALS: return "omap-get-vals";
-        case CEPH_OSD_OP_OMAPGETHEADER: return "omap-get-header";
-        case CEPH_OSD_OP_OMAPGETVALSBYKEYS: return "omap-get-vals-by-keys";
-        case CEPH_OSD_OP_OMAPSETVALS: return "omap-set-vals";
-        case CEPH_OSD_OP_OMAPSETHEADER: return "omap-set-header";
-        case CEPH_OSD_OP_OMAPCLEAR: return "omap-clear";
-        case CEPH_OSD_OP_OMAPRMKEYS: return "omap-rm-keys";
+#define GENERATE_CASE(op, opcode, str)  case CEPH_OSD_OP_##op: return (str);
+__CEPH_FORALL_OSD_OPS(GENERATE_CASE)
+#undef GENERATE_CASE
+        default:
+                return "???";
         }
-        return "???";
 }
 
 const char *ceph_osd_state_name(int s)
diff --git a/net/ceph/debugfs.c b/net/ceph/debugfs.c
index d1a62c69a9f4..d2d525529f87 100644
--- a/net/ceph/debugfs.c
+++ b/net/ceph/debugfs.c
@@ -169,7 +169,8 @@ static int osdc_show(struct seq_file *s, void *pp)
 
                 for (i = 0; i < req->r_num_ops; i++) {
                         opcode = req->r_ops[i].op;
-                        seq_printf(s, "\t%s", ceph_osd_op_name(opcode));
+                        seq_printf(s, "%s%s", (i == 0 ? "\t" : ","),
+                                   ceph_osd_op_name(opcode));
                 }
 
                 seq_printf(s, "\n");
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index b2f571dd933d..559c9f619c20 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -292,7 +292,11 @@ int ceph_msgr_init(void)
         if (ceph_msgr_slab_init())
                 return -ENOMEM;
 
-        ceph_msgr_wq = alloc_workqueue("ceph-msgr", 0, 0);
+        /*
+         * The number of active work items is limited by the number of
+         * connections, so leave @max_active at default.
+         */
+        ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_MEM_RECLAIM, 0);
         if (ceph_msgr_wq)
                 return 0;
 
@@ -1937,11 +1941,11 @@ static int process_banner(struct ceph_connection *con)
                    sizeof(con->peer_addr)) != 0 &&
             !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
               con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
-                pr_warning("wrong peer, want %s/%d, got %s/%d\n",
-                           ceph_pr_addr(&con->peer_addr.in_addr),
-                           (int)le32_to_cpu(con->peer_addr.nonce),
-                           ceph_pr_addr(&con->actual_peer_addr.in_addr),
-                           (int)le32_to_cpu(con->actual_peer_addr.nonce));
+                pr_warn("wrong peer, want %s/%d, got %s/%d\n",
+                        ceph_pr_addr(&con->peer_addr.in_addr),
+                        (int)le32_to_cpu(con->peer_addr.nonce),
+                        ceph_pr_addr(&con->actual_peer_addr.in_addr),
+                        (int)le32_to_cpu(con->actual_peer_addr.nonce));
                 con->error_msg = "wrong peer at address";
                 return -1;
         }
@@ -2302,7 +2306,7 @@ static int read_partial_message(struct ceph_connection *con)
 
                 BUG_ON(!con->in_msg ^ skip);
                 if (con->in_msg && data_len > con->in_msg->data_length) {
-                        pr_warning("%s skipping long message (%u > %zd)\n",
+                        pr_warn("%s skipping long message (%u > %zd)\n",
                                 __func__, data_len, con->in_msg->data_length);
                         ceph_msg_put(con->in_msg);
                         con->in_msg = NULL;
@@ -2712,7 +2716,7 @@ static bool con_sock_closed(struct ceph_connection *con)
         CASE(OPEN);
         CASE(STANDBY);
         default:
-                pr_warning("%s con %p unrecognized state %lu\n",
+                pr_warn("%s con %p unrecognized state %lu\n",
                         __func__, con, con->state);
                 con->error_msg = "unrecognized con state";
                 BUG();
@@ -2828,8 +2832,8 @@ static void con_work(struct work_struct *work)
  */
 static void con_fault(struct ceph_connection *con)
 {
-        pr_warning("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
-               ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
+        pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
+                ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
         dout("fault %p state %lu to peer %s\n",
              con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
 
@@ -3071,10 +3075,8 @@ static void ceph_msg_data_destroy(struct ceph_msg_data *data)
                 return;
 
         WARN_ON(!list_empty(&data->links));
-        if (data->type == CEPH_MSG_DATA_PAGELIST) {
+        if (data->type == CEPH_MSG_DATA_PAGELIST)
                 ceph_pagelist_release(data->pagelist);
-                kfree(data->pagelist);
-        }
         kmem_cache_free(ceph_msg_data_cache, data);
 }
 
diff --git a/net/ceph/mon_client.c b/net/ceph/mon_client.c
index 61fcfc304f68..a83062ceeec9 100644
--- a/net/ceph/mon_client.c
+++ b/net/ceph/mon_client.c
@@ -1182,10 +1182,10 @@ static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
                 pr_info("alloc_msg unknown type %d\n", type);
                 *skip = 1;
         } else if (front_len > m->front_alloc_len) {
-                pr_warning("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
-                           front_len, m->front_alloc_len,
-                           (unsigned int)con->peer_name.type,
-                           le64_to_cpu(con->peer_name.num));
+                pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
+                        front_len, m->front_alloc_len,
+                        (unsigned int)con->peer_name.type,
+                        le64_to_cpu(con->peer_name.num));
                 ceph_msg_put(m);
                 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
         }
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 30f6faf3584f..f3fc54eac09d 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -30,8 +30,11 @@ static void __send_queued(struct ceph_osd_client *osdc);
 static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd);
 static void __register_request(struct ceph_osd_client *osdc,
                                struct ceph_osd_request *req);
+static void __unregister_request(struct ceph_osd_client *osdc,
+                                 struct ceph_osd_request *req);
 static void __unregister_linger_request(struct ceph_osd_client *osdc,
                                         struct ceph_osd_request *req);
+static void __enqueue_request(struct ceph_osd_request *req);
 static void __send_request(struct ceph_osd_client *osdc,
                            struct ceph_osd_request *req);
 
@@ -428,68 +431,9 @@ EXPORT_SYMBOL(ceph_osdc_alloc_request);
 static bool osd_req_opcode_valid(u16 opcode)
 {
         switch (opcode) {
-        case CEPH_OSD_OP_READ:
-        case CEPH_OSD_OP_STAT:
-        case CEPH_OSD_OP_MAPEXT:
-        case CEPH_OSD_OP_MASKTRUNC:
-        case CEPH_OSD_OP_SPARSE_READ:
-        case CEPH_OSD_OP_NOTIFY:
-        case CEPH_OSD_OP_NOTIFY_ACK:
-        case CEPH_OSD_OP_ASSERT_VER:
-        case CEPH_OSD_OP_WRITE:
-        case CEPH_OSD_OP_WRITEFULL:
-        case CEPH_OSD_OP_TRUNCATE:
-        case CEPH_OSD_OP_ZERO:
-        case CEPH_OSD_OP_DELETE:
-        case CEPH_OSD_OP_APPEND:
-        case CEPH_OSD_OP_STARTSYNC:
-        case CEPH_OSD_OP_SETTRUNC:
-        case CEPH_OSD_OP_TRIMTRUNC:
-        case CEPH_OSD_OP_TMAPUP:
-        case CEPH_OSD_OP_TMAPPUT:
-        case CEPH_OSD_OP_TMAPGET:
-        case CEPH_OSD_OP_CREATE:
-        case CEPH_OSD_OP_ROLLBACK:
-        case CEPH_OSD_OP_WATCH:
-        case CEPH_OSD_OP_OMAPGETKEYS:
-        case CEPH_OSD_OP_OMAPGETVALS:
-        case CEPH_OSD_OP_OMAPGETHEADER:
-        case CEPH_OSD_OP_OMAPGETVALSBYKEYS:
-        case CEPH_OSD_OP_OMAPSETVALS:
-        case CEPH_OSD_OP_OMAPSETHEADER:
-        case CEPH_OSD_OP_OMAPCLEAR:
-        case CEPH_OSD_OP_OMAPRMKEYS:
-        case CEPH_OSD_OP_OMAP_CMP:
-        case CEPH_OSD_OP_SETALLOCHINT:
-        case CEPH_OSD_OP_CLONERANGE:
-        case CEPH_OSD_OP_ASSERT_SRC_VERSION:
-        case CEPH_OSD_OP_SRC_CMPXATTR:
-        case CEPH_OSD_OP_GETXATTR:
-        case CEPH_OSD_OP_GETXATTRS:
-        case CEPH_OSD_OP_CMPXATTR:
-        case CEPH_OSD_OP_SETXATTR:
-        case CEPH_OSD_OP_SETXATTRS:
-        case CEPH_OSD_OP_RESETXATTRS:
-        case CEPH_OSD_OP_RMXATTR:
-        case CEPH_OSD_OP_PULL:
-        case CEPH_OSD_OP_PUSH:
-        case CEPH_OSD_OP_BALANCEREADS:
-        case CEPH_OSD_OP_UNBALANCEREADS:
-        case CEPH_OSD_OP_SCRUB:
-        case CEPH_OSD_OP_SCRUB_RESERVE:
-        case CEPH_OSD_OP_SCRUB_UNRESERVE:
-        case CEPH_OSD_OP_SCRUB_STOP:
-        case CEPH_OSD_OP_SCRUB_MAP:
-        case CEPH_OSD_OP_WRLOCK:
-        case CEPH_OSD_OP_WRUNLOCK:
-        case CEPH_OSD_OP_RDLOCK:
-        case CEPH_OSD_OP_RDUNLOCK:
-        case CEPH_OSD_OP_UPLOCK:
-        case CEPH_OSD_OP_DNLOCK:
-        case CEPH_OSD_OP_CALL:
-        case CEPH_OSD_OP_PGLS:
-        case CEPH_OSD_OP_PGLS_FILTER:
-                return true;
+#define GENERATE_CASE(op, opcode, str)  case CEPH_OSD_OP_##op: return true;
+__CEPH_FORALL_OSD_OPS(GENERATE_CASE)
+#undef GENERATE_CASE
         default:
                 return false;
         }
@@ -892,6 +836,37 @@ __lookup_request_ge(struct ceph_osd_client *osdc,
         return NULL;
 }
 
+static void __kick_linger_request(struct ceph_osd_request *req)
+{
+        struct ceph_osd_client *osdc = req->r_osdc;
+        struct ceph_osd *osd = req->r_osd;
+
+        /*
+         * Linger requests need to be resent with a new tid to avoid
+         * the dup op detection logic on the OSDs.  Achieve this with
+         * a re-register dance instead of open-coding.
+         */
+        ceph_osdc_get_request(req);
+        if (!list_empty(&req->r_linger_item))
+                __unregister_linger_request(osdc, req);
+        else
+                __unregister_request(osdc, req);
+        __register_request(osdc, req);
+        ceph_osdc_put_request(req);
+
+        /*
+         * Unless request has been registered as both normal and
+         * lingering, __unregister{,_linger}_request clears r_osd.
+         * However, here we need to preserve r_osd to make sure we
+         * requeue on the same OSD.
+         */
+        WARN_ON(req->r_osd || !osd);
+        req->r_osd = osd;
+
+        dout("%s requeueing %p tid %llu\n", __func__, req, req->r_tid);
+        __enqueue_request(req);
+}
+
 /*
  * Resubmit requests pending on the given osd.
  */
@@ -900,12 +875,14 @@ static void __kick_osd_requests(struct ceph_osd_client *osdc,
 {
         struct ceph_osd_request *req, *nreq;
         LIST_HEAD(resend);
+        LIST_HEAD(resend_linger);
         int err;
 
-        dout("__kick_osd_requests osd%d\n", osd->o_osd);
+        dout("%s osd%d\n", __func__, osd->o_osd);
         err = __reset_osd(osdc, osd);
         if (err)
                 return;
+
         /*
          * Build up a list of requests to resend by traversing the
          * osd's list of requests.  Requests for a given object are
@@ -926,33 +903,32 @@ static void __kick_osd_requests(struct ceph_osd_client *osdc,
         list_for_each_entry(req, &osd->o_requests, r_osd_item) {
                 if (!req->r_sent)
                         break;
-                list_move_tail(&req->r_req_lru_item, &resend);
-                dout("requeueing %p tid %llu osd%d\n", req, req->r_tid,
-                     osd->o_osd);
-                if (!req->r_linger)
+
+                if (!req->r_linger) {
+                        dout("%s requeueing %p tid %llu\n", __func__, req,
+                             req->r_tid);
+                        list_move_tail(&req->r_req_lru_item, &resend);
                         req->r_flags |= CEPH_OSD_FLAG_RETRY;
+                } else {
+                        list_move_tail(&req->r_req_lru_item, &resend_linger);
+                }
         }
         list_splice(&resend, &osdc->req_unsent);
 
         /*
-         * Linger requests are re-registered before sending, which
-         * sets up a new tid for each.  We add them to the unsent
-         * list at the end to keep things in tid order.
+         * Both registered and not yet registered linger requests are
+         * enqueued with a new tid on the same OSD.  We add/move them
+         * to req_unsent/o_requests at the end to keep things in tid
+         * order.
          */
         list_for_each_entry_safe(req, nreq, &osd->o_linger_requests,
                                  r_linger_osd_item) {
-                /*
-                 * reregister request prior to unregistering linger so
-                 * that r_osd is preserved.
-                 */
-                BUG_ON(!list_empty(&req->r_req_lru_item));
-                __register_request(osdc, req);
-                list_add_tail(&req->r_req_lru_item, &osdc->req_unsent);
-                list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
-                __unregister_linger_request(osdc, req);
-                dout("requeued lingering %p tid %llu osd%d\n", req, req->r_tid,
-                     osd->o_osd);
+                WARN_ON(!list_empty(&req->r_req_lru_item));
+                __kick_linger_request(req);
         }
+
+        list_for_each_entry_safe(req, nreq, &resend_linger, r_req_lru_item)
+                __kick_linger_request(req);
 }
 
 /*
@@ -1346,6 +1322,22 @@ static int __calc_request_pg(struct ceph_osdmap *osdmap,
                                    &req->r_target_oid, pg_out);
 }
 
+static void __enqueue_request(struct ceph_osd_request *req)
+{
+        struct ceph_osd_client *osdc = req->r_osdc;
+
+        dout("%s %p tid %llu to osd%d\n", __func__, req, req->r_tid,
+             req->r_osd ? req->r_osd->o_osd : -1);
+
+        if (req->r_osd) {
+                __remove_osd_from_lru(req->r_osd);
+                list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
+                list_move_tail(&req->r_req_lru_item, &osdc->req_unsent);
+        } else {
+                list_move_tail(&req->r_req_lru_item, &osdc->req_notarget);
+        }
+}
+
 /*
  * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
  * (as needed), and set the request r_osd appropriately.  If there is
@@ -1423,13 +1415,7 @@ static int __map_request(struct ceph_osd_client *osdc,
                               &osdc->osdmap->osd_addr[o]);
         }
 
-        if (req->r_osd) {
-                __remove_osd_from_lru(req->r_osd);
-                list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
-                list_move_tail(&req->r_req_lru_item, &osdc->req_unsent);
-        } else {
-                list_move_tail(&req->r_req_lru_item, &osdc->req_notarget);
-        }
+        __enqueue_request(req);
         err = 1;   /* osd or pg changed */
 
 out:
@@ -1774,8 +1760,8 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
         }
         bytes = le32_to_cpu(msg->hdr.data_len);
         if (payload_len != bytes) {
-                pr_warning("sum of op payload lens %d != data_len %d",
-                           payload_len, bytes);
+                pr_warn("sum of op payload lens %d != data_len %d\n",
+                        payload_len, bytes);
                 goto bad_put;
         }
 
@@ -2313,24 +2299,19 @@ static void handle_watch_notify(struct ceph_osd_client *osdc,
         if (event) {
                 event_work = kmalloc(sizeof(*event_work), GFP_NOIO);
                 if (!event_work) {
-                        dout("ERROR: could not allocate event_work\n");
-                        goto done_err;
+                        pr_err("couldn't allocate event_work\n");
+                        ceph_osdc_put_event(event);
+                        return;
                 }
                 INIT_WORK(&event_work->work, do_event_work);
                 event_work->event = event;
                 event_work->ver = ver;
                 event_work->notify_id = notify_id;
                 event_work->opcode = opcode;
-                if (!queue_work(osdc->notify_wq, &event_work->work)) {
-                        dout("WARNING: failed to queue notify event work\n");
-                        goto done_err;
-                }
-        }
 
-        return;
+                queue_work(osdc->notify_wq, &event_work->work);
+        }
 
-done_err:
-        ceph_osdc_put_event(event);
         return;
 
 bad:
@@ -2797,10 +2778,10 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
         ceph_msg_revoke_incoming(req->r_reply);
 
         if (front_len > req->r_reply->front_alloc_len) {
-                pr_warning("get_reply front %d > preallocated %d (%u#%llu)\n",
-                           front_len, req->r_reply->front_alloc_len,
-                           (unsigned int)con->peer_name.type,
-                           le64_to_cpu(con->peer_name.num));
+                pr_warn("get_reply front %d > preallocated %d (%u#%llu)\n",
+                        front_len, req->r_reply->front_alloc_len,
+                        (unsigned int)con->peer_name.type,
+                        le64_to_cpu(con->peer_name.num));
                 m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front_len, GFP_NOFS,
                                  false);
                 if (!m)
@@ -2823,8 +2804,7 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
                         if (osd_data->pages &&
                                 unlikely(osd_data->length < data_len)) {
 
-                                pr_warning("tid %lld reply has %d bytes "
-                                        "we had only %llu bytes ready\n",
+                                pr_warn("tid %lld reply has %d bytes we had only %llu bytes ready\n",
                                         tid, data_len, osd_data->length);
                                 *skip = 1;
                                 ceph_msg_put(m);
diff --git a/net/ceph/osdmap.c b/net/ceph/osdmap.c
index c547e46084d3..b8c3fde5b04f 100644
--- a/net/ceph/osdmap.c
+++ b/net/ceph/osdmap.c
@@ -521,11 +521,11 @@ static int decode_pool(void **p, void *end, struct ceph_pg_pool_info *pi)
         ev = ceph_decode_8(p);  /* encoding version */
         cv = ceph_decode_8(p); /* compat version */
         if (ev < 5) {
-                pr_warning("got v %d < 5 cv %d of ceph_pg_pool\n", ev, cv);
+                pr_warn("got v %d < 5 cv %d of ceph_pg_pool\n", ev, cv);
                 return -EINVAL;
         }
         if (cv > 9) {
-                pr_warning("got v %d cv %d > 9 of ceph_pg_pool\n", ev, cv);
+                pr_warn("got v %d cv %d > 9 of ceph_pg_pool\n", ev, cv);
                 return -EINVAL;
         }
         len = ceph_decode_32(p);
@@ -671,26 +671,26 @@ static int osdmap_set_max_osd(struct ceph_osdmap *map, int max)
         int i;
 
         state = krealloc(map->osd_state, max*sizeof(*state), GFP_NOFS);
+        if (!state)
+                return -ENOMEM;
+        map->osd_state = state;
+
         weight = krealloc(map->osd_weight, max*sizeof(*weight), GFP_NOFS);
-        addr = krealloc(map->osd_addr, max*sizeof(*addr), GFP_NOFS);
-        if (!state || !weight || !addr) {
-                kfree(state);
-                kfree(weight);
-                kfree(addr);
+        if (!weight)
+                return -ENOMEM;
+        map->osd_weight = weight;
 
+        addr = krealloc(map->osd_addr, max*sizeof(*addr), GFP_NOFS);
+        if (!addr)
                 return -ENOMEM;
-        }
+        map->osd_addr = addr;
 
         for (i = map->max_osd; i < max; i++) {
-                state[i] = 0;
-                weight[i] = CEPH_OSD_OUT;
-                memset(addr + i, 0, sizeof(*addr));
+                map->osd_state[i] = 0;
+                map->osd_weight[i] = CEPH_OSD_OUT;
+                memset(map->osd_addr + i, 0, sizeof(*map->osd_addr));
         }
 
-        map->osd_state = state;
-        map->osd_weight = weight;
-        map->osd_addr = addr;
-
         if (map->osd_primary_affinity) {
                 u32 *affinity;
 
@@ -698,11 +698,11 @@ static int osdmap_set_max_osd(struct ceph_osdmap *map, int max)
                                     max*sizeof(*affinity), GFP_NOFS);
                 if (!affinity)
                         return -ENOMEM;
+                map->osd_primary_affinity = affinity;
 
                 for (i = map->max_osd; i < max; i++)
-                        affinity[i] = CEPH_OSD_DEFAULT_PRIMARY_AFFINITY;
-
-                map->osd_primary_affinity = affinity;
+                        map->osd_primary_affinity[i] =
+                            CEPH_OSD_DEFAULT_PRIMARY_AFFINITY;
         }
 
         map->max_osd = max;
@@ -729,9 +729,9 @@ static int get_osdmap_client_data_v(void **p, void *end,
 
                 ceph_decode_8_safe(p, end, struct_compat, e_inval);
                 if (struct_compat > OSDMAP_WRAPPER_COMPAT_VER) {
-                        pr_warning("got v %d cv %d > %d of %s ceph_osdmap\n",
-                                   struct_v, struct_compat,
-                                   OSDMAP_WRAPPER_COMPAT_VER, prefix);
+                        pr_warn("got v %d cv %d > %d of %s ceph_osdmap\n",
+                                struct_v, struct_compat,
+                                OSDMAP_WRAPPER_COMPAT_VER, prefix);
                         return -EINVAL;
                 }
                 *p += 4; /* ignore wrapper struct_len */
@@ -739,9 +739,9 @@ static int get_osdmap_client_data_v(void **p, void *end,
                 ceph_decode_8_safe(p, end, struct_v, e_inval);
                 ceph_decode_8_safe(p, end, struct_compat, e_inval);
                 if (struct_compat > OSDMAP_CLIENT_DATA_COMPAT_VER) {
-                        pr_warning("got v %d cv %d > %d of %s ceph_osdmap client data\n",
-                                   struct_v, struct_compat,
-                                   OSDMAP_CLIENT_DATA_COMPAT_VER, prefix);
+                        pr_warn("got v %d cv %d > %d of %s ceph_osdmap client data\n",
+                                struct_v, struct_compat,
+                                OSDMAP_CLIENT_DATA_COMPAT_VER, prefix);
                         return -EINVAL;
                 }
                 *p += 4; /* ignore client data struct_len */
@@ -751,8 +751,8 @@ static int get_osdmap_client_data_v(void **p, void *end,
                 *p -= 1;
                 ceph_decode_16_safe(p, end, version, e_inval);
                 if (version < 6) {
-                        pr_warning("got v %d < 6 of %s ceph_osdmap\n", version,
-                                   prefix);
+                        pr_warn("got v %d < 6 of %s ceph_osdmap\n",
+                                version, prefix);
                         return -EINVAL;
                 }
 
diff --git a/net/ceph/pagelist.c b/net/ceph/pagelist.c
index 92866bebb65f..c7c220a736e5 100644
--- a/net/ceph/pagelist.c
+++ b/net/ceph/pagelist.c
@@ -1,5 +1,6 @@
 #include <linux/module.h>
 #include <linux/gfp.h>
+#include <linux/slab.h>
 #include <linux/pagemap.h>
 #include <linux/highmem.h>
 #include <linux/ceph/pagelist.h>
@@ -13,8 +14,10 @@ static void ceph_pagelist_unmap_tail(struct ceph_pagelist *pl)
         }
 }
 
-int ceph_pagelist_release(struct ceph_pagelist *pl)
+void ceph_pagelist_release(struct ceph_pagelist *pl)
 {
+        if (!atomic_dec_and_test(&pl->refcnt))
+                return;
         ceph_pagelist_unmap_tail(pl);
         while (!list_empty(&pl->head)) {
                 struct page *page = list_first_entry(&pl->head, struct page,
@@ -23,7 +26,7 @@ int ceph_pagelist_release(struct ceph_pagelist *pl)
                 __free_page(page);
         }
         ceph_pagelist_free_reserve(pl);
-        return 0;
+        kfree(pl);
 }
 EXPORT_SYMBOL(ceph_pagelist_release);