import.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,348 行 · 第 1/4 页

        ptlrpc_daemonize("ll_imp_inval");
        
        CDEBUG(D_HA, "thread invalidate import %s to %s@%s\n",
               imp->imp_obd->obd_name, obd2cli_tgt(imp->imp_obd),
               imp->imp_connection->c_remote_uuid.uuid);

        ptlrpc_invalidate_import(imp);

        if (obd_dump_on_eviction) {
                CERROR("dump the log upon eviction\n");
                libcfs_debug_dumplog();
        }

        IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
        ptlrpc_import_recovery_state_machine(imp);

        RETURN(0);
}
#endif

int ptlrpc_import_recovery_state_machine(struct obd_import *imp)
{
        int rc = 0;
        int inflight;
        char *target_start;
        int target_len;

        ENTRY;
        if (imp->imp_state == LUSTRE_IMP_EVICTED) {
                deuuidify(obd2cli_tgt(imp->imp_obd), NULL,
                          &target_start, &target_len);
                /* Don't care about MGC eviction */
                if (strcmp(imp->imp_obd->obd_type->typ_name,
                           LUSTRE_MGC_NAME) != 0) {
                        LCONSOLE_ERROR_MSG(0x167, "This client was evicted by "
                                           "%.*s; in progress operations using "
                                           "this service will fail.\n",
                                           target_len, target_start);
                }
                CDEBUG(D_HA, "evicted from %s@%s; invalidating\n",
                       obd2cli_tgt(imp->imp_obd),
                       imp->imp_connection->c_remote_uuid.uuid);

#ifdef __KERNEL__
                rc = cfs_kernel_thread(ptlrpc_invalidate_import_thread, imp,
                                   CLONE_VM | CLONE_FILES);
                if (rc < 0)
                        CERROR("error starting invalidate thread: %d\n", rc);
                else
                        rc = 0;
                RETURN(rc);
#else
                ptlrpc_invalidate_import(imp);

                IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
#endif
        }

        if (imp->imp_state == LUSTRE_IMP_REPLAY) {
                CDEBUG(D_HA, "replay requested by %s\n",
                       obd2cli_tgt(imp->imp_obd));
                rc = ptlrpc_replay_next(imp, &inflight);
                if (inflight == 0 &&
                    atomic_read(&imp->imp_replay_inflight) == 0) {
                        IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY_LOCKS);
                        rc = ldlm_replay_locks(imp);
                        if (rc)
                                GOTO(out, rc);
                }
                rc = 0;
        }

        if (imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS) {
                if (atomic_read(&imp->imp_replay_inflight) == 0) {
                        IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY_WAIT);
                        rc = signal_completed_replay(imp);
                        if (rc)
                                GOTO(out, rc);
                }

        }

        if (imp->imp_state == LUSTRE_IMP_REPLAY_WAIT) {
                if (atomic_read(&imp->imp_replay_inflight) == 0) {
                        IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
                }
        }

        if (imp->imp_state == LUSTRE_IMP_RECOVER) {
                CDEBUG(D_HA, "reconnected to %s@%s\n",
                       obd2cli_tgt(imp->imp_obd),
                       imp->imp_connection->c_remote_uuid.uuid);

                rc = ptlrpc_resend(imp);
                if (rc)
                        GOTO(out, rc);
                IMPORT_SET_STATE(imp, LUSTRE_IMP_FULL);
                ptlrpc_activate_import(imp);

                deuuidify(obd2cli_tgt(imp->imp_obd), NULL,
                          &target_start, &target_len);
                LCONSOLE_INFO("%s: Connection restored to service %.*s "
                              "using nid %s.\n", imp->imp_obd->obd_name,
                              target_len, target_start,
                              libcfs_nid2str(imp->imp_connection->c_peer.nid));
        }

        if (imp->imp_state == LUSTRE_IMP_FULL) {
                cfs_waitq_signal(&imp->imp_recovery_waitq);
                ptlrpc_wake_delayed(imp);
        }

 out:
        RETURN(rc);
}

static int back_to_sleep(void *unused)
{
        return 0;
}

int ptlrpc_disconnect_import(struct obd_import *imp, int noclose)
{
        struct ptlrpc_request *req;
        int rq_opc, rc = 0;
        int nowait = imp->imp_obd->obd_force;
        ENTRY;

        if (nowait)
                GOTO(set_state, rc);

        switch (imp->imp_connect_op) {
        case OST_CONNECT: rq_opc = OST_DISCONNECT; break;
        case MDS_CONNECT: rq_opc = MDS_DISCONNECT; break;
        case MGS_CONNECT: rq_opc = MGS_DISCONNECT; break;
        default:
                CERROR("don't know how to disconnect from %s (connect_op %d)\n",
                       obd2cli_tgt(imp->imp_obd), imp->imp_connect_op);
                RETURN(-EINVAL);
        }

        if (ptlrpc_import_in_recovery(imp)) {
                struct l_wait_info lwi;
                cfs_duration_t timeout;

                if (AT_OFF) {
                        timeout = cfs_time_seconds(obd_timeout);
                } else {
                        int idx = import_at_get_index(imp, 
                                imp->imp_client->cli_request_portal);
                        timeout = cfs_time_seconds(
                                at_get(&imp->imp_at.iat_service_estimate[idx]));
                }
                lwi = LWI_TIMEOUT_INTR(cfs_timeout_cap(timeout), 
                                       back_to_sleep, LWI_ON_SIGNAL_NOOP, NULL);
                rc = l_wait_event(imp->imp_recovery_waitq,
                                  !ptlrpc_import_in_recovery(imp), &lwi);
        }

        spin_lock(&imp->imp_lock);
        if (imp->imp_state != LUSTRE_IMP_FULL)
                GOTO(out, 0);

        spin_unlock(&imp->imp_lock);

        req = ptlrpc_prep_req(imp, LUSTRE_OBD_VERSION, rq_opc, 1, NULL, NULL);
        if (req) {
                /* We are disconnecting, do not retry a failed DISCONNECT rpc if
                 * it fails.  We can get through the above with a down server
                 * if the client doesn't know the server is gone yet. */
                req->rq_no_resend = 1;
                
#ifndef CRAY_XT3
                /* We want client umounts to happen quickly, no matter the 
                   server state... */
                req->rq_timeout = min_t(int, req->rq_timeout,
                                        INITIAL_CONNECT_TIMEOUT);
#else
                /* ... but we always want liblustre clients to nicely 
                   disconnect, so only use the adaptive value. */
                if (AT_OFF)
                        req->rq_timeout = obd_timeout / 3;
#endif

                IMPORT_SET_STATE(imp, LUSTRE_IMP_CONNECTING);
                req->rq_send_state =  LUSTRE_IMP_CONNECTING;
                ptlrpc_req_set_repsize(req, 1, NULL);
                rc = ptlrpc_queue_wait(req);
                ptlrpc_req_finished(req);
        }

set_state:
        spin_lock(&imp->imp_lock);
out:
        if (noclose) 
                IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_DISCON);
        else
                IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_CLOSED);
        memset(&imp->imp_remote_handle, 0, sizeof(imp->imp_remote_handle));
        /* Try all connections in the future - bz 12758 */ 
        imp->imp_last_recon = 0;
        spin_unlock(&imp->imp_lock);

        RETURN(rc);
}

/* Sets maximal number of RPCs possible originating from other side of this
   import (server) to us and number of async RPC replies that we are not waiting
   for arriving */
void ptlrpc_import_setasync(struct obd_import *imp, int count)
{
        LNetSetAsync(imp->imp_connection->c_peer, count);
}


/* Adaptive Timeout utils */
extern unsigned int at_min, at_max, at_history;

/* Bin into timeslices using AT_BINS bins.
   This gives us a max of the last binlimit*AT_BINS secs without the storage,
   but still smoothing out a return to normalcy from a slow response.
   (E.g. remember the maximum latency in each minute of the last 4 minutes.) */
int at_add(struct adaptive_timeout *at, unsigned int val) 
{
        unsigned int old = at->at_current;
        time_t now = cfs_time_current_sec();
        time_t binlimit = max_t(time_t, at_history / AT_BINS, 1);

        LASSERT(at);
#if 0
        CDEBUG(D_INFO, "add %u to %p time=%lu v=%u (%u %u %u %u)\n", 
               val, at, now - at->at_binstart, at->at_current,
               at->at_hist[0], at->at_hist[1], at->at_hist[2], at->at_hist[3]);
#endif
        if (val == 0) 
                /* 0's don't count, because we never want our timeout to 
                   drop to 0, and because 0 could mean an error */
                return 0;

        spin_lock(&at->at_lock);

        if (unlikely(at->at_binstart == 0)) {
                /* Special case to remove default from history */
                at->at_current = val;
                at->at_worst_ever = val;
                at->at_worst_time = now;
                at->at_hist[0] = val;
                at->at_binstart = now;
        } else if (now - at->at_binstart < binlimit ) {
                /* in bin 0 */
                at->at_hist[0] = max(val, at->at_hist[0]);
                at->at_current = max(val, at->at_current);
        } else {
                int i, shift;
                unsigned int maxv = val;
                /* move bins over */
                shift = (now - at->at_binstart) / binlimit;
                LASSERT(shift > 0);
                for(i = AT_BINS - 1; i >= 0; i--) {
                        if (i >= shift) {
                                at->at_hist[i] = at->at_hist[i - shift];
                                maxv = max(maxv, at->at_hist[i]);
                        } else {
                                at->at_hist[i] = 0;
                        }
                }
                at->at_hist[0] = val;
                at->at_current = maxv;
                at->at_binstart += shift * binlimit;
        }

        if (at->at_current > at->at_worst_ever) {
                at->at_worst_ever = at->at_current;
                at->at_worst_time = now;
        }

        if (at->at_flags & AT_FLG_NOHIST)
                /* Only keep last reported val; keeping the rest of the history
                   for proc only */
                at->at_current = val;

        if (at_max > 0)
                at->at_current =  min(at->at_current, at_max);
        at->at_current =  max(at->at_current, at_min);

#if 0
        if (at->at_current != old)
                CDEBUG(D_ADAPTTO, "AT %p change: old=%u new=%u delta=%d "
                       "(val=%u) hist %u %u %u %u\n", at,
                       old, at->at_current, at->at_current - old, val,
                       at->at_hist[0], at->at_hist[1], at->at_hist[2],
                       at->at_hist[3]);
#endif
        
        /* if we changed, report the old value */
        old = (at->at_current != old) ? old : 0;
        
        spin_unlock(&at->at_lock);
        return old;
}

/* Find the imp_at index for a given portal; assign if space available */
int import_at_get_index(struct obd_import *imp, int portal) 
{
        struct imp_at *at = &imp->imp_at;
        int i;

        for (i = 0; i < IMP_AT_MAX_PORTALS; i++) {
                if (at->iat_portal[i] == portal) 
                        return i;
                if (at->iat_portal[i] == 0)
                        /* unused */
                        break;
        }

        /* Not found in list, add it under a lock */
        spin_lock(&imp->imp_lock);

        /* Check unused under lock */
        for (; i < IMP_AT_MAX_PORTALS; i++) {
                if (at->iat_portal[i] == portal) 
                        goto out;
                if (at->iat_portal[i] == 0)
                        /* unused */
                        break;
        }
        
        /* Not enough portals? */
        LASSERT(i < IMP_AT_MAX_PORTALS);

        at->iat_portal[i] = portal;
out:
        spin_unlock(&imp->imp_lock);
        return i;
}