import.c

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

        list_for_each_entry(conn, &imp->imp_conn_list, oic_item) {
                CDEBUG(D_HA, "%s: connect to NID %s last attempt "LPU64"\n",
                       imp->imp_obd->obd_name,
                       libcfs_nid2str(conn->oic_conn->c_peer.nid),
                       conn->oic_last_attempt);
                
                /* Don't thrash connections */
                if (cfs_time_before_64(cfs_time_current_64(),
                                     conn->oic_last_attempt + 
                                     cfs_time_seconds(CONNECTION_SWITCH_MIN))) {
                        continue;
                }

                /* If we have not tried this connection since the
                   the last successful attempt, go with this one */
                if ((conn->oic_last_attempt == 0) ||
                    cfs_time_beforeq_64(conn->oic_last_attempt,
                                       imp->imp_last_success_conn)) {
                        imp_conn = conn;
                        tried_all = 0;
                        break;
                }

                /* If all of the connections have already been tried
                   since the last successful connection; just choose the
                   least recently used */
                if (!imp_conn)
                        imp_conn = conn;
                else if (cfs_time_before_64(conn->oic_last_attempt,
                                            imp_conn->oic_last_attempt))
                        imp_conn = conn;
        }

        /* if not found, simply choose the current one */
        if (!imp_conn) {
                LASSERT(imp->imp_conn_current);
                imp_conn = imp->imp_conn_current;
                tried_all = 0;
        }
        LASSERT(imp_conn->oic_conn);

        /* If we've tried everything, and we're back to the beginning of the
           list, increase our timeout and try again. It will be reset when
           we do finally connect. (FIXME: really we should wait for all network
           state associated with the last connection attempt to drain before
           trying to reconnect on it.) */
        if (tried_all && (imp->imp_conn_list.next == &imp_conn->oic_item) &&
            !imp->imp_recon_bk /* not retrying */) {
                if (at_get(&imp->imp_at.iat_net_latency) <
                    CONNECTION_SWITCH_MAX) {
                        at_add(&imp->imp_at.iat_net_latency,
                               at_get(&imp->imp_at.iat_net_latency) +
                               CONNECTION_SWITCH_INC);
                }
                LASSERT(imp_conn->oic_last_attempt);
                CWARN("%s: tried all connections, increasing latency to %ds\n",
                      imp->imp_obd->obd_name,
                      at_get(&imp->imp_at.iat_net_latency));
        }

        imp_conn->oic_last_attempt = cfs_time_current_64();

        /* switch connection, don't mind if it's same as the current one */
        if (imp->imp_connection)
                ptlrpc_put_connection(imp->imp_connection);
        imp->imp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);

        dlmexp =  class_conn2export(&imp->imp_dlm_handle);
        LASSERT(dlmexp != NULL);
        if (dlmexp->exp_connection)
                ptlrpc_put_connection(dlmexp->exp_connection);
        dlmexp->exp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);
        class_export_put(dlmexp);

        if (imp->imp_conn_current != imp_conn) {
                if (imp->imp_conn_current)
                        LCONSOLE_INFO("Changing connection for %s to %s/%s\n",
                                      imp->imp_obd->obd_name,
                                      imp_conn->oic_uuid.uuid,
                                      libcfs_nid2str(imp_conn->oic_conn->c_peer.nid));
                imp->imp_conn_current = imp_conn;
        }

        CDEBUG(D_HA, "%s: import %p using connection %s/%s\n",
               imp->imp_obd->obd_name, imp, imp_conn->oic_uuid.uuid,
               libcfs_nid2str(imp_conn->oic_conn->c_peer.nid));

        spin_unlock(&imp->imp_lock);

        RETURN(0);
}

int ptlrpc_connect_import(struct obd_import *imp, char *new_uuid)
{
        struct obd_device *obd = imp->imp_obd;
        int initial_connect = 0;
        int rc;
        __u64 committed_before_reconnect = 0;
        struct ptlrpc_request *request;
        int size[] = { sizeof(struct ptlrpc_body),
                       sizeof(imp->imp_obd->u.cli.cl_target_uuid),
                       sizeof(obd->obd_uuid),
                       sizeof(imp->imp_dlm_handle),
                       sizeof(imp->imp_connect_data) };
        char *tmp[] = { NULL,
                        obd2cli_tgt(imp->imp_obd),
                        obd->obd_uuid.uuid,
                        (char *)&imp->imp_dlm_handle,
                        (char *)&imp->imp_connect_data };
        struct ptlrpc_connect_async_args *aa;

        ENTRY;
        spin_lock(&imp->imp_lock);
        if (imp->imp_state == LUSTRE_IMP_CLOSED) {
                spin_unlock(&imp->imp_lock);
                CERROR("can't connect to a closed import\n");
                RETURN(-EINVAL);
        } else if (imp->imp_state == LUSTRE_IMP_FULL) {
                spin_unlock(&imp->imp_lock);
                CERROR("already connected\n");
                RETURN(0);
        } else if (imp->imp_state == LUSTRE_IMP_CONNECTING) {
                spin_unlock(&imp->imp_lock);
                CERROR("already connecting\n");
                RETURN(-EALREADY);
        }

        IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_CONNECTING);

        imp->imp_conn_cnt++;
        imp->imp_resend_replay = 0;

        if (!lustre_handle_is_used(&imp->imp_remote_handle))
                initial_connect = 1;
        else
                committed_before_reconnect = imp->imp_peer_committed_transno;

        spin_unlock(&imp->imp_lock);

        if (new_uuid) {
                struct obd_uuid uuid;

                obd_str2uuid(&uuid, new_uuid);
                rc = import_set_conn_priority(imp, &uuid);
                if (rc)
                        GOTO(out, rc);
        }

        rc = import_select_connection(imp);
        if (rc)
                GOTO(out, rc);

        /* last in connection list */
        if (imp->imp_conn_current->oic_item.next == &imp->imp_conn_list) {
                if (imp->imp_initial_recov_bk && initial_connect) {
                        CDEBUG(D_HA, "Last connection attempt (%d) for %s\n",
                               imp->imp_conn_cnt, obd2cli_tgt(imp->imp_obd));
                        /* Don't retry if connect fails */
                        rc = 0;
                        obd_set_info_async(obd->obd_self_export,
                                           strlen(KEY_INIT_RECOV),
                                           KEY_INIT_RECOV,
                                           sizeof(rc), &rc, NULL);
                }
                if (imp->imp_recon_bk) {
                        CDEBUG(D_HA, "Last reconnection attempt (%d) for %s\n",
                               imp->imp_conn_cnt, obd2cli_tgt(imp->imp_obd));
                        spin_lock(&imp->imp_lock);
                        imp->imp_last_recon = 1;
                        spin_unlock(&imp->imp_lock);
                }
        }

        /* Reset connect flags to the originally requested flags, in case
         * the server is updated on-the-fly we will get the new features. */
        imp->imp_connect_data.ocd_connect_flags = imp->imp_connect_flags_orig;
        imp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;

        rc = obd_reconnect(imp->imp_obd->obd_self_export, obd,
                           &obd->obd_uuid, &imp->imp_connect_data);
        if (rc)
                GOTO(out, rc);

        request = ptlrpc_prep_req(imp, LUSTRE_OBD_VERSION, imp->imp_connect_op,
                                  5, size, tmp);
        if (!request)
                GOTO(out, rc = -ENOMEM);

#ifndef __KERNEL__
        lustre_msg_add_op_flags(request->rq_reqmsg, MSG_CONNECT_LIBCLIENT);
#endif
        if (imp->imp_msg_magic == LUSTRE_MSG_MAGIC_V1)
                lustre_msg_add_op_flags(request->rq_reqmsg,
                                        MSG_CONNECT_NEXT_VER);

        request->rq_send_state = LUSTRE_IMP_CONNECTING;
        /* Allow a slightly larger reply for future growth compatibility */
        size[REPLY_REC_OFF] = sizeof(struct obd_connect_data) +
                              16 * sizeof(__u64);
        ptlrpc_req_set_repsize(request, 2, size);
        request->rq_interpret_reply = ptlrpc_connect_interpret;

        CLASSERT(sizeof (*aa) <= sizeof (request->rq_async_args));
        aa = (struct ptlrpc_connect_async_args *)&request->rq_async_args;
        memset(aa, 0, sizeof *aa);

        aa->pcaa_peer_committed = committed_before_reconnect;
        aa->pcaa_initial_connect = initial_connect;
        if (aa->pcaa_initial_connect) {
                spin_lock(&imp->imp_lock);
                imp->imp_replayable = 1;
                spin_unlock(&imp->imp_lock);
                if (AT_OFF)
                        /* AT will use INITIAL_CONNECT_TIMEOUT the first
                           time, adaptive after that. */
                        request->rq_timeout = INITIAL_CONNECT_TIMEOUT;
        }

        DEBUG_REQ(D_RPCTRACE, request, "%sconnect request %d",
                  aa->pcaa_initial_connect ? "initial " : "re", 
                  imp->imp_conn_cnt);
        ptlrpcd_add_req(request);
        rc = 0;
out:
        if (rc != 0) {
                IMPORT_SET_STATE(imp, LUSTRE_IMP_DISCON);
        }

        RETURN(rc);
}
EXPORT_SYMBOL(ptlrpc_connect_import);

static void ptlrpc_maybe_ping_import_soon(struct obd_import *imp)
{
#ifdef __KERNEL__
        struct obd_import_conn *imp_conn;
#endif
        int wake_pinger = 0;

        ENTRY;

        spin_lock(&imp->imp_lock);
        if (list_empty(&imp->imp_conn_list))
                GOTO(unlock, 0);

#ifdef __KERNEL__
        imp_conn = list_entry(imp->imp_conn_list.prev,
                              struct obd_import_conn,
                              oic_item);

        /* XXX: When the failover node is the primary node, it is possible
         * to have two identical connections in imp_conn_list. We must 
         * compare not conn's pointers but NIDs, otherwise we can defeat
         * connection throttling. (See bug 14774.) */
        if (imp->imp_conn_current->oic_conn->c_self != 
                                imp_conn->oic_conn->c_self) {
                ptlrpc_ping_import_soon(imp);
                wake_pinger = 1;
        }

#else
        /* liblustre has no pinger thead, so we wakup pinger anyway */
        wake_pinger = 1;
#endif 
 unlock:
        spin_unlock(&imp->imp_lock);

        if (wake_pinger)
                ptlrpc_pinger_wake_up();

        EXIT;
}

static int ptlrpc_connect_interpret(struct ptlrpc_request *request,
                                    void * data, int rc)
{
        struct ptlrpc_connect_async_args *aa = data;
        struct obd_import *imp = request->rq_import;
        struct client_obd *cli = &imp->imp_obd->u.cli;
        struct lustre_handle old_hdl;
        int msg_flags;
        ENTRY;

        spin_lock(&imp->imp_lock);
        if (imp->imp_state == LUSTRE_IMP_CLOSED) {
                spin_unlock(&imp->imp_lock);
                RETURN(0);
        }
        spin_unlock(&imp->imp_lock);

        if (rc)
                GOTO(out, rc);

        LASSERT(imp->imp_conn_current);

        msg_flags = lustre_msg_get_op_flags(request->rq_repmsg);

        /* All imports are pingable */
        spin_lock(&imp->imp_lock);
        imp->imp_pingable = 1;

        if (aa->pcaa_initial_connect) {
                if (msg_flags & MSG_CONNECT_REPLAYABLE) {
                        imp->imp_replayable = 1;
                        spin_unlock(&imp->imp_lock);
                        CDEBUG(D_HA, "connected to replayable target: %s\n",
                               obd2cli_tgt(imp->imp_obd));
                } else {
                        imp->imp_replayable = 0;
                        spin_unlock(&imp->imp_lock);
                }

                if ((request->rq_reqmsg->lm_magic == LUSTRE_MSG_MAGIC_V1 &&
                     msg_flags & MSG_CONNECT_NEXT_VER) ||
                    request->rq_reqmsg->lm_magic == LUSTRE_MSG_MAGIC_V2) {
                        imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
                        CDEBUG(D_RPCTRACE, "connect to %s with lustre_msg_v2\n",
                               obd2cli_tgt(imp->imp_obd));
                } else {
                        CDEBUG(D_RPCTRACE, "connect to %s with lustre_msg_v1\n",
                               obd2cli_tgt(imp->imp_obd));
                }

                imp->imp_remote_handle =
                                *lustre_msg_get_handle(request->rq_repmsg);

                IMPORT_SET_STATE(imp, LUSTRE_IMP_FULL);
                spin_lock(&imp->imp_lock);
                if (imp->imp_invalid) {
                        spin_unlock(&imp->imp_lock);
                        ptlrpc_activate_import(imp);
                } else {
                        spin_unlock(&imp->imp_lock);
                }

                GOTO(finish, rc = 0);
        } else {