osl.c

   1 /*
   2  * Copyright (C) 2007 Andre Noll <maan@systemlinux.org>
   3  *
   4  * Licensed under the GPL v2. For licencing details see COPYING.
   5  */
   6
   7 /** \file osl.c Object storage layer functions. */
   8 #include "para.h"
   9 #include "error.h"
  10 #include "list.h"
  11 #include "osl_core.h"
  12 #include <dirent.h> /* readdir() */
  13 #include <assert.h>
  14
  15 /**
  16  * A wrapper for lseek(2).
  17  *
  18  * \param fd The filedescriptor whose offset is to be to repositioned.
  19  * \param offset A value-result parameter.
  20  * \param whence Usual repositioning directive.
  21  *
  22  * Reposition the offset of the file descriptor \a fd to the argument \a offset
  23  * according to the directive \a whence. Upon successful return, \a offset
  24  * contains the resulting offset location as measured in bytes from the
  25  * beginning of the file.
  26  *
  27  * \return Positive on success. Otherwise, the function returns \p -E_LSEEK.
  28  *
  29  * \sa lseek(2).
  30  */
  31 int para_lseek(int fd, off_t *offset, int whence)
  32 {
  33         *offset = lseek(fd, *offset, whence);
  34         int ret = -E_LSEEK;
  35         if (*offset == -1)
  36                 return ret;
  37         return 1;
  38 }
  39
  40 /**
  41  * Wrapper for the write system call.
  42  *
  43  * \param fd The file descriptor to write to.
  44  * \param buf The buffer to write.
  45  * \param size The length of \a buf in bytes.
  46  *
  47  * This function writes out the given bufffer and retries if an interrupt
  48  * occured during the write.
  49  *
  50  * \return On success, the number of bytes written is returned, otherwise, the
  51  * function returns \p -E_WRITE.
  52  *
  53  * \sa write(2).
  54  */
  55 ssize_t para_write(int fd, const void *buf, size_t size)
  56 {
  57         ssize_t ret;
  58
  59         for (;;) {
  60                 ret = write(fd, buf, size);
  61                 if ((ret < 0) && (errno == EAGAIN || errno == EINTR))
  62                         continue;
  63                 return ret >= 0? ret : -E_WRITE;
  64         }
  65 }
  66
  67 /**
  68  * Write the whole buffer to a file descriptor.
  69  *
  70  * \param fd The file descriptor to write to.
  71  * \param buf The buffer to write.
  72  * \param size The length of \a buf in bytes.
  73  *
  74  * This function writes the given buffer and continues on short writes and
  75  * when interrupted by a signal.
  76  *
  77  * \return Positive on success, negative on errors. Possible errors: any
  78  * errors returned by para_write().
  79  *
  80  * \sa para_write().
  81  */
  82 ssize_t para_write_all(int fd, const void *buf, size_t size)
  83 {
  84 //      PARA_DEBUG_LOG("writing %zu bytes\n", size);
  85         const char *b = buf;
  86         while (size) {
  87                 ssize_t ret = para_write(fd, b, size);
  88 //              PARA_DEBUG_LOG("ret: %zd\n", ret);
  89                 if (ret < 0)
  90                         return ret;
  91                 b += ret;
  92                 size -= ret;
  93         }
  94         return 1;
  95 }
  96 /**
  97  * Wrapper for the open(2) system call.
  98  *
  99  * \param path The filename.
 100  * \param flags The usual open(2) flags.
 101  * \param mode Specifies the permissions to use.
 102  *
 103  * The mode parameter must be specified when O_CREAT is in the flags, and is ignored
 104  * otherwise.
 105  *
 106  * \return Positive on success, negative on errors. Possible errors: \p
 107  * E_EXIST, \p E_ISDIR, \p E_NOENT, \p E_OSL_PERM.
 108  *
 109  * \sa open(2).
 110  */
 111 int para_open(const char *path, int flags, mode_t mode)
 112 {
 113         PARA_DEBUG_LOG("opening %s\n", path);
 114         int ret = open(path, flags, mode);
 115
 116         if (ret >= 0)
 117                 return ret;
 118         switch (errno) {
 119         case EEXIST:
 120                 ret = -E_EXIST;
 121                 break;
 122         case EISDIR:
 123                 ret = -E_ISDIR;
 124                 break;
 125         case ENOENT:
 126                 ret = -E_NOENT;
 127                 break;
 128         case EPERM:
 129                 ret = -E_OSL_PERM;
 130                 break;
 131         };
 132         PARA_ERROR_LOG("failed to open %s: %s\n", path, strerror(errno));
 133         return ret;
 134 }
 135
 136 /**
 137  * Open a file, write the given buffer and close the file.
 138  *
 139  * \param filename Full path to the file to open.
 140  * \param buf The buffer to write to the file.
 141  * \param size The size of \a buf.
 142  *
 143  * \return Positive on success, negative on errors. Possible errors include:
 144  * any errors from para_open() or para_write().
 145  *
 146  * \sa para_open(), para_write().
 147  */
 148 int para_write_file(const char *filename, const void *buf, size_t size)
 149 {
 150         int ret, fd;
 151
 152         ret = para_open(filename, O_WRONLY | O_CREAT | O_EXCL, 0644);
 153         if (ret < 0)
 154                 return ret;
 155         fd = ret;
 156         ret = para_write_all(fd, buf, size);
 157         if (ret < 0)
 158                 goto out;
 159         ret = 1;
 160 out:
 161         close(fd);
 162         return ret;
 163 }
 164
 165 static int append_file(const char *filename, char *header, size_t header_size,
 166         char *data, size_t data_size, uint32_t *new_pos)
 167 {
 168         int ret, fd;
 169
 170 //      PARA_DEBUG_LOG("appending %zu  + %zu bytes\n", header_size, data_size);
 171         ret = para_open(filename, O_WRONLY | O_CREAT | O_APPEND, 0644);
 172         if (ret < 0)
 173                 return ret;
 174         fd = ret;
 175         if (header && header_size) {
 176                 ret = para_write_all(fd, header, header_size);
 177                 if (ret < 0)
 178                         goto out;
 179         }
 180         ret = para_write_all(fd, data, data_size);
 181         if (ret < 0)
 182                 goto out;
 183         if (new_pos) {
 184                 off_t offset = 0;
 185                 ret = para_lseek(fd, &offset, SEEK_END);
 186                 if (ret < 0)
 187                         goto out;
 188 //              PARA_DEBUG_LOG("new file size: " FMT_OFF_T "\n", offset);
 189                 *new_pos = offset;
 190         }
 191         ret = 1;
 192 out:
 193         close(fd);
 194         return ret;
 195 }
 196
 197 /**
 198  * Map a file into memory.
 199  *
 200  * \param path Name of the regular file to map.
 201  * \param open_mode Either \p O_RDONLY or \p O_RDWR.
 202  * \param obj On success, the mapping is returned here.
 203  *
 204  * \return Positive on success, negative on errors. Possible errors include: \p
 205  * E_FSTAT, any errors returned by para_open(), \p E_EMPTY, \p E_MMAP.
 206  *
 207  * \sa para_open(), mmap(2).
 208  */
 209 int mmap_full_file(const char *path, int open_mode, struct osl_object *obj)
 210 {
 211         int fd, ret, mmap_prot, mmap_flags;
 212         struct stat file_status;
 213
 214         if (open_mode == O_RDONLY) {
 215                 mmap_prot = PROT_READ;
 216                 mmap_flags = MAP_PRIVATE;
 217         } else {
 218                 mmap_prot = PROT_READ | PROT_WRITE;
 219                 mmap_flags = MAP_SHARED;
 220         }
 221         ret = para_open(path, open_mode, 0);
 222         if (ret < 0)
 223                 return ret;
 224         fd = ret;
 225         ret = -E_FSTAT;
 226         if (fstat(fd, &file_status) < 0)
 227                 goto out;
 228         obj->size = file_status.st_size;
 229         ret = -E_EMPTY;
 230         PARA_DEBUG_LOG("%s: size %zu\n", path, obj->size);
 231         if (!obj->size)
 232                 goto out;
 233         obj->data = mmap(NULL, obj->size, mmap_prot, mmap_flags, fd, 0);
 234         if (obj->data == MAP_FAILED) {
 235                 obj->data = NULL;
 236                 ret = -E_MMAP;
 237                 goto out;
 238         }
 239         ret = 1;
 240 out:
 241         close(fd);
 242         return ret;
 243 }
 244
 245 /**
 246  * Traverse the given directory recursively.
 247  *
 248  * \param dirname The directory to traverse.
 249  * \param func The function to call for each entry.
 250  * \param private_data Pointer to an arbitrary data structure.
 251  *
 252  * For each regular file  in \a dirname, the supplied function \a func is
 253  * called.  The full path of the regular file and the \a private_data pointer
 254  * are passed to \a func.
 255  *
 256  * \return On success, 1 is returned. Otherwise, this function returns a
 257  * negative value which indicates the kind of the error.
 258  */
 259 int for_each_file_in_dir(const char *dirname,
 260                 int (*func)(const char *, const void *), const void *private_data)
 261 {
 262         DIR *dir = NULL;
 263         struct dirent *entry;
 264         /*
 265          * Opening the current directory (".") and calling fchdir() to return
 266          * is usually faster and more reliable than saving cwd in some buffer
 267          * and calling chdir() afterwards (see man 3 getcwd).
 268          */
 269         int cwd_fd = open(".", O_RDONLY);
 270         int ret = -1;
 271
 272 //      PARA_DEBUG_LOG("dirname: %s\n", dirname);
 273         if (cwd_fd < 0)
 274                 return -E_OSL_GETCWD;
 275         ret = -E_OSL_CHDIR;
 276         if (chdir(dirname) < 0)
 277                 goto out;
 278         ret = -E_OSL_OPENDIR;
 279         dir = opendir(".");
 280         if (!dir)
 281                 goto out;
 282         /* scan cwd recursively */
 283         while ((entry = readdir(dir))) {
 284                 mode_t m;
 285                 char *tmp;
 286                 struct stat s;
 287
 288                 if (!strcmp(entry->d_name, "."))
 289                         continue;
 290                 if (!strcmp(entry->d_name, ".."))
 291                         continue;
 292                 if (lstat(entry->d_name, &s) == -1)
 293                         continue;
 294                 m = s.st_mode;
 295                 if (!S_ISREG(m) && !S_ISDIR(m))
 296                         continue;
 297                 tmp = make_message("%s/%s", dirname, entry->d_name);
 298                 if (!S_ISDIR(m)) {
 299                         ret = func(tmp, private_data);
 300                         free(tmp);
 301                         if (ret < 0)
 302                                 goto out;
 303                         continue;
 304                 }
 305                 /* directory */
 306                 ret = for_each_file_in_dir(tmp, func, private_data);
 307                 free(tmp);
 308                 if (ret < 0)
 309                         goto out;
 310         }
 311         ret = 1;
 312 out:
 313         if (dir)
 314                 closedir(dir);
 315         if (fchdir(cwd_fd) < 0 && ret >= 0)
 316                 ret = -E_OSL_CHDIR;
 317         close(cwd_fd);
 318         return ret;
 319 }
 320
 321 int para_mkdir(const char *path, mode_t mode)
 322 {
 323         if (!mkdir(path, mode))
 324                 return 1;
 325         if (errno == EEXIST)
 326                 return -E_EXIST;
 327         if (errno == ENOSPC)
 328                 return -E_NOSPC;
 329         if (errno == ENOTDIR)
 330                 return -E_NOTDIR;
 331         if (errno == EPERM)
 332                 return E_OSL_PERM;
 333         return -E_MKDIR;
 334 }
 335
 336 static int verify_name(const char *name)
 337 {
 338         if (!name)
 339                 return -E_BAD_NAME;
 340         if (!*name)
 341                 return -E_BAD_NAME;
 342         if (strchr(name, '/'))
 343                 return -E_BAD_NAME;
 344         if (!strcmp(name, ".."))
 345                 return -E_BAD_NAME;
 346         if (!strcmp(name, "."))
 347                 return -E_BAD_NAME;
 348         return 1;
 349 }
 350
 351 /**
 352  * Compare two osl objects pointing to unsigned integers of 32 bit size.
 353  *
 354  * \param obj1 Pointer to the first integer.
 355  * \param obj2 Pointer to the second integer.
 356  *
 357  * \return The values required for an osl compare function.
 358  *
 359  * \sa osl_compare_func, osl_hash_compare().
 360  */
 361 int uint32_compare(const struct osl_object *obj1, const struct osl_object *obj2)
 362 {
 363         uint32_t d1 = read_u32((const char *)obj1->data);
 364         uint32_t d2 = read_u32((const char *)obj2->data);
 365
 366         if (d1 < d2)
 367                 return 1;
 368         if (d1 > d2)
 369                 return -1;
 370         return 0;
 371 }
 372
 373 /**
 374  * Compare two osl objects pointing to hash values.
 375  *
 376  * \param obj1 Pointer to the first hash object.
 377  * \param obj2 Pointer to the second hash object.
 378  *
 379  * \return The values required for an osl compare function.
 380  *
 381  * \sa osl_compare_func, uint32_compare().
 382  */
 383 int osl_hash_compare(const struct osl_object *obj1, const struct osl_object *obj2)
 384 {
 385         return hash_compare((HASH_TYPE *)obj1->data, (HASH_TYPE *)obj2->data);
 386 }
 387
 388 static char *disk_storage_dirname(const struct osl_table *t, unsigned col_num,
 389                 const char *ds_name)
 390 {
 391         char *dirname, *column_name = column_filename(t, col_num);
 392
 393         if (!(t->desc->flags & OSL_LARGE_TABLE))
 394                 return column_name;
 395         dirname = make_message("%s/%.2s", column_name, ds_name);
 396         free(column_name);
 397         return dirname;
 398 }
 399
 400 static char *disk_storage_name_of_object(const struct osl_table *t,
 401         const struct osl_object *obj)
 402 {
 403         HASH_TYPE hash[HASH_SIZE];
 404         hash_object(obj, hash);
 405         return disk_storage_name_of_hash(t, hash);
 406 }
 407
 408 static int disk_storage_name_of_row(const struct osl_table *t,
 409                 const struct osl_row *row, char **name)
 410 {
 411         struct osl_object obj;
 412         int ret = osl_get_object(t, row, t->disk_storage_name_column, &obj);
 413
 414         if (ret < 0)
 415                 return ret;
 416         *name = disk_storage_name_of_object(t, &obj);
 417         return 1;
 418 }
 419
 420 static void column_name_hash(const char *col_name, HASH_TYPE *hash)
 421 {
 422         return hash_function(col_name, strlen(col_name), hash);
 423 }
 424
 425 static int init_column_descriptions(struct osl_table *t)
 426 {
 427         int i, j, ret;
 428         const struct osl_column_description *cd;
 429
 430         ret = -E_BAD_TABLE_DESC;
 431         ret = verify_name(t->desc->name);
 432         if (ret < 0)
 433                 goto err;
 434         ret = -E_BAD_DB_DIR;
 435         if (!t->desc->dir)
 436                 goto err;
 437         /* the size of the index header without column descriptions */
 438         t->index_header_size = IDX_COLUMN_DESCRIPTIONS;
 439         FOR_EACH_COLUMN(i, t->desc, cd) {
 440                 struct osl_column *col = t->columns + i;
 441                 if (cd->storage_flags & OSL_RBTREE) {
 442                         if (!cd->compare_function)
 443                                 return -E_NO_COMPARE_FUNC;
 444                 }
 445                 if (cd->storage_type == OSL_NO_STORAGE)
 446                         continue;
 447                 ret = -E_NO_COLUMN_NAME;
 448                 if (!cd->name || !cd->name[0])
 449                         goto err;
 450                 ret = verify_name(cd->name);
 451                 if (ret < 0)
 452                         goto err;
 453                 t->index_header_size += index_column_description_size(cd->name);
 454                 column_name_hash(cd->name, col->name_hash);
 455                 ret = -E_DUPLICATE_COL_NAME;
 456                 for (j = i + 1; j < t->desc->num_columns; j++) {
 457                         const char *name2 = get_column_description(t->desc,
 458                                 j)->name;
 459                         if (cd->name && name2 && !strcmp(cd->name, name2))
 460                                 goto err;
 461                 }
 462         }
 463         return 1;
 464 err:
 465         return ret;
 466 }
 467
 468 /**
 469  * Initialize a struct table from given table description.
 470  *
 471  * \param desc The description of the osl table.
 472  * \param table_ptr Result is returned here.
 473  *
 474  * This function performs several sanity checks on \p desc and returns if any
 475  * of these tests fail. On success, a struct \p osl_table is allocated and
 476  * initialized with data derived from \p desc.
 477  *
 478  * \return Positive on success, negative on errors. Possible errors include: \p
 479  * E_BAD_TABLE_DESC, \p E_NO_COLUMN_DESC, \p E_NO_COLUMNS, \p
 480  * E_BAD_STORAGE_TYPE, \p E_BAD_STORAGE_FLAGS, \p E_BAD_STORAGE_SIZE, \p
 481  * E_NO_UNIQUE_RBTREE_COLUMN, \p E_NO_RBTREE_COL.
 482  *
 483  * \sa struct osl_table.
 484  */
 485 int init_table_structure(const struct osl_table_description *desc,
 486                 struct osl_table **table_ptr)
 487 {
 488         const struct osl_column_description *cd;
 489         struct osl_table *t = para_calloc(sizeof(*t));
 490         int i, ret = -E_BAD_TABLE_DESC, have_disk_storage_name_column = 0;
 491
 492         if (!desc)
 493                 goto err;
 494         PARA_DEBUG_LOG("creating table structure for '%s' from table "
 495                 "description\n", desc->name);
 496         ret = -E_NO_COLUMN_DESC;
 497         if (!desc->column_descriptions)
 498                 goto err;
 499         ret = -E_NO_COLUMNS;
 500         if (!desc->num_columns)
 501                 goto err;
 502         t->columns = para_calloc(desc->num_columns * sizeof(struct osl_column));
 503         t->desc = desc;
 504         FOR_EACH_COLUMN(i, t->desc, cd) {
 505                 enum osl_storage_type st = cd->storage_type;
 506                 enum osl_storage_flags sf = cd->storage_flags;
 507                 struct osl_column *col = &t->columns[i];
 508
 509                 ret = -E_BAD_STORAGE_TYPE;
 510                 if (st != OSL_MAPPED_STORAGE && st != OSL_DISK_STORAGE
 511                                 && st != OSL_NO_STORAGE)
 512                         goto err;
 513                 ret = -E_BAD_STORAGE_FLAGS;
 514                 if (st == OSL_DISK_STORAGE && sf & OSL_RBTREE)
 515                         goto err;
 516                 ret = -E_BAD_STORAGE_SIZE;
 517                 if (sf & OSL_FIXED_SIZE && !cd->data_size)
 518                         goto err;
 519                 switch (st) {
 520                 case OSL_DISK_STORAGE:
 521                         t->num_disk_storage_columns++;
 522                         break;
 523                 case OSL_MAPPED_STORAGE:
 524                         t->num_mapped_columns++;
 525                         col->index_offset = t->row_index_size;
 526                         t->row_index_size += 8;
 527                         break;
 528                 case OSL_NO_STORAGE:
 529                         col->volatile_num = t->num_volatile_columns;
 530                         t->num_volatile_columns++;
 531                         break;
 532                 }
 533                 if (sf & OSL_RBTREE) {
 534                         col->rbtree_num = t->num_rbtrees;
 535                         t->num_rbtrees++;
 536                         if ((sf & OSL_UNIQUE) && (st == OSL_MAPPED_STORAGE)) {
 537                                 if (!have_disk_storage_name_column)
 538                                         t->disk_storage_name_column = i;
 539                                 have_disk_storage_name_column = 1;
 540                         }
 541                 }
 542         }
 543         ret = -E_NO_UNIQUE_RBTREE_COLUMN;
 544         if (t->num_disk_storage_columns && !have_disk_storage_name_column)
 545                 goto err;
 546         ret = -E_NO_RBTREE_COL;
 547         if (!t->num_rbtrees)
 548                 goto err;
 549         /* success */
 550         PARA_DEBUG_LOG("OK. Index entry size: %u\n", t->row_index_size);
 551         ret = init_column_descriptions(t);
 552         if (ret < 0)
 553                 goto err;
 554         *table_ptr = t;
 555         return 1;
 556 err:
 557         free(t->columns);
 558         free(t);
 559         return ret;
 560 }
 561
 562 /**
 563  * Read the table description from index header.
 564  *
 565  * \param map The memory mapping of the index file.
 566  * \param desc The values found in the index header are returned here.
 567  *
 568  * Read the index header, check for the paraslash magic string and the table version number.
 569  * Read all information stored in the index header into \a desc.
 570  *
 571  * \return Positive on success, negative on errors.
 572  *
 573  * \sa struct osl_table_description, osl_create_table.
 574  */
 575 int read_table_desc(struct osl_object *map, struct osl_table_description *desc)
 576 {
 577         char *buf = map->data;
 578         uint8_t version;
 579         uint16_t header_size;
 580         int ret, i;
 581         unsigned offset;
 582         struct osl_column_description *cd;
 583
 584         if (map->size < MIN_INDEX_HEADER_SIZE(1))
 585                 return -E_SHORT_TABLE;
 586         if (strncmp(buf + IDX_PARA_MAGIC, PARA_MAGIC, strlen(PARA_MAGIC)))
 587                 return -E_NO_MAGIC;
 588         version = read_u8(buf + IDX_VERSION);
 589         if (version < MIN_TABLE_VERSION || version > MAX_TABLE_VERSION)
 590                 return -E_VERSION_MISMATCH;
 591         desc->num_columns = read_u8(buf + IDX_TABLE_FLAGS);
 592         desc->flags = read_u8(buf + IDX_TABLE_FLAGS);
 593         desc->num_columns = read_u16(buf + IDX_NUM_COLUMNS);
 594         PARA_DEBUG_LOG("%u columns\n", desc->num_columns);
 595         if (!desc->num_columns)
 596                 return -E_NO_COLUMNS;
 597         header_size = read_u16(buf + IDX_HEADER_SIZE);
 598         if (map->size < header_size)
 599                 return -E_BAD_SIZE;
 600         desc->column_descriptions = para_calloc(desc->num_columns
 601                 * sizeof(struct osl_column_description));
 602         offset = IDX_COLUMN_DESCRIPTIONS;
 603         FOR_EACH_COLUMN(i, desc, cd) {
 604                 char *null_byte;
 605
 606                 ret = -E_SHORT_TABLE;
 607                 if (map->size < offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE) {
 608                         PARA_ERROR_LOG("map size = %zu < %u = offset + min desc size\n",
 609                                 map->size, offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE);
 610                         goto err;
 611                 }
 612                 cd->storage_type = read_u16(buf + offset + IDX_CD_STORAGE_TYPE);
 613                 cd->storage_flags = read_u16(buf + offset +
 614                         IDX_CD_STORAGE_FLAGS);
 615                 cd->data_size = read_u32(buf + offset + IDX_CD_DATA_SIZE);
 616                 null_byte = memchr(buf + offset + IDX_CD_NAME, '\0',
 617                         map->size - offset - IDX_CD_NAME);
 618                 ret = -E_INDEX_CORRUPTION;
 619                 if (!null_byte)
 620                         goto err;
 621                 cd->name = para_strdup(buf + offset + IDX_CD_NAME);
 622                 offset += index_column_description_size(cd->name);
 623         }
 624         if (offset != header_size) {
 625                 ret = -E_INDEX_CORRUPTION;
 626                 PARA_ERROR_LOG("real header size = %u != %u = stored header size\n",
 627                         offset, header_size);
 628                 goto err;
 629         }
 630         return 1;
 631 err:
 632         FOR_EACH_COLUMN(i, desc, cd)
 633                 free(cd->name);
 634         return ret;
 635 }
 636
 637 /*
 638  * check whether the table description given by \p t->desc matches the on-disk
 639  * table structure stored in the index of \a t.
 640  */
 641 static int compare_table_descriptions(struct osl_table *t)
 642 {
 643         int i, ret;
 644         struct osl_table_description desc;
 645         const struct osl_column_description *cd1, *cd2;
 646
 647         /* read the on-disk structure into desc */
 648         ret = read_table_desc(&t->index_map, &desc);
 649         if (ret < 0)
 650                 return ret;
 651         ret = -E_BAD_TABLE_FLAGS;
 652         if (desc.flags != t->desc->flags)
 653                 goto out;
 654         ret = E_BAD_COLUMN_NUM;
 655         if (desc.num_columns != t->desc->num_columns)
 656                 goto out;
 657         FOR_EACH_COLUMN(i, t->desc, cd1) {
 658                 cd2 = get_column_description(&desc, i);
 659                 ret = -E_BAD_STORAGE_TYPE;
 660                 if (cd1->storage_type != cd2->storage_type)
 661                         goto out;
 662                 ret = -E_BAD_STORAGE_FLAGS;
 663                 if (cd1->storage_flags != cd2->storage_flags) {
 664                         PARA_ERROR_LOG("sf1 = %u != %u = sf2\n",
 665                                 cd1->storage_flags, cd2->storage_flags);
 666                         goto out;
 667                 }
 668                 ret = -E_BAD_DATA_SIZE;
 669                 if (cd1->storage_flags & OSL_FIXED_SIZE)
 670                         if (cd1->data_size != cd2->data_size)
 671                                 goto out;
 672                 ret = -E_BAD_COLUMN_NAME;
 673                 if (strcmp(cd1->name, cd2->name))
 674                         goto out;
 675         }
 676         PARA_DEBUG_LOG("table description of '%s' matches on-disk data, good\n",
 677                 t->desc->name);
 678         ret = 1;
 679 out:
 680         FOR_EACH_COLUMN(i, &desc, cd1)
 681                 free(cd1->name);
 682         free(desc.column_descriptions);
 683         return ret;
 684 }
 685
 686 static int create_table_index(struct osl_table *t)
 687 {
 688         char *buf, *filename;
 689         int i, ret;
 690         size_t size = t->index_header_size;
 691         const struct osl_column_description *cd;
 692         unsigned offset;
 693
 694         PARA_INFO_LOG("creating %zu byte index for table %s\n", size,
 695                 t->desc->name);
 696         buf = para_calloc(size);
 697         sprintf(buf + IDX_PARA_MAGIC, "%s", PARA_MAGIC);
 698         write_u8(buf + IDX_TABLE_FLAGS, t->desc->flags);
 699         write_u8(buf + IDX_DIRTY_FLAG, 0);
 700         write_u8(buf + IDX_VERSION, CURRENT_TABLE_VERSION);
 701         write_u16(buf + IDX_NUM_COLUMNS, t->desc->num_columns);
 702         write_u16(buf + IDX_HEADER_SIZE, t->index_header_size);
 703         offset = IDX_COLUMN_DESCRIPTIONS;
 704         FOR_EACH_COLUMN(i, t->desc, cd) {
 705                 write_u16(buf + offset + IDX_CD_STORAGE_TYPE,
 706                         cd->storage_type);
 707                 write_u16(buf + offset + IDX_CD_STORAGE_FLAGS,
 708                         cd->storage_flags);
 709                 if (cd->storage_flags & OSL_FIXED_SIZE)
 710                         write_u32(buf + offset + IDX_CD_DATA_SIZE,
 711                                 cd->data_size);
 712                 strcpy(buf + offset + IDX_CD_NAME, cd->name);
 713                 offset += index_column_description_size(cd->name);
 714         }
 715         assert(offset = size);
 716         filename = index_filename(t->desc);
 717         ret = para_write_file(filename, buf, size);
 718         free(buf);
 719         free(filename);
 720         return ret;
 721 }
 722
 723 /**
 724  * Create a new osl table.
 725  *
 726  * \param desc Pointer to the table description.
 727  *
 728  * \return Positive on success, negative on errors. Possible errors include: \p
 729  * E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_BAD_NAME, \p E_NO_COMPARE_FUNC, \p
 730  * E_NO_COLUMN_NAME, \p E_DUPLICATE_COL_NAME, \p E_MKDIR, any errors returned
 731  * by para_open().
 732  */
 733 int osl_create_table(const struct osl_table_description *desc)
 734 {
 735         const struct osl_column_description *cd;
 736         char *table_dir = NULL, *filename;
 737         struct osl_table *t;
 738         int i, ret = init_table_structure(desc, &t);
 739
 740         if (ret < 0)
 741                 return ret;
 742         PARA_INFO_LOG("creating %s\n", desc->name);
 743         FOR_EACH_COLUMN(i, t->desc, cd) {
 744                 if (cd->storage_type == OSL_NO_STORAGE)
 745                         continue;
 746                 if (!table_dir) {
 747                         ret = para_mkdir(desc->dir, 0777);
 748                         if (ret < 0 && ret != -E_EXIST)
 749                                 goto out;
 750                         table_dir = make_message("%s/%s", desc->dir,
 751                                 desc->name);
 752                         ret = para_mkdir(table_dir, 0777);
 753                         if (ret < 0)
 754                                 goto out;
 755                 }
 756                 filename = column_filename(t, i);
 757                 PARA_INFO_LOG("filename: %s\n", filename);
 758                 if (cd->storage_type == OSL_MAPPED_STORAGE) {
 759                         ret = para_open(filename, O_RDWR | O_CREAT | O_EXCL,
 760                                 0644);
 761                         free(filename);
 762                         if (ret < 0)
 763                                 goto out;
 764                         close(ret);
 765                         continue;
 766                 }
 767                 /* DISK STORAGE */
 768                 ret = para_mkdir(filename, 0777);
 769                 free(filename);
 770                 if (ret < 0)
 771                         goto out;
 772         }
 773         if (t->num_mapped_columns) {
 774                 ret = create_table_index(t);
 775                 if (ret < 0)
 776                         goto out;
 777         }
 778         ret = 1;
 779 out:
 780         free(table_dir);
 781         free(t->columns);
 782         free(t);
 783         return ret;
 784 }
 785
 786 static int table_is_dirty(struct osl_table *t)
 787 {
 788         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 789         uint8_t dirty = read_u8(buf) & 0x1;
 790         return !!dirty;
 791 }
 792
 793 static void mark_table_dirty(struct osl_table *t)
 794 {
 795         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 796         write_u8(buf, read_u8(buf) | 1);
 797 }
 798
 799 static void mark_table_clean(struct osl_table *t)
 800 {
 801         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 802         write_u8(buf, read_u8(buf) & 0xfe);
 803 }
 804
 805 static void unmap_column(struct osl_table *t, unsigned col_num)
 806 {
 807         struct osl_object map = t->columns[col_num].data_map;
 808         int ret;
 809         if (!map.data)
 810                 return;
 811         ret = para_munmap(map.data, map.size);
 812         assert(ret > 0);
 813         map.data = NULL;
 814 }
 815
 816 /**
 817  * Unmap all mapped files of an osl table.
 818  *
 819  * \param t Pointer to a mapped table.
 820  * \param flags Options for unmapping.
 821  *
 822  * \return Positive on success, negative on errors. Possible errors include:
 823  * E_NOT_MAPPED, E_MUNMAP.
 824  *
 825  * \sa map_table(), enum osl_close_flags, para_munmap().
 826  */
 827 int unmap_table(struct osl_table *t, enum osl_close_flags flags)
 828 {
 829         unsigned i;
 830         const struct osl_column_description *cd;
 831         int ret;
 832
 833         if (!t->num_mapped_columns) /* can this ever happen? */
 834                 return 1;
 835         PARA_DEBUG_LOG("unmapping table '%s'\n", t->desc->name);
 836         if (!t->index_map.data)
 837                 return -E_NOT_MAPPED;
 838         if (flags & OSL_MARK_CLEAN)
 839                 mark_table_clean(t);
 840         ret = para_munmap(t->index_map.data, t->index_map.size);
 841         if (ret < 0)
 842                 return ret;
 843         t->index_map.data = NULL;
 844         if (!t->num_rows)
 845                 return 1;
 846         FOR_EACH_MAPPED_COLUMN(i, t, cd)
 847                 unmap_column(t, i);
 848         return 1;
 849 }
 850
 851 static int map_column(struct osl_table *t, unsigned col_num)
 852 {
 853         struct stat statbuf;
 854         char *filename = column_filename(t, col_num);
 855         int ret = -E_STAT;
 856         if (stat(filename, &statbuf) < 0) {
 857                 free(filename);
 858                 return ret;
 859         }
 860         if (!(S_IFREG & statbuf.st_mode)) {
 861                 free(filename);
 862                 return ret;
 863         }
 864         ret = mmap_full_file(filename, O_RDWR,
 865                 &t->columns[col_num].data_map);
 866         free(filename);
 867         return ret;
 868 }
 869
 870 /**
 871  * Map the index file and all columns of type \p OSL_MAPPED_STORAGE into memory.
 872  *
 873  * \param t Pointer to an initialized table structure.
 874  * \param flags Mapping options.
 875  *
 876  * \return Negative return value on errors; on success the number of rows
 877  * (including invalid rows) is returned.
 878  *
 879  * \sa unmap_table(), enum map_table_flags, osl_open_table(), mmap(2).
 880  */
 881 int map_table(struct osl_table *t, enum map_table_flags flags)
 882 {
 883         char *filename;
 884         const struct osl_column_description *cd;
 885         int i = 0, ret, num_rows = 0;
 886
 887         if (!t->num_mapped_columns)
 888                 return 0;
 889         if (t->index_map.data)
 890                 return -E_ALREADY_MAPPED;
 891         filename = index_filename(t->desc);
 892         PARA_DEBUG_LOG("mapping table '%s' (index: %s)\n", t->desc->name, filename);
 893         ret = mmap_full_file(filename, flags & MAP_TBL_FL_MAP_RDONLY?
 894                 O_RDONLY : O_RDWR, &t->index_map);
 895         free(filename);
 896         if (ret < 0)
 897                 return ret;
 898         if (flags & MAP_TBL_FL_VERIFY_INDEX) {
 899                 ret = compare_table_descriptions(t);
 900                 if (ret < 0)
 901                         goto err;
 902         }
 903         ret = -E_BUSY;
 904         if (!(flags & MAP_TBL_FL_IGNORE_DIRTY)) {
 905                 if (table_is_dirty(t)) {
 906                         PARA_ERROR_LOG("%s is dirty\n", t->desc->name);
 907                         goto err;
 908                 }
 909         }
 910         mark_table_dirty(t);
 911         num_rows = table_num_rows(t);
 912         if (!num_rows)
 913                 return num_rows;
 914         /* map data files */
 915         FOR_EACH_MAPPED_COLUMN(i, t, cd) {
 916                 ret = map_column(t, i);
 917                 if (ret < 0)
 918                         goto err;
 919         }
 920         return num_rows;
 921 err:    /* unmap what is already mapped */
 922         for (i--; i >= 0; i--) {
 923                 struct osl_object map = t->columns[i].data_map;
 924                 para_munmap(map.data, map.size);
 925                 map.data = NULL;
 926         }
 927         para_munmap(t->index_map.data, t->index_map.size);
 928         t->index_map.data = NULL;
 929         return ret;
 930 }
 931
 932 /**
 933  * Retrieve a mapped object by row and column number.
 934  *
 935  * \param t Pointer to an open osl table.
 936  * \param col_num Number of the mapped column containing the object to retrieve.
 937  * \param row_num Number of the row containing the object to retrieve.
 938  * \param obj The result is returned here.
 939  *
 940  * It is considered an error if \a col_num does not refer to a column
 941  * of storage type \p OSL_MAPPED_STORAGE.
 942  *
 943  * \return Positive on success, negative on errors. Possible errors include:
 944  * \p E_BAD_ROW_NUM, \p E_INVALID_OBJECT.
 945  *
 946  * \sa osl_storage_type.
 947  */
 948 int get_mapped_object(const struct osl_table *t, unsigned col_num,
 949         uint32_t row_num, struct osl_object *obj)
 950 {
 951         struct osl_column *col = &t->columns[col_num];
 952         uint32_t offset;
 953         char *header;
 954         char *cell_index;
 955         int ret;
 956
 957         if (t->num_rows <= row_num)
 958                 return -E_BAD_ROW_NUM;
 959         ret = get_cell_index(t, row_num, col_num, &cell_index);
 960         if (ret < 0)
 961                 return ret;
 962         offset = read_u32(cell_index);
 963         obj->size = read_u32(cell_index + 4) - 1;
 964         header = col->data_map.data + offset;
 965         obj->data = header + 1;
 966         if (read_u8(header) == 0xff) {
 967                 PARA_ERROR_LOG("col %u, size %zu, offset %u\n", col_num,
 968                         obj->size, offset);
 969                 return -E_INVALID_OBJECT;
 970         }
 971         return 1;
 972 }
 973
 974 static int search_rbtree(const struct osl_object *obj,
 975                 const struct osl_table *t, unsigned col_num,
 976                 struct rb_node **result, struct rb_node ***rb_link)
 977 {
 978         struct osl_column *col = &t->columns[col_num];
 979         struct rb_node **new = &col->rbtree.rb_node, *parent = NULL;
 980         const struct osl_column_description *cd =
 981                 get_column_description(t->desc, col_num);
 982         enum osl_storage_type st = cd->storage_type;
 983         while (*new) {
 984                 struct osl_row *this_row = get_row_pointer(*new,
 985                         col->rbtree_num);
 986                 int ret;
 987                 struct osl_object this_obj;
 988                 parent = *new;
 989                 if (st == OSL_MAPPED_STORAGE) {
 990                         ret = get_mapped_object(t, col_num, this_row->num,
 991                                 &this_obj);
 992                         if (ret < 0)
 993                                 return ret;
 994                 } else
 995                         this_obj = this_row->volatile_objects[col->volatile_num];
 996                 ret = cd->compare_function(obj, &this_obj);
 997                 if (!ret) {
 998                         if (result)
 999                                 *result = get_rb_node_pointer(this_row,
1000                                         col->rbtree_num);
1001                         return 1;
1002                 }
1003                 if (ret < 0)
1004                         new = &((*new)->rb_left);
1005                 else
1006                         new = &((*new)->rb_right);
1007         }
1008         if (result)
1009                 *result = parent;
1010         if (rb_link)
1011                 *rb_link = new;
1012         return -E_RB_KEY_NOT_FOUND;
1013 }
1014
1015 static int insert_rbtree(struct osl_table *t, unsigned col_num,
1016         const struct osl_row *row, const struct osl_object *obj)
1017 {
1018         struct rb_node *parent, **rb_link;
1019         unsigned rbtree_num;
1020         struct rb_node *n;
1021         int ret = search_rbtree(obj, t, col_num, &parent, &rb_link);
1022
1023         if (ret > 0)
1024                 return -E_RB_KEY_EXISTS;
1025         rbtree_num = t->columns[col_num].rbtree_num;
1026         n = get_rb_node_pointer(row, rbtree_num);
1027         rb_link_node(n, parent, rb_link);
1028         rb_insert_color(n, &t->columns[col_num].rbtree);
1029         return 1;
1030 }
1031
1032 static void remove_rb_node(struct osl_table *t, unsigned col_num,
1033                 const struct osl_row *row)
1034 {
1035         struct osl_column *col = &t->columns[col_num];
1036         const struct osl_column_description *cd =
1037                 get_column_description(t->desc, col_num);
1038         enum osl_storage_flags sf = cd->storage_flags;
1039         struct rb_node *victim, *splice_out_node, *tmp;
1040         if (!(sf & OSL_RBTREE))
1041                 return;
1042         /*
1043          * Which node is removed/spliced out actually depends on how many
1044          * children the victim node has: If it has no children, it gets
1045          * deleted. If it has one child, it gets spliced out. If it has two
1046          * children, its successor (which has at most a right child) gets
1047          * spliced out.
1048          */
1049         victim = get_rb_node_pointer(row, col->rbtree_num);
1050         if (victim->rb_left && victim->rb_right)
1051                 splice_out_node = rb_next(victim);
1052         else
1053                 splice_out_node = victim;
1054         /* Go up to the root and decrement the size of each node in the path. */
1055         for (tmp = splice_out_node; tmp; tmp = rb_parent(tmp))
1056                 tmp->size--;
1057         rb_erase(victim, &col->rbtree);
1058 }
1059
1060 static int add_row_to_rbtrees(struct osl_table *t, uint32_t row_num,
1061                 struct osl_object *volatile_objs, struct osl_row **row_ptr)
1062 {
1063         unsigned i;
1064         int ret;
1065         struct osl_row *row = allocate_row(t->num_rbtrees);
1066         const struct osl_column_description *cd;
1067
1068         row->num = row_num;
1069         row->volatile_objects = volatile_objs;
1070         FOR_EACH_RBTREE_COLUMN(i, t, cd) {
1071                 if (cd->storage_type == OSL_MAPPED_STORAGE) {
1072                         struct osl_object obj;
1073                         ret = get_mapped_object(t, i, row_num, &obj);
1074                         if (ret < 0)
1075                                 goto err;
1076                         ret = insert_rbtree(t, i, row, &obj);
1077                 } else { /* volatile */
1078                         const struct osl_object *obj
1079                                 = volatile_objs + t->columns[i].volatile_num;
1080                         ret = insert_rbtree(t, i, row, obj);
1081                 }
1082                 if (ret < 0)
1083                         goto err;
1084         }
1085         if (row_ptr)
1086                 *row_ptr = row;
1087         return 1;
1088 err: /* rollback changes, i.e. remove added entries from rbtrees */
1089         while (i)
1090                 remove_rb_node(t, i--, row);
1091         free(row);
1092         return ret;
1093 }
1094
1095 static void free_volatile_objects(const struct osl_table *t,
1096                 enum osl_close_flags flags)
1097 {
1098         int i, j;
1099         struct rb_node *n;
1100         struct osl_column *rb_col;
1101         const struct osl_column_description *cd;
1102
1103         if (!t->num_volatile_columns)
1104                 return;
1105         /* find the first rbtree column (any will do) */
1106         FOR_EACH_RBTREE_COLUMN(i, t, cd)
1107                 break;
1108         rb_col = t->columns + i;
1109         /* walk that rbtree and free all volatile objects */
1110         for (n = rb_first(&rb_col->rbtree); n; n = rb_next(n)) {
1111                 struct osl_row *r = get_row_pointer(n, rb_col->rbtree_num);
1112                 if (flags & OSL_FREE_VOLATILE)
1113                         for (j = 0; j < t->num_volatile_columns; j++)
1114                                 free(r->volatile_objects[j].data);
1115                 free(r->volatile_objects);
1116         }
1117 }
1118
1119 /**
1120  * Erase all rbtree nodes and free resources.
1121  *
1122  * \param t Pointer to an open osl table.
1123  *
1124  * This function is called by osl_close_table().
1125  */
1126 void clear_rbtrees(struct osl_table *t)
1127 {
1128         const struct osl_column_description *cd;
1129         unsigned i, rbtrees_cleared = 0;
1130
1131         FOR_EACH_RBTREE_COLUMN(i, t, cd) {
1132                 struct osl_column *col = &t->columns[i];
1133                 struct rb_node *n;
1134                 rbtrees_cleared++;
1135                 for (n = rb_first(&col->rbtree); n;) {
1136                         struct osl_row *r;
1137                         rb_erase(n, &col->rbtree);
1138                         if (rbtrees_cleared == t->num_rbtrees) {
1139                                 r = get_row_pointer(n, col->rbtree_num);
1140                                 n = rb_next(n);
1141                                 free(r);
1142                         } else
1143                                 n = rb_next(n);
1144                 }
1145         }
1146
1147 }
1148
1149 /**
1150  * Close an osl table.
1151  *
1152  * \param t Pointer to the table to be closed.
1153  * \param flags Options for what should be cleaned up.
1154  *
1155  * If osl_open_table() succeeds, the resulting table pointer must later be
1156  * passed to this function in order to flush all changes to the filesystem and
1157  * to free the resources that were allocated by osl_open_table().
1158  *
1159  * \return Positive on success, negative on errors. Possible errors: \p E_BAD_TABLE,
1160  * errors returned by unmap_table().
1161  *
1162  * \sa osl_open_table(), unmap_table().
1163  */
1164 int osl_close_table(struct osl_table *t, enum osl_close_flags flags)
1165 {
1166         int ret;
1167
1168         if (!t)
1169                 return -E_BAD_TABLE;
1170         free_volatile_objects(t, flags);
1171         clear_rbtrees(t);
1172         ret = unmap_table(t, flags);
1173         if (ret < 0)
1174                 PARA_ERROR_LOG("unmap_table failed: %d\n", ret);
1175         free(t->columns);
1176         free(t);
1177         return ret;
1178 }
1179
1180 /**
1181  * Find out whether the given row number corresponds to an invalid row.
1182  *
1183  * \param t Pointer to the osl table.
1184  * \param row_num The number of the row in question.
1185  *
1186  * By definition, a row is considered invalid if all its index entries
1187  * are invalid.
1188  *
1189  * \return Positive if \a row_num corresponds to an invalid row,
1190  * zero if it corresponds to a valid row, negative on errors.
1191  */
1192 int row_is_invalid(struct osl_table *t, uint32_t row_num)
1193 {
1194         char *row_index;
1195         int i, ret = get_row_index(t, row_num, &row_index);
1196
1197         if (ret < 0)
1198                 return ret;
1199         for (i = 0; i < t->row_index_size; i++) {
1200                 if ((unsigned char)row_index[i] != 0xff)
1201                         return 0;
1202         }
1203         PARA_INFO_LOG("row %d is invalid\n", row_num);
1204         return 1;
1205 }
1206
1207 /**
1208  * Invalidate a row of an osl table.
1209  *
1210  * \param t Pointer to an open osl table.
1211  * \param row_num Number of the row to mark as invalid.
1212  *
1213  * This function marks each mapped object in the index entry of \a row as
1214  * invalid.
1215  *
1216  * \return Positive on success, negative on errors.
1217  */
1218 int mark_row_invalid(struct osl_table *t, uint32_t row_num)
1219 {
1220         char *row_index;
1221         int ret = get_row_index(t, row_num, &row_index);
1222
1223         if (ret < 0)
1224                 return ret;
1225         PARA_INFO_LOG("marking row %d as invalid\n", row_num);
1226         memset(row_index, 0xff, t->row_index_size);
1227         return 1;
1228 }
1229
1230 /**
1231  * Initialize all rbtrees and compute number of invalid rows.
1232  *
1233  * \param t The table containing the rbtrees to be initialized.
1234  *
1235  * \return Positive on success, negative on errors.
1236  */
1237 int init_rbtrees(struct osl_table *t)
1238 {
1239         int i, ret;
1240         const struct osl_column_description *cd;
1241
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);
1249                 if (ret < 0)
1250                         return ret;
1251                 if (ret) {
1252                         t->num_invalid_rows++;
1253                         continue;
1254                 }
1255                 ret = add_row_to_rbtrees(t, i, NULL, NULL);
1256                 if (ret < 0)
1257                         return ret;
1258         }
1259         return 1;
1260 }
1261
1262 /**
1263  * Open an osl table.
1264  *
1265  * Each osl table must be opened before its data can be accessed.
1266  *
1267  * \param table_desc Describes the table to be opened.
1268  * \param result Contains a pointer to the open table on success.
1269  *
1270  * The table description given by \a desc should coincide with the
1271  * description used at creation time.
1272  *
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().
1277  */
1278 int osl_open_table(const struct osl_table_description *table_desc,
1279                 struct osl_table **result)
1280 {
1281         int i, ret;
1282         struct osl_table *t;
1283         const struct osl_column_description *cd;
1284
1285         PARA_INFO_LOG("opening table %s\n", table_desc->name);
1286         ret = init_table_structure(table_desc, &t);
1287         if (ret < 0)
1288                 return ret;
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);
1294                 free(dirname);
1295                 if (ret < 0) {
1296                         if (errno == ENOENT)
1297                                 ret = -E_NOENT;
1298                         else
1299                                 ret = -E_STAT;
1300                         goto err;
1301                 }
1302                 ret = -E_NOTDIR;
1303                 if (!S_ISDIR(statbuf.st_mode))
1304                         goto err;
1305         }
1306         ret = map_table(t, MAP_TBL_FL_VERIFY_INDEX);
1307         if (ret < 0)
1308                 goto err;
1309         t->num_rows = ret;
1310         PARA_DEBUG_LOG("num rows: %d\n", t->num_rows);
1311         ret = init_rbtrees(t);
1312         if (ret < 0) {
1313                 osl_close_table(t, OSL_MARK_CLEAN); /* ignore further errors */
1314                 return ret;
1315         }
1316         *result = t;
1317         return 1;
1318 err:
1319         free(t->columns);
1320         free(t);
1321         return ret;
1322 }
1323
1324 static int create_disk_storage_object_dir(const struct osl_table *t,
1325                 unsigned col_num, const char *ds_name)
1326 {
1327         char *dirname;
1328         int ret;
1329
1330         if (!(t->desc->flags & OSL_LARGE_TABLE))
1331                 return 1;
1332         dirname = disk_storage_dirname(t, col_num, ds_name);
1333         ret = para_mkdir(dirname, 0777);
1334         free(dirname);
1335         if (ret < 0 && ret != -E_EXIST)
1336                 return ret;
1337         return 1;
1338 }
1339
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)
1342 {
1343         int ret;
1344         char *filename;
1345
1346         ret = create_disk_storage_object_dir(t, col_num, ds_name);
1347         if (ret < 0)
1348                 return ret;
1349         filename = disk_storage_path(t, col_num, ds_name);
1350         ret = para_write_file(filename, obj->data, obj->size);
1351         free(filename);
1352         return ret;
1353 }
1354
1355 static int append_map_file(const struct osl_table *t, unsigned col_num,
1356         const struct osl_object *obj, uint32_t *new_size)
1357 {
1358         char *filename = column_filename(t, col_num);
1359         int ret;
1360         char header = 0; /* zero means valid object */
1361
1362 //      PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj->size);
1363         ret = append_file(filename, &header, 1, obj->data, obj->size,
1364                 new_size);
1365         free(filename);
1366         return ret;
1367 }
1368
1369 static int append_row_index(const struct osl_table *t, char *row_index)
1370 {
1371         char *filename;
1372         int ret;
1373
1374         if (!t->num_mapped_columns)
1375                 return 1;
1376         filename = index_filename(t->desc);
1377         ret = append_file(filename, NULL, 0, row_index,
1378                 t->row_index_size, NULL);
1379         free(filename);
1380         return ret;
1381 }
1382
1383 /**
1384  * A wrapper for truncate(2)
1385  *
1386  * \param path Name of the regular file to truncate
1387  * \param size Number of bytes to \b shave \b off
1388  *
1389  * Truncate the regular file named by \a path by \a size bytes.
1390  *
1391  * \return Positive on success, negative on errors. Possible errors include: \p
1392  * E_STAT, \p E_BAD_SIZE, \p E_TRUNC.
1393  *
1394  * \sa truncate(2)
1395  */
1396 int para_truncate(const char *path, off_t size)
1397 {
1398         int ret;
1399         struct stat statbuf;
1400
1401         ret = -E_STAT;
1402         if (stat(path, &statbuf) < 0)
1403                 goto out;
1404         ret = -E_BAD_SIZE;
1405         if (statbuf.st_size < size)
1406                 goto out;
1407         ret = -E_TRUNC;
1408         if (truncate(path, statbuf.st_size - size) < 0)
1409                 goto out;
1410         ret = 1;
1411 out:
1412         return ret;
1413 }
1414
1415 static int truncate_mapped_file(const struct osl_table *t, unsigned col_num,
1416                 off_t size)
1417 {
1418         char *filename = column_filename(t, col_num);
1419         int ret = para_truncate(filename, size);
1420         free(filename);
1421         return ret;
1422 }
1423
1424 static int delete_disk_storage_file(const struct osl_table *t, unsigned col_num,
1425                 const char *ds_name)
1426 {
1427         char *dirname, *filename = disk_storage_path(t, col_num, ds_name);
1428         int ret = unlink(filename);
1429
1430         PARA_DEBUG_LOG("deleted %s\n", filename);
1431         free(filename);
1432         if (ret < 0) {
1433                 if (errno == ENOENT)
1434                         return -E_NOENT;
1435                 return -E_UNLINK;
1436         }
1437         if (!(t->desc->flags & OSL_LARGE_TABLE))
1438                 return 1;
1439         dirname = disk_storage_dirname(t, col_num, ds_name);
1440         rmdir(dirname);
1441         free(dirname);
1442         return 1;
1443 }
1444
1445 /**
1446  * Add a new row to an osl table and retrieve this row.
1447  *
1448  * \param t Pointer to an open osl table.
1449  * \param objects Array of objects to be added.
1450  * \param row Result pointer.
1451  *
1452  * The \a objects parameter must point to an array containing one object per
1453  * column.  The order of the objects in the array is given by the table
1454  * description of \a table. Several sanity checks are performed during object
1455  * insertion and the function returns without modifying the table if any of
1456  * these tests fail.  In fact, it is atomic in the sense that it either
1457  * succeeds or leaves the table unchanged (i.e. either all or none of the
1458  * objects are added to the table).
1459  *
1460  * It is considered an error if an object is added to a column with associated
1461  * rbtree if this object is equal to an object already contained in that column
1462  * (i.e. the compare function for the column's rbtree returns zero).
1463  *
1464  * Possible errors include: \p E_RB_KEY_EXISTS, \p E_BAD_DATA_SIZE.
1465  *
1466  * \return Positive on success, negative on errors.
1467  *
1468  * \sa struct osl_table_description, osl_compare_func, osl_add_row().
1469  */
1470 int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
1471                 struct osl_row **row)
1472 {
1473         int i, ret;
1474         char *ds_name = NULL;
1475         struct rb_node **rb_parents = NULL, ***rb_links = NULL;
1476         char *new_row_index = NULL;
1477         struct osl_object *volatile_objs = NULL;
1478         const struct osl_column_description *cd;
1479
1480         if (!t)
1481                 return -E_BAD_TABLE;
1482         rb_parents = para_malloc(t->num_rbtrees * sizeof(struct rn_node*));
1483         rb_links = para_malloc(t->num_rbtrees * sizeof(struct rn_node**));
1484         if (t->num_mapped_columns)
1485                 new_row_index = para_malloc(t->row_index_size);
1486         /* pass 1: sanity checks */
1487 //      PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
1488 //              objects[1].data);
1489         FOR_EACH_COLUMN(i, t->desc, cd) {
1490                 enum osl_storage_type st = cd->storage_type;
1491                 enum osl_storage_flags sf = cd->storage_flags;
1492
1493 //              ret = -E_NULL_OBJECT;
1494 //              if (!objects[i])
1495 //                      goto out;
1496                 if (st == OSL_DISK_STORAGE)
1497                         continue;
1498                 if (sf & OSL_RBTREE) {
1499                         unsigned rbtree_num = t->columns[i].rbtree_num;
1500                         ret = -E_RB_KEY_EXISTS;
1501 //                      PARA_DEBUG_LOG("checking whether %p exists\n",
1502 //                              objects[i].data);
1503                         if (search_rbtree(objects + i, t, i,
1504                                         &rb_parents[rbtree_num],
1505                                         &rb_links[rbtree_num]) > 0)
1506                                 goto out;
1507                 }
1508                 if (sf & OSL_FIXED_SIZE) {
1509 //                      PARA_DEBUG_LOG("fixed size. need: %zu, have: %d\n",
1510 //                              objects[i].size, cd->data_size);
1511                         ret = -E_BAD_DATA_SIZE;
1512                         if (objects[i].size != cd->data_size)
1513                                 goto out;
1514                 }
1515         }
1516         if (t->num_disk_storage_columns)
1517                 ds_name = disk_storage_name_of_object(t,
1518                         &objects[t->disk_storage_name_column]);
1519         ret = unmap_table(t, OSL_MARK_CLEAN);
1520         if (ret < 0)
1521                 goto out;
1522 //      PARA_DEBUG_LOG("sanity tests passed%s\n", "");
1523         /* pass 2: create data files, append map data */
1524         FOR_EACH_COLUMN(i, t->desc, cd) {
1525                 enum osl_storage_type st = cd->storage_type;
1526                 if (st == OSL_NO_STORAGE)
1527                         continue;
1528                 if (st == OSL_MAPPED_STORAGE) {
1529                         uint32_t new_size;
1530                         struct osl_column *col = &t->columns[i];
1531 //                      PARA_DEBUG_LOG("appending object of size %zu\n",
1532 //                              objects[i].size);
1533                         ret = append_map_file(t, i, objects + i, &new_size);
1534                         if (ret < 0)
1535                                 goto rollback;
1536                         update_cell_index(new_row_index, col, new_size,
1537                                 objects[i].size);
1538                         continue;
1539                 }
1540                 /* DISK_STORAGE */
1541                 ret = write_disk_storage_file(t, i, objects + i, ds_name);
1542                 if (ret < 0)
1543                         goto rollback;
1544         }
1545         ret = append_row_index(t, new_row_index);
1546         if (ret < 0)
1547                 goto rollback;
1548         ret = map_table(t, MAP_TBL_FL_VERIFY_INDEX);
1549         if (ret < 0) { /* truncate index and rollback changes */
1550                 char *filename = index_filename(t->desc);
1551                 para_truncate(filename, t->row_index_size);
1552                 free(filename);
1553                 goto rollback;
1554         }
1555         /* pass 3: add entry to rbtrees */
1556         if (t->num_volatile_columns) {
1557                 volatile_objs = para_calloc(t->num_volatile_columns
1558                         * sizeof(struct osl_object));
1559                 FOR_EACH_VOLATILE_COLUMN(i, t, cd)
1560                         volatile_objs[t->columns[i].volatile_num] = objects[i];
1561         }
1562         t->num_rows++;
1563 //      PARA_DEBUG_LOG("adding new entry as row #%d\n", t->num_rows - 1);
1564         ret = add_row_to_rbtrees(t, t->num_rows - 1, volatile_objs, row);
1565         if (ret < 0)
1566                 goto out;
1567 //      PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
1568         ret = 1;
1569         goto out;
1570 rollback: /* rollback all changes made, ignore further errors */
1571         for (i--; i >= 0; i--) {
1572                 cd = get_column_description(t->desc, i);
1573                 enum osl_storage_type st = cd->storage_type;
1574                 if (st == OSL_NO_STORAGE)
1575                         continue;
1576
1577                 if (st == OSL_MAPPED_STORAGE)
1578                         truncate_mapped_file(t, i, objects[i].size);
1579                 else /* disk storage */
1580                         delete_disk_storage_file(t, i, ds_name);
1581         }
1582         /* ignore error and return previous error value */
1583         map_table(t, MAP_TBL_FL_VERIFY_INDEX);
1584 out:
1585         free(new_row_index);
1586         free(ds_name);
1587         free(rb_parents);
1588         free(rb_links);
1589         return ret;
1590 }
1591
1592 /**
1593  * Add a new row to an osl table.
1594  *
1595  * \param t Same meaning as osl_add_and_get_row().
1596  * \param objects Same meaning as osl_add_and_get_row().
1597  *
1598  * \return The return value of the underlying call to osl_add_and_get_row().
1599  *
1600  * This is equivalent to osl_add_and_get_row(t, objects, NULL).
1601  */
1602 int osl_add_row(struct osl_table *t, struct osl_object *objects)
1603 {
1604         return osl_add_and_get_row(t, objects, NULL);
1605 }
1606
1607 /**
1608  * Retrieve an object identified by row and column
1609  *
1610  * \param t Pointer to an open osl table.
1611  * \param r Pointer to the row.
1612  * \param col_num The column number.
1613  * \param object The result pointer.
1614  *
1615  * The column determined by \a col_num must be of type \p OSL_MAPPED_STORAGE
1616  * or \p OSL_NO_STORAGE, i.e. no disk storage objects may be retrieved by this
1617  * function.
1618  *
1619  * \return Positive if object was found, negative on errors. Possible errors
1620  * include: \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE.
1621  *
1622  * \sa osl_storage_type, osl_open_disk_object().
1623  */
1624 int osl_get_object(const struct osl_table *t, const struct osl_row *r,
1625         unsigned col_num, struct osl_object *object)
1626 {
1627         const struct osl_column_description *cd;
1628
1629         if (!t)
1630                 return -E_BAD_TABLE;
1631         cd = get_column_description(t->desc, col_num);
1632         /* col must not be disk storage */
1633         if (cd->storage_type == OSL_DISK_STORAGE)
1634                 return -E_BAD_STORAGE_TYPE;
1635         if (cd->storage_type == OSL_MAPPED_STORAGE)
1636                 return get_mapped_object(t, col_num, r->num, object);
1637         /* volatile */
1638         *object = r->volatile_objects[t->columns[col_num].volatile_num];
1639         return 1;
1640 }
1641
1642 static int mark_mapped_object_invalid(const struct osl_table *t,
1643                 uint32_t row_num, unsigned col_num)
1644 {
1645         struct osl_object obj;
1646         char *p;
1647         int ret = get_mapped_object(t, col_num, row_num, &obj);
1648
1649         if (ret < 0)
1650                 return ret;
1651         p = obj.data;
1652         p--;
1653         *p = 0xff;
1654         return 1;
1655 }
1656
1657 /**
1658  * Delete a row from an osl table.
1659  *
1660  * \param t Pointer to an open osl table.
1661  * \param row Pointer to the row to delete.
1662  *
1663  * This removes all disk storage objects, removes all rbtree nodes,  and frees
1664  * all volatile objects belonging to the given row. For mapped columns, the
1665  * data is merely marked invalid and may be pruned from time to time by
1666  * para_fsck.
1667  *
1668  * \return Positive on success, negative on errors. Possible errors include:
1669  * \p E_BAD_TABLE, errors returned by osl_get_object().
1670  */
1671 int osl_del_row(struct osl_table *t, struct osl_row *row)
1672 {
1673         struct osl_row *r = row;
1674         int i, ret;
1675         const struct osl_column_description *cd;
1676
1677         if (!t)
1678                 return -E_BAD_TABLE;
1679         PARA_INFO_LOG("deleting row %p\n", row);
1680
1681         if (t->num_disk_storage_columns) {
1682                 char *ds_name;
1683                 ret = disk_storage_name_of_row(t, r, &ds_name);
1684                 if (ret < 0)
1685                         goto out;
1686                 FOR_EACH_DISK_STORAGE_COLUMN(i, t, cd)
1687                         delete_disk_storage_file(t, i, ds_name);
1688                 free(ds_name);
1689         }
1690         FOR_EACH_COLUMN(i, t->desc, cd) {
1691                 struct osl_column *col = t->columns + i;
1692                 enum osl_storage_type st = cd->storage_type;
1693                 remove_rb_node(t, i, r);
1694                 if (st == OSL_MAPPED_STORAGE) {
1695                         mark_mapped_object_invalid(t, r->num, i);
1696                         continue;
1697                 }
1698                 if (st == OSL_NO_STORAGE)
1699                         free(r->volatile_objects[col->volatile_num].data);
1700         }
1701         if (t->num_mapped_columns) {
1702                 ret = mark_row_invalid(t, r->num);
1703                 if (ret < 0)
1704                         goto out;
1705                 t->num_invalid_rows++;
1706         } else
1707                 t->num_rows--;
1708         ret = 1;
1709 out:
1710         free(r->volatile_objects);
1711         free(r);
1712         return ret;
1713 }
1714
1715 /* test if column has an rbtree */
1716 static int check_rbtree_col(const struct osl_table *t, unsigned col_num,
1717                 struct osl_column **col)
1718 {
1719         if (!t)
1720                 return -E_BAD_TABLE;
1721         if (!(get_column_description(t->desc, col_num)->storage_flags & OSL_RBTREE))
1722                 return -E_BAD_STORAGE_FLAGS;
1723         *col = t->columns + col_num;
1724         return 1;
1725 }
1726
1727 /**
1728  * Get the row that contains the given object.
1729  *
1730  * \param t Pointer to an open osl table.
1731  * \param col_num The number of the column to be searched.
1732  * \param obj The object to be looked up.
1733  * \param result Points to the row containing \a obj.
1734  *
1735  * Lookup \a obj in \a t and return the row containing \a obj. The column
1736  * specified by \a col_num must have an associated rbtree.
1737  *
1738  * \return Positive on success, negative on errors. If an error occured, \a
1739  * result is set to \p NULL. Possible errors include: \p E_BAD_TABLE, \p
1740  * E_BAD_STORAGE_FLAGS, errors returned by get_mapped_object(), \p
1741  * E_RB_KEY_NOT_FOUND.
1742  *
1743  * \sa osl_storage_flags
1744  */
1745 int osl_get_row(const struct osl_table *t, unsigned col_num,
1746                 const struct osl_object *obj, struct osl_row **result)
1747 {
1748         int ret;
1749         struct rb_node *node;
1750         struct osl_row *row;
1751         struct osl_column *col;
1752
1753         *result = NULL;
1754         ret = check_rbtree_col(t, col_num, &col);
1755         if (ret < 0)
1756                 return ret;
1757         ret = search_rbtree(obj, t, col_num, &node, NULL);
1758         if (ret < 0)
1759                 return ret;
1760         row = get_row_pointer(node, t->columns[col_num].rbtree_num);
1761         *result = row;
1762         return 1;
1763 }
1764
1765 static int rbtree_loop(struct osl_column *col,  void *private_data,
1766                 osl_rbtree_loop_func *func)
1767 {
1768         struct rb_node *n;
1769
1770         for (n = rb_first(&col->rbtree); n; n = rb_next(n)) {
1771                 struct osl_row *r = get_row_pointer(n, col->rbtree_num);
1772                 int ret = func(r, private_data);
1773                 if (ret < 0)
1774                         return ret;
1775         }
1776         return 1;
1777 }
1778
1779 static int rbtree_loop_reverse(struct osl_column *col,  void *private_data,
1780                 osl_rbtree_loop_func *func)
1781 {
1782         struct rb_node *n;
1783
1784         for (n = rb_last(&col->rbtree); n; n = rb_prev(n)) {
1785                 struct osl_row *r = get_row_pointer(n, col->rbtree_num);
1786                 int ret = func(r, private_data);
1787                 if (ret < 0)
1788                         return ret;
1789         }
1790         return 1;
1791 }
1792
1793 /**
1794  * Loop over all nodes in an rbtree.
1795  *
1796  * \param t Pointer to an open osl table.
1797  * \param col_num The column to use for iterating over the elements.
1798  * \param private_data Pointer that gets passed to \a func.
1799  * \param func The function to be called for each node in the rbtree.
1800  *
1801  * This function does an in-order walk of the rbtree associated with \a
1802  * col_num. It is an error if the \p OSL_RBTREE flag is not set for this
1803  * column. For each node in the rbtree, the given function \a func is called
1804  * with two \p void* pointers as arguments: The first argument points to the
1805  * row that contains the object corresponding to the rbtree node currently
1806  * traversed, and the \a private_data pointer is passed to \a func as the
1807  * second argument. The loop terminates either if \a func returns a negative
1808  * value, or if all nodes of the tree have been visited.
1809  *
1810  *
1811  * \return Positive on success, negative on errors. If the termination of the
1812  * loop was caused by \a func returning a negative value, this value is
1813  * returned.
1814  *
1815  * \sa osl_storage_flags, osl_rbtree_loop_reverse(), osl_compare_func.
1816  */
1817 int osl_rbtree_loop(const struct osl_table *t, unsigned col_num,
1818         void *private_data, osl_rbtree_loop_func *func)
1819 {
1820         struct osl_column *col;
1821
1822         int ret = check_rbtree_col(t, col_num, &col);
1823         if (ret < 0)
1824                 return ret;
1825         return rbtree_loop(col, private_data, func);
1826 }
1827
1828 /**
1829  * Loop over all nodes in an rbtree in reverse order.
1830  *
1831  * \param t Identical meaning as in \p osl_rbtree_loop().
1832  * \param col_num Identical meaning as in \p osl_rbtree_loop().
1833  * \param private_data Identical meaning as in \p osl_rbtree_loop().
1834  * \param func Identical meaning as in \p osl_rbtree_loop().
1835  *
1836  * This function is identical to \p osl_rbtree_loop(), the only difference
1837  * is that the tree is walked in reverse order.
1838  *
1839  * \return The same return value as \p osl_rbtree_loop().
1840  *
1841  * \sa osl_rbtree_loop().
1842  */
1843 int osl_rbtree_loop_reverse(const struct osl_table *t, unsigned col_num,
1844         void *private_data, osl_rbtree_loop_func *func)
1845 {
1846         struct osl_column *col;
1847
1848         int ret = check_rbtree_col(t, col_num, &col);
1849         if (ret < 0)
1850                 return ret;
1851         return rbtree_loop_reverse(col, private_data, func);
1852 }
1853
1854 /* TODO: Rollback changes on errors */
1855 static int rename_disk_storage_objects(struct osl_table *t,
1856                 struct osl_object *old_obj, struct osl_object *new_obj)
1857 {
1858         int i, ret;
1859         const struct osl_column_description *cd;
1860         char *old_ds_name, *new_ds_name;
1861
1862         if (!t->num_disk_storage_columns)
1863                 return 1; /* nothing to do */
1864         if (old_obj->size == new_obj->size && !memcmp(new_obj->data,
1865                         old_obj->data, new_obj->size))
1866                 return 1; /* object did not change */
1867         old_ds_name = disk_storage_name_of_object(t, old_obj);
1868         new_ds_name = disk_storage_name_of_object(t, new_obj);
1869         FOR_EACH_DISK_STORAGE_COLUMN(i, t, cd) {
1870                 char *old_filename, *new_filename;
1871                 ret = create_disk_storage_object_dir(t, i, new_ds_name);
1872                 if (ret < 0)
1873                         goto out;
1874                 old_filename = disk_storage_path(t, i, old_ds_name);
1875                 new_filename = disk_storage_path(t, i, new_ds_name);
1876                 ret = para_rename(old_filename, new_filename);
1877                 free(old_filename);
1878                 free(new_filename);
1879                 if (ret < 0)
1880                         goto out;
1881         }
1882         ret = 1;
1883 out:
1884         free(old_ds_name);
1885         free(new_ds_name);
1886         return ret;
1887
1888 }
1889
1890 /**
1891  * Change an object in an osl table.
1892  *
1893  * \param t Pointer to an open osl table.
1894  * \param r Pointer to the row containing the object to be updated.
1895  * \param col_num Number of the column containing the object to be updated.
1896  * \param obj Pointer to the replacement object.
1897  *
1898  * This function  gets rid of all references to the old object. This includes
1899  * removal of the rbtree node in case there is an rbtree associated with \a
1900  * col_num. It then inserts \a obj into the table and the rbtree if neccessary.
1901  *
1902  * If the \p OSL_RBTREE flag is set for \a col_num, you \b MUST call this
1903  * function in order to change the contents of an object, even for volatile or
1904  * mapped columns of constant size (which may be updated directly if \p
1905  * OSL_RBTREE is not set).  Otherwise the rbtree might become corrupted.
1906  *
1907  * \return Positive on success, negative on errors. Possible errors include: \p
1908  * E_BAD_TABLE, \p E_RB_KEY_EXISTS, \p E_BAD_SIZE, \p E_NOENT, \p E_UNLINK,
1909  * errors returned by para_write_file(), \p E_MKDIR.
1910  */
1911 int osl_update_object(struct osl_table *t, const struct osl_row *r,
1912                 unsigned col_num, struct osl_object *obj)
1913 {
1914         struct osl_column *col;
1915         const struct osl_column_description *cd;
1916         int ret;
1917
1918         if (!t)
1919                 return -E_BAD_TABLE;
1920         col = &t->columns[col_num];
1921         cd = get_column_description(t->desc, col_num);
1922         PARA_DEBUG_LOG("updating column %u of %s\n", col_num, t->desc->name);
1923         if (cd->storage_flags & OSL_RBTREE) {
1924                 if (search_rbtree(obj, t, col_num, NULL, NULL) > 0)
1925                         return -E_RB_KEY_EXISTS;
1926         }
1927         if (cd->storage_flags & OSL_FIXED_SIZE) {
1928                 if (obj->size != cd->data_size)
1929                         return -E_BAD_DATA_SIZE;
1930         }
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) {
1936                 char *ds_name;
1937                 ret = disk_storage_name_of_row(t, r, &ds_name);
1938                 if (ret < 0)
1939                         return ret;
1940                 ret = delete_disk_storage_file(t, col_num, ds_name);
1941                 if (ret < 0 && ret != -E_NOENT) {
1942                         free(ds_name);
1943                         return ret;
1944                 }
1945                 ret = write_disk_storage_file(t, col_num, obj, ds_name);
1946                 free(ds_name);
1947                 if (ret < 0)
1948                         return ret;
1949         } else { /* mapped storage */
1950                 struct osl_object old_obj;
1951                 ret = get_mapped_object(t, col_num, r->num, &old_obj);
1952                 if (ret < 0)
1953                         return ret;
1954                 /*
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.
1958                  */
1959                 if (col_num == t->disk_storage_name_column) {
1960                         ret = rename_disk_storage_objects(t, &old_obj, obj);
1961                         if (ret < 0)
1962                                 return ret;
1963                 }
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;
1968                         char *row_index;
1969                         ret = get_row_index(t, r->num, &row_index);
1970                         if (ret < 0)
1971                                 return ret;
1972                         ret = mark_mapped_object_invalid(t, r->num, col_num);
1973                         if (ret < 0)
1974                                 return ret;
1975                         unmap_column(t, col_num);
1976                         ret = append_map_file(t, col_num, obj,
1977                                 &new_data_map_size);
1978                         if (ret < 0)
1979                                 return ret;
1980                         ret = map_column(t, col_num);
1981                         if (ret < 0)
1982                                 return ret;
1983                         update_cell_index(row_index, col, new_data_map_size,
1984                                 obj->size);
1985                 }
1986         }
1987         if (cd->storage_flags & OSL_RBTREE) {
1988                 ret = insert_rbtree(t, col_num, r, obj);
1989                 if (ret < 0)
1990                         return ret;
1991         }
1992         return 1;
1993 }
1994
1995 /**
1996  * Retrieve an object of type \p OSL_DISK_STORAGE by row and column.
1997  *
1998  * \param t Pointer to an open osl table.
1999  * \param r Pointer to the row containing the object.
2000  * \param col_num The column number.
2001  * \param obj Points to the result upon successful return.
2002  *
2003  * For columns of type \p OSL_DISK_STORAGE, this function must be used to
2004  * retrieve one of its containing objects. Afterwards, osl_close_disk_object()
2005  * must be called in order to deallocate the resources.
2006  *
2007  * \return Positive on success, negative on errors. Possible errors include:
2008  * \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE, errors returned by osl_get_object().
2009  *
2010  * \sa osl_get_object(), osl_storage_type, osl_close_disk_object().
2011  */
2012 int osl_open_disk_object(const struct osl_table *t, const struct osl_row *r,
2013                 unsigned col_num, struct osl_object *obj)
2014 {
2015         const struct osl_column_description *cd;
2016         char *ds_name, *filename;
2017         int ret;
2018
2019         if (!t)
2020                 return -E_BAD_TABLE;
2021         cd = get_column_description(t->desc, col_num);
2022         if (cd->storage_type != OSL_DISK_STORAGE)
2023                 return -E_BAD_STORAGE_TYPE;
2024
2025         ret = disk_storage_name_of_row(t, r, &ds_name);
2026         if (ret < 0)
2027                 return ret;
2028         filename = disk_storage_path(t, col_num, ds_name);
2029         free(ds_name);
2030         PARA_DEBUG_LOG("filename: %s\n", filename);
2031         ret = mmap_full_file(filename, O_RDONLY, obj);
2032         free(filename);
2033         return ret;
2034 }
2035
2036 /**
2037  * Free resources that were allocated during osl_open_disk_object().
2038  *
2039  * \param obj Pointer to the object previously returned by open_disk_object().
2040  *
2041  * \return The return value of the underlying call to para_munmap().
2042  *
2043  * \sa para_munmap().
2044  */
2045 int osl_close_disk_object(struct osl_object *obj)
2046 {
2047         return para_munmap(obj->data, obj->size);
2048 }
2049
2050 /**
2051  * Get the number of rows of the given table.
2052  *
2053  * \param t Pointer to an open osl table.
2054  * \param num_rows Result is returned here.
2055  *
2056  * The number of rows returned via \a num_rows excluding any invalid rows.
2057  *
2058  * \return Positive on success, \p -E_BAD_TABLE if \a t is \p NULL.
2059  */
2060 int osl_get_num_rows(const struct osl_table *t, unsigned *num_rows)
2061 {
2062         if (!t)
2063                 return -E_BAD_TABLE;
2064         assert(t->num_rows >= t->num_invalid_rows);
2065         *num_rows = t->num_rows - t->num_invalid_rows;
2066         return 1;
2067 }
2068
2069 /**
2070  * Get the rank of a row.
2071  *
2072  * \param t An open osl table.
2073  * \param r The row to get the rank of.
2074  * \param col_num The number of an rbtree column.
2075  * \param rank Result pointer.
2076  *
2077  * The rank is, by definition, the position of the row in the linear order
2078  * determined by an inorder tree walk of the rbtree associated with column
2079  * number \a col_num of \a table.
2080  *
2081  * \return Positive on success, negative on errors.
2082  *
2083  * \sa osl_get_nth_row().
2084  */
2085 int osl_get_rank(const struct osl_table *t, struct osl_row *r,
2086                 unsigned col_num, unsigned *rank)
2087 {
2088         struct osl_object obj;
2089         struct osl_column *col;
2090         struct rb_node *node;
2091         int ret = check_rbtree_col(t, col_num, &col);
2092
2093         if (ret < 0)
2094                 return ret;
2095         ret = osl_get_object(t, r, col_num, &obj);
2096         if (ret < 0)
2097                 return ret;
2098         ret = search_rbtree(&obj, t, col_num, &node, NULL);
2099         if (ret < 0)
2100                 return ret;
2101         ret = rb_rank(node, rank);
2102         if (ret < 0)
2103                 return -E_BAD_ROW;
2104         return 1;
2105 }
2106
2107 /**
2108  * Get the row with n-th greatest value.
2109  *
2110  * \param t Pointer to an open osl table.
2111  * \param col_num The column number.
2112  * \param n The rank of the desired row.
2113  * \param result Row is returned here.
2114  *
2115  * Retrieve the n-th order statistic with respect to the compare function
2116  * of the rbtree column \a col_num. In other words, get that row with
2117  * \a n th greatest value in column \a col_num. It's an error if
2118  * \a col_num is not a rbtree column, or if \a n is larger than the
2119  * number of rows in the table.
2120  *
2121  * \return Positive on success, negative on errors. Possible errors:
2122  * \p E_BAD_TABLE, \p E_BAD_STORAGE_FLAGS, \p E_RB_KEY_NOT_FOUND.
2123  *
2124  * \sa osl_storage_flags, osl_compare_func, osl_get_row(),
2125  * osl_rbtree_last_row(), osl_rbtree_first_row(), osl_get_rank().
2126  */
2127 int osl_get_nth_row(const struct osl_table *t, unsigned col_num,
2128                 unsigned n, struct osl_row **result)
2129 {
2130         struct osl_column *col;
2131         struct rb_node *node;
2132         int ret = check_rbtree_col(t, col_num, &col);
2133
2134         if (ret < 0)
2135                 return ret;
2136         node = rb_nth(col->rbtree.rb_node, n);
2137         if (!node)
2138                 return -E_RB_KEY_NOT_FOUND;
2139         *result = get_row_pointer(node, col->rbtree_num);
2140         return 1;
2141 }
2142
2143 /**
2144  * Get the row corresponding to the smallest rbtree node of a column.
2145  *
2146  * \param t An open rbtree table.
2147  * \param col_num The number of the rbtree column.
2148  * \param result A pointer to the first row is returned here.
2149  *
2150  * The rbtree node of the smallest object (with respect to the corresponding
2151  * compare function) is selected and the row containing this object is
2152  * returned. It is an error if \a col_num refers to a column without an
2153  * associated rbtree.
2154  *
2155  * \return Positive on success, negative on errors.
2156  *
2157  * \sa osl_get_nth_row(), osl_rbtree_last_row().
2158  */
2159 int osl_rbtree_first_row(const struct osl_table *t, unsigned col_num,
2160                 struct osl_row **result)
2161 {
2162         return osl_get_nth_row(t, col_num, 1, result);
2163 }
2164
2165 /**
2166  * Get the row corresponding to the greatest rbtree node of a column.
2167  *
2168  * \param t The same meaning as in \p osl_rbtree_first_row().
2169  * \param col_num The same meaning as in \p osl_rbtree_first_row().
2170  * \param result The same meaning as in \p osl_rbtree_first_row().
2171  *
2172  * This function works just like osl_rbtree_first_row(), the only difference
2173  * is that the row containing the greatest rather than the smallest object is
2174  * returned.
2175  *
2176  * \return Positive on success, negative on errors.
2177  *
2178  * \sa osl_get_nth_row(), osl_rbtree_first_row().
2179  */
2180 int osl_rbtree_last_row(const struct osl_table *t, unsigned col_num,
2181                 struct osl_row **result)
2182 {
2183         unsigned num_rows;
2184         int ret = osl_get_num_rows(t, &num_rows);
2185
2186         if (ret < 0)
2187                 return ret;
2188         return osl_get_nth_row(t, col_num, num_rows, result);
2189 }