Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client

* 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client:
  ceph: no .snap inside of snapped namespace
  libceph: fix msgr standby handling
  libceph: fix msgr keepalive flag
  libceph: fix msgr backoff
  libceph: retry after authorization failure
  libceph: fix handling of short returns from get_user_pages
  ceph: do not clear I_COMPLETE from d_release
  ceph: do not set I_COMPLETE
  Revert "ceph: keep reference to parent inode on ceph_dentry"

2 files changed, 67 insertions, 22 deletions

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 35b36b86d762..05f357828a2f 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -336,7 +336,6 @@ static void reset_connection(struct ceph_connection *con)
                 ceph_msg_put(con->out_msg);
                 con->out_msg = NULL;
         }
-        con->out_keepalive_pending = false;
         con->in_seq = 0;
         con->in_seq_acked = 0;
 }
@@ -1248,8 +1247,6 @@ static int process_connect(struct ceph_connection *con)
                      con->auth_retry);
                 if (con->auth_retry == 2) {
                         con->error_msg = "connect authorization failure";
-                        reset_connection(con);
-                        set_bit(CLOSED, &con->state);
                         return -1;
                 }
                 con->auth_retry = 1;
@@ -1715,14 +1712,6 @@ more:
 
         /* open the socket first? */
         if (con->sock == NULL) {
-                /*
-                 * if we were STANDBY and are reconnecting _this_
-                 * connection, bump connect_seq now.  Always bump
-                 * global_seq.
-                 */
-                if (test_and_clear_bit(STANDBY, &con->state))
-                        con->connect_seq++;
-
                 prepare_write_banner(msgr, con);
                 prepare_write_connect(msgr, con, 1);
                 prepare_read_banner(con);
@@ -1951,7 +1940,24 @@ static void con_work(struct work_struct *work)
                                                    work.work);
 
         mutex_lock(&con->mutex);
+        if (test_and_clear_bit(BACKOFF, &con->state)) {
+                dout("con_work %p backing off\n", con);
+                if (queue_delayed_work(ceph_msgr_wq, &con->work,
+                                       round_jiffies_relative(con->delay))) {
+                        dout("con_work %p backoff %lu\n", con, con->delay);
+                        mutex_unlock(&con->mutex);
+                        return;
+                } else {
+                        con->ops->put(con);
+                        dout("con_work %p FAILED to back off %lu\n", con,
+                             con->delay);
+                }
+        }
 
+        if (test_bit(STANDBY, &con->state)) {
+                dout("con_work %p STANDBY\n", con);
+                goto done;
+        }
         if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
                 dout("con_work CLOSED\n");
                 con_close_socket(con);
@@ -2008,10 +2014,12 @@ static void ceph_fault(struct ceph_connection *con)
         /* Requeue anything that hasn't been acked */
         list_splice_init(&con->out_sent, &con->out_queue);
 
-        /* If there are no messages in the queue, place the connection
-         * in a STANDBY state (i.e., don't try to reconnect just yet). */
-        if (list_empty(&con->out_queue) && !con->out_keepalive_pending) {
-                dout("fault setting STANDBY\n");
+        /* If there are no messages queued or keepalive pending, place
+         * the connection in a STANDBY state */
+        if (list_empty(&con->out_queue) &&
+            !test_bit(KEEPALIVE_PENDING, &con->state)) {
+                dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
+                clear_bit(WRITE_PENDING, &con->state);
                 set_bit(STANDBY, &con->state);
         } else {
                 /* retry after a delay. */
@@ -2019,11 +2027,24 @@ static void ceph_fault(struct ceph_connection *con)
                         con->delay = BASE_DELAY_INTERVAL;
                 else if (con->delay < MAX_DELAY_INTERVAL)
                         con->delay *= 2;
-                dout("fault queueing %p delay %lu\n", con, con->delay);
                 con->ops->get(con);
                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
-                                       round_jiffies_relative(con->delay)) == 0)
+                                       round_jiffies_relative(con->delay))) {
+                        dout("fault queued %p delay %lu\n", con, con->delay);
+                } else {
                         con->ops->put(con);
+                        dout("fault failed to queue %p delay %lu, backoff\n",
+                             con, con->delay);
+                        /*
+                         * In many cases we see a socket state change
+                         * while con_work is running and end up
+                         * queuing (non-delayed) work, such that we
+                         * can't backoff with a delay.  Set a flag so
+                         * that when con_work restarts we schedule the
+                         * delay then.
+                         */
+                        set_bit(BACKOFF, &con->state);
+                }
         }
 
 out_unlock:
@@ -2094,6 +2115,19 @@ void ceph_messenger_destroy(struct ceph_messenger *msgr)
 }
 EXPORT_SYMBOL(ceph_messenger_destroy);
 
+static void clear_standby(struct ceph_connection *con)
+{
+        /* come back from STANDBY? */
+        if (test_and_clear_bit(STANDBY, &con->state)) {
+                mutex_lock(&con->mutex);
+                dout("clear_standby %p and ++connect_seq\n", con);
+                con->connect_seq++;
+                WARN_ON(test_bit(WRITE_PENDING, &con->state));
+                WARN_ON(test_bit(KEEPALIVE_PENDING, &con->state));
+                mutex_unlock(&con->mutex);
+        }
+}
+
 /*
  * Queue up an outgoing message on the given connection.
  */
@@ -2126,6 +2160,7 @@ void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
 
         /* if there wasn't anything waiting to send before, queue
          * new work */
+        clear_standby(con);
         if (test_and_set_bit(WRITE_PENDING, &con->state) == 0)
                 queue_con(con);
 }
@@ -2191,6 +2226,8 @@ void ceph_con_revoke_message(struct ceph_connection *con, struct ceph_msg *msg)
  */
 void ceph_con_keepalive(struct ceph_connection *con)
 {
+        dout("con_keepalive %p\n", con);
+        clear_standby(con);
         if (test_and_set_bit(KEEPALIVE_PENDING, &con->state) == 0 &&
             test_and_set_bit(WRITE_PENDING, &con->state) == 0)
                 queue_con(con);
diff --git a/net/ceph/pagevec.c b/net/ceph/pagevec.c
index 1a040e64c69f..cd9c21df87d1 100644
--- a/net/ceph/pagevec.c
+++ b/net/ceph/pagevec.c
@@ -16,22 +16,30 @@ struct page **ceph_get_direct_page_vector(const char __user *data,
                                           int num_pages, bool write_page)
 {
         struct page **pages;
-        int rc;
+        int got = 0;
+        int rc = 0;
 
         pages = kmalloc(sizeof(*pages) * num_pages, GFP_NOFS);
         if (!pages)
                 return ERR_PTR(-ENOMEM);
 
         down_read(&current->mm->mmap_sem);
-        rc = get_user_pages(current, current->mm, (unsigned long)data,
-                            num_pages, write_page, 0, pages, NULL);
+        while (got < num_pages) {
+                rc = get_user_pages(current, current->mm,
+                    (unsigned long)data + ((unsigned long)got * PAGE_SIZE),
+                    num_pages - got, write_page, 0, pages + got, NULL);
+                if (rc < 0)
+                        break;
+                BUG_ON(rc == 0);
+                got += rc;
+        }
         up_read(&current->mm->mmap_sem);
-        if (rc < num_pages)
+        if (rc < 0)
                 goto fail;
         return pages;
 
 fail:
-        ceph_put_page_vector(pages, rc > 0 ? rc : 0, false);
+        ceph_put_page_vector(pages, got, false);
         return ERR_PTR(rc);
 }
 EXPORT_SYMBOL(ceph_get_direct_page_vector);