Context Navigation

mkspace.c @ 8760

Revision 8760, 32.9 KB (checked in by elaine, 2 years ago)
Add a "type" field to keys in the keyval database so both directory entries and attributes can be stored for dirdata handles.

Line
1	/*
2	* (C) 2002 Clemson University and The University of Chicago
3	*
4	* See COPYING in top-level directory.
5	*/
6
7	#include <unistd.h>
8	#include <string.h>
9	#include <stdio.h>
10	#include <limits.h>
11	#include <string.h>
12	#include <stdlib.h>
13	#include <time.h>
14	#include <assert.h>
15
16	#include "pvfs2-attr.h"
17	#include "trove.h"
18	#include "mkspace.h"
19	#include "pvfs2-debug.h"
20	#include "gossip.h"
21	#include "server-config.h"
22	#include "str-utils.h"
23	#include "extent-utils.h"
24	#include "pvfs2-util.h"
25	#include "pvfs2-internal.h"
26	#include "pint-util.h"
27	#include "pint-event.h"
28	#include "dist-dir-utils.h"
29
30	static char *lost_and_found_string = "lost+found";
31
32	static TROVE_handle s_used_handles[4] =
33	{
34	TROVE_HANDLE_NULL, TROVE_HANDLE_NULL,
35	TROVE_HANDLE_NULL, TROVE_HANDLE_NULL
36	};
37
38	#define mkspace_print(v, format, f...) \
39	do { \
40	if (v == PVFS2_MKSPACE_GOSSIP_VERBOSE) \
41	gossip_debug(GOSSIP_SERVER_DEBUG, format, ##f); \
42	else if (v == PVFS2_MKSPACE_STDERR_VERBOSE) \
43	fprintf(stderr,format, ##f); \
44	} while(0)
45
46	static int handle_is_excluded(
47	TROVE_handle handle, TROVE_handle *handles_to_exclude,
48	int num_handles_to_exclude)
49	{
50	int excluded = 0;
51
52	while((num_handles_to_exclude - 1) > -1)
53	{
54	if (handle == handles_to_exclude[num_handles_to_exclude-1])
55	{
56	excluded = 1;
57	break;
58	}
59	num_handles_to_exclude--;
60	}
61	return excluded;
62	}
63
64	static void get_handle_extent_from_ranges(
65	char handle_ranges, TROVE_handle_extent out_extent,
66	TROVE_handle *handles_to_exclude, int num_handles_to_exclude)
67	{
68	PINT_llist *cur = NULL;
69	TROVE_handle_extent *tmp_extent = NULL;
70	PINT_llist *extent_list = NULL;
71
72	if (handle_ranges && out_extent)
73	{
74	out_extent->first = TROVE_HANDLE_NULL;
75	out_extent->last = TROVE_HANDLE_NULL;
76
77	extent_list = PINT_create_extent_list(handle_ranges);
78	if (extent_list)
79	{
80	cur = extent_list;
81	while(cur)
82	{
83	tmp_extent = PINT_llist_head(cur);
84	if (!tmp_extent)
85	{
86	break;
87	}
88
89	/*
90	allow any handle range in this list that can allow
91	at least the single handle allocation to pass. a
92	range of 1 is ok, so long as it's not a handle that
93	was previously allocated (i.e. in the specified
94	excluded list)
95	*/
96	if (((tmp_extent->last - tmp_extent->first) > 0) \|\|
97	((tmp_extent->last > 0) &&
98	(tmp_extent->last == tmp_extent->first) &&
99	!handle_is_excluded(
100	tmp_extent->last, handles_to_exclude,
101	num_handles_to_exclude)))
102	{
103	out_extent->first = tmp_extent->first;
104	out_extent->last = tmp_extent->last;
105	break;
106	}
107	cur = PINT_llist_next(cur);
108	}
109	PINT_release_extent_list(extent_list);
110	}
111	}
112	}
113
114	int pvfs2_mkspace(
115	char *data_path,
116	char *meta_path,
117	char *collection,
118	TROVE_coll_id coll_id,
119	TROVE_handle root_handle,
120	char *meta_handle_ranges,
121	char *data_handle_ranges,
122	int create_collection_only,
123	int verbose)
124	{
125	int ret = - 1, count = 0, keyval_count, i;
126	TROVE_op_id op_id;
127	TROVE_ds_state state;
128	TROVE_keyval_s key, val;
129	TROVE_keyval_s key_a = NULL, val_a = NULL;
130	TROVE_ds_attributes_s attr;
131	TROVE_handle_extent cur_extent;
132	TROVE_handle_extent_array extent_array;
133	TROVE_context_id trove_context = -1;
134	char *merged_handle_ranges = NULL;
135	unsigned char *c = NULL;
136	TROVE_handle new_root_handle = TROVE_HANDLE_NULL;
137	TROVE_handle root_dirdata_handle = TROVE_HANDLE_NULL;
138	TROVE_handle lost_and_found_handle = TROVE_HANDLE_NULL;
139	TROVE_handle lost_and_found_dirdata_handle = TROVE_HANDLE_NULL;
140
141	/* distributed directory parameters */
142	PVFS_dist_dir_attr meta_dist_dir_attr;
143	PVFS_dist_dir_attr dirdata_dist_dir_attr;
144	PVFS_dist_dir_bitmap dist_dir_bitmap = NULL;
145	/* only one dirdata_handle for root and lost_and_found, not using arrays
146	PVFS_handle *dirdata_handles;
147	*/
148
149	mkspace_print(verbose,"Data storage space : %s\n",data_path);
150	mkspace_print(verbose,"Metadata storage space : %s\n", meta_path);
151	mkspace_print(verbose,"Collection : %s\n",collection);
152	mkspace_print(verbose,"ID : %d\n",coll_id);
153	mkspace_print(verbose,"Root Handle : %llu\n",llu(root_handle));
154	mkspace_print(verbose,"Meta Handles : %s\n",
155	(meta_handle_ranges && strlen(meta_handle_ranges) ?
156	meta_handle_ranges : "NONE"));
157	mkspace_print(verbose,"Data Handles : %s\n",
158	(data_handle_ranges && strlen(data_handle_ranges) ?
159	data_handle_ranges : "NONE"));
160
161	new_root_handle = root_handle;
162
163	/*
164	if we're only creating a collection inside an existing
165	storage space, we need to assume that it exists already
166	*/
167	if (!create_collection_only)
168	{
169	/*
170	try to initialize; fails if storage space isn't there, which
171	is exactly what we're expecting in this case.
172	*/
173	ret = trove_initialize(TROVE_METHOD_DBPF,
174	NULL,
175	data_path,
176	meta_path,
177	0);
178	if (ret > -1)
179	{
180	gossip_err("error: storage space %s or %s already "
181	"exists; aborting!\n",data_path,meta_path);
182	return -1;
183	}
184
185	ret = trove_storage_create(TROVE_METHOD_DBPF, data_path, meta_path, NULL, &op_id);
186	if (ret != 1)
187	{
188	gossip_err("error: storage create failed; aborting!\n");
189	return -1;
190	}
191	}
192
193	/* now that the storage space exists, initialize trove properly */
194	ret = trove_initialize(
195	TROVE_METHOD_DBPF, NULL,
196	data_path, meta_path, 0);
197	if (ret < 0)
198	{
199	gossip_err("error: trove initialize failed; aborting!\n");
200	return -1;
201	}
202
203	mkspace_print(verbose,"info: created data storage space '%s'.\n",
204	data_path);
205	mkspace_print(verbose,"info: created metadata storage space '%s'.\n",
206	meta_path);
207
208	/* try to look up collection used to store file system */
209	ret = trove_collection_lookup(
210	TROVE_METHOD_DBPF, collection, &coll_id, NULL, &op_id);
211	if (ret == 1)
212	{
213	mkspace_print(verbose, "warning: collection lookup succeeded "
214	"before it should; aborting!\n");
215	trove_finalize(TROVE_METHOD_DBPF);
216	return -1;
217	}
218
219	/* create the collection for the fs */
220	ret = trove_collection_create(collection, coll_id, NULL, &op_id);
221	if (ret != 1)
222	{
223	mkspace_print(verbose,"error: collection create failed for "
224	"collection '%s'.\n",collection);
225	return -1;
226	}
227
228	/* make sure a collection lookup succeeds */
229	ret = trove_collection_lookup(
230	TROVE_METHOD_DBPF, collection, &coll_id, NULL, &op_id);
231	if (ret != 1)
232	{
233	mkspace_print(verbose,"error: collection lookup failed for "
234	"collection '%s' after create.\n",collection);
235	return -1;
236	}
237
238	mkspace_print(verbose, "info: created collection '%s'.\n",collection);
239
240	ret = trove_open_context(coll_id, &trove_context);
241	if (ret < 0)
242	{
243	mkspace_print(verbose,"trove_open_context() failure.\n");
244	return -1;
245	}
246
247	/* merge the specified ranges to pass to the handle allocator */
248	if ((meta_handle_ranges && strlen(meta_handle_ranges)) &&
249	(data_handle_ranges && strlen(data_handle_ranges)))
250	{
251	merged_handle_ranges = PINT_merge_handle_range_strs(
252	meta_handle_ranges, data_handle_ranges);
253	}
254	else if (meta_handle_ranges && strlen(meta_handle_ranges))
255	{
256	merged_handle_ranges = strdup(meta_handle_ranges);
257	}
258	else if (data_handle_ranges && strlen(data_handle_ranges))
259	{
260	merged_handle_ranges = strdup(data_handle_ranges);
261	}
262
263	if (!merged_handle_ranges)
264	{
265	gossip_err("Failed to merge the handle range! Format invalid\n");
266	return -1;
267	}
268
269	/*
270	set the trove handle ranges; this initializes the handle
271	allocator with the ranges we were told to use
272	*/
273	ret = trove_collection_setinfo(
274	coll_id, trove_context, TROVE_COLLECTION_HANDLE_RANGES,
275	merged_handle_ranges);
276
277	if (ret < 0)
278	{
279	mkspace_print(verbose, "Error adding handle ranges: %s\n",
280	merged_handle_ranges);
281	free(merged_handle_ranges);
282	return -1;
283	}
284
285	mkspace_print(verbose,"info: set handle ranges to %s\n",
286	merged_handle_ranges);
287
288	free(merged_handle_ranges);
289
290
291	/*
292	if a root_handle is specified, 1) create a dataspace to hold the
293	root directory 2) create the dspace for dir entries, 3) set
294	attributes on the dspace
295	*/
296	if (new_root_handle != TROVE_HANDLE_NULL)
297	{
298	cur_extent.first = cur_extent.last = new_root_handle;
299	extent_array.extent_count = 1;
300	extent_array.extent_array = &cur_extent;
301
302	ret = trove_dspace_create(
303	coll_id, &extent_array, &new_root_handle,
304	PVFS_TYPE_DIRECTORY, NULL,
305	(TROVE_SYNC \| TROVE_FORCE_REQUESTED_HANDLE),
306	NULL, trove_context, &op_id, NULL);
307
308	while (ret == 0)
309	{
310	ret = trove_dspace_test(coll_id, op_id, trove_context,
311	&count, NULL, NULL, &state,
312	TROVE_DEFAULT_TEST_TIMEOUT);
313	}
314
315	if ((ret != 1) && (state != 0))
316	{
317	mkspace_print(verbose,
318	"dspace create (for root dir) failed.\n");
319	return -1;
320	}
321
322	mkspace_print(verbose,"info: created root directory "
323	"with handle %llu.\n", llu(new_root_handle));
324	s_used_handles[0] = new_root_handle;
325
326	/* set collection attribute for root handle */
327	key.buffer = ROOT_HANDLE_KEYSTR;
328	key.buffer_sz = ROOT_HANDLE_KEYLEN;
329	val.buffer = &new_root_handle;
330	val.buffer_sz = sizeof(new_root_handle);
331	ret = trove_collection_seteattr(coll_id, &key, &val, 0,
332	NULL, trove_context, &op_id);
333	while (ret == 0)
334	{
335	ret = trove_dspace_test(
336	coll_id, op_id, trove_context, &count, NULL, NULL,
337	&state, TROVE_DEFAULT_TEST_TIMEOUT);
338	}
339
340	if (ret < 0)
341	{
342	gossip_err("error: collection seteattr (for root handle) "
343	"failed; aborting!\n");
344	return -1;
345	}
346
347	/* set root directory dspace attributes */
348	memset(&attr, 0, sizeof(TROVE_ds_attributes_s));
349	attr.uid = getuid();
350	attr.gid = getgid();
351	attr.mode = 0777;
352	attr.type = PVFS_TYPE_DIRECTORY;
353	attr.atime = attr.ctime = PINT_util_get_current_time();
354	attr.mtime = PINT_util_mktime_version(attr.ctime);
355
356	ret = trove_dspace_setattr(
357	coll_id, new_root_handle, &attr, TROVE_SYNC, NULL,
358	trove_context, &op_id, NULL);
359
360	while (ret == 0)
361	{
362	ret = trove_dspace_test(
363	coll_id, op_id, trove_context, &count, NULL, NULL,
364	&state, TROVE_DEFAULT_TEST_TIMEOUT);
365	}
366
367	if (ret < 0)
368	{
369	gossip_err("error: dspace setattr for root handle "
370	"attributes failed; aborting!\n");
371	return -1;
372	}
373
374	/*
375	create a dataspace to hold directory entries; if we have a
376	meta handle range, use that one of those ranges (being
377	careful to make sure the range has enough space for an
378	allocation) to allocate a dataspace to hold directory
379	entries. if we don't have a meta handle range, use
380	TROVE_HANDLE_NULL which tells the allocator to use any
381	handle available
382	*/
383	cur_extent.first = cur_extent.last = TROVE_HANDLE_NULL;
384	if (meta_handle_ranges)
385	{
386	get_handle_extent_from_ranges(
387	meta_handle_ranges, &cur_extent, s_used_handles, 1);
388
389	if ((cur_extent.first == TROVE_HANDLE_NULL) &&
390	(cur_extent.last == TROVE_HANDLE_NULL))
391	{
392	gossip_err("No valid meta handle ranges available! "
393	"Using a default\n");
394	}
395	else
396	{
397	mkspace_print(
398	verbose, "info: using meta handle range %llu-%llu for "
399	"root dirent dspace\n", llu(cur_extent.first),
400	llu(cur_extent.last));
401	}
402	}
403
404	extent_array.extent_count = 1;
405	extent_array.extent_array = &cur_extent;
406
407	ret = trove_dspace_create(
408	coll_id, &extent_array, &root_dirdata_handle,
409	PVFS_TYPE_DIRDATA, NULL, TROVE_SYNC, NULL,
410	trove_context, &op_id, NULL);
411
412	while (ret == 0)
413	{
414	ret = trove_dspace_test(
415	coll_id, op_id, trove_context, &count, NULL, NULL,
416	&state, TROVE_DEFAULT_TEST_TIMEOUT);
417	}
418
419	if ((ret != 1) && (state != 0))
420	{
421	gossip_err("dspace create (for dirent storage) failed.\n");
422	return -1;
423	}
424
425	mkspace_print(verbose, "info: created dspace for dirents "
426	"with handle %llu\n", llu(root_dirdata_handle));
427	s_used_handles[1] = root_dirdata_handle;
428
429	#if 0
430	/* set root dirdata dspace attributes */
431	memset(&attr, 0, sizeof(TROVE_ds_attributes_s));
432	attr.uid = getuid();
433	attr.gid = getgid();
434	attr.mode = 0777;
435	attr.type = PVFS_TYPE_DIRDATA;
436	attr.atime = attr.ctime = PINT_util_get_current_time();
437	attr.mtime = PINT_util_mktime_version(attr.ctime);
438
439	ret = trove_dspace_setattr(
440	coll_id, root_dirdata_handle, &attr, TROVE_SYNC, NULL,
441	trove_context, &op_id, NULL);
442
443	while (ret == 0)
444	{
445	ret = trove_dspace_test(
446	coll_id, op_id, trove_context, &count, NULL, NULL,
447	&state, TROVE_DEFAULT_TEST_TIMEOUT);
448	}
449
450	if (ret < 0)
451	{
452	gossip_err("error: dspace setattr for root dirdata handle "
453	"attributes failed; aborting!\n");
454	return -1;
455	}
456	#endif
457
458
459	key.buffer = DIRECTORY_ENTRY_KEYSTR;
460	key.buffer_sz = DIRECTORY_ENTRY_KEYLEN;
461	val.buffer = &root_dirdata_handle;
462	val.buffer_sz = sizeof(TROVE_handle);
463
464	ret = trove_keyval_write(
465	coll_id, new_root_handle, &key, &val,
466	TROVE_SYNC, 0, NULL,
467	trove_context, &op_id, NULL);
468
469	while (ret == 0)
470	{
471	ret = trove_dspace_test(
472	coll_id, op_id, trove_context, &count, NULL, NULL,
473	&state, TROVE_DEFAULT_TEST_TIMEOUT);
474	}
475
476	if (ret < 0)
477	{
478	gossip_err("error: keyval write for handle used to store "
479	"dirents failed; aborting!\n");
480	return -1;
481	}
482
483	mkspace_print(
484	verbose, "info: wrote attributes for root directory.\n");
485
486
487	/* !!! set dist-dir-struct keyvals for root handle and its dirdata handle */
488	/* init meta handle dist_dir_struct, num_servers=1, server_no = -1, pre_dsg_num_server=1 */
489	/* should be reusable when init lost+found */
490	ret = PINT_init_dist_dir_state(&meta_dist_dir_attr,
491	&dist_dir_bitmap, 1, 0, 1);
492	assert(ret == 0);
493
494	/* gossip dist_dir_attr and dist_dir_bitmap */
495	mkspace_print(verbose, "info: init dist_dir_attr for root handle %llu: \n"
496	"with tree_height=%d, num_servers=%d, bitmap_size=%d, "
497	"split_size=%d, server_no=%d and branch_level=%d\n",
498	llu(new_root_handle),
499	meta_dist_dir_attr.tree_height,
500	meta_dist_dir_attr.num_servers,
501	meta_dist_dir_attr.bitmap_size,
502	meta_dist_dir_attr.split_size,
503	meta_dist_dir_attr.server_no,
504	meta_dist_dir_attr.branch_level);
505	mkspace_print(verbose, "info: init dist_dir_bitmap as:\n");
506	for(i = meta_dist_dir_attr.bitmap_size - 1;
507	i >= 0 ; i--)
508	{
509	c = (unsigned char *)(dist_dir_bitmap + i);
510	mkspace_print(verbose,
511	" i=%d : %02x %02x %02x %02x\n",
512	i, c[3], c[2], c[1], c[0]);
513	}
514	mkspace_print(verbose, "\n");
515
516	/* init dirdata_dist_dir_attr */
517	PINT_dist_dir_attr_copyto(dirdata_dist_dir_attr, meta_dist_dir_attr);
518	PINT_dist_dir_set_serverno(0, &dirdata_dist_dir_attr, dist_dir_bitmap);
519
520	mkspace_print(verbose, "info: init dist_dir_attr for root dirdata handle %llu: \n"
521	"with tree_height=%d, num_servers=%d, bitmap_size=%d, "
522	"split_size=%d, server_no=%d and branch_level=%d\n",
523	llu(root_dirdata_handle),
524	dirdata_dist_dir_attr.tree_height,
525	dirdata_dist_dir_attr.num_servers,
526	dirdata_dist_dir_attr.bitmap_size,
527	dirdata_dist_dir_attr.split_size,
528	dirdata_dist_dir_attr.server_no,
529	dirdata_dist_dir_attr.branch_level);
530
531	keyval_count = 3;
532
533	key_a = malloc(sizeof(TROVE_keyval_s) * keyval_count);
534	if(!key_a)
535	{
536	gossip_err("keyval space create (for root and lost+found storage) failed.\n");
537	return -1;
538	}
539	memset(key_a, 0, sizeof(TROVE_keyval_s) * keyval_count);
540
541	val_a = malloc(sizeof(TROVE_keyval_s) * keyval_count);
542	if(!val_a)
543	{
544	free(key_a);
545	gossip_err("keyval space create (for root and lost+found storage) failed.\n");
546	return -1;
547	}
548	memset(val_a, 0, sizeof(TROVE_keyval_s) * keyval_count);
549
550	/* set keyval for root meta handle */
551	key_a[0].buffer = DIST_DIR_ATTR_KEYSTR;
552	key_a[0].buffer_sz = DIST_DIR_ATTR_KEYLEN;
553
554	val_a[0].buffer = &meta_dist_dir_attr;
555	val_a[0].buffer_sz =
556	sizeof(meta_dist_dir_attr);
557
558	key_a[1].buffer = DIST_DIRDATA_BITMAP_KEYSTR;
559	key_a[1].buffer_sz = DIST_DIRDATA_BITMAP_KEYLEN;
560
561	val_a[1].buffer_sz =
562	meta_dist_dir_attr.bitmap_size * /* bitmap_size = 1 */
563	sizeof(PVFS_dist_dir_bitmap_basetype);
564	val_a[1].buffer = dist_dir_bitmap;
565
566	key_a[2].buffer = DIST_DIRDATA_HANDLES_KEYSTR;
567	key_a[2].buffer_sz = DIST_DIRDATA_HANDLES_KEYLEN;
568
569	val_a[2].buffer = &root_dirdata_handle; /* only one dirdata server */
570	val_a[2].buffer_sz = meta_dist_dir_attr.num_servers *
571	sizeof(root_dirdata_handle);
572
573	mkspace_print(verbose,
574	" info: writing dist-dir-struct keyvals for root handle: %llu "
575	" with server_no=%d and branch_level=%d \n",
576	llu(new_root_handle),
577	meta_dist_dir_attr.server_no,
578	meta_dist_dir_attr.branch_level);
579
580	/* write to root meta handle keyval space */
581	ret = trove_keyval_write_list(
582	coll_id, new_root_handle, key_a, val_a, keyval_count,
583	TROVE_SYNC, 0, NULL,
584	trove_context, &op_id, NULL);
585
586	while (ret == 0)
587	{
588	ret = trove_dspace_test(
589	coll_id, op_id, trove_context, &count, NULL, NULL,
590	&state, TROVE_DEFAULT_TEST_TIMEOUT);
591	}
592
593	if (ret < 0)
594	{
595	gossip_err("error: keyval write for handle used to store "
596	"dirents failed; aborting!\n");
597	return -1;
598	}
599
600	#if 0
601	/* adjust dist_dir_attr val_a */
602	val_a[0].buffer = &dirdata_dist_dir_attr;
603
604	mkspace_print(verbose,
605	" info: writing dist-dir-struct keyvals for root dirdata handle: %llu "
606	" with server_no=%d and branch_level=%d \n",
607	llu(root_dirdata_handle),
608	dirdata_dist_dir_attr.server_no,
609	dirdata_dist_dir_attr.branch_level);
610
611	/* write to root dirdata handle keyval space */
612	ret = trove_keyval_write_list(
613	coll_id, root_dirdata_handle, key_a, val_a, keyval_count,
614	TROVE_SYNC, 0, NULL,
615	trove_context, &op_id, NULL);
616
617	while (ret == 0)
618	{
619	ret = trove_dspace_test(
620	coll_id, op_id, trove_context, &count, NULL, NULL,
621	&state, TROVE_DEFAULT_TEST_TIMEOUT);
622	}
623
624	if (ret < 0)
625	{
626	gossip_err("error: dist-dir-struct keyval write for root dirdata handle "
627	"failed; aborting!\n");
628	return -1;
629	}
630
631	mkspace_print(
632	verbose, "info: wrote attributes for root directory.\n");
633	#endif
634
635	/****************************************************
636	at this point we need to create and initialize the
637	lost+found directory as well
638	*****************************************************/
639	cur_extent.first = cur_extent.last = TROVE_HANDLE_NULL;
640	if (meta_handle_ranges)
641	{
642	get_handle_extent_from_ranges(
643	meta_handle_ranges, &cur_extent, s_used_handles, 2);
644
645	if ((cur_extent.first == TROVE_HANDLE_NULL) &&
646	(cur_extent.last == TROVE_HANDLE_NULL))
647	{
648	gossip_err("No valid meta handle ranges available! "
649	"Using a default\n");
650	}
651	else
652	{
653	mkspace_print(
654	verbose, "info: using meta handle range %llu-%llu for "
655	"lost+found directory dspace\n", llu(cur_extent.first),
656	llu(cur_extent.last));
657	}
658	}
659	extent_array.extent_count = 1;
660	extent_array.extent_array = &cur_extent;
661
662	ret = trove_dspace_create(
663	coll_id, &extent_array, &lost_and_found_handle,
664	PVFS_TYPE_DIRECTORY, NULL, TROVE_SYNC, NULL,
665	trove_context, &op_id, NULL);
666
667	while (ret == 0)
668	{
669	ret = trove_dspace_test(
670	coll_id, op_id, trove_context, &count, NULL, NULL,
671	&state, TROVE_DEFAULT_TEST_TIMEOUT);
672	}
673
674	if ((ret != 1) && (state != 0))
675	{
676	mkspace_print(verbose,
677	"dspace create (for lost+found dir) failed.\n");
678	return -1;
679	}
680
681	mkspace_print(verbose,"info: created lost+found directory "
682	"with handle %llu.\n", llu(lost_and_found_handle));
683	s_used_handles[2] = lost_and_found_handle;
684
685	/* set lost+found directory dspace attributes */
686	memset(&attr, 0, sizeof(TROVE_ds_attributes_s));
687	attr.uid = getuid();
688	attr.gid = getgid();
689	attr.mode = 0777;
690	attr.type = PVFS_TYPE_DIRECTORY;
691	attr.atime = attr.ctime = PINT_util_get_current_time();
692	attr.mtime = PINT_util_mktime_version(attr.ctime);
693
694	ret = trove_dspace_setattr(
695	coll_id, lost_and_found_handle, &attr, TROVE_SYNC, NULL,
696	trove_context, &op_id, NULL);
697
698	while (ret == 0)
699	{
700	ret = trove_dspace_test(
701	coll_id, op_id, trove_context, &count, NULL, NULL,
702	&state, TROVE_DEFAULT_TEST_TIMEOUT);
703	}
704
705	if (ret < 0)
706	{
707	gossip_err("error: dspace setattr for lost+found handle "
708	"attributes failed; aborting!\n");
709	return -1;
710	}
711
712	/* create a dataspace to hold directory entries */
713	cur_extent.first = cur_extent.last = TROVE_HANDLE_NULL;
714	if (meta_handle_ranges)
715	{
716	get_handle_extent_from_ranges(
717	meta_handle_ranges, &cur_extent, s_used_handles, 3);
718
719	if ((cur_extent.first == TROVE_HANDLE_NULL) &&
720	(cur_extent.last == TROVE_HANDLE_NULL))
721	{
722	gossip_err("No valid meta handle ranges available! "
723	"Using a default\n");
724	}
725	else
726	{
727	mkspace_print(
728	verbose, "info: using meta handle range %llu-%llu for "
729	"lost+found dirent dspace\n", llu(cur_extent.first),
730	llu(cur_extent.last));
731	}
732	}
733	extent_array.extent_count = 1;
734	extent_array.extent_array = &cur_extent;
735
736	ret = trove_dspace_create(
737	coll_id, &extent_array, &lost_and_found_dirdata_handle,
738	PVFS_TYPE_DIRDATA, NULL, TROVE_SYNC, NULL,
739	trove_context, &op_id, NULL);
740
741	while (ret == 0)
742	{
743	ret = trove_dspace_test(
744	coll_id, op_id, trove_context, &count, NULL, NULL,
745	&state, TROVE_DEFAULT_TEST_TIMEOUT);
746	}
747
748	if ((ret != 1) && (state != 0))
749	{
750	gossip_err("dspace create (for dirent storage) failed.\n");
751	return -1;
752	}
753
754	mkspace_print(
755	verbose, "info: created dspace for dirents "
756	"with handle %llu\n", llu(lost_and_found_dirdata_handle));
757	s_used_handles[3] = lost_and_found_dirdata_handle;
758
759	#if 0
760	/* set lost+found dirdata dspace attributes */
761	memset(&attr, 0, sizeof(TROVE_ds_attributes_s));
762	attr.uid = getuid();
763	attr.gid = getgid();
764	attr.mode = 0777;
765	attr.type = PVFS_TYPE_DIRDATA;
766	attr.atime = attr.ctime = PINT_util_get_current_time();
767	attr.mtime = PINT_util_mktime_version(attr.ctime);
768
769	ret = trove_dspace_setattr(
770	coll_id, lost_and_found_dirdata_handle, &attr, TROVE_SYNC, NULL,
771	trove_context, &op_id, NULL);
772
773	while (ret == 0)
774	{
775	ret = trove_dspace_test(
776	coll_id, op_id, trove_context, &count, NULL, NULL,
777	&state, TROVE_DEFAULT_TEST_TIMEOUT);
778	}
779
780	if (ret < 0)
781	{
782	gossip_err("error: dspace setattr for lost+found handle "
783	"attributes failed; aborting!\n");
784	return -1;
785	}
786	#endif
787
788	key.buffer = DIRECTORY_ENTRY_KEYSTR;
789	key.buffer_sz = DIRECTORY_ENTRY_KEYLEN;
790	val.buffer = &lost_and_found_dirdata_handle;
791	val.buffer_sz = sizeof(TROVE_handle);
792
793	ret = trove_keyval_write(
794	coll_id, lost_and_found_handle, &key, &val,
795	TROVE_SYNC,
796	0, NULL, trove_context, &op_id, NULL);
797
798	while (ret == 0)
799	{
800	ret = trove_dspace_test(
801	coll_id, op_id, trove_context, &count, NULL, NULL,
802	&state, TROVE_DEFAULT_TEST_TIMEOUT);
803	}
804
805	if (ret < 0)
806	{
807	gossip_err("error: keyval write for handle used to store "
808	"dirents failed; aborting!\n");
809	return -1;
810	}
811
812	mkspace_print(verbose, "info: wrote attributes for "
813	"lost+found directory.\n");
814
815
816	/* set dist-dir-struct keyvals for lost+find handle and its dirdata handle
817	* use 1 dirdata server now, just as root handle
818	* */
819
820	/* adjust keyval values, others should be the same as the root handle */
821	val_a[0].buffer = &meta_dist_dir_attr;
822	val_a[2].buffer = &lost_and_found_dirdata_handle;
823	/* bitmap should be the same as the root handle for now */
824
825	mkspace_print(verbose,
826	" info: writing dist-dir-struct keyvals for lost+found handle: %llu "
827	" with server_no=%d and branch_level=%d \n",
828	llu(lost_and_found_handle),
829	meta_dist_dir_attr.server_no,
830	meta_dist_dir_attr.branch_level);
831
832	/* write to lost+found meta handle keyval space */
833	ret = trove_keyval_write_list(
834	coll_id, lost_and_found_handle, key_a, val_a, keyval_count,
835	TROVE_SYNC, 0, NULL,
836	trove_context, &op_id, NULL);
837
838	while (ret == 0)
839	{
840	ret = trove_dspace_test(
841	coll_id, op_id, trove_context, &count, NULL, NULL,
842	&state, TROVE_DEFAULT_TEST_TIMEOUT);
843	}
844
845	if (ret < 0)
846	{
847	gossip_err("error: keyval write for handle used to store "
848	"dirents failed; aborting!\n");
849	return -1;
850	}
851
852	#if 0
853	/* adjust dist_dir_attr val */
854	val_a[0].buffer = &dirdata_dist_dir_attr;
855
856	mkspace_print(verbose,
857	" info: writing dist-dir-struct keyvals for lost+found dirdata handle: %llu "
858	" with server_no=%d and branch_level=%d \n",
859	llu(lost_and_found_dirdata_handle),
860	dirdata_dist_dir_attr.server_no,
861	dirdata_dist_dir_attr.branch_level);
862
863	/* write to lost+found dirdata handle keyval space */
864	ret = trove_keyval_write_list(
865	coll_id, lost_and_found_dirdata_handle, key_a, val_a, keyval_count,
866	TROVE_SYNC, 0, NULL,
867	trove_context, &op_id, NULL);
868
869	while (ret == 0)
870	{
871	ret = trove_dspace_test(
872	coll_id, op_id, trove_context, &count, NULL, NULL,
873	&state, TROVE_DEFAULT_TEST_TIMEOUT);
874	}
875
876	if (ret < 0)
877	{
878	gossip_err("error: dist-dir-struct keyval write for lost+found dirdata handle "
879	"failed; aborting!\n");
880	return -1;
881	}
882
883	mkspace_print(verbose, "info: wrote attributes for "
884	"lost+found directory.\n");
885	#endif
886
887	/*
888	finally, crdirent the lost+found directory into the root
889	directory by writing the appropriate keyval info into the
890	root dirdata handle space
891	*/
892	key.buffer = lost_and_found_string;
893	key.buffer_sz = strlen(lost_and_found_string) + 1;
894	val.buffer = &lost_and_found_handle;
895	val.buffer_sz = sizeof(TROVE_handle);
896
897	ret = trove_keyval_write(
898	coll_id, root_dirdata_handle, &key, &val,
899	TROVE_SYNC \| TROVE_NOOVERWRITE \| TROVE_KEYVAL_HANDLE_COUNT \| TROVE_KEYVAL_DIRECTORY_ENTRY,
900	0,
901	NULL, trove_context, &op_id, NULL);
902
903	while (ret == 0)
904	{
905	ret = trove_dspace_test(
906	coll_id, op_id, trove_context, &count, NULL, NULL,
907	&state, TROVE_DEFAULT_TEST_TIMEOUT);
908	}
909
910	if (ret < 0)
911	{
912	gossip_err("error: keyval write for handle used to store "
913	"dirents failed; aborting!\n");
914	return -1;
915	}
916
917	mkspace_print(verbose, "info: adding lost+found directory to "
918	"the root directory.\n");
919	}
920
921	if (trove_context != -1)
922	{
923	trove_close_context(coll_id, trove_context);
924	}
925	trove_finalize(TROVE_METHOD_DBPF);
926
927	mkspace_print(verbose, "collection created:\n"
928	"\troot handle = %llu, coll id = %d, "
929	"root string = \"%s\"\n",
930	llu(root_handle), coll_id, ROOT_HANDLE_KEYSTR);
931
932	/* free space */
933	if (dist_dir_bitmap)
934	{
935	free(dist_dir_bitmap);
936	}
937	if (key_a)
938	{
939	free(key_a);
940	}
941	if (val_a)
942	{
943	free(val_a);
944	}
945
946	return 0;
947	}
948
949	int pvfs2_rmspace(
950	char *data_path,
951	char *meta_path,
952	char *collection,
953	TROVE_coll_id coll_id,
954	int remove_collection_only,
955	int verbose)
956	{
957	int ret = -1;
958	TROVE_op_id op_id;
959	static int trove_is_initialized = 0;
960
961	/* try to initialize; fails if storage space isn't there */
962	if (!trove_is_initialized)
963	{
964	ret = trove_initialize(
965	TROVE_METHOD_DBPF, NULL,
966	data_path, meta_path, 0);
967	if (ret == -1)
968	{
969	gossip_err("error: storage space %s or %s does not "
970	"exist; aborting!\n", data_path, meta_path);
971	return -1;
972	}
973	trove_is_initialized = 1;
974	}
975
976	mkspace_print(verbose, "Attempting to remove collection %s\n",
977	collection);
978
979	ret = trove_collection_remove(
980	TROVE_METHOD_DBPF, collection, NULL, &op_id);
981	mkspace_print(
982	verbose, "PVFS2 Collection %s removed %s\n", collection,
983	(((ret == 1) \|\| (ret == -TROVE_ENOENT)) ? "successfully" :
984	"with errors"));
985
986	if (!remove_collection_only)
987	{
988	ret = trove_storage_remove(TROVE_METHOD_DBPF, data_path, meta_path,
989	NULL, &op_id);
990	/*
991	* it is a bit weird to do a trove_finaliz() prior to blowing away
992	* the storage space, but this allows the __db files of the DB env
993	* to be blown away for the rmdir() to work correctly!
994	*/
995	trove_finalize(TROVE_METHOD_DBPF);
996	mkspace_print(
997	verbose, "PVFS2 Storage Space %s and %s removed %s\n",
998	data_path, meta_path, (((ret == 1) \|\| (ret == -TROVE_ENOENT)) ?
999	"successfully" : "with errors"));
1000
1001	trove_is_initialized = 0;
1002	}
1003	return ret;
1004	}
1005
1006	/*
1007	* Local variables:
1008	* c-indent-level: 4
1009	* c-basic-offset: 4
1010	* End:
1011	*
1012	* vim: ts=8 sts=4 sw=4 expandtab
1013	*/

Note: See TracBrowser for help on using the browser.

Context Navigation

root/branches/Orange-Elaine-Distr-Dir-Branch/src/common/misc/mkspace.c @ 8760

Download in other formats: