2 * Copyright (C) 2007 Andre Noll <maan@systemlinux.org>
4 * Licensed under the GPL v2. For licencing details see COPYING.
7 /** \file osl.c Object storage layer functions. */
8 #include <dirent.h> /* readdir() */
18 * A wrapper for lseek(2).
20 * \param fd The file descriptor whose offset is to be to repositioned.
21 * \param offset A value-result parameter.
22 * \param whence Usual repositioning directive.
24 * Reposition the offset of the file descriptor \a fd to the argument \a offset
25 * according to the directive \a whence. Upon successful return, \a offset
26 * contains the resulting offset location as measured in bytes from the
27 * beginning of the file.
29 * \return Positive on success. Otherwise, the function returns \p -E_LSEEK.
33 int para_lseek(int fd
, off_t
*offset
, int whence
)
35 *offset
= lseek(fd
, *offset
, whence
);
43 * Wrapper for the write system call.
45 * \param fd The file descriptor to write to.
46 * \param buf The buffer to write.
47 * \param size The length of \a buf in bytes.
49 * This function writes out the given buffer and retries if an interrupt
50 * occurred during the write.
52 * \return On success, the number of bytes written is returned, otherwise, the
53 * function returns \p -E_WRITE.
57 ssize_t
para_write(int fd
, const void *buf
, size_t size
)
62 ret
= write(fd
, buf
, size
);
63 if ((ret
< 0) && (errno
== EAGAIN
|| errno
== EINTR
))
65 return ret
>= 0? ret
: -E_WRITE
;
70 * Write the whole buffer to a file descriptor.
72 * \param fd The file descriptor to write to.
73 * \param buf The buffer to write.
74 * \param size The length of \a buf in bytes.
76 * This function writes the given buffer and continues on short writes and
77 * when interrupted by a signal.
79 * \return Positive on success, negative on errors. Possible errors: any
80 * errors returned by para_write().
84 ssize_t
para_write_all(int fd
, const void *buf
, size_t size
)
86 // PARA_DEBUG_LOG("writing %zu bytes\n", size);
89 ssize_t ret
= para_write(fd
, b
, size
);
90 // PARA_DEBUG_LOG("ret: %zd\n", ret);
99 * Open a file, write the given buffer and close the file.
101 * \param filename Full path to the file to open.
102 * \param buf The buffer to write to the file.
103 * \param size The size of \a buf.
105 * \return Positive on success, negative on errors. Possible errors include:
106 * any errors from para_open() or para_write().
108 * \sa para_open(), para_write().
110 int para_write_file(const char *filename
, const void *buf
, size_t size
)
114 ret
= para_open(filename
, O_WRONLY
| O_CREAT
| O_EXCL
, 0644);
118 ret
= para_write_all(fd
, buf
, size
);
127 static int append_file(const char *filename
, char *header
, size_t header_size
,
128 char *data
, size_t data_size
, uint32_t *new_pos
)
132 // PARA_DEBUG_LOG("appending %zu + %zu bytes\n", header_size, data_size);
133 ret
= para_open(filename
, O_WRONLY
| O_CREAT
| O_APPEND
, 0644);
137 if (header
&& header_size
) {
138 ret
= para_write_all(fd
, header
, header_size
);
142 ret
= para_write_all(fd
, data
, data_size
);
147 ret
= para_lseek(fd
, &offset
, SEEK_END
);
150 // PARA_DEBUG_LOG("new file size: " FMT_OFF_T "\n", offset);
160 * Map a file into memory.
162 * \param path Name of the regular file to map.
163 * \param open_mode Either \p O_RDONLY or \p O_RDWR.
164 * \param obj On success, the mapping is returned here.
166 * \return Positive on success, negative on errors. Possible errors include: \p
167 * E_FSTAT, any errors returned by para_open(), \p E_EMPTY, \p E_MMAP.
169 * \sa para_open(), mmap(2).
171 int mmap_full_file(const char *path
, int open_mode
, struct osl_object
*obj
)
173 int fd
, ret
, mmap_prot
, mmap_flags
;
174 struct stat file_status
;
176 if (open_mode
== O_RDONLY
) {
177 mmap_prot
= PROT_READ
;
178 mmap_flags
= MAP_PRIVATE
;
180 mmap_prot
= PROT_READ
| PROT_WRITE
;
181 mmap_flags
= MAP_SHARED
;
183 ret
= para_open(path
, open_mode
, 0);
188 if (fstat(fd
, &file_status
) < 0)
190 obj
->size
= file_status
.st_size
;
192 PARA_DEBUG_LOG("%s: size %zu\n", path
, obj
->size
);
195 obj
->data
= mmap(NULL
, obj
->size
, mmap_prot
, mmap_flags
, fd
, 0);
196 if (obj
->data
== MAP_FAILED
) {
208 * Traverse the given directory recursively.
210 * \param dirname The directory to traverse.
211 * \param func The function to call for each entry.
212 * \param private_data Pointer to an arbitrary data structure.
214 * For each regular file under \a dirname, the supplied function \a func is
215 * called. The full path of the regular file and the \a private_data pointer
216 * are passed to \a func. Directories for which the calling process has no
217 * permissions to change to are silently ignored.
219 * \return On success, 1 is returned. Otherwise, this function returns a
220 * negative value which indicates the kind of the error.
222 int for_each_file_in_dir(const char *dirname
,
223 int (*func
)(const char *, const void *), const void *private_data
)
226 struct dirent
*entry
;
227 int cwd_fd
, ret2
, ret
= para_opendir(dirname
, &dir
, &cwd_fd
);
230 return ret
== -E_CHDIR_PERM
? 1 : ret
;
231 /* scan cwd recursively */
232 while ((entry
= readdir(dir
))) {
237 if (!strcmp(entry
->d_name
, "."))
239 if (!strcmp(entry
->d_name
, ".."))
241 if (lstat(entry
->d_name
, &s
) == -1)
244 if (!S_ISREG(m
) && !S_ISDIR(m
))
246 tmp
= make_message("%s/%s", dirname
, entry
->d_name
);
248 ret
= func(tmp
, private_data
);
255 ret
= for_each_file_in_dir(tmp
, func
, private_data
);
263 ret2
= para_fchdir(cwd_fd
);
264 if (ret2
< 0 && ret
>= 0)
270 static int verify_name(const char *name
)
276 if (strchr(name
, '/'))
278 if (!strcmp(name
, ".."))
280 if (!strcmp(name
, "."))
286 * Compare two osl objects pointing to unsigned integers of 32 bit size.
288 * \param obj1 Pointer to the first integer.
289 * \param obj2 Pointer to the second integer.
291 * \return The values required for an osl compare function.
293 * \sa osl_compare_func, osl_hash_compare().
295 int uint32_compare(const struct osl_object
*obj1
, const struct osl_object
*obj2
)
297 uint32_t d1
= read_u32((const char *)obj1
->data
);
298 uint32_t d2
= read_u32((const char *)obj2
->data
);
308 * Compare two osl objects pointing to hash values.
310 * \param obj1 Pointer to the first hash object.
311 * \param obj2 Pointer to the second hash object.
313 * \return The values required for an osl compare function.
315 * \sa osl_compare_func, uint32_compare().
317 int osl_hash_compare(const struct osl_object
*obj1
, const struct osl_object
*obj2
)
319 return hash_compare((HASH_TYPE
*)obj1
->data
, (HASH_TYPE
*)obj2
->data
);
322 static char *disk_storage_dirname(const struct osl_table
*t
, unsigned col_num
,
325 char *dirname
, *column_name
= column_filename(t
, col_num
);
327 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
329 dirname
= make_message("%s/%.2s", column_name
, ds_name
);
334 static char *disk_storage_name_of_object(const struct osl_table
*t
,
335 const struct osl_object
*obj
)
337 HASH_TYPE hash
[HASH_SIZE
];
338 hash_object(obj
, hash
);
339 return disk_storage_name_of_hash(t
, hash
);
342 static int disk_storage_name_of_row(const struct osl_table
*t
,
343 const struct osl_row
*row
, char **name
)
345 struct osl_object obj
;
346 int ret
= osl_get_object(t
, row
, t
->disk_storage_name_column
, &obj
);
350 *name
= disk_storage_name_of_object(t
, &obj
);
354 static void column_name_hash(const char *col_name
, HASH_TYPE
*hash
)
356 return hash_function(col_name
, strlen(col_name
), hash
);
359 static int init_column_descriptions(struct osl_table
*t
)
362 const struct osl_column_description
*cd
;
364 ret
= -E_BAD_TABLE_DESC
;
365 ret
= verify_name(t
->desc
->name
);
371 /* the size of the index header without column descriptions */
372 t
->index_header_size
= IDX_COLUMN_DESCRIPTIONS
;
373 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
374 struct osl_column
*col
= t
->columns
+ i
;
375 if (cd
->storage_flags
& OSL_RBTREE
) {
376 if (!cd
->compare_function
)
377 return -E_NO_COMPARE_FUNC
;
379 if (cd
->storage_type
== OSL_NO_STORAGE
)
381 ret
= -E_NO_COLUMN_NAME
;
382 if (!cd
->name
|| !cd
->name
[0])
384 ret
= verify_name(cd
->name
);
387 t
->index_header_size
+= index_column_description_size(cd
->name
);
388 column_name_hash(cd
->name
, col
->name_hash
);
389 ret
= -E_DUPLICATE_COL_NAME
;
390 for (j
= i
+ 1; j
< t
->desc
->num_columns
; j
++) {
391 const char *name2
= get_column_description(t
->desc
,
393 if (cd
->name
&& name2
&& !strcmp(cd
->name
, name2
))
403 * Initialize a struct table from given table description.
405 * \param desc The description of the osl table.
406 * \param table_ptr Result is returned here.
408 * This function performs several sanity checks on \p desc and returns if any
409 * of these tests fail. On success, a struct \p osl_table is allocated and
410 * initialized with data derived from \p desc.
412 * \return Positive on success, negative on errors. Possible errors include: \p
413 * E_BAD_TABLE_DESC, \p E_NO_COLUMN_DESC, \p E_NO_COLUMNS, \p
414 * E_BAD_STORAGE_TYPE, \p E_BAD_STORAGE_FLAGS, \p E_BAD_STORAGE_SIZE, \p
415 * E_NO_UNIQUE_RBTREE_COLUMN, \p E_NO_RBTREE_COL.
417 * \sa struct osl_table.
419 int init_table_structure(const struct osl_table_description
*desc
,
420 struct osl_table
**table_ptr
)
422 const struct osl_column_description
*cd
;
423 struct osl_table
*t
= para_calloc(sizeof(*t
));
424 int i
, ret
= -E_BAD_TABLE_DESC
, have_disk_storage_name_column
= 0;
428 PARA_DEBUG_LOG("creating table structure for '%s' from table "
429 "description\n", desc
->name
);
430 ret
= -E_NO_COLUMN_DESC
;
431 if (!desc
->column_descriptions
)
434 if (!desc
->num_columns
)
436 t
->columns
= para_calloc(desc
->num_columns
* sizeof(struct osl_column
));
438 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
439 enum osl_storage_type st
= cd
->storage_type
;
440 enum osl_storage_flags sf
= cd
->storage_flags
;
441 struct osl_column
*col
= &t
->columns
[i
];
443 ret
= -E_BAD_STORAGE_TYPE
;
444 if (st
!= OSL_MAPPED_STORAGE
&& st
!= OSL_DISK_STORAGE
445 && st
!= OSL_NO_STORAGE
)
447 ret
= -E_BAD_STORAGE_FLAGS
;
448 if (st
== OSL_DISK_STORAGE
&& sf
& OSL_RBTREE
)
450 ret
= -E_BAD_STORAGE_SIZE
;
451 if (sf
& OSL_FIXED_SIZE
&& !cd
->data_size
)
454 case OSL_DISK_STORAGE
:
455 t
->num_disk_storage_columns
++;
457 case OSL_MAPPED_STORAGE
:
458 t
->num_mapped_columns
++;
459 col
->index_offset
= t
->row_index_size
;
460 t
->row_index_size
+= 8;
463 col
->volatile_num
= t
->num_volatile_columns
;
464 t
->num_volatile_columns
++;
467 if (sf
& OSL_RBTREE
) {
468 col
->rbtree_num
= t
->num_rbtrees
;
470 if ((sf
& OSL_UNIQUE
) && (st
== OSL_MAPPED_STORAGE
)) {
471 if (!have_disk_storage_name_column
)
472 t
->disk_storage_name_column
= i
;
473 have_disk_storage_name_column
= 1;
477 ret
= -E_NO_UNIQUE_RBTREE_COLUMN
;
478 if (t
->num_disk_storage_columns
&& !have_disk_storage_name_column
)
480 ret
= -E_NO_RBTREE_COL
;
484 PARA_DEBUG_LOG("OK. Index entry size: %u\n", t
->row_index_size
);
485 ret
= init_column_descriptions(t
);
497 * Read the table description from index header.
499 * \param map The memory mapping of the index file.
500 * \param desc The values found in the index header are returned here.
502 * Read the index header, check for the paraslash magic string and the table version number.
503 * Read all information stored in the index header into \a desc.
505 * \return Positive on success, negative on errors.
507 * \sa struct osl_table_description, osl_create_table.
509 int read_table_desc(struct osl_object
*map
, struct osl_table_description
*desc
)
511 char *buf
= map
->data
;
513 uint16_t header_size
;
516 struct osl_column_description
*cd
;
518 if (map
->size
< MIN_INDEX_HEADER_SIZE(1))
519 return -E_SHORT_TABLE
;
520 if (strncmp(buf
+ IDX_PARA_MAGIC
, PARA_MAGIC
, strlen(PARA_MAGIC
)))
522 version
= read_u8(buf
+ IDX_VERSION
);
523 if (version
< MIN_TABLE_VERSION
|| version
> MAX_TABLE_VERSION
)
524 return -E_VERSION_MISMATCH
;
525 desc
->num_columns
= read_u8(buf
+ IDX_TABLE_FLAGS
);
526 desc
->flags
= read_u8(buf
+ IDX_TABLE_FLAGS
);
527 desc
->num_columns
= read_u16(buf
+ IDX_NUM_COLUMNS
);
528 PARA_DEBUG_LOG("%u columns\n", desc
->num_columns
);
529 if (!desc
->num_columns
)
530 return -E_NO_COLUMNS
;
531 header_size
= read_u16(buf
+ IDX_HEADER_SIZE
);
532 if (map
->size
< header_size
)
534 desc
->column_descriptions
= para_calloc(desc
->num_columns
535 * sizeof(struct osl_column_description
));
536 offset
= IDX_COLUMN_DESCRIPTIONS
;
537 FOR_EACH_COLUMN(i
, desc
, cd
) {
540 ret
= -E_SHORT_TABLE
;
541 if (map
->size
< offset
+ MIN_IDX_COLUMN_DESCRIPTION_SIZE
) {
542 PARA_ERROR_LOG("map size = %zu < %u = offset + min desc size\n",
543 map
->size
, offset
+ MIN_IDX_COLUMN_DESCRIPTION_SIZE
);
546 cd
->storage_type
= read_u16(buf
+ offset
+ IDX_CD_STORAGE_TYPE
);
547 cd
->storage_flags
= read_u16(buf
+ offset
+
548 IDX_CD_STORAGE_FLAGS
);
549 cd
->data_size
= read_u32(buf
+ offset
+ IDX_CD_DATA_SIZE
);
550 null_byte
= memchr(buf
+ offset
+ IDX_CD_NAME
, '\0',
551 map
->size
- offset
- IDX_CD_NAME
);
552 ret
= -E_INDEX_CORRUPTION
;
555 cd
->name
= para_strdup(buf
+ offset
+ IDX_CD_NAME
);
556 offset
+= index_column_description_size(cd
->name
);
558 if (offset
!= header_size
) {
559 ret
= -E_INDEX_CORRUPTION
;
560 PARA_ERROR_LOG("real header size = %u != %u = stored header size\n",
561 offset
, header_size
);
566 FOR_EACH_COLUMN(i
, desc
, cd
)
572 * check whether the table description given by \p t->desc matches the on-disk
573 * table structure stored in the index of \a t.
575 static int compare_table_descriptions(struct osl_table
*t
)
578 struct osl_table_description desc
;
579 const struct osl_column_description
*cd1
, *cd2
;
581 /* read the on-disk structure into desc */
582 ret
= read_table_desc(&t
->index_map
, &desc
);
585 ret
= -E_BAD_TABLE_FLAGS
;
586 if (desc
.flags
!= t
->desc
->flags
)
588 ret
= -E_BAD_COLUMN_NUM
;
589 if (desc
.num_columns
!= t
->desc
->num_columns
)
591 FOR_EACH_COLUMN(i
, t
->desc
, cd1
) {
592 cd2
= get_column_description(&desc
, i
);
593 ret
= -E_BAD_STORAGE_TYPE
;
594 if (cd1
->storage_type
!= cd2
->storage_type
)
596 ret
= -E_BAD_STORAGE_FLAGS
;
597 if (cd1
->storage_flags
!= cd2
->storage_flags
) {
598 PARA_ERROR_LOG("sf1 = %u != %u = sf2\n",
599 cd1
->storage_flags
, cd2
->storage_flags
);
602 ret
= -E_BAD_DATA_SIZE
;
603 if (cd1
->storage_flags
& OSL_FIXED_SIZE
)
604 if (cd1
->data_size
!= cd2
->data_size
)
606 ret
= -E_BAD_COLUMN_NAME
;
607 if (strcmp(cd1
->name
, cd2
->name
))
610 PARA_DEBUG_LOG("table description of '%s' matches on-disk data, good\n",
614 FOR_EACH_COLUMN(i
, &desc
, cd1
)
616 free(desc
.column_descriptions
);
620 static int create_table_index(struct osl_table
*t
)
622 char *buf
, *filename
;
624 size_t size
= t
->index_header_size
;
625 const struct osl_column_description
*cd
;
628 PARA_INFO_LOG("creating %zu byte index for table %s\n", size
,
630 buf
= para_calloc(size
);
631 sprintf(buf
+ IDX_PARA_MAGIC
, "%s", PARA_MAGIC
);
632 write_u8(buf
+ IDX_TABLE_FLAGS
, t
->desc
->flags
);
633 write_u8(buf
+ IDX_DIRTY_FLAG
, 0);
634 write_u8(buf
+ IDX_VERSION
, CURRENT_TABLE_VERSION
);
635 write_u16(buf
+ IDX_NUM_COLUMNS
, t
->desc
->num_columns
);
636 write_u16(buf
+ IDX_HEADER_SIZE
, t
->index_header_size
);
637 offset
= IDX_COLUMN_DESCRIPTIONS
;
638 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
639 write_u16(buf
+ offset
+ IDX_CD_STORAGE_TYPE
,
641 write_u16(buf
+ offset
+ IDX_CD_STORAGE_FLAGS
,
643 if (cd
->storage_flags
& OSL_FIXED_SIZE
)
644 write_u32(buf
+ offset
+ IDX_CD_DATA_SIZE
,
646 strcpy(buf
+ offset
+ IDX_CD_NAME
, cd
->name
);
647 offset
+= index_column_description_size(cd
->name
);
649 assert(offset
= size
);
650 filename
= index_filename(t
->desc
);
651 ret
= para_write_file(filename
, buf
, size
);
658 * Create a new osl table.
660 * \param desc Pointer to the table description.
662 * \return Positive on success, negative on errors. Possible errors include: \p
663 * E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_BAD_NAME, \p E_NO_COMPARE_FUNC, \p
664 * E_NO_COLUMN_NAME, \p E_DUPLICATE_COL_NAME, \p E_MKDIR, any errors returned
667 int osl_create_table(const struct osl_table_description
*desc
)
669 const struct osl_column_description
*cd
;
670 char *table_dir
= NULL
, *filename
;
672 int i
, ret
= init_table_structure(desc
, &t
);
676 PARA_INFO_LOG("creating %s\n", desc
->name
);
677 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
678 if (cd
->storage_type
== OSL_NO_STORAGE
)
681 ret
= para_mkdir(desc
->dir
, 0777);
682 if (ret
< 0 && !is_errno(-ret
, EEXIST
))
684 table_dir
= make_message("%s/%s", desc
->dir
,
686 ret
= para_mkdir(table_dir
, 0777);
690 filename
= column_filename(t
, i
);
691 PARA_INFO_LOG("filename: %s\n", filename
);
692 if (cd
->storage_type
== OSL_MAPPED_STORAGE
) {
693 ret
= para_open(filename
, O_RDWR
| O_CREAT
| O_EXCL
,
702 ret
= para_mkdir(filename
, 0777);
707 if (t
->num_mapped_columns
) {
708 ret
= create_table_index(t
);
720 static int table_is_dirty(struct osl_table
*t
)
722 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
723 uint8_t dirty
= read_u8(buf
) & 0x1;
727 static void mark_table_dirty(struct osl_table
*t
)
729 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
730 write_u8(buf
, read_u8(buf
) | 1);
733 static void mark_table_clean(struct osl_table
*t
)
735 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
736 write_u8(buf
, read_u8(buf
) & 0xfe);
739 static void unmap_column(struct osl_table
*t
, unsigned col_num
)
741 struct osl_object map
= t
->columns
[col_num
].data_map
;
745 ret
= para_munmap(map
.data
, map
.size
);
751 * Unmap all mapped files of an osl table.
753 * \param t Pointer to a mapped table.
754 * \param flags Options for unmapping.
756 * \return Positive on success, negative on errors.
758 * \sa map_table(), enum osl_close_flags, para_munmap().
760 int unmap_table(struct osl_table
*t
, enum osl_close_flags flags
)
763 const struct osl_column_description
*cd
;
766 if (!t
->num_mapped_columns
) /* can this ever happen? */
768 PARA_DEBUG_LOG("unmapping table '%s'\n", t
->desc
->name
);
769 if (!t
->index_map
.data
)
770 return -E_NOT_MAPPED
;
771 if (flags
& OSL_MARK_CLEAN
)
773 ret
= para_munmap(t
->index_map
.data
, t
->index_map
.size
);
776 t
->index_map
.data
= NULL
;
779 FOR_EACH_MAPPED_COLUMN(i
, t
, cd
)
784 static int map_column(struct osl_table
*t
, unsigned col_num
)
787 char *filename
= column_filename(t
, col_num
);
789 if (stat(filename
, &statbuf
) < 0) {
793 if (!(S_IFREG
& statbuf
.st_mode
)) {
797 ret
= mmap_full_file(filename
, O_RDWR
,
798 &t
->columns
[col_num
].data_map
);
804 * Map the index file and all columns of type \p OSL_MAPPED_STORAGE into memory.
806 * \param t Pointer to an initialized table structure.
807 * \param flags Mapping options.
809 * \return Negative return value on errors; on success the number of rows
810 * (including invalid rows) is returned.
812 * \sa unmap_table(), enum map_table_flags, osl_open_table(), mmap(2).
814 int map_table(struct osl_table
*t
, enum map_table_flags flags
)
817 const struct osl_column_description
*cd
;
818 int i
= 0, ret
, num_rows
= 0;
820 if (!t
->num_mapped_columns
)
822 if (t
->index_map
.data
)
823 return -E_ALREADY_MAPPED
;
824 filename
= index_filename(t
->desc
);
825 PARA_DEBUG_LOG("mapping table '%s' (index: %s)\n", t
->desc
->name
, filename
);
826 ret
= mmap_full_file(filename
, flags
& MAP_TBL_FL_MAP_RDONLY
?
827 O_RDONLY
: O_RDWR
, &t
->index_map
);
831 if (flags
& MAP_TBL_FL_VERIFY_INDEX
) {
832 ret
= compare_table_descriptions(t
);
837 if (!(flags
& MAP_TBL_FL_IGNORE_DIRTY
)) {
838 if (table_is_dirty(t
)) {
839 PARA_ERROR_LOG("%s is dirty\n", t
->desc
->name
);
844 num_rows
= table_num_rows(t
);
848 FOR_EACH_MAPPED_COLUMN(i
, t
, cd
) {
849 ret
= map_column(t
, i
);
854 err
: /* unmap what is already mapped */
855 for (i
--; i
>= 0; i
--) {
856 struct osl_object map
= t
->columns
[i
].data_map
;
857 para_munmap(map
.data
, map
.size
);
860 para_munmap(t
->index_map
.data
, t
->index_map
.size
);
861 t
->index_map
.data
= NULL
;
866 * Retrieve a mapped object by row and column number.
868 * \param t Pointer to an open osl table.
869 * \param col_num Number of the mapped column containing the object to retrieve.
870 * \param row_num Number of the row containing the object to retrieve.
871 * \param obj The result is returned here.
873 * It is considered an error if \a col_num does not refer to a column
874 * of storage type \p OSL_MAPPED_STORAGE.
876 * \return Positive on success, negative on errors. Possible errors include:
877 * \p E_BAD_ROW_NUM, \p E_INVALID_OBJECT.
879 * \sa osl_storage_type.
881 int get_mapped_object(const struct osl_table
*t
, unsigned col_num
,
882 uint32_t row_num
, struct osl_object
*obj
)
884 struct osl_column
*col
= &t
->columns
[col_num
];
890 if (t
->num_rows
<= row_num
)
891 return -E_BAD_ROW_NUM
;
892 ret
= get_cell_index(t
, row_num
, col_num
, &cell_index
);
895 offset
= read_u32(cell_index
);
896 obj
->size
= read_u32(cell_index
+ 4) - 1;
897 header
= col
->data_map
.data
+ offset
;
898 obj
->data
= header
+ 1;
899 if (read_u8(header
) == 0xff) {
900 PARA_ERROR_LOG("col %u, size %zu, offset %u\n", col_num
,
902 return -E_INVALID_OBJECT
;
907 static int search_rbtree(const struct osl_object
*obj
,
908 const struct osl_table
*t
, unsigned col_num
,
909 struct rb_node
**result
, struct rb_node
***rb_link
)
911 struct osl_column
*col
= &t
->columns
[col_num
];
912 struct rb_node
**new = &col
->rbtree
.rb_node
, *parent
= NULL
;
913 const struct osl_column_description
*cd
=
914 get_column_description(t
->desc
, col_num
);
915 enum osl_storage_type st
= cd
->storage_type
;
917 struct osl_row
*this_row
= get_row_pointer(*new,
920 struct osl_object this_obj
;
922 if (st
== OSL_MAPPED_STORAGE
) {
923 ret
= get_mapped_object(t
, col_num
, this_row
->num
,
928 this_obj
= this_row
->volatile_objects
[col
->volatile_num
];
929 ret
= cd
->compare_function(obj
, &this_obj
);
932 *result
= get_rb_node_pointer(this_row
,
937 new = &((*new)->rb_left
);
939 new = &((*new)->rb_right
);
945 return -E_RB_KEY_NOT_FOUND
;
948 static int insert_rbtree(struct osl_table
*t
, unsigned col_num
,
949 const struct osl_row
*row
, const struct osl_object
*obj
)
951 struct rb_node
*parent
, **rb_link
;
954 int ret
= search_rbtree(obj
, t
, col_num
, &parent
, &rb_link
);
957 return -E_RB_KEY_EXISTS
;
958 rbtree_num
= t
->columns
[col_num
].rbtree_num
;
959 n
= get_rb_node_pointer(row
, rbtree_num
);
960 rb_link_node(n
, parent
, rb_link
);
961 rb_insert_color(n
, &t
->columns
[col_num
].rbtree
);
965 static void remove_rb_node(struct osl_table
*t
, unsigned col_num
,
966 const struct osl_row
*row
)
968 struct osl_column
*col
= &t
->columns
[col_num
];
969 const struct osl_column_description
*cd
=
970 get_column_description(t
->desc
, col_num
);
971 enum osl_storage_flags sf
= cd
->storage_flags
;
972 struct rb_node
*victim
, *splice_out_node
, *tmp
;
973 if (!(sf
& OSL_RBTREE
))
976 * Which node is removed/spliced out actually depends on how many
977 * children the victim node has: If it has no children, it gets
978 * deleted. If it has one child, it gets spliced out. If it has two
979 * children, its successor (which has at most a right child) gets
982 victim
= get_rb_node_pointer(row
, col
->rbtree_num
);
983 if (victim
->rb_left
&& victim
->rb_right
)
984 splice_out_node
= rb_next(victim
);
986 splice_out_node
= victim
;
987 /* Go up to the root and decrement the size of each node in the path. */
988 for (tmp
= splice_out_node
; tmp
; tmp
= rb_parent(tmp
))
990 rb_erase(victim
, &col
->rbtree
);
993 static int add_row_to_rbtrees(struct osl_table
*t
, uint32_t row_num
,
994 struct osl_object
*volatile_objs
, struct osl_row
**row_ptr
)
998 struct osl_row
*row
= allocate_row(t
->num_rbtrees
);
999 const struct osl_column_description
*cd
;
1002 row
->volatile_objects
= volatile_objs
;
1003 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
) {
1004 if (cd
->storage_type
== OSL_MAPPED_STORAGE
) {
1005 struct osl_object obj
;
1006 ret
= get_mapped_object(t
, i
, row_num
, &obj
);
1009 ret
= insert_rbtree(t
, i
, row
, &obj
);
1010 } else { /* volatile */
1011 const struct osl_object
*obj
1012 = volatile_objs
+ t
->columns
[i
].volatile_num
;
1013 ret
= insert_rbtree(t
, i
, row
, obj
);
1021 err
: /* rollback changes, i.e. remove added entries from rbtrees */
1023 remove_rb_node(t
, i
--, row
);
1028 static void free_volatile_objects(const struct osl_table
*t
,
1029 enum osl_close_flags flags
)
1033 struct osl_column
*rb_col
;
1034 const struct osl_column_description
*cd
;
1036 if (!t
->num_volatile_columns
)
1038 /* find the first rbtree column (any will do) */
1039 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
)
1041 rb_col
= t
->columns
+ i
;
1042 /* walk that rbtree and free all volatile objects */
1043 for (n
= rb_first(&rb_col
->rbtree
); n
; n
= rb_next(n
)) {
1044 struct osl_row
*r
= get_row_pointer(n
, rb_col
->rbtree_num
);
1045 if (flags
& OSL_FREE_VOLATILE
)
1046 for (j
= 0; j
< t
->num_volatile_columns
; j
++)
1047 free(r
->volatile_objects
[j
].data
);
1048 free(r
->volatile_objects
);
1053 * Erase all rbtree nodes and free resources.
1055 * \param t Pointer to an open osl table.
1057 * This function is called by osl_close_table().
1059 void clear_rbtrees(struct osl_table
*t
)
1061 const struct osl_column_description
*cd
;
1062 unsigned i
, rbtrees_cleared
= 0;
1064 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
) {
1065 struct osl_column
*col
= &t
->columns
[i
];
1068 for (n
= rb_first(&col
->rbtree
); n
;) {
1070 rb_erase(n
, &col
->rbtree
);
1071 if (rbtrees_cleared
== t
->num_rbtrees
) {
1072 r
= get_row_pointer(n
, col
->rbtree_num
);
1083 * Close an osl table.
1085 * \param t Pointer to the table to be closed.
1086 * \param flags Options for what should be cleaned up.
1088 * If osl_open_table() succeeds, the resulting table pointer must later be
1089 * passed to this function in order to flush all changes to the file system and
1090 * to free the resources that were allocated by osl_open_table().
1092 * \return Positive on success, negative on errors. Possible errors: \p E_BAD_TABLE,
1093 * errors returned by unmap_table().
1095 * \sa osl_open_table(), unmap_table().
1097 int osl_close_table(struct osl_table
*t
, enum osl_close_flags flags
)
1102 return -E_BAD_TABLE
;
1103 free_volatile_objects(t
, flags
);
1105 ret
= unmap_table(t
, flags
);
1107 PARA_ERROR_LOG("unmap_table failed: %d\n", ret
);
1114 * Find out whether the given row number corresponds to an invalid row.
1116 * \param t Pointer to the osl table.
1117 * \param row_num The number of the row in question.
1119 * By definition, a row is considered invalid if all its index entries
1122 * \return Positive if \a row_num corresponds to an invalid row,
1123 * zero if it corresponds to a valid row, negative on errors.
1125 int row_is_invalid(struct osl_table
*t
, uint32_t row_num
)
1128 int i
, ret
= get_row_index(t
, row_num
, &row_index
);
1132 for (i
= 0; i
< t
->row_index_size
; i
++) {
1133 if ((unsigned char)row_index
[i
] != 0xff)
1136 PARA_INFO_LOG("row %d is invalid\n", row_num
);
1141 * Invalidate a row of an osl table.
1143 * \param t Pointer to an open osl table.
1144 * \param row_num Number of the row to mark as invalid.
1146 * This function marks each mapped object in the index entry of \a row as
1149 * \return Positive on success, negative on errors.
1151 int mark_row_invalid(struct osl_table
*t
, uint32_t row_num
)
1154 int ret
= get_row_index(t
, row_num
, &row_index
);
1158 PARA_INFO_LOG("marking row %d as invalid\n", row_num
);
1159 memset(row_index
, 0xff, t
->row_index_size
);
1164 * Initialize all rbtrees and compute number of invalid rows.
1166 * \param t The table containing the rbtrees to be initialized.
1168 * \return Positive on success, negative on errors.
1170 int init_rbtrees(struct osl_table
*t
)
1173 const struct osl_column_description
*cd
;
1175 /* create rbtrees */
1176 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
)
1177 t
->columns
[i
].rbtree
= RB_ROOT
;
1178 /* add valid rows to rbtrees */
1179 t
->num_invalid_rows
= 0;
1180 for (i
= 0; i
< t
->num_rows
; i
++) {
1181 ret
= row_is_invalid(t
, i
);
1185 t
->num_invalid_rows
++;
1188 ret
= add_row_to_rbtrees(t
, i
, NULL
, NULL
);
1196 * Open an osl table.
1198 * Each osl table must be opened before its data can be accessed.
1200 * \param table_desc Describes the table to be opened.
1201 * \param result Contains a pointer to the open table on success.
1203 * The table description given by \a desc should coincide with the
1204 * description used at creation time.
1206 * \return Positive on success, negative on errors. Possible errors include:
1207 * errors returned by init_table_structure(), \p E_NOENT, \p E_STAT, \p \p
1208 * E_NOTDIR, \p E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_NO_COMPARE_FUNC, \p
1209 * E_NO_COLUMN_NAME, errors returned by init_rbtrees().
1211 int osl_open_table(const struct osl_table_description
*table_desc
,
1212 struct osl_table
**result
)
1215 struct osl_table
*t
;
1216 const struct osl_column_description
*cd
;
1218 PARA_INFO_LOG("opening table %s\n", table_desc
->name
);
1219 ret
= init_table_structure(table_desc
, &t
);
1222 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
) {
1223 /* check if directory exists */
1224 char *dirname
= column_filename(t
, i
);
1225 struct stat statbuf
;
1226 ret
= stat(dirname
, &statbuf
);
1229 ret
= -ERRNO_TO_PARA_ERROR(errno
);
1232 ret
= -ERRNO_TO_PARA_ERROR(ENOTDIR
);
1233 if (!S_ISDIR(statbuf
.st_mode
))
1236 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1240 PARA_DEBUG_LOG("num rows: %d\n", t
->num_rows
);
1241 ret
= init_rbtrees(t
);
1243 osl_close_table(t
, OSL_MARK_CLEAN
); /* ignore further errors */
1254 static int create_disk_storage_object_dir(const struct osl_table
*t
,
1255 unsigned col_num
, const char *ds_name
)
1260 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1262 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1263 ret
= para_mkdir(dirname
, 0777);
1265 if (ret
< 0 && !is_errno(-ret
, EEXIST
))
1270 static int write_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1271 const struct osl_object
*obj
, const char *ds_name
)
1276 ret
= create_disk_storage_object_dir(t
, col_num
, ds_name
);
1279 filename
= disk_storage_path(t
, col_num
, ds_name
);
1280 ret
= para_write_file(filename
, obj
->data
, obj
->size
);
1285 static int append_map_file(const struct osl_table
*t
, unsigned col_num
,
1286 const struct osl_object
*obj
, uint32_t *new_size
)
1288 char *filename
= column_filename(t
, col_num
);
1290 char header
= 0; /* zero means valid object */
1292 // PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj->size);
1293 ret
= append_file(filename
, &header
, 1, obj
->data
, obj
->size
,
1299 static int append_row_index(const struct osl_table
*t
, char *row_index
)
1304 if (!t
->num_mapped_columns
)
1306 filename
= index_filename(t
->desc
);
1307 ret
= append_file(filename
, NULL
, 0, row_index
,
1308 t
->row_index_size
, NULL
);
1314 * A wrapper for truncate(2)
1316 * \param path Name of the regular file to truncate
1317 * \param size Number of bytes to \b shave \b off
1319 * Truncate the regular file named by \a path by \a size bytes.
1321 * \return Positive on success, negative on errors. Possible errors include: \p
1322 * E_STAT, \p E_BAD_SIZE, \p E_TRUNC.
1326 int para_truncate(const char *path
, off_t size
)
1329 struct stat statbuf
;
1332 if (stat(path
, &statbuf
) < 0)
1335 if (statbuf
.st_size
< size
)
1338 if (truncate(path
, statbuf
.st_size
- size
) < 0)
1345 static int truncate_mapped_file(const struct osl_table
*t
, unsigned col_num
,
1348 char *filename
= column_filename(t
, col_num
);
1349 int ret
= para_truncate(filename
, size
);
1354 static int delete_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1355 const char *ds_name
)
1357 char *dirname
, *filename
= disk_storage_path(t
, col_num
, ds_name
);
1358 int ret
= unlink(filename
), err
= errno
;
1362 return -ERRNO_TO_PARA_ERROR(err
);
1363 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1365 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1372 * Add a new row to an osl table and retrieve this row.
1374 * \param t Pointer to an open osl table.
1375 * \param objects Array of objects to be added.
1376 * \param row Result pointer.
1378 * The \a objects parameter must point to an array containing one object per
1379 * column. The order of the objects in the array is given by the table
1380 * description of \a table. Several sanity checks are performed during object
1381 * insertion and the function returns without modifying the table if any of
1382 * these tests fail. In fact, it is atomic in the sense that it either
1383 * succeeds or leaves the table unchanged (i.e. either all or none of the
1384 * objects are added to the table).
1386 * It is considered an error if an object is added to a column with associated
1387 * rbtree if this object is equal to an object already contained in that column
1388 * (i.e. the compare function for the column's rbtree returns zero).
1390 * Possible errors include: \p E_RB_KEY_EXISTS, \p E_BAD_DATA_SIZE.
1392 * \return Positive on success, negative on errors.
1394 * \sa struct osl_table_description, osl_compare_func, osl_add_row().
1396 int osl_add_and_get_row(struct osl_table
*t
, struct osl_object
*objects
,
1397 struct osl_row
**row
)
1400 char *ds_name
= NULL
;
1401 struct rb_node
**rb_parents
= NULL
, ***rb_links
= NULL
;
1402 char *new_row_index
= NULL
;
1403 struct osl_object
*volatile_objs
= NULL
;
1404 const struct osl_column_description
*cd
;
1407 return -E_BAD_TABLE
;
1408 rb_parents
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
*));
1409 rb_links
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
**));
1410 if (t
->num_mapped_columns
)
1411 new_row_index
= para_malloc(t
->row_index_size
);
1412 /* pass 1: sanity checks */
1413 // PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
1414 // objects[1].data);
1415 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1416 enum osl_storage_type st
= cd
->storage_type
;
1417 enum osl_storage_flags sf
= cd
->storage_flags
;
1419 // ret = -E_NULL_OBJECT;
1422 if (st
== OSL_DISK_STORAGE
)
1424 if (sf
& OSL_RBTREE
) {
1425 unsigned rbtree_num
= t
->columns
[i
].rbtree_num
;
1426 ret
= -E_RB_KEY_EXISTS
;
1427 // PARA_DEBUG_LOG("checking whether %p exists\n",
1428 // objects[i].data);
1429 if (search_rbtree(objects
+ i
, t
, i
,
1430 &rb_parents
[rbtree_num
],
1431 &rb_links
[rbtree_num
]) > 0)
1434 if (sf
& OSL_FIXED_SIZE
) {
1435 // PARA_DEBUG_LOG("fixed size. need: %zu, have: %d\n",
1436 // objects[i].size, cd->data_size);
1437 ret
= -E_BAD_DATA_SIZE
;
1438 if (objects
[i
].size
!= cd
->data_size
)
1442 if (t
->num_disk_storage_columns
)
1443 ds_name
= disk_storage_name_of_object(t
,
1444 &objects
[t
->disk_storage_name_column
]);
1445 ret
= unmap_table(t
, OSL_MARK_CLEAN
);
1448 // PARA_DEBUG_LOG("sanity tests passed%s\n", "");
1449 /* pass 2: create data files, append map data */
1450 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1451 enum osl_storage_type st
= cd
->storage_type
;
1452 if (st
== OSL_NO_STORAGE
)
1454 if (st
== OSL_MAPPED_STORAGE
) {
1456 struct osl_column
*col
= &t
->columns
[i
];
1457 // PARA_DEBUG_LOG("appending object of size %zu\n",
1458 // objects[i].size);
1459 ret
= append_map_file(t
, i
, objects
+ i
, &new_size
);
1462 update_cell_index(new_row_index
, col
, new_size
,
1467 ret
= write_disk_storage_file(t
, i
, objects
+ i
, ds_name
);
1471 ret
= append_row_index(t
, new_row_index
);
1474 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1475 if (ret
< 0) { /* truncate index and rollback changes */
1476 char *filename
= index_filename(t
->desc
);
1477 para_truncate(filename
, t
->row_index_size
);
1481 /* pass 3: add entry to rbtrees */
1482 if (t
->num_volatile_columns
) {
1483 volatile_objs
= para_calloc(t
->num_volatile_columns
1484 * sizeof(struct osl_object
));
1485 FOR_EACH_VOLATILE_COLUMN(i
, t
, cd
)
1486 volatile_objs
[t
->columns
[i
].volatile_num
] = objects
[i
];
1489 // PARA_DEBUG_LOG("adding new entry as row #%d\n", t->num_rows - 1);
1490 ret
= add_row_to_rbtrees(t
, t
->num_rows
- 1, volatile_objs
, row
);
1493 // PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
1496 rollback
: /* rollback all changes made, ignore further errors */
1497 for (i
--; i
>= 0; i
--) {
1498 cd
= get_column_description(t
->desc
, i
);
1499 enum osl_storage_type st
= cd
->storage_type
;
1500 if (st
== OSL_NO_STORAGE
)
1503 if (st
== OSL_MAPPED_STORAGE
)
1504 truncate_mapped_file(t
, i
, objects
[i
].size
);
1505 else /* disk storage */
1506 delete_disk_storage_file(t
, i
, ds_name
);
1508 /* ignore error and return previous error value */
1509 map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1511 free(new_row_index
);
1519 * Add a new row to an osl table.
1521 * \param t Same meaning as osl_add_and_get_row().
1522 * \param objects Same meaning as osl_add_and_get_row().
1524 * \return The return value of the underlying call to osl_add_and_get_row().
1526 * This is equivalent to osl_add_and_get_row(t, objects, NULL).
1528 int osl_add_row(struct osl_table
*t
, struct osl_object
*objects
)
1530 return osl_add_and_get_row(t
, objects
, NULL
);
1534 * Retrieve an object identified by row and column
1536 * \param t Pointer to an open osl table.
1537 * \param r Pointer to the row.
1538 * \param col_num The column number.
1539 * \param object The result pointer.
1541 * The column determined by \a col_num must be of type \p OSL_MAPPED_STORAGE
1542 * or \p OSL_NO_STORAGE, i.e. no disk storage objects may be retrieved by this
1545 * \return Positive if object was found, negative on errors. Possible errors
1546 * include: \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE.
1548 * \sa osl_storage_type, osl_open_disk_object().
1550 int osl_get_object(const struct osl_table
*t
, const struct osl_row
*r
,
1551 unsigned col_num
, struct osl_object
*object
)
1553 const struct osl_column_description
*cd
;
1556 return -E_BAD_TABLE
;
1557 cd
= get_column_description(t
->desc
, col_num
);
1558 /* col must not be disk storage */
1559 if (cd
->storage_type
== OSL_DISK_STORAGE
)
1560 return -E_BAD_STORAGE_TYPE
;
1561 if (cd
->storage_type
== OSL_MAPPED_STORAGE
)
1562 return get_mapped_object(t
, col_num
, r
->num
, object
);
1564 *object
= r
->volatile_objects
[t
->columns
[col_num
].volatile_num
];
1568 static int mark_mapped_object_invalid(const struct osl_table
*t
,
1569 uint32_t row_num
, unsigned col_num
)
1571 struct osl_object obj
;
1573 int ret
= get_mapped_object(t
, col_num
, row_num
, &obj
);
1584 * Delete a row from an osl table.
1586 * \param t Pointer to an open osl table.
1587 * \param row Pointer to the row to delete.
1589 * This removes all disk storage objects, removes all rbtree nodes, and frees
1590 * all volatile objects belonging to the given row. For mapped columns, the
1591 * data is merely marked invalid and may be pruned from time to time by
1594 * \return Positive on success, negative on errors. Possible errors include:
1595 * \p E_BAD_TABLE, errors returned by osl_get_object().
1597 int osl_del_row(struct osl_table
*t
, struct osl_row
*row
)
1599 struct osl_row
*r
= row
;
1601 const struct osl_column_description
*cd
;
1604 return -E_BAD_TABLE
;
1605 PARA_INFO_LOG("deleting row %p\n", row
);
1607 if (t
->num_disk_storage_columns
) {
1609 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1612 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
)
1613 delete_disk_storage_file(t
, i
, ds_name
);
1616 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1617 struct osl_column
*col
= t
->columns
+ i
;
1618 enum osl_storage_type st
= cd
->storage_type
;
1619 remove_rb_node(t
, i
, r
);
1620 if (st
== OSL_MAPPED_STORAGE
) {
1621 mark_mapped_object_invalid(t
, r
->num
, i
);
1624 if (st
== OSL_NO_STORAGE
)
1625 free(r
->volatile_objects
[col
->volatile_num
].data
);
1627 if (t
->num_mapped_columns
) {
1628 ret
= mark_row_invalid(t
, r
->num
);
1631 t
->num_invalid_rows
++;
1636 free(r
->volatile_objects
);
1641 /* test if column has an rbtree */
1642 static int check_rbtree_col(const struct osl_table
*t
, unsigned col_num
,
1643 struct osl_column
**col
)
1646 return -E_BAD_TABLE
;
1647 if (!(get_column_description(t
->desc
, col_num
)->storage_flags
& OSL_RBTREE
))
1648 return -E_BAD_STORAGE_FLAGS
;
1649 *col
= t
->columns
+ col_num
;
1654 * Get the row that contains the given object.
1656 * \param t Pointer to an open osl table.
1657 * \param col_num The number of the column to be searched.
1658 * \param obj The object to be looked up.
1659 * \param result Points to the row containing \a obj.
1661 * Lookup \a obj in \a t and return the row containing \a obj. The column
1662 * specified by \a col_num must have an associated rbtree.
1664 * \return Positive on success, negative on errors. If an error occurred, \a
1665 * result is set to \p NULL. Possible errors include: \p E_BAD_TABLE, \p
1666 * E_BAD_STORAGE_FLAGS, errors returned by get_mapped_object(), \p
1667 * E_RB_KEY_NOT_FOUND.
1669 * \sa osl_storage_flags
1671 int osl_get_row(const struct osl_table
*t
, unsigned col_num
,
1672 const struct osl_object
*obj
, struct osl_row
**result
)
1675 struct rb_node
*node
;
1676 struct osl_row
*row
;
1677 struct osl_column
*col
;
1680 ret
= check_rbtree_col(t
, col_num
, &col
);
1683 ret
= search_rbtree(obj
, t
, col_num
, &node
, NULL
);
1686 row
= get_row_pointer(node
, t
->columns
[col_num
].rbtree_num
);
1691 static int rbtree_loop(struct osl_column
*col
, void *private_data
,
1692 osl_rbtree_loop_func
*func
)
1694 struct rb_node
*n
, *tmp
;
1696 /* this for-loop is safe against removal of an entry */
1697 for (n
= rb_first(&col
->rbtree
), tmp
= n
? rb_next(n
) : NULL
;
1699 n
= tmp
, tmp
= tmp
? rb_next(tmp
) : NULL
) {
1700 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1701 int ret
= func(r
, private_data
);
1708 static int rbtree_loop_reverse(struct osl_column
*col
, void *private_data
,
1709 osl_rbtree_loop_func
*func
)
1711 struct rb_node
*n
, *tmp
;
1713 /* safe against removal of an entry */
1714 for (n
= rb_last(&col
->rbtree
), tmp
= n
? rb_prev(n
) : NULL
;
1716 n
= tmp
, tmp
= tmp
? rb_prev(tmp
) : NULL
) {
1717 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1718 int ret
= func(r
, private_data
);
1726 * Loop over all nodes in an rbtree.
1728 * \param t Pointer to an open osl table.
1729 * \param col_num The column to use for iterating over the elements.
1730 * \param private_data Pointer that gets passed to \a func.
1731 * \param func The function to be called for each node in the rbtree.
1733 * This function does an in-order walk of the rbtree associated with \a
1734 * col_num. It is an error if the \p OSL_RBTREE flag is not set for this
1735 * column. For each node in the rbtree, the given function \a func is called
1736 * with two pointers as arguments: The first osl_row* argument points to the
1737 * row that contains the object corresponding to the rbtree node currently
1738 * traversed, and the \a private_data pointer is passed verbatim to \a func as the
1739 * second argument. The loop terminates either if \a func returns a negative
1740 * value, or if all nodes of the tree have been visited.
1743 * \return Positive on success, negative on errors. If the termination of the
1744 * loop was caused by \a func returning a negative value, this value is
1747 * \sa osl_storage_flags, osl_rbtree_loop_reverse(), osl_compare_func.
1749 int osl_rbtree_loop(const struct osl_table
*t
, unsigned col_num
,
1750 void *private_data
, osl_rbtree_loop_func
*func
)
1752 struct osl_column
*col
;
1754 int ret
= check_rbtree_col(t
, col_num
, &col
);
1757 return rbtree_loop(col
, private_data
, func
);
1761 * Loop over all nodes in an rbtree in reverse order.
1763 * \param t Identical meaning as in \p osl_rbtree_loop().
1764 * \param col_num Identical meaning as in \p osl_rbtree_loop().
1765 * \param private_data Identical meaning as in \p osl_rbtree_loop().
1766 * \param func Identical meaning as in \p osl_rbtree_loop().
1768 * This function is identical to \p osl_rbtree_loop(), the only difference
1769 * is that the tree is walked in reverse order.
1771 * \return The same return value as \p osl_rbtree_loop().
1773 * \sa osl_rbtree_loop().
1775 int osl_rbtree_loop_reverse(const struct osl_table
*t
, unsigned col_num
,
1776 void *private_data
, osl_rbtree_loop_func
*func
)
1778 struct osl_column
*col
;
1780 int ret
= check_rbtree_col(t
, col_num
, &col
);
1783 return rbtree_loop_reverse(col
, private_data
, func
);
1786 /* TODO: Rollback changes on errors */
1787 static int rename_disk_storage_objects(struct osl_table
*t
,
1788 struct osl_object
*old_obj
, struct osl_object
*new_obj
)
1791 const struct osl_column_description
*cd
;
1792 char *old_ds_name
, *new_ds_name
;
1794 if (!t
->num_disk_storage_columns
)
1795 return 1; /* nothing to do */
1796 if (old_obj
->size
== new_obj
->size
&& !memcmp(new_obj
->data
,
1797 old_obj
->data
, new_obj
->size
))
1798 return 1; /* object did not change */
1799 old_ds_name
= disk_storage_name_of_object(t
, old_obj
);
1800 new_ds_name
= disk_storage_name_of_object(t
, new_obj
);
1801 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
) {
1802 char *old_filename
, *new_filename
;
1803 ret
= create_disk_storage_object_dir(t
, i
, new_ds_name
);
1806 old_filename
= disk_storage_path(t
, i
, old_ds_name
);
1807 new_filename
= disk_storage_path(t
, i
, new_ds_name
);
1808 ret
= para_rename(old_filename
, new_filename
);
1823 * Change an object in an osl table.
1825 * \param t Pointer to an open osl table.
1826 * \param r Pointer to the row containing the object to be updated.
1827 * \param col_num Number of the column containing the object to be updated.
1828 * \param obj Pointer to the replacement object.
1830 * This function gets rid of all references to the old object. This includes
1831 * removal of the rbtree node in case there is an rbtree associated with \a
1832 * col_num. It then inserts \a obj into the table and the rbtree if necessary.
1834 * If the \p OSL_RBTREE flag is set for \a col_num, you \b MUST call this
1835 * function in order to change the contents of an object, even for volatile or
1836 * mapped columns of constant size (which may be updated directly if \p
1837 * OSL_RBTREE is not set). Otherwise the rbtree might become corrupted.
1839 * \return Positive on success, negative on errors. Possible errors include: \p
1840 * E_BAD_TABLE, \p E_RB_KEY_EXISTS, \p E_BAD_SIZE, \p E_NOENT, \p E_UNLINK,
1841 * errors returned by para_write_file(), \p E_MKDIR.
1843 int osl_update_object(struct osl_table
*t
, const struct osl_row
*r
,
1844 unsigned col_num
, struct osl_object
*obj
)
1846 struct osl_column
*col
;
1847 const struct osl_column_description
*cd
;
1851 return -E_BAD_TABLE
;
1852 col
= &t
->columns
[col_num
];
1853 cd
= get_column_description(t
->desc
, col_num
);
1854 PARA_DEBUG_LOG("updating column %u of %s\n", col_num
, t
->desc
->name
);
1855 if (cd
->storage_flags
& OSL_RBTREE
) {
1856 if (search_rbtree(obj
, t
, col_num
, NULL
, NULL
) > 0)
1857 return -E_RB_KEY_EXISTS
;
1859 if (cd
->storage_flags
& OSL_FIXED_SIZE
) {
1860 if (obj
->size
!= cd
->data_size
)
1861 return -E_BAD_DATA_SIZE
;
1863 remove_rb_node(t
, col_num
, r
);
1864 if (cd
->storage_type
== OSL_NO_STORAGE
) { /* TODO: If fixed size, reuse object? */
1865 free(r
->volatile_objects
[col
->volatile_num
].data
);
1866 r
->volatile_objects
[col
->volatile_num
] = *obj
;
1867 } else if (cd
->storage_type
== OSL_DISK_STORAGE
) {
1869 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1872 ret
= delete_disk_storage_file(t
, col_num
, ds_name
);
1873 if (ret
< 0 && !is_errno(-ret
, ENOENT
)) {
1877 ret
= write_disk_storage_file(t
, col_num
, obj
, ds_name
);
1881 } else { /* mapped storage */
1882 struct osl_object old_obj
;
1883 ret
= get_mapped_object(t
, col_num
, r
->num
, &old_obj
);
1887 * If the updated column is the disk storage name column, the
1888 * disk storage name changes, so we have to rename all disk
1889 * storage objects accordingly.
1891 if (col_num
== t
->disk_storage_name_column
) {
1892 ret
= rename_disk_storage_objects(t
, &old_obj
, obj
);
1896 if (cd
->storage_flags
& OSL_FIXED_SIZE
)
1897 memcpy(old_obj
.data
, obj
->data
, cd
->data_size
);
1898 else { /* TODO: if the size doesn't change, use old space */
1899 uint32_t new_data_map_size
;
1901 ret
= get_row_index(t
, r
->num
, &row_index
);
1904 ret
= mark_mapped_object_invalid(t
, r
->num
, col_num
);
1907 unmap_column(t
, col_num
);
1908 ret
= append_map_file(t
, col_num
, obj
,
1909 &new_data_map_size
);
1912 ret
= map_column(t
, col_num
);
1915 update_cell_index(row_index
, col
, new_data_map_size
,
1919 if (cd
->storage_flags
& OSL_RBTREE
) {
1920 ret
= insert_rbtree(t
, col_num
, r
, obj
);
1928 * Retrieve an object of type \p OSL_DISK_STORAGE by row and column.
1930 * \param t Pointer to an open osl table.
1931 * \param r Pointer to the row containing the object.
1932 * \param col_num The column number.
1933 * \param obj Points to the result upon successful return.
1935 * For columns of type \p OSL_DISK_STORAGE, this function must be used to
1936 * retrieve one of its containing objects. Afterwards, osl_close_disk_object()
1937 * must be called in order to deallocate the resources.
1939 * \return Positive on success, negative on errors. Possible errors include:
1940 * \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE, errors returned by osl_get_object().
1942 * \sa osl_get_object(), osl_storage_type, osl_close_disk_object().
1944 int osl_open_disk_object(const struct osl_table
*t
, const struct osl_row
*r
,
1945 unsigned col_num
, struct osl_object
*obj
)
1947 const struct osl_column_description
*cd
;
1948 char *ds_name
, *filename
;
1952 return -E_BAD_TABLE
;
1953 cd
= get_column_description(t
->desc
, col_num
);
1954 if (cd
->storage_type
!= OSL_DISK_STORAGE
)
1955 return -E_BAD_STORAGE_TYPE
;
1957 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1960 filename
= disk_storage_path(t
, col_num
, ds_name
);
1962 PARA_DEBUG_LOG("filename: %s\n", filename
);
1963 ret
= mmap_full_file(filename
, O_RDONLY
, obj
);
1969 * Free resources that were allocated during osl_open_disk_object().
1971 * \param obj Pointer to the object previously returned by open_disk_object().
1973 * \return The return value of the underlying call to para_munmap().
1975 * \sa para_munmap().
1977 int osl_close_disk_object(struct osl_object
*obj
)
1979 return para_munmap(obj
->data
, obj
->size
);
1983 * Get the number of rows of the given table.
1985 * \param t Pointer to an open osl table.
1986 * \param num_rows Result is returned here.
1988 * The number of rows returned via \a num_rows excluding any invalid rows.
1990 * \return Positive on success, \p -E_BAD_TABLE if \a t is \p NULL.
1992 int osl_get_num_rows(const struct osl_table
*t
, unsigned *num_rows
)
1995 return -E_BAD_TABLE
;
1996 assert(t
->num_rows
>= t
->num_invalid_rows
);
1997 *num_rows
= t
->num_rows
- t
->num_invalid_rows
;
2002 * Get the rank of a row.
2004 * \param t An open osl table.
2005 * \param r The row to get the rank of.
2006 * \param col_num The number of an rbtree column.
2007 * \param rank Result pointer.
2009 * The rank is, by definition, the position of the row in the linear order
2010 * determined by an inorder tree walk of the rbtree associated with column
2011 * number \a col_num of \a table.
2013 * \return Positive on success, negative on errors.
2015 * \sa osl_get_nth_row().
2017 int osl_get_rank(const struct osl_table
*t
, struct osl_row
*r
,
2018 unsigned col_num
, unsigned *rank
)
2020 struct osl_object obj
;
2021 struct osl_column
*col
;
2022 struct rb_node
*node
;
2023 int ret
= check_rbtree_col(t
, col_num
, &col
);
2027 ret
= osl_get_object(t
, r
, col_num
, &obj
);
2030 ret
= search_rbtree(&obj
, t
, col_num
, &node
, NULL
);
2033 ret
= rb_rank(node
, rank
);
2040 * Get the row with n-th greatest value.
2042 * \param t Pointer to an open osl table.
2043 * \param col_num The column number.
2044 * \param n The rank of the desired row.
2045 * \param result Row is returned here.
2047 * Retrieve the n-th order statistic with respect to the compare function
2048 * of the rbtree column \a col_num. In other words, get that row with
2049 * \a n th greatest value in column \a col_num. It's an error if
2050 * \a col_num is not a rbtree column, or if \a n is larger than the
2051 * number of rows in the table.
2053 * \return Positive on success, negative on errors. Possible errors:
2054 * \p E_BAD_TABLE, \p E_BAD_STORAGE_FLAGS, \p E_RB_KEY_NOT_FOUND.
2056 * \sa osl_storage_flags, osl_compare_func, osl_get_row(),
2057 * osl_rbtree_last_row(), osl_rbtree_first_row(), osl_get_rank().
2059 int osl_get_nth_row(const struct osl_table
*t
, unsigned col_num
,
2060 unsigned n
, struct osl_row
**result
)
2062 struct osl_column
*col
;
2063 struct rb_node
*node
;
2064 int ret
= check_rbtree_col(t
, col_num
, &col
);
2068 node
= rb_nth(col
->rbtree
.rb_node
, n
);
2070 return -E_RB_KEY_NOT_FOUND
;
2071 *result
= get_row_pointer(node
, col
->rbtree_num
);
2076 * Get the row corresponding to the smallest rbtree node of a column.
2078 * \param t An open rbtree table.
2079 * \param col_num The number of the rbtree column.
2080 * \param result A pointer to the first row is returned here.
2082 * The rbtree node of the smallest object (with respect to the corresponding
2083 * compare function) is selected and the row containing this object is
2084 * returned. It is an error if \a col_num refers to a column without an
2085 * associated rbtree.
2087 * \return Positive on success, negative on errors.
2089 * \sa osl_get_nth_row(), osl_rbtree_last_row().
2091 int osl_rbtree_first_row(const struct osl_table
*t
, unsigned col_num
,
2092 struct osl_row
**result
)
2094 return osl_get_nth_row(t
, col_num
, 1, result
);
2098 * Get the row corresponding to the greatest rbtree node of a column.
2100 * \param t The same meaning as in \p osl_rbtree_first_row().
2101 * \param col_num The same meaning as in \p osl_rbtree_first_row().
2102 * \param result The same meaning as in \p osl_rbtree_first_row().
2104 * This function works just like osl_rbtree_first_row(), the only difference
2105 * is that the row containing the greatest rather than the smallest object is
2108 * \return Positive on success, negative on errors.
2110 * \sa osl_get_nth_row(), osl_rbtree_first_row().
2112 int osl_rbtree_last_row(const struct osl_table
*t
, unsigned col_num
,
2113 struct osl_row
**result
)
2116 int ret
= osl_get_num_rows(t
, &num_rows
);
2120 return osl_get_nth_row(t
, col_num
, num_rows
, result
);