blob: fix issue with blobid handling

Fixes github issue #29.

Because of how we handle the blobid and pagenum in blobstore,
it was possible to have blobstore inadvertently open the wrong
blob if open is provided a blobid where the lower 32 bits match
an existing blob but the upper 32 are clear.

Patch does the following:
- removes assert() that caught this on MD load and replace with
an error given that this condition can be induced via the API
- cleanup of pagenum and blobid conversion/handling to make it
clearer how they're related and converted
- new UTs that would have failed w/o the new check in place

Change-Id: I2b49b237922b3b8cfc4df296f5bc20195e41dc41
Signed-off-by: Paul Luse <paul.e.luse@intel.com>
Reviewed-on: https://review.gerrithub.io/380872
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Daniel Verkamp <daniel.verkamp@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>

lib/blob/blobstore.c

@@ -270,9 +270,17 @@ _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count,
 	assert(blob != NULL);
 	assert(blob->state == SPDK_BLOB_STATE_LOADING);
 	assert(blob->active.clusters == NULL);
-	assert(blob->id == pages[0].id);
 	assert(blob->state == SPDK_BLOB_STATE_LOADING);
 
+	/* The blobid provided doesn't match what's in the MD, this can
+	 * happen for example if a bogus blobid is passed in through open.
+	 */
+	if (blob->id != pages[0].id) {
+		SPDK_ERRLOG("Blobid (%lu) doesn't match what's in metadata (%lu)\n",
+			    blob->id, pages[0].id);
+		return -ENOENT;
+	}
+
 	for (i = 0; i < page_count; i++) {
 		page = &pages[i];
 
@@ -579,6 +587,13 @@ _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno)
 
 	/* Parse the pages */
 	rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob);
+	if (rc) {
+		_spdk_blob_free(blob);
+		ctx->cb_fn(seq, NULL, rc);
+		spdk_dma_free(ctx->pages);
+		free(ctx);
+		return;
+	}
 
 	_spdk_blob_mark_clean(blob);
 
@@ -2131,11 +2146,7 @@ void spdk_bs_md_create_blob(struct spdk_blob_store *bs,
 	}
 	spdk_bit_array_set(bs->used_md_pages, page_idx);
 
-	/* The blob id is a 64 bit number. The lower 32 bits are the page_idx. The upper
-	 * 32 bits are not currently used. Stick a 1 there just to catch bugs where the
-	 * code assumes blob id == page_idx.
-	 */
-	id = (1ULL << 32) | page_idx;
+	id = _spdk_bs_page_to_blobid(page_idx);
 
 	SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx);
 
@@ -2496,7 +2507,7 @@ _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno)
 		return;
 	}
 
-	id = (1ULL << 32) | ctx->page_num;
+	id = _spdk_bs_page_to_blobid(ctx->page_num);
 
 	blob = _spdk_blob_lookup(bs, id);
 	if (blob) {

lib/blob/blobstore.h

@@ -50,6 +50,7 @@
 #define SPDK_BLOB_OPTS_NUM_MD_PAGES UINT32_MAX
 #define SPDK_BLOB_OPTS_MAX_MD_OPS 32
 #define SPDK_BLOB_OPTS_MAX_CHANNEL_OPS 512
+#define SPDK_BLOB_BLOBID_HIGH_BIT (1ULL << 32)
 
 struct spdk_xattr {
 	/* TODO: reorder for best packing */
@@ -350,6 +351,16 @@ _spdk_bs_blobid_to_page(spdk_blob_id id)
 	return id & 0xFFFFFFFF;
 }
 
+/* The blob id is a 64 bit number. The lower 32 bits are the page_idx. The upper
+ * 32 bits are not currently used. Stick a 1 there just to catch bugs where the
+ * code assumes blob id == page_idx.
+ */
+static inline spdk_blob_id
+_spdk_bs_page_to_blobid(uint32_t page_idx)
+{
+	return SPDK_BLOB_BLOBID_HIGH_BIT | page_idx;
+}
+
 /* Given a page offset into a blob, look up the LBA for the
  * start of that page.
  */

test/unit/lib/blob/blob.c/blob_ut.c

@@ -855,6 +855,11 @@ bs_load(void)
 	CU_ASSERT(g_bserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_bs != NULL);
 
+	/* Try to open a blobid that does not exist */
+	spdk_bs_md_open_blob(g_bs, 0, blob_op_with_handle_complete, NULL);
+	CU_ASSERT(g_bserrno == -ENOENT);
+	CU_ASSERT(g_blob == NULL);
+
 	/* Create a blob */
 	spdk_bs_md_create_blob(g_bs, blob_op_with_id_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
@@ -866,6 +871,11 @@ bs_load(void)
 	CU_ASSERT(g_blob != NULL);
 	blob = g_blob;
 
+	/* Try again to open valid blob but without the upper bit set */
+	spdk_bs_md_open_blob(g_bs, blobid & 0xFFFFFFFF, blob_op_with_handle_complete, NULL);
+	CU_ASSERT(g_bserrno == -ENOENT);
+	CU_ASSERT(g_blob == NULL);
+
 	/* Set some xattrs */
 	rc = spdk_blob_md_set_xattr(blob, "name", "log.txt", strlen("log.txt") + 1);
 	CU_ASSERT(rc == 0);