6 files changed, 54 insertions, 162 deletions

diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 5d5ab67f516d..ec565508e904 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -239,6 +239,8 @@ enum {
         Opt_nocrc,
         Opt_cephx_require_signatures,
         Opt_nocephx_require_signatures,
+        Opt_tcp_nodelay,
+        Opt_notcp_nodelay,
 };
 
 static match_table_t opt_tokens = {
@@ -259,6 +261,8 @@ static match_table_t opt_tokens = {
         {Opt_nocrc, "nocrc"},
         {Opt_cephx_require_signatures, "cephx_require_signatures"},
         {Opt_nocephx_require_signatures, "nocephx_require_signatures"},
+        {Opt_tcp_nodelay, "tcp_nodelay"},
+        {Opt_notcp_nodelay, "notcp_nodelay"},
         {-1, NULL}
 };
 
@@ -457,6 +461,7 @@ ceph_parse_options(char *options, const char *dev_name,
                 case Opt_nocrc:
                         opt->flags |= CEPH_OPT_NOCRC;
                         break;
+
                 case Opt_cephx_require_signatures:
                         opt->flags &= ~CEPH_OPT_NOMSGAUTH;
                         break;
@@ -464,6 +469,13 @@ ceph_parse_options(char *options, const char *dev_name,
                         opt->flags |= CEPH_OPT_NOMSGAUTH;
                         break;
 
+                case Opt_tcp_nodelay:
+                        opt->flags |= CEPH_OPT_TCP_NODELAY;
+                        break;
+                case Opt_notcp_nodelay:
+                        opt->flags &= ~CEPH_OPT_TCP_NODELAY;
+                        break;
+
                 default:
                         BUG_ON(token);
                 }
@@ -518,10 +530,12 @@ struct ceph_client *ceph_create_client(struct ceph_options *opt, void *private,
         /* msgr */
         if (ceph_test_opt(client, MYIP))
                 myaddr = &client->options->my_addr;
+
         ceph_messenger_init(&client->msgr, myaddr,
                 client->supported_features,
                 client->required_features,
-                ceph_test_opt(client, NOCRC));
+                ceph_test_opt(client, NOCRC),
+                ceph_test_opt(client, TCP_NODELAY));
 
         /* subsystems */
         err = ceph_monc_init(&client->monc, client);
diff --git a/net/ceph/ceph_strings.c b/net/ceph/ceph_strings.c
index 30560202f57b..139a9cb19b0c 100644
--- a/net/ceph/ceph_strings.c
+++ b/net/ceph/ceph_strings.c
@@ -42,17 +42,3 @@ const char *ceph_osd_state_name(int s)
                 return "???";
         }
 }
-
-const char *ceph_pool_op_name(int op)
-{
-        switch (op) {
-        case POOL_OP_CREATE: return "create";
-        case POOL_OP_DELETE: return "delete";
-        case POOL_OP_AUID_CHANGE: return "auid change";
-        case POOL_OP_CREATE_SNAP: return "create snap";
-        case POOL_OP_DELETE_SNAP: return "delete snap";
-        case POOL_OP_CREATE_UNMANAGED_SNAP: return "create unmanaged snap";
-        case POOL_OP_DELETE_UNMANAGED_SNAP: return "delete unmanaged snap";
-        }
-        return "???";
-}
diff --git a/net/ceph/debugfs.c b/net/ceph/debugfs.c
index d2d525529f87..14d9995097cc 100644
--- a/net/ceph/debugfs.c
+++ b/net/ceph/debugfs.c
@@ -127,8 +127,6 @@ static int monc_show(struct seq_file *s, void *p)
                 op = le16_to_cpu(req->request->hdr.type);
                 if (op == CEPH_MSG_STATFS)
                         seq_printf(s, "%llu statfs\n", req->tid);
-                else if (op == CEPH_MSG_POOLOP)
-                        seq_printf(s, "%llu poolop\n", req->tid);
                 else if (op == CEPH_MSG_MON_GET_VERSION)
                         seq_printf(s, "%llu mon_get_version", req->tid);
                 else
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 33a2f201e460..6b3f54ed65ba 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -510,6 +510,16 @@ static int ceph_tcp_connect(struct ceph_connection *con)
                 return ret;
         }
 
+        if (con->msgr->tcp_nodelay) {
+                int optval = 1;
+
+                ret = kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
+                                        (char *)&optval, sizeof(optval));
+                if (ret)
+                        pr_err("kernel_setsockopt(TCP_NODELAY) failed: %d",
+                               ret);
+        }
+
         sk_set_memalloc(sock->sk);
 
         con->sock = sock;
@@ -2922,7 +2932,8 @@ void ceph_messenger_init(struct ceph_messenger *msgr,
                         struct ceph_entity_addr *myaddr,
                         u64 supported_features,
                         u64 required_features,
-                        bool nocrc)
+                        bool nocrc,
+                        bool tcp_nodelay)
 {
         msgr->supported_features = supported_features;
         msgr->required_features = required_features;
@@ -2937,6 +2948,7 @@ void ceph_messenger_init(struct ceph_messenger *msgr,
         get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
         encode_my_addr(msgr);
         msgr->nocrc = nocrc;
+        msgr->tcp_nodelay = tcp_nodelay;
 
         atomic_set(&msgr->stopping, 0);
 
diff --git a/net/ceph/mon_client.c b/net/ceph/mon_client.c
index f2148e22b148..2b3cf05e87b0 100644
--- a/net/ceph/mon_client.c
+++ b/net/ceph/mon_client.c
@@ -410,7 +410,7 @@ out_unlocked:
 }
 
 /*
- * generic requests (e.g., statfs, poolop)
+ * generic requests (currently statfs, mon_get_version)
  */
 static struct ceph_mon_generic_request *__lookup_generic_req(
         struct ceph_mon_client *monc, u64 tid)
@@ -569,7 +569,7 @@ static void handle_statfs_reply(struct ceph_mon_client *monc,
         return;
 
 bad:
-        pr_err("corrupt generic reply, tid %llu\n", tid);
+        pr_err("corrupt statfs reply, tid %llu\n", tid);
         ceph_msg_dump(msg);
 }
 
@@ -588,7 +588,6 @@ int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
 
         kref_init(&req->kref);
         req->buf = buf;
-        req->buf_len = sizeof(*buf);
         init_completion(&req->completion);
 
         err = -ENOMEM;
@@ -611,7 +610,7 @@ int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
         err = do_generic_request(monc, req);
 
 out:
-        kref_put(&req->kref, release_generic_request);
+        put_generic_request(req);
         return err;
 }
 EXPORT_SYMBOL(ceph_monc_do_statfs);
@@ -647,7 +646,7 @@ static void handle_get_version_reply(struct ceph_mon_client *monc,
 
         return;
 bad:
-        pr_err("corrupt mon_get_version reply\n");
+        pr_err("corrupt mon_get_version reply, tid %llu\n", tid);
         ceph_msg_dump(msg);
 }
 
@@ -670,7 +669,6 @@ int ceph_monc_do_get_version(struct ceph_mon_client *monc, const char *what,
 
         kref_init(&req->kref);
         req->buf = newest;
-        req->buf_len = sizeof(*newest);
         init_completion(&req->completion);
 
         req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
@@ -701,134 +699,12 @@ int ceph_monc_do_get_version(struct ceph_mon_client *monc, const char *what,
 
         mutex_unlock(&monc->mutex);
 out:
-        kref_put(&req->kref, release_generic_request);
+        put_generic_request(req);
         return err;
 }
 EXPORT_SYMBOL(ceph_monc_do_get_version);
 
 /*
- * pool ops
- */
-static int get_poolop_reply_buf(const char *src, size_t src_len,
-                                char *dst, size_t dst_len)
-{
-        u32 buf_len;
-
-        if (src_len != sizeof(u32) + dst_len)
-                return -EINVAL;
-
-        buf_len = le32_to_cpu(*(__le32 *)src);
-        if (buf_len != dst_len)
-                return -EINVAL;
-
-        memcpy(dst, src + sizeof(u32), dst_len);
-        return 0;
-}
-
-static void handle_poolop_reply(struct ceph_mon_client *monc,
-                                struct ceph_msg *msg)
-{
-        struct ceph_mon_generic_request *req;
-        struct ceph_mon_poolop_reply *reply = msg->front.iov_base;
-        u64 tid = le64_to_cpu(msg->hdr.tid);
-
-        if (msg->front.iov_len < sizeof(*reply))
-                goto bad;
-        dout("handle_poolop_reply %p tid %llu\n", msg, tid);
-
-        mutex_lock(&monc->mutex);
-        req = __lookup_generic_req(monc, tid);
-        if (req) {
-                if (req->buf_len &&
-                    get_poolop_reply_buf(msg->front.iov_base + sizeof(*reply),
-                                     msg->front.iov_len - sizeof(*reply),
-                                     req->buf, req->buf_len) < 0) {
-                        mutex_unlock(&monc->mutex);
-                        goto bad;
-                }
-                req->result = le32_to_cpu(reply->reply_code);
-                get_generic_request(req);
-        }
-        mutex_unlock(&monc->mutex);
-        if (req) {
-                complete(&req->completion);
-                put_generic_request(req);
-        }
-        return;
-
-bad:
-        pr_err("corrupt generic reply, tid %llu\n", tid);
-        ceph_msg_dump(msg);
-}
-
-/*
- * Do a synchronous pool op.
- */
-static int do_poolop(struct ceph_mon_client *monc, u32 op,
-                        u32 pool, u64 snapid,
-                        char *buf, int len)
-{
-        struct ceph_mon_generic_request *req;
-        struct ceph_mon_poolop *h;
-        int err;
-
-        req = kzalloc(sizeof(*req), GFP_NOFS);
-        if (!req)
-                return -ENOMEM;
-
-        kref_init(&req->kref);
-        req->buf = buf;
-        req->buf_len = len;
-        init_completion(&req->completion);
-
-        err = -ENOMEM;
-        req->request = ceph_msg_new(CEPH_MSG_POOLOP, sizeof(*h), GFP_NOFS,
-                                    true);
-        if (!req->request)
-                goto out;
-        req->reply = ceph_msg_new(CEPH_MSG_POOLOP_REPLY, 1024, GFP_NOFS,
-                                  true);
-        if (!req->reply)
-                goto out;
-
-        /* fill out request */
-        req->request->hdr.version = cpu_to_le16(2);
-        h = req->request->front.iov_base;
-        h->monhdr.have_version = 0;
-        h->monhdr.session_mon = cpu_to_le16(-1);
-        h->monhdr.session_mon_tid = 0;
-        h->fsid = monc->monmap->fsid;
-        h->pool = cpu_to_le32(pool);
-        h->op = cpu_to_le32(op);
-        h->auid = 0;
-        h->snapid = cpu_to_le64(snapid);
-        h->name_len = 0;
-
-        err = do_generic_request(monc, req);
-
-out:
-        kref_put(&req->kref, release_generic_request);
-        return err;
-}
-
-int ceph_monc_create_snapid(struct ceph_mon_client *monc,
-                            u32 pool, u64 *snapid)
-{
-        return do_poolop(monc,  POOL_OP_CREATE_UNMANAGED_SNAP,
-                                   pool, 0, (char *)snapid, sizeof(*snapid));
-
-}
-EXPORT_SYMBOL(ceph_monc_create_snapid);
-
-int ceph_monc_delete_snapid(struct ceph_mon_client *monc,
-                            u32 pool, u64 snapid)
-{
-        return do_poolop(monc,  POOL_OP_CREATE_UNMANAGED_SNAP,
-                                   pool, snapid, NULL, 0);
-
-}
-
-/*
  * Resend pending generic requests.
  */
 static void __resend_generic_request(struct ceph_mon_client *monc)
@@ -1112,10 +988,6 @@ static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
                 handle_get_version_reply(monc, msg);
                 break;
 
-        case CEPH_MSG_POOLOP_REPLY:
-                handle_poolop_reply(monc, msg);
-                break;
-
         case CEPH_MSG_MON_MAP:
                 ceph_monc_handle_map(monc, msg);
                 break;
@@ -1154,7 +1026,6 @@ static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
         case CEPH_MSG_MON_SUBSCRIBE_ACK:
                 m = ceph_msg_get(monc->m_subscribe_ack);
                 break;
-        case CEPH_MSG_POOLOP_REPLY:
         case CEPH_MSG_STATFS_REPLY:
                 return get_generic_reply(con, hdr, skip);
         case CEPH_MSG_AUTH_REPLY:
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 53299c7b0ca4..41a4abc7e98e 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -1035,10 +1035,11 @@ static void put_osd(struct ceph_osd *osd)
 {
         dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
              atomic_read(&osd->o_ref) - 1);
-        if (atomic_dec_and_test(&osd->o_ref) && osd->o_auth.authorizer) {
+        if (atomic_dec_and_test(&osd->o_ref)) {
                 struct ceph_auth_client *ac = osd->o_osdc->client->monc.auth;
 
-                ceph_auth_destroy_authorizer(ac, osd->o_auth.authorizer);
+                if (osd->o_auth.authorizer)
+                        ceph_auth_destroy_authorizer(ac, osd->o_auth.authorizer);
                 kfree(osd);
         }
 }
@@ -1048,14 +1049,24 @@ static void put_osd(struct ceph_osd *osd)
  */
 static void __remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
 {
-        dout("__remove_osd %p\n", osd);
+        dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
         WARN_ON(!list_empty(&osd->o_requests));
         WARN_ON(!list_empty(&osd->o_linger_requests));
 
-        rb_erase(&osd->o_node, &osdc->osds);
         list_del_init(&osd->o_osd_lru);
-        ceph_con_close(&osd->o_con);
-        put_osd(osd);
+        rb_erase(&osd->o_node, &osdc->osds);
+        RB_CLEAR_NODE(&osd->o_node);
+}
+
+static void remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
+{
+        dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
+
+        if (!RB_EMPTY_NODE(&osd->o_node)) {
+                ceph_con_close(&osd->o_con);
+                __remove_osd(osdc, osd);
+                put_osd(osd);
+        }
 }
 
 static void remove_all_osds(struct ceph_osd_client *osdc)
@@ -1065,7 +1076,7 @@ static void remove_all_osds(struct ceph_osd_client *osdc)
         while (!RB_EMPTY_ROOT(&osdc->osds)) {
                 struct ceph_osd *osd = rb_entry(rb_first(&osdc->osds),
                                                 struct ceph_osd, o_node);
-                __remove_osd(osdc, osd);
+                remove_osd(osdc, osd);
         }
         mutex_unlock(&osdc->request_mutex);
 }
@@ -1106,7 +1117,7 @@ static void remove_old_osds(struct ceph_osd_client *osdc)
         list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
                 if (time_before(jiffies, osd->lru_ttl))
                         break;
-                __remove_osd(osdc, osd);
+                remove_osd(osdc, osd);
         }
         mutex_unlock(&osdc->request_mutex);
 }
@@ -1121,8 +1132,7 @@ static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
         dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
         if (list_empty(&osd->o_requests) &&
             list_empty(&osd->o_linger_requests)) {
-                __remove_osd(osdc, osd);
-
+                remove_osd(osdc, osd);
                 return -ENODEV;
         }
 
@@ -1926,6 +1936,7 @@ static void reset_changed_osds(struct ceph_osd_client *osdc)
 {
         struct rb_node *p, *n;
 
+        dout("%s %p\n", __func__, osdc);
         for (p = rb_first(&osdc->osds); p; p = n) {
                 struct ceph_osd *osd = rb_entry(p, struct ceph_osd, o_node);