tchdb.c

高性能嵌入式数据库在高并发的环境下使用最好是64位系统比较好

    hdb->ba64[bidx] = TCHTOILL(llnum);
  } else {
    uint32_t lnum = off >> hdb->apow;
    hdb->ba32[bidx] = TCHTOIL(lnum);
  }
}


/* Load the free block pool from the file.
   The return value is true if successful, else, it is false. */
static bool tchdbsavefbp(TCHDB *hdb){
  assert(hdb);
  if(hdb->fbpnum > (hdb->fbpmax >> 1)){
    tchdbfbpmerge(hdb);
  } else if(hdb->fbpnum > 1){
    tcfbpsortbyoff(hdb->fbpool, hdb->fbpnum);
  }
  int bsiz = hdb->frec - hdb->msiz;
  char *buf;
  TCMALLOC(buf, bsiz);
  char *wp = buf;
  HDBFB *cur = hdb->fbpool;
  HDBFB *end = cur + hdb->fbpnum;
  uint64_t base = 0;
  bsiz -= sizeof(HDBFB) + sizeof(uint8_t) + sizeof(uint8_t);
  while(cur < end && bsiz > 0){
    uint64_t noff = cur->off >> hdb->apow;
    int step;
    uint64_t llnum = noff - base;
    TCSETVNUMBUF64(step, wp, llnum);
    wp += step;
    bsiz -= step;
    uint32_t lnum = cur->rsiz >> hdb->apow;
    TCSETVNUMBUF(step, wp, lnum);
    wp += step;
    bsiz -= step;
    base = noff;
    cur++;
  }
  *(wp++) = '\0';
  *(wp++) = '\0';
  if(!tcseekwrite(hdb, hdb->msiz, buf, wp - buf)){
    TCFREE(buf);
    return false;
  }
  TCFREE(buf);
  return true;
}


/* Save the free block pool into the file.
   The return value is true if successful, else, it is false. */
static bool tchdbloadfbp(TCHDB *hdb){
  int bsiz = hdb->frec - hdb->msiz;
  char *buf;
  TCMALLOC(buf, bsiz);
  if(!tcseekread(hdb, hdb->msiz, buf, bsiz)){
    TCFREE(buf);
    return false;
  }
  const char *rp = buf;
  HDBFB *cur = hdb->fbpool;
  HDBFB *end = cur + hdb->fbpmax;
  uint64_t base = 0;
  while(cur < end && *rp != '\0'){
    int step;
    uint64_t llnum;
    TCREADVNUMBUF64(rp, llnum, step);
    base += llnum << hdb->apow;
    cur->off = base;
    rp += step;
    uint32_t lnum;
    TCREADVNUMBUF(rp, lnum, step);
    cur->rsiz = lnum << hdb->apow;
    rp += step;
    cur++;
  }
  hdb->fbpnum = cur - (HDBFB *)hdb->fbpool;
  TCFREE(buf);
  return true;
}


/* Sort the free block pool by offset.
   `fbpool' specifies the free block pool.
   `fbpnum' specifies the number of blocks. */
static void tcfbpsortbyoff(HDBFB *fbpool, int fbpnum){
  assert(fbpool && fbpnum >= 0);
  fbpnum--;
  int bottom = fbpnum / 2 + 1;
  int top = fbpnum;
  while(bottom > 0){
    bottom--;
    int mybot = bottom;
    int i = mybot * 2;
    while(i <= top){
      if(i < top && fbpool[i+1].off > fbpool[i].off) i++;
      if(fbpool[mybot].off >= fbpool[i].off) break;
      HDBFB swap = fbpool[mybot];
      fbpool[mybot] = fbpool[i];
      fbpool[i] = swap;
      mybot = i;
      i = mybot * 2;
    }
  }
  while(top > 0){
    HDBFB swap = fbpool[0];
    fbpool[0] = fbpool[top];
    fbpool[top] = swap;
    top--;
    int mybot = bottom;
    int i = mybot * 2;
    while(i <= top){
      if(i < top && fbpool[i+1].off > fbpool[i].off) i++;
      if(fbpool[mybot].off >= fbpool[i].off) break;
      swap = fbpool[mybot];
      fbpool[mybot] = fbpool[i];
      fbpool[i] = swap;
      mybot = i;
      i = mybot * 2;
    }
  }
}


/* Sort the free block pool by record size.
   `fbpool' specifies the free block pool.
   `fbpnum' specifies the number of blocks. */
static void tcfbpsortbyrsiz(HDBFB *fbpool, int fbpnum){
  assert(fbpool && fbpnum >= 0);
  fbpnum--;
  int bottom = fbpnum / 2 + 1;
  int top = fbpnum;
  while(bottom > 0){
    bottom--;
    int mybot = bottom;
    int i = mybot * 2;
    while(i <= top){
      if(i < top && fbpool[i+1].rsiz > fbpool[i].rsiz) i++;
      if(fbpool[mybot].rsiz >= fbpool[i].rsiz) break;
      HDBFB swap = fbpool[mybot];
      fbpool[mybot] = fbpool[i];
      fbpool[i] = swap;
      mybot = i;
      i = mybot * 2;
    }
  }
  while(top > 0){
    HDBFB swap = fbpool[0];
    fbpool[0] = fbpool[top];
    fbpool[top] = swap;
    top--;
    int mybot = bottom;
    int i = mybot * 2;
    while(i <= top){
      if(i < top && fbpool[i+1].rsiz > fbpool[i].rsiz) i++;
      if(fbpool[mybot].rsiz >= fbpool[i].rsiz) break;
      swap = fbpool[mybot];
      fbpool[mybot] = fbpool[i];
      fbpool[i] = swap;
      mybot = i;
      i = mybot * 2;
    }
  }
}


/* Merge successive records in the free block pool.
   `hdb' specifies the hash database object. */
static void tchdbfbpmerge(TCHDB *hdb){
  assert(hdb);
  TCDODEBUG(hdb->cnt_mergefbp++);
  int32_t onum = hdb->fbpnum;
  tcfbpsortbyoff(hdb->fbpool, hdb->fbpnum);
  HDBFB *wp = hdb->fbpool;;
  HDBFB *cur = wp;
  HDBFB *end = wp + hdb->fbpnum - 1;
  while(cur < end){
    if(cur->off > 0){
      HDBFB *next = cur + 1;
      if(cur->off + cur->rsiz == next->off){
        if(hdb->iter == next->off) hdb->iter += next->rsiz;
        cur->rsiz += next->rsiz;
        next->off = 0;
      }
      *(wp++) = *cur;
    }
    cur++;
  }
  if(end->off > 0) *(wp++) = *end;
  hdb->fbpnum = wp - (HDBFB *)hdb->fbpool;
  hdb->fbpmis = (hdb->fbpnum < onum) ? 0 : hdb->fbpnum * -2;
}


/* Insert a block into the free block pool.
   `hdb' specifies the hash database object.
   `off' specifies the offset of the block.
   `rsiz' specifies the size of the block. */
static void tchdbfbpinsert(TCHDB *hdb, uint64_t off, uint32_t rsiz){
  assert(hdb && off > 0 && rsiz > 0);
  TCDODEBUG(hdb->cnt_insertfbp++);
  if(hdb->fpow < 1) return;
  HDBFB *pv = hdb->fbpool;
  if(hdb->fbpnum >= hdb->fbpmax){
    tchdbfbpmerge(hdb);
    tcfbpsortbyrsiz(hdb->fbpool, hdb->fbpnum);
    if(hdb->fbpnum >= hdb->fbpmax){
      TCDODEBUG(hdb->cnt_reducefbp++);
      int32_t dnum = (hdb->fbpmax >> 2) + 1;
      memmove(pv, pv + dnum, (hdb->fbpnum - dnum) * sizeof(*pv));
      hdb->fbpnum -= dnum;
    }
    hdb->fbpmis = 0;
  }
  pv = pv + hdb->fbpnum;
  pv->off = off;
  pv->rsiz = rsiz;
  hdb->fbpnum++;
}


/* Search the free block pool for the minimum region.
   `hdb' specifies the hash database object.
   `rec' specifies the record object to be stored.
   The return value is true if successful, else, it is false. */
static bool tchdbfbpsearch(TCHDB *hdb, TCHREC *rec){
  assert(hdb && rec);
  TCDODEBUG(hdb->cnt_searchfbp++);
  if(hdb->fbpnum < 1){
    rec->off = hdb->fsiz;
    rec->rsiz = 0;
    return true;
  }
  uint32_t rsiz = rec->rsiz;
  HDBFB *pv = hdb->fbpool;
  HDBFB *ep = pv + hdb->fbpnum;
  while(pv < ep){
    if(pv->rsiz > rsiz){
      if(pv->rsiz > rsiz * 2){
        uint32_t psiz = tchdbpadsize(hdb, pv->off + rsiz);
        uint64_t noff = pv->off + rsiz + psiz;
        if(pv->rsiz >= (noff - pv->off) * 2){
          TCDODEBUG(hdb->cnt_dividefbp++);
          rec->off = pv->off;
          rec->rsiz = noff - pv->off;
          pv->off = noff;
          pv->rsiz -= rec->rsiz;
          return tchdbwritefb(hdb, pv->off, pv->rsiz);
        }
      }
      rec->off = pv->off;
      rec->rsiz = pv->rsiz;
      ep--;
      pv->off = ep->off;
      pv->rsiz = ep->rsiz;
      hdb->fbpnum--;
      return true;
    }
    pv++;
  }
  rec->off = hdb->fsiz;
  rec->rsiz = 0;
  hdb->fbpmis++;
  if(hdb->fbpmis >= HDBFBPMGFREQ){
    tchdbfbpmerge(hdb);
    tcfbpsortbyrsiz(hdb->fbpool, hdb->fbpnum);
  }
  return true;
}


/* Splice the trailing free block
   `hdb' specifies the hash database object.
   `rec' specifies the record object to be stored.
   `nsiz' specifies the needed size.
   The return value is whether splicing succeeded or not. */
static bool tchdbfbpsplice(TCHDB *hdb, TCHREC *rec, uint32_t nsiz){
  assert(hdb && rec && nsiz > 0);
  if(hdb->fbpnum < 1) return false;
  uint64_t off = rec->off + rec->rsiz;
  uint32_t rsiz = rec->rsiz;
  HDBFB *pv = hdb->fbpool;
  HDBFB *ep = pv + hdb->fbpnum;
  while(pv < ep){
    if(pv->off == off && rsiz + pv->rsiz >= nsiz){
      if(hdb->iter == pv->off) hdb->iter += pv->rsiz;
      rec->rsiz += pv->rsiz;
      ep--;
      pv->off = ep->off;
      pv->rsiz = ep->rsiz;
      hdb->fbpnum--;
      return true;
    }
    pv++;
  }
  return false;
}


/* Write a free block into the file.
   `hdb' specifies the hash database object.
   `off' specifies the offset of the block.
   `rsiz' specifies the size of the block.
   The return value is true if successful, else, it is false. */
static bool tchdbwritefb(TCHDB *hdb, uint64_t off, uint32_t rsiz){
  assert(hdb && off > 0 && rsiz > 0);
  char rbuf[HDBMAXHSIZ];
  char *wp = rbuf;
  *(uint8_t *)(wp++) = HDBMAGICFB;
  uint32_t lnum = TCHTOIL(rsiz);
  memcpy(wp, &lnum, sizeof(lnum));
  wp += sizeof(lnum);
  if(!tcseekwrite(hdb, off, rbuf, wp - rbuf)) return false;
  return true;
}


/* Write a record into the file.
   `hdb' specifies the hash database object.
   `rec' specifies the record object.
   `bidx' specifies the index of the bucket.
   `entoff' specifies the offset of the tree entry.
   The return value is true if successful, else, it is false. */
static bool tchdbwriterec(TCHDB *hdb, TCHREC *rec, uint64_t bidx, off_t entoff){
  assert(hdb && rec);
  TCDODEBUG(hdb->cnt_writerec++);
  char stack[HDBIOBUFSIZ];
  int bsiz = (rec->rsiz > 0) ? rec->rsiz : HDBMAXHSIZ + rec->ksiz + rec->vsiz + hdb->align;
  char *rbuf;
  if(bsiz <= HDBIOBUFSIZ){
    rbuf = stack;
  } else {
    TCMALLOC(rbuf, bsiz);
  }
  char *wp = rbuf;
  *(uint8_t *)(wp++) = HDBMAGICREC;
  *(uint8_t *)(wp++) = rec->hash;
  if(hdb->ba64){
    uint64_t llnum;
    llnum = rec->left >> hdb->apow;
    llnum = TCHTOILL(llnum);
    memcpy(wp, &llnum, sizeof(llnum));
    wp += sizeof(llnum);
    llnum = rec->right >> hdb->apow;
    llnum = TCHTOILL(llnum);
    memcpy(wp, &llnum, sizeof(llnum));
    wp += sizeof(llnum);
  } else {
    uint32_t lnum;
    lnum = rec->left >> hdb->apow;
    lnum = TCHTOIL(lnum);
    memcpy(wp, &lnum, sizeof(lnum));
    wp += sizeof(lnum);
    lnum = rec->right >> hdb->apow;
    lnum = TCHTOIL(lnum);
    memcpy(wp, &lnum, sizeof(lnum));
    wp += sizeof(lnum);
  }
  uint16_t snum;
  char *pwp = wp;
  wp += sizeof(snum);
  int step;
  TCSETVNUMBUF(step, wp, rec->ksiz);
  wp += step;
  TCSETVNUMBUF(step, wp, rec->vsiz);
  wp += step;
  int32_t hsiz = wp - rbuf;
  int32_t rsiz = hsiz + rec->ksiz + rec->vsiz;
  int32_t finc = 0;
  if(rec->rsiz < 1){
    uint16_t psiz = tchdbpadsize(hdb, hdb->fsiz + rsiz);
    rec->rsiz = rsiz + psiz;
    rec->psiz = psiz;
    finc = rec->rsiz;
  } else if(rsiz > rec->rsiz){
    if(rbuf != stack) TCFREE(rbuf);
    if(tchdbfbpsplice(hdb, rec, rsiz)){
      TCDODEBUG(hdb->cnt_splicefbp++);
      return tchdbwriterec(hdb, rec, bidx, entoff);
    }
    TCDODEBUG(hdb->cnt_moverec++);
    if(!tchdbwritefb(hdb, rec->off, rec->rsiz)) return false;
    tchdbfbpinsert(hdb, rec->off, rec->rsiz);
    rec->rsiz = rsiz;
    if(!tchdbfbpsearch(hdb, rec)) return false;
    return tchdbwriterec(hdb, rec, bidx, entoff);
  } else {
    TCDODEBUG(hdb->cnt_reuserec++);
    uint32_t psiz = rec->rsiz - rsiz;
    if(psiz > UINT16_MAX){
      TCDODEBUG(hdb->cnt_dividefbp++);
      psiz = tchdbpadsize(hdb, rec->off + rsiz);
      uint64_t noff = rec->off + rsiz + psiz;
      uint32_t nsiz = rec->rsiz - rsiz - psiz;
      rec->rsiz = noff - rec->off;
      rec->psiz = psiz;
      if(!tchdbwritefb(hdb, noff, nsiz)) return false;
      tchdbfbpinsert(hdb, noff, nsiz);
    }
    rec->psiz = psiz;
  }
  snum = rec->psiz;
  snum = TCHTOIS(snum);
  memcpy(pwp, &snum, sizeof(snum));
  rsiz = rec->rsiz;
  rsiz -= hsiz;
  memcpy(wp, rec->kbuf, rec->ksiz);
  wp += rec->ksiz;
  rsiz -= rec->ksiz;
  memcpy(wp, rec->vbuf, rec->vsiz);
  wp += rec->vsiz;
  rsiz -= rec->vsiz;
  memset(wp, 0, rsiz);
  if(!tcseekwrite(hdb, rec->off, rbuf, rec->rsiz)){
    if(rbuf != stack) TCFREE(rbuf);
    return false;
  }
  if(finc != 0){
    hdb->fsiz += finc;
    uint64_t lln