tcbdb.c

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

  llnum = bdb->first;
  llnum = TCHTOILL(llnum);
  memcpy(wp, &llnum, sizeof(llnum));
  wp += sizeof(llnum);
  llnum = bdb->last;
  llnum = TCHTOILL(llnum);
  memcpy(wp, &llnum, sizeof(llnum));
  wp += sizeof(llnum);
  llnum = bdb->lnum;
  llnum = TCHTOILL(llnum);
  memcpy(wp, &llnum, sizeof(llnum));
  wp += sizeof(llnum);
  llnum = bdb->nnum;
  llnum = TCHTOILL(llnum);
  memcpy(wp, &llnum, sizeof(llnum));
  wp += sizeof(llnum);
  llnum = bdb->rnum;
  llnum = TCHTOILL(llnum);
  memcpy(wp, &llnum, sizeof(llnum));
  wp += sizeof(llnum);
}


/* Deserialize meta data from the opaque field.
   `bdb' specifies the B+ tree database object. */
static void tcbdbloadmeta(TCBDB *bdb){
  const char *rp = bdb->opaque;
  uint8_t cnum = *(uint8_t *)(rp++);
  if(cnum == 0x0){
    bdb->cmp = tccmplexical;
  } else if(cnum == 0x1){
    bdb->cmp = tccmpdecimal;
  } else if(cnum == 0x2){
    bdb->cmp = tccmpint32;
  } else if(cnum == 0x3){
    bdb->cmp = tccmpint64;
  }
  rp += 7;
  uint32_t lnum;
  memcpy(&lnum, rp, sizeof(lnum));
  rp += sizeof(lnum);
  bdb->lmemb = TCITOHL(lnum);
  memcpy(&lnum, rp, sizeof(lnum));
  rp += sizeof(lnum);
  bdb->nmemb = TCITOHL(lnum);
  uint64_t llnum;
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->root = TCITOHLL(llnum);
  rp += sizeof(llnum);
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->first = TCITOHLL(llnum);
  rp += sizeof(llnum);
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->last = TCITOHLL(llnum);
  rp += sizeof(llnum);
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->lnum = TCITOHLL(llnum);
  rp += sizeof(llnum);
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->nnum = TCITOHLL(llnum);
  rp += sizeof(llnum);
  memcpy(&llnum, rp, sizeof(llnum));
  bdb->rnum = TCITOHLL(llnum);
  rp += sizeof(llnum);
}


/* Create a new leaf.
   `bdb' specifies the B+ tree database object.
   `prev' specifies the ID number of the previous leaf.
   `next' specifies the ID number of the next leaf.
   The return value is the new leaf object. */
static BDBLEAF *tcbdbleafnew(TCBDB *bdb, uint64_t prev, uint64_t next){
  assert(bdb);
  BDBLEAF lent;
  lent.id = ++bdb->lnum;
  lent.recs = tcptrlistnew2(bdb->lmemb + 1);
  lent.size = 0;
  lent.prev = prev;
  lent.next = next;
  lent.dirty = true;
  lent.dead = false;
  tcmapputkeep(bdb->leafc, &(lent.id), sizeof(lent.id), &lent, sizeof(lent));
  int rsiz;
  return (BDBLEAF *)tcmapget(bdb->leafc, &(lent.id), sizeof(lent.id), &rsiz);
}


/* Remove a leaf from the cache.
   `bdb' specifies the B+ tree database object.
   `leaf' specifies the leaf object.
   If successful, the return value is true, else, it is false. */
static bool tcbdbleafcacheout(TCBDB *bdb, BDBLEAF *leaf){
  assert(bdb && leaf);
  bool err = false;
  if(leaf->dirty && !tcbdbleafsave(bdb, leaf)) err = true;
  TCPTRLIST *recs = leaf->recs;
  int ln = TCPTRLISTNUM(recs);
  for(int i = 0; i < ln; i++){
    BDBREC *rec = TCPTRLISTVAL(recs, i);
    if(rec->rest) tclistdel(rec->rest);
    TCFREE(rec);
  }
  tcptrlistdel(recs);
  tcmapout(bdb->leafc, &(leaf->id), sizeof(leaf->id));
  return !err;
}


/* Save a leaf into the internal database.
   `bdb' specifies the B+ tree database object.
   `leaf' specifies the leaf object.
   If successful, the return value is true, else, it is false. */
static bool tcbdbleafsave(TCBDB *bdb, BDBLEAF *leaf){
  assert(bdb && leaf);
  TCDODEBUG(bdb->cnt_saveleaf++);
  TCXSTR *rbuf = tcxstrnew3(BDBPAGEBUFSIZ);
  char hbuf[(sizeof(uint64_t)+1)*3];
  char *wp = hbuf;
  uint64_t llnum;
  int step;
  llnum = leaf->prev;
  TCSETVNUMBUF64(step, wp, llnum);
  wp += step;
  llnum = leaf->next;
  TCSETVNUMBUF64(step, wp, llnum);
  wp += step;
  TCXSTRCAT(rbuf, hbuf, wp - hbuf);
  TCPTRLIST *recs = leaf->recs;
  int ln = TCPTRLISTNUM(recs);
  for(int i = 0; i < ln; i++){
    BDBREC *rec = TCPTRLISTVAL(recs, i);
    char *dbuf = (char *)rec + sizeof(*rec);
    int lnum;
    wp = hbuf;
    lnum = rec->ksiz;
    TCSETVNUMBUF(step, wp, lnum);
    wp += step;
    lnum = rec->vsiz;
    TCSETVNUMBUF(step, wp, lnum);
    wp += step;
    TCLIST *rest = rec->rest;
    int rnum = rest ? TCLISTNUM(rest) : 0;
    TCSETVNUMBUF(step, wp, rnum);
    wp += step;
    TCXSTRCAT(rbuf, hbuf, wp - hbuf);
    TCXSTRCAT(rbuf, dbuf, rec->ksiz);
    TCXSTRCAT(rbuf, dbuf + rec->ksiz + TCALIGNPAD(rec->ksiz), rec->vsiz);
    for(int j = 0; j < rnum; j++){
      const char *vbuf;
      int vsiz;
      TCLISTVAL(vbuf, rest, j, vsiz);
      TCSETVNUMBUF(step, hbuf, vsiz);
      TCXSTRCAT(rbuf, hbuf, step);
      TCXSTRCAT(rbuf, vbuf, vsiz);
    }
  }
  bool err = false;
  step = sprintf(hbuf, "%llx", (unsigned long long)leaf->id);
  if(ln < 1 && !tchdbout(bdb->hdb, hbuf, step) && tchdbecode(bdb->hdb) != TCENOREC)
    err = true;
  if(!leaf->dead && !tchdbput(bdb->hdb, hbuf, step, TCXSTRPTR(rbuf), TCXSTRSIZE(rbuf)))
    err = true;
  tcxstrdel(rbuf);
  leaf->dirty = false;
  leaf->dead = false;
  return !err;
}


/* Load a leaf from the internal database.
   `bdb' specifies the B+ tree database object.
   `id' specifies the ID number of the leaf.
   The return value is the leaf object or `NULL' on failure. */
static BDBLEAF *tcbdbleafload(TCBDB *bdb, uint64_t id){
  assert(bdb && id > 0);
  bool clk = BDBLOCKCACHE(bdb);
  int rsiz;
  BDBLEAF *leaf = (BDBLEAF *)tcmapget3(bdb->leafc, &id, sizeof(id), &rsiz);
  if(leaf){
    if(clk) BDBUNLOCKCACHE(bdb);
    return leaf;
  }
  if(clk) BDBUNLOCKCACHE(bdb);
  TCDODEBUG(bdb->cnt_loadleaf++);
  char hbuf[(sizeof(uint64_t)+1)*3];
  int step;
  step = sprintf(hbuf, "%llx", (unsigned long long)id);
  char *rbuf = NULL;
  char wbuf[BDBPAGEBUFSIZ];
  const char *rp = NULL;
  rsiz = tchdbget3(bdb->hdb, hbuf, step, wbuf, BDBPAGEBUFSIZ);
  if(rsiz < 1){
    tcbdbsetecode(bdb, TCEMISC, __FILE__, __LINE__, __func__);
    return false;
  } else if(rsiz < BDBPAGEBUFSIZ){
    rp = wbuf;
  } else {
    if(!(rbuf = tchdbget(bdb->hdb, hbuf, step, &rsiz))){
      tcbdbsetecode(bdb, TCEMISC, __FILE__, __LINE__, __func__);
      return false;
    }
    rp = rbuf;
  }
  BDBLEAF lent;
  lent.id = id;
  uint64_t llnum;
  TCREADVNUMBUF64(rp, llnum, step);
  lent.prev = llnum;
  rp += step;
  rsiz -= step;
  TCREADVNUMBUF64(rp, llnum, step);
  lent.next = llnum;
  rp += step;
  rsiz -= step;
  lent.dirty = false;
  lent.dead = false;
  lent.recs = tcptrlistnew2(bdb->lmemb + 1);
  lent.size = 0;
  bool err = false;
  while(rsiz >= 3){
    int ksiz;
    TCREADVNUMBUF(rp, ksiz, step);
    rp += step;
    rsiz -= step;
    int vsiz;
    TCREADVNUMBUF(rp, vsiz, step);
    rp += step;
    rsiz -= step;
    int rnum;
    TCREADVNUMBUF(rp, rnum, step);
    rp += step;
    rsiz -= step;
    if(rsiz < ksiz + vsiz + rnum){
      err = true;
      break;
    }
    int psiz = TCALIGNPAD(ksiz);
    BDBREC *nrec;
    TCMALLOC(nrec, sizeof(*nrec) + ksiz + psiz + vsiz + 1);
    char *dbuf = (char *)nrec + sizeof(*nrec);
    memcpy(dbuf, rp, ksiz);
    dbuf[ksiz] = '\0';
    nrec->ksiz = ksiz;
    rp += ksiz;
    rsiz -= ksiz;
    memcpy(dbuf + ksiz + psiz, rp, vsiz);
    dbuf[ksiz+psiz+vsiz] = '\0';
    nrec->vsiz = vsiz;
    rp += vsiz;
    rsiz -= vsiz;
    lent.size += ksiz;
    lent.size += vsiz;
    if(rnum > 0){
      nrec->rest = tclistnew2(rnum);
      while(rnum-- > 0 && rsiz > 0){
        TCREADVNUMBUF(rp, vsiz, step);
        rp += step;
        rsiz -= step;
        if(rsiz < vsiz){
          err = true;
          break;
        }
        TCLISTPUSH(nrec->rest, rp, vsiz);
        rp += vsiz;
        rsiz -= vsiz;
        lent.size += vsiz;
      }
    } else {
      nrec->rest = NULL;
    }
    TCPTRLISTPUSH(lent.recs, nrec);
  }
  TCFREE(rbuf);
  if(err || rsiz != 0){
    tcbdbsetecode(bdb, TCEMISC, __FILE__, __LINE__, __func__);
    return NULL;
  }
  clk = BDBLOCKCACHE(bdb);
  if(!tcmapputkeep(bdb->leafc, &(lent.id), sizeof(lent.id), &lent, sizeof(lent))){
    int ln = TCPTRLISTNUM(lent.recs);
    for(int i = 0; i < ln; i++){
      BDBREC *rec = TCPTRLISTVAL(lent.recs, i);
      if(rec->rest) tclistdel(rec->rest);
      TCFREE(rec);
    }
    tcptrlistdel(lent.recs);
  }
  leaf = (BDBLEAF *)tcmapget(bdb->leafc, &(lent.id), sizeof(lent.id), &rsiz);
  if(clk) BDBUNLOCKCACHE(bdb);
  return leaf;
}


/* Check existence of a leaf in the internal database.
   `bdb' specifies the B+ tree database object.
   `id' specifies the ID number of the leaf.
   The return value is true if the leaf exists, else, it is false. */
static bool tcbdbleafcheck(TCBDB *bdb, uint64_t id){
  assert(bdb && id > 0);
  bool clk = BDBLOCKCACHE(bdb);
  int rsiz;
  BDBLEAF *leaf = (BDBLEAF *)tcmapget(bdb->leafc, &id, sizeof(id), &rsiz);
  if(clk) BDBUNLOCKCACHE(bdb);
  if(leaf) return true;
  char hbuf[(sizeof(uint64_t)+1)*3];
  int step = sprintf(hbuf, "%llx", (unsigned long long)id);
  return tchdbvsiz(bdb->hdb, hbuf, step) > 0;
}


/* Load the historical leaf from the internal database.
   `bdb' specifies the B+ tree database object.
   `kbuf' specifies the pointer to the region of the key.
   `ksiz' specifies the size of the region of the key.
   `id' specifies the ID number of the historical leaf.
   If successful, the return value is the pointer to the leaf, else, it is `NULL'. */
static BDBLEAF *tcbdbgethistleaf(TCBDB *bdb, const char *kbuf, int ksiz, uint64_t id){
  assert(bdb && kbuf && ksiz >= 0 && id > 0);
  BDBLEAF *leaf = tcbdbleafload(bdb, id);
  if(!leaf) return NULL;
  int ln = TCPTRLISTNUM(leaf->recs);
  if(ln < 2) return NULL;
  BDBREC *rec = TCPTRLISTVAL(leaf->recs, 0);
  char *dbuf = (char *)rec + sizeof(*rec);
  int rv;
  if(bdb->cmp == tccmplexical){
    TCCMPLEXICAL(rv, kbuf, ksiz, dbuf, rec->ksiz);
  } else {
    rv = bdb->cmp(kbuf, ksiz, dbuf, rec->ksiz, bdb->cmpop);
  }
  if(rv == 0) return leaf;
  if(rv < 0) return NULL;
  rec = TCPTRLISTVAL(leaf->recs, ln - 1);
  dbuf = (char *)rec + sizeof(*rec);
  if(bdb->cmp == tccmplexical){
    TCCMPLEXICAL(rv, kbuf, ksiz, dbuf, rec->ksiz);
  } else {
    rv = bdb->cmp(kbuf, ksiz, dbuf, rec->ksiz, bdb->cmpop);
  }
  if(rv <= 0 || leaf->next < 1) return leaf;
  return NULL;
}


/* Add a record to a leaf.
   `bdb' specifies the B+ tree database object.
   `leaf' specifies the leaf object.
   `dmode' specifies behavior when the key overlaps.
   `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.
   If successful, the return value is true, else, it is false. */
static bool tcbdbleafaddrec(TCBDB *bdb, BDBLEAF *leaf, int dmode,
                            const char *kbuf, int ksiz, const char *vbuf, int vsiz){
  assert(bdb && leaf && kbuf && ksiz >= 0);
  TCCMP cmp = bdb->cmp;
  void *cmpop = bdb->cmpop;
  TCPTRLIST *recs = leaf->recs;
  int ln = TCPTRLISTNUM(recs);
  int left = 0;
  int right = ln;
  int i = (left + right) / 2;
  while(right >= left && i < ln){
    BDBREC *rec = TCPTRLISTVAL(recs, i);
    char *dbuf = (char *)rec + sizeof(*rec);
    int rv;
    if(cmp == tccmplexical){
      TCCMPLEXICAL(rv, kbuf, ksiz, dbuf, rec->ksiz);
    } else {
      rv = cmp(kbuf, ksiz, dbuf, rec->ksiz, cmpop);
    }
    if(rv == 0){
      break;
    } else if(rv <= 0){
      right = i - 1;
    } else {
      left = i + 1;
    }
    i = (left + right) / 2;
  }
  while(i < ln){
    BDBREC *rec = TCPTRLISTVAL(recs, i);
    char *dbuf = (char *)rec + sizeof(*rec);
    int rv;
    if(cmp == tccmplexical){
      TCCMPLEXICAL(rv, kbuf, ksiz, dbuf, rec->ksiz);
    } else {
      rv = cmp(kbuf, ksiz, dbuf, rec->ksiz, cmpop);
    }
    if(rv == 0){
      int psiz = TCALIGNPAD(rec->ksiz);
      BDBREC *orec = rec;
      BDBPDPROCOP *procptr;
      int nvsiz;
      char *nvbuf;
      switch(dmode){
      case BDBPDKEEP:
        tcbdbsetecode(bdb, TCEKEEP, __FILE__, __LINE__, __func__);
        return false;
      case BDBPDCAT:
        leaf->size += vsiz;
        TCREALLOC(rec, rec, sizeof(*rec) + rec->ksiz + psiz + rec->vsiz + vsiz + 1);
        if(rec != orec){
          tcptrlistover(recs, i, rec);
          dbuf = (char *)rec + sizeof(*rec);
        }
        memcpy(dbuf + rec->ksiz + psiz + rec->vsiz, vbuf, vsiz);
        rec->vsiz += vsiz;
        dbuf[rec->ksiz+psiz+rec->vsiz] = '\0';
        break;
      case BDBPDDUP:
        leaf->size += vsiz;
        if(!rec->rest) rec->rest = tclistnew2(1);
        TCLISTPUSH(rec->rest, vbuf, vsiz);
        bdb->rnum++;
        break;
      case BDBPDDUPB:
        leaf->size += vsiz;
        if(!rec->rest) rec->rest = tclistnew2(1);
        tclistunshift(rec->rest, dbuf + rec->ksiz + psiz, rec->vsiz);
        if(vsiz > rec->vsiz){
          TCREALLOC(rec, rec, sizeof(*rec) + rec->ksiz + psiz + vsiz + 1);
          if(rec != orec){
            tcptrlistover(recs, i, rec);
            dbuf = (char *)rec + sizeof(*rec);
          }
        }
        memcpy(dbuf + rec->ksiz + psiz, vbuf, vsiz);
        dbuf[rec->ksiz+psiz+vsiz] = '\0';
        rec->vsiz = vsiz;
        bdb->rnum++;
        break;
      case BDBPDADDINT:
        if(re