rbd: mark ceph_msgr_wq reentrant

The ceph messenger workqueue is used to activate execution of the
messenger state machine loop con_work().  The work queue used is
allocated with flag WQ_NON_REENTRANT, which only allows a single
work item on the queue to execute at a time.

There is no need for this restriction, because each connection
maintains its own state and manipulation of each connection is
protected by the connection's mutex.

So drop the WQ_NON_REENTRANT flag when allocating this work queue.
This will allow different connections to be operated on at the same
time.

Signed-off-by: Alex Elder <elder@inktank.com>
rbd: mark ceph_msgr_wq unbounded

Andreas Bluemle observed an undesirable behavior in the ceph
messenger when using networking hardware that binds received data
handling to a single CPU.  The messenger uses a workqueue to process
message activity, and in this case all work was being queued on the
same per-cpu workqueue.  Consequently that CPU becomes a bottleneck,
forced to process all incoming messages.

Marking the workqueue UNBOUND allows any available CPU to process a
work item on the queue.  The concurrency of the workqueue will be
limited by the number of active connections.

Reported-by: Andreas Bluemle <andreas.bluemle@itxperts.de>
Signed-off-by: Alex Elder <elder@inktank.com>
rbd: reset BACKOFF if unable to re-queue

If ceph_fault() is unable to queue work after a delay, it sets the 
BACKOFF connection flag so con_work() will attempt to do so.

In con_work(), when BACKOFF is set, if queue_delayed_work() doesn't
result in newly-queued work, it simply ignores this condition and
proceeds as if no backoff delay were desired.  There are two
problems with this--one of which is a bug.

The first problem is simply that the intended behavior is to back
off, and if we aren't able queue the work item to run after a delay
we're not doing that.

The only reason queue_delayed_work() won't queue work is if the
provided work item is already queued.  In the messenger, this
means that con_work() is already scheduled to be run again.  So
if we simply set the BACKOFF flag again when this occurs, we know
the next con_work() call will again attempt to hold off activity
on the connection until after the delay.

The second problem--the bug--is a leak of a reference count.  If
queue_delayed_work() returns 0 in con_work(), con->ops->put() drops
the connection reference held on entry to con_work().  However,
processing is (was) allowed to continue, and at the end of the
function a second con->ops->put() is called.

This patch fixes both problems.

Signed-off-by: Alex Elder <elder@inktank.com>
rbd: let con_work() handle backoff

Both ceph_fault() and con_work() include handling for imposing a
delay before doing further processing on a faulted connection.
The latter is used only if ceph_fault() is unable to.

Instead, just let con_work() always be responsible for implementing
the delay.  After setting up the delay value, set the BACKOFF flag
on the connection unconditionally and call queue_con() to ensure
con_work() will get called to handle it.

Signed-off-by: Alex Elder <elder@inktank.com>
rbd: define common queue_con_delay()

This patch defines a single function, queue_con_delay() to call
queue_delayed_work() for a connection.  It basically generalizes
what was previously queue_con() by adding the delay argument.
queue_con() is now a simple helper that passes 0 for its delay.
queue_con_delay() returns 0 if it queued work or an errno if it
did not for some reason.

If con_work() finds the BACKOFF flag set for a connection, it now
calls queue_con_delay() to handle arranging to start again after a
delay.


Note about connection reference counts:  con_work() only ever gets
called as a work item function.  At the time that work is scheduled,
a reference to the connection is acquired, and the corresponding
con_work() call is then responsible for dropping that reference
before it returns.

Previously, the backoff handling inside con_work() silently handed
off its reference to delayed work it scheduled.  Now that
queue_con_delay() is used, a new reference is acquired for the
newly-scheduled work, and the original reference is dropped by the
con->ops->put() call at the end of the function.

Signed-off-by: Alex Elder <elder@inktank.com>