X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ftarget%2Ftgt_lastrcvd.c;h=55e59959fab633f02ab6800ed845865f72eb1026;hp=1c8f2068a270ac388a1c7357cac013b32f94bf26;hb=9197849bc761f7fbbdee65916ad1346eec79d412;hpb=70bb27b746c1e80196815e335c08bdc113ef052a;ds=sidebyside

diff --git a/lustre/target/tgt_lastrcvd.c b/lustre/target/tgt_lastrcvd.c
index 1c8f206..55e5995 100644
--- a/lustre/target/tgt_lastrcvd.c
+++ b/lustre/target/tgt_lastrcvd.c
@@ -50,7 +50,8 @@ static int tgt_bitmap_chunk_alloc(struct lu_target *lut, int chunk)
 {
 	unsigned long *bm;
 
-	OBD_ALLOC(bm, BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) * sizeof(long));
+	OBD_ALLOC_LARGE(bm, BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
+			sizeof(long));
 	if (bm == NULL)
 		return -ENOMEM;
 
@@ -59,7 +60,7 @@ static int tgt_bitmap_chunk_alloc(struct lu_target *lut, int chunk)
 	if (lut->lut_reply_bitmap[chunk] != NULL) {
 		/* someone else already allocated the bitmap for this chunk */
 		spin_unlock(&lut->lut_client_bitmap_lock);
-		OBD_FREE(bm, BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
+		OBD_FREE_LARGE(bm, BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
 			 sizeof(long));
 		return 0;
 	}
@@ -155,6 +156,12 @@ static int tgt_clear_reply_slot(struct lu_target *lut, int idx)
 	LASSERT(chunk < LUT_REPLY_SLOTS_MAX_CHUNKS);
 	LASSERT(b < LUT_REPLY_SLOTS_PER_CHUNK);
 
+	if (lut->lut_reply_bitmap[chunk] == NULL) {
+		CERROR("%s: slot %d not allocated\n",
+		       tgt_name(lut), idx);
+		return -ENOENT;
+	}
+
 	if (test_and_clear_bit(b, lut->lut_reply_bitmap[chunk]) == 0) {
 		CERROR("%s: slot %d already clear in bitmap\n",
 		       tgt_name(lut), idx);
@@ -378,6 +385,9 @@ int tgt_client_alloc(struct obd_export *exp)
 	ENTRY;
 	LASSERT(exp != exp->exp_obd->obd_self_export);
 
+	spin_lock_init(&exp->exp_target_data.ted_nodemap_lock);
+	INIT_LIST_HEAD(&exp->exp_target_data.ted_nodemap_member);
+
 	OBD_ALLOC_PTR(exp->exp_target_data.ted_lcd);
 	if (exp->exp_target_data.ted_lcd == NULL)
 		RETURN(-ENOMEM);
@@ -507,6 +517,7 @@ static int tgt_client_data_update(const struct lu_env *env,
 		RETURN(PTR_ERR(th));
 
 	tti_buf_lcd(tti);
+	mutex_lock(&ted->ted_lcd_lock);
 	rc = dt_declare_record_write(env, tgt->lut_last_rcvd,
 				     &tti->tti_buf,
 				     ted->ted_lr_off, th);
@@ -536,6 +547,7 @@ static int tgt_client_data_update(const struct lu_env *env,
 	rc = tgt_client_data_write(env, tgt, ted->ted_lcd, &tti->tti_off, th);
 	EXIT;
 out:
+	mutex_unlock(&ted->ted_lcd_lock);
 	dt_trans_stop(env, tgt->lut_bottom, th);
 	CDEBUG(D_INFO, "%s: update last_rcvd client data for UUID = %s, "
 	       "last_transno = "LPU64": rc = %d\n", tgt->lut_obd->obd_name,
@@ -781,7 +793,9 @@ static void tgt_cb_last_committed(struct lu_env *env, struct thandle *th,
 	if (ccb->llcc_transno > ccb->llcc_exp->exp_last_committed) {
 		ccb->llcc_exp->exp_last_committed = ccb->llcc_transno;
 		spin_unlock(&ccb->llcc_tgt->lut_translock);
+
 		ptlrpc_commit_replies(ccb->llcc_exp);
+		tgt_cancel_slc_locks(ccb->llcc_transno);
 	} else {
 		spin_unlock(&ccb->llcc_tgt->lut_translock);
 	}
@@ -1057,10 +1071,8 @@ int tgt_client_del(const struct lu_env *env, struct obd_export *exp)
 		RETURN(rc);
 	}
 
-	mutex_lock(&ted->ted_lcd_lock);
 	memset(ted->ted_lcd->lcd_uuid, 0, sizeof ted->ted_lcd->lcd_uuid);
 	rc = tgt_client_data_update(env, exp);
-	mutex_unlock(&ted->ted_lcd_lock);
 
 	CDEBUG(rc == 0 ? D_INFO : D_ERROR,
 	       "%s: zeroing out client %s at idx %u (%llu), rc %d\n",
@@ -1712,23 +1724,16 @@ int tgt_txn_start_cb(const struct lu_env *env, struct thandle *th,
 	if (tsi->tsi_exp == NULL)
 		return 0;
 
-	dto = dt_object_locate(tgt->lut_last_rcvd, th->th_dev);
-	tti_buf_lcd(tti);
-
-	rc = dt_declare_record_write(env, dto, &tti->tti_buf,
-				     tsi->tsi_exp->exp_target_data.ted_lr_off,
-				     th);
-	if (rc)
-		return rc;
-
-	tti_buf_lsd(tti);
-	rc = dt_declare_record_write(env, dto, &tti->tti_buf, 0, th);
-	if (rc)
-		return rc;
-
 	if (tgt_is_multimodrpcs_client(tsi->tsi_exp)) {
-		tti->tti_off = atomic_read(&tgt->lut_num_clients) * 8
-				* sizeof(struct lsd_reply_data);
+		/*
+		 * Use maximum possible file offset for declaration to ensure
+		 * ZFS will reserve enough credits for a write anywhere in this
+		 * file, since we don't know where in the file the write will be
+		 * because a replay slot has not been assigned.  This should be
+		 * replaced by dmu_tx_hold_append() when available.
+		 */
+		tti->tti_off = atomic_read(&tgt->lut_num_clients) * 8 *
+				sizeof(struct lsd_reply_data);
 		tti->tti_buf.lb_buf = NULL;
 		tti->tti_buf.lb_len = sizeof(struct lsd_reply_data);
 		dto = dt_object_locate(tgt->lut_reply_data, th->th_dev);
@@ -1736,6 +1741,14 @@ int tgt_txn_start_cb(const struct lu_env *env, struct thandle *th,
 					     tti->tti_off, th);
 		if (rc)
 			return rc;
+	} else {
+		dto = dt_object_locate(tgt->lut_last_rcvd, th->th_dev);
+		tti_buf_lcd(tti);
+		tti->tti_off = tsi->tsi_exp->exp_target_data.ted_lr_off;
+		rc = dt_declare_record_write(env, dto, &tti->tti_buf,
+					     tti->tti_off, th);
+		if (rc)
+			return rc;
 	}
 
 	if (tsi->tsi_vbr_obj != NULL &&
@@ -1879,7 +1892,11 @@ int tgt_reply_data_init(const struct lu_env *env, struct lu_target *tgt)
 
 			/* create in-memory reply_data and link it to
 			 * target export's reply list */
-			tgt_set_reply_slot(tgt, idx);
+			rc = tgt_set_reply_slot(tgt, idx);
+			if (rc != 0) {
+				mutex_unlock(&ted->ted_lcd_lock);
+				GOTO(out, rc);
+			}
 			trd->trd_reply = *lrd;
 			trd->trd_pre_versions[0] = 0;
 			trd->trd_pre_versions[1] = 0;