ldlm_lib.c

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

        RETURN(rc);

err_import:
        class_destroy_import(imp);
err_ldlm:
        ldlm_put_ref(0);
err:
        RETURN(rc);

}

int client_obd_cleanup(struct obd_device *obddev)
{
        ENTRY;
        ldlm_put_ref(obddev->obd_force);
        RETURN(0);
}

/* ->o_connect() method for client side (OSC and MDC and MGC) */
int client_connect_import(struct lustre_handle *dlm_handle,
                          struct obd_device *obd, struct obd_uuid *cluuid,
                          struct obd_connect_data *data, void *localdata)
{
        struct client_obd *cli = &obd->u.cli;
        struct obd_import *imp = cli->cl_import;
        struct obd_export *exp;
        struct obd_connect_data *ocd;
        struct ldlm_namespace *to_be_freed = NULL;
        int rc;
        ENTRY;

        mutex_down(&cli->cl_sem);
        rc = class_connect(dlm_handle, obd, cluuid);
        if (rc)
                GOTO(out_sem, rc);

        cli->cl_conn_count++;
        if (cli->cl_conn_count > 1)
                GOTO(out_sem, rc);
        exp = class_conn2export(dlm_handle);

        if (obd->obd_namespace != NULL)
                CERROR("already have namespace!\n");
        obd->obd_namespace = ldlm_namespace_new(obd->obd_name,
                                                LDLM_NAMESPACE_CLIENT,
                                                LDLM_NAMESPACE_GREEDY);
        if (obd->obd_namespace == NULL)
                GOTO(out_disco, rc = -ENOMEM);

        imp->imp_dlm_handle = *dlm_handle;
        rc = ptlrpc_init_import(imp);
        if (rc != 0)
                GOTO(out_ldlm, rc);

        ocd = &imp->imp_connect_data;
        if (data) {
                *ocd = *data;
                imp->imp_connect_flags_orig = data->ocd_connect_flags;
        }

        rc = ptlrpc_connect_import(imp, NULL);
        if (rc != 0) {
                LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
                GOTO(out_ldlm, rc);
        }
        LASSERT(exp->exp_connection);

        if (data) {
                LASSERT((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
                        ocd->ocd_connect_flags);
                data->ocd_connect_flags = ocd->ocd_connect_flags;
        }

        ptlrpc_pinger_add_import(imp);
        EXIT;

        if (rc) {
out_ldlm:
                ldlm_namespace_free_prior(obd->obd_namespace);
                to_be_freed = obd->obd_namespace;
                obd->obd_namespace = NULL;
out_disco:
                cli->cl_conn_count--;
                class_disconnect(exp);
        } else {
                class_export_put(exp);
        }
out_sem:
        mutex_up(&cli->cl_sem);
        if (to_be_freed)
                ldlm_namespace_free_post(to_be_freed, 0);
        return rc;
}

int client_disconnect_export(struct obd_export *exp)
{
        struct obd_device *obd = class_exp2obd(exp);
        struct client_obd *cli;
        struct obd_import *imp;
        struct ldlm_namespace *to_be_freed = NULL;
        int rc = 0, err;
        ENTRY;

        if (!obd) {
                CERROR("invalid export for disconnect: exp %p cookie "LPX64"\n",
                       exp, exp ? exp->exp_handle.h_cookie : -1);
                RETURN(-EINVAL);
        }

        cli = &obd->u.cli;
        imp = cli->cl_import;

        mutex_down(&cli->cl_sem);
        if (!cli->cl_conn_count) {
                CERROR("disconnecting disconnected device (%s)\n",
                       obd->obd_name);
                GOTO(out_sem, rc = -EINVAL);
        }

        cli->cl_conn_count--;
        if (cli->cl_conn_count)
                GOTO(out_no_disconnect, rc = 0);

        /* Mark import deactivated now, so we don't try to reconnect if any
         * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
         * fully deactivate the import, or that would drop all requests. */
        spin_lock(&imp->imp_lock);
        imp->imp_deactive = 1;
        spin_unlock(&imp->imp_lock);

        /* Some non-replayable imports (MDS's OSCs) are pinged, so just
         * delete it regardless.  (It's safe to delete an import that was
         * never added.) */
        (void)ptlrpc_pinger_del_import(imp);

        if (obd->obd_namespace != NULL) {
                /* obd_force == local only */
                ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
                                       obd->obd_force ? LDLM_FL_LOCAL_ONLY:0,
                                       NULL);
                ldlm_namespace_free_prior(obd->obd_namespace);
                to_be_freed = obd->obd_namespace;
        }

        rc = ptlrpc_disconnect_import(imp, 0);

        ptlrpc_invalidate_import(imp);
        /* set obd_namespace to NULL only after invalidate, because we can have
         * some connect requests in flight, and his need store a connect flags
         * in obd_namespace. bug 14260 */
        obd->obd_namespace = NULL;
	
        ptlrpc_free_rq_pool(imp->imp_rq_pool);
        class_destroy_import(imp);
        cli->cl_import = NULL;

        EXIT;
 out_no_disconnect:
        err = class_disconnect(exp);
        if (!rc && err)
                rc = err;
 out_sem:
        mutex_up(&cli->cl_sem);
        if (to_be_freed)
                ldlm_namespace_free_post(to_be_freed, obd->obd_force);
        RETURN(rc);
}

/* --------------------------------------------------------------------------
 * from old lib/target.c
 * -------------------------------------------------------------------------- */

int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
                            struct obd_uuid *cluuid)
{
        ENTRY;
        if (exp->exp_connection && exp->exp_imp_reverse) {
                struct lustre_handle *hdl;
                hdl = &exp->exp_imp_reverse->imp_remote_handle;
                /* Might be a re-connect after a partition. */
                if (!memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
                        CWARN("%s: %s reconnecting\n", exp->exp_obd->obd_name,
                              cluuid->uuid);
                        conn->cookie = exp->exp_handle.h_cookie;
                        /* target_handle_connect() treats EALREADY and
                         * -EALREADY differently.  EALREADY means we are
                         * doing a valid reconnect from the same client. */
                        RETURN(EALREADY);
                } else {
                        CERROR("%s reconnecting from %s, "
                               "handle mismatch (ours "LPX64", theirs "
                               LPX64")\n", cluuid->uuid,
                               exp->exp_connection->c_remote_uuid.uuid,
                               hdl->cookie, conn->cookie);
                        memset(conn, 0, sizeof *conn);
                        /* target_handle_connect() treats EALREADY and
                         * -EALREADY differently.  -EALREADY is an error
                         * (same UUID, different handle). */
                        RETURN(-EALREADY);
                }
        }

        conn->cookie = exp->exp_handle.h_cookie;
        CDEBUG(D_HA, "connect export for UUID '%s' at %p, cookie "LPX64"\n",
               cluuid->uuid, exp, conn->cookie);
        RETURN(0);
}

void target_client_add_cb(struct obd_device *obd, __u64 transno, void *cb_data,
                          int error)
{
        struct obd_export *exp = cb_data;

        CDEBUG(D_RPCTRACE, "%s: committing for initial connect of %s\n",
               obd->obd_name, exp->exp_client_uuid.uuid);

        spin_lock(&exp->exp_lock);
        exp->exp_need_sync = 0;
        spin_unlock(&exp->exp_lock);
}
EXPORT_SYMBOL(target_client_add_cb);

static void 
target_start_and_reset_recovery_timer(struct obd_device *obd,
                                      svc_handler_t handler,
                                      struct ptlrpc_request *req,
                                      int new_client);

int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
{
        struct obd_device *target, *targref = NULL;
        struct obd_export *export = NULL;
        struct obd_import *revimp;
        struct lustre_handle conn;
        struct obd_uuid tgtuuid;
        struct obd_uuid cluuid;
        struct obd_uuid remote_uuid;
        char *str, *tmp;
        int rc = 0, abort_recovery;
        struct obd_connect_data *data;
        int size[2] = { sizeof(struct ptlrpc_body), sizeof(*data) };
        lnet_nid_t client_nid = 0;
        ENTRY;

        OBD_RACE(OBD_FAIL_TGT_CONN_RACE);

        lustre_set_req_swabbed(req, REQ_REC_OFF);
        str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, sizeof(tgtuuid)-1);
        if (str == NULL) {
                DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
                GOTO(out, rc = -EINVAL);
        }

        obd_str2uuid (&tgtuuid, str);
        target = class_uuid2obd(&tgtuuid);
        /* COMPAT_146 */
        /* old (pre 1.6) lustre_process_log tries to connect to mdsname
           (eg. mdsA) instead of uuid. */
        if (!target) {
                snprintf((char *)tgtuuid.uuid, sizeof(tgtuuid), "%s_UUID", str);
                target = class_uuid2obd(&tgtuuid);
        }
        if (!target)
                target = class_name2obd(str);
        /* end COMPAT_146 */

        if (!target || target->obd_stopping || !target->obd_set_up) {
                LCONSOLE_ERROR_MSG(0x137, "UUID '%s' is not available "
                                   " for connect (%s)\n", str,
                                   !target ? "no target" :
                                   (target->obd_stopping ? "stopping" :
                                   "not set up"));
                GOTO(out, rc = -ENODEV);
        }

        if (target->obd_no_conn) {
                LCONSOLE_WARN("%s: temporarily refusing client connection "
                              "from %s\n", target->obd_name, 
                              libcfs_nid2str(req->rq_peer.nid));
                GOTO(out, rc = -EAGAIN);
        }

        /* Make sure the target isn't cleaned up while we're here. Yes, 
           there's still a race between the above check and our incref here. 
           Really, class_uuid2obd should take the ref. */
        targref = class_incref(target);

        lustre_set_req_swabbed(req, REQ_REC_OFF + 1);
        str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF + 1,
                                sizeof(cluuid) - 1);
        if (str == NULL) {
                DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
                GOTO(out, rc = -EINVAL);
        }

        obd_str2uuid (&cluuid, str);

        /* XXX extract a nettype and format accordingly */
        switch (sizeof(lnet_nid_t)) {
                /* NB the casts only avoid compiler warnings */
        case 8:
                snprintf(remote_uuid.uuid, sizeof remote_uuid,
                         "NET_"LPX64"_UUID", (__u64)req->rq_peer.nid);
                break;
        case 4:
                snprintf(remote_uuid.uuid, sizeof remote_uuid,
                         "NET_%x_UUID", (__u32)req->rq_peer.nid);
                break;
        default:
                LBUG();
        }

        spin_lock_bh(&target->obd_processing_task_lock);
        abort_recovery = target->obd_abort_recovery;
        spin_unlock_bh(&target->obd_processing_task_lock);
        if (abort_recovery)
                target_abort_recovery(target);

        tmp = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn);
        if (tmp == NULL)
                GOTO(out, rc = -EPROTO);

        memcpy(&conn, tmp, sizeof conn);

        data = lustre_swab_reqbuf(req, REQ_REC_OFF + 3, sizeof(*data),
                                  lustre_swab_connect);
        rc = lustre_pack_reply(req, 2, size, NULL);
        if (rc)
                GOTO(out, rc);

        if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
                if (!data) {
                        DEBUG_REQ(D_WARNING, req, "Refusing old (unversioned) "
                                  "libclient connection attempt");
                        GOTO(out, rc = -EPROTO);
                } else if (data->ocd_version < LUSTRE_VERSION_CODE -
                                               LUSTRE_VERSION_ALLOWED_OFFSET ||
                           data->ocd_version > LUSTRE_VERSION_CODE +
                                               LUSTRE_VERSION_ALLOWED_OFFSET) {
                        DEBUG_REQ(D_WARNING, req, "Refusing %s (%d.%d.%d.%d) "