X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Flinkea.c;h=cf17a50999f8df898d750a5c5c343c0e43e59b1d;hb=8e498d3f23ea9bcbef524153c6613f93a6229431;hp=eed6fd21dad1a18ef0bb6c1d213d12ec1a85d124;hpb=1b2547843817b4b7adbeb87ea9b070d9cac35c90;p=fs%2Flustre-release.git

diff --git a/lustre/obdclass/linkea.c b/lustre/obdclass/linkea.c
index eed6fd2..cf17a50 100644
--- a/lustre/obdclass/linkea.c
+++ b/lustre/obdclass/linkea.c
@@ -21,25 +21,26 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2013, Intel Corporation.
+ * Copyright (c) 2013, 2017, Intel Corporation.
  * Use is subject to license terms.
  *
  * Author: Di Wang <di.wang@intel.com>
  */
 
-#include <lustre/lustre_idl.h>
 #include <obd.h>
 #include <lustre_linkea.h>
 
 int linkea_data_new(struct linkea_data *ldata, struct lu_buf *buf)
 {
-	ldata->ld_buf = lu_buf_check_and_alloc(buf, PAGE_CACHE_SIZE);
+	ldata->ld_buf = lu_buf_check_and_alloc(buf, PAGE_SIZE);
 	if (ldata->ld_buf->lb_buf == NULL)
 		return -ENOMEM;
 	ldata->ld_leh = ldata->ld_buf->lb_buf;
 	ldata->ld_leh->leh_magic = LINK_EA_MAGIC;
-	ldata->ld_leh->leh_len = sizeof(struct link_ea_header);
 	ldata->ld_leh->leh_reccount = 0;
+	ldata->ld_leh->leh_len = sizeof(struct link_ea_header);
+	ldata->ld_leh->leh_overflow_time = 0;
+	ldata->ld_leh->leh_padding = 0;
 	return 0;
 }
 EXPORT_SYMBOL(linkea_data_new);
@@ -54,11 +55,15 @@ int linkea_init(struct linkea_data *ldata)
 		leh->leh_magic = LINK_EA_MAGIC;
 		leh->leh_reccount = __swab32(leh->leh_reccount);
 		leh->leh_len = __swab64(leh->leh_len);
-		/* entries are swabbed by linkea_entry_unpack */
+		leh->leh_overflow_time = __swab32(leh->leh_overflow_time);
+		leh->leh_padding = __swab32(leh->leh_padding);
+		/* individual entries are swabbed by linkea_entry_unpack() */
 	}
+
 	if (leh->leh_magic != LINK_EA_MAGIC)
 		return -EINVAL;
-	if (leh->leh_reccount == 0)
+
+	if (leh->leh_reccount == 0 && leh->leh_overflow_time == 0)
 		return -ENODATA;
 
 	ldata->ld_leh = leh;
@@ -66,22 +71,36 @@ int linkea_init(struct linkea_data *ldata)
 }
 EXPORT_SYMBOL(linkea_init);
 
+int linkea_init_with_rec(struct linkea_data *ldata)
+{
+	int rc;
+
+	rc = linkea_init(ldata);
+	if (!rc && ldata->ld_leh->leh_reccount == 0)
+		rc = -ENODATA;
+
+	return rc;
+}
+EXPORT_SYMBOL(linkea_init_with_rec);
+
 /**
  * Pack a link_ea_entry.
  * All elements are stored as chars to avoid alignment issues.
  * Numbers are always big-endian
  * \retval record length
  */
-static int linkea_entry_pack(struct link_ea_entry *lee,
-			     const struct lu_name *lname,
-			     const struct lu_fid *pfid)
+int linkea_entry_pack(struct link_ea_entry *lee, const struct lu_name *lname,
+		      const struct lu_fid *pfid)
 {
 	struct lu_fid   tmpfid;
 	int             reclen;
 
-	fid_cpu_to_be(&tmpfid, pfid);
+	tmpfid = *pfid;
+	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MUL_REF))
+		tmpfid.f_oid--;
 	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_CRASH))
 		tmpfid.f_ver = ~0;
+	fid_cpu_to_be(&tmpfid, &tmpfid);
 	memcpy(&lee->lee_parent_fid, &tmpfid, sizeof(tmpfid));
 	memcpy(lee->lee_name, lname->ln_name, lname->ln_namelen);
 	reclen = sizeof(struct link_ea_entry) + lname->ln_namelen;
@@ -90,15 +109,20 @@ static int linkea_entry_pack(struct link_ea_entry *lee,
 	lee->lee_reclen[1] = reclen & 0xff;
 	return reclen;
 }
+EXPORT_SYMBOL(linkea_entry_pack);
 
 void linkea_entry_unpack(const struct link_ea_entry *lee, int *reclen,
 			 struct lu_name *lname, struct lu_fid *pfid)
 {
+	LASSERT(lee != NULL);
+
 	*reclen = (lee->lee_reclen[0] << 8) | lee->lee_reclen[1];
 	memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
 	fid_be_to_cpu(pfid, pfid);
-	lname->ln_name = lee->lee_name;
-	lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
+	if (lname != NULL) {
+		lname->ln_name = lee->lee_name;
+		lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
+	}
 }
 EXPORT_SYMBOL(linkea_entry_unpack);
 
@@ -108,27 +132,45 @@ EXPORT_SYMBOL(linkea_entry_unpack);
 int linkea_add_buf(struct linkea_data *ldata, const struct lu_name *lname,
 		   const struct lu_fid *pfid)
 {
-	LASSERT(ldata->ld_leh != NULL);
+	struct link_ea_header *leh = ldata->ld_leh;
+	int reclen;
+
+	LASSERT(leh != NULL);
 
 	if (lname == NULL || pfid == NULL)
 		return -EINVAL;
 
-	ldata->ld_reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
-	if (ldata->ld_leh->leh_len + ldata->ld_reclen >
-	    ldata->ld_buf->lb_len) {
+	reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
+	if (unlikely(leh->leh_len + reclen > MAX_LINKEA_SIZE)) {
+		/* Use 32-bits to save the overflow time, although it will
+		 * shrink the ktime_get_real_seconds() returned 64-bits value
+		 * to 32-bits value, it is still quite large and can be used
+		 * for about 140 years. That is enough.
+		 */
+		leh->leh_overflow_time = ktime_get_real_seconds();
+		if (unlikely(leh->leh_overflow_time == 0))
+			leh->leh_overflow_time++;
+
+		CDEBUG(D_INODE, "No enough space to hold linkea entry '"
+		       DFID": %.*s' at %u\n", PFID(pfid), lname->ln_namelen,
+		       lname->ln_name, leh->leh_overflow_time);
+		return 0;
+	}
+
+	if (leh->leh_len + reclen > ldata->ld_buf->lb_len) {
 		if (lu_buf_check_and_grow(ldata->ld_buf,
-					  ldata->ld_leh->leh_len +
-					  ldata->ld_reclen) < 0)
+					  leh->leh_len + reclen) < 0)
 			return -ENOMEM;
+
+		leh = ldata->ld_leh = ldata->ld_buf->lb_buf;
 	}
 
-	ldata->ld_leh = ldata->ld_buf->lb_buf;
-	ldata->ld_lee = ldata->ld_buf->lb_buf + ldata->ld_leh->leh_len;
+	ldata->ld_lee = ldata->ld_buf->lb_buf + leh->leh_len;
 	ldata->ld_reclen = linkea_entry_pack(ldata->ld_lee, lname, pfid);
-	ldata->ld_leh->leh_len += ldata->ld_reclen;
-	ldata->ld_leh->leh_reccount++;
-	CDEBUG(D_INODE, "New link_ea name '%.*s' is added\n",
-	       lname->ln_namelen, lname->ln_name);
+	leh->leh_len += ldata->ld_reclen;
+	leh->leh_reccount++;
+	CDEBUG(D_INODE, "New link_ea name '"DFID":%.*s' is added\n",
+	       PFID(pfid), lname->ln_namelen, lname->ln_name);
 	return 0;
 }
 EXPORT_SYMBOL(linkea_add_buf);
@@ -137,6 +179,7 @@ EXPORT_SYMBOL(linkea_add_buf);
 void linkea_del_buf(struct linkea_data *ldata, const struct lu_name *lname)
 {
 	LASSERT(ldata->ld_leh != NULL && ldata->ld_lee != NULL);
+	LASSERT(ldata->ld_leh->leh_reccount > 0);
 
 	ldata->ld_leh->leh_reccount--;
 	ldata->ld_leh->leh_len -= ldata->ld_reclen;
@@ -152,6 +195,70 @@ void linkea_del_buf(struct linkea_data *ldata, const struct lu_name *lname)
 }
 EXPORT_SYMBOL(linkea_del_buf);
 
+int linkea_links_new(struct linkea_data *ldata, struct lu_buf *buf,
+		     const struct lu_name *cname, const struct lu_fid *pfid)
+{
+	int rc;
+
+	rc = linkea_data_new(ldata, buf);
+	if (!rc)
+		rc = linkea_add_buf(ldata, cname, pfid);
+
+	return rc;
+}
+EXPORT_SYMBOL(linkea_links_new);
+
+/**
+ * Mark the linkEA as overflow with current timestamp,
+ * and remove the last linkEA entry.
+ *
+ * Return the new linkEA size.
+ */
+int linkea_overflow_shrink(struct linkea_data *ldata)
+{
+	struct link_ea_header *leh;
+	struct lu_name tname;
+	struct lu_fid tfid;
+	int count;
+
+	leh = ldata->ld_leh = ldata->ld_buf->lb_buf;
+	if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
+		leh->leh_magic = LINK_EA_MAGIC;
+		leh->leh_reccount = __swab32(leh->leh_reccount);
+		leh->leh_overflow_time = __swab32(leh->leh_overflow_time);
+		leh->leh_padding = __swab32(leh->leh_padding);
+	}
+
+	LASSERT(leh->leh_reccount > 0);
+
+	leh->leh_len = sizeof(struct link_ea_header);
+	leh->leh_reccount--;
+	if (unlikely(leh->leh_reccount == 0))
+		return 0;
+
+	leh->leh_overflow_time = ktime_get_real_seconds();
+	if (unlikely(leh->leh_overflow_time == 0))
+		leh->leh_overflow_time++;
+	ldata->ld_reclen = 0;
+	ldata->ld_lee = (struct link_ea_entry *)(leh + 1);
+	for (count = 0; count < leh->leh_reccount; count++) {
+		linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
+				    &tname, &tfid);
+		leh->leh_len += ldata->ld_reclen;
+		ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
+							 ldata->ld_reclen);
+	}
+
+	linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &tname, &tfid);
+	CDEBUG(D_INODE, "No enough space to hold the last linkea entry '"
+	       DFID": %.*s', shrink it, left %d linkea entries, size %llu\n",
+	       PFID(&tfid), tname.ln_namelen, tname.ln_name,
+	       leh->leh_reccount, leh->leh_len);
+
+	return leh->leh_len;
+}
+EXPORT_SYMBOL(linkea_overflow_shrink);
+
 /**
  * Check if such a link exists in linkEA.
  *
@@ -172,8 +279,9 @@ int linkea_links_find(struct linkea_data *ldata, const struct lu_name *lname,
 
 	LASSERT(ldata->ld_leh != NULL);
 
-	/* link #0 */
-	ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
+	/* link #0, if leh_reccount == 0 we skip the loop and return -ENOENT */
+	if (likely(ldata->ld_leh->leh_reccount > 0))
+		ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
 
 	for (count = 0; count < ldata->ld_leh->leh_reccount; count++) {
 		linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,