X-Git-Url: http://git.tuebingen.mpg.de/?a=blobdiff_plain;ds=sidebyside;f=osl.c;h=1587b04c5fa06201ce5e32c024c7a8e189a24b8b;hb=a3b43ea4ccc39b56ec4de003f83836b75905afe2;hp=1fc45bd4367506943a99e9789679ee1d93409ca6;hpb=f6f50d03a09d6bc423324206d274336e9905bbb4;p=paraslash.git

diff --git a/osl.c b/osl.c
index 1fc45bd4..1587b04c 100644
--- a/osl.c
+++ b/osl.c
@@ -84,11 +84,11 @@ ssize_t para_write(int fd, const void *buf, size_t size)
  */
 ssize_t para_write_all(int fd, const void *buf, size_t size)
 {
-	PARA_DEBUG_LOG("writing %zu bytes\n", size);
+//	PARA_DEBUG_LOG("writing %zu bytes\n", size);
 	const char *b = buf;
 	while (size) {
 		ssize_t ret = para_write(fd, b, size);
-		PARA_DEBUG_LOG("ret: %d\n", ret);
+//		PARA_DEBUG_LOG("ret: %zd\n", ret);
 		if (ret < 0)
 			return ret;
 		b += ret;
@@ -170,7 +170,7 @@ static int append_file(const char *filename, char *header, size_t header_size,
 {
 	int ret, fd;
 
-	PARA_DEBUG_LOG("appending %lu  + %ld bytes\n", header_size, data_size);
+//	PARA_DEBUG_LOG("appending %zu  + %zu bytes\n", header_size, data_size);
 	ret = para_open(filename, O_WRONLY | O_CREAT | O_APPEND, 0644);
 	if (ret < 0)
 		return ret;
@@ -493,10 +493,10 @@ int init_table_structure(const struct osl_table_description *desc,
 	struct osl_table *t = para_calloc(sizeof(*t));
 	int i, ret = -E_BAD_TABLE_DESC, have_disk_storage_name_column = 0;
 
-	PARA_INFO_LOG("creating table structure for '%s' from table "
-		"description\n", desc->name);
 	if (!desc)
 		goto err;
+	PARA_DEBUG_LOG("creating table structure for '%s' from table "
+		"description\n", desc->name);
 	ret = -E_NO_COLUMN_DESC;
 	if (!desc->column_descriptions)
 		goto err;
@@ -526,8 +526,8 @@ int init_table_structure(const struct osl_table_description *desc,
 			break;
 		case OSL_MAPPED_STORAGE:
 			t->num_mapped_columns++;
-			col->index_offset = t->index_entry_size;
-			t->index_entry_size += 8;
+			col->index_offset = t->row_index_size;
+			t->row_index_size += 8;
 			break;
 		case OSL_NO_STORAGE:
 			col->volatile_num = t->num_volatile_columns;
@@ -551,7 +551,7 @@ int init_table_structure(const struct osl_table_description *desc,
 	if (!t->num_rbtrees)
 		goto err;
 	/* success */
-	PARA_INFO_LOG("OK. Index entry size: %u\n", t->index_entry_size);
+	PARA_DEBUG_LOG("OK. Index entry size: %u\n", t->row_index_size);
 	ret = init_column_descriptions(t);
 	if (ret < 0)
 		goto err;
@@ -609,7 +609,7 @@ int read_table_desc(struct osl_object *map, struct osl_table_description *desc)
 
 		ret = -E_SHORT_TABLE;
 		if (map->size < offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE) {
-			PARA_ERROR_LOG("map size = %u < %u = offset + min desc size\n",
+			PARA_ERROR_LOG("map size = %zu < %u = offset + min desc size\n",
 				map->size, offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE);
 			goto err;
 		}
@@ -677,7 +677,7 @@ static int compare_table_descriptions(struct osl_table *t)
 		if (strcmp(cd1->name, cd2->name))
 			goto out;
 	}
-	PARA_INFO_LOG("table description of '%s' matches on-disk data, good\n",
+	PARA_DEBUG_LOG("table description of '%s' matches on-disk data, good\n",
 		t->desc->name);
 	ret = 1;
 out:
@@ -806,6 +806,17 @@ static void mark_table_clean(struct osl_table *t)
 	write_u8(buf, read_u8(buf) & 0xfe);
 }
 
+static void unmap_column(struct osl_table *t, unsigned col_num)
+{
+	struct osl_object map = t->columns[col_num].data_map;
+	int ret;
+	if (!map.data)
+		return;
+	ret = para_munmap(map.data, map.size);
+	assert(ret > 0);
+	map.data = NULL;
+}
+
 /**
  * Unmap all mapped files of an osl table.
  *
@@ -825,7 +836,7 @@ int unmap_table(struct osl_table *t, enum osl_close_flags flags)
 
 	if (!t->num_mapped_columns) /* can this ever happen? */
 		return 1;
-	PARA_INFO_LOG("unmapping table '%s'\n", t->desc->name);
+	PARA_DEBUG_LOG("unmapping table '%s'\n", t->desc->name);
 	if (!t->index_map.data)
 		return -E_NOT_MAPPED;
 	if (flags & OSL_MARK_CLEAN)
@@ -836,18 +847,30 @@ int unmap_table(struct osl_table *t, enum osl_close_flags flags)
 	t->index_map.data = NULL;
 	if (!t->num_rows)
 		return 1;
-	FOR_EACH_MAPPED_COLUMN(i, t, cd) {
-		struct osl_object map = t->columns[i].data_map;
-		if (!map.data)
-			continue;
-		ret = para_munmap(map.data, map.size);
-		if (ret < 0)
-			return ret;
-		map.data = NULL;
-	}
+	FOR_EACH_MAPPED_COLUMN(i, t, cd)
+		unmap_column(t, i);
 	return 1;
 }
 
+static int map_column(struct osl_table *t, unsigned col_num)
+{
+	struct stat statbuf;
+	char *filename = column_filename(t, col_num);
+	int ret = -E_STAT;
+	if (stat(filename, &statbuf) < 0) {
+		free(filename);
+		return ret;
+	}
+	if (!(S_IFREG & statbuf.st_mode)) {
+		free(filename);
+		return ret;
+	}
+	ret = mmap_full_file(filename, O_RDWR,
+		&t->columns[col_num].data_map);
+	free(filename);
+	return ret;
+}
+
 /**
  * Map the index file and all columns of type \p OSL_MAPPED_STORAGE into memory.
  *
@@ -894,20 +917,7 @@ int map_table(struct osl_table *t, enum map_table_flags flags)
 		return num_rows;
 	/* map data files */
 	FOR_EACH_MAPPED_COLUMN(i, t, cd) {
-		struct stat statbuf;
-		filename = column_filename(t, i);
-		ret = -E_STAT;
-		if (stat(filename, &statbuf) < 0) {
-			free(filename);
-			goto err;
-		}
-		if (!(S_IFREG & statbuf.st_mode)) {
-			free(filename);
-			goto err;
-		}
-		ret = mmap_full_file(filename, O_RDWR,
-			&t->columns[i].data_map);
-		free(filename);
+		ret = map_column(t, i);
 		if (ret < 0)
 			goto err;
 	}
@@ -945,26 +955,23 @@ int get_mapped_object(const struct osl_table *t, unsigned col_num,
 	struct osl_column *col = &t->columns[col_num];
 	uint32_t offset;
 	char *header;
-	unsigned char *index_entry;
+	char *cell_index;
 	int ret;
 
 	if (t->num_rows <= row_num)
 		return -E_BAD_ID;
-	ret = get_index_entry(t, row_num, col_num, &index_entry);
+	ret = get_cell_index(t, row_num, col_num, &cell_index);
 	if (ret < 0)
 		return ret;
-	offset = read_u32(index_entry);
-	obj->size = read_u32(index_entry + 4) - 1;
-	PARA_DEBUG_LOG("index_entry: %p\n", index_entry);
+	offset = read_u32(cell_index);
+	obj->size = read_u32(cell_index + 4) - 1;
 	header = col->data_map.data + offset;
 	obj->data = header + 1;
 	if (read_u8(header) == 0xff) {
-		PARA_ERROR_LOG("col %d, size %ld, offset %ld\n", col_num,
+		PARA_ERROR_LOG("col %u, size %zu, offset %u\n", col_num,
 			obj->size, offset);
 		return -E_INVALID_OBJECT;
 	}
-	PARA_DEBUG_LOG("mapped obj row_num: %u, col %u, size: %d\n", row_num,
-		col_num, obj->size);
 	return 1;
 }
 
@@ -1062,7 +1069,6 @@ static int add_row_to_rbtrees(struct osl_table *t, uint32_t id,
 	struct osl_row *row = allocate_row(t->num_rbtrees);
 	const struct osl_column_description *cd;
 
-	PARA_DEBUG_LOG("row: %p, id: %u\n", row, id);
 	row->id = id;
 	row->volatile_objects = volatile_objs;
 	FOR_EACH_RBTREE_COLUMN(i, t, cd) {
@@ -1075,7 +1081,6 @@ static int add_row_to_rbtrees(struct osl_table *t, uint32_t id,
 		} else { /* volatile */
 			const struct osl_object *obj
 				= volatile_objs + t->columns[i].volatile_num;
-			PARA_DEBUG_LOG("inserting %p\n", obj->data);
 			ret = insert_rbtree(t, i, row, obj);
 		}
 		if (ret < 0)
@@ -1190,13 +1195,13 @@ int osl_close_table(struct osl_table *t, enum osl_close_flags flags)
  */
 int row_is_invalid(struct osl_table *t, uint32_t row_num)
 {
-	unsigned char *index_entry;
-	int i, ret = get_index_entry_start(t, row_num, &index_entry);
+	char *row_index;
+	int i, ret = get_row_index(t, row_num, &row_index);
 
 	if (ret < 0)
 		return ret;
-	for (i = 0; i < t->index_entry_size; i++) {
-		if (index_entry[i] != 0xff)
+	for (i = 0; i < t->row_index_size; i++) {
+		if ((unsigned char)row_index[i] != 0xff)
 			return 0;
 	}
 	PARA_INFO_LOG("row %d is invalid\n", row_num);
@@ -1216,14 +1221,13 @@ int row_is_invalid(struct osl_table *t, uint32_t row_num)
  */
 int mark_row_invalid(struct osl_table *t, uint32_t row_num)
 {
-	unsigned char *index_entry;
-	int i, ret = get_index_entry_start(t, row_num, &index_entry);
+	char *row_index;
+	int ret = get_row_index(t, row_num, &row_index);
 
-	PARA_INFO_LOG("marking row %d as invalid\n", row_num);
 	if (ret < 0)
 		return ret;
-	for (i = 0; i < t->index_entry_size; i++)
-		index_entry[i] = 0xff;
+	PARA_INFO_LOG("marking row %d as invalid\n", row_num);
+	memset(row_index, 0xff, t->row_index_size);
 	return 1;
 }
 
@@ -1359,14 +1363,14 @@ static int append_map_file(const struct osl_table *t, unsigned col_num,
 	int ret;
 	char header = 0; /* zero means valid object */
 
-	PARA_DEBUG_LOG("appending %ld + 1 byte\n", obj->size);
+//	PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj->size);
 	ret = append_file(filename, &header, 1, obj->data, obj->size,
 		new_size);
 	free(filename);
 	return ret;
 }
 
-static int append_index_entry(const struct osl_table *t, char *new_index_entry)
+static int append_row_index(const struct osl_table *t, char *row_index)
 {
 	char *filename;
 	int ret;
@@ -1374,9 +1378,8 @@ static int append_index_entry(const struct osl_table *t, char *new_index_entry)
 	if (!t->num_mapped_columns)
 		return 1;
 	filename = index_filename(t->desc);
-	PARA_DEBUG_LOG("appending %ld bytes\n", t->index_entry_size);
-	ret = append_file(filename, NULL, 0, new_index_entry,
-		t->index_entry_size, NULL);
+	ret = append_file(filename, NULL, 0, row_index,
+		t->row_index_size, NULL);
 	free(filename);
 	return ret;
 }
@@ -1428,7 +1431,7 @@ static int delete_disk_storage_file(const struct osl_table *t, unsigned col_num,
 	char *dirname, *filename = disk_storage_path(t, col_num, ds_name);
 	int ret = unlink(filename);
 
-	PARA_INFO_LOG("deleted %s\n", filename);
+	PARA_DEBUG_LOG("deleted %s\n", filename);
 	free(filename);
 	if (ret < 0) {
 		if (errno == ENOENT)
@@ -1474,7 +1477,7 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 	int i, ret;
 	char *ds_name = NULL;
 	struct rb_node **rb_parents = NULL, ***rb_links = NULL;
-	char *new_index_entry = NULL;
+	char *new_row_index = NULL;
 	struct osl_object *volatile_objs = NULL;
 	const struct osl_column_description *cd;
 
@@ -1483,10 +1486,10 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 	rb_parents = para_malloc(t->num_rbtrees * sizeof(struct rn_node*));
 	rb_links = para_malloc(t->num_rbtrees * sizeof(struct rn_node**));
 	if (t->num_mapped_columns)
-		new_index_entry = para_malloc(t->index_entry_size);
+		new_row_index = para_malloc(t->row_index_size);
 	/* pass 1: sanity checks */
-	PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
-		objects[1].data);
+//	PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
+//		objects[1].data);
 	FOR_EACH_COLUMN(i, t->desc, cd) {
 		enum osl_storage_type st = cd->storage_type;
 		enum osl_storage_flags sf = cd->storage_flags;
@@ -1499,16 +1502,16 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 		if (sf & OSL_RBTREE) {
 			unsigned rbtree_num = t->columns[i].rbtree_num;
 			ret = -E_RB_KEY_EXISTS;
-			PARA_DEBUG_LOG("checking whether %p exists\n",
-				objects[i].data);
+//			PARA_DEBUG_LOG("checking whether %p exists\n",
+//				objects[i].data);
 			if (search_rbtree(objects + i, t, i,
 					&rb_parents[rbtree_num],
 					&rb_links[rbtree_num]) > 0)
 				goto out;
 		}
 		if (sf & OSL_FIXED_SIZE) {
-			PARA_DEBUG_LOG("fixed size. need: %d, have: %d\n",
-				objects[i].size, cd->data_size);
+//			PARA_DEBUG_LOG("fixed size. need: %zu, have: %d\n",
+//				objects[i].size, cd->data_size);
 			ret = -E_BAD_DATA_SIZE;
 			if (objects[i].size != cd->data_size)
 				goto out;
@@ -1520,7 +1523,7 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 	ret = unmap_table(t, OSL_MARK_CLEAN);
 	if (ret < 0)
 		goto out;
-	PARA_DEBUG_LOG("sanity tests passed%s\n", "");
+//	PARA_DEBUG_LOG("sanity tests passed%s\n", "");
 	/* pass 2: create data files, append map data */
 	FOR_EACH_COLUMN(i, t->desc, cd) {
 		enum osl_storage_type st = cd->storage_type;
@@ -1529,12 +1532,12 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 		if (st == OSL_MAPPED_STORAGE) {
 			uint32_t new_size;
 			struct osl_column *col = &t->columns[i];
-			PARA_DEBUG_LOG("appending object of size %ld\n",
-				objects[i].size);
+//			PARA_DEBUG_LOG("appending object of size %zu\n",
+//				objects[i].size);
 			ret = append_map_file(t, i, objects + i, &new_size);
 			if (ret < 0)
 				goto rollback;
-			update_index_entry(new_index_entry, col, new_size,
+			update_cell_index(new_row_index, col, new_size,
 				objects[i].size);
 			continue;
 		}
@@ -1543,13 +1546,13 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 		if (ret < 0)
 			goto rollback;
 	}
-	ret = append_index_entry(t, new_index_entry);
+	ret = append_row_index(t, new_row_index);
 	if (ret < 0)
 		goto rollback;
 	ret = map_table(t, MAP_TBL_FL_VERIFY_INDEX);
 	if (ret < 0) { /* truncate index and rollback changes */
 		char *filename = index_filename(t->desc);
-		para_truncate(filename, t->index_entry_size);
+		para_truncate(filename, t->row_index_size);
 		free(filename);
 		goto rollback;
 	}
@@ -1561,11 +1564,11 @@ int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
 			volatile_objs[t->columns[i].volatile_num] = objects[i];
 	}
 	t->num_rows++;
-	PARA_DEBUG_LOG("adding new entry as row #%d\n", t->num_rows - 1);
+//	PARA_DEBUG_LOG("adding new entry as row #%d\n", t->num_rows - 1);
 	ret = add_row_to_rbtrees(t, t->num_rows - 1, volatile_objs, row);
 	if (ret < 0)
 		goto out;
-	PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
+//	PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
 	ret = 1;
 	goto out;
 rollback: /* rollback all changes made, ignore further errors */
@@ -1583,7 +1586,7 @@ rollback: /* rollback all changes made, ignore further errors */
 	/* ignore error and return previous error value */
 	map_table(t, MAP_TBL_FL_VERIFY_INDEX);
 out:
-	free(new_index_entry);
+	free(new_row_index);
 	free(ds_name);
 	free(rb_parents);
 	free(rb_links);
@@ -1920,6 +1923,7 @@ int osl_update_object(struct osl_table *t, const struct osl_row *r,
 		return -E_BAD_TABLE;
 	col = &t->columns[col_num];
 	cd = get_column_description(t->desc, col_num);
+	PARA_DEBUG_LOG("updating column %u of %s\n", col_num, t->desc->name);
 	if (cd->storage_flags & OSL_RBTREE) {
 		if (search_rbtree(obj, t, col_num, NULL, NULL) > 0)
 			return -E_RB_KEY_EXISTS;
@@ -1965,18 +1969,22 @@ int osl_update_object(struct osl_table *t, const struct osl_row *r,
 			memcpy(old_obj.data, obj->data, cd->data_size);
 		else { /* TODO: if the size doesn't change, use old space */
 			uint32_t new_data_map_size;
-			unsigned char *index_entry;
-			ret = get_index_entry_start(t, r->id, &index_entry);
+			char *row_index;
+			ret = get_row_index(t, r->id, &row_index);
 			if (ret < 0)
 				return ret;
 			ret = mark_mapped_object_invalid(t, r->id, col_num);
 			if (ret < 0)
 				return ret;
+			unmap_column(t, col_num);
 			ret = append_map_file(t, col_num, obj,
 				&new_data_map_size);
 			if (ret < 0)
 				return ret;
-			update_index_entry(index_entry, col, new_data_map_size,
+			ret = map_column(t, col_num);
+			if (ret < 0)
+				return ret;
+			update_cell_index(row_index, col, new_data_map_size,
 				obj->size);
 		}
 	}