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. */
12 #include <dirent.h> /* readdir() */
15 //#define FMT_OFF_T "%li"
19 * A wrapper for lseek(2).
21 * \param fd The filedescriptor whose offset is to be to repositioned.
22 * \param offset A value-result parameter.
23 * \param whence Usual repositioning directive.
25 * Reposition the offset of the file descriptor \a fd to the argument \a offset
26 * according to the directive \a whence. Upon successful return, \a offset
27 * contains the resulting offset location as measured in bytes from the
28 * beginning of the file.
30 * \return Positive on success. Otherwise, the function returns \p -E_LSEEK.
34 int para_lseek(int fd
, off_t
*offset
, int whence
)
36 *offset
= lseek(fd
, *offset
, whence
);
44 * Waraper for the write system call.
46 * \param fd The file descriptor to write to.
47 * \param buf The buffer to write.
48 * \param size The length of \a buf in bytes.
50 * This function writes out the given bufffer and retries if an interrupt
51 * occured during the write.
53 * \return On success, the number of bytes written is returned, otherwise, the
54 * function returns \p -E_WRITE.
58 ssize_t
para_write(int fd
, const void *buf
, size_t size
)
63 ret
= write(fd
, buf
, size
);
64 if ((ret
< 0) && (errno
== EAGAIN
|| errno
== EINTR
))
66 return ret
>= 0? ret
: -E_WRITE
;
71 * Write the whole buffer to a file descriptor.
73 * \param fd The file descriptor to write to.
74 * \param buf The buffer to write.
75 * \param size The length of \a buf in bytes.
77 * This function writes the given buffer and continues on short writes and
78 * when interrupted by a signal.
80 * \return Positive on success, negative on errors. Possible errors: any
81 * errors returned by para_write().
85 ssize_t
para_write_all(int fd
, const void *buf
, size_t size
)
87 PARA_DEBUG_LOG("writing %zu bytes\n", size
);
90 ssize_t ret
= para_write(fd
, b
, size
);
91 PARA_DEBUG_LOG("ret: %zd\n", ret
);
100 * Wrapper for the open(2) system call.
102 * \param path The filename.
103 * \param flags The usual open(2) flags.
104 * \param mode Specifies the permissions to use.
106 * The mode parameter must be specified when O_CREAT is in the flags, and is ignored
109 * \return Positive on success, negative on errors. Possible errors: \p
110 * E_EXIST, \p E_ISDIR, \p E_NOENT, \p E_OSL_PERM.
114 int para_open(const char *path
, int flags
, mode_t mode
)
116 PARA_DEBUG_LOG("opening %s\n", path
);
117 int ret
= open(path
, flags
, mode
);
135 PARA_ERROR_LOG("failed to open %s: %s\n", path
, strerror(errno
));
140 * Open a file, write the given buffer and close the file.
142 * \param filename Full path to the file to open.
143 * \param buf The buffer to write to the file.
144 * \param size The size of \a buf.
146 * \return Positive on success, negative on errors. Possible errors include:
147 * any errors from para_open() or para_write().
149 * \sa para_open(), para_write().
151 int para_write_file(const char *filename
, const void *buf
, size_t size
)
155 ret
= para_open(filename
, O_WRONLY
| O_CREAT
| O_EXCL
, 0644);
159 ret
= para_write_all(fd
, buf
, size
);
168 static int append_file(const char *filename
, char *header
, size_t header_size
,
169 char *data
, size_t data_size
, uint32_t *new_pos
)
173 PARA_DEBUG_LOG("appending %zu + %zu bytes\n", header_size
, data_size
);
174 ret
= para_open(filename
, O_WRONLY
| O_CREAT
| O_APPEND
, 0644);
178 if (header
&& header_size
) {
179 ret
= para_write_all(fd
, header
, header_size
);
183 ret
= para_write_all(fd
, data
, data_size
);
188 ret
= para_lseek(fd
, &offset
, SEEK_END
);
191 // PARA_DEBUG_LOG("new file size: " FMT_OFF_T "\n", offset);
201 * Map a file into memory.
203 * \param path Name of the regular file to map.
204 * \param open_mode Either \p O_RDONLY or \p O_RDWR.
205 * \param obj On success, the mapping is returned here.
207 * \return Positive on success, negative on errors. Possible errors include: \p
208 * E_FSTAT, any errors returned by para_open(), \p E_EMPTY, \p E_MMAP.
210 * \sa para_open(), mmap(2).
212 int mmap_full_file(const char *path
, int open_mode
, struct osl_object
*obj
)
214 int fd
, ret
, mmap_prot
, mmap_flags
;
215 struct stat file_status
;
217 if (open_mode
== O_RDONLY
) {
218 mmap_prot
= PROT_READ
;
219 mmap_flags
= MAP_PRIVATE
;
221 mmap_prot
= PROT_READ
| PROT_WRITE
;
222 mmap_flags
= MAP_SHARED
;
224 ret
= para_open(path
, open_mode
, 0);
229 if (fstat(fd
, &file_status
) < 0)
231 obj
->size
= file_status
.st_size
;
233 PARA_DEBUG_LOG("%s: size %zu\n", path
, obj
->size
);
236 obj
->data
= mmap(NULL
, obj
->size
, mmap_prot
, mmap_flags
, fd
, 0);
237 if (obj
->data
== MAP_FAILED
) {
249 * Traverse the given directory recursively.
251 * \param dirname The directory to traverse.
252 * \param func The function to call for each entry.
253 * \param private_data Pointer to an arbitrary data structure.
255 * For each regular file in \a dirname, the supplied function \a func is
256 * called. The full path of the regular file and the \a private_data pointer
257 * are passed to \a func.
259 * \return On success, 1 is returned. Otherwise, this function returns a
260 * negative value which indicates the kind of the error.
262 int for_each_file_in_dir(const char *dirname
,
263 int (*func
)(const char *, const void *), const void *private_data
)
266 struct dirent
*entry
;
268 * Opening the current directory (".") and calling fchdir() to return
269 * is usually faster and more reliable than saving cwd in some buffer
270 * and calling chdir() afterwards (see man 3 getcwd).
272 int cwd_fd
= open(".", O_RDONLY
);
276 // PARA_DEBUG_LOG("dirname: %s\n", dirname);
278 return -E_OSL_GETCWD
;
280 if (chdir(dirname
) < 0)
282 ret
= -E_OSL_OPENDIR
;
286 /* scan cwd recursively */
287 while ((entry
= readdir(dir
))) {
291 if (!strcmp(entry
->d_name
, "."))
293 if (!strcmp(entry
->d_name
, ".."))
296 if (lstat(entry
->d_name
, &s
) == -1)
299 if (!S_ISREG(m
) && !S_ISDIR(m
))
301 tmp
= make_message("%s/%s", dirname
, entry
->d_name
);
303 ret
= func(tmp
, private_data
);
310 ret
= for_each_file_in_dir(tmp
, func
, private_data
);
319 if (fchdir(cwd_fd
) < 0 && ret
>= 0)
325 int para_mkdir(const char *path
, mode_t mode
)
327 if (!mkdir(path
, mode
))
333 if (errno
== ENOTDIR
)
340 static int verify_basename(const char *name
)
343 return -E_BAD_BASENAME
;
345 return -E_BAD_BASENAME
;
346 if (strchr(name
, '/'))
347 return -E_BAD_BASENAME
;
348 if (!strcmp(name
, ".."))
349 return -E_BAD_BASENAME
;
350 if (!strcmp(name
, "."))
351 return -E_BAD_BASENAME
;
356 * Compare two osl objects pointing to unsigned integers of 32 bit size.
358 * \param obj1 Pointer to the first integer.
359 * \param obj2 Pointer to the second integer.
361 * \return The values required for an osl compare function.
363 * \sa osl_compare_func, osl_hash_compare().
365 int uint32_compare(const struct osl_object
*obj1
, const struct osl_object
*obj2
)
367 uint32_t d1
= read_u32((const char *)obj1
->data
);
368 uint32_t d2
= read_u32((const char *)obj2
->data
);
378 * Compare two osl objects pointing to hash values.
380 * \param obj1 Pointer to the first hash object.
381 * \param obj2 Pointer to the second hash object.
383 * \return The values required for an osl compare function.
385 * \sa osl_compare_func, uint32_compare().
387 int osl_hash_compare(const struct osl_object
*obj1
, const struct osl_object
*obj2
)
389 return hash_compare((HASH_TYPE
*)obj1
->data
, (HASH_TYPE
*)obj2
->data
);
392 static char *disk_storage_dirname(const struct osl_table
*t
, unsigned col_num
,
395 char *dirname
, *column_name
= column_filename(t
, col_num
);
397 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
399 dirname
= make_message("%s/%.2s", column_name
, ds_name
);
404 static char *disk_storage_name_of_object(const struct osl_table
*t
,
405 const struct osl_object
*obj
)
407 HASH_TYPE hash
[HASH_SIZE
];
408 hash_object(obj
, hash
);
409 return disk_storage_name_of_hash(t
, hash
);
412 static int disk_storage_name_of_row(const struct osl_table
*t
,
413 const struct osl_row
*row
, char **name
)
415 struct osl_object obj
;
416 int ret
= osl_get_object(t
, row
, t
->disk_storage_name_column
, &obj
);
420 *name
= disk_storage_name_of_object(t
, &obj
);
424 static void column_name_hash(const char *col_name
, HASH_TYPE
*hash
)
426 return hash_function(col_name
, strlen(col_name
), hash
);
429 static int init_column_descriptions(struct osl_table
*t
)
432 const struct osl_column_description
*cd
;
434 ret
= -E_BAD_TABLE_DESC
;
435 ret
= verify_basename(t
->desc
->name
);
441 /* the size of the index header without column descriptions */
442 t
->index_header_size
= IDX_COLUMN_DESCRIPTIONS
;
443 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
444 struct osl_column
*col
= t
->columns
+ i
;
445 if (cd
->storage_flags
& OSL_RBTREE
) {
446 if (!cd
->compare_function
)
447 return -E_NO_COMPARE_FUNC
;
449 if (cd
->storage_type
== OSL_NO_STORAGE
)
451 ret
= -E_NO_COLUMN_NAME
;
452 if (!cd
->name
|| !cd
->name
[0])
454 ret
= verify_basename(cd
->name
);
457 t
->index_header_size
+= index_column_description_size(cd
->name
);
458 column_name_hash(cd
->name
, col
->name_hash
);
459 ret
= -E_DUPLICATE_COL_NAME
;
460 for (j
= i
+ 1; j
< t
->desc
->num_columns
; j
++) {
461 const char *name2
= get_column_description(t
->desc
,
463 if (cd
->name
&& name2
&& !strcmp(cd
->name
, name2
))
473 * Initialize a struct table from given table description.
475 * \param desc The description of the osl table.
476 * \param table_ptr Result is returned here.
478 * This function performs several sanity checks on \p desc and returns if any
479 * of these tests fail. On success, a struct \p osl_table is allocated and
480 * initialized with data derived from \p desc.
482 * \return Positive on success, negative on errors. Possible errors include: \p
483 * E_BAD_TABLE_DESC, \p E_NO_COLUMN_DESC, \p E_NO_COLUMNS, \p
484 * E_BAD_STORAGE_TYPE, \p E_BAD_STORAGE_FLAGS, \p E_BAD_STORAGE_SIZE, \p
485 * E_NO_UNIQUE_RBTREE_COLUMN, \p E_NO_RBTREE_COL.
487 * \sa struct osl_table.
489 int init_table_structure(const struct osl_table_description
*desc
,
490 struct osl_table
**table_ptr
)
492 const struct osl_column_description
*cd
;
493 struct osl_table
*t
= para_calloc(sizeof(*t
));
494 int i
, ret
= -E_BAD_TABLE_DESC
, have_disk_storage_name_column
= 0;
498 PARA_INFO_LOG("creating table structure for '%s' from table "
499 "description\n", desc
->name
);
500 ret
= -E_NO_COLUMN_DESC
;
501 if (!desc
->column_descriptions
)
504 if (!desc
->num_columns
)
506 t
->columns
= para_calloc(desc
->num_columns
* sizeof(struct osl_column
));
508 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
509 enum osl_storage_type st
= cd
->storage_type
;
510 enum osl_storage_flags sf
= cd
->storage_flags
;
511 struct osl_column
*col
= &t
->columns
[i
];
513 ret
= -E_BAD_STORAGE_TYPE
;
514 if (st
!= OSL_MAPPED_STORAGE
&& st
!= OSL_DISK_STORAGE
515 && st
!= OSL_NO_STORAGE
)
517 ret
= -E_BAD_STORAGE_FLAGS
;
518 if (st
== OSL_DISK_STORAGE
&& sf
& OSL_RBTREE
)
520 ret
= -E_BAD_STORAGE_SIZE
;
521 if (sf
& OSL_FIXED_SIZE
&& !cd
->data_size
)
524 case OSL_DISK_STORAGE
:
525 t
->num_disk_storage_columns
++;
527 case OSL_MAPPED_STORAGE
:
528 t
->num_mapped_columns
++;
529 col
->index_offset
= t
->index_entry_size
;
530 t
->index_entry_size
+= 8;
533 col
->volatile_num
= t
->num_volatile_columns
;
534 t
->num_volatile_columns
++;
537 if (sf
& OSL_RBTREE
) {
538 col
->rbtree_num
= t
->num_rbtrees
;
540 if ((sf
& OSL_UNIQUE
) && (st
== OSL_MAPPED_STORAGE
)) {
541 if (!have_disk_storage_name_column
)
542 t
->disk_storage_name_column
= i
;
543 have_disk_storage_name_column
= 1;
547 ret
= -E_NO_UNIQUE_RBTREE_COLUMN
;
548 if (t
->num_disk_storage_columns
&& !have_disk_storage_name_column
)
550 ret
= -E_NO_RBTREE_COL
;
554 PARA_INFO_LOG("OK. Index entry size: %u\n", t
->index_entry_size
);
555 ret
= init_column_descriptions(t
);
567 * Read the table description from index header.
569 * \param map The memory mapping of the index file.
570 * \param desc The values found in the index header are returned here.
572 * Read the index header, check for the paraslash magic string and the table version number.
573 * Read all information stored in the index header into \a desc.
575 * \return Positive on success, negative on errors.
577 * \sa struct osl_table_description, osl_create_table.
579 int read_table_desc(struct osl_object
*map
, struct osl_table_description
*desc
)
581 char *buf
= map
->data
;
583 uint16_t header_size
;
586 struct osl_column_description
*cd
;
588 if (map
->size
< MIN_INDEX_HEADER_SIZE(1))
589 return -E_SHORT_TABLE
;
590 if (strncmp(buf
+ IDX_PARA_MAGIC
, PARA_MAGIC
, strlen(PARA_MAGIC
)))
592 version
= read_u8(buf
+ IDX_VERSION
);
593 if (version
< MIN_TABLE_VERSION
|| version
> MAX_TABLE_VERSION
)
594 return -E_VERSION_MISMATCH
;
595 desc
->num_columns
= read_u8(buf
+ IDX_TABLE_FLAGS
);
596 desc
->flags
= read_u8(buf
+ IDX_TABLE_FLAGS
);
597 desc
->num_columns
= read_u16(buf
+ IDX_NUM_COLUMNS
);
598 PARA_DEBUG_LOG("%u columns\n", desc
->num_columns
);
599 if (!desc
->num_columns
)
600 return -E_NO_COLUMNS
;
601 header_size
= read_u16(buf
+ IDX_HEADER_SIZE
);
602 if (map
->size
< header_size
)
604 desc
->column_descriptions
= para_calloc(desc
->num_columns
605 * sizeof(struct osl_column_description
));
606 offset
= IDX_COLUMN_DESCRIPTIONS
;
607 FOR_EACH_COLUMN(i
, desc
, cd
) {
610 ret
= -E_SHORT_TABLE
;
611 if (map
->size
< offset
+ MIN_IDX_COLUMN_DESCRIPTION_SIZE
) {
612 PARA_ERROR_LOG("map size = %zu < %u = offset + min desc size\n",
613 map
->size
, offset
+ MIN_IDX_COLUMN_DESCRIPTION_SIZE
);
616 cd
->storage_type
= read_u16(buf
+ offset
+ IDX_CD_STORAGE_TYPE
);
617 cd
->storage_flags
= read_u16(buf
+ offset
+
618 IDX_CD_STORAGE_FLAGS
);
619 cd
->data_size
= read_u32(buf
+ offset
+ IDX_CD_DATA_SIZE
);
620 null_byte
= memchr(buf
+ offset
+ IDX_CD_NAME
, '\0',
621 map
->size
- offset
- IDX_CD_NAME
);
622 ret
= -E_INDEX_CORRUPTION
;
625 cd
->name
= para_strdup(buf
+ offset
+ IDX_CD_NAME
);
626 offset
+= index_column_description_size(cd
->name
);
628 if (offset
!= header_size
) {
629 ret
= -E_INDEX_CORRUPTION
;
630 PARA_ERROR_LOG("real header size = %u != %u = stored header size\n",
631 offset
, header_size
);
636 FOR_EACH_COLUMN(i
, desc
, cd
)
642 * check whether the table description given by \p t->desc matches the on-disk
643 * table structure stored in the index of \a t.
645 static int compare_table_descriptions(struct osl_table
*t
)
648 struct osl_table_description desc
;
649 const struct osl_column_description
*cd1
, *cd2
;
651 /* read the on-disk structure into desc */
652 ret
= read_table_desc(&t
->index_map
, &desc
);
655 ret
= -E_BAD_TABLE_FLAGS
;
656 if (desc
.flags
!= t
->desc
->flags
)
658 ret
= E_BAD_COLUMN_NUM
;
659 if (desc
.num_columns
!= t
->desc
->num_columns
)
661 FOR_EACH_COLUMN(i
, t
->desc
, cd1
) {
662 cd2
= get_column_description(&desc
, i
);
663 ret
= -E_BAD_STORAGE_TYPE
;
664 if (cd1
->storage_type
!= cd2
->storage_type
)
666 ret
= -E_BAD_STORAGE_FLAGS
;
667 if (cd1
->storage_flags
!= cd2
->storage_flags
) {
668 PARA_ERROR_LOG("sf1 = %u != %u = sf2\n",
669 cd1
->storage_flags
, cd2
->storage_flags
);
672 ret
= -E_BAD_DATA_SIZE
;
673 if (cd1
->storage_flags
& OSL_FIXED_SIZE
)
674 if (cd1
->data_size
!= cd2
->data_size
)
676 ret
= -E_BAD_COLUMN_NAME
;
677 if (strcmp(cd1
->name
, cd2
->name
))
680 PARA_INFO_LOG("table description of '%s' matches on-disk data, good\n",
684 FOR_EACH_COLUMN(i
, &desc
, cd1
)
686 free(desc
.column_descriptions
);
690 static int create_table_index(struct osl_table
*t
)
692 char *buf
, *filename
;
694 size_t size
= t
->index_header_size
;
695 const struct osl_column_description
*cd
;
698 PARA_INFO_LOG("creating %zu byte index for table %s\n", size
,
700 buf
= para_calloc(size
);
701 sprintf(buf
+ IDX_PARA_MAGIC
, "%s", PARA_MAGIC
);
702 write_u8(buf
+ IDX_TABLE_FLAGS
, t
->desc
->flags
);
703 write_u8(buf
+ IDX_DIRTY_FLAG
, 0);
704 write_u8(buf
+ IDX_VERSION
, CURRENT_TABLE_VERSION
);
705 write_u16(buf
+ IDX_NUM_COLUMNS
, t
->desc
->num_columns
);
706 write_u16(buf
+ IDX_HEADER_SIZE
, t
->index_header_size
);
707 offset
= IDX_COLUMN_DESCRIPTIONS
;
708 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
709 write_u16(buf
+ offset
+ IDX_CD_STORAGE_TYPE
,
711 write_u16(buf
+ offset
+ IDX_CD_STORAGE_FLAGS
,
713 if (cd
->storage_flags
& OSL_FIXED_SIZE
)
714 write_u32(buf
+ offset
+ IDX_CD_DATA_SIZE
,
716 strcpy(buf
+ offset
+ IDX_CD_NAME
, cd
->name
);
717 offset
+= index_column_description_size(cd
->name
);
719 assert(offset
= size
);
720 filename
= index_filename(t
->desc
);
721 ret
= para_write_file(filename
, buf
, size
);
728 * Create a new osl table.
730 * \param desc Pointer to the table description.
732 * \return Positive on success, negative on errors. Possible errors include: \p
733 * E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_BAD_BASENAME, \p E_NO_COMPARE_FUNC, \p
734 * E_NO_COLUMN_NAME, \p E_DUPLICATE_COL_NAME, \p E_MKDIR, any errors returned
737 int osl_create_table(const struct osl_table_description
*desc
)
739 const struct osl_column_description
*cd
;
740 char *table_dir
= NULL
, *filename
;
742 int i
, ret
= init_table_structure(desc
, &t
);
746 PARA_INFO_LOG("creating %s\n", desc
->name
);
747 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
748 if (cd
->storage_type
== OSL_NO_STORAGE
)
751 ret
= para_mkdir(desc
->dir
, 0777);
752 if (ret
< 0 && ret
!= -E_EXIST
)
754 table_dir
= make_message("%s/%s", desc
->dir
,
756 ret
= para_mkdir(table_dir
, 0777);
760 filename
= column_filename(t
, i
);
761 PARA_INFO_LOG("filename: %s\n", filename
);
762 if (cd
->storage_type
== OSL_MAPPED_STORAGE
) {
763 ret
= para_open(filename
, O_RDWR
| O_CREAT
| O_EXCL
,
772 ret
= para_mkdir(filename
, 0777);
777 if (t
->num_mapped_columns
) {
778 ret
= create_table_index(t
);
790 static int table_is_dirty(struct osl_table
*t
)
792 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
793 uint8_t dirty
= read_u8(buf
) & 0x1;
797 static void mark_table_dirty(struct osl_table
*t
)
799 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
800 write_u8(buf
, read_u8(buf
) | 1);
803 static void mark_table_clean(struct osl_table
*t
)
805 char *buf
= (char *)t
->index_map
.data
+ IDX_DIRTY_FLAG
;
806 write_u8(buf
, read_u8(buf
) & 0xfe);
810 * Unmap all mapped files of an osl table.
812 * \param t Pointer to a mapped table.
813 * \param flags Options for unmapping.
815 * \return Positive on success, negative on errors. Possible errors include:
816 * E_NOT_MAPPED, E_MUNMAP.
818 * \sa map_table(), enum osl_close_flags, para_munmap().
820 int unmap_table(struct osl_table
*t
, enum osl_close_flags flags
)
823 const struct osl_column_description
*cd
;
826 if (!t
->num_mapped_columns
) /* can this ever happen? */
828 PARA_INFO_LOG("unmapping table '%s'\n", t
->desc
->name
);
829 if (!t
->index_map
.data
)
830 return -E_NOT_MAPPED
;
831 if (flags
& OSL_MARK_CLEAN
)
833 ret
= para_munmap(t
->index_map
.data
, t
->index_map
.size
);
836 t
->index_map
.data
= NULL
;
839 FOR_EACH_MAPPED_COLUMN(i
, t
, cd
) {
840 struct osl_object map
= t
->columns
[i
].data_map
;
843 ret
= para_munmap(map
.data
, map
.size
);
852 * Map the index file and all columns of type \p OSL_MAPPED_STORAGE into memory.
854 * \param t Pointer to an initialized table structure.
855 * \param flags Mapping options.
857 * \return Negative return value on errors; on success the number of rows
858 * (including invalid rows) is returned.
860 * \sa unmap_table(), enum map_table_flags, osl_open_table(), mmap(2).
862 int map_table(struct osl_table
*t
, enum map_table_flags flags
)
865 const struct osl_column_description
*cd
;
866 int i
= 0, ret
, num_rows
= 0;
868 if (!t
->num_mapped_columns
)
870 if (t
->index_map
.data
)
871 return -E_ALREADY_MAPPED
;
872 filename
= index_filename(t
->desc
);
873 PARA_DEBUG_LOG("mapping table '%s' (index: %s)\n", t
->desc
->name
, filename
);
874 ret
= mmap_full_file(filename
, flags
& MAP_TBL_FL_MAP_RDONLY
?
875 O_RDONLY
: O_RDWR
, &t
->index_map
);
879 if (flags
& MAP_TBL_FL_VERIFY_INDEX
) {
880 ret
= compare_table_descriptions(t
);
885 if (!(flags
& MAP_TBL_FL_IGNORE_DIRTY
)) {
886 if (table_is_dirty(t
)) {
887 PARA_ERROR_LOG("%s is dirty\n", t
->desc
->name
);
892 num_rows
= table_num_rows(t
);
896 FOR_EACH_MAPPED_COLUMN(i
, t
, cd
) {
898 filename
= column_filename(t
, i
);
900 if (stat(filename
, &statbuf
) < 0) {
904 if (!(S_IFREG
& statbuf
.st_mode
)) {
908 ret
= mmap_full_file(filename
, O_RDWR
,
909 &t
->columns
[i
].data_map
);
915 err
: /* unmap what is already mapped */
916 for (i
--; i
>= 0; i
--) {
917 struct osl_object map
= t
->columns
[i
].data_map
;
918 para_munmap(map
.data
, map
.size
);
921 para_munmap(t
->index_map
.data
, t
->index_map
.size
);
922 t
->index_map
.data
= NULL
;
927 * Retrieve a mapped object by row and column number.
929 * \param t Pointer to an open osl table.
930 * \param col_num Number of the mapped column containing the object to retrieve.
931 * \param row_num Number of the row containing the object to retrieve.
932 * \param obj The result is returned here.
934 * It is considered an error if \a col_num does not refer to a column
935 * of storage type \p OSL_MAPPED_STORAGE.
937 * \return Positive on success, negative on errors. Possible errors include:
938 * \p E_BAD_ID, \p E_INVALID_OBJECT.
940 * \sa osl_storage_type.
942 int get_mapped_object(const struct osl_table
*t
, unsigned col_num
,
943 uint32_t row_num
, struct osl_object
*obj
)
945 struct osl_column
*col
= &t
->columns
[col_num
];
951 if (t
->num_rows
<= row_num
)
953 ret
= get_index_entry(t
, row_num
, col_num
, &index_entry
);
956 offset
= read_u32(index_entry
);
957 obj
->size
= read_u32(index_entry
+ 4) - 1;
958 PARA_DEBUG_LOG("index_entry: %p\n", index_entry
);
959 header
= col
->data_map
.data
+ offset
;
960 obj
->data
= header
+ 1;
961 if (read_u8(header
) == 0xff) {
962 PARA_ERROR_LOG("col %u, size %zu, offset %u\n", col_num
,
964 return -E_INVALID_OBJECT
;
966 PARA_DEBUG_LOG("mapped obj row_num: %u, col %u, size: %zu\n", row_num
,
971 static int search_rbtree(const struct osl_object
*obj
,
972 const struct osl_table
*t
, unsigned col_num
,
973 struct rb_node
**result
, struct rb_node
***rb_link
)
975 struct osl_column
*col
= &t
->columns
[col_num
];
976 struct rb_node
**new = &col
->rbtree
.rb_node
, *parent
= NULL
;
977 const struct osl_column_description
*cd
=
978 get_column_description(t
->desc
, col_num
);
979 enum osl_storage_type st
= cd
->storage_type
;
981 struct osl_row
*this_row
= get_row_pointer(*new,
984 struct osl_object this_obj
;
986 if (st
== OSL_MAPPED_STORAGE
) {
987 ret
= get_mapped_object(t
, col_num
, this_row
->id
,
992 this_obj
= this_row
->volatile_objects
[col
->volatile_num
];
993 ret
= cd
->compare_function(obj
, &this_obj
);
996 *result
= get_rb_node_pointer(this_row
,
1001 new = &((*new)->rb_left
);
1003 new = &((*new)->rb_right
);
1009 return -E_RB_KEY_NOT_FOUND
;
1012 static int insert_rbtree(struct osl_table
*t
, unsigned col_num
,
1013 const struct osl_row
*row
, const struct osl_object
*obj
)
1015 struct rb_node
*parent
, **rb_link
;
1016 unsigned rbtree_num
;
1018 int ret
= search_rbtree(obj
, t
, col_num
, &parent
, &rb_link
);
1021 return -E_RB_KEY_EXISTS
;
1022 rbtree_num
= t
->columns
[col_num
].rbtree_num
;
1023 n
= get_rb_node_pointer(row
, rbtree_num
);
1024 rb_link_node(n
, parent
, rb_link
);
1025 rb_insert_color(n
, &t
->columns
[col_num
].rbtree
);
1029 static void remove_rb_node(struct osl_table
*t
, unsigned col_num
,
1030 const struct osl_row
*row
)
1032 struct osl_column
*col
= &t
->columns
[col_num
];
1033 const struct osl_column_description
*cd
=
1034 get_column_description(t
->desc
, col_num
);
1035 enum osl_storage_flags sf
= cd
->storage_flags
;
1036 struct rb_node
*victim
, *splice_out_node
, *tmp
;
1037 if (!(sf
& OSL_RBTREE
))
1040 * Which node is removed/spliced out actually depends on how many
1041 * children the victim node has: If it has no children, it gets
1042 * deleted. If it has one child, it gets spliced out. If it has two
1043 * children, its successor (which has at most a right child) gets
1046 victim
= get_rb_node_pointer(row
, col
->rbtree_num
);
1047 if (victim
->rb_left
&& victim
->rb_right
)
1048 splice_out_node
= rb_next(victim
);
1050 splice_out_node
= victim
;
1051 /* Go up to the root and decrement the size of each node in the path. */
1052 for (tmp
= splice_out_node
; tmp
; tmp
= rb_parent(tmp
))
1054 rb_erase(victim
, &col
->rbtree
);
1057 static int add_row_to_rbtrees(struct osl_table
*t
, uint32_t id
,
1058 struct osl_object
*volatile_objs
, struct osl_row
**row_ptr
)
1062 struct osl_row
*row
= allocate_row(t
->num_rbtrees
);
1063 const struct osl_column_description
*cd
;
1065 PARA_DEBUG_LOG("row: %p, id: %u\n", row
, id
);
1067 row
->volatile_objects
= volatile_objs
;
1068 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
) {
1069 if (cd
->storage_type
== OSL_MAPPED_STORAGE
) {
1070 struct osl_object obj
;
1071 ret
= get_mapped_object(t
, i
, id
, &obj
);
1074 ret
= insert_rbtree(t
, i
, row
, &obj
);
1075 } else { /* volatile */
1076 const struct osl_object
*obj
1077 = volatile_objs
+ t
->columns
[i
].volatile_num
;
1078 PARA_DEBUG_LOG("inserting %p\n", obj
->data
);
1079 ret
= insert_rbtree(t
, i
, row
, obj
);
1087 err
: /* rollback changes, i.e. remove added entries from rbtrees */
1089 remove_rb_node(t
, i
--, row
);
1094 static void free_volatile_objects(const struct osl_table
*t
,
1095 enum osl_close_flags flags
)
1099 struct osl_column
*rb_col
;
1100 const struct osl_column_description
*cd
;
1102 if (!t
->num_volatile_columns
)
1104 /* find the first rbtree column (any will do) */
1105 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
)
1107 rb_col
= t
->columns
+ i
;
1108 /* walk that rbtree and free all volatile objects */
1109 for (n
= rb_first(&rb_col
->rbtree
); n
; n
= rb_next(n
)) {
1110 struct osl_row
*r
= get_row_pointer(n
, rb_col
->rbtree_num
);
1111 if (flags
& OSL_FREE_VOLATILE
)
1112 for (j
= 0; j
< t
->num_volatile_columns
; j
++)
1113 free(r
->volatile_objects
[j
].data
);
1114 free(r
->volatile_objects
);
1119 * Erase all rbtree nodes and free resources.
1121 * \param t Pointer to an open osl table.
1123 * This function is called by osl_close_table().
1125 void clear_rbtrees(struct osl_table
*t
)
1127 const struct osl_column_description
*cd
;
1128 unsigned i
, rbtrees_cleared
= 0;
1130 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
) {
1131 struct osl_column
*col
= &t
->columns
[i
];
1134 for (n
= rb_first(&col
->rbtree
); n
;) {
1136 rb_erase(n
, &col
->rbtree
);
1137 if (rbtrees_cleared
== t
->num_rbtrees
) {
1138 r
= get_row_pointer(n
, col
->rbtree_num
);
1149 * Close an osl table.
1151 * \param t Pointer to the table to be closed.
1152 * \param flags Options for what should be cleaned up.
1154 * If osl_open_table() succeeds, the resulting table pointer must later be
1155 * passed to this function in order to flush all changes to the filesystem and
1156 * to free the resources that were allocated by osl_open_table().
1158 * \return Positive on success, negative on errors. Possible errors: \p E_BAD_TABLE,
1159 * errors returned by unmap_table().
1161 * \sa osl_open_table(), unmap_table().
1163 int osl_close_table(struct osl_table
*t
, enum osl_close_flags flags
)
1168 return -E_BAD_TABLE
;
1169 free_volatile_objects(t
, flags
);
1171 ret
= unmap_table(t
, flags
);
1173 PARA_ERROR_LOG("unmap_table failed: %d\n", ret
);
1180 * Find out whether the given row number corresponds to an invalid row.
1182 * \param t Pointer to the osl table.
1183 * \param row_num The number of the row in question.
1185 * By definition, a row is considered invalid if all its index entries
1188 * \return Positive if \a row_num corresponds to an invalid row,
1189 * zero if it corresponds to a valid row, negative on errors.
1191 int row_is_invalid(struct osl_table
*t
, uint32_t row_num
)
1194 int i
, ret
= get_index_entry_start(t
, row_num
, &index_entry
);
1198 for (i
= 0; i
< t
->index_entry_size
; i
++) {
1199 if ((unsigned char)index_entry
[i
] != 0xff)
1202 PARA_INFO_LOG("row %d is invalid\n", row_num
);
1207 * Invalidate a row of an osl table.
1209 * \param t Pointer to an open osl table.
1210 * \param row_num Number of the row to mark as invalid.
1212 * This function marks each mapped object in the index entry of \a row as
1215 * \return Positive on success, negative on errors.
1217 int mark_row_invalid(struct osl_table
*t
, uint32_t row_num
)
1220 int i
, ret
= get_index_entry_start(t
, row_num
, &index_entry
);
1222 PARA_INFO_LOG("marking row %d as invalid\n", row_num
);
1225 for (i
= 0; i
< t
->index_entry_size
; i
++)
1226 index_entry
[i
] = 0xff;
1231 * Initialize all rbtrees and compute number of invalid rows.
1233 * \param t The table containing the rbtrees to be initialized.
1235 * \return Positive on success, negative on errors.
1237 int init_rbtrees(struct osl_table
*t
)
1240 const struct osl_column_description
*cd
;
1242 /* create rbtrees */
1243 FOR_EACH_RBTREE_COLUMN(i
, t
, cd
)
1244 t
->columns
[i
].rbtree
= RB_ROOT
;
1245 /* add valid rows to rbtrees */
1246 t
->num_invalid_rows
= 0;
1247 for (i
= 0; i
< t
->num_rows
; i
++) {
1248 ret
= row_is_invalid(t
, i
);
1252 t
->num_invalid_rows
++;
1255 ret
= add_row_to_rbtrees(t
, i
, NULL
, NULL
);
1263 * Open an osl table.
1265 * Each osl table must be opened before its data can be accessed.
1267 * \param table_desc Describes the table to be opened.
1268 * \param result Contains a pointer to the open table on success.
1270 * The table description given by \a desc should coincide with the
1271 * description used at creation time.
1273 * \return Positive on success, negative on errors. Possible errors include:
1274 * errors returned by init_table_structure(), \p E_NOENT, \p E_STAT, \p \p
1275 * E_NOTDIR, \p E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_NO_COMPARE_FUNC, \p
1276 * E_NO_COLUMN_NAME, errors returned by init_rbtrees().
1278 int osl_open_table(const struct osl_table_description
*table_desc
,
1279 struct osl_table
**result
)
1282 struct osl_table
*t
;
1283 const struct osl_column_description
*cd
;
1285 PARA_INFO_LOG("opening table %s\n", table_desc
->name
);
1286 ret
= init_table_structure(table_desc
, &t
);
1289 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
) {
1290 /* check if directory exists */
1291 char *dirname
= column_filename(t
, i
);
1292 struct stat statbuf
;
1293 ret
= stat(dirname
, &statbuf
);
1296 if (errno
== ENOENT
)
1303 if (!S_ISDIR(statbuf
.st_mode
))
1306 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1310 PARA_DEBUG_LOG("num rows: %d\n", t
->num_rows
);
1311 ret
= init_rbtrees(t
);
1313 osl_close_table(t
, OSL_MARK_CLEAN
); /* ignore further errors */
1324 static int create_disk_storage_object_dir(const struct osl_table
*t
,
1325 unsigned col_num
, const char *ds_name
)
1330 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1332 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1333 ret
= para_mkdir(dirname
, 0777);
1335 if (ret
< 0 && ret
!= -E_EXIST
)
1340 static int write_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1341 const struct osl_object
*obj
, const char *ds_name
)
1346 ret
= create_disk_storage_object_dir(t
, col_num
, ds_name
);
1349 filename
= disk_storage_path(t
, col_num
, ds_name
);
1350 ret
= para_write_file(filename
, obj
->data
, obj
->size
);
1355 static int append_map_file(const struct osl_table
*t
, unsigned col_num
,
1356 const struct osl_object
*obj
, uint32_t *new_size
)
1358 char *filename
= column_filename(t
, col_num
);
1360 char header
= 0; /* zero means valid object */
1362 PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj
->size
);
1363 ret
= append_file(filename
, &header
, 1, obj
->data
, obj
->size
,
1369 static int append_index_entry(const struct osl_table
*t
, char *new_index_entry
)
1374 if (!t
->num_mapped_columns
)
1376 filename
= index_filename(t
->desc
);
1377 PARA_DEBUG_LOG("appending %u bytes\n", t
->index_entry_size
);
1378 ret
= append_file(filename
, NULL
, 0, new_index_entry
,
1379 t
->index_entry_size
, NULL
);
1385 * A wrapper for truncate(2)
1387 * \param path Name of the regular file to truncate
1388 * \param size Number of bytes to \b shave \b off
1390 * Truncate the regular file named by \a path by \a size bytes.
1392 * \return Positive on success, negative on errors. Possible errors include: \p
1393 * E_STAT, \p E_BAD_SIZE, \p E_TRUNC.
1397 int para_truncate(const char *path
, off_t size
)
1400 struct stat statbuf
;
1403 if (stat(path
, &statbuf
) < 0)
1406 if (statbuf
.st_size
< size
)
1409 if (truncate(path
, statbuf
.st_size
- size
) < 0)
1416 static int truncate_mapped_file(const struct osl_table
*t
, unsigned col_num
,
1419 char *filename
= column_filename(t
, col_num
);
1420 int ret
= para_truncate(filename
, size
);
1425 static int delete_disk_storage_file(const struct osl_table
*t
, unsigned col_num
,
1426 const char *ds_name
)
1428 char *dirname
, *filename
= disk_storage_path(t
, col_num
, ds_name
);
1429 int ret
= unlink(filename
);
1431 PARA_INFO_LOG("deleted %s\n", filename
);
1434 if (errno
== ENOENT
)
1438 if (!(t
->desc
->flags
& OSL_LARGE_TABLE
))
1440 dirname
= disk_storage_dirname(t
, col_num
, ds_name
);
1447 * Add a new row to an osl table and retrieve this row.
1449 * \param t Pointer to an open osl table.
1450 * \param objects Array of objects to be added.
1451 * \param row Result pointer.
1453 * The \a objects parameter must point to an array containing one object per
1454 * column. The order of the objects in the array is given by the table
1455 * description of \a table. Several sanity checks are performed during object
1456 * insertion and the function returns without modifying the table if any of
1457 * these tests fail. In fact, it is atomic in the sense that it either
1458 * succeeds or leaves the table unchanged (i.e. either all or none of the
1459 * objects are added to the table).
1461 * It is considered an error if an object is added to a column with associated
1462 * rbtree if this object is equal to an object already contained in that column
1463 * (i.e. the compare function for the column's rbtree returns zero).
1465 * Possible errors include: \p E_RB_KEY_EXISTS, \p E_BAD_DATA_SIZE.
1467 * \return Positive on success, negative on errors.
1469 * \sa struct osl_table_description, osl_compare_func, osl_add_row().
1471 int osl_add_and_get_row(struct osl_table
*t
, struct osl_object
*objects
,
1472 struct osl_row
**row
)
1475 char *ds_name
= NULL
;
1476 struct rb_node
**rb_parents
= NULL
, ***rb_links
= NULL
;
1477 char *new_index_entry
= NULL
;
1478 struct osl_object
*volatile_objs
= NULL
;
1479 const struct osl_column_description
*cd
;
1482 return -E_BAD_TABLE
;
1483 rb_parents
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
*));
1484 rb_links
= para_malloc(t
->num_rbtrees
* sizeof(struct rn_node
**));
1485 if (t
->num_mapped_columns
)
1486 new_index_entry
= para_malloc(t
->index_entry_size
);
1487 /* pass 1: sanity checks */
1488 PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects
[0].data
,
1490 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1491 enum osl_storage_type st
= cd
->storage_type
;
1492 enum osl_storage_flags sf
= cd
->storage_flags
;
1494 // ret = -E_NULL_OBJECT;
1497 if (st
== OSL_DISK_STORAGE
)
1499 if (sf
& OSL_RBTREE
) {
1500 unsigned rbtree_num
= t
->columns
[i
].rbtree_num
;
1501 ret
= -E_RB_KEY_EXISTS
;
1502 PARA_DEBUG_LOG("checking whether %p exists\n",
1504 if (search_rbtree(objects
+ i
, t
, i
,
1505 &rb_parents
[rbtree_num
],
1506 &rb_links
[rbtree_num
]) > 0)
1509 if (sf
& OSL_FIXED_SIZE
) {
1510 PARA_DEBUG_LOG("fixed size. need: %zu, have: %d\n",
1511 objects
[i
].size
, cd
->data_size
);
1512 ret
= -E_BAD_DATA_SIZE
;
1513 if (objects
[i
].size
!= cd
->data_size
)
1517 if (t
->num_disk_storage_columns
)
1518 ds_name
= disk_storage_name_of_object(t
,
1519 &objects
[t
->disk_storage_name_column
]);
1520 ret
= unmap_table(t
, OSL_MARK_CLEAN
);
1523 PARA_DEBUG_LOG("sanity tests passed%s\n", "");
1524 /* pass 2: create data files, append map data */
1525 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1526 enum osl_storage_type st
= cd
->storage_type
;
1527 if (st
== OSL_NO_STORAGE
)
1529 if (st
== OSL_MAPPED_STORAGE
) {
1531 struct osl_column
*col
= &t
->columns
[i
];
1532 PARA_DEBUG_LOG("appending object of size %zu\n",
1534 ret
= append_map_file(t
, i
, objects
+ i
, &new_size
);
1537 update_index_entry(new_index_entry
, col
, new_size
,
1542 ret
= write_disk_storage_file(t
, i
, objects
+ i
, ds_name
);
1546 ret
= append_index_entry(t
, new_index_entry
);
1549 ret
= map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1550 if (ret
< 0) { /* truncate index and rollback changes */
1551 char *filename
= index_filename(t
->desc
);
1552 para_truncate(filename
, t
->index_entry_size
);
1556 /* pass 3: add entry to rbtrees */
1557 if (t
->num_volatile_columns
) {
1558 volatile_objs
= para_calloc(t
->num_volatile_columns
1559 * sizeof(struct osl_object
));
1560 FOR_EACH_VOLATILE_COLUMN(i
, t
, cd
)
1561 volatile_objs
[t
->columns
[i
].volatile_num
] = objects
[i
];
1564 PARA_DEBUG_LOG("adding new entry as row #%d\n", t
->num_rows
- 1);
1565 ret
= add_row_to_rbtrees(t
, t
->num_rows
- 1, volatile_objs
, row
);
1568 PARA_DEBUG_LOG("added new entry as row #%d\n", t
->num_rows
- 1);
1571 rollback
: /* rollback all changes made, ignore further errors */
1572 for (i
--; i
>= 0; i
--) {
1573 cd
= get_column_description(t
->desc
, i
);
1574 enum osl_storage_type st
= cd
->storage_type
;
1575 if (st
== OSL_NO_STORAGE
)
1578 if (st
== OSL_MAPPED_STORAGE
)
1579 truncate_mapped_file(t
, i
, objects
[i
].size
);
1580 else /* disk storage */
1581 delete_disk_storage_file(t
, i
, ds_name
);
1583 /* ignore error and return previous error value */
1584 map_table(t
, MAP_TBL_FL_VERIFY_INDEX
);
1586 free(new_index_entry
);
1594 * Add a new row to an osl table.
1596 * \param t Same meaning as osl_add_and_get_row().
1597 * \param objects Same meaning as osl_add_and_get_row().
1599 * \return The return value of the underlying call to osl_add_and_get_row().
1601 * This is equivalent to osl_add_and_get_row(t, objects, NULL).
1603 int osl_add_row(struct osl_table
*t
, struct osl_object
*objects
)
1605 return osl_add_and_get_row(t
, objects
, NULL
);
1609 * Retrieve an object identified by row and column
1611 * \param t Pointer to an open osl table.
1612 * \param r Pointer to the row.
1613 * \param col_num The column number.
1614 * \param object The result pointer.
1616 * The column determined by \a col_num must be of type \p OSL_MAPPED_STORAGE
1617 * or \p OSL_NO_STORAGE, i.e. no disk storage objects may be retrieved by this
1620 * \return Positive if object was found, negative on errors. Possible errors
1621 * include: \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE.
1623 * \sa osl_storage_type, osl_open_disk_object().
1625 int osl_get_object(const struct osl_table
*t
, const struct osl_row
*r
,
1626 unsigned col_num
, struct osl_object
*object
)
1628 const struct osl_column_description
*cd
;
1631 return -E_BAD_TABLE
;
1632 cd
= get_column_description(t
->desc
, col_num
);
1633 /* col must not be disk storage */
1634 if (cd
->storage_type
== OSL_DISK_STORAGE
)
1635 return -E_BAD_STORAGE_TYPE
;
1636 if (cd
->storage_type
== OSL_MAPPED_STORAGE
)
1637 return get_mapped_object(t
, col_num
, r
->id
, object
);
1639 *object
= r
->volatile_objects
[t
->columns
[col_num
].volatile_num
];
1643 static int mark_mapped_object_invalid(const struct osl_table
*t
, uint32_t id
,
1646 struct osl_object obj
;
1648 int ret
= get_mapped_object(t
, col_num
, id
, &obj
);
1659 * Delete a row from an osl table.
1661 * \param t Pointer to an open osl table.
1662 * \param row Pointer to the row to delete.
1664 * This removes all disk storage objects, removes all rbtree nodes, and frees
1665 * all volatile objects belonging to the given row. For mapped columns, the
1666 * data is merely marked invalid and may be pruned from time to time by
1669 * \return Positive on success, negative on errors. Possible errors include:
1670 * \p E_BAD_TABLE, errors returned by osl_get_object().
1672 int osl_del_row(struct osl_table
*t
, struct osl_row
*row
)
1674 struct osl_row
*r
= row
;
1676 const struct osl_column_description
*cd
;
1679 return -E_BAD_TABLE
;
1680 PARA_INFO_LOG("deleting row %p\n", row
);
1682 if (t
->num_disk_storage_columns
) {
1684 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1687 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
)
1688 delete_disk_storage_file(t
, i
, ds_name
);
1691 FOR_EACH_COLUMN(i
, t
->desc
, cd
) {
1692 struct osl_column
*col
= t
->columns
+ i
;
1693 enum osl_storage_type st
= cd
->storage_type
;
1694 remove_rb_node(t
, i
, r
);
1695 if (st
== OSL_MAPPED_STORAGE
) {
1696 mark_mapped_object_invalid(t
, r
->id
, i
);
1699 if (st
== OSL_NO_STORAGE
)
1700 free(r
->volatile_objects
[col
->volatile_num
].data
);
1702 if (t
->num_mapped_columns
) {
1703 ret
= mark_row_invalid(t
, r
->id
);
1706 t
->num_invalid_rows
++;
1711 free(r
->volatile_objects
);
1716 /* test if column has an rbtree */
1717 static int check_rbtree_col(const struct osl_table
*t
, unsigned col_num
,
1718 struct osl_column
**col
)
1721 return -E_BAD_TABLE
;
1722 if (!(get_column_description(t
->desc
, col_num
)->storage_flags
& OSL_RBTREE
))
1723 return -E_BAD_STORAGE_FLAGS
;
1724 *col
= t
->columns
+ col_num
;
1729 * Get the row that contains the given object.
1731 * \param t Pointer to an open osl table.
1732 * \param col_num The number of the column to be searched.
1733 * \param obj The object to be looked up.
1734 * \param result Points to the row containing \a obj.
1736 * Lookup \a obj in \a t and return the row containing \a obj. The column
1737 * specified by \a col_num must have an associated rbtree.
1739 * \return Positive on success, negative on errors. If an error occured, \a
1740 * result is set to \p NULL. Possible errors include: \p E_BAD_TABLE, \p
1741 * E_BAD_STORAGE_FLAGS, errors returned by get_mapped_object(), \p
1742 * E_RB_KEY_NOT_FOUND.
1744 * \sa osl_storage_flags
1746 int osl_get_row(const struct osl_table
*t
, unsigned col_num
,
1747 const struct osl_object
*obj
, struct osl_row
**result
)
1750 struct rb_node
*node
;
1751 struct osl_row
*row
;
1752 struct osl_column
*col
;
1755 ret
= check_rbtree_col(t
, col_num
, &col
);
1758 ret
= search_rbtree(obj
, t
, col_num
, &node
, NULL
);
1761 row
= get_row_pointer(node
, t
->columns
[col_num
].rbtree_num
);
1766 static int rbtree_loop(struct osl_column
*col
, void *private_data
,
1767 osl_rbtree_loop_func
*func
)
1771 for (n
= rb_first(&col
->rbtree
); n
; n
= rb_next(n
)) {
1772 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1773 int ret
= func(r
, private_data
);
1780 static int rbtree_loop_reverse(struct osl_column
*col
, void *private_data
,
1781 osl_rbtree_loop_func
*func
)
1785 for (n
= rb_last(&col
->rbtree
); n
; n
= rb_prev(n
)) {
1786 struct osl_row
*r
= get_row_pointer(n
, col
->rbtree_num
);
1787 int ret
= func(r
, private_data
);
1795 * Loop over all nodes in an rbtree.
1797 * \param t Pointer to an open osl table.
1798 * \param col_num The column to use for iterating over the elements.
1799 * \param private_data Pointer that gets passed to \a func.
1800 * \param func The function to be called for each node in the rbtree.
1802 * This function does an in-order walk of the rbtree associated with \a
1803 * col_num. It is an error if the \p OSL_RBTREE flag is not set for this
1804 * column. For each node in the rbtree, the given function \a func is called
1805 * with two \p void* pointers as arguments: The first argument points to the
1806 * row that contains the object corresponding to the rbtree node currently
1807 * traversed, and the \a private_data pointer is passed to \a func as the
1808 * second argument. The loop terminates either if \a func returns a negative
1809 * value, or if all nodes of the tree have been visited.
1812 * \return Positive on success, negative on errors. If the termination of the
1813 * loop was caused by \a func returning a negative value, this value is
1816 * \sa osl_storage_flags, osl_rbtree_loop_reverse(), osl_compare_func.
1818 int osl_rbtree_loop(const struct osl_table
*t
, unsigned col_num
,
1819 void *private_data
, osl_rbtree_loop_func
*func
)
1821 struct osl_column
*col
;
1823 int ret
= check_rbtree_col(t
, col_num
, &col
);
1826 return rbtree_loop(col
, private_data
, func
);
1830 * Loop over all nodes in an rbtree in reverse order.
1832 * \param t Identical meaning as in \p osl_rbtree_loop().
1833 * \param col_num Identical meaning as in \p osl_rbtree_loop().
1834 * \param private_data Identical meaning as in \p osl_rbtree_loop().
1835 * \param func Identical meaning as in \p osl_rbtree_loop().
1837 * This function is identical to \p osl_rbtree_loop(), the only difference
1838 * is that the tree is walked in reverse order.
1840 * \return The same return value as \p osl_rbtree_loop().
1842 * \sa osl_rbtree_loop().
1844 int osl_rbtree_loop_reverse(const struct osl_table
*t
, unsigned col_num
,
1845 void *private_data
, osl_rbtree_loop_func
*func
)
1847 struct osl_column
*col
;
1849 int ret
= check_rbtree_col(t
, col_num
, &col
);
1852 return rbtree_loop_reverse(col
, private_data
, func
);
1855 /* TODO: Rollback changes on errors */
1856 static int rename_disk_storage_objects(struct osl_table
*t
,
1857 struct osl_object
*old_obj
, struct osl_object
*new_obj
)
1860 const struct osl_column_description
*cd
;
1861 char *old_ds_name
, *new_ds_name
;
1863 if (!t
->num_disk_storage_columns
)
1864 return 1; /* nothing to do */
1865 if (old_obj
->size
== new_obj
->size
&& !memcmp(new_obj
->data
,
1866 old_obj
->data
, new_obj
->size
))
1867 return 1; /* object did not change */
1868 old_ds_name
= disk_storage_name_of_object(t
, old_obj
);
1869 new_ds_name
= disk_storage_name_of_object(t
, new_obj
);
1870 FOR_EACH_DISK_STORAGE_COLUMN(i
, t
, cd
) {
1871 char *old_filename
, *new_filename
;
1872 ret
= create_disk_storage_object_dir(t
, i
, new_ds_name
);
1875 old_filename
= disk_storage_path(t
, i
, old_ds_name
);
1876 new_filename
= disk_storage_path(t
, i
, new_ds_name
);
1877 ret
= para_rename(old_filename
, new_filename
);
1892 * Change an object in an osl table.
1894 * \param t Pointer to an open osl table.
1895 * \param r Pointer to the row containing the object to be updated.
1896 * \param col_num Number of the column containing the object to be updated.
1897 * \param obj Pointer to the replacement object.
1899 * This function gets rid of all references to the old object. This includes
1900 * removal of the rbtree node in case there is an rbtree associated with \a
1901 * col_num. It then inserts \a obj into the table and the rbtree if neccessary.
1903 * If the \p OSL_RBTREE flag is set for \a col_num, you \b MUST call this
1904 * function in order to change the contents of an object, even for volatile or
1905 * mapped columns of constant size (which may be updated directly if \p
1906 * OSL_RBTREE is not set). Otherwise the rbtree might become corrupted.
1908 * \return Positive on success, negative on errors. Possible errors include: \p
1909 * E_BAD_TABLE, \p E_RB_KEY_EXISTS, \p E_BAD_SIZE, \p E_NOENT, \p E_UNLINK,
1910 * errors returned by para_write_file(), \p E_MKDIR.
1912 int osl_update_object(struct osl_table
*t
, const struct osl_row
*r
,
1913 unsigned col_num
, struct osl_object
*obj
)
1915 struct osl_column
*col
;
1916 const struct osl_column_description
*cd
;
1920 return -E_BAD_TABLE
;
1921 col
= &t
->columns
[col_num
];
1922 cd
= get_column_description(t
->desc
, col_num
);
1923 if (cd
->storage_flags
& OSL_RBTREE
) {
1924 if (search_rbtree(obj
, t
, col_num
, NULL
, NULL
) > 0)
1925 return -E_RB_KEY_EXISTS
;
1927 if (cd
->storage_flags
& OSL_FIXED_SIZE
) {
1928 if (obj
->size
!= cd
->data_size
)
1931 remove_rb_node(t
, col_num
, r
);
1932 if (cd
->storage_type
== OSL_NO_STORAGE
) { /* TODO: If fixed size, reuse object? */
1933 free(r
->volatile_objects
[col
->volatile_num
].data
);
1934 r
->volatile_objects
[col
->volatile_num
] = *obj
;
1935 } else if (cd
->storage_type
== OSL_DISK_STORAGE
) {
1937 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
1940 ret
= delete_disk_storage_file(t
, col_num
, ds_name
);
1941 if (ret
< 0 && ret
!= -E_NOENT
) {
1945 ret
= write_disk_storage_file(t
, col_num
, obj
, ds_name
);
1949 } else { /* mapped storage */
1950 struct osl_object old_obj
;
1951 ret
= get_mapped_object(t
, col_num
, r
->id
, &old_obj
);
1955 * If the updated column is the disk storage name column, the
1956 * disk storage name changes, so we have to rename all disk
1957 * storage objects accordingly.
1959 if (col_num
== t
->disk_storage_name_column
) {
1960 ret
= rename_disk_storage_objects(t
, &old_obj
, obj
);
1964 if (cd
->storage_flags
& OSL_FIXED_SIZE
)
1965 memcpy(old_obj
.data
, obj
->data
, cd
->data_size
);
1966 else { /* TODO: if the size doesn't change, use old space */
1967 uint32_t new_data_map_size
;
1969 ret
= get_index_entry_start(t
, r
->id
, &index_entry
);
1972 ret
= mark_mapped_object_invalid(t
, r
->id
, col_num
);
1975 ret
= append_map_file(t
, col_num
, obj
,
1976 &new_data_map_size
);
1979 update_index_entry(index_entry
, col
, new_data_map_size
,
1983 if (cd
->storage_flags
& OSL_RBTREE
) {
1984 ret
= insert_rbtree(t
, col_num
, r
, obj
);
1992 * Retrieve an object of type \p OSL_DISK_STORAGE by row and column.
1994 * \param t Pointer to an open osl table.
1995 * \param r Pointer to the row containing the object.
1996 * \param col_num The column number.
1997 * \param obj Points to the result upon successful return.
1999 * For columns of type \p OSL_DISK_STORAGE, this function must be used to
2000 * retrieve one of its containing objects. Afterwards, osl_close_disk_object()
2001 * must be called in order to deallocate the resources.
2003 * \return Positive on success, negative on errors. Possible errors include:
2004 * \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE, errors returned by osl_get_object().
2006 * \sa osl_get_object(), osl_storage_type, osl_close_disk_object().
2008 int osl_open_disk_object(const struct osl_table
*t
, const struct osl_row
*r
,
2009 unsigned col_num
, struct osl_object
*obj
)
2011 const struct osl_column_description
*cd
;
2012 char *ds_name
, *filename
;
2016 return -E_BAD_TABLE
;
2017 cd
= get_column_description(t
->desc
, col_num
);
2018 if (cd
->storage_type
!= OSL_DISK_STORAGE
)
2019 return -E_BAD_STORAGE_TYPE
;
2021 ret
= disk_storage_name_of_row(t
, r
, &ds_name
);
2024 filename
= disk_storage_path(t
, col_num
, ds_name
);
2026 PARA_DEBUG_LOG("filename: %s\n", filename
);
2027 ret
= mmap_full_file(filename
, O_RDONLY
, obj
);
2033 * Free resources that were allocated during osl_open_disk_object().
2035 * \param obj Pointer to the object previously returned by open_disk_object().
2037 * \return The return value of the underlying call to para_munmap().
2039 * \sa para_munmap().
2041 int osl_close_disk_object(struct osl_object
*obj
)
2043 return para_munmap(obj
->data
, obj
->size
);
2047 * Get the number of rows of the given table.
2049 * \param t Pointer to an open osl table.
2050 * \param num_rows Result is returned here.
2052 * The number of rows returned via \a num_rows excluding any invalid rows.
2054 * \return Positive on success, \p -E_BAD_TABLE if \a t is \p NULL.
2056 int osl_get_num_rows(const struct osl_table
*t
, unsigned *num_rows
)
2059 return -E_BAD_TABLE
;
2060 assert(t
->num_rows
>= t
->num_invalid_rows
);
2061 *num_rows
= t
->num_rows
- t
->num_invalid_rows
;
2066 * Get the rank of a row.
2068 * \param t An open osl table.
2069 * \param r The row to get the rank of.
2070 * \param col_num The number of an rbtree column.
2071 * \param rank Result pointer.
2073 * The rank is, by definition, the position of the row in the linear order
2074 * determined by an inorder tree walk of the rbtree associated with column
2075 * number \a col_num of \a table.
2077 * \return Positive on success, negative on errors.
2079 * \sa osl_get_nth_row().
2081 int osl_get_rank(const struct osl_table
*t
, struct osl_row
*r
,
2082 unsigned col_num
, unsigned *rank
)
2084 struct osl_object obj
;
2085 struct osl_column
*col
;
2086 struct rb_node
*node
;
2087 int ret
= check_rbtree_col(t
, col_num
, &col
);
2091 ret
= osl_get_object(t
, r
, col_num
, &obj
);
2094 ret
= search_rbtree(&obj
, t
, col_num
, &node
, NULL
);
2097 ret
= rb_rank(node
, rank
);
2104 * Get the row with n-th greatest value.
2106 * \param t Pointer to an open osl table.
2107 * \param col_num The column number.
2108 * \param n The rank of the desired row.
2109 * \param result Row is returned here.
2111 * Retrieve the n-th order statistic with respect to the compare function
2112 * of the rbtree column \a col_num. In other words, get that row with
2113 * \a n th greatest value in column \a col_num. It's an error if
2114 * \a col_num is not a rbtree column, or if \a n is larger than the
2115 * number of rows in the table.
2117 * \return Positive on success, negative on errors. Possible errors:
2118 * \p E_BAD_TABLE, \p E_BAD_STORAGE_FLAGS, \p E_RB_KEY_NOT_FOUND.
2120 * \sa osl_storage_flags, osl_compare_func, osl_get_row(),
2121 * osl_rbtree_last_row(), osl_rbtree_first_row(), osl_get_rank().
2123 int osl_get_nth_row(const struct osl_table
*t
, unsigned col_num
,
2124 unsigned n
, struct osl_row
**result
)
2126 struct osl_column
*col
;
2127 struct rb_node
*node
;
2128 int ret
= check_rbtree_col(t
, col_num
, &col
);
2132 node
= rb_nth(col
->rbtree
.rb_node
, n
);
2134 return -E_RB_KEY_NOT_FOUND
;
2135 *result
= get_row_pointer(node
, col
->rbtree_num
);
2140 * Get the row corresponding to the smallest rbtree node of a column.
2142 * \param t An open rbtree table.
2143 * \param col_num The number of the rbtree column.
2144 * \param result A pointer to the first row is returned here.
2146 * The rbtree node of the smallest object (with respect to the corresponding
2147 * compare function) is selected and the row containing this object is
2148 * returned. It is an error if \a col_num refers to a column without an
2149 * associated rbtree.
2151 * \return Positive on success, negative on errors.
2153 * \sa osl_get_nth_row(), osl_rbtree_last_row().
2155 int osl_rbtree_first_row(const struct osl_table
*t
, unsigned col_num
,
2156 struct osl_row
**result
)
2158 return osl_get_nth_row(t
, col_num
, 1, result
);
2162 * Get the row corresponding to the greatest rbtree node of a column.
2164 * \param t The same meaning as in \p osl_rbtree_first_row().
2165 * \param col_num The same meaning as in \p osl_rbtree_first_row().
2166 * \param result The same meaning as in \p osl_rbtree_first_row().
2168 * This function works just like osl_rbtree_first_row(), the only difference
2169 * is that the row containing the greatest rather than the smallest object is
2172 * \return Positive on success, negative on errors.
2174 * \sa osl_get_nth_row(), osl_rbtree_first_row().
2176 int osl_rbtree_last_row(const struct osl_table
*t
, unsigned col_num
,
2177 struct osl_row
**result
)
2180 int ret
= osl_get_num_rows(t
, &num_rows
);
2184 return osl_get_nth_row(t
, col_num
, num_rows
, result
);