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 && ret
!= -E_EXIST
)
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 if (errno
== ENOENT
)
1236 if (!S_ISDIR(statbuf
.st_mode
))
1239 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1243 PARA_DEBUG_LOG("num rows: %d\n", t
->num_rows
);
1244 ret
= init_rbtrees(t
);
1246 osl_close_table(t
, OSL_MARK_CLEAN
); /* ignore further errors */
1257 static int create_disk_storage_object_dir(const struct osl_table
*t
,
1258 unsigned col_num
, const char *ds_name
)
1263 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1265 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1266 ret
= para_mkdir(dirname
, 0777);
1268 if (ret
< 0 && ret
!= -E_EXIST
)
1273 static int write_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1274 const struct osl_object
*obj
, const char *ds_name
)
1279 ret
= create_disk_storage_object_dir(t
, col_num
, ds_name
);
1282 filename
= disk_storage_path(t
, col_num
, ds_name
);
1283 ret
= para_write_file(filename
, obj
->data
, obj
->size
);
1288 static int append_map_file(const struct osl_table
*t
, unsigned col_num
,
1289 const struct osl_object
*obj
, uint32_t *new_size
)
1291 char *filename
= column_filename(t
, col_num
);
1293 char header
= 0; /* zero means valid object */
1295 // PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj->size);
1296 ret
= append_file(filename
, &header
, 1, obj
->data
, obj
->size
,
1302 static int append_row_index(const struct osl_table
*t
, char *row_index
)
1307 if (!t
->num_mapped_columns
)
1309 filename
= index_filename(t
->desc
);
1310 ret
= append_file(filename
, NULL
, 0, row_index
,
1311 t
->row_index_size
, NULL
);
1317 * A wrapper for truncate(2)
1319 * \param path Name of the regular file to truncate
1320 * \param size Number of bytes to \b shave \b off
1322 * Truncate the regular file named by \a path by \a size bytes.
1324 * \return Positive on success, negative on errors. Possible errors include: \p
1325 * E_STAT, \p E_BAD_SIZE, \p E_TRUNC.
1329 int para_truncate(const char *path
, off_t size
)
1332 struct stat statbuf
;
1335 if (stat(path
, &statbuf
) < 0)
1338 if (statbuf
.st_size
< size
)
1341 if (truncate(path
, statbuf
.st_size
- size
) < 0)
1348 static int truncate_mapped_file(const struct osl_table
*t
, unsigned col_num
,
1351 char *filename
= column_filename(t
, col_num
);
1352 int ret
= para_truncate(filename
, size
);
1357 static int delete_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1358 const char *ds_name
)
1360 char *dirname
, *filename
= disk_storage_path(t
, col_num
, ds_name
);
1361 int ret
= unlink(filename
);
1363 PARA_DEBUG_LOG("deleted %s\n", filename
);
1366 if (errno
== ENOENT
)
1370 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1372 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1379 * Add a new row to an osl table and retrieve this row.
1381 * \param t Pointer to an open osl table.
1382 * \param objects Array of objects to be added.
1383 * \param row Result pointer.
1385 * The \a objects parameter must point to an array containing one object per
1386 * column. The order of the objects in the array is given by the table
1387 * description of \a table. Several sanity checks are performed during object
1388 * insertion and the function returns without modifying the table if any of
1389 * these tests fail. In fact, it is atomic in the sense that it either
1390 * succeeds or leaves the table unchanged (i.e. either all or none of the
1391 * objects are added to the table).
1393 * It is considered an error if an object is added to a column with associated
1394 * rbtree if this object is equal to an object already contained in that column
1395 * (i.e. the compare function for the column's rbtree returns zero).
1397 * Possible errors include: \p E_RB_KEY_EXISTS, \p E_BAD_DATA_SIZE.
1399 * \return Positive on success, negative on errors.
1401 * \sa struct osl_table_description, osl_compare_func, osl_add_row().
1403 int osl_add_and_get_row(struct osl_table
*t
, struct osl_object
*objects
,
1404 struct osl_row
**row
)
1407 char *ds_name
= NULL
;
1408 struct rb_node
**rb_parents
= NULL
, ***rb_links
= NULL
;
1409 char *new_row_index
= NULL
;
1410 struct osl_object
*volatile_objs
= NULL
;
1411 const struct osl_column_description
*cd
;
1414 return -E_BAD_TABLE
;
1415 rb_parents
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
*));
1416 rb_links
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
**));
1417 if (t
->num_mapped_columns
)
1418 new_row_index
= para_malloc(t
->row_index_size
);
1419 /* pass 1: sanity checks */
1420 // PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
1421 // objects[1].data);
1422 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1423 enum osl_storage_type st
= cd
->storage_type
;
1424 enum osl_storage_flags sf
= cd
->storage_flags
;
1426 // ret = -E_NULL_OBJECT;
1429 if (st
== OSL_DISK_STORAGE
)
1431 if (sf
& OSL_RBTREE
) {
1432 unsigned rbtree_num
= t
->columns
[i
].rbtree_num
;
1433 ret
= -E_RB_KEY_EXISTS
;
1434 // PARA_DEBUG_LOG("checking whether %p exists\n",
1435 // objects[i].data);
1436 if (search_rbtree(objects
+ i
, t
, i
,
1437 &rb_parents
[rbtree_num
],
1438 &rb_links
[rbtree_num
]) > 0)
1441 if (sf
& OSL_FIXED_SIZE
) {
1442 // PARA_DEBUG_LOG("fixed size. need: %zu, have: %d\n",
1443 // objects[i].size, cd->data_size);
1444 ret
= -E_BAD_DATA_SIZE
;
1445 if (objects
[i
].size
!= cd
->data_size
)
1449 if (t
->num_disk_storage_columns
)
1450 ds_name
= disk_storage_name_of_object(t
,
1451 &objects
[t
->disk_storage_name_column
]);
1452 ret
= unmap_table(t
, OSL_MARK_CLEAN
);
1455 // PARA_DEBUG_LOG("sanity tests passed%s\n", "");
1456 /* pass 2: create data files, append map data */
1457 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1458 enum osl_storage_type st
= cd
->storage_type
;
1459 if (st
== OSL_NO_STORAGE
)
1461 if (st
== OSL_MAPPED_STORAGE
) {
1463 struct osl_column
*col
= &t
->columns
[i
];
1464 // PARA_DEBUG_LOG("appending object of size %zu\n",
1465 // objects[i].size);
1466 ret
= append_map_file(t
, i
, objects
+ i
, &new_size
);
1469 update_cell_index(new_row_index
, col
, new_size
,
1474 ret
= write_disk_storage_file(t
, i
, objects
+ i
, ds_name
);
1478 ret
= append_row_index(t
, new_row_index
);
1481 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1482 if (ret
< 0) { /* truncate index and rollback changes */
1483 char *filename
= index_filename(t
->desc
);
1484 para_truncate(filename
, t
->row_index_size
);
1488 /* pass 3: add entry to rbtrees */
1489 if (t
->num_volatile_columns
) {
1490 volatile_objs
= para_calloc(t
->num_volatile_columns
1491 * sizeof(struct osl_object
));
1492 FOR_EACH_VOLATILE_COLUMN(i
, t
, cd
)
1493 volatile_objs
[t
->columns
[i
].volatile_num
] = objects
[i
];
1496 // PARA_DEBUG_LOG("adding new entry as row #%d\n", t->num_rows - 1);
1497 ret
= add_row_to_rbtrees(t
, t
->num_rows
- 1, volatile_objs
, row
);
1500 // PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
1503 rollback
: /* rollback all changes made, ignore further errors */
1504 for (i
--; i
>= 0; i
--) {
1505 cd
= get_column_description(t
->desc
, i
);
1506 enum osl_storage_type st
= cd
->storage_type
;
1507 if (st
== OSL_NO_STORAGE
)
1510 if (st
== OSL_MAPPED_STORAGE
)
1511 truncate_mapped_file(t
, i
, objects
[i
].size
);
1512 else /* disk storage */
1513 delete_disk_storage_file(t
, i
, ds_name
);
1515 /* ignore error and return previous error value */
1516 map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1518 free(new_row_index
);
1526 * Add a new row to an osl table.
1528 * \param t Same meaning as osl_add_and_get_row().
1529 * \param objects Same meaning as osl_add_and_get_row().
1531 * \return The return value of the underlying call to osl_add_and_get_row().
1533 * This is equivalent to osl_add_and_get_row(t, objects, NULL).
1535 int osl_add_row(struct osl_table
*t
, struct osl_object
*objects
)
1537 return osl_add_and_get_row(t
, objects
, NULL
);
1541 * Retrieve an object identified by row and column
1543 * \param t Pointer to an open osl table.
1544 * \param r Pointer to the row.
1545 * \param col_num The column number.
1546 * \param object The result pointer.
1548 * The column determined by \a col_num must be of type \p OSL_MAPPED_STORAGE
1549 * or \p OSL_NO_STORAGE, i.e. no disk storage objects may be retrieved by this
1552 * \return Positive if object was found, negative on errors. Possible errors
1553 * include: \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE.
1555 * \sa osl_storage_type, osl_open_disk_object().
1557 int osl_get_object(const struct osl_table
*t
, const struct osl_row
*r
,
1558 unsigned col_num
, struct osl_object
*object
)
1560 const struct osl_column_description
*cd
;
1563 return -E_BAD_TABLE
;
1564 cd
= get_column_description(t
->desc
, col_num
);
1565 /* col must not be disk storage */
1566 if (cd
->storage_type
== OSL_DISK_STORAGE
)
1567 return -E_BAD_STORAGE_TYPE
;
1568 if (cd
->storage_type
== OSL_MAPPED_STORAGE
)
1569 return get_mapped_object(t
, col_num
, r
->num
, object
);
1571 *object
= r
->volatile_objects
[t
->columns
[col_num
].volatile_num
];
1575 static int mark_mapped_object_invalid(const struct osl_table
*t
,
1576 uint32_t row_num
, unsigned col_num
)
1578 struct osl_object obj
;
1580 int ret
= get_mapped_object(t
, col_num
, row_num
, &obj
);
1591 * Delete a row from an osl table.
1593 * \param t Pointer to an open osl table.
1594 * \param row Pointer to the row to delete.
1596 * This removes all disk storage objects, removes all rbtree nodes, and frees
1597 * all volatile objects belonging to the given row. For mapped columns, the
1598 * data is merely marked invalid and may be pruned from time to time by
1601 * \return Positive on success, negative on errors. Possible errors include:
1602 * \p E_BAD_TABLE, errors returned by osl_get_object().
1604 int osl_del_row(struct osl_table
*t
, struct osl_row
*row
)
1606 struct osl_row
*r
= row
;
1608 const struct osl_column_description
*cd
;
1611 return -E_BAD_TABLE
;
1612 PARA_INFO_LOG("deleting row %p\n", row
);
1614 if (t
->num_disk_storage_columns
) {
1616 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1619 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
)
1620 delete_disk_storage_file(t
, i
, ds_name
);
1623 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1624 struct osl_column
*col
= t
->columns
+ i
;
1625 enum osl_storage_type st
= cd
->storage_type
;
1626 remove_rb_node(t
, i
, r
);
1627 if (st
== OSL_MAPPED_STORAGE
) {
1628 mark_mapped_object_invalid(t
, r
->num
, i
);
1631 if (st
== OSL_NO_STORAGE
)
1632 free(r
->volatile_objects
[col
->volatile_num
].data
);
1634 if (t
->num_mapped_columns
) {
1635 ret
= mark_row_invalid(t
, r
->num
);
1638 t
->num_invalid_rows
++;
1643 free(r
->volatile_objects
);
1648 /* test if column has an rbtree */
1649 static int check_rbtree_col(const struct osl_table
*t
, unsigned col_num
,
1650 struct osl_column
**col
)
1653 return -E_BAD_TABLE
;
1654 if (!(get_column_description(t
->desc
, col_num
)->storage_flags
& OSL_RBTREE
))
1655 return -E_BAD_STORAGE_FLAGS
;
1656 *col
= t
->columns
+ col_num
;
1661 * Get the row that contains the given object.
1663 * \param t Pointer to an open osl table.
1664 * \param col_num The number of the column to be searched.
1665 * \param obj The object to be looked up.
1666 * \param result Points to the row containing \a obj.
1668 * Lookup \a obj in \a t and return the row containing \a obj. The column
1669 * specified by \a col_num must have an associated rbtree.
1671 * \return Positive on success, negative on errors. If an error occurred, \a
1672 * result is set to \p NULL. Possible errors include: \p E_BAD_TABLE, \p
1673 * E_BAD_STORAGE_FLAGS, errors returned by get_mapped_object(), \p
1674 * E_RB_KEY_NOT_FOUND.
1676 * \sa osl_storage_flags
1678 int osl_get_row(const struct osl_table
*t
, unsigned col_num
,
1679 const struct osl_object
*obj
, struct osl_row
**result
)
1682 struct rb_node
*node
;
1683 struct osl_row
*row
;
1684 struct osl_column
*col
;
1687 ret
= check_rbtree_col(t
, col_num
, &col
);
1690 ret
= search_rbtree(obj
, t
, col_num
, &node
, NULL
);
1693 row
= get_row_pointer(node
, t
->columns
[col_num
].rbtree_num
);
1698 static int rbtree_loop(struct osl_column
*col
, void *private_data
,
1699 osl_rbtree_loop_func
*func
)
1701 struct rb_node
*n
, *tmp
;
1703 /* this for-loop is safe against removal of an entry */
1704 for (n
= rb_first(&col
->rbtree
), tmp
= n
? rb_next(n
) : NULL
;
1706 n
= tmp
, tmp
= tmp
? rb_next(tmp
) : NULL
) {
1707 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1708 int ret
= func(r
, private_data
);
1715 static int rbtree_loop_reverse(struct osl_column
*col
, void *private_data
,
1716 osl_rbtree_loop_func
*func
)
1718 struct rb_node
*n
, *tmp
;
1720 /* safe against removal of an entry */
1721 for (n
= rb_last(&col
->rbtree
), tmp
= n
? rb_prev(n
) : NULL
;
1723 n
= tmp
, tmp
= tmp
? rb_prev(tmp
) : NULL
) {
1724 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1725 int ret
= func(r
, private_data
);
1733 * Loop over all nodes in an rbtree.
1735 * \param t Pointer to an open osl table.
1736 * \param col_num The column to use for iterating over the elements.
1737 * \param private_data Pointer that gets passed to \a func.
1738 * \param func The function to be called for each node in the rbtree.
1740 * This function does an in-order walk of the rbtree associated with \a
1741 * col_num. It is an error if the \p OSL_RBTREE flag is not set for this
1742 * column. For each node in the rbtree, the given function \a func is called
1743 * with two pointers as arguments: The first osl_row* argument points to the
1744 * row that contains the object corresponding to the rbtree node currently
1745 * traversed, and the \a private_data pointer is passed verbatim to \a func as the
1746 * second argument. The loop terminates either if \a func returns a negative
1747 * value, or if all nodes of the tree have been visited.
1750 * \return Positive on success, negative on errors. If the termination of the
1751 * loop was caused by \a func returning a negative value, this value is
1754 * \sa osl_storage_flags, osl_rbtree_loop_reverse(), osl_compare_func.
1756 int osl_rbtree_loop(const struct osl_table
*t
, unsigned col_num
,
1757 void *private_data
, osl_rbtree_loop_func
*func
)
1759 struct osl_column
*col
;
1761 int ret
= check_rbtree_col(t
, col_num
, &col
);
1764 return rbtree_loop(col
, private_data
, func
);
1768 * Loop over all nodes in an rbtree in reverse order.
1770 * \param t Identical meaning as in \p osl_rbtree_loop().
1771 * \param col_num Identical meaning as in \p osl_rbtree_loop().
1772 * \param private_data Identical meaning as in \p osl_rbtree_loop().
1773 * \param func Identical meaning as in \p osl_rbtree_loop().
1775 * This function is identical to \p osl_rbtree_loop(), the only difference
1776 * is that the tree is walked in reverse order.
1778 * \return The same return value as \p osl_rbtree_loop().
1780 * \sa osl_rbtree_loop().
1782 int osl_rbtree_loop_reverse(const struct osl_table
*t
, unsigned col_num
,
1783 void *private_data
, osl_rbtree_loop_func
*func
)
1785 struct osl_column
*col
;
1787 int ret
= check_rbtree_col(t
, col_num
, &col
);
1790 return rbtree_loop_reverse(col
, private_data
, func
);
1793 /* TODO: Rollback changes on errors */
1794 static int rename_disk_storage_objects(struct osl_table
*t
,
1795 struct osl_object
*old_obj
, struct osl_object
*new_obj
)
1798 const struct osl_column_description
*cd
;
1799 char *old_ds_name
, *new_ds_name
;
1801 if (!t
->num_disk_storage_columns
)
1802 return 1; /* nothing to do */
1803 if (old_obj
->size
== new_obj
->size
&& !memcmp(new_obj
->data
,
1804 old_obj
->data
, new_obj
->size
))
1805 return 1; /* object did not change */
1806 old_ds_name
= disk_storage_name_of_object(t
, old_obj
);
1807 new_ds_name
= disk_storage_name_of_object(t
, new_obj
);
1808 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
) {
1809 char *old_filename
, *new_filename
;
1810 ret
= create_disk_storage_object_dir(t
, i
, new_ds_name
);
1813 old_filename
= disk_storage_path(t
, i
, old_ds_name
);
1814 new_filename
= disk_storage_path(t
, i
, new_ds_name
);
1815 ret
= para_rename(old_filename
, new_filename
);
1830 * Change an object in an osl table.
1832 * \param t Pointer to an open osl table.
1833 * \param r Pointer to the row containing the object to be updated.
1834 * \param col_num Number of the column containing the object to be updated.
1835 * \param obj Pointer to the replacement object.
1837 * This function gets rid of all references to the old object. This includes
1838 * removal of the rbtree node in case there is an rbtree associated with \a
1839 * col_num. It then inserts \a obj into the table and the rbtree if necessary.
1841 * If the \p OSL_RBTREE flag is set for \a col_num, you \b MUST call this
1842 * function in order to change the contents of an object, even for volatile or
1843 * mapped columns of constant size (which may be updated directly if \p
1844 * OSL_RBTREE is not set). Otherwise the rbtree might become corrupted.
1846 * \return Positive on success, negative on errors. Possible errors include: \p
1847 * E_BAD_TABLE, \p E_RB_KEY_EXISTS, \p E_BAD_SIZE, \p E_NOENT, \p E_UNLINK,
1848 * errors returned by para_write_file(), \p E_MKDIR.
1850 int osl_update_object(struct osl_table
*t
, const struct osl_row
*r
,
1851 unsigned col_num
, struct osl_object
*obj
)
1853 struct osl_column
*col
;
1854 const struct osl_column_description
*cd
;
1858 return -E_BAD_TABLE
;
1859 col
= &t
->columns
[col_num
];
1860 cd
= get_column_description(t
->desc
, col_num
);
1861 PARA_DEBUG_LOG("updating column %u of %s\n", col_num
, t
->desc
->name
);
1862 if (cd
->storage_flags
& OSL_RBTREE
) {
1863 if (search_rbtree(obj
, t
, col_num
, NULL
, NULL
) > 0)
1864 return -E_RB_KEY_EXISTS
;
1866 if (cd
->storage_flags
& OSL_FIXED_SIZE
) {
1867 if (obj
->size
!= cd
->data_size
)
1868 return -E_BAD_DATA_SIZE
;
1870 remove_rb_node(t
, col_num
, r
);
1871 if (cd
->storage_type
== OSL_NO_STORAGE
) { /* TODO: If fixed size, reuse object? */
1872 free(r
->volatile_objects
[col
->volatile_num
].data
);
1873 r
->volatile_objects
[col
->volatile_num
] = *obj
;
1874 } else if (cd
->storage_type
== OSL_DISK_STORAGE
) {
1876 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1879 ret
= delete_disk_storage_file(t
, col_num
, ds_name
);
1880 if (ret
< 0 && ret
!= -E_NOENT
) {
1884 ret
= write_disk_storage_file(t
, col_num
, obj
, ds_name
);
1888 } else { /* mapped storage */
1889 struct osl_object old_obj
;
1890 ret
= get_mapped_object(t
, col_num
, r
->num
, &old_obj
);
1894 * If the updated column is the disk storage name column, the
1895 * disk storage name changes, so we have to rename all disk
1896 * storage objects accordingly.
1898 if (col_num
== t
->disk_storage_name_column
) {
1899 ret
= rename_disk_storage_objects(t
, &old_obj
, obj
);
1903 if (cd
->storage_flags
& OSL_FIXED_SIZE
)
1904 memcpy(old_obj
.data
, obj
->data
, cd
->data_size
);
1905 else { /* TODO: if the size doesn't change, use old space */
1906 uint32_t new_data_map_size
;
1908 ret
= get_row_index(t
, r
->num
, &row_index
);
1911 ret
= mark_mapped_object_invalid(t
, r
->num
, col_num
);
1914 unmap_column(t
, col_num
);
1915 ret
= append_map_file(t
, col_num
, obj
,
1916 &new_data_map_size
);
1919 ret
= map_column(t
, col_num
);
1922 update_cell_index(row_index
, col
, new_data_map_size
,
1926 if (cd
->storage_flags
& OSL_RBTREE
) {
1927 ret
= insert_rbtree(t
, col_num
, r
, obj
);
1935 * Retrieve an object of type \p OSL_DISK_STORAGE by row and column.
1937 * \param t Pointer to an open osl table.
1938 * \param r Pointer to the row containing the object.
1939 * \param col_num The column number.
1940 * \param obj Points to the result upon successful return.
1942 * For columns of type \p OSL_DISK_STORAGE, this function must be used to
1943 * retrieve one of its containing objects. Afterwards, osl_close_disk_object()
1944 * must be called in order to deallocate the resources.
1946 * \return Positive on success, negative on errors. Possible errors include:
1947 * \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE, errors returned by osl_get_object().
1949 * \sa osl_get_object(), osl_storage_type, osl_close_disk_object().
1951 int osl_open_disk_object(const struct osl_table
*t
, const struct osl_row
*r
,
1952 unsigned col_num
, struct osl_object
*obj
)
1954 const struct osl_column_description
*cd
;
1955 char *ds_name
, *filename
;
1959 return -E_BAD_TABLE
;
1960 cd
= get_column_description(t
->desc
, col_num
);
1961 if (cd
->storage_type
!= OSL_DISK_STORAGE
)
1962 return -E_BAD_STORAGE_TYPE
;
1964 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1967 filename
= disk_storage_path(t
, col_num
, ds_name
);
1969 PARA_DEBUG_LOG("filename: %s\n", filename
);
1970 ret
= mmap_full_file(filename
, O_RDONLY
, obj
);
1976 * Free resources that were allocated during osl_open_disk_object().
1978 * \param obj Pointer to the object previously returned by open_disk_object().
1980 * \return The return value of the underlying call to para_munmap().
1982 * \sa para_munmap().
1984 int osl_close_disk_object(struct osl_object
*obj
)
1986 return para_munmap(obj
->data
, obj
->size
);
1990 * Get the number of rows of the given table.
1992 * \param t Pointer to an open osl table.
1993 * \param num_rows Result is returned here.
1995 * The number of rows returned via \a num_rows excluding any invalid rows.
1997 * \return Positive on success, \p -E_BAD_TABLE if \a t is \p NULL.
1999 int osl_get_num_rows(const struct osl_table
*t
, unsigned *num_rows
)
2002 return -E_BAD_TABLE
;
2003 assert(t
->num_rows
>= t
->num_invalid_rows
);
2004 *num_rows
= t
->num_rows
- t
->num_invalid_rows
;
2009 * Get the rank of a row.
2011 * \param t An open osl table.
2012 * \param r The row to get the rank of.
2013 * \param col_num The number of an rbtree column.
2014 * \param rank Result pointer.
2016 * The rank is, by definition, the position of the row in the linear order
2017 * determined by an inorder tree walk of the rbtree associated with column
2018 * number \a col_num of \a table.
2020 * \return Positive on success, negative on errors.
2022 * \sa osl_get_nth_row().
2024 int osl_get_rank(const struct osl_table
*t
, struct osl_row
*r
,
2025 unsigned col_num
, unsigned *rank
)
2027 struct osl_object obj
;
2028 struct osl_column
*col
;
2029 struct rb_node
*node
;
2030 int ret
= check_rbtree_col(t
, col_num
, &col
);
2034 ret
= osl_get_object(t
, r
, col_num
, &obj
);
2037 ret
= search_rbtree(&obj
, t
, col_num
, &node
, NULL
);
2040 ret
= rb_rank(node
, rank
);
2047 * Get the row with n-th greatest value.
2049 * \param t Pointer to an open osl table.
2050 * \param col_num The column number.
2051 * \param n The rank of the desired row.
2052 * \param result Row is returned here.
2054 * Retrieve the n-th order statistic with respect to the compare function
2055 * of the rbtree column \a col_num. In other words, get that row with
2056 * \a n th greatest value in column \a col_num. It's an error if
2057 * \a col_num is not a rbtree column, or if \a n is larger than the
2058 * number of rows in the table.
2060 * \return Positive on success, negative on errors. Possible errors:
2061 * \p E_BAD_TABLE, \p E_BAD_STORAGE_FLAGS, \p E_RB_KEY_NOT_FOUND.
2063 * \sa osl_storage_flags, osl_compare_func, osl_get_row(),
2064 * osl_rbtree_last_row(), osl_rbtree_first_row(), osl_get_rank().
2066 int osl_get_nth_row(const struct osl_table
*t
, unsigned col_num
,
2067 unsigned n
, struct osl_row
**result
)
2069 struct osl_column
*col
;
2070 struct rb_node
*node
;
2071 int ret
= check_rbtree_col(t
, col_num
, &col
);
2075 node
= rb_nth(col
->rbtree
.rb_node
, n
);
2077 return -E_RB_KEY_NOT_FOUND
;
2078 *result
= get_row_pointer(node
, col
->rbtree_num
);
2083 * Get the row corresponding to the smallest rbtree node of a column.
2085 * \param t An open rbtree table.
2086 * \param col_num The number of the rbtree column.
2087 * \param result A pointer to the first row is returned here.
2089 * The rbtree node of the smallest object (with respect to the corresponding
2090 * compare function) is selected and the row containing this object is
2091 * returned. It is an error if \a col_num refers to a column without an
2092 * associated rbtree.
2094 * \return Positive on success, negative on errors.
2096 * \sa osl_get_nth_row(), osl_rbtree_last_row().
2098 int osl_rbtree_first_row(const struct osl_table
*t
, unsigned col_num
,
2099 struct osl_row
**result
)
2101 return osl_get_nth_row(t
, col_num
, 1, result
);
2105 * Get the row corresponding to the greatest rbtree node of a column.
2107 * \param t The same meaning as in \p osl_rbtree_first_row().
2108 * \param col_num The same meaning as in \p osl_rbtree_first_row().
2109 * \param result The same meaning as in \p osl_rbtree_first_row().
2111 * This function works just like osl_rbtree_first_row(), the only difference
2112 * is that the row containing the greatest rather than the smallest object is
2115 * \return Positive on success, negative on errors.
2117 * \sa osl_get_nth_row(), osl_rbtree_first_row().
2119 int osl_rbtree_last_row(const struct osl_table
*t
, unsigned col_num
,
2120 struct osl_row
**result
)
2123 int ret
= osl_get_num_rows(t
, &num_rows
);
2127 return osl_get_nth_row(t
, col_num
, num_rows
, result
);