extern int (*mds_reint_p)(int offset, struct ptlrpc_request *req);
extern int (*mds_getattr_name_p)(int offset, struct ptlrpc_request *req);
-static int ldlm_handle_enqueue(struct obd_device *obddev, struct ptlrpc_service *svc,
- struct ptlrpc_request *req)
+static int ldlm_handle_enqueue(struct ptlrpc_request *req)
{
+ struct obd_device *obddev = req->rq_export->export_obd;
struct ldlm_reply *dlm_rep;
struct ldlm_request *dlm_req;
int rc, size = sizeof(*dlm_rep), cookielen = 0;
__u32 flags;
ldlm_error_t err;
struct ldlm_lock *lock = NULL;
- struct lustre_handle lockh;
void *cookie = NULL;
ENTRY;
}
}
- lock = ldlm_local_lock_create(obddev->obd_namespace,
- &dlm_req->lock_handle2,
- dlm_req->lock_desc.l_resource.lr_name,
- dlm_req->lock_desc.l_resource.lr_type,
- dlm_req->lock_desc.l_req_mode,
- NULL, 0);
+ lock = ldlm_lock_create(obddev->obd_namespace,
+ &dlm_req->lock_handle2,
+ dlm_req->lock_desc.l_resource.lr_name,
+ dlm_req->lock_desc.l_resource.lr_type,
+ dlm_req->lock_desc.l_req_mode, NULL, 0);
if (!lock)
- GOTO(out, -ENOMEM);
+ GOTO(out, err = -ENOMEM);
memcpy(&lock->l_remote_handle, &dlm_req->lock_handle1,
sizeof(lock->l_remote_handle));
flags = dlm_req->lock_flags;
err = ldlm_lock_enqueue(lock, cookie, cookielen, &flags,
- ldlm_cli_callback, ldlm_cli_callback);
+ ldlm_server_ast, ldlm_server_ast);
if (err != ELDLM_OK)
GOTO(out, err);
lock->l_connection = ptlrpc_connection_addref(req->rq_connection);
EXIT;
out:
- if (lock)
+ if (lock) {
+ LDLM_DEBUG(lock, "server-side enqueue handler, sending reply");
ldlm_lock_put(lock);
+ }
req->rq_status = err;
CDEBUG(D_INFO, "err = %d\n", err);
- if (ptlrpc_reply(svc, req))
+ if (ptlrpc_reply(req->rq_svc, req))
LBUG();
- if (err)
- LDLM_DEBUG_NOLOCK("server-side enqueue handler END");
- else {
+ if (!err)
ldlm_reprocess_all(lock->l_resource);
- LDLM_DEBUG(lock, "server-side enqueue handler END");
- }
+ LDLM_DEBUG_NOLOCK("server-side enqueue handler END");
return 0;
}
-static int ldlm_handle_convert(struct ptlrpc_service *svc, struct ptlrpc_request *req)
+static int ldlm_handle_convert(struct ptlrpc_request *req)
{
struct ldlm_request *dlm_req;
struct ldlm_reply *dlm_rep;
- struct ldlm_resource *res;
struct ldlm_lock *lock;
int rc, size = sizeof(*dlm_rep);
ENTRY;
req->rq_status = EINVAL;
} else {
LDLM_DEBUG(lock, "server-side convert handler START");
- ldlm_lock_convert(&dlm_req->lock_handle1,
- dlm_req->lock_desc.l_req_mode,
+ ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
&dlm_rep->lock_flags);
req->rq_status = 0;
}
- if (ptlrpc_reply(svc, req) != 0)
+ if (ptlrpc_reply(req->rq_svc, req) != 0)
LBUG();
- ldlm_reprocess_all(lock->res);
+ ldlm_reprocess_all(lock->l_resource);
ldlm_lock_put(lock);
LDLM_DEBUG(lock, "server-side convert handler END");
RETURN(0);
}
-static int ldlm_handle_cancel(struct ptlrpc_service *svc, struct ptlrpc_request *req)
+static int ldlm_handle_cancel(struct ptlrpc_request *req)
{
struct ldlm_request *dlm_req;
struct ldlm_lock *lock;
- struct ldlm_resource *res;
int rc;
ENTRY;
req->rq_status = ESTALE;
} else {
LDLM_DEBUG(lock, "server-side cancel handler START");
- ldlm_local_lock_cancel(lock);
+ ldlm_lock_cancel(lock);
req->rq_status = 0;
}
- if (ptlrpc_reply(svc, req) != 0)
+ if (ptlrpc_reply(req->rq_svc, req) != 0)
LBUG();
ldlm_reprocess_all(lock->l_resource);
- ldlm_lock_put(put);
+ ldlm_lock_put(lock);
LDLM_DEBUG_NOLOCK("server-side cancel handler END");
RETURN(0);
}
-static int ldlm_handle_callback(struct ptlrpc_service *svc,
- struct ptlrpc_request *req)
+static int ldlm_handle_callback(struct ptlrpc_request *req)
{
struct ldlm_request *dlm_req;
- struct ldlm_lock_desc desc;
- struct ldlm_lock_desc *descp;
+ struct ldlm_lock_desc *descp = NULL;
struct ldlm_lock *lock;
- __u64 is_blocking_ast;
+ __u64 is_blocking_ast = 0;
int rc;
ENTRY;
/* We must send the reply first, so that the thread is free to handle
* any requests made in common_callback() */
- rc = ptlrpc_reply(svc, req);
+ rc = ptlrpc_reply(req->rq_svc, req);
if (rc != 0)
RETURN(rc);
-
+
lock = ldlm_handle2lock(&dlm_req->lock_handle1);
- is_blocking_ast = dlm_req->lock_handle2.addr;
- if (is_blocking_ast) {
- /* FIXME: copy lock information into desc here */
- descp = &desc;
- } else
- descp = NULL;
-
if (!lock) {
CERROR("callback on lock %Lx - lock disappeared\n",
- dlm_req->lock_handle.addr);
+ dlm_req->lock_handle1.addr);
RETURN(0);
}
+ /* check if this is a blocking AST */
+ if (dlm_req->lock_desc.l_req_mode !=
+ dlm_req->lock_desc.l_granted_mode) {
+ descp = &dlm_req->lock_desc;
+ is_blocking_ast = 1;
+ }
+
LDLM_DEBUG(lock, "client %s callback handler START",
- new == NULL ? "completion" : "blocked");
+ is_blocking_ast ? "blocked" : "completion");
- if (is_blocking_ast) {
+ if (descp) {
int do_ast;
l_lock(&lock->l_resource->lr_namespace->ns_lock);
lock->l_flags |= LDLM_FL_CBPENDING;
do_ast = (!lock->l_readers && !lock->l_writers);
l_unlock(&lock->l_resource->lr_namespace->ns_lock);
- if (do_ast) {
- CDEBUG(D_INFO, "Lock already unused, calling "
- "callback (%p).\n", lock->l_blocking_ast);
- if (lock->l_blocking_ast != NULL)
- lock->l_blocking_ast(lock, descp, lock->l_data,
- lock->l_data_len, NULL);
+ if (do_ast) {
+ LDLM_DEBUG(lock, "already unused, calling "
+ "callback (%p)", lock->l_blocking_ast);
+ if (lock->l_blocking_ast != NULL) {
+ struct lustre_handle lockh;
+ ldlm_lock2handle(lock, &lockh);
+ lock->l_blocking_ast(&lockh, descp,
+ lock->l_data,
+ lock->l_data_len);
+ }
} else {
LDLM_DEBUG(lock, "Lock still has references, will be"
" cancelled later");
}
ldlm_lock_put(lock);
} else {
+ struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
+
+ l_lock(&lock->l_resource->lr_namespace->ns_lock);
lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
/* If we receive the completion AST before the actual enqueue
ldlm_lock_change_resource(lock, dlm_req->lock_desc.l_resource.lr_name);
LDLM_DEBUG(lock, "completion AST, new resource");
}
+ lock->l_resource->lr_tmp = &rpc_list;
+ ldlm_resource_unlink_lock(lock);
ldlm_grant_lock(lock);
/* FIXME: we want any completion function, not just wake_up */
wake_up(&lock->l_waitq);
ldlm_lock_put(lock);
+ lock->l_resource->lr_tmp = NULL;
+ l_unlock(&lock->l_resource->lr_namespace->ns_lock);
+
+ ldlm_run_ast_work(&rpc_list);
}
LDLM_DEBUG_NOLOCK("client %s callback handler END (lock: %p)",
- new == NULL ? "completion" : "blocked", lock);
+ is_blocking_ast ? "blocked" : "completion", lock);
RETURN(0);
}
-static int lustre_handle(struct obd_device *dev, struct ptlrpc_service *svc,
- struct ptlrpc_request *req)
+static int ldlm_handle(struct ptlrpc_request *req)
{
- struct obd_device *req_dev;
- int id, rc;
+ int rc;
ENTRY;
rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
GOTO(out, rc = -EINVAL);
}
- id = req->rq_reqmsg->target_id;
- if (id < 0 || id > MAX_OBD_DEVICES)
- GOTO(out, rc = -ENODEV);
- req_dev = req->rq_obd = &obd_dev[id];
-
+ if (!req->rq_export &&
+ req->rq_reqmsg->opc == LDLM_ENQUEUE) {
+ CERROR("No export handle for enqueue request.\n");
+ GOTO(out, rc = -ENOTCONN);
+ }
switch (req->rq_reqmsg->opc) {
case LDLM_ENQUEUE:
CDEBUG(D_INODE, "enqueue\n");
OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
- rc = ldlm_handle_enqueue(req_dev, svc, req);
+ rc = ldlm_handle_enqueue(req);
break;
case LDLM_CONVERT:
CDEBUG(D_INODE, "convert\n");
OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
- rc = ldlm_handle_convert(svc, req);
+ rc = ldlm_handle_convert(req);
break;
case LDLM_CANCEL:
CDEBUG(D_INODE, "cancel\n");
OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
- rc = ldlm_hanel_cancel(svc, req);
+ rc = ldlm_handle_cancel(req);
break;
case LDLM_CALLBACK:
CDEBUG(D_INODE, "callback\n");
OBD_FAIL_RETURN(OBD_FAIL_LDLM_CALLBACK, 0);
- rc = ldlm_handle_callback(svc, req);
+ rc = ldlm_handle_callback(req);
break;
default:
- rc = ptlrpc_error(svc, req);
+ rc = ptlrpc_error(req->rq_svc, req);
RETURN(rc);
}
EXIT;
out:
if (rc)
- RETURN(ptlrpc_error(svc, req));
+ RETURN(ptlrpc_error(req->rq_svc, req));
return 0;
}
-static int ldlm_iocontrol(long cmd, struct obd_conn *conn, int len, void *karg,
+static int ldlm_iocontrol(long cmd, struct lustre_handle *conn, int len, void *karg,
void *uarg)
{
- struct obd_device *obddev = conn->oc_dev;
+ struct obd_device *obddev = class_conn2obd(conn);
struct ptlrpc_connection *connection;
int err;
ENTRY;
MOD_INC_USE_COUNT;
ldlm->ldlm_service =
ptlrpc_init_svc(64 * 1024, LDLM_REQUEST_PORTAL,
- LDLM_REPLY_PORTAL, "self", lustre_handle);
+ LDLM_REPLY_PORTAL, "self", ldlm_handle);
if (!ldlm->ldlm_service) {
LBUG();
GOTO(out_dec, rc = -ENOMEM);
for (i = 0; i < LDLM_NUM_THREADS; i++) {
rc = ptlrpc_start_thread(obddev, ldlm->ldlm_service,
"lustre_dlm");
- /* XXX We could just continue if we had started at least
- * a few threads here.
- */
if (rc) {
CERROR("cannot start LDLM thread #%d: rc %d\n", i, rc);
LBUG();
o_iocontrol: ldlm_iocontrol,
o_setup: ldlm_setup,
o_cleanup: ldlm_cleanup,
- o_connect: gen_connect,
- o_disconnect: gen_disconnect
+ o_connect: class_connect,
+ o_disconnect: class_disconnect
};
static int __init ldlm_init(void)
{
- int rc = obd_register_type(&ldlm_obd_ops, OBD_LDLM_DEVICENAME);
+ int rc = class_register_type(&ldlm_obd_ops, OBD_LDLM_DEVICENAME);
if (rc != 0)
return rc;
static void __exit ldlm_exit(void)
{
- obd_unregister_type(OBD_LDLM_DEVICENAME);
+ class_unregister_type(OBD_LDLM_DEVICENAME);
if (kmem_cache_destroy(ldlm_resource_slab) != 0)
CERROR("couldn't free ldlm resource slab\n");
if (kmem_cache_destroy(ldlm_lock_slab) != 0)
CERROR("couldn't free ldlm lock slab\n");
}
-EXPORT_SYMBOL(ldlm_local_lock_match);
+EXPORT_SYMBOL(ldlm_lockname);
+EXPORT_SYMBOL(ldlm_typename);
+EXPORT_SYMBOL(ldlm_handle2lock);
+EXPORT_SYMBOL(ldlm_lock_match);
EXPORT_SYMBOL(ldlm_lock_addref);
EXPORT_SYMBOL(ldlm_lock_decref);
EXPORT_SYMBOL(ldlm_cli_convert);
EXPORT_SYMBOL(ldlm_cli_enqueue);
EXPORT_SYMBOL(ldlm_cli_cancel);
-EXPORT_SYMBOL(lustre_handle2object);
EXPORT_SYMBOL(ldlm_test);
EXPORT_SYMBOL(ldlm_lock_dump);
EXPORT_SYMBOL(ldlm_namespace_new);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff