My previous article was the first in a series on developing a custom database system. To recap briefly, we're developing a key-value embedded database. In the previous article I argued in favour of developing a new database system and presented a project design along with the requirements and logical model. Just to remind you, Figure 1 presents an overview of the LhimkDB database system. In this article we will look at the lowest database layer - a data access layer called UDB (Unordered Database). UDB is of course part of LhimkDB, but both the tasks it performs and its implementation are flexible enough to warrant using UDB as an independent library providing persistent data storage.

Figure 1. Overview of LhimkDB architecture

LhimkDB is developed in a language of my devising called Lhimk. It is so similar to C/C++ that there is little point in learning it separately, let alone writing tutorials for it. Any language features found in Lhimk and not found in C/C++ are described in the inset Lhimk or within the article itself. Anyone who doesn't fancy using Lhimk (which I can well understand) can very easily rewrite all Lhimk code in C++. Lhimk code can also be linked to from within C/C++ code.

Design goals

To start with, let's formulate our expectations of UDB. The basic idea behind UDB is that of a more flexible heap memory model, with added persistence and transaction support. We'll start by declaring an interface, or the set of routines exported by UDB. Here are the necessary methods:

• pid_t put(char d[[]]) - loads data into persistent storage and returns a PID corresponding to the location of the data.

• putAt(pid_t pid, char d[[]]) - loads data into persistent storage at the specified PID, effectively altering an existing entry.

• char at(pid_t pid)[[]] - reads data from the specified PID (the location returned by put()).

• remove(pid_t pid) - removes data from the specified PID (the PID can later be reused for other data).

• beginTransaction() - starts a write transaction.

• commitTransaction() - finalises a write transaction and commits it to disk. Note that no rollback routine is required, as aborting a transaction is simply a matter of not calling commitTransaction().

And that's basically the whole API. The char[[]] type is equivalent to (char* buf, uint len) - a resizable array with a specified initial size. This array declaration syntax is one of the few constructs specific to Lhimk and not found in C/C++. (Note that the peculiar-looking notation char fun()[[]] simply means that fun returns char[[]] and is actually valid C syntax).

Transaction support means that we're allowing one writer thread and many reader threads, with the readers only seeing data that has been added to the database and committed with a call to commitTransaction(). If a thread requests data from the same PID twice while that PID is being written, it should get back the same data for both calls, meaning that only changes committed prior to the start of the read operation are visible to the reader. It is also necessary for the database to automatically recover consistency after a system failure, so that once it is up and running again it will only contain values entered by committed transactions. In practice this means that any transactions interrupted by a system failure without being committed should be rolled back.

Finally, the database system should stipulate no limitations on database size in memory (up to the system memory limit), database size on disk (up to the maximum permissible file size) or transaction size.

Listing 1. Minimalist implementation of the MFile class

 
 
class MFile
{
   MFilePrivate* _p;
 
   int init(char* path, char* ext)
   {
      this->_p = MFilePrivate::new();
      // field initialisation skipped
      this->open();
      return 0;
   };
 
   constructor new(char* path, char* ext)
   {
      MFile* this = lhcnew(MFile*);
      this->init(path, ext);
      return this;
   };
   
   void free()
   {
      if (this->canWrite())
      {
         this->finishWrite();
      }
      /* The entire block has to be locked because other
 threads can alter usage at the same time */
      LOCK_USAGE;
      if (this->_p->usage<0)
      {
         ERROR("USAGE<0", ERROR_FATAL);
      }
      if (this->_p->usage > 0)
      {
         this->_p->usage--;
         UNLOCK_USAGE;
      }
      else
      {
         this->close();
         UNLOCK_USAGE;
         free(this->_p);
         free(this);
      }
   };
      
   MFile *clone()
   {
      LOCK_USAGE;
      this->_mfile_p->usage++;
      UNLOCK_USAGE;
      return this;
   };
 
   int open()
   {
      this->_p->needsSync = 0;
      this->_p->file->open("a+");
      this->_p->maxPos = this->_p->file->size();
      return 0;
   };
   
   int close()
   {
      this->_p->file->munmap(this->_p->buf, this->_p->maxPos);
      this->_p->file->close();
      return 0;
   };
 
   int _truncate_(unsigned long long off)
   {
      // increment file by GROWSTEP
      unsigned long long noff = (off/GROWSTEP +1) * GROWSTEP;
      if (this->_p->buf != 0x0 && this->_p->maxPos>0)
      {
         this->_p->file->munmap(this->_p->buf, this->_p->maxPos);
      }
      this->_p->file->ftruncate(noff);
      this->_p->maxPos = noff;
      this->_p->buf = this->_p->file->mmap(pthis->buf, pthis->maxPos, "rw", "s", 0);
      return 0;
   };
   
   int truncate(unsigned long long off)
   {
      LOCK;
      this->_truncate_(off);
      UNLOCK;
   };
   
   int write(unsigned long long pos, void *buf, unsigned int size)
   {
      if (!this->canWrite())
      {
         ERROR("MFile::write, thread can not write", ERROR_MINOR);
      }
      LOCK;
      if (pos+size >= this->_p->maxPos)
      {
         // truncate without locking
         this->_truncate_(pos+size);
      }
      memmove(this->_p->buf + pos, buf, size);
      this->_p->needsSync = 1;
      UNLOCK;
      return 0;
   };
      
   int read(unsigned long long pos, void *buf, unsigned int size)
   {
       if (pos + size > this->_p->maxPos)
       {
          ERROR("Read out of range", ERROR_MINOR);
       }
      LOCK;
      memmove(buf, this->_p->buf + pos, size);
      UNLOCK;
      return 0;
   };
   
   int sync()
   {
      this->_p->file->msync(this->_p->buf, this->_p->maxPos);
      return 0;
   };
   
   int commit()
   {
      return this->sync();
   };
   
   int startWrite()
   {
      LOCK_WRITE;
      this->_p->write_thread = lpthThread::self();
   };
   
   int finishWrite()
   {
      this->_p->write_thread = 0x0;
      UNLOCK_WRITE;
   };
   
   int canWrite()
   {
      int ret = 0;
      if (((void*)this->_p->write_thread) != 0x0)
      {
         ret = this->_p->write_thread == lpthThread::self();
      }
      return ret;
   };
   
};