LU-2263 ptlrpc: decrease ptlrpcd threads number

By default Lustre creates ptlrpcd for each hyper-thread which is
too many for fat-core machine, this patch changed the default
value to CPU core number, and set the upper limit to 32.

Another change in this patch is, ptlrpc_abort_inflight() may release
imp_lock and reschedule when it's necessary, to avoid one thread
holding imp_lock for too long.

This patch didn't apply CPU partition APIs to ptlrpcd, that will
need more efforts.

Signed-off-by: Liang Zhen <liang@whamcloud.com>
Change-Id: I9c395d086258a66f8853dc432b11d49d0db03659
Revision-Id: 2

lustre/include/lustre_net.h

@@ -2274,6 +2274,8 @@ typedef enum {
         PDL_POLICY_PREFERRED    = 4,
 } pdl_policy_t;
 
+#define PTLRPCD_NTHRS_MAX	32
+
 /* ptlrpc/ptlrpcd.c */
 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force);
 void ptlrpcd_wake(struct ptlrpc_request *req);

lustre/ptlrpc/client.c

@@ -2809,18 +2809,32 @@ int ptlrpc_replay_req(struct ptlrpc_request *req)
 }
 EXPORT_SYMBOL(ptlrpc_replay_req);
 
+static void
+ptlrpc_abort_cond_resched(struct obd_import *imp)
+{
+	/* called with hold of imp_lock */
+	if (!cfs_need_resched())
+		return;
+
+	cfs_spin_unlock(&imp->imp_lock);
+
+	CDEBUG(D_RPCTRACE, "reschedule abort_inflight for %s:%s",
+	       imp->imp_obd->obd_uuid.uuid,
+	       libcfs_nid2str(imp->imp_connection->c_peer.nid));
+	cfs_schedule();
+
+	cfs_spin_lock(&imp->imp_lock);
+}
+
 /**
  * Aborts all in-flight request on import \a imp sending and delayed lists
  */
 void ptlrpc_abort_inflight(struct obd_import *imp)
 {
-        cfs_list_t *tmp, *n;
-        ENTRY;
+	cfs_list_t	*tmp;
+	cfs_list_t	*n;
+	ENTRY;
 
-        /* Make sure that no new requests get processed for this import.
-         * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
-         * this flag and then putting requests on sending_list or delayed_list.
-         */
         cfs_spin_lock(&imp->imp_lock);
 
         /* XXX locking?  Maybe we should remove each request with the list
@@ -2831,40 +2845,50 @@ void ptlrpc_abort_inflight(struct obd_import *imp)
                 struct ptlrpc_request *req =
                         cfs_list_entry(tmp, struct ptlrpc_request, rq_list);
 
-                DEBUG_REQ(D_RPCTRACE, req, "inflight");
+		if (req->rq_import_generation >= imp->imp_generation)
+			continue;
 
-                cfs_spin_lock (&req->rq_lock);
-                if (req->rq_import_generation < imp->imp_generation) {
-                        req->rq_err = 1;
-			req->rq_status = -EIO;
-                        ptlrpc_client_wake_req(req);
-                }
-                cfs_spin_unlock (&req->rq_lock);
-        }
+		DEBUG_REQ(D_RPCTRACE, req, "abort inflight req");
 
-        cfs_list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request, rq_list);
+		cfs_spin_lock(&req->rq_lock);
 
-                DEBUG_REQ(D_RPCTRACE, req, "aborting waiting req");
+		req->rq_err = 1;
+		req->rq_status = -EIO;
+		ptlrpc_client_wake_req(req);
 
-                cfs_spin_lock (&req->rq_lock);
-                if (req->rq_import_generation < imp->imp_generation) {
-                        req->rq_err = 1;
-			req->rq_status = -EIO;
-                        ptlrpc_client_wake_req(req);
-                }
-                cfs_spin_unlock (&req->rq_lock);
-        }
+		cfs_spin_unlock(&req->rq_lock);
+	}
 
-        /* Last chance to free reqs left on the replay list, but we
-         * will still leak reqs that haven't committed.  */
-        if (imp->imp_replayable)
-                ptlrpc_free_committed(imp);
+	ptlrpc_abort_cond_resched(imp);
 
-        cfs_spin_unlock(&imp->imp_lock);
+	cfs_list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
+		struct ptlrpc_request *req =
+			cfs_list_entry(tmp, struct ptlrpc_request, rq_list);
 
-        EXIT;
+		if (req->rq_import_generation >= imp->imp_generation)
+			continue;
+
+		DEBUG_REQ(D_RPCTRACE, req, "aborting waiting req");
+
+		cfs_spin_lock(&req->rq_lock);
+
+		req->rq_err = 1;
+		req->rq_status = -EIO;
+		ptlrpc_client_wake_req(req);
+
+		cfs_spin_unlock(&req->rq_lock);
+	}
+
+	/* Last chance to free reqs left on the replay list, but we
+	 * will still leak reqs that haven't committed.  */
+	if (imp->imp_replayable) {
+		ptlrpc_abort_cond_resched(imp);
+		ptlrpc_free_committed(imp);
+	}
+
+	cfs_spin_unlock(&imp->imp_lock);
+
+	EXIT;
 }
 EXPORT_SYMBOL(ptlrpc_abort_inflight);
 

lustre/ptlrpc/import.c

@@ -202,6 +202,11 @@ static void ptlrpc_deactivate_and_unlock_import(struct obd_import *imp)
         LASSERT_SPIN_LOCKED(&imp->imp_lock);
 
         CDEBUG(D_HA, "setting import %s INVALID\n", obd2cli_tgt(imp->imp_obd));
+	/*
+	 * Make sure that no new requests get processed for this import.
+	 * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
+	 * this flag and then putting requests on sending_list or delayed_list.
+	 */
         imp->imp_invalid = 1;
         imp->imp_generation++;
         cfs_spin_unlock(&imp->imp_lock);

lustre/ptlrpc/ptlrpcd.c

@@ -807,12 +807,19 @@ static void ptlrpcd_fini(void)
 
 static int ptlrpcd_init(void)
 {
-        int nthreads = cfs_num_online_cpus();
-        char name[16];
-        int size, i = -1, j, rc = 0;
-        ENTRY;
+	char	name[16];
+	int	rc = 0;
+	int	i = -1;
+	int	j;
+	int	nthreads;
+	int	size;
+	ENTRY;
 
 #ifdef __KERNEL__
+	nthreads = cfs_cpt_weight(cfs_cpt_table, CFS_CPT_ANY) /
+		   cfs_cpu_ht_nsiblings(0);
+	nthreads = min_t(int, PTLRPCD_NTHRS_MAX, nthreads);
+
         if (max_ptlrpcds > 0 && max_ptlrpcds < nthreads)
                 nthreads = max_ptlrpcds;
         if (nthreads < 2)