libceph: workqueue changes

These patches change the flags passed when creating the ceph messenger work queue. They're done in two distinct steps so their separate effects can more easily be isolated.

[PATCH 1/2] libceph: mark ceph_msgr_wq reentrant [PATCH 2/2] libceph: mark ceph_msgr_wq unbounded

-Alex

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.

---
 net/ceph/messenger.c |    2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

index 159aa8b..28ed904 100644
@@ -190,7 +190,7 @@ int ceph_msgr_init(void)
 	zero_page = ZERO_PAGE(0);
 	page_cache_get(zero_page);
 
-	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_NON_REENTRANT, 0);
+	ceph_msgr_wq = alloc_workqueue("ceph-msgr", 0, 0);
 	if (ceph_msgr_wq)
 		return 0;
 

1.7.9.5

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.

---
 net/ceph/messenger.c |    2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

index 28ed904..f9f65fe 100644
@@ -190,7 +190,7 @@ int ceph_msgr_init(void)
 	zero_page = ZERO_PAGE(0);
 	page_cache_get(zero_page);
 
-	ceph_msgr_wq = alloc_workqueue("ceph-msgr", 0, 0);
+	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_UNBOUND, 0);
 	if (ceph_msgr_wq)
 		return 0;
 

1.7.9.5