tcfdb.c

Tokyo Cabinet的Tokyo Cabinet 是一个DBM的实现。这里的数据库由一系列key-value对的记录构成。key和value都可以是任意长度的字节序列,既可以是二进制也可以是字符

/*************************************************************************************************
 * The fixed-length database API of Tokyo Cabinet
 *                                                      Copyright (C) 2006-2009 Mikio Hirabayashi
 * This file is part of Tokyo Cabinet.
 * Tokyo Cabinet is free software; you can redistribute it and/or modify it under the terms of
 * the GNU Lesser General Public License as published by the Free Software Foundation; either
 * version 2.1 of the License or any later version.  Tokyo Cabinet is distributed in the hope
 * that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
 * License for more details.
 * You should have received a copy of the GNU Lesser General Public License along with Tokyo
 * Cabinet; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307 USA.
 *************************************************************************************************/


#include "tcutil.h"
#include "tcfdb.h"
#include "myconf.h"

#define FDBFILEMODE    00644             // permission of created files
#define FDBIOBUFSIZ    8192              // size of an I/O buffer

#define FDBMAGICDATA   "ToKyO CaBiNeT"   // magic data for identification
#define FDBHEADSIZ     256               // size of the reagion of the header
#define FDBTYPEOFF     32                // offset of the region for the database type
#define FDBFLAGSOFF    33                // offset of the region for the additional flags
#define FDBRNUMOFF     48                // offset of the region for the record number
#define FDBFSIZOFF     56                // offset of the region for the file size
#define FDBWIDTHOFF    64                // offset of the region for the record width
#define FDBLIMSIZOFF   72                // offset of the region for the limit size
#define FDBMINOFF      80                // offset of the region for the minimum ID offset
#define FDBMAXOFF      88                // offset of the region for the maximum ID offset
#define FDBOPAQUEOFF   128               // offset of the region for the opaque field

#define FDBDEFWIDTH    255               // default value width
#define FDBDEFLIMSIZ   (256*(1LL<<20))   // default limit size
#define FDBRMTXNUM     127               // number of record mutexes
#define FDBTRUNCALW    256               // number of record for truncate allowance
#define FDBIDARYUNIT   2048              // size of ID array allocation unit
#define FDBWALSUFFIX   "wal"             // suffix of write ahead logging file

enum {                                   // enumeration for duplication behavior
  FDBPDOVER,                             // overwrite an existing value
  FDBPDKEEP,                             // keep the existing value
  FDBPDCAT,                              // concatenate values
  FDBPDADDINT,                           // add an integer
  FDBPDADDDBL,                           // add a real number
  FDBPDPROC                              // process by a callback function
};

typedef struct {                         // type of structure for a duplication callback
  TCPDPROC proc;                         // function pointer
  void *op;                              // opaque pointer
} FDBPDPROCOP;


/* private macros */
#define FDBLOCKMETHOD(TC_fdb, TC_wr) \
  ((TC_fdb)->mmtx ? tcfdblockmethod((TC_fdb), (TC_wr)) : true)
#define FDBUNLOCKMETHOD(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdbunlockmethod(TC_fdb) : true)
#define FDBLOCKATTR(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdblockattr(TC_fdb) : true)
#define FDBUNLOCKATTR(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdbunlockattr(TC_fdb) : true)
#define FDBLOCKRECORD(TC_fdb, TC_wr, TC_id) \
  ((TC_fdb)->mmtx ? tcfdblockrecord((TC_fdb), (TC_wr), (TC_id)) : true)
#define FDBUNLOCKRECORD(TC_fdb, TC_id) \
  ((TC_fdb)->mmtx ? tcfdbunlockrecord((TC_fdb), (TC_id)) : true)
#define FDBLOCKALLRECORDS(TC_fdb, TC_wr) \
  ((TC_fdb)->mmtx ? tcfdblockallrecords((TC_fdb), (TC_wr)) : true)
#define FDBUNLOCKALLRECORDS(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdbunlockallrecords(TC_fdb) : true)
#define FDBLOCKWAL(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdblockwal(TC_fdb) : true)
#define FDBUNLOCKWAL(TC_fdb) \
  ((TC_fdb)->mmtx ? tcfdbunlockwal(TC_fdb) : true)
#define FDBTHREADYIELD(TC_fdb) \
  do { if((TC_fdb)->mmtx) sched_yield(); } while(false)


/* private function prototypes */
static void tcfdbdumpmeta(TCFDB *fdb, char *hbuf);
static void tcfdbloadmeta(TCFDB *fdb, const char *hbuf);
static void tcfdbclear(TCFDB *fdb);
static void tcfdbsetflag(TCFDB *fdb, int flag, bool sign);
static bool tcfdbwalinit(TCFDB *fdb);
static bool tcfdbwalwrite(TCFDB *fdb, uint64_t off, int64_t size);
static int tcfdbwalrestore(TCFDB *fdb, const char *path);
static bool tcfdbwalremove(TCFDB *fdb, const char *path);
static bool tcfdbopenimpl(TCFDB *fdb, const char *path, int omode);
static bool tcfdbcloseimpl(TCFDB *fdb);
static int64_t tcfdbprevid(TCFDB *fdb, int64_t id);
static int64_t tcfdbnextid(TCFDB *fdb, int64_t id);
static bool tcfdbputimpl(TCFDB *fdb, int64_t id, const void *vbuf, int vsiz, int dmode);
static bool tcfdboutimpl(TCFDB *fdb, int64_t id);
static const void *tcfdbgetimpl(TCFDB *fdb, int64_t id, int *sp);
static bool tcfdbiterinitimpl(TCFDB *fdb);
static uint64_t tcfdbiternextimpl(TCFDB *fdb);
static uint64_t *tcfdbrangeimpl(TCFDB *fdb, int64_t lower, int64_t upper, int max, int *np);
static bool tcfdboptimizeimpl(TCFDB *fdb, int32_t width, int64_t limsiz);
static bool tcfdbvanishimpl(TCFDB *fdb);
static bool tcfdbcopyimpl(TCFDB *fdb, const char *path);
static bool tcfdbforeachimpl(TCFDB *fdb, TCITER iter, void *op);
static bool tcfdblockmethod(TCFDB *fdb, bool wr);
static bool tcfdbunlockmethod(TCFDB *fdb);
static bool tcfdblockattr(TCFDB *fdb);
static bool tcfdbunlockattr(TCFDB *fdb);
static bool tcfdblockrecord(TCFDB *fdb, bool wr, uint64_t id);
static bool tcfdbunlockrecord(TCFDB *fdb, uint64_t id);
static bool tcfdblockallrecords(TCFDB *fdb, bool wr);
static bool tcfdbunlockallrecords(TCFDB *fdb);
static bool tcfdblockwal(TCFDB *fdb);
static bool tcfdbunlockwal(TCFDB *fdb);


/* debugging function prototypes */
void tcfdbprintmeta(TCFDB *fdb);



/*************************************************************************************************
 * API
 *************************************************************************************************/


/* Get the message string corresponding to an error code. */
const char *tcfdberrmsg(int ecode){
  return tcerrmsg(ecode);
}


/* Create a fixed-length database object. */
TCFDB *tcfdbnew(void){
  TCFDB *fdb;
  TCMALLOC(fdb, sizeof(*fdb));
  tcfdbclear(fdb);
  return fdb;
}


/* Delete a fixed-length database object. */
void tcfdbdel(TCFDB *fdb){
  assert(fdb);
  if(fdb->fd >= 0) tcfdbclose(fdb);
  if(fdb->mmtx){
    pthread_key_delete(*(pthread_key_t *)fdb->eckey);
    pthread_mutex_destroy(fdb->wmtx);
    pthread_mutex_destroy(fdb->tmtx);
    for(int i = FDBRMTXNUM - 1; i >= 0; i--){
      pthread_rwlock_destroy((pthread_rwlock_t *)fdb->rmtxs + i);
    }
    pthread_mutex_destroy(fdb->amtx);
    pthread_rwlock_destroy(fdb->mmtx);
    TCFREE(fdb->eckey);
    TCFREE(fdb->wmtx);
    TCFREE(fdb->tmtx);
    TCFREE(fdb->rmtxs);
    TCFREE(fdb->amtx);
    TCFREE(fdb->mmtx);
  }
  TCFREE(fdb);
}


/* Get the last happened error code of a fixed-length database object. */
int tcfdbecode(TCFDB *fdb){
  assert(fdb);
  return fdb->mmtx ?
    (int)(intptr_t)pthread_getspecific(*(pthread_key_t *)fdb->eckey) : fdb->ecode;
}


/* Set mutual exclusion control of a fixed-length database object for threading. */
bool tcfdbsetmutex(TCFDB *fdb){
  assert(fdb);
  if(!TCUSEPTHREAD) return true;
  if(fdb->mmtx || fdb->fd >= 0){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    return false;
  }
  TCMALLOC(fdb->mmtx, sizeof(pthread_rwlock_t));
  TCMALLOC(fdb->amtx, sizeof(pthread_mutex_t));
  TCMALLOC(fdb->rmtxs, sizeof(pthread_rwlock_t) * FDBRMTXNUM);
  TCMALLOC(fdb->tmtx, sizeof(pthread_mutex_t));
  TCMALLOC(fdb->wmtx, sizeof(pthread_mutex_t));
  TCMALLOC(fdb->eckey, sizeof(pthread_key_t));
  bool err = false;
  if(pthread_rwlock_init(fdb->mmtx, NULL) != 0) err = true;
  if(pthread_mutex_init(fdb->amtx, NULL) != 0) err = true;
  for(int i = 0; i < FDBRMTXNUM; i++){
    if(pthread_rwlock_init((pthread_rwlock_t *)fdb->rmtxs + i, NULL) != 0) err = true;
  }
  if(pthread_mutex_init(fdb->tmtx, NULL) != 0) err = true;
  if(pthread_mutex_init(fdb->wmtx, NULL) != 0) err = true;
  if(pthread_key_create(fdb->eckey, NULL) != 0) err = true;
  if(err){
    TCFREE(fdb->eckey);
    TCFREE(fdb->wmtx);
    TCFREE(fdb->tmtx);
    TCFREE(fdb->rmtxs);
    TCFREE(fdb->amtx);
    TCFREE(fdb->mmtx);
    fdb->eckey = NULL;
    fdb->wmtx = NULL;
    fdb->tmtx = NULL;
    fdb->rmtxs = NULL;
    fdb->amtx = NULL;
    fdb->mmtx = NULL;
    return false;
  }
  return true;
}


/* Set the tuning parameters of a fixed-length database object. */
bool tcfdbtune(TCFDB *fdb, int32_t width, int64_t limsiz){
  assert(fdb);
  if(fdb->fd >= 0){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    return false;
  }
  fdb->width = (width > 0) ? width : FDBDEFWIDTH;
  fdb->limsiz = (limsiz > 0) ? limsiz : FDBDEFLIMSIZ;
  if(fdb->limsiz < FDBHEADSIZ + fdb->width + sizeof(uint32_t))
    fdb->limsiz = FDBHEADSIZ + fdb->width + sizeof(uint32_t);
  fdb->limsiz = tcpagealign(fdb->limsiz);
  return true;
}


/* Open a database file and connect a fixed-length database object. */
bool tcfdbopen(TCFDB *fdb, const char *path, int omode){
  assert(fdb && path);
  if(!FDBLOCKMETHOD(fdb, true)) return false;
  if(fdb->fd >= 0){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdbopenimpl(fdb, path, omode);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Close a fixed-length database object. */
bool tcfdbclose(TCFDB *fdb){
  assert(fdb);
  if(!FDBLOCKMETHOD(fdb, true)) return false;
  if(fdb->fd < 0){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdbcloseimpl(fdb);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Store a record into a fixed-length database object. */
bool tcfdbput(TCFDB *fdb, int64_t id, const void *vbuf, int vsiz){
  assert(fdb && vbuf && vsiz >= 0);
  if(!FDBLOCKMETHOD(fdb, id < 1)) return false;
  if(fdb->fd < 0 || !(fdb->omode & FDBOWRITER)){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(id == FDBIDMIN){
    id = fdb->min;
  } else if(id == FDBIDPREV){
    id = fdb->min - 1;
  } else if(id == FDBIDMAX){
    id = fdb->max;
  } else if(id == FDBIDNEXT){
    id = fdb->max + 1;
  }
  if(id < 1 || id > fdb->limid){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(!FDBLOCKRECORD(fdb, true, id)){
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdbputimpl(fdb, id, vbuf, vsiz, FDBPDOVER);
  FDBUNLOCKRECORD(fdb, id);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Store a record with a decimal key into a fixed-length database object. */
bool tcfdbput2(TCFDB *fdb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(fdb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  return tcfdbput(fdb, tcfdbkeytoid(kbuf, ksiz), vbuf, vsiz);
}


/* Store a string record with a decimal key into a fixed-length database object. */
bool tcfdbput3(TCFDB *fdb, const char *kstr, const void *vstr){
  assert(fdb && kstr && vstr);
  return tcfdbput(fdb, tcfdbkeytoid(kstr, strlen(kstr)), vstr, strlen(vstr));
}


/* Store a new record into a fixed-length database object. */
bool tcfdbputkeep(TCFDB *fdb, int64_t id, const void *vbuf, int vsiz){
  assert(fdb && vbuf && vsiz >= 0);
  if(!FDBLOCKMETHOD(fdb, id < 1)) return false;
  if(fdb->fd < 0 || !(fdb->omode & FDBOWRITER)){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(id == FDBIDMIN){
    id = fdb->min;
  } else if(id == FDBIDPREV){
    id = fdb->min - 1;
  } else if(id == FDBIDMAX){
    id = fdb->max;
  } else if(id == FDBIDNEXT){
    id = fdb->max + 1;
  }
  if(id < 1 || id > fdb->limid){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(!FDBLOCKRECORD(fdb, true, id)){
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdbputimpl(fdb, id, vbuf, vsiz, FDBPDKEEP);
  FDBUNLOCKRECORD(fdb, id);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Store a new record with a decimal key into a fixed-length database object. */
bool tcfdbputkeep2(TCFDB *fdb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(fdb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  return tcfdbputkeep(fdb, tcfdbkeytoid(kbuf, ksiz), vbuf, vsiz);
}


/* Store a new string record with a decimal key into a fixed-length database object. */
bool tcfdbputkeep3(TCFDB *fdb, const char *kstr, const void *vstr){
  assert(fdb && kstr && vstr);
  return tcfdbputkeep(fdb, tcfdbkeytoid(kstr, strlen(kstr)), vstr, strlen(vstr));
}


/* Concatenate a value at the end of the existing record in a fixed-length database object. */
bool tcfdbputcat(TCFDB *fdb, int64_t id, const void *vbuf, int vsiz){
  assert(fdb && vbuf && vsiz >= 0);
  if(!FDBLOCKMETHOD(fdb, id < 1)) return false;
  if(fdb->fd < 0 || !(fdb->omode & FDBOWRITER)){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(id == FDBIDMIN){
    id = fdb->min;
  } else if(id == FDBIDPREV){
    id = fdb->min - 1;
  } else if(id == FDBIDMAX){
    id = fdb->max;
  } else if(id == FDBIDNEXT){
    id = fdb->max + 1;
  }
  if(id < 1 || id > fdb->limid){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(!FDBLOCKRECORD(fdb, true, id)){
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdbputimpl(fdb, id, vbuf, vsiz, FDBPDCAT);
  FDBUNLOCKRECORD(fdb, id);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Concatenate a value with a decimal key in a fixed-length database object. */
bool tcfdbputcat2(TCFDB *fdb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(fdb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  return tcfdbputcat(fdb, tcfdbkeytoid(kbuf, ksiz), vbuf, vsiz);
}


/* Concatenate a string value with a decimal key in a fixed-length database object. */
bool tcfdbputcat3(TCFDB *fdb, const char *kstr, const void *vstr){
  assert(fdb && kstr && vstr);
  return tcfdbputcat(fdb, tcfdbkeytoid(kstr, strlen(kstr)), vstr, strlen(vstr));
}


/* Remove a record of a fixed-length database object. */
bool tcfdbout(TCFDB *fdb, int64_t id){
  assert(fdb);
  if(!FDBLOCKMETHOD(fdb, true)) return false;
  if(fdb->fd < 0 || !(fdb->omode & FDBOWRITER)){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(id == FDBIDMIN){
    id = fdb->min;
  } else if(id == FDBIDMAX){
    id = fdb->max;
  }
  if(id < 1 || id > fdb->limid){
    tcfdbsetecode(fdb, TCEINVALID, __FILE__, __LINE__, __func__);
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  if(!FDBLOCKRECORD(fdb, true, id)){
    FDBUNLOCKMETHOD(fdb);
    return false;
  }
  bool rv = tcfdboutimpl(fdb, id);
  FDBUNLOCKRECORD(fdb, id);
  FDBUNLOCKMETHOD(fdb);
  return rv;
}


/* Remove a record with a decimal key of a fixed-length database object. */
bool tcfdbout2(TCFDB *fdb, const void *kbuf, int ksiz){
  assert(fdb && kbuf && ksiz >= 0);
  return tcfdbout(fdb, tcfdbkeytoid(kbuf, ksiz));
}


/* Remove a string record with a decimal key of a fixed-length database object. */
bool tcfdbout3(TCFDB *fdb, const char *kstr){