tcadb.c

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

/*************************************************************************************************
 * The abstract 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 "tchdb.h"
#include "tcbdb.h"
#include "tcfdb.h"
#include "tctdb.h"
#include "tcadb.h"
#include "myconf.h"

typedef struct {                         // type of structure for mapper to B+ tree database
  TCADB *adb;                            // source database object
  TCBDB *bdb;                            // destination database object
  TCLIST *recs;                          // cached records
  int64_t rsiz;                          // total size of cached records
  int64_t csiz;                          // capacity of cached records
  ADBMAPPROC proc;                       // mapping function
  void *op;                              // opaque object for the mapping function
} ADBMAPBDB;


/* private function prototypes */
static bool tcadbmapbdbiter(const void *kbuf, int ksiz, const void *vbuf, int vsiz, void *op);
static bool tcadbmapbdbdump(ADBMAPBDB *map);
static int tcadbmapreccmplexical(const TCLISTDATUM *a, const TCLISTDATUM *b);
static int tcadbmapreccmpdecimal(const TCLISTDATUM *a, const TCLISTDATUM *b);
static int tcadbmapreccmpint32(const TCLISTDATUM *a, const TCLISTDATUM *b);
static int tcadbmapreccmpint64(const TCLISTDATUM *a, const TCLISTDATUM *b);
static int tcadbtdbqrygetout(const void *pkbuf, int pksiz, TCMAP *cols, void *op);



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


/* Create an abstract database object. */
TCADB *tcadbnew(void){
  TCADB *adb;
  TCMALLOC(adb, sizeof(*adb));
  adb->name = NULL;
  adb->omode = ADBOVOID;
  adb->mdb = NULL;
  adb->ndb = NULL;
  adb->hdb = NULL;
  adb->bdb = NULL;
  adb->fdb = NULL;
  adb->tdb = NULL;
  adb->capnum = -1;
  adb->capsiz = -1;
  adb->capcnt = 0;
  adb->cur = NULL;
  return adb;
}


/* Delete an abstract database object. */
void tcadbdel(TCADB *adb){
  assert(adb);
  if(adb->name) tcadbclose(adb);
  TCFREE(adb);
}


/* Open an abstract database. */
bool tcadbopen(TCADB *adb, const char *name){
  assert(adb && name);
  if(adb->name) return false;
  TCLIST *elems = tcstrsplit(name, "#");
  char *path = tclistshift2(elems);
  if(!path){
    tclistdel(elems);
    return false;
  }
  int64_t bnum = -1;
  int64_t capnum = -1;
  int64_t capsiz = -1;
  bool owmode = true;
  bool ocmode = true;
  bool otmode = false;
  bool onlmode = false;
  bool onbmode = false;
  int8_t apow = -1;
  int8_t fpow = -1;
  bool tlmode = false;
  bool tdmode = false;
  bool tbmode = false;
  bool ttmode = false;
  int32_t rcnum = -1;
  int64_t xmsiz = -1;
  int32_t lmemb = -1;
  int32_t nmemb = -1;
  int32_t lcnum = -1;
  int32_t ncnum = -1;
  int32_t width = -1;
  int64_t limsiz = -1;
  TCLIST *idxs = NULL;
  int ln = TCLISTNUM(elems);
  for(int i = 0; i < ln; i++){
    const char *elem = TCLISTVALPTR(elems, i);
    char *pv = strchr(elem, '=');
    if(!pv) continue;
    *(pv++) = '\0';
    if(!tcstricmp(elem, "bnum")){
      bnum = tcatoix(pv);
    } else if(!tcstricmp(elem, "capnum")){
      capnum = tcatoix(pv);
    } else if(!tcstricmp(elem, "capsiz")){
      capsiz = tcatoix(pv);
    } else if(!tcstricmp(elem, "mode")){
      owmode = strchr(pv, 'w') || strchr(pv, 'W');
      ocmode = strchr(pv, 'c') || strchr(pv, 'C');
      otmode = strchr(pv, 't') || strchr(pv, 'T');
      onlmode = strchr(pv, 'e') || strchr(pv, 'E');
      onbmode = strchr(pv, 'f') || strchr(pv, 'F');
    } else if(!tcstricmp(elem, "apow")){
      apow = tcatoix(pv);
    } else if(!tcstricmp(elem, "fpow")){
      fpow = tcatoix(pv);
    } else if(!tcstricmp(elem, "opts")){
      if(strchr(pv, 'l') || strchr(pv, 'L')) tlmode = true;
      if(strchr(pv, 'd') || strchr(pv, 'D')) tdmode = true;
      if(strchr(pv, 'b') || strchr(pv, 'B')) tbmode = true;
      if(strchr(pv, 't') || strchr(pv, 'T')) ttmode = true;
    } else if(!tcstricmp(elem, "rcnum")){
      rcnum = tcatoix(pv);
    } else if(!tcstricmp(elem, "xmsiz")){
      xmsiz = tcatoix(pv);
    } else if(!tcstricmp(elem, "lmemb")){
      lmemb = tcatoix(pv);
    } else if(!tcstricmp(elem, "nmemb")){
      nmemb = tcatoix(pv);
    } else if(!tcstricmp(elem, "lcnum")){
      lcnum = tcatoix(pv);
    } else if(!tcstricmp(elem, "ncnum")){
      ncnum = tcatoix(pv);
    } else if(!tcstricmp(elem, "width")){
      width = tcatoix(pv);
    } else if(!tcstricmp(elem, "limsiz")){
      limsiz = tcatoix(pv);
    } else if(!tcstricmp(elem, "idx")){
      if(!idxs) idxs = tclistnew();
      TCLISTPUSH(idxs, pv, strlen(pv));
    }
  }
  tclistdel(elems);
  adb->omode = ADBOVOID;
  if(!tcstricmp(path, "*")){
    adb->mdb = bnum > 0 ? tcmdbnew2(bnum) : tcmdbnew();
    adb->capnum = capnum;
    adb->capsiz = capsiz;
    adb->capcnt = 0;
    adb->omode = ADBOMDB;
  } else if(!tcstricmp(path, "+")){
    adb->ndb = tcndbnew();
    adb->capnum = capnum;
    adb->capsiz = capsiz;
    adb->capcnt = 0;
    adb->omode = ADBONDB;
  } else if(tcstribwm(path, ".tch") || tcstribwm(path, ".hdb")){
    TCHDB *hdb = tchdbnew();
    tchdbsetmutex(hdb);
    int opts = 0;
    if(tlmode) opts |= HDBTLARGE;
    if(tdmode) opts |= HDBTDEFLATE;
    if(tbmode) opts |= HDBTBZIP;
    if(ttmode) opts |= HDBTTCBS;
    tchdbtune(hdb, bnum, apow, fpow, opts);
    tchdbsetcache(hdb, rcnum);
    if(xmsiz >= 0) tchdbsetxmsiz(hdb, xmsiz);
    int omode = owmode ? HDBOWRITER : HDBOREADER;
    if(ocmode) omode |= HDBOCREAT;
    if(otmode) omode |= HDBOTRUNC;
    if(onlmode) omode |= HDBONOLCK;
    if(onbmode) omode |= HDBOLCKNB;
    if(!tchdbopen(hdb, path, omode)){
      tchdbdel(hdb);
      TCFREE(path);
      return false;
    }
    adb->hdb = hdb;
    adb->omode = ADBOHDB;
  } else if(tcstribwm(path, ".tcb") || tcstribwm(path, ".bdb")){
    TCBDB *bdb = tcbdbnew();
    tcbdbsetmutex(bdb);
    int opts = 0;
    if(tlmode) opts |= BDBTLARGE;
    if(tdmode) opts |= BDBTDEFLATE;
    if(tbmode) opts |= BDBTBZIP;
    if(ttmode) opts |= BDBTTCBS;
    tcbdbtune(bdb, lmemb, nmemb, bnum, apow, fpow, opts);
    tcbdbsetcache(bdb, lcnum, ncnum);
    if(xmsiz >= 0) tcbdbsetxmsiz(bdb, xmsiz);
    if(capnum > 0) tcbdbsetcapnum(bdb, capnum);
    int omode = owmode ? BDBOWRITER : BDBOREADER;
    if(ocmode) omode |= BDBOCREAT;
    if(otmode) omode |= BDBOTRUNC;
    if(onlmode) omode |= BDBONOLCK;
    if(onbmode) omode |= BDBOLCKNB;
    if(!tcbdbopen(bdb, path, omode)){
      tcbdbdel(bdb);
      TCFREE(path);
      return false;
    }
    adb->bdb = bdb;
    adb->cur = tcbdbcurnew(bdb);
    adb->omode = ADBOBDB;
  } else if(tcstribwm(path, ".tcf") || tcstribwm(path, ".fdb")){
    TCFDB *fdb = tcfdbnew();
    tcfdbsetmutex(fdb);
    tcfdbtune(fdb, width, limsiz);
    int omode = owmode ? FDBOWRITER : FDBOREADER;
    if(ocmode) omode |= FDBOCREAT;
    if(otmode) omode |= FDBOTRUNC;
    if(onlmode) omode |= FDBONOLCK;
    if(onbmode) omode |= FDBOLCKNB;
    if(!tcfdbopen(fdb, path, omode)){
      tcfdbdel(fdb);
      TCFREE(path);
      return false;
    }
    adb->fdb = fdb;
    adb->omode = ADBOFDB;
  } else if(tcstribwm(path, ".tct") || tcstribwm(path, ".tdb")){
    TCTDB *tdb = tctdbnew();
    tctdbsetmutex(tdb);
    int opts = 0;
    if(tlmode) opts |= TDBTLARGE;
    if(tdmode) opts |= TDBTDEFLATE;
    if(tbmode) opts |= TDBTBZIP;
    if(ttmode) opts |= TDBTTCBS;
    tctdbtune(tdb, bnum, apow, fpow, opts);
    tctdbsetcache(tdb, rcnum, lcnum, ncnum);
    if(xmsiz >= 0) tctdbsetxmsiz(tdb, xmsiz);
    int omode = owmode ? TDBOWRITER : TDBOREADER;
    if(ocmode) omode |= TDBOCREAT;
    if(otmode) omode |= TDBOTRUNC;
    if(onlmode) omode |= TDBONOLCK;
    if(onbmode) omode |= TDBOLCKNB;
    if(!tctdbopen(tdb, path, omode)){
      tctdbdel(tdb);
      TCFREE(path);
      return false;
    }
    if(idxs){
      int xnum = TCLISTNUM(idxs);
      for(int i = 0; i < xnum; i++){
        const char *expr = TCLISTVALPTR(idxs, i);
        int type = TDBITLEXICAL;
        char *pv = strchr(expr, ':');
        if(pv){
          *(pv++) = '\0';
          type = tctdbstrtoindextype(pv);
        }
        if(type >= 0) tctdbsetindex(tdb, expr, type | TDBITKEEP);
      }
    }
    adb->tdb = tdb;
    adb->omode = ADBOTDB;
  }
  if(idxs) tclistdel(idxs);
  TCFREE(path);
  if(adb->omode == ADBOVOID) return false;
  adb->name = tcstrdup(name);
  return true;
}


/* Close an abstract database object. */
bool tcadbclose(TCADB *adb){
  assert(adb);
  int err = false;
  if(!adb->name) return false;
  switch(adb->omode){
  case ADBOMDB:
    tcmdbdel(adb->mdb);
    adb->mdb = NULL;
    break;
  case ADBONDB:
    tcndbdel(adb->ndb);
    adb->ndb = NULL;
    break;
  case ADBOHDB:
    if(!tchdbclose(adb->hdb)) err = true;
    tchdbdel(adb->hdb);
    adb->hdb = NULL;
    break;
  case ADBOBDB:
    tcbdbcurdel(adb->cur);
    if(!tcbdbclose(adb->bdb)) err = true;
    tcbdbdel(adb->bdb);
    adb->bdb = NULL;
    break;
  case ADBOFDB:
    if(!tcfdbclose(adb->fdb)) err = true;
    tcfdbdel(adb->fdb);
    adb->fdb = NULL;
    break;
  case ADBOTDB:
    if(!tctdbclose(adb->tdb)) err = true;
    tctdbdel(adb->tdb);
    adb->tdb = NULL;
    break;
  default:
    err = true;
    break;
  }
  TCFREE(adb->name);
  adb->name = NULL;
  adb->omode = ADBOVOID;
  return !err;
}


/* Store a record into an abstract database object. */
bool tcadbput(TCADB *adb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(adb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  bool err = false;
  char numbuf[TCNUMBUFSIZ];
  switch(adb->omode){
  case ADBOMDB:
    if(adb->capnum > 0 || adb->capsiz > 0){
      tcmdbput3(adb->mdb, kbuf, ksiz, vbuf, vsiz);
      adb->capcnt++;
      if((adb->capcnt & 0xff) == 0){
        if(adb->capnum > 0 && tcmdbrnum(adb->mdb) > adb->capnum)
          tcmdbcutfront(adb->mdb, 0x100);
        if(adb->capsiz > 0 && tcmdbmsiz(adb->mdb) > adb->capsiz)
          tcmdbcutfront(adb->mdb, 0x200);
      }
    } else {
      tcmdbput(adb->mdb, kbuf, ksiz, vbuf, vsiz);
    }
    break;
  case ADBONDB:
    tcndbput(adb->ndb, kbuf, ksiz, vbuf, vsiz);
    if(adb->capnum > 0 || adb->capsiz > 0){
      adb->capcnt++;
      if((adb->capcnt & 0xff) == 0){
        if(adb->capnum > 0 && tcndbrnum(adb->ndb) > adb->capnum)
          tcndbcutfringe(adb->ndb, 0x100);
        if(adb->capsiz > 0 && tcndbmsiz(adb->ndb) > adb->capsiz)
          tcndbcutfringe(adb->ndb, 0x200);
      }
    }
    break;
  case ADBOHDB:
    if(!tchdbput(adb->hdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOBDB:
    if(!tcbdbput(adb->bdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOFDB:
    if(!tcfdbput2(adb->fdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOTDB:
    if(ksiz < 1){
      ksiz = sprintf(numbuf, "%lld", (long long)tctdbgenuid(adb->tdb));
      kbuf = numbuf;
    }
    if(!tctdbput2(adb->tdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  default:
    err = true;
    break;
  }
  return !err;
}


/* Store a string record into an abstract object. */
bool tcadbput2(TCADB *adb, const char *kstr, const char *vstr){
  assert(adb && kstr && vstr);
  return tcadbput(adb, kstr, strlen(kstr), vstr, strlen(vstr));
}


/* Store a new record into an abstract database object. */
bool tcadbputkeep(TCADB *adb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(adb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  bool err = false;
  char numbuf[TCNUMBUFSIZ];
  switch(adb->omode){
  case ADBOMDB:
    if(tcmdbputkeep(adb->mdb, kbuf, ksiz, vbuf, vsiz)){
      if(adb->capnum > 0 || adb->capsiz > 0){
        adb->capcnt++;
        if((adb->capcnt & 0xff) == 0){
          if(adb->capnum > 0 && tcmdbrnum(adb->mdb) > adb->capnum)
            tcmdbcutfront(adb->mdb, 0x100);
          if(adb->capsiz > 0 && tcmdbmsiz(adb->mdb) > adb->capsiz)
            tcmdbcutfront(adb->mdb, 0x200);
        }
      }
    } else {
      err = true;
    }
    break;
  case ADBONDB:
    if(tcndbputkeep(adb->ndb, kbuf, ksiz, vbuf, vsiz)){
      if(adb->capnum > 0 || adb->capsiz > 0){
        adb->capcnt++;
        if((adb->capcnt & 0xff) == 0){
          if(adb->capnum > 0 && tcndbrnum(adb->ndb) > adb->capnum)
            tcndbcutfringe(adb->ndb, 0x100);
          if(adb->capsiz > 0 && tcndbmsiz(adb->ndb) > adb->capsiz)
            tcndbcutfringe(adb->ndb, 0x200);
        }
      }
    } else {
      err = true;
    }
    break;
  case ADBOHDB:
    if(!tchdbputkeep(adb->hdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOBDB:
    if(!tcbdbputkeep(adb->bdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOFDB:
    if(!tcfdbputkeep2(adb->fdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  case ADBOTDB:
    if(ksiz < 1){
      ksiz = sprintf(numbuf, "%lld", (long long)tctdbgenuid(adb->tdb));
      kbuf = numbuf;
    }
    if(!tctdbputkeep2(adb->tdb, kbuf, ksiz, vbuf, vsiz)) err = true;
    break;
  default:
    err = true;
    break;
  }
  return !err;
}


/* Store a new string record into an abstract database object. */
bool tcadbputkeep2(TCADB *adb, const char *kstr, const char *vstr){
  assert(adb && kstr && vstr);
  return tcadbputkeep(adb, kstr, strlen(kstr), vstr, strlen(vstr));
}


/* Concatenate a value at the end of the existing record in an abstract database object. */
bool tcadbputcat(TCADB *adb, const void *kbuf, int ksiz, const void *vbuf, int vsiz){
  assert(adb && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  bool err = false;
  char numbuf[TCNUMBUFSIZ];
  switch(adb->omode){
  case ADBOMDB:
    if(adb->capnum > 0 || adb->capsiz > 0){
      tcmdbputcat3(adb->mdb, kbuf, ksiz, vbuf, vsiz);
      adb->capcnt++;
      if((adb->capcnt & 0xff) == 0){
        if(adb->capnum > 0 && tcmdbrnum(adb->mdb) > adb->capnum)
          tcmdbcutfront(adb->mdb, 0x100);
        if(adb->capsiz > 0 && tcmdbmsiz(adb->mdb) > adb->capsiz)
          tcmdbcutfront(adb->mdb, 0x200);
      }
    } else {
      tcmdbputcat(adb->mdb, kbuf, ksiz, vbuf, vsiz);
    }
    break;