* GPL HEADER END
*/
/*
- * Copyright (c) 2012, 2014, Intel Corporation.
+ * Copyright (c) 2012, 2015, Intel Corporation.
*/
/*
* lustre/target/tgt_main.c
#include "tgt_internal.h"
#include "../ptlrpc/ptlrpc_internal.h"
+static spinlock_t uncommitted_slc_locks_guard;
+static struct list_head uncommitted_slc_locks;
+
+/*
+ * Save cross-MDT lock in uncommitted_slc_locks.
+ *
+ * Lock R/W count is not saved, but released in unlock (not canceled remotely),
+ * instead only a refcount is taken, so that the remote MDT where the object
+ * resides can detect conflict with this lock there.
+ *
+ * \param lock cross-MDT lock to save
+ * \param transno when the transaction with this transno is committed, this lock
+ * can be canceled.
+ */
+void tgt_save_slc_lock(struct ldlm_lock *lock, __u64 transno)
+{
+ spin_lock(&uncommitted_slc_locks_guard);
+ lock_res_and_lock(lock);
+ if (ldlm_is_cbpending(lock)) {
+ /* if it was canceld by server, don't save, because remote MDT
+ * will do Sync-on-Cancel. */
+ LDLM_LOCK_PUT(lock);
+ } else {
+ lock->l_transno = transno;
+ /* if this lock is in the list already, there are two operations
+ * both use this lock, and save it after use, so for the second
+ * one, just put the refcount. */
+ if (list_empty(&lock->l_slc_link))
+ list_add_tail(&lock->l_slc_link,
+ &uncommitted_slc_locks);
+ else
+ LDLM_LOCK_PUT(lock);
+ }
+ unlock_res_and_lock(lock);
+ spin_unlock(&uncommitted_slc_locks_guard);
+}
+EXPORT_SYMBOL(tgt_save_slc_lock);
+
+/*
+ * Discard cross-MDT lock from uncommitted_slc_locks.
+ *
+ * This is called upon BAST, just remove lock from uncommitted_slc_locks and put
+ * lock refcount. The BAST will cancel this lock.
+ *
+ * \param lock cross-MDT lock to discard
+ */
+void tgt_discard_slc_lock(struct ldlm_lock *lock)
+{
+ spin_lock(&uncommitted_slc_locks_guard);
+ lock_res_and_lock(lock);
+ /* may race with tgt_cancel_slc_locks() */
+ if (lock->l_transno != 0) {
+ LASSERT(!list_empty(&lock->l_slc_link));
+ LASSERT(ldlm_is_cbpending(lock));
+ list_del_init(&lock->l_slc_link);
+ lock->l_transno = 0;
+ LDLM_LOCK_PUT(lock);
+ }
+ unlock_res_and_lock(lock);
+ spin_unlock(&uncommitted_slc_locks_guard);
+}
+EXPORT_SYMBOL(tgt_discard_slc_lock);
+
+/*
+ * Cancel cross-MDT locks upon transaction commit.
+ *
+ * Remove cross-MDT locks from uncommitted_slc_locks, cancel them and put lock
+ * refcount.
+ *
+ * \param transno transaction with this number was committed.
+ */
+void tgt_cancel_slc_locks(__u64 transno)
+{
+ struct ldlm_lock *lock, *next;
+ LIST_HEAD(list);
+ struct lustre_handle lockh;
+
+ spin_lock(&uncommitted_slc_locks_guard);
+ list_for_each_entry_safe(lock, next, &uncommitted_slc_locks,
+ l_slc_link) {
+ lock_res_and_lock(lock);
+ LASSERT(lock->l_transno != 0);
+ if (lock->l_transno > transno) {
+ unlock_res_and_lock(lock);
+ continue;
+ }
+ /* ouch, another operation is using it after it's saved */
+ if (lock->l_readers != 0 || lock->l_writers != 0) {
+ unlock_res_and_lock(lock);
+ continue;
+ }
+ /* set CBPENDING so that this lock won't be used again */
+ ldlm_set_cbpending(lock);
+ lock->l_transno = 0;
+ list_move(&lock->l_slc_link, &list);
+ unlock_res_and_lock(lock);
+ }
+ spin_unlock(&uncommitted_slc_locks_guard);
+
+ list_for_each_entry_safe(lock, next, &list, l_slc_link) {
+ list_del_init(&lock->l_slc_link);
+ ldlm_lock2handle(lock, &lockh);
+ ldlm_cli_cancel(&lockh, LCF_ASYNC);
+ LDLM_LOCK_PUT(lock);
+ }
+}
+
int tgt_init(const struct lu_env *env, struct lu_target *lut,
struct obd_device *obd, struct dt_device *dt,
struct tgt_opc_slice *slice, int request_fail_id,
struct lu_attr attr;
struct lu_fid fid;
struct dt_object *o;
- int rc = 0;
+ int i, rc = 0;
ENTRY;
RETURN(0);
spin_lock_init(&lut->lut_translock);
+ spin_lock_init(&lut->lut_client_bitmap_lock);
OBD_ALLOC(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
if (lut->lut_client_bitmap == NULL)
rc = PTR_ERR(o);
CERROR("%s: cannot open LAST_RCVD: rc = %d\n", tgt_name(lut),
rc);
- GOTO(out, rc);
+ GOTO(out_put, rc);
}
lut->lut_last_rcvd = o;
rc = tgt_server_data_init(env, lut);
if (rc < 0)
- GOTO(out, rc);
+ GOTO(out_put, rc);
/* prepare transactions callbacks */
lut->lut_txn_cb.dtc_txn_start = tgt_txn_start_cb;
OBD_ALLOC(lut->lut_reply_bitmap,
LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));
if (lut->lut_reply_bitmap == NULL)
- GOTO(out, rc);
+ GOTO(out, rc = -ENOMEM);
memset(&attr, 0, sizeof(attr));
attr.la_valid = LA_MODE;
if (rc < 0)
GOTO(out, rc);
+ atomic_set(&lut->lut_sync_count, 0);
+
RETURN(0);
+
out:
+ dt_txn_callback_del(lut->lut_bottom, &lut->lut_txn_cb);
+out_put:
+ obd->u.obt.obt_magic = 0;
+ obd->u.obt.obt_lut = NULL;
if (lut->lut_last_rcvd != NULL) {
lu_object_put(env, &lut->lut_last_rcvd->do_lu);
- dt_txn_callback_del(lut->lut_bottom, &lut->lut_txn_cb);
+ lut->lut_last_rcvd = NULL;
}
- lut->lut_last_rcvd = NULL;
if (lut->lut_client_bitmap != NULL)
OBD_FREE(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
lut->lut_client_bitmap = NULL;
if (lut->lut_reply_data != NULL)
lu_object_put(env, &lut->lut_reply_data->do_lu);
lut->lut_reply_data = NULL;
- if (lut->lut_reply_bitmap != NULL)
+ if (lut->lut_reply_bitmap != NULL) {
+ for (i = 0; i < LUT_REPLY_SLOTS_MAX_CHUNKS; i++) {
+ if (lut->lut_reply_bitmap[i] != NULL)
+ OBD_FREE_LARGE(lut->lut_reply_bitmap[i],
+ BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
+ sizeof(long));
+ lut->lut_reply_bitmap[i] = NULL;
+ }
OBD_FREE(lut->lut_reply_bitmap,
LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));
+ }
lut->lut_reply_bitmap = NULL;
return rc;
}
if (lut->lut_reply_bitmap != NULL) {
for (i = 0; i < LUT_REPLY_SLOTS_MAX_CHUNKS; i++) {
if (lut->lut_reply_bitmap[i] != NULL)
- OBD_FREE(lut->lut_reply_bitmap[i],
+ OBD_FREE_LARGE(lut->lut_reply_bitmap[i],
BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
sizeof(long));
lut->lut_reply_bitmap[i] = NULL;
{
ENTRY;
- tgt_page_to_corrupt = alloc_page(GFP_IOFS);
+ tgt_page_to_corrupt = alloc_page(GFP_KERNEL);
tgt_key_init_generic(&tgt_thread_key, NULL);
lu_context_key_register_many(&tgt_thread_key, NULL);
update_info_init();
+ spin_lock_init(&uncommitted_slc_locks_guard);
+ INIT_LIST_HEAD(&uncommitted_slc_locks);
+
RETURN(0);
}
void tgt_mod_exit(void)
{
if (tgt_page_to_corrupt != NULL)
- page_cache_release(tgt_page_to_corrupt);
+ put_page(tgt_page_to_corrupt);
lu_context_key_degister(&tgt_thread_key);
lu_context_key_degister(&tgt_session_key);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff