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         struct stat s;
 271         int ret = -1;
 272
 273 //      PARA_DEBUG_LOG("dirname: %s\n", dirname);
 274         if (cwd_fd < 0)
 275                 return -E_OSL_GETCWD;
 276         ret = -E_OSL_CHDIR;
 277         if (chdir(dirname) < 0)
 278                 goto out;
 279         ret = -E_OSL_OPENDIR;
 280         dir = opendir(".");
 281         if (!dir)
 282                 goto out;
 283         /* scan cwd recursively */
 284         while ((entry = readdir(dir))) {
 285                 mode_t m;
 286                 char *tmp;
 287
 288                 if (!strcmp(entry->d_name, "."))
 289                         continue;
 290                 if (!strcmp(entry->d_name, ".."))
 291                         continue;
 292                 ret = -E_OSL_LSTAT;
 293                 if (lstat(entry->d_name, &s) == -1)
 294                         continue;
 295                 m = s.st_mode;
 296                 if (!S_ISREG(m) && !S_ISDIR(m))
 297                         continue;
 298                 tmp = make_message("%s/%s", dirname, entry->d_name);
 299                 if (!S_ISDIR(m)) {
 300                         ret = func(tmp, private_data);
 301                         free(tmp);
 302                         if (ret < 0)
 303                                 goto out;
 304                         continue;
 305                 }
 306                 /* directory */
 307                 ret = for_each_file_in_dir(tmp, func, private_data);
 308                 free(tmp);
 309                 if (ret < 0)
 310                         goto out;
 311         }
 312         ret = 1;
 313 out:
 314         if (dir)
 315                 closedir(dir);
 316         if (fchdir(cwd_fd) < 0 && ret >= 0)
 317                 ret = -E_OSL_CHDIR;
 318         close(cwd_fd);
 319         return ret;
 320 }
 321
 322 int para_mkdir(const char *path, mode_t mode)
 323 {
 324         if (!mkdir(path, mode))
 325                 return 1;
 326         if (errno == EEXIST)
 327                 return -E_EXIST;
 328         if (errno == ENOSPC)
 329                 return -E_NOSPC;
 330         if (errno == ENOTDIR)
 331                 return -E_NOTDIR;
 332         if (errno == EPERM)
 333                 return E_OSL_PERM;
 334         return -E_MKDIR;
 335 }
 336
 337 static int verify_name(const char *name)
 338 {
 339         if (!name)
 340                 return -E_BAD_NAME;
 341         if (!*name)
 342                 return -E_BAD_NAME;
 343         if (strchr(name, '/'))
 344                 return -E_BAD_NAME;
 345         if (!strcmp(name, ".."))
 346                 return -E_BAD_NAME;
 347         if (!strcmp(name, "."))
 348                 return -E_BAD_NAME;
 349         return 1;
 350 }
 351
 352 /**
 353  * Compare two osl objects pointing to unsigned integers of 32 bit size.
 354  *
 355  * \param obj1 Pointer to the first integer.
 356  * \param obj2 Pointer to the second integer.
 357  *
 358  * \return The values required for an osl compare function.
 359  *
 360  * \sa osl_compare_func, osl_hash_compare().
 361  */
 362 int uint32_compare(const struct osl_object *obj1, const struct osl_object *obj2)
 363 {
 364         uint32_t d1 = read_u32((const char *)obj1->data);
 365         uint32_t d2 = read_u32((const char *)obj2->data);
 366
 367         if (d1 < d2)
 368                 return 1;
 369         if (d1 > d2)
 370                 return -1;
 371         return 0;
 372 }
 373
 374 /**
 375  * Compare two osl objects pointing to hash values.
 376  *
 377  * \param obj1 Pointer to the first hash object.
 378  * \param obj2 Pointer to the second hash object.
 379  *
 380  * \return The values required for an osl compare function.
 381  *
 382  * \sa osl_compare_func, uint32_compare().
 383  */
 384 int osl_hash_compare(const struct osl_object *obj1, const struct osl_object *obj2)
 385 {
 386         return hash_compare((HASH_TYPE *)obj1->data, (HASH_TYPE *)obj2->data);
 387 }
 388
 389 static char *disk_storage_dirname(const struct osl_table *t, unsigned col_num,
 390                 const char *ds_name)
 391 {
 392         char *dirname, *column_name = column_filename(t, col_num);
 393
 394         if (!(t->desc->flags & OSL_LARGE_TABLE))
 395                 return column_name;
 396         dirname = make_message("%s/%.2s", column_name, ds_name);
 397         free(column_name);
 398         return dirname;
 399 }
 400
 401 static char *disk_storage_name_of_object(const struct osl_table *t,
 402         const struct osl_object *obj)
 403 {
 404         HASH_TYPE hash[HASH_SIZE];
 405         hash_object(obj, hash);
 406         return disk_storage_name_of_hash(t, hash);
 407 }
 408
 409 static int disk_storage_name_of_row(const struct osl_table *t,
 410                 const struct osl_row *row, char **name)
 411 {
 412         struct osl_object obj;
 413         int ret = osl_get_object(t, row, t->disk_storage_name_column, &obj);
 414
 415         if (ret < 0)
 416                 return ret;
 417         *name = disk_storage_name_of_object(t, &obj);
 418         return 1;
 419 }
 420
 421 static void column_name_hash(const char *col_name, HASH_TYPE *hash)
 422 {
 423         return hash_function(col_name, strlen(col_name), hash);
 424 }
 425
 426 static int init_column_descriptions(struct osl_table *t)
 427 {
 428         int i, j, ret;
 429         const struct osl_column_description *cd;
 430
 431         ret = -E_BAD_TABLE_DESC;
 432         ret = verify_name(t->desc->name);
 433         if (ret < 0)
 434                 goto err;
 435         ret = -E_BAD_DB_DIR;
 436         if (!t->desc->dir)
 437                 goto err;
 438         /* the size of the index header without column descriptions */
 439         t->index_header_size = IDX_COLUMN_DESCRIPTIONS;
 440         FOR_EACH_COLUMN(i, t->desc, cd) {
 441                 struct osl_column *col = t->columns + i;
 442                 if (cd->storage_flags & OSL_RBTREE) {
 443                         if (!cd->compare_function)
 444                                 return -E_NO_COMPARE_FUNC;
 445                 }
 446                 if (cd->storage_type == OSL_NO_STORAGE)
 447                         continue;
 448                 ret = -E_NO_COLUMN_NAME;
 449                 if (!cd->name || !cd->name[0])
 450                         goto err;
 451                 ret = verify_name(cd->name);
 452                 if (ret < 0)
 453                         goto err;
 454                 t->index_header_size += index_column_description_size(cd->name);
 455                 column_name_hash(cd->name, col->name_hash);
 456                 ret = -E_DUPLICATE_COL_NAME;
 457                 for (j = i + 1; j < t->desc->num_columns; j++) {
 458                         const char *name2 = get_column_description(t->desc,
 459                                 j)->name;
 460                         if (cd->name && name2 && !strcmp(cd->name, name2))
 461                                 goto err;
 462                 }
 463         }
 464         return 1;
 465 err:
 466         return ret;
 467 }
 468
 469 /**
 470  * Initialize a struct table from given table description.
 471  *
 472  * \param desc The description of the osl table.
 473  * \param table_ptr Result is returned here.
 474  *
 475  * This function performs several sanity checks on \p desc and returns if any
 476  * of these tests fail. On success, a struct \p osl_table is allocated and
 477  * initialized with data derived from \p desc.
 478  *
 479  * \return Positive on success, negative on errors. Possible errors include: \p
 480  * E_BAD_TABLE_DESC, \p E_NO_COLUMN_DESC, \p E_NO_COLUMNS, \p
 481  * E_BAD_STORAGE_TYPE, \p E_BAD_STORAGE_FLAGS, \p E_BAD_STORAGE_SIZE, \p
 482  * E_NO_UNIQUE_RBTREE_COLUMN, \p E_NO_RBTREE_COL.
 483  *
 484  * \sa struct osl_table.
 485  */
 486 int init_table_structure(const struct osl_table_description *desc,
 487                 struct osl_table **table_ptr)
 488 {
 489         const struct osl_column_description *cd;
 490         struct osl_table *t = para_calloc(sizeof(*t));
 491         int i, ret = -E_BAD_TABLE_DESC, have_disk_storage_name_column = 0;
 492
 493         if (!desc)
 494                 goto err;
 495         PARA_DEBUG_LOG("creating table structure for '%s' from table "
 496                 "description\n", desc->name);
 497         ret = -E_NO_COLUMN_DESC;
 498         if (!desc->column_descriptions)
 499                 goto err;
 500         ret = -E_NO_COLUMNS;
 501         if (!desc->num_columns)
 502                 goto err;
 503         t->columns = para_calloc(desc->num_columns * sizeof(struct osl_column));
 504         t->desc = desc;
 505         FOR_EACH_COLUMN(i, t->desc, cd) {
 506                 enum osl_storage_type st = cd->storage_type;
 507                 enum osl_storage_flags sf = cd->storage_flags;
 508                 struct osl_column *col = &t->columns[i];
 509
 510                 ret = -E_BAD_STORAGE_TYPE;
 511                 if (st != OSL_MAPPED_STORAGE && st != OSL_DISK_STORAGE
 512                                 && st != OSL_NO_STORAGE)
 513                         goto err;
 514                 ret = -E_BAD_STORAGE_FLAGS;
 515                 if (st == OSL_DISK_STORAGE && sf & OSL_RBTREE)
 516                         goto err;
 517                 ret = -E_BAD_STORAGE_SIZE;
 518                 if (sf & OSL_FIXED_SIZE && !cd->data_size)
 519                         goto err;
 520                 switch (st) {
 521                 case OSL_DISK_STORAGE:
 522                         t->num_disk_storage_columns++;
 523                         break;
 524                 case OSL_MAPPED_STORAGE:
 525                         t->num_mapped_columns++;
 526                         col->index_offset = t->row_index_size;
 527                         t->row_index_size += 8;
 528                         break;
 529                 case OSL_NO_STORAGE:
 530                         col->volatile_num = t->num_volatile_columns;
 531                         t->num_volatile_columns++;
 532                         break;
 533                 }
 534                 if (sf & OSL_RBTREE) {
 535                         col->rbtree_num = t->num_rbtrees;
 536                         t->num_rbtrees++;
 537                         if ((sf & OSL_UNIQUE) && (st == OSL_MAPPED_STORAGE)) {
 538                                 if (!have_disk_storage_name_column)
 539                                         t->disk_storage_name_column = i;
 540                                 have_disk_storage_name_column = 1;
 541                         }
 542                 }
 543         }
 544         ret = -E_NO_UNIQUE_RBTREE_COLUMN;
 545         if (t->num_disk_storage_columns && !have_disk_storage_name_column)
 546                 goto err;
 547         ret = -E_NO_RBTREE_COL;
 548         if (!t->num_rbtrees)
 549                 goto err;
 550         /* success */
 551         PARA_DEBUG_LOG("OK. Index entry size: %u\n", t->row_index_size);
 552         ret = init_column_descriptions(t);
 553         if (ret < 0)
 554                 goto err;
 555         *table_ptr = t;
 556         return 1;
 557 err:
 558         free(t->columns);
 559         free(t);
 560         return ret;
 561 }
 562
 563 /**
 564  * Read the table description from index header.
 565  *
 566  * \param map The memory mapping of the index file.
 567  * \param desc The values found in the index header are returned here.
 568  *
 569  * Read the index header, check for the paraslash magic string and the table version number.
 570  * Read all information stored in the index header into \a desc.
 571  *
 572  * \return Positive on success, negative on errors.
 573  *
 574  * \sa struct osl_table_description, osl_create_table.
 575  */
 576 int read_table_desc(struct osl_object *map, struct osl_table_description *desc)
 577 {
 578         char *buf = map->data;
 579         uint8_t version;
 580         uint16_t header_size;
 581         int ret, i;
 582         unsigned offset;
 583         struct osl_column_description *cd;
 584
 585         if (map->size < MIN_INDEX_HEADER_SIZE(1))
 586                 return -E_SHORT_TABLE;
 587         if (strncmp(buf + IDX_PARA_MAGIC, PARA_MAGIC, strlen(PARA_MAGIC)))
 588                 return -E_NO_MAGIC;
 589         version = read_u8(buf + IDX_VERSION);
 590         if (version < MIN_TABLE_VERSION || version > MAX_TABLE_VERSION)
 591                 return -E_VERSION_MISMATCH;
 592         desc->num_columns = read_u8(buf + IDX_TABLE_FLAGS);
 593         desc->flags = read_u8(buf + IDX_TABLE_FLAGS);
 594         desc->num_columns = read_u16(buf + IDX_NUM_COLUMNS);
 595         PARA_DEBUG_LOG("%u columns\n", desc->num_columns);
 596         if (!desc->num_columns)
 597                 return -E_NO_COLUMNS;
 598         header_size = read_u16(buf + IDX_HEADER_SIZE);
 599         if (map->size < header_size)
 600                 return -E_BAD_SIZE;
 601         desc->column_descriptions = para_calloc(desc->num_columns
 602                 * sizeof(struct osl_column_description));
 603         offset = IDX_COLUMN_DESCRIPTIONS;
 604         FOR_EACH_COLUMN(i, desc, cd) {
 605                 char *null_byte;
 606
 607                 ret = -E_SHORT_TABLE;
 608                 if (map->size < offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE) {
 609                         PARA_ERROR_LOG("map size = %zu < %u = offset + min desc size\n",
 610                                 map->size, offset + MIN_IDX_COLUMN_DESCRIPTION_SIZE);
 611                         goto err;
 612                 }
 613                 cd->storage_type = read_u16(buf + offset + IDX_CD_STORAGE_TYPE);
 614                 cd->storage_flags = read_u16(buf + offset +
 615                         IDX_CD_STORAGE_FLAGS);
 616                 cd->data_size = read_u32(buf + offset + IDX_CD_DATA_SIZE);
 617                 null_byte = memchr(buf + offset + IDX_CD_NAME, '\0',
 618                         map->size - offset - IDX_CD_NAME);
 619                 ret = -E_INDEX_CORRUPTION;
 620                 if (!null_byte)
 621                         goto err;
 622                 cd->name = para_strdup(buf + offset + IDX_CD_NAME);
 623                 offset += index_column_description_size(cd->name);
 624         }
 625         if (offset != header_size) {
 626                 ret = -E_INDEX_CORRUPTION;
 627                 PARA_ERROR_LOG("real header size = %u != %u = stored header size\n",
 628                         offset, header_size);
 629                 goto err;
 630         }
 631         return 1;
 632 err:
 633         FOR_EACH_COLUMN(i, desc, cd)
 634                 free(cd->name);
 635         return ret;
 636 }
 637
 638 /*
 639  * check whether the table description given by \p t->desc matches the on-disk
 640  * table structure stored in the index of \a t.
 641  */
 642 static int compare_table_descriptions(struct osl_table *t)
 643 {
 644         int i, ret;
 645         struct osl_table_description desc;
 646         const struct osl_column_description *cd1, *cd2;
 647
 648         /* read the on-disk structure into desc */
 649         ret = read_table_desc(&t->index_map, &desc);
 650         if (ret < 0)
 651                 return ret;
 652         ret = -E_BAD_TABLE_FLAGS;
 653         if (desc.flags != t->desc->flags)
 654                 goto out;
 655         ret = E_BAD_COLUMN_NUM;
 656         if (desc.num_columns != t->desc->num_columns)
 657                 goto out;
 658         FOR_EACH_COLUMN(i, t->desc, cd1) {
 659                 cd2 = get_column_description(&desc, i);
 660                 ret = -E_BAD_STORAGE_TYPE;
 661                 if (cd1->storage_type != cd2->storage_type)
 662                         goto out;
 663                 ret = -E_BAD_STORAGE_FLAGS;
 664                 if (cd1->storage_flags != cd2->storage_flags) {
 665                         PARA_ERROR_LOG("sf1 = %u != %u = sf2\n",
 666                                 cd1->storage_flags, cd2->storage_flags);
 667                         goto out;
 668                 }
 669                 ret = -E_BAD_DATA_SIZE;
 670                 if (cd1->storage_flags & OSL_FIXED_SIZE)
 671                         if (cd1->data_size != cd2->data_size)
 672                                 goto out;
 673                 ret = -E_BAD_COLUMN_NAME;
 674                 if (strcmp(cd1->name, cd2->name))
 675                         goto out;
 676         }
 677         PARA_DEBUG_LOG("table description of '%s' matches on-disk data, good\n",
 678                 t->desc->name);
 679         ret = 1;
 680 out:
 681         FOR_EACH_COLUMN(i, &desc, cd1)
 682                 free(cd1->name);
 683         free(desc.column_descriptions);
 684         return ret;
 685 }
 686
 687 static int create_table_index(struct osl_table *t)
 688 {
 689         char *buf, *filename;
 690         int i, ret;
 691         size_t size = t->index_header_size;
 692         const struct osl_column_description *cd;
 693         unsigned offset;
 694
 695         PARA_INFO_LOG("creating %zu byte index for table %s\n", size,
 696                 t->desc->name);
 697         buf = para_calloc(size);
 698         sprintf(buf + IDX_PARA_MAGIC, "%s", PARA_MAGIC);
 699         write_u8(buf + IDX_TABLE_FLAGS, t->desc->flags);
 700         write_u8(buf + IDX_DIRTY_FLAG, 0);
 701         write_u8(buf + IDX_VERSION, CURRENT_TABLE_VERSION);
 702         write_u16(buf + IDX_NUM_COLUMNS, t->desc->num_columns);
 703         write_u16(buf + IDX_HEADER_SIZE, t->index_header_size);
 704         offset = IDX_COLUMN_DESCRIPTIONS;
 705         FOR_EACH_COLUMN(i, t->desc, cd) {
 706                 write_u16(buf + offset + IDX_CD_STORAGE_TYPE,
 707                         cd->storage_type);
 708                 write_u16(buf + offset + IDX_CD_STORAGE_FLAGS,
 709                         cd->storage_flags);
 710                 if (cd->storage_flags & OSL_FIXED_SIZE)
 711                         write_u32(buf + offset + IDX_CD_DATA_SIZE,
 712                                 cd->data_size);
 713                 strcpy(buf + offset + IDX_CD_NAME, cd->name);
 714                 offset += index_column_description_size(cd->name);
 715         }
 716         assert(offset = size);
 717         filename = index_filename(t->desc);
 718         ret = para_write_file(filename, buf, size);
 719         free(buf);
 720         free(filename);
 721         return ret;
 722 }
 723
 724 /**
 725  * Create a new osl table.
 726  *
 727  * \param desc Pointer to the table description.
 728  *
 729  * \return Positive on success, negative on errors. Possible errors include: \p
 730  * E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_BAD_NAME, \p E_NO_COMPARE_FUNC, \p
 731  * E_NO_COLUMN_NAME, \p E_DUPLICATE_COL_NAME, \p E_MKDIR, any errors returned
 732  * by para_open().
 733  */
 734 int osl_create_table(const struct osl_table_description *desc)
 735 {
 736         const struct osl_column_description *cd;
 737         char *table_dir = NULL, *filename;
 738         struct osl_table *t;
 739         int i, ret = init_table_structure(desc, &t);
 740
 741         if (ret < 0)
 742                 return ret;
 743         PARA_INFO_LOG("creating %s\n", desc->name);
 744         FOR_EACH_COLUMN(i, t->desc, cd) {
 745                 if (cd->storage_type == OSL_NO_STORAGE)
 746                         continue;
 747                 if (!table_dir) {
 748                         ret = para_mkdir(desc->dir, 0777);
 749                         if (ret < 0 && ret != -E_EXIST)
 750                                 goto out;
 751                         table_dir = make_message("%s/%s", desc->dir,
 752                                 desc->name);
 753                         ret = para_mkdir(table_dir, 0777);
 754                         if (ret < 0)
 755                                 goto out;
 756                 }
 757                 filename = column_filename(t, i);
 758                 PARA_INFO_LOG("filename: %s\n", filename);
 759                 if (cd->storage_type == OSL_MAPPED_STORAGE) {
 760                         ret = para_open(filename, O_RDWR | O_CREAT | O_EXCL,
 761                                 0644);
 762                         free(filename);
 763                         if (ret < 0)
 764                                 goto out;
 765                         close(ret);
 766                         continue;
 767                 }
 768                 /* DISK STORAGE */
 769                 ret = para_mkdir(filename, 0777);
 770                 free(filename);
 771                 if (ret < 0)
 772                         goto out;
 773         }
 774         if (t->num_mapped_columns) {
 775                 ret = create_table_index(t);
 776                 if (ret < 0)
 777                         goto out;
 778         }
 779         ret = 1;
 780 out:
 781         free(table_dir);
 782         free(t->columns);
 783         free(t);
 784         return ret;
 785 }
 786
 787 static int table_is_dirty(struct osl_table *t)
 788 {
 789         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 790         uint8_t dirty = read_u8(buf) & 0x1;
 791         return !!dirty;
 792 }
 793
 794 static void mark_table_dirty(struct osl_table *t)
 795 {
 796         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 797         write_u8(buf, read_u8(buf) | 1);
 798 }
 799
 800 static void mark_table_clean(struct osl_table *t)
 801 {
 802         char *buf = (char *)t->index_map.data + IDX_DIRTY_FLAG;
 803         write_u8(buf, read_u8(buf) & 0xfe);
 804 }
 805
 806 static void unmap_column(struct osl_table *t, unsigned col_num)
 807 {
 808         struct osl_object map = t->columns[col_num].data_map;
 809         int ret;
 810         if (!map.data)
 811                 return;
 812         ret = para_munmap(map.data, map.size);
 813         assert(ret > 0);
 814         map.data = NULL;
 815 }
 816
 817 /**
 818  * Unmap all mapped files of an osl table.
 819  *
 820  * \param t Pointer to a mapped table.
 821  * \param flags Options for unmapping.
 822  *
 823  * \return Positive on success, negative on errors. Possible errors include:
 824  * E_NOT_MAPPED, E_MUNMAP.
 825  *
 826  * \sa map_table(), enum osl_close_flags, para_munmap().
 827  */
 828 int unmap_table(struct osl_table *t, enum osl_close_flags flags)
 829 {
 830         unsigned i;
 831         const struct osl_column_description *cd;
 832         int ret;
 833
 834         if (!t->num_mapped_columns) /* can this ever happen? */
 835                 return 1;
 836         PARA_DEBUG_LOG("unmapping table '%s'\n", t->desc->name);
 837         if (!t->index_map.data)
 838                 return -E_NOT_MAPPED;
 839         if (flags & OSL_MARK_CLEAN)
 840                 mark_table_clean(t);
 841         ret = para_munmap(t->index_map.data, t->index_map.size);
 842         if (ret < 0)
 843                 return ret;
 844         t->index_map.data = NULL;
 845         if (!t->num_rows)
 846                 return 1;
 847         FOR_EACH_MAPPED_COLUMN(i, t, cd)
 848                 unmap_column(t, i);
 849         return 1;
 850 }
 851
 852 static int map_column(struct osl_table *t, unsigned col_num)
 853 {
 854         struct stat statbuf;
 855         char *filename = column_filename(t, col_num);
 856         int ret = -E_STAT;
 857         if (stat(filename, &statbuf) < 0) {
 858                 free(filename);
 859                 return ret;
 860         }
 861         if (!(S_IFREG & statbuf.st_mode)) {
 862                 free(filename);
 863                 return ret;
 864         }
 865         ret = mmap_full_file(filename, O_RDWR,
 866                 &t->columns[col_num].data_map);
 867         free(filename);
 868         return ret;
 869 }
 870
 871 /**
 872  * Map the index file and all columns of type \p OSL_MAPPED_STORAGE into memory.
 873  *
 874  * \param t Pointer to an initialized table structure.
 875  * \param flags Mapping options.
 876  *
 877  * \return Negative return value on errors; on success the number of rows
 878  * (including invalid rows) is returned.
 879  *
 880  * \sa unmap_table(), enum map_table_flags, osl_open_table(), mmap(2).
 881  */
 882 int map_table(struct osl_table *t, enum map_table_flags flags)
 883 {
 884         char *filename;
 885         const struct osl_column_description *cd;
 886         int i = 0, ret, num_rows = 0;
 887
 888         if (!t->num_mapped_columns)
 889                 return 0;
 890         if (t->index_map.data)
 891                 return -E_ALREADY_MAPPED;
 892         filename = index_filename(t->desc);
 893         PARA_DEBUG_LOG("mapping table '%s' (index: %s)\n", t->desc->name, filename);
 894         ret = mmap_full_file(filename, flags & MAP_TBL_FL_MAP_RDONLY?
 895                 O_RDONLY : O_RDWR, &t->index_map);
 896         free(filename);
 897         if (ret < 0)
 898                 return ret;
 899         if (flags & MAP_TBL_FL_VERIFY_INDEX) {
 900                 ret = compare_table_descriptions(t);
 901                 if (ret < 0)
 902                         goto err;
 903         }
 904         ret = -E_BUSY;
 905         if (!(flags & MAP_TBL_FL_IGNORE_DIRTY)) {
 906                 if (table_is_dirty(t)) {
 907                         PARA_ERROR_LOG("%s is dirty\n", t->desc->name);
 908                         goto err;
 909                 }
 910         }
 911         mark_table_dirty(t);
 912         num_rows = table_num_rows(t);
 913         if (!num_rows)
 914                 return num_rows;
 915         /* map data files */
 916         FOR_EACH_MAPPED_COLUMN(i, t, cd) {
 917                 ret = map_column(t, i);
 918                 if (ret < 0)
 919                         goto err;
 920         }
 921         return num_rows;
 922 err:    /* unmap what is already mapped */
 923         for (i--; i >= 0; i--) {
 924                 struct osl_object map = t->columns[i].data_map;
 925                 para_munmap(map.data, map.size);
 926                 map.data = NULL;
 927         }
 928         para_munmap(t->index_map.data, t->index_map.size);
 929         t->index_map.data = NULL;
 930         return ret;
 931 }
 932
 933 /**
 934  * Retrieve a mapped object by row and column number.
 935  *
 936  * \param t Pointer to an open osl table.
 937  * \param col_num Number of the mapped column containing the object to retrieve.
 938  * \param row_num Number of the row containing the object to retrieve.
 939  * \param obj The result is returned here.
 940  *
 941  * It is considered an error if \a col_num does not refer to a column
 942  * of storage type \p OSL_MAPPED_STORAGE.
 943  *
 944  * \return Positive on success, negative on errors. Possible errors include:
 945  * \p E_BAD_ROW_NUM, \p E_INVALID_OBJECT.
 946  *
 947  * \sa osl_storage_type.
 948  */
 949 int get_mapped_object(const struct osl_table *t, unsigned col_num,
 950         uint32_t row_num, struct osl_object *obj)
 951 {
 952         struct osl_column *col = &t->columns[col_num];
 953         uint32_t offset;
 954         char *header;
 955         char *cell_index;
 956         int ret;
 957
 958         if (t->num_rows <= row_num)
 959                 return -E_BAD_ROW_NUM;
 960         ret = get_cell_index(t, row_num, col_num, &cell_index);
 961         if (ret < 0)
 962                 return ret;
 963         offset = read_u32(cell_index);
 964         obj->size = read_u32(cell_index + 4) - 1;
 965         header = col->data_map.data + offset;
 966         obj->data = header + 1;
 967         if (read_u8(header) == 0xff) {
 968                 PARA_ERROR_LOG("col %u, size %zu, offset %u\n", col_num,
 969                         obj->size, offset);
 970                 return -E_INVALID_OBJECT;
 971         }
 972         return 1;
 973 }
 974
 975 static int search_rbtree(const struct osl_object *obj,
 976                 const struct osl_table *t, unsigned col_num,
 977                 struct rb_node **result, struct rb_node ***rb_link)
 978 {
 979         struct osl_column *col = &t->columns[col_num];
 980         struct rb_node **new = &col->rbtree.rb_node, *parent = NULL;
 981         const struct osl_column_description *cd =
 982                 get_column_description(t->desc, col_num);
 983         enum osl_storage_type st = cd->storage_type;
 984         while (*new) {
 985                 struct osl_row *this_row = get_row_pointer(*new,
 986                         col->rbtree_num);
 987                 int ret;
 988                 struct osl_object this_obj;
 989                 parent = *new;
 990                 if (st == OSL_MAPPED_STORAGE) {
 991                         ret = get_mapped_object(t, col_num, this_row->num,
 992                                 &this_obj);
 993                         if (ret < 0)
 994                                 return ret;
 995                 } else
 996                         this_obj = this_row->volatile_objects[col->volatile_num];
 997                 ret = cd->compare_function(obj, &this_obj);
 998                 if (!ret) {
 999                         if (result)
1000                                 *result = get_rb_node_pointer(this_row,
1001                                         col->rbtree_num);
1002                         return 1;
1003                 }
1004                 if (ret < 0)
1005                         new = &((*new)->rb_left);
1006                 else
1007                         new = &((*new)->rb_right);
1008         }
1009         if (result)
1010                 *result = parent;
1011         if (rb_link)
1012                 *rb_link = new;
1013         return -E_RB_KEY_NOT_FOUND;
1014 }
1015
1016 static int insert_rbtree(struct osl_table *t, unsigned col_num,
1017         const struct osl_row *row, const struct osl_object *obj)
1018 {
1019         struct rb_node *parent, **rb_link;
1020         unsigned rbtree_num;
1021         struct rb_node *n;
1022         int ret = search_rbtree(obj, t, col_num, &parent, &rb_link);
1023
1024         if (ret > 0)
1025                 return -E_RB_KEY_EXISTS;
1026         rbtree_num = t->columns[col_num].rbtree_num;
1027         n = get_rb_node_pointer(row, rbtree_num);
1028         rb_link_node(n, parent, rb_link);
1029         rb_insert_color(n, &t->columns[col_num].rbtree);
1030         return 1;
1031 }
1032
1033 static void remove_rb_node(struct osl_table *t, unsigned col_num,
1034                 const struct osl_row *row)
1035 {
1036         struct osl_column *col = &t->columns[col_num];
1037         const struct osl_column_description *cd =
1038                 get_column_description(t->desc, col_num);
1039         enum osl_storage_flags sf = cd->storage_flags;
1040         struct rb_node *victim, *splice_out_node, *tmp;
1041         if (!(sf & OSL_RBTREE))
1042                 return;
1043         /*
1044          * Which node is removed/spliced out actually depends on how many
1045          * children the victim node has: If it has no children, it gets
1046          * deleted. If it has one child, it gets spliced out. If it has two
1047          * children, its successor (which has at most a right child) gets
1048          * spliced out.
1049          */
1050         victim = get_rb_node_pointer(row, col->rbtree_num);
1051         if (victim->rb_left && victim->rb_right)
1052                 splice_out_node = rb_next(victim);
1053         else
1054                 splice_out_node = victim;
1055         /* Go up to the root and decrement the size of each node in the path. */
1056         for (tmp = splice_out_node; tmp; tmp = rb_parent(tmp))
1057                 tmp->size--;
1058         rb_erase(victim, &col->rbtree);
1059 }
1060
1061 static int add_row_to_rbtrees(struct osl_table *t, uint32_t row_num,
1062                 struct osl_object *volatile_objs, struct osl_row **row_ptr)
1063 {
1064         unsigned i;
1065         int ret;
1066         struct osl_row *row = allocate_row(t->num_rbtrees);
1067         const struct osl_column_description *cd;
1068
1069         row->num = row_num;
1070         row->volatile_objects = volatile_objs;
1071         FOR_EACH_RBTREE_COLUMN(i, t, cd) {
1072                 if (cd->storage_type == OSL_MAPPED_STORAGE) {
1073                         struct osl_object obj;
1074                         ret = get_mapped_object(t, i, row_num, &obj);
1075                         if (ret < 0)
1076                                 goto err;
1077                         ret = insert_rbtree(t, i, row, &obj);
1078                 } else { /* volatile */
1079                         const struct osl_object *obj
1080                                 = volatile_objs + t->columns[i].volatile_num;
1081                         ret = insert_rbtree(t, i, row, obj);
1082                 }
1083                 if (ret < 0)
1084                         goto err;
1085         }
1086         if (row_ptr)
1087                 *row_ptr = row;
1088         return 1;
1089 err: /* rollback changes, i.e. remove added entries from rbtrees */
1090         while (i)
1091                 remove_rb_node(t, i--, row);
1092         free(row);
1093         return ret;
1094 }
1095
1096 static void free_volatile_objects(const struct osl_table *t,
1097                 enum osl_close_flags flags)
1098 {
1099         int i, j;
1100         struct rb_node *n;
1101         struct osl_column *rb_col;
1102         const struct osl_column_description *cd;
1103
1104         if (!t->num_volatile_columns)
1105                 return;
1106         /* find the first rbtree column (any will do) */
1107         FOR_EACH_RBTREE_COLUMN(i, t, cd)
1108                 break;
1109         rb_col = t->columns + i;
1110         /* walk that rbtree and free all volatile objects */
1111         for (n = rb_first(&rb_col->rbtree); n; n = rb_next(n)) {
1112                 struct osl_row *r = get_row_pointer(n, rb_col->rbtree_num);
1113                 if (flags & OSL_FREE_VOLATILE)
1114                         for (j = 0; j < t->num_volatile_columns; j++)
1115                                 free(r->volatile_objects[j].data);
1116                 free(r->volatile_objects);
1117         }
1118 }
1119
1120 /**
1121  * Erase all rbtree nodes and free resources.
1122  *
1123  * \param t Pointer to an open osl table.
1124  *
1125  * This function is called by osl_close_table().
1126  */
1127 void clear_rbtrees(struct osl_table *t)
1128 {
1129         const struct osl_column_description *cd;
1130         unsigned i, rbtrees_cleared = 0;
1131
1132         FOR_EACH_RBTREE_COLUMN(i, t, cd) {
1133                 struct osl_column *col = &t->columns[i];
1134                 struct rb_node *n;
1135                 rbtrees_cleared++;
1136                 for (n = rb_first(&col->rbtree); n;) {
1137                         struct osl_row *r;
1138                         rb_erase(n, &col->rbtree);
1139                         if (rbtrees_cleared == t->num_rbtrees) {
1140                                 r = get_row_pointer(n, col->rbtree_num);
1141                                 n = rb_next(n);
1142                                 free(r);
1143                         } else
1144                                 n = rb_next(n);
1145                 }
1146         }
1147
1148 }
1149
1150 /**
1151  * Close an osl table.
1152  *
1153  * \param t Pointer to the table to be closed.
1154  * \param flags Options for what should be cleaned up.
1155  *
1156  * If osl_open_table() succeeds, the resulting table pointer must later be
1157  * passed to this function in order to flush all changes to the filesystem and
1158  * to free the resources that were allocated by osl_open_table().
1159  *
1160  * \return Positive on success, negative on errors. Possible errors: \p E_BAD_TABLE,
1161  * errors returned by unmap_table().
1162  *
1163  * \sa osl_open_table(), unmap_table().
1164  */
1165 int osl_close_table(struct osl_table *t, enum osl_close_flags flags)
1166 {
1167         int ret;
1168
1169         if (!t)
1170                 return -E_BAD_TABLE;
1171         free_volatile_objects(t, flags);
1172         clear_rbtrees(t);
1173         ret = unmap_table(t, flags);
1174         if (ret < 0)
1175                 PARA_ERROR_LOG("unmap_table failed: %d\n", ret);
1176         free(t->columns);
1177         free(t);
1178         return ret;
1179 }
1180
1181 /**
1182  * Find out whether the given row number corresponds to an invalid row.
1183  *
1184  * \param t Pointer to the osl table.
1185  * \param row_num The number of the row in question.
1186  *
1187  * By definition, a row is considered invalid if all its index entries
1188  * are invalid.
1189  *
1190  * \return Positive if \a row_num corresponds to an invalid row,
1191  * zero if it corresponds to a valid row, negative on errors.
1192  */
1193 int row_is_invalid(struct osl_table *t, uint32_t row_num)
1194 {
1195         char *row_index;
1196         int i, ret = get_row_index(t, row_num, &row_index);
1197
1198         if (ret < 0)
1199                 return ret;
1200         for (i = 0; i < t->row_index_size; i++) {
1201                 if ((unsigned char)row_index[i] != 0xff)
1202                         return 0;
1203         }
1204         PARA_INFO_LOG("row %d is invalid\n", row_num);
1205         return 1;
1206 }
1207
1208 /**
1209  * Invalidate a row of an osl table.
1210  *
1211  * \param t Pointer to an open osl table.
1212  * \param row_num Number of the row to mark as invalid.
1213  *
1214  * This function marks each mapped object in the index entry of \a row as
1215  * invalid.
1216  *
1217  * \return Positive on success, negative on errors.
1218  */
1219 int mark_row_invalid(struct osl_table *t, uint32_t row_num)
1220 {
1221         char *row_index;
1222         int ret = get_row_index(t, row_num, &row_index);
1223
1224         if (ret < 0)
1225                 return ret;
1226         PARA_INFO_LOG("marking row %d as invalid\n", row_num);
1227         memset(row_index, 0xff, t->row_index_size);
1228         return 1;
1229 }
1230
1231 /**
1232  * Initialize all rbtrees and compute number of invalid rows.
1233  *
1234  * \param t The table containing the rbtrees to be initialized.
1235  *
1236  * \return Positive on success, negative on errors.
1237  */
1238 int init_rbtrees(struct osl_table *t)
1239 {
1240         int i, ret;
1241         const struct osl_column_description *cd;
1242
1243         /* create rbtrees */
1244         FOR_EACH_RBTREE_COLUMN(i, t, cd)
1245                 t->columns[i].rbtree = RB_ROOT;
1246         /* add valid rows to rbtrees */
1247         t->num_invalid_rows = 0;
1248         for (i = 0; i < t->num_rows; i++) {
1249                 ret = row_is_invalid(t, i);
1250                 if (ret < 0)
1251                         return ret;
1252                 if (ret) {
1253                         t->num_invalid_rows++;
1254                         continue;
1255                 }
1256                 ret = add_row_to_rbtrees(t, i, NULL, NULL);
1257                 if (ret < 0)
1258                         return ret;
1259         }
1260         return 1;
1261 }
1262
1263 /**
1264  * Open an osl table.
1265  *
1266  * Each osl table must be opened before its data can be accessed.
1267  *
1268  * \param table_desc Describes the table to be opened.
1269  * \param result Contains a pointer to the open table on success.
1270  *
1271  * The table description given by \a desc should coincide with the
1272  * description used at creation time.
1273  *
1274  * \return Positive on success, negative on errors. Possible errors include:
1275  * errors returned by init_table_structure(), \p E_NOENT, \p E_STAT, \p \p
1276  * E_NOTDIR, \p E_BAD_TABLE_DESC, \p E_BAD_DB_DIR, \p E_NO_COMPARE_FUNC, \p
1277  * E_NO_COLUMN_NAME, errors returned by init_rbtrees().
1278  */
1279 int osl_open_table(const struct osl_table_description *table_desc,
1280                 struct osl_table **result)
1281 {
1282         int i, ret;
1283         struct osl_table *t;
1284         const struct osl_column_description *cd;
1285
1286         PARA_INFO_LOG("opening table %s\n", table_desc->name);
1287         ret = init_table_structure(table_desc, &t);
1288         if (ret < 0)
1289                 return ret;
1290         FOR_EACH_DISK_STORAGE_COLUMN(i, t, cd) {
1291                 /* check if directory exists */
1292                 char *dirname = column_filename(t, i);
1293                 struct stat statbuf;
1294                 ret = stat(dirname, &statbuf);
1295                 free(dirname);
1296                 if (ret < 0) {
1297                         if (errno == ENOENT)
1298                                 ret = -E_NOENT;
1299                         else
1300                                 ret = -E_STAT;
1301                         goto err;
1302                 }
1303                 ret = -E_NOTDIR;
1304                 if (!S_ISDIR(statbuf.st_mode))
1305                         goto err;
1306         }
1307         ret = map_table(t, MAP_TBL_FL_VERIFY_INDEX);
1308         if (ret < 0)
1309                 goto err;
1310         t->num_rows = ret;
1311         PARA_DEBUG_LOG("num rows: %d\n", t->num_rows);
1312         ret = init_rbtrees(t);
1313         if (ret < 0) {
1314                 osl_close_table(t, OSL_MARK_CLEAN); /* ignore further errors */
1315                 return ret;
1316         }
1317         *result = t;
1318         return 1;
1319 err:
1320         free(t->columns);
1321         free(t);
1322         return ret;
1323 }
1324
1325 static int create_disk_storage_object_dir(const struct osl_table *t,
1326                 unsigned col_num, const char *ds_name)
1327 {
1328         char *dirname;
1329         int ret;
1330
1331         if (!(t->desc->flags & OSL_LARGE_TABLE))
1332                 return 1;
1333         dirname = disk_storage_dirname(t, col_num, ds_name);
1334         ret = para_mkdir(dirname, 0777);
1335         free(dirname);
1336         if (ret < 0 && ret != -E_EXIST)
1337                 return ret;
1338         return 1;
1339 }
1340
1341 static int write_disk_storage_file(const struct osl_table *t, unsigned col_num,
1342         const struct osl_object *obj, const char *ds_name)
1343 {
1344         int ret;
1345         char *filename;
1346
1347         ret = create_disk_storage_object_dir(t, col_num, ds_name);
1348         if (ret < 0)
1349                 return ret;
1350         filename = disk_storage_path(t, col_num, ds_name);
1351         ret = para_write_file(filename, obj->data, obj->size);
1352         free(filename);
1353         return ret;
1354 }
1355
1356 static int append_map_file(const struct osl_table *t, unsigned col_num,
1357         const struct osl_object *obj, uint32_t *new_size)
1358 {
1359         char *filename = column_filename(t, col_num);
1360         int ret;
1361         char header = 0; /* zero means valid object */
1362
1363 //      PARA_DEBUG_LOG("appending %zu + 1 byte\n", obj->size);
1364         ret = append_file(filename, &header, 1, obj->data, obj->size,
1365                 new_size);
1366         free(filename);
1367         return ret;
1368 }
1369
1370 static int append_row_index(const struct osl_table *t, char *row_index)
1371 {
1372         char *filename;
1373         int ret;
1374
1375         if (!t->num_mapped_columns)
1376                 return 1;
1377         filename = index_filename(t->desc);
1378         ret = append_file(filename, NULL, 0, row_index,
1379                 t->row_index_size, NULL);
1380         free(filename);
1381         return ret;
1382 }
1383
1384 /**
1385  * A wrapper for truncate(2)
1386  *
1387  * \param path Name of the regular file to truncate
1388  * \param size Number of bytes to \b shave \b off
1389  *
1390  * Truncate the regular file named by \a path by \a size bytes.
1391  *
1392  * \return Positive on success, negative on errors. Possible errors include: \p
1393  * E_STAT, \p E_BAD_SIZE, \p E_TRUNC.
1394  *
1395  * \sa truncate(2)
1396  */
1397 int para_truncate(const char *path, off_t size)
1398 {
1399         int ret;
1400         struct stat statbuf;
1401
1402         ret = -E_STAT;
1403         if (stat(path, &statbuf) < 0)
1404                 goto out;
1405         ret = -E_BAD_SIZE;
1406         if (statbuf.st_size < size)
1407                 goto out;
1408         ret = -E_TRUNC;
1409         if (truncate(path, statbuf.st_size - size) < 0)
1410                 goto out;
1411         ret = 1;
1412 out:
1413         return ret;
1414 }
1415
1416 static int truncate_mapped_file(const struct osl_table *t, unsigned col_num,
1417                 off_t size)
1418 {
1419         char *filename = column_filename(t, col_num);
1420         int ret = para_truncate(filename, size);
1421         free(filename);
1422         return ret;
1423 }
1424
1425 static int delete_disk_storage_file(const struct osl_table *t, unsigned col_num,
1426                 const char *ds_name)
1427 {
1428         char *dirname, *filename = disk_storage_path(t, col_num, ds_name);
1429         int ret = unlink(filename);
1430
1431         PARA_DEBUG_LOG("deleted %s\n", filename);
1432         free(filename);
1433         if (ret < 0) {
1434                 if (errno == ENOENT)
1435                         return -E_NOENT;
1436                 return -E_UNLINK;
1437         }
1438         if (!(t->desc->flags & OSL_LARGE_TABLE))
1439                 return 1;
1440         dirname = disk_storage_dirname(t, col_num, ds_name);
1441         rmdir(dirname);
1442         free(dirname);
1443         return 1;
1444 }
1445
1446 /**
1447  * Add a new row to an osl table and retrieve this row.
1448  *
1449  * \param t Pointer to an open osl table.
1450  * \param objects Array of objects to be added.
1451  * \param row Result pointer.
1452  *
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).
1460  *
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).
1464  *
1465  * Possible errors include: \p E_RB_KEY_EXISTS, \p E_BAD_DATA_SIZE.
1466  *
1467  * \return Positive on success, negative on errors.
1468  *
1469  * \sa struct osl_table_description, osl_compare_func, osl_add_row().
1470  */
1471 int osl_add_and_get_row(struct osl_table *t, struct osl_object *objects,
1472                 struct osl_row **row)
1473 {
1474         int i, ret;
1475         char *ds_name = NULL;
1476         struct rb_node **rb_parents = NULL, ***rb_links = NULL;
1477         char *new_row_index = NULL;
1478         struct osl_object *volatile_objs = NULL;
1479         const struct osl_column_description *cd;
1480
1481         if (!t)
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_row_index = para_malloc(t->row_index_size);
1487         /* pass 1: sanity checks */
1488 //      PARA_DEBUG_LOG("sanity tests: %p:%p\n", objects[0].data,
1489 //              objects[1].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;
1493
1494 //              ret = -E_NULL_OBJECT;
1495 //              if (!objects[i])
1496 //                      goto out;
1497                 if (st == OSL_DISK_STORAGE)
1498                         continue;
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",
1503 //                              objects[i].data);
1504                         if (search_rbtree(objects + i, t, i,
1505                                         &rb_parents[rbtree_num],
1506                                         &rb_links[rbtree_num]) > 0)
1507                                 goto out;
1508                 }
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)
1514                                 goto out;
1515                 }
1516         }
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);
1521         if (ret < 0)
1522                 goto out;
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)
1528                         continue;
1529                 if (st == OSL_MAPPED_STORAGE) {
1530                         uint32_t new_size;
1531                         struct osl_column *col = &t->columns[i];
1532 //                      PARA_DEBUG_LOG("appending object of size %zu\n",
1533 //                              objects[i].size);
1534                         ret = append_map_file(t, i, objects + i, &new_size);
1535                         if (ret < 0)
1536                                 goto rollback;
1537                         update_cell_index(new_row_index, col, new_size,
1538                                 objects[i].size);
1539                         continue;
1540                 }
1541                 /* DISK_STORAGE */
1542                 ret = write_disk_storage_file(t, i, objects + i, ds_name);
1543                 if (ret < 0)
1544                         goto rollback;
1545         }
1546         ret = append_row_index(t, new_row_index);
1547         if (ret < 0)
1548                 goto rollback;
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->row_index_size);
1553                 free(filename);
1554                 goto rollback;
1555         }
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];
1562         }
1563         t->num_rows++;
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);
1566         if (ret < 0)
1567                 goto out;
1568 //      PARA_DEBUG_LOG("added new entry as row #%d\n", t->num_rows - 1);
1569         ret = 1;
1570         goto out;
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)
1576                         continue;
1577
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);
1582         }
1583         /* ignore error and return previous error value */
1584         map_table(t, MAP_TBL_FL_VERIFY_INDEX);
1585 out:
1586         free(new_row_index);
1587         free(ds_name);
1588         free(rb_parents);
1589         free(rb_links);
1590         return ret;
1591 }
1592
1593 /**
1594  * Add a new row to an osl table.
1595  *
1596  * \param t Same meaning as osl_add_and_get_row().
1597  * \param objects Same meaning as osl_add_and_get_row().
1598  *
1599  * \return The return value of the underlying call to osl_add_and_get_row().
1600  *
1601  * This is equivalent to osl_add_and_get_row(t, objects, NULL).
1602  */
1603 int osl_add_row(struct osl_table *t, struct osl_object *objects)
1604 {
1605         return osl_add_and_get_row(t, objects, NULL);
1606 }
1607
1608 /**
1609  * Retrieve an object identified by row and column
1610  *
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.
1615  *
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
1618  * function.
1619  *
1620  * \return Positive if object was found, negative on errors. Possible errors
1621  * include: \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE.
1622  *
1623  * \sa osl_storage_type, osl_open_disk_object().
1624  */
1625 int osl_get_object(const struct osl_table *t, const struct osl_row *r,
1626         unsigned col_num, struct osl_object *object)
1627 {
1628         const struct osl_column_description *cd;
1629
1630         if (!t)
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->num, object);
1638         /* volatile */
1639         *object = r->volatile_objects[t->columns[col_num].volatile_num];
1640         return 1;
1641 }
1642
1643 static int mark_mapped_object_invalid(const struct osl_table *t,
1644                 uint32_t row_num, unsigned col_num)
1645 {
1646         struct osl_object obj;
1647         char *p;
1648         int ret = get_mapped_object(t, col_num, row_num, &obj);
1649
1650         if (ret < 0)
1651                 return ret;
1652         p = obj.data;
1653         p--;
1654         *p = 0xff;
1655         return 1;
1656 }
1657
1658 /**
1659  * Delete a row from an osl table.
1660  *
1661  * \param t Pointer to an open osl table.
1662  * \param row Pointer to the row to delete.
1663  *
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
1667  * para_fsck.
1668  *
1669  * \return Positive on success, negative on errors. Possible errors include:
1670  * \p E_BAD_TABLE, errors returned by osl_get_object().
1671  */
1672 int osl_del_row(struct osl_table *t, struct osl_row *row)
1673 {
1674         struct osl_row *r = row;
1675         int i, ret;
1676         const struct osl_column_description *cd;
1677
1678         if (!t)
1679                 return -E_BAD_TABLE;
1680         PARA_INFO_LOG("deleting row %p\n", row);
1681
1682         if (t->num_disk_storage_columns) {
1683                 char *ds_name;
1684                 ret = disk_storage_name_of_row(t, r, &ds_name);
1685                 if (ret < 0)
1686                         goto out;
1687                 FOR_EACH_DISK_STORAGE_COLUMN(i, t, cd)
1688                         delete_disk_storage_file(t, i, ds_name);
1689                 free(ds_name);
1690         }
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->num, i);
1697                         continue;
1698                 }
1699                 if (st == OSL_NO_STORAGE)
1700                         free(r->volatile_objects[col->volatile_num].data);
1701         }
1702         if (t->num_mapped_columns) {
1703                 ret = mark_row_invalid(t, r->num);
1704                 if (ret < 0)
1705                         goto out;
1706                 t->num_invalid_rows++;
1707         } else
1708                 t->num_rows--;
1709         ret = 1;
1710 out:
1711         free(r->volatile_objects);
1712         free(r);
1713         return ret;
1714 }
1715
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)
1719 {
1720         if (!t)
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;
1725         return 1;
1726 }
1727
1728 /**
1729  * Get the row that contains the given object.
1730  *
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.
1735  *
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.
1738  *
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.
1743  *
1744  * \sa osl_storage_flags
1745  */
1746 int osl_get_row(const struct osl_table *t, unsigned col_num,
1747                 const struct osl_object *obj, struct osl_row **result)
1748 {
1749         int ret;
1750         struct rb_node *node;
1751         struct osl_row *row;
1752         struct osl_column *col;
1753
1754         *result = NULL;
1755         ret = check_rbtree_col(t, col_num, &col);
1756         if (ret < 0)
1757                 return ret;
1758         ret = search_rbtree(obj, t, col_num, &node, NULL);
1759         if (ret < 0)
1760                 return ret;
1761         row = get_row_pointer(node, t->columns[col_num].rbtree_num);
1762         *result = row;
1763         return 1;
1764 }
1765
1766 static int rbtree_loop(struct osl_column *col,  void *private_data,
1767                 osl_rbtree_loop_func *func)
1768 {
1769         struct rb_node *n;
1770
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);
1774                 if (ret < 0)
1775                         return ret;
1776         }
1777         return 1;
1778 }
1779
1780 static int rbtree_loop_reverse(struct osl_column *col,  void *private_data,
1781                 osl_rbtree_loop_func *func)
1782 {
1783         struct rb_node *n;
1784
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);
1788                 if (ret < 0)
1789                         return ret;
1790         }
1791         return 1;
1792 }
1793
1794 /**
1795  * Loop over all nodes in an rbtree.
1796  *
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.
1801  *
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.
1810  *
1811  *
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
1814  * returned.
1815  *
1816  * \sa osl_storage_flags, osl_rbtree_loop_reverse(), osl_compare_func.
1817  */
1818 int osl_rbtree_loop(const struct osl_table *t, unsigned col_num,
1819         void *private_data, osl_rbtree_loop_func *func)
1820 {
1821         struct osl_column *col;
1822
1823         int ret = check_rbtree_col(t, col_num, &col);
1824         if (ret < 0)
1825                 return ret;
1826         return rbtree_loop(col, private_data, func);
1827 }
1828
1829 /**
1830  * Loop over all nodes in an rbtree in reverse order.
1831  *
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().
1836  *
1837  * This function is identical to \p osl_rbtree_loop(), the only difference
1838  * is that the tree is walked in reverse order.
1839  *
1840  * \return The same return value as \p osl_rbtree_loop().
1841  *
1842  * \sa osl_rbtree_loop().
1843  */
1844 int osl_rbtree_loop_reverse(const struct osl_table *t, unsigned col_num,
1845         void *private_data, osl_rbtree_loop_func *func)
1846 {
1847         struct osl_column *col;
1848
1849         int ret = check_rbtree_col(t, col_num, &col);
1850         if (ret < 0)
1851                 return ret;
1852         return rbtree_loop_reverse(col, private_data, func);
1853 }
1854
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)
1858 {
1859         int i, ret;
1860         const struct osl_column_description *cd;
1861         char *old_ds_name, *new_ds_name;
1862
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);
1873                 if (ret < 0)
1874                         goto out;
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);
1878                 free(old_filename);
1879                 free(new_filename);
1880                 if (ret < 0)
1881                         goto out;
1882         }
1883         ret = 1;
1884 out:
1885         free(old_ds_name);
1886         free(new_ds_name);
1887         return ret;
1888
1889 }
1890
1891 /**
1892  * Change an object in an osl table.
1893  *
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.
1898  *
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.
1902  *
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.
1907  *
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.
1911  */
1912 int osl_update_object(struct osl_table *t, const struct osl_row *r,
1913                 unsigned col_num, struct osl_object *obj)
1914 {
1915         struct osl_column *col;
1916         const struct osl_column_description *cd;
1917         int ret;
1918
1919         if (!t)
1920                 return -E_BAD_TABLE;
1921         col = &t->columns[col_num];
1922         cd = get_column_description(t->desc, col_num);
1923         PARA_DEBUG_LOG("updating column %u of %s\n", col_num, t->desc->name);
1924         if (cd->storage_flags & OSL_RBTREE) {
1925                 if (search_rbtree(obj, t, col_num, NULL, NULL) > 0)
1926                         return -E_RB_KEY_EXISTS;
1927         }
1928         if (cd->storage_flags & OSL_FIXED_SIZE) {
1929                 if (obj->size != cd->data_size)
1930                         return -E_BAD_DATA_SIZE;
1931         }
1932         remove_rb_node(t, col_num, r);
1933         if (cd->storage_type == OSL_NO_STORAGE) { /* TODO: If fixed size, reuse object? */
1934                 free(r->volatile_objects[col->volatile_num].data);
1935                 r->volatile_objects[col->volatile_num] = *obj;
1936         } else if (cd->storage_type == OSL_DISK_STORAGE) {
1937                 char *ds_name;
1938                 ret = disk_storage_name_of_row(t, r, &ds_name);
1939                 if (ret < 0)
1940                         return ret;
1941                 ret = delete_disk_storage_file(t, col_num, ds_name);
1942                 if (ret < 0 && ret != -E_NOENT) {
1943                         free(ds_name);
1944                         return ret;
1945                 }
1946                 ret = write_disk_storage_file(t, col_num, obj, ds_name);
1947                 free(ds_name);
1948                 if (ret < 0)
1949                         return ret;
1950         } else { /* mapped storage */
1951                 struct osl_object old_obj;
1952                 ret = get_mapped_object(t, col_num, r->num, &old_obj);
1953                 if (ret < 0)
1954                         return ret;
1955                 /*
1956                  * If the updated column is the disk storage name column, the
1957                  * disk storage name changes, so we have to rename all disk
1958                  * storage objects accordingly.
1959                  */
1960                 if (col_num == t->disk_storage_name_column) {
1961                         ret = rename_disk_storage_objects(t, &old_obj, obj);
1962                         if (ret < 0)
1963                                 return ret;
1964                 }
1965                 if (cd->storage_flags & OSL_FIXED_SIZE)
1966                         memcpy(old_obj.data, obj->data, cd->data_size);
1967                 else { /* TODO: if the size doesn't change, use old space */
1968                         uint32_t new_data_map_size;
1969                         char *row_index;
1970                         ret = get_row_index(t, r->num, &row_index);
1971                         if (ret < 0)
1972                                 return ret;
1973                         ret = mark_mapped_object_invalid(t, r->num, col_num);
1974                         if (ret < 0)
1975                                 return ret;
1976                         unmap_column(t, col_num);
1977                         ret = append_map_file(t, col_num, obj,
1978                                 &new_data_map_size);
1979                         if (ret < 0)
1980                                 return ret;
1981                         ret = map_column(t, col_num);
1982                         if (ret < 0)
1983                                 return ret;
1984                         update_cell_index(row_index, col, new_data_map_size,
1985                                 obj->size);
1986                 }
1987         }
1988         if (cd->storage_flags & OSL_RBTREE) {
1989                 ret = insert_rbtree(t, col_num, r, obj);
1990                 if (ret < 0)
1991                         return ret;
1992         }
1993         return 1;
1994 }
1995
1996 /**
1997  * Retrieve an object of type \p OSL_DISK_STORAGE by row and column.
1998  *
1999  * \param t Pointer to an open osl table.
2000  * \param r Pointer to the row containing the object.
2001  * \param col_num The column number.
2002  * \param obj Points to the result upon successful return.
2003  *
2004  * For columns of type \p OSL_DISK_STORAGE, this function must be used to
2005  * retrieve one of its containing objects. Afterwards, osl_close_disk_object()
2006  * must be called in order to deallocate the resources.
2007  *
2008  * \return Positive on success, negative on errors. Possible errors include:
2009  * \p E_BAD_TABLE, \p E_BAD_STORAGE_TYPE, errors returned by osl_get_object().
2010  *
2011  * \sa osl_get_object(), osl_storage_type, osl_close_disk_object().
2012  */
2013 int osl_open_disk_object(const struct osl_table *t, const struct osl_row *r,
2014                 unsigned col_num, struct osl_object *obj)
2015 {
2016         const struct osl_column_description *cd;
2017         char *ds_name, *filename;
2018         int ret;
2019
2020         if (!t)
2021                 return -E_BAD_TABLE;
2022         cd = get_column_description(t->desc, col_num);
2023         if (cd->storage_type != OSL_DISK_STORAGE)
2024                 return -E_BAD_STORAGE_TYPE;
2025
2026         ret = disk_storage_name_of_row(t, r, &ds_name);
2027         if (ret < 0)
2028                 return ret;
2029         filename = disk_storage_path(t, col_num, ds_name);
2030         free(ds_name);
2031         PARA_DEBUG_LOG("filename: %s\n", filename);
2032         ret = mmap_full_file(filename, O_RDONLY, obj);
2033         free(filename);
2034         return ret;
2035 }
2036
2037 /**
2038  * Free resources that were allocated during osl_open_disk_object().
2039  *
2040  * \param obj Pointer to the object previously returned by open_disk_object().
2041  *
2042  * \return The return value of the underlying call to para_munmap().
2043  *
2044  * \sa para_munmap().
2045  */
2046 int osl_close_disk_object(struct osl_object *obj)
2047 {
2048         return para_munmap(obj->data, obj->size);
2049 }
2050
2051 /**
2052  * Get the number of rows of the given table.
2053  *
2054  * \param t Pointer to an open osl table.
2055  * \param num_rows Result is returned here.
2056  *
2057  * The number of rows returned via \a num_rows excluding any invalid rows.
2058  *
2059  * \return Positive on success, \p -E_BAD_TABLE if \a t is \p NULL.
2060  */
2061 int osl_get_num_rows(const struct osl_table *t, unsigned *num_rows)
2062 {
2063         if (!t)
2064                 return -E_BAD_TABLE;
2065         assert(t->num_rows >= t->num_invalid_rows);
2066         *num_rows = t->num_rows - t->num_invalid_rows;
2067         return 1;
2068 }
2069
2070 /**
2071  * Get the rank of a row.
2072  *
2073  * \param t An open osl table.
2074  * \param r The row to get the rank of.
2075  * \param col_num The number of an rbtree column.
2076  * \param rank Result pointer.
2077  *
2078  * The rank is, by definition, the position of the row in the linear order
2079  * determined by an inorder tree walk of the rbtree associated with column
2080  * number \a col_num of \a table.
2081  *
2082  * \return Positive on success, negative on errors.
2083  *
2084  * \sa osl_get_nth_row().
2085  */
2086 int osl_get_rank(const struct osl_table *t, struct osl_row *r,
2087                 unsigned col_num, unsigned *rank)
2088 {
2089         struct osl_object obj;
2090         struct osl_column *col;
2091         struct rb_node *node;
2092         int ret = check_rbtree_col(t, col_num, &col);
2093
2094         if (ret < 0)
2095                 return ret;
2096         ret = osl_get_object(t, r, col_num, &obj);
2097         if (ret < 0)
2098                 return ret;
2099         ret = search_rbtree(&obj, t, col_num, &node, NULL);
2100         if (ret < 0)
2101                 return ret;
2102         ret = rb_rank(node, rank);
2103         if (ret < 0)
2104                 return -E_BAD_ROW;
2105         return 1;
2106 }
2107
2108 /**
2109  * Get the row with n-th greatest value.
2110  *
2111  * \param t Pointer to an open osl table.
2112  * \param col_num The column number.
2113  * \param n The rank of the desired row.
2114  * \param result Row is returned here.
2115  *
2116  * Retrieve the n-th order statistic with respect to the compare function
2117  * of the rbtree column \a col_num. In other words, get that row with
2118  * \a n th greatest value in column \a col_num. It's an error if
2119  * \a col_num is not a rbtree column, or if \a n is larger than the
2120  * number of rows in the table.
2121  *
2122  * \return Positive on success, negative on errors. Possible errors:
2123  * \p E_BAD_TABLE, \p E_BAD_STORAGE_FLAGS, \p E_RB_KEY_NOT_FOUND.
2124  *
2125  * \sa osl_storage_flags, osl_compare_func, osl_get_row(),
2126  * osl_rbtree_last_row(), osl_rbtree_first_row(), osl_get_rank().
2127  */
2128 int osl_get_nth_row(const struct osl_table *t, unsigned col_num,
2129                 unsigned n, struct osl_row **result)
2130 {
2131         struct osl_column *col;
2132         struct rb_node *node;
2133         int ret = check_rbtree_col(t, col_num, &col);
2134
2135         if (ret < 0)
2136                 return ret;
2137         node = rb_nth(col->rbtree.rb_node, n);
2138         if (!node)
2139                 return -E_RB_KEY_NOT_FOUND;
2140         *result = get_row_pointer(node, col->rbtree_num);
2141         return 1;
2142 }
2143
2144 /**
2145  * Get the row corresponding to the smallest rbtree node of a column.
2146  *
2147  * \param t An open rbtree table.
2148  * \param col_num The number of the rbtree column.
2149  * \param result A pointer to the first row is returned here.
2150  *
2151  * The rbtree node of the smallest object (with respect to the corresponding
2152  * compare function) is selected and the row containing this object is
2153  * returned. It is an error if \a col_num refers to a column without an
2154  * associated rbtree.
2155  *
2156  * \return Positive on success, negative on errors.
2157  *
2158  * \sa osl_get_nth_row(), osl_rbtree_last_row().
2159  */
2160 int osl_rbtree_first_row(const struct osl_table *t, unsigned col_num,
2161                 struct osl_row **result)
2162 {
2163         return osl_get_nth_row(t, col_num, 1, result);
2164 }
2165
2166 /**
2167  * Get the row corresponding to the greatest rbtree node of a column.
2168  *
2169  * \param t The same meaning as in \p osl_rbtree_first_row().
2170  * \param col_num The same meaning as in \p osl_rbtree_first_row().
2171  * \param result The same meaning as in \p osl_rbtree_first_row().
2172  *
2173  * This function works just like osl_rbtree_first_row(), the only difference
2174  * is that the row containing the greatest rather than the smallest object is
2175  * returned.
2176  *
2177  * \return Positive on success, negative on errors.
2178  *
2179  * \sa osl_get_nth_row(), osl_rbtree_first_row().
2180  */
2181 int osl_rbtree_last_row(const struct osl_table *t, unsigned col_num,
2182                 struct osl_row **result)
2183 {
2184         unsigned num_rows;
2185         int ret = osl_get_num_rows(t, &num_rows);
2186
2187         if (ret < 0)
2188                 return ret;
2189         return osl_get_nth_row(t, col_num, num_rows, result);
2190 }