tcadb.c

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

        const char *kbuf;
        int ksiz;
        TCLISTVAL(kbuf, args, i, ksiz);
        int vsiz;
        const char *vbuf = tclistval(args, i + 1, &vsiz);
        if(!tctdbput2(adb->tdb, kbuf, ksiz, vbuf, vsiz)){
          err = true;
          break;
        }
      }
      if(err){
        tclistdel(rv);
        rv = NULL;
      }
    } else if(!strcmp(name, "outlist")){
      rv = tclistnew2(1);
      bool err = false;
      for(int i = 0; i < argc; i++){
        const char *kbuf;
        int ksiz;
        TCLISTVAL(kbuf, args, i, ksiz);
        if(!tctdbout(adb->tdb, kbuf, ksiz) && tctdbecode(adb->tdb) != TCENOREC){
          err = true;
          break;
        }
      }
      if(err){
        tclistdel(rv);
        rv = NULL;
      }
    } else if(!strcmp(name, "getlist")){
      rv = tclistnew2(argc);
      bool err = false;
      for(int i = 0; i < argc; i++){
        const char *kbuf;
        int ksiz;
        TCLISTVAL(kbuf, args, i, ksiz);
        int vsiz;
        char *vbuf = tctdbget2(adb->tdb, kbuf, ksiz, &vsiz);
        if(vbuf){
          TCLISTPUSH(rv, kbuf, ksiz);
          TCLISTPUSH(rv, vbuf, vsiz);
          TCFREE(vbuf);
        } else if(tctdbecode(adb->tdb) != TCENOREC){
          err = true;
        }
      }
      if(err){
        tclistdel(rv);
        rv = NULL;
      }
    } else if(!strcmp(name, "setindex")){
      rv = tclistnew2(1);
      bool err = false;
      argc--;
      for(int i = 0; i < argc; i += 2){
        const char *kbuf;
        int ksiz;
        TCLISTVAL(kbuf, args, i, ksiz);
        int vsiz;
        const char *vbuf = tclistval(args, i + 1, &vsiz);
        int type = tctdbstrtoindextype(vbuf);
        if(type >= 0){
          if(!tctdbsetindex(adb->tdb, kbuf, type)) err = true;
        } else {
          err = true;
        }
      }
      if(err){
        tclistdel(rv);
        rv = NULL;
      }
    } else if(!strcmp(name, "search")){
      bool toout = false;
      bool tocnt = false;
      TDBQRY *qry = tctdbqrynew(adb->tdb);
      TCLIST *cnames = NULL;
      for(int i = 0; i < argc; i++){
        const char *arg;
        int asiz;
        TCLISTVAL(arg, args, i, asiz);
        TCLIST *tokens = tcstrsplit2(arg, asiz);
        int tnum = TCLISTNUM(tokens);
        if(tnum > 0){
          const char *cmd = TCLISTVALPTR(tokens, 0);
          if((!strcmp(cmd, "addcond") || !strcmp(cmd, "cond")) && tnum > 3){
            const char *name = TCLISTVALPTR(tokens, 1);
            const char *opstr = TCLISTVALPTR(tokens, 2);
            const char *expr = TCLISTVALPTR(tokens, 3);
            int op = tctdbqrystrtocondop(opstr);
            if(op >= 0) tctdbqryaddcond(qry, name, op, expr);
          } else if((!strcmp(cmd, "setorder") || !strcmp(cmd, "order")) && tnum > 2){
            const char *name = TCLISTVALPTR(tokens, 1);
            const char *typestr = TCLISTVALPTR(tokens, 2);
            int type = tctdbqrystrtoordertype(typestr);
            if(type >= 0) tctdbqrysetorder(qry, name, type);
          } else if((!strcmp(cmd, "setlimit") || !strcmp(cmd, "limit") ||
                     !strcmp(cmd, "setmax") || !strcmp(cmd, "max") ) && tnum > 1){
            const char *maxstr = TCLISTVALPTR(tokens, 1);
            int max = tcatoi(maxstr);
            int skip = 0;
            if(tnum > 2){
              maxstr = TCLISTVALPTR(tokens, 2);
              skip = tcatoi(maxstr);
            }
            tctdbqrysetlimit(qry, max, skip);
          } else if(!strcmp(cmd, "get") || !strcmp(cmd, "columns")){
            if(!cnames) cnames = tclistnew();
            for(int j = 1; j < tnum; j++){
              const char *token;
              int tsiz;
              TCLISTVAL(token, tokens, j, tsiz);
              TCLISTPUSH(cnames, token, tsiz);
            }
          } else if(!strcmp(cmd, "out") || !strcmp(cmd, "remove")){
            toout = true;
          } else if(!strcmp(cmd, "count")){
            tocnt = true;
          }
        }
        tclistdel(tokens);
      }
      if(toout){
        if(cnames){
          rv = tclistnew2(1);
          void *opq[2];
          opq[0] = rv;
          opq[1] = cnames;
          if(!tctdbqryproc(qry, tcadbtdbqrygetout, opq)){
            tclistdel(rv);
            rv = NULL;
          }
        } else {
          if(tctdbqrysearchout(qry)){
            rv = tclistnew2(1);
          } else {
            rv = NULL;
          }
        }
      } else {
        rv = tctdbqrysearch(qry);
        if(cnames){
          int cnnum = TCLISTNUM(cnames);
          int rnum = TCLISTNUM(rv);
          TCLIST *nrv = tclistnew2(rnum);
          for(int i = 0; i < rnum; i++){
            const char *pkbuf;
            int pksiz;
            TCLISTVAL(pkbuf, rv, i, pksiz);
            TCMAP *cols = tctdbget(adb->tdb, pkbuf, pksiz);
            if(cols){
              tcmapput(cols, "", 0, pkbuf, pksiz);
              tcmapmove(cols, "", 0, true);
              if(cnnum > 0){
                TCMAP *ncols = tcmapnew2(cnnum + 1);
                for(int j = 0; j < cnnum; j++){
                  const char *cname;
                  int cnsiz;
                  TCLISTVAL(cname, cnames, j, cnsiz);
                  int cvsiz;
                  const char *cvalue = tcmapget(cols, cname, cnsiz, &cvsiz);
                  if(cvalue) tcmapput(ncols, cname, cnsiz, cvalue, cvsiz);
                }
                tcmapdel(cols);
                cols = ncols;
              }
              int csiz;
              char *cbuf = tcstrjoin4(cols, &csiz);
              tclistpushmalloc(nrv, cbuf, csiz);
              tcmapdel(cols);
            }
          }
          tclistdel(rv);
          rv = nrv;
        }
      }
      if(tocnt && rv){
        tclistclear(rv);
        char numbuf[TCNUMBUFSIZ];
        int len = sprintf(numbuf, "%d", tctdbqrycount(qry));
        tclistpush(rv, numbuf, len);
      }
      if(cnames) tclistdel(cnames);
      tctdbqrydel(qry);
    } else if(!strcmp(name, "genuid")){
      rv = tclistnew2(1);
      char numbuf[TCNUMBUFSIZ];
      int nsiz = sprintf(numbuf, "%lld", (long long)tctdbgenuid(adb->tdb));
      TCLISTPUSH(rv, numbuf, nsiz);
    } else {
      rv = NULL;
    }
    break;
  default:
    rv = NULL;
    break;
  }
  return rv;
}



/*************************************************************************************************
 * features for experts
 *************************************************************************************************/


/* Get the open mode of an abstract database object. */
int tcadbomode(TCADB *adb){
  assert(adb);
  return adb->omode;
}


/* Get the concrete database object of an abstract database object. */
void *tcadbreveal(TCADB *adb){
  assert(adb);
  void *rv;
  switch(adb->omode){
  case ADBOMDB:
    rv = adb->mdb;
    break;
  case ADBONDB:
    rv = adb->ndb;
    break;
  case ADBOHDB:
    rv = adb->hdb;
    break;
  case ADBOBDB:
    rv = adb->bdb;
    break;
  case ADBOFDB:
    rv = adb->fdb;
    break;
  case ADBOTDB:
    rv = adb->tdb;
    break;
  default:
    rv = NULL;
    break;
  }
  return rv;
}


/* Store a record into an abstract database object with a duplication handler. */
bool tcadbputproc(TCADB *adb, const void *kbuf, int ksiz, const char *vbuf, int vsiz,
                  TCPDPROC proc, void *op){
  assert(adb && kbuf && ksiz >= 0 && proc);
  bool err = false;
  switch(adb->omode){
  case ADBOMDB:
    if(tcmdbputproc(adb->mdb, kbuf, ksiz, vbuf, vsiz, proc, op)){
      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(tcndbputproc(adb->ndb, kbuf, ksiz, vbuf, vsiz, proc, op)){
      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(!tchdbputproc(adb->hdb, kbuf, ksiz, vbuf, vsiz, proc, op)) err = true;
    break;
  case ADBOBDB:
    if(!tcbdbputproc(adb->bdb, kbuf, ksiz, vbuf, vsiz, proc, op)) err = true;
    break;
  case ADBOFDB:
    if(!tcfdbputproc(adb->fdb, tcfdbkeytoid(kbuf, ksiz), vbuf, vsiz, proc, op)) err = true;
    break;
  case ADBOTDB:
    err = true;
    break;
  default:
    err = true;
    break;
  }
  return !err;
}


/* Process each record atomically of an abstract database object. */
bool tcadbforeach(TCADB *adb, TCITER iter, void *op){
  assert(adb && iter);
  bool rv;
  switch(adb->omode){
  case ADBOMDB:
    tcmdbforeach(adb->mdb, iter, op);
    rv = true;
    break;
  case ADBONDB:
    tcndbforeach(adb->ndb, iter, op);
    rv = true;
    break;
  case ADBOHDB:
    rv = tchdbforeach(adb->hdb, iter, op);
    break;
  case ADBOBDB:
    rv = tcbdbforeach(adb->bdb, iter, op);
    break;
  case ADBOFDB:
    rv = tcfdbforeach(adb->fdb, iter, op);
    break;
  case ADBOTDB:
    rv = tctdbforeach(adb->tdb, iter, op);
    break;
  default:
    rv = false;
    break;
  }
  return rv;
}


/* Map records of an abstract database object into another B+ tree database. */
bool tcadbmapbdb(TCADB *adb, TCLIST *keys, TCBDB *bdb, ADBMAPPROC proc, void *op, int64_t csiz){
  assert(adb && bdb && proc);
  if(csiz < 0) csiz = 256 * 1024 * 1024;
  TCLIST *recs = tclistnew2(tclmin(csiz / 64 + 256, INT_MAX / 4));
  ADBMAPBDB map;
  map.adb = adb;
  map.bdb = bdb;
  map.recs = recs;
  map.proc = proc;
  map.op = op;
  map.rsiz = 0;
  map.csiz = csiz;
  bool err = false;
  if(keys){
    int knum = TCLISTNUM(keys);
    for(int i = 0; i < knum && !err; i++){
      const char *kbuf;
      int ksiz;
      TCLISTVAL(kbuf, keys, i, ksiz);
      int vsiz;
      char *vbuf = tcadbget(adb, kbuf, ksiz, &vsiz);
      if(vbuf){
        if(!tcadbmapbdbiter(kbuf, ksiz, vbuf, vsiz, &map)) err = true;
        TCFREE(vbuf);
        if(map.rsiz > map.csiz && !tcadbmapbdbdump(&map)) err = true;
      }
      if(map.rsiz > 0 && !tcadbmapbdbdump(&map)) err = true;
    }
  } else {
    if(!tcadbforeach(adb, tcadbmapbdbiter, &map)) err = true;
  }
  if(map.rsiz > 0 && !tcadbmapbdbdump(&map)) err = true;
  tclistdel(recs);
  return !err;
}


/* Emit records generated by the mapping function into the result map. */
bool tcadbmapbdbemit(void *map, const char *kbuf, int ksiz, const char *vbuf, int vsiz){
  assert(map && kbuf && ksiz >= 0 && vbuf && vsiz >= 0);
  ADBMAPBDB *mymap = map;
  int rsiz = sizeof(ksiz) + ksiz + vsiz;
  char stack[TCNUMBUFSIZ*8];
  char *rbuf;
  if(rsiz <= sizeof(stack)){
    rbuf = stack;
  } else {
    TCMALLOC(rbuf, rsiz);
  }
  bool err = false;
  char *wp = rbuf;
  memcpy(wp, &ksiz, sizeof(ksiz));
  wp += sizeof(ksiz);
  memcpy(wp, kbuf, ksiz);
  wp += ksiz;
  memcpy(wp, vbuf, vsiz);
  tclistpush(mymap->recs, rbuf, rsiz);
  mymap->rsiz += rsiz + sizeof(TCLISTDATUM);
  if(rbuf != stack) TCFREE(rbuf);
  if(mymap->rsiz > mymap->csiz && !tcadbmapbdbdump(map)) err = true;
  return !err;
}


/* Call the mapping function for every record of an abstract database object.
   `kbuf' specifies the pointer to the region of the key.
   `ksiz' specifies the size of the region of the key.
   `vbuf' specifies the pointer to the region of the value.
   `vsiz' specifies the size of the region of the value.
   `op' specifies the pointer to the optional opaque object.
   The return value is true to continue iteration or false to stop iteration. */
static bool tcadbmapbdbiter(const void *kbuf, int ksiz, const void *vbuf, int vsiz, void *op){
  assert(kbuf && ksiz >= 0 && vbuf && vsiz >= 0 && op);
  ADBMAPBDB *map = op;
  bool err = false;
  if(!map->proc(map, kbuf, ksiz, vbuf, vsiz, map->op)) err = true;
  return !err;
}


/* Dump all cached records into the B+ tree database.
   `map' specifies the mapper object for the B+ tree database.
   The return value is true if successful, else, it is false. */
static bool tcadbmapbdbdump(ADBMAPBDB *map){
  assert(map);
  TCBDB *bdb = map->bdb;
  TCLIST *recs = map->recs;
  int rnum = TCLISTNUM(recs);
  TCCMP cmp = tcbdbcmpfunc(bdb);
  if(cmp == tccmplexical){
    tclistsortex(recs, tcadbmapreccmplexical);
  } else if(cmp == tccmpdecimal){
    tclistsortex(recs, tcadbmapreccmpdecimal);
  } else if(cmp == tccmpint32){
    tclistsortex(recs, tcadbmapreccmpint32);
  } else if(cmp == tccmpint64){
    tclistsortex(recs, tcadbmapreccmpint64);
  }
  bool err = false;
  for(int i = 0; i < rnum; i++){
    const char *rbuf;
    int rsiz;
    TCLISTVAL(rbuf, recs, i, rsiz);
    int ksiz;
    memcpy(&ksiz, rbuf, sizeof(ksiz));
    const char *kbuf = rbuf + sizeof(ksiz);
    if(!tcbdbputdup(bdb, kbuf, ksiz, kbuf + ksiz, rsiz - sizeof(ksiz) - ksiz)){
      err = true;
      break;
    }
  }
  tclistclear(recs);
  map->rsiz = 0;
  return !err;
}


/* Compare two list elements by lexical order for mapping.
   `a' specifies the pointer to one element.
   `b' specifies the pointer to the other element.
   The return value is positive if the former is big, negative if the latter is big, 0 if both
   are equivalent. */
static int tcadbmapreccmplexical(const TCLISTDATUM *a, const TCLISTDATUM *b){
  assert(a && b);
  unsigned char *ao = (unsigned char *)((TCLISTDATUM *)a)->ptr;
  unsigned char *bo = (unsigned char *)((TCLISTDATUM *)b)->ptr;
  int size = (((TCLISTDATUM *)a)->size < ((TCLISTDATUM *)b)->size) ?
    ((TCLISTDATUM *)a)->size : ((TCLISTDATUM *)b)->size;
  for(int i = sizeof(int); i < size; i++){
    if(ao[i] > bo[i]) return 1;
    if(ao[i] < bo[i]) return -1;
  }
  return ((TCLISTDATUM *)a)->size - ((TCLISTDATUM *)b)->size;
}


/* Compare two keys as decimal strings of real numbers for mapping.
   `a' specifies the pointer to one element.
   `b' specifies the pointer to the other element.
   The return value is positive if the form