- struct ptlrpcd_ctl *pc = arg;
- unsigned long flags;
- ENTRY;
-
- kportal_daemonize(pc->pc_name);
-
- SIGNAL_MASK_LOCK(current, flags);
- sigfillset(¤t->blocked);
- RECALC_SIGPENDING;
- SIGNAL_MASK_UNLOCK(current, flags);
-
- complete(&pc->pc_starting);
-
- /* this mainloop strongly resembles ptlrpc_set_wait except
- * that our set never completes. ptlrpcd_check calls ptlrpc_check_set
- * when there are requests in the set. new requests come in
- * on the set's new_req_list and ptlrpcd_check moves them into
- * the set. */
- while (1) {
- wait_queue_t set_wait;
- struct l_wait_info lwi;
- int timeout;
-
- timeout = ptlrpc_set_next_timeout(pc->pc_set) * HZ;
- lwi = LWI_TIMEOUT(timeout, ptlrpc_expired_set, pc->pc_set);
-
- /* ala the pinger, wait on pc's waitqueue and the set's */
- init_waitqueue_entry(&set_wait, current);
- add_wait_queue(&pc->pc_set->set_waitq, &set_wait);
- l_wait_event(pc->pc_waitq, ptlrpcd_check(pc), &lwi);
- remove_wait_queue(&pc->pc_set->set_waitq, &set_wait);
-
- if (test_bit(LIOD_STOP, &pc->pc_flags))
- break;
- }
- /* wait for inflight requests to drain */
- if (!list_empty(&pc->pc_set->set_requests))
- ptlrpc_set_wait(pc->pc_set);
- complete(&pc->pc_finishing);
- return 0;
+ struct ptlrpcd_ctl *pc = arg;
+ struct ptlrpc_request_set *set;
+ struct lu_context ses = { 0 };
+ struct lu_env env = { .le_ses = &ses };
+ int rc = 0;
+ int exit = 0;
+
+ ENTRY;
+ if (cfs_cpt_bind(cfs_cpt_tab, pc->pc_cpt) != 0)
+ CWARN("Failed to bind %s on CPT %d\n", pc->pc_name, pc->pc_cpt);
+
+ /*
+ * Allocate the request set after the thread has been bound
+ * above. This is safe because no requests will be queued
+ * until all ptlrpcd threads have confirmed that they have
+ * successfully started.
+ */
+ set = ptlrpc_prep_set();
+ if (set == NULL)
+ GOTO(failed, rc = -ENOMEM);
+ spin_lock(&pc->pc_lock);
+ pc->pc_set = set;
+ spin_unlock(&pc->pc_lock);
+
+ /* Both client and server (MDT/OST) may use the environment. */
+ rc = lu_context_init(&env.le_ctx, LCT_MD_THREAD |
+ LCT_DT_THREAD |
+ LCT_CL_THREAD |
+ LCT_REMEMBER |
+ LCT_NOREF);
+ if (rc != 0)
+ GOTO(failed, rc);
+ rc = lu_context_init(env.le_ses, LCT_SESSION |
+ LCT_REMEMBER |
+ LCT_NOREF);
+ if (rc != 0) {
+ lu_context_fini(&env.le_ctx);
+ GOTO(failed, rc);
+ }
+
+ complete(&pc->pc_starting);
+
+ /*
+ * This mainloop strongly resembles ptlrpc_set_wait() except that our
+ * set never completes. ptlrpcd_check() calls ptlrpc_check_set() when
+ * there are requests in the set. New requests come in on the set's
+ * new_req_list and ptlrpcd_check() moves them into the set.
+ */
+ do {
+ DEFINE_WAIT_FUNC(wait, woken_wake_function);
+ time64_t timeout;
+
+ timeout = ptlrpc_set_next_timeout(set);
+
+ lu_context_enter(&env.le_ctx);
+ lu_context_enter(env.le_ses);
+
+ add_wait_queue(&set->set_waitq, &wait);
+ while (!ptlrpcd_check(&env, pc)) {
+ int ret;
+
+ if (timeout == 0)
+ ret = wait_woken(&wait, TASK_IDLE,
+ MAX_SCHEDULE_TIMEOUT);
+ else
+ ret = wait_woken(&wait, TASK_IDLE,
+ cfs_time_seconds(timeout));
+ if (ret != 0)
+ continue;
+ /* Timed out */
+ ptlrpc_expired_set(set);
+ break;
+ }
+ remove_wait_queue(&set->set_waitq, &wait);
+
+ lu_context_exit(&env.le_ctx);
+ lu_context_exit(env.le_ses);
+
+ /*
+ * Abort inflight rpcs for forced stop case.
+ */
+ if (test_bit(LIOD_STOP, &pc->pc_flags)) {
+ if (test_bit(LIOD_FORCE, &pc->pc_flags))
+ ptlrpc_abort_set(set);
+ exit++;
+ }
+
+ /*
+ * Let's make one more loop to make sure that ptlrpcd_check()
+ * copied all raced new rpcs into the set so we can kill them.
+ */
+ } while (exit < 2);
+
+ /*
+ * Wait for inflight requests to drain.
+ */
+ if (!list_empty(&set->set_requests))
+ ptlrpc_set_wait(&env, set);
+ lu_context_fini(&env.le_ctx);
+ lu_context_fini(env.le_ses);
+
+ complete(&pc->pc_finishing);
+
+ return 0;
+
+failed:
+ pc->pc_error = rc;
+ complete(&pc->pc_starting);
+ RETURN(rc);
+}
+
+static void ptlrpcd_ctl_init(struct ptlrpcd_ctl *pc, int index, int cpt)
+{
+ ENTRY;
+
+ pc->pc_index = index;
+ pc->pc_cpt = cpt;
+ init_completion(&pc->pc_starting);
+ init_completion(&pc->pc_finishing);
+ spin_lock_init(&pc->pc_lock);
+
+ if (index < 0) {
+ /* Recovery thread. */
+ snprintf(pc->pc_name, sizeof(pc->pc_name), "ptlrpcd_rcv");
+ } else {
+ /* Regular thread. */
+ snprintf(pc->pc_name, sizeof(pc->pc_name),
+ "ptlrpcd_%02d_%02d", cpt, index);
+ }
+
+ EXIT;
+}
+
+/* XXX: We want multiple CPU cores to share the async RPC load. So we
+ * start many ptlrpcd threads. We also want to reduce the ptlrpcd
+ * overhead caused by data transfer cross-CPU cores. So we bind
+ * all ptlrpcd threads to a CPT, in the expectation that CPTs
+ * will be defined in a way that matches these boundaries. Within
+ * a CPT a ptlrpcd thread can be scheduled on any available core.
+ *
+ * Each ptlrpcd thread has its own request queue. This can cause
+ * response delay if the thread is already busy. To help with
+ * this we define partner threads: these are other threads bound
+ * to the same CPT which will check for work in each other's
+ * request queues if they have no work to do.
+ *
+ * The desired number of partner threads can be tuned by setting
+ * ptlrpcd_partner_group_size. The default is to create pairs of
+ * partner threads.
+ */
+static int ptlrpcd_partners(struct ptlrpcd *pd, int index)
+{
+ struct ptlrpcd_ctl *pc;
+ struct ptlrpcd_ctl **ppc;
+ int first;
+ int i;
+ int rc = 0;
+
+ ENTRY;
+
+ LASSERT(index >= 0 && index < pd->pd_nthreads);
+ pc = &pd->pd_threads[index];
+ pc->pc_npartners = pd->pd_groupsize - 1;
+
+ if (pc->pc_npartners <= 0)
+ GOTO(out, rc);
+
+ OBD_CPT_ALLOC(pc->pc_partners, cfs_cpt_tab, pc->pc_cpt,
+ sizeof(struct ptlrpcd_ctl *) * pc->pc_npartners);
+ if (pc->pc_partners == NULL) {
+ pc->pc_npartners = 0;
+ GOTO(out, rc = -ENOMEM);
+ }
+
+ first = index - index % pd->pd_groupsize;
+ ppc = pc->pc_partners;
+ for (i = first; i < first + pd->pd_groupsize; i++) {
+ if (i != index)
+ *ppc++ = &pd->pd_threads[i];
+ }
+out:
+ RETURN(rc);
+}
+
+int ptlrpcd_start(struct ptlrpcd_ctl *pc)
+{
+ struct task_struct *task;
+ int rc = 0;
+
+ ENTRY;
+
+ /*
+ * Do not allow starting a second thread for one pc.
+ */
+ if (test_and_set_bit(LIOD_START, &pc->pc_flags)) {
+ CWARN("Starting second thread (%s) for same pc %p\n",
+ pc->pc_name, pc);
+ RETURN(0);
+ }
+
+ task = kthread_run(ptlrpcd, pc, "%s", pc->pc_name);
+ if (IS_ERR(task))
+ GOTO(out_set, rc = PTR_ERR(task));
+
+ wait_for_completion(&pc->pc_starting);
+ rc = pc->pc_error;
+ if (rc != 0)
+ GOTO(out_set, rc);
+
+ RETURN(0);
+
+out_set:
+ if (pc->pc_set != NULL) {
+ struct ptlrpc_request_set *set = pc->pc_set;
+
+ spin_lock(&pc->pc_lock);
+ pc->pc_set = NULL;
+ spin_unlock(&pc->pc_lock);
+ ptlrpc_set_destroy(set);
+ }
+ clear_bit(LIOD_START, &pc->pc_flags);
+ RETURN(rc);