db2/hash/hash_page.c

   1 /*-
   2  * See the file LICENSE for redistribution information.
   3  *
   4  * Copyright (c) 1996, 1997
   5  *      Sleepycat Software.  All rights reserved.
   6  */
   7 /*
   8  * Copyright (c) 1990, 1993, 1994
   9  *      Margo Seltzer.  All rights reserved.
  10  */
  11 /*
  12  * Copyright (c) 1990, 1993, 1994
  13  *      The Regents of the University of California.  All rights reserved.
  14  *
  15  * This code is derived from software contributed to Berkeley by
  16  * Margo Seltzer.
  17  *
  18  * Redistribution and use in source and binary forms, with or without
  19  * modification, are permitted provided that the following conditions
  20  * are met:
  21  * 1. Redistributions of source code must retain the above copyright
  22  *    notice, this list of conditions and the following disclaimer.
  23  * 2. Redistributions in binary form must reproduce the above copyright
  24  *    notice, this list of conditions and the following disclaimer in the
  25  *    documentation and/or other materials provided with the distribution.
  26  * 3. All advertising materials mentioning features or use of this software
  27  *    must display the following acknowledgement:
  28  *      This product includes software developed by the University of
  29  *      California, Berkeley and its contributors.
  30  * 4. Neither the name of the University nor the names of its contributors
  31  *    may be used to endorse or promote products derived from this software
  32  *    without specific prior written permission.
  33  *
  34  * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
  35  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  36  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  37  * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  38  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  39  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  40  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  41  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  42  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  43  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  44  * SUCH DAMAGE.
  45  */
  46
  47 #include "config.h"
  48
  49 #ifndef lint
  50 static const char sccsid[] = "@(#)hash_page.c   10.18 (Sleepycat) 8/21/97";
  51 #endif /* not lint */
  52
  53
  54 /*
  55  * PACKAGE:  hashing
  56  *
  57  * DESCRIPTION:
  58  *      Page manipulation for hashing package.
  59  *
  60  * ROUTINES:
  61  *
  62  * External
  63  *      __get_page
  64  *      __add_ovflpage
  65  *      __overflow_page
  66  * Internal
  67  *      open_temp
  68  */
  69
  70 #ifndef NO_SYSTEM_INCLUDES
  71 #include <sys/types.h>
  72
  73 #include <errno.h>
  74 #include <stdio.h>
  75 #include <stdlib.h>
  76 #include <string.h>
  77 #include <unistd.h>
  78 #endif
  79
  80 #include "db_int.h"
  81 #include "db_page.h"
  82 #include "db_swap.h"
  83 #include "hash.h"
  84
  85 static int __ham_lock_bucket __P((DB *, HASH_CURSOR *, db_lockmode_t));
  86
  87 #ifdef DEBUG_SLOW
  88 static void      account_page(HTAB *, db_pgno_t, int);
  89 #endif
  90
  91 /*
  92  * PUBLIC: int __ham_item __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
  93  */
  94 int
  95 __ham_item(hashp, cursorp, mode)
  96         HTAB *hashp;
  97         HASH_CURSOR *cursorp;
  98         db_lockmode_t mode;
  99 {
 100         db_pgno_t next_pgno;
 101         int ret;
 102
 103         if (F_ISSET(cursorp, H_DELETED))
 104                 return (EINVAL);
 105         F_CLR(cursorp, H_OK | H_NOMORE);
 106
 107         /* Check if we need to get a page for this cursor. */
 108         if ((ret = __ham_get_cpage(hashp, cursorp, mode)) != 0)
 109                 return (ret);
 110
 111         /* Check if we are looking for space in which to insert an item. */
 112         if (cursorp->seek_size && cursorp->seek_found_page == PGNO_INVALID
 113             && cursorp->seek_size < P_FREESPACE(cursorp->pagep))
 114                 cursorp->seek_found_page = cursorp->pgno;
 115
 116         /* Check if we need to go on to the next page. */
 117         if (F_ISSET(cursorp, H_ISDUP) && cursorp->dpgno == PGNO_INVALID)
 118                 /*
 119                  * ISDUP is set, and offset is at the beginning of the datum.
 120                  * We need to grab the length of the datum, then set the datum
 121                  * pointer to be the beginning of the datum.
 122                  */
 123                 memcpy(&cursorp->dup_len,
 124                     H_PAIRDATA(cursorp->pagep, cursorp->bndx)->data +
 125                     cursorp->dup_off, sizeof(db_indx_t));
 126         else if (F_ISSET(cursorp, H_ISDUP)) {
 127                 /* Make sure we're not about to run off the page. */
 128                 if (cursorp->dpagep == NULL && (ret = __ham_get_page(hashp->dbp,
 129                     cursorp->dpgno, &cursorp->dpagep)) != 0)
 130                         return (ret);
 131
 132                 if (cursorp->dndx >= NUM_ENT(cursorp->dpagep)) {
 133                         if (NEXT_PGNO(cursorp->dpagep) == PGNO_INVALID) {
 134                                 if ((ret = __ham_put_page(hashp->dbp,
 135                                     cursorp->dpagep, 0)) != 0)
 136                                         return (ret);
 137                                 F_CLR(cursorp, H_ISDUP);
 138                                 cursorp->dpagep = NULL;
 139                                 cursorp->dpgno = PGNO_INVALID;
 140                                 cursorp->dndx = NDX_INVALID;
 141                                 cursorp->bndx++;
 142                         } else if ((ret = __ham_next_cpage(hashp, cursorp,
 143                             NEXT_PGNO(cursorp->dpagep), 0, H_ISDUP)) != 0)
 144                                 return (ret);
 145                 }
 146         }
 147
 148         if (cursorp->bndx >= (db_indx_t)H_NUMPAIRS(cursorp->pagep)) {
 149                 /* Fetch next page. */
 150                 if (NEXT_PGNO(cursorp->pagep) == PGNO_INVALID) {
 151                         F_SET(cursorp, H_NOMORE);
 152                         if (cursorp->dpagep != NULL &&
 153                             (ret = __ham_put_page(hashp->dbp,
 154                             cursorp->dpagep, 0)) != 0)
 155                                 return (ret);
 156                         cursorp->dpgno = PGNO_INVALID;
 157                         return (DB_NOTFOUND);
 158                 }
 159                 next_pgno = NEXT_PGNO(cursorp->pagep);
 160                 cursorp->bndx = 0;
 161                 if ((ret = __ham_next_cpage(hashp,
 162                     cursorp, next_pgno, 0, 0)) != 0)
 163                         return (ret);
 164         }
 165
 166         F_SET(cursorp, H_OK);
 167         return (0);
 168 }
 169
 170 /*
 171  * PUBLIC: int __ham_item_reset __P((HTAB *, HASH_CURSOR *));
 172  */
 173 int
 174 __ham_item_reset(hashp, cursorp)
 175         HTAB *hashp;
 176         HASH_CURSOR *cursorp;
 177 {
 178         int ret;
 179
 180         if (cursorp->pagep)
 181                 ret = __ham_put_page(hashp->dbp, cursorp->pagep, 0);
 182         else
 183                 ret = 0;
 184
 185         __ham_item_init(cursorp);
 186         return (ret);
 187 }
 188
 189 /*
 190  * PUBLIC: void __ham_item_init __P((HASH_CURSOR *));
 191  */
 192 void
 193 __ham_item_init(cursorp)
 194         HASH_CURSOR *cursorp;
 195 {
 196         cursorp->pagep = NULL;
 197         cursorp->bucket = BUCKET_INVALID;
 198         cursorp->lock = 0;
 199         cursorp->bndx = NDX_INVALID;
 200         cursorp->pgno = PGNO_INVALID;
 201         cursorp->dpgno = PGNO_INVALID;
 202         cursorp->dndx = NDX_INVALID;
 203         cursorp->dpagep = NULL;
 204         cursorp->flags = 0;
 205         cursorp->seek_size = 0;
 206         cursorp->seek_found_page = PGNO_INVALID;
 207 }
 208
 209 /*
 210  * PUBLIC: int __ham_item_done __P((HTAB *, HASH_CURSOR *, int));
 211  */
 212 int
 213 __ham_item_done(hashp, cursorp, dirty)
 214         HTAB *hashp;
 215         HASH_CURSOR *cursorp;
 216         int dirty;
 217 {
 218         int ret, t_ret;
 219
 220         t_ret = ret = 0;
 221
 222         if (cursorp->pagep)
 223                 ret = __ham_put_page(hashp->dbp, cursorp->pagep,
 224                     dirty && cursorp->dpagep == NULL);
 225         cursorp->pagep = NULL;
 226
 227         if (cursorp->dpagep)
 228                 t_ret = __ham_put_page(hashp->dbp, cursorp->dpagep, dirty);
 229         cursorp->dpagep = NULL;
 230
 231         if (ret == 0 && t_ret != 0)
 232                 ret = t_ret;
 233
 234         /*
 235          * If we are running with transactions, then we must
 236          * not relinquish locks explicitly.
 237          */
 238         if (cursorp->lock && hashp->dbp->txn == NULL)
 239             t_ret = lock_put(hashp->dbp->dbenv->lk_info, cursorp->lock);
 240         cursorp->lock = 0;
 241
 242
 243         /*
 244          * We don't throw out the page number since we might want to
 245          * continue getting on this page.
 246          */
 247         return (ret != 0 ? ret : t_ret);
 248 }
 249
 250 /*
 251  * Returns the last item in a bucket.
 252  *
 253  * PUBLIC: int __ham_item_last __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
 254  */
 255 int
 256 __ham_item_last(hashp, cursorp, mode)
 257         HTAB *hashp;
 258         HASH_CURSOR *cursorp;
 259         db_lockmode_t mode;
 260 {
 261         int ret;
 262
 263         if ((ret = __ham_item_reset(hashp, cursorp)) != 0)
 264                 return (ret);
 265
 266         cursorp->bucket = hashp->hdr->max_bucket;
 267         F_SET(cursorp, H_OK);
 268         return (__ham_item_prev(hashp, cursorp, mode));
 269 }
 270 /*
 271  * PUBLIC: int __ham_item_first __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
 272  */
 273 int
 274 __ham_item_first(hashp, cursorp, mode)
 275         HTAB *hashp;
 276         HASH_CURSOR *cursorp;
 277         db_lockmode_t mode;
 278 {
 279         int ret;
 280
 281         if ((ret = __ham_item_reset(hashp, cursorp)) != 0)
 282                 return (ret);
 283         F_SET(cursorp, H_OK);
 284         cursorp->bucket = 0;
 285         return (__ham_item_next(hashp, cursorp, mode));
 286 }
 287
 288 /*
 289  * Returns a pointer to key/data pair on a page.  In the case of bigkeys,
 290  * just returns the page number and index of the bigkey pointer pair.
 291  *
 292  * PUBLIC: int __ham_item_prev __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
 293  */
 294 int
 295 __ham_item_prev(hashp, cursorp, mode)
 296         HTAB *hashp;
 297         HASH_CURSOR *cursorp;
 298         db_lockmode_t mode;
 299 {
 300         db_pgno_t next_pgno;
 301         int ret;
 302
 303         /*
 304          * There are N cases for backing up in a hash file.
 305          * Case 1: In the middle of a page, no duplicates, just dec the index.
 306          * Case 2: In the middle of a duplicate set, back up one.
 307          * Case 3: At the beginning of a duplicate set, get out of set and
 308          *      back up to next key.
 309          * Case 4: At the beginning of a page; go to previous page.
 310          * Case 5: At the beginning of a bucket; go to prev bucket.
 311          */
 312         F_CLR(cursorp, H_OK | H_NOMORE | H_DELETED);
 313
 314         /*
 315          * First handle the duplicates.  Either you'll get the key here
 316          * or you'll exit the duplicate set and drop into the code below
 317          * to handle backing up through keys.
 318          */
 319         if (F_ISSET(cursorp, H_ISDUP)) {
 320                 if (cursorp->dpgno == PGNO_INVALID) {
 321                         /* Duplicates are on-page. */
 322                         if (cursorp->dup_off != 0)
 323                                 if ((ret = __ham_get_cpage(hashp,
 324                                     cursorp, mode)) != 0)
 325                                         return (ret);
 326                                 else {
 327                                         HASH_CURSOR *h;
 328                                         h = cursorp;
 329                                         memcpy(&h->dup_len,
 330                                             H_PAIRDATA(h->pagep, h->bndx)->data
 331                                             + h->dup_off - sizeof(db_indx_t),
 332                                             sizeof(db_indx_t));
 333                                         cursorp->dup_off -=
 334                                             DUP_SIZE(cursorp->dup_len);
 335                                         cursorp->dndx--;
 336                                         return (__ham_item(hashp,
 337                                             cursorp, mode));
 338                                 }
 339                 } else if (cursorp->dndx > 0) { /* Duplicates are off-page. */
 340                         cursorp->dndx--;
 341                         return (__ham_item(hashp, cursorp, mode));
 342                 } else if ((ret = __ham_get_cpage(hashp, cursorp, mode)) != 0)
 343                         return (ret);
 344                 else if (PREV_PGNO(cursorp->dpagep) == PGNO_INVALID) {
 345                         F_CLR(cursorp, H_ISDUP); /* End of dups */
 346                         cursorp->dpgno = PGNO_INVALID;
 347                         if (cursorp->dpagep != NULL)
 348                                 (void)__ham_put_page(hashp->dbp,
 349                                     cursorp->dpagep, 0);
 350                         cursorp->dpagep = NULL;
 351                 } else if ((ret = __ham_next_cpage(hashp, cursorp,
 352                     PREV_PGNO(cursorp->dpagep), 0, H_ISDUP)) != 0)
 353                         return (ret);
 354                 else {
 355                         cursorp->dndx = NUM_ENT(cursorp->pagep) - 1;
 356                         return (__ham_item(hashp, cursorp, mode));
 357                 }
 358         }
 359
 360         /*
 361          * If we get here, we are not in a duplicate set, and just need
 362          * to back up the cursor.  There are still three cases:
 363          * midpage, beginning of page, beginning of bucket.
 364          */
 365
 366         if (cursorp->bndx == 0) {               /* Beginning of page. */
 367                 if ((ret = __ham_get_cpage(hashp, cursorp, mode)) != 0)
 368                         return (ret);
 369                 cursorp->pgno = PREV_PGNO(cursorp->pagep);
 370                 if (cursorp->pgno == PGNO_INVALID) {
 371                         /* Beginning of bucket. */
 372                         F_SET(cursorp, H_NOMORE);
 373                         return (DB_NOTFOUND);
 374                 } else if ((ret = __ham_next_cpage(hashp,
 375                     cursorp, cursorp->pgno, 0, 0)) != 0)
 376                         return (ret);
 377                 else
 378                         cursorp->bndx = H_NUMPAIRS(cursorp->pagep);
 379         }
 380
 381         /*
 382          * Either we've got the cursor set up to be decremented, or we
 383          * have to find the end of a bucket.
 384          */
 385         if (cursorp->bndx == NDX_INVALID) {
 386                 if (cursorp->pagep == NULL)
 387                         next_pgno = BUCKET_TO_PAGE(hashp, cursorp->bucket);
 388                 else
 389                         goto got_page;
 390
 391                 do {
 392                         if ((ret = __ham_next_cpage(hashp,
 393                             cursorp, next_pgno, 0, 0)) != 0)
 394                                 return (ret);
 395 got_page:               next_pgno = NEXT_PGNO(cursorp->pagep);
 396                         cursorp->bndx = H_NUMPAIRS(cursorp->pagep);
 397                 } while (next_pgno != PGNO_INVALID);
 398
 399                 if (cursorp->bndx == 0) {
 400                         /* Bucket was empty. */
 401                         F_SET(cursorp, H_NOMORE);
 402                         return (DB_NOTFOUND);
 403                 }
 404         }
 405
 406         cursorp->bndx--;
 407
 408         return (__ham_item(hashp, cursorp, mode));
 409 }
 410
 411 /*
 412  * Sets the cursor to the next key/data pair on a page.
 413  *
 414  * PUBLIC: int __ham_item_next __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
 415  */
 416 int
 417 __ham_item_next(hashp, cursorp, mode)
 418         HTAB *hashp;
 419         HASH_CURSOR *cursorp;
 420         db_lockmode_t mode;
 421 {
 422         /*
 423          * Deleted on-page duplicates are a weird case. If we delete the last
 424          * one, then our cursor is at the very end of a duplicate set and
 425          * we actually need to go on to the next key.
 426          */
 427         if (F_ISSET(cursorp, H_DELETED)) {
 428                 if (cursorp->bndx != NDX_INVALID &&
 429                     F_ISSET(cursorp, H_ISDUP) &&
 430                     cursorp->dpgno == PGNO_INVALID &&
 431                     cursorp->dup_tlen == cursorp->dup_off) {
 432                         F_CLR(cursorp, H_ISDUP);
 433                         cursorp->dpgno = PGNO_INVALID;
 434                         cursorp->bndx++;
 435                 }
 436                 F_CLR(cursorp, H_DELETED);
 437         } else if (cursorp->bndx == NDX_INVALID) {
 438                 cursorp->bndx = 0;
 439                 cursorp->dpgno = PGNO_INVALID;
 440                 F_CLR(cursorp, H_ISDUP);
 441         } else if (F_ISSET(cursorp, H_ISDUP) && cursorp->dpgno != PGNO_INVALID)
 442                 cursorp->dndx++;
 443         else if (F_ISSET(cursorp, H_ISDUP)) {
 444                 cursorp->dndx++;
 445                 cursorp->dup_off += DUP_SIZE(cursorp->dup_len);
 446                 if (cursorp->dup_off >= cursorp->dup_tlen) {
 447                         F_CLR(cursorp, H_ISDUP);
 448                         cursorp->dpgno = PGNO_INVALID;
 449                         cursorp->bndx++;
 450                 }
 451         } else
 452                 cursorp->bndx++;
 453
 454         return (__ham_item(hashp, cursorp, mode));
 455 }
 456
 457 /*
 458  * PUBLIC: void __ham_putitem __P((PAGE *p, const DBT *, int));
 459  *
 460  * This is a little bit sleazy in that we're overloading the meaning
 461  * of the H_OFFPAGE type here.  When we recover deletes, we have the
 462  * entire entry instead of having only the DBT, so we'll pass type
 463  * H_OFFPAGE to mean, "copy the whole entry" as opposed to constructing
 464  * an H_KEYDATA around it.
 465  */
 466 void
 467 __ham_putitem(p, dbt, type)
 468         PAGE *p;
 469         const DBT *dbt;
 470         int type;
 471 {
 472         u_int16_t n, off;
 473
 474         n = NUM_ENT(p);
 475
 476         /* Put the item element on the page. */
 477         if (type == H_OFFPAGE) {
 478                 off = HOFFSET(p) - dbt->size;
 479                 HOFFSET(p) = p->inp[n] = off;
 480                 memcpy(P_ENTRY(p, n), dbt->data, dbt->size);
 481         } else {
 482                 off = HOFFSET(p) - HKEYDATA_SIZE(dbt->size);
 483                 HOFFSET(p) = p->inp[n] = off;
 484                 PUT_HKEYDATA(GET_HKEYDATA(p, n), dbt->data, dbt->size, type);
 485         }
 486
 487         /* Adjust page info. */
 488         NUM_ENT(p) += 1;
 489 }
 490
 491
 492 /*
 493  * PUBLIC: int __ham_del_pair __P((HTAB *, HASH_CURSOR *));
 494  * XXX TODO: if the item is an offdup, delete the other pages and
 495  * then remove the pair. If the offpage page is 0, then you can
 496  * just remove the pair.
 497  */
 498 int
 499 __ham_del_pair(hashp, cursorp)
 500         HTAB *hashp;
 501         HASH_CURSOR *cursorp;
 502 {
 503         DBT data_dbt, key_dbt;
 504         DB_ENV *dbenv;
 505         DB_LSN new_lsn, *n_lsn;
 506         PAGE *p;
 507         db_indx_t ndx;
 508         db_pgno_t chg_pgno, pgno;
 509         int ret, tret;
 510
 511         dbenv = hashp->dbp->dbenv;
 512         ndx = cursorp->bndx;
 513         if (cursorp->pagep == NULL && (ret =
 514             __ham_get_page(hashp->dbp, cursorp->pgno, &cursorp->pagep)) != 0)
 515                 return (ret);
 516
 517         p = cursorp->pagep;
 518
 519         /*
 520          * We optimize for the normal case which is when neither the key nor
 521          * the data are large.  In this case, we write a single log record
 522          * and do the delete.  If either is large, we'll call __big_delete
 523          * to remove the big item and then update the page to remove the
 524          * entry referring to the big item.
 525          */
 526         ret = 0;
 527         if (H_PAIRKEY(p, ndx)->type == H_OFFPAGE) {
 528                 memcpy(&pgno, (u_int8_t *)GET_HOFFPAGE(p, H_KEYINDEX(ndx)) +
 529                     SSZ(HOFFPAGE, pgno), sizeof(db_pgno_t));
 530                 ret = __db_doff(hashp->dbp, pgno, __ham_del_page);
 531         }
 532
 533         if (ret == 0)
 534                 switch (H_PAIRDATA(p, ndx)->type) {
 535                 case H_OFFPAGE:
 536                         memcpy(&pgno,
 537                             (u_int8_t *)GET_HOFFPAGE(p, H_DATAINDEX(ndx)) +
 538                             SSZ(HOFFPAGE, pgno), sizeof(db_pgno_t));
 539                         ret = __db_doff(hashp->dbp, pgno, __ham_del_page);
 540                         break;
 541                 case H_OFFDUP:
 542                         memcpy(&pgno,
 543                             (u_int8_t *)GET_HOFFDUP(p, H_DATAINDEX(ndx)) +
 544                             SSZ(HOFFDUP, pgno), sizeof(db_pgno_t));
 545                         ret = __db_ddup(hashp->dbp, pgno, __ham_del_page);
 546                         break;
 547                 }
 548
 549         if (ret)
 550                 return (ret);
 551
 552         /* Now log the delete off this page. */
 553         if (DB_LOGGING(hashp->dbp)) {
 554                 key_dbt.data = P_ENTRY(p, H_KEYINDEX(ndx));
 555                 key_dbt.size =
 556                     LEN_HITEM(p, hashp->hdr->pagesize, H_KEYINDEX(ndx));
 557                 data_dbt.data = P_ENTRY(p, H_DATAINDEX(ndx));
 558                 data_dbt.size =
 559                     LEN_HITEM(p, hashp->hdr->pagesize, H_DATAINDEX(ndx));
 560
 561                 if ((ret = __ham_insdel_log(dbenv->lg_info,
 562                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, DELPAIR,
 563                     hashp->dbp->log_fileid, PGNO(p), (u_int32_t)ndx,
 564                     &LSN(p), &key_dbt, &data_dbt)) != 0)
 565                         return (ret);
 566
 567                 /* Move lsn onto page. */
 568                 LSN(p) = new_lsn;
 569         }
 570
 571         __ham_dpair(hashp->dbp, p, ndx);
 572
 573         /*
 574          * If we are locking, we will not maintain this.
 575          * XXXX perhaps we can retain incremental numbers and apply them
 576          * later.
 577          */
 578         if (!F_ISSET(hashp->dbp, DB_AM_LOCKING))
 579                 --hashp->hdr->nelem;
 580
 581         /*
 582          * Check if the page is empty.  There are two cases.  If it's
 583          * empty and it's not the first chain in the bucket (i.e., the
 584          * bucket page) then we can simply remove it. If it is the first
 585          * chain in the bucket, then we need to copy the second page into
 586          * it and remove the second page.
 587          */
 588         if (NUM_ENT(p) == 0 && PREV_PGNO(p) == PGNO_INVALID &&
 589             NEXT_PGNO(p) != PGNO_INVALID) {
 590                 PAGE *n_pagep, *nn_pagep;
 591                 db_pgno_t tmp_pgno;
 592
 593                 /*
 594                  * First page in chain is empty and we know that there
 595                  * are more pages in the chain.
 596                  * XXX Need to log this.
 597                  */
 598                 if ((ret =
 599                     __ham_get_page(hashp->dbp, NEXT_PGNO(p), &n_pagep)) != 0)
 600                         return (ret);
 601
 602                 if (NEXT_PGNO(n_pagep) != PGNO_INVALID) {
 603                         if ((ret =
 604                             __ham_get_page(hashp->dbp, NEXT_PGNO(n_pagep),
 605                             &nn_pagep)) != 0) {
 606                                 (void) __ham_put_page(hashp->dbp, n_pagep, 0);
 607                                 return (ret);
 608                         }
 609                         PREV_PGNO(nn_pagep) = PGNO(p);
 610                         (void)__ham_put_page(hashp->dbp, nn_pagep, 1);
 611                 }
 612
 613                 tmp_pgno = PGNO(p);
 614                 memcpy(p, n_pagep, hashp->hdr->pagesize);
 615                 PGNO(p) = tmp_pgno;
 616                 PREV_PGNO(p) = PGNO_INVALID;
 617
 618                 /*
 619                  * Cursor is advanced to the beginning of the next page.
 620                  */
 621                 cursorp->bndx = NDX_INVALID;
 622                 cursorp->pgno = PGNO(p);
 623                 chg_pgno = PGNO(p);
 624                 if ((ret = __ham_dirty_page(hashp, p)) != 0 ||
 625                     (ret = __ham_del_page(hashp->dbp, n_pagep)) != 0)
 626                         return (ret);
 627         } else if (NUM_ENT(p) == 0 && PREV_PGNO(p) != PGNO_INVALID) {
 628                 PAGE *n_pagep, *p_pagep;
 629
 630                 if ((ret =
 631                     __ham_get_page(hashp->dbp, PREV_PGNO(p), &p_pagep)) != 0)
 632                         return (ret);
 633
 634                 if (NEXT_PGNO(p) != PGNO_INVALID) {
 635                         if ((ret = __ham_get_page(hashp->dbp,
 636                             NEXT_PGNO(p), &n_pagep)) != 0) {
 637                                 (void)__ham_put_page(hashp->dbp, p_pagep, 0);
 638                                 return (ret);
 639                         }
 640                         n_lsn = &LSN(n_pagep);
 641                 } else {
 642                         n_pagep = NULL;
 643                         n_lsn = NULL;
 644                 }
 645
 646                 NEXT_PGNO(p_pagep) = NEXT_PGNO(p);
 647                 if (n_pagep != NULL)
 648                         PREV_PGNO(n_pagep) = PGNO(p_pagep);
 649
 650                 if (DB_LOGGING(hashp->dbp)) {
 651                         if ((ret = __ham_newpage_log(dbenv->lg_info,
 652                             (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, DELOVFL,
 653                             hashp->dbp->log_fileid, PREV_PGNO(p), &LSN(p_pagep),
 654                             PGNO(p), &LSN(p), NEXT_PGNO(p), n_lsn)) != 0)
 655                                 return (ret);
 656
 657                         /* Move lsn onto page. */
 658                         LSN(p_pagep) = new_lsn; /* Structure assignment. */
 659                         if (n_pagep)
 660                                 LSN(n_pagep) = new_lsn;
 661                         LSN(p) = new_lsn;
 662                 }
 663                 cursorp->pgno = NEXT_PGNO(p);
 664                 cursorp->bndx = 0;
 665                 /*
 666                  * Since we are about to delete the cursor page and we have
 667                  * just moved the cursor, we need to make sure that the
 668                  * old page pointer isn't left hanging around in the cursor.
 669                  */
 670                 cursorp->pagep = NULL;
 671                 chg_pgno = PGNO(p);
 672                 ret = __ham_del_page(hashp->dbp, p);
 673                 if ((tret = __ham_put_page(hashp->dbp, p_pagep, 1)) != 0 &&
 674                     ret == 0)
 675                         ret = tret;
 676                 if (n_pagep != NULL &&
 677                     (tret = __ham_put_page(hashp->dbp, n_pagep, 1)) != 0 &&
 678                     ret == 0)
 679                         ret = tret;
 680                 if (ret != 0)
 681                         return (ret);
 682         } else {
 683                 /*
 684                  * Mark item deleted so that we don't try to return it, and
 685                  * so that we update the cursor correctly on the next call
 686                  * to next.
 687                  */
 688                 F_SET(cursorp, H_DELETED);
 689                 chg_pgno = cursorp->pgno;
 690                 ret = __ham_dirty_page(hashp, p);
 691         }
 692         __ham_c_update(hashp, cursorp, chg_pgno, 0, 0, 0);
 693
 694         F_CLR(cursorp, H_OK);
 695         return (ret);
 696 }
 697 /*
 698  * PUBLIC: int __ham_replpair __P((HTAB *, HASH_CURSOR *, DBT *, u_int32_t));
 699  * Given the key data indicated by the cursor, replace part/all of it
 700  * according to the fields in the dbt.
 701  */
 702 int
 703 __ham_replpair(hashp, hcp, dbt, make_dup)
 704         HTAB *hashp;
 705         HASH_CURSOR *hcp;
 706         DBT *dbt;
 707         u_int32_t make_dup;
 708 {
 709         DBT old_dbt, tmp;
 710         DB_LSN  new_lsn;
 711         HKEYDATA *hk;
 712         u_int32_t len;
 713         int32_t change;
 714         int is_big, ret, type;
 715         u_int8_t *beg, *dest, *end, *src;
 716
 717         /*
 718          * Big item replacements are handled in generic code.
 719          * Items that fit on the current page fall into 4 classes.
 720          * 1. On-page element, same size
 721          * 2. On-page element, new is bigger (fits)
 722          * 3. On-page element, new is bigger (does not fit)
 723          * 4. On-page element, old is bigger
 724          * Numbers 1, 2, and 4 are essentially the same (and should
 725          * be the common case).  We handle case 3 as a delete and
 726          * add.
 727          */
 728
 729         /*
 730          * We need to compute the number of bytes that we are adding or
 731          * removing from the entry.  Normally, we can simply substract
 732          * the number of bytes we are replacing (dbt->dlen) from the
 733          * number of bytes we are inserting (dbt->size).  However, if
 734          * we are doing a partial put off the end of a record, then this
 735          * formula doesn't work, because we are essentially adding
 736          * new bytes.
 737          */
 738         change = dbt->size - dbt->dlen;
 739
 740         hk = H_PAIRDATA(hcp->pagep, hcp->bndx);
 741         is_big = hk->type == H_OFFPAGE;
 742
 743         if (is_big)
 744                 memcpy(&len, (u_int8_t *)hk + SSZ(HOFFPAGE, tlen),
 745                     sizeof(u_int32_t));
 746         else
 747                 len = LEN_HKEYDATA(hcp->pagep,
 748                     hashp->dbp->pgsize, H_DATAINDEX(hcp->bndx));
 749
 750         if (dbt->doff + dbt->dlen > len)
 751                 change += dbt->doff + dbt->dlen - len;
 752
 753
 754         if (change > (int)P_FREESPACE(hcp->pagep) || is_big) {
 755                 /*
 756                  * Case 3 -- two subcases.
 757                  * A. This is not really a partial operation, but an overwrite.
 758                  *    Simple del and add works.
 759                  * B. This is a partial and we need to construct the data that
 760                  *    we are really inserting (yuck).
 761                  * In both cases, we need to grab the key off the page (in
 762                  * some cases we could do this outside of this routine; for
 763                  * cleanliness we do it here.  If you happen to be on a big
 764                  * key, this could be a performance hit).
 765                  */
 766                 tmp.flags = 0;
 767                 F_SET(&tmp, DB_DBT_MALLOC | DB_DBT_INTERNAL);
 768                 if ((ret =
 769                     __db_ret(hashp->dbp, hcp->pagep, H_KEYINDEX(hcp->bndx),
 770                     &tmp, &hcp->big_key, &hcp->big_keylen)) != 0)
 771                         return (ret);
 772
 773                 type = hk->type;
 774                 if (dbt->doff == 0 && dbt->dlen == len) {
 775                         ret = __ham_del_pair(hashp, hcp);
 776                         if (ret == 0)
 777                             ret = __ham_add_el(hashp, hcp, &tmp, dbt, type);
 778                 } else {                                        /* Case B */
 779                         DBT tdata;
 780                         tdata.flags = 0;
 781                         F_SET(&tdata, DB_DBT_MALLOC | DB_DBT_INTERNAL);
 782
 783                         if ((ret = __db_ret(hashp->dbp, hcp->pagep,
 784                             H_DATAINDEX(hcp->bndx), &tdata, &hcp->big_data,
 785                             &hcp->big_datalen)) != 0)
 786                                 goto err;
 787
 788                         /* Now we can delete the item. */
 789                         if ((ret = __ham_del_pair(hashp, hcp)) != 0) {
 790                                 free(tdata.data);
 791                                 goto err;
 792                         }
 793
 794                         /* Now shift old data around to make room for new. */
 795                         if (change > 0) {
 796                                 tdata.data = (void *)
 797                                     realloc(tdata.data, tdata.size + change);
 798                                 memset((u_int8_t *)tdata.data + tdata.size,
 799                                     0, change);
 800                         }
 801                         if (tdata.data == NULL)
 802                                 return (ENOMEM);
 803                         end = (u_int8_t *)tdata.data + tdata.size;
 804
 805                         src = (u_int8_t *)tdata.data + dbt->doff + dbt->dlen;
 806                         if (src < end && tdata.size > dbt->doff + dbt->dlen) {
 807                                 len = tdata.size - dbt->doff - dbt->dlen;
 808                                 dest = src + change;
 809                                 memmove(dest, src, len);
 810                         }
 811                         memcpy((u_int8_t *)tdata.data + dbt->doff,
 812                             dbt->data, dbt->size);
 813                         tdata.size += change;
 814
 815                         /* Now add the pair. */
 816                         ret = __ham_add_el(hashp, hcp, &tmp, &tdata, type);
 817                         free(tdata.data);
 818                 }
 819 err:            free(tmp.data);
 820                 return (ret);
 821         }
 822
 823         /*
 824          * Set up pointer into existing data. Do it before the log
 825          * message so we can use it inside of the log setup.
 826          */
 827         beg = H_PAIRDATA(hcp->pagep, hcp->bndx)->data;
 828         beg += dbt->doff;
 829
 830         /*
 831          * If we are going to have to move bytes at all, figure out
 832          * all the parameters here.  Then log the call before moving
 833          * anything around.
 834          */
 835         if (DB_LOGGING(hashp->dbp)) {
 836                 old_dbt.data = beg;
 837                 old_dbt.size = dbt->dlen;
 838                 if ((ret = __ham_replace_log(hashp->dbp->dbenv->lg_info,
 839                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0,
 840                     hashp->dbp->log_fileid, PGNO(hcp->pagep),
 841                     (u_int32_t)H_DATAINDEX(hcp->bndx), &LSN(hcp->pagep),
 842                     (u_int32_t)dbt->doff, &old_dbt, dbt, make_dup)) != 0)
 843                         return (ret);
 844
 845                 LSN(hcp->pagep) = new_lsn;      /* Structure assignment. */
 846         }
 847
 848         __ham_onpage_replace(hcp->pagep, hashp->dbp->pgsize,
 849             (u_int32_t)H_DATAINDEX(hcp->bndx), (int32_t)dbt->doff, change, dbt);
 850
 851         return (0);
 852 }
 853
 854 /*
 855  * Replace data on a page with new data, possibly growing or shrinking what's
 856  * there.  This is called on two different occasions. On one (from replpair)
 857  * we are interested in changing only the data.  On the other (from recovery)
 858  * we are replacing the entire data (header and all) with a new element.  In
 859  * the latter case, the off argument is negative.
 860  * pagep: the page that we're changing
 861  * ndx: page index of the element that is growing/shrinking.
 862  * off: Offset at which we are beginning the replacement.
 863  * change: the number of bytes (+ or -) that the element is growing/shrinking.
 864  * dbt: the new data that gets written at beg.
 865  * PUBLIC: void __ham_onpage_replace __P((PAGE *, size_t, u_int32_t, int32_t,
 866  * PUBLIC:     int32_t,  DBT *));
 867  */
 868 void
 869 __ham_onpage_replace(pagep, pgsize, ndx, off, change, dbt)
 870         PAGE *pagep;
 871         size_t pgsize;
 872         u_int32_t ndx;
 873         int32_t off;
 874         int32_t change;
 875         DBT *dbt;
 876 {
 877         db_indx_t i;
 878         int32_t len;
 879         u_int8_t *src, *dest;
 880         int zero_me;
 881
 882         if (change != 0) {
 883                 zero_me = 0;
 884                 src = (u_int8_t *)(pagep) + HOFFSET(pagep);
 885                 if (off < 0)
 886                         len = pagep->inp[ndx] - HOFFSET(pagep);
 887                 else if ((u_int32_t)off >= LEN_HKEYDATA(pagep, pgsize, ndx)) {
 888                         len = GET_HKEYDATA(pagep, ndx)->data +
 889                             LEN_HKEYDATA(pagep, pgsize, ndx) - src;
 890                         zero_me = 1;
 891                 } else
 892                         len = (GET_HKEYDATA(pagep, ndx)->data + off) - src;
 893                 dest = src - change;
 894                 memmove(dest, src, len);
 895                 if (zero_me)
 896                         memset(dest + len, 0, change);
 897
 898                 /* Now update the indices. */
 899                 for (i = ndx; i < NUM_ENT(pagep); i++)
 900                         pagep->inp[i] -= change;
 901                 HOFFSET(pagep) -= change;
 902         }
 903         if (off >= 0)
 904                 memcpy(GET_HKEYDATA(pagep, ndx)->data + off,
 905                     dbt->data, dbt->size);
 906         else
 907                 memcpy(P_ENTRY(pagep, ndx), dbt->data, dbt->size);
 908 }
 909
 910 /*
 911  * PUBLIC: int __ham_split_page __P((HTAB *, u_int32_t, u_int32_t));
 912  */
 913 int
 914 __ham_split_page(hashp, obucket, nbucket)
 915         HTAB *hashp;
 916         u_int32_t obucket, nbucket;
 917 {
 918         DBT key, val, page_dbt;
 919         DB_ENV *dbenv;
 920         DB_LSN new_lsn;
 921         PAGE **pp, *old_pagep, *temp_pagep, *new_pagep;
 922         db_indx_t n;
 923         db_pgno_t bucket_pgno, next_pgno;
 924         u_int32_t big_len, len;
 925         int ret, tret;
 926         void *big_buf;
 927
 928         dbenv = hashp->dbp->dbenv;
 929         temp_pagep = old_pagep = new_pagep = NULL;
 930
 931         bucket_pgno = BUCKET_TO_PAGE(hashp, obucket);
 932         if ((ret = __ham_get_page(hashp->dbp, bucket_pgno, &old_pagep)) != 0)
 933                 return (ret);
 934         if ((ret = __ham_new_page(hashp, BUCKET_TO_PAGE(hashp, nbucket), P_HASH,
 935             &new_pagep)) != 0)
 936                 goto err;
 937
 938         temp_pagep = hashp->split_buf;
 939         memcpy(temp_pagep, old_pagep, hashp->hdr->pagesize);
 940
 941         if (DB_LOGGING(hashp->dbp)) {
 942                 page_dbt.size = hashp->hdr->pagesize;
 943                 page_dbt.data = old_pagep;
 944                 if ((ret = __ham_splitdata_log(dbenv->lg_info,
 945                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0,
 946                     hashp->dbp->log_fileid, SPLITOLD, PGNO(old_pagep),
 947                     &page_dbt, &LSN(old_pagep))) != 0)
 948                         goto err;
 949         }
 950
 951         P_INIT(old_pagep, hashp->hdr->pagesize, PGNO(old_pagep), PGNO_INVALID,
 952             PGNO_INVALID, 0, P_HASH);
 953
 954         if (DB_LOGGING(hashp->dbp))
 955                 LSN(old_pagep) = new_lsn;       /* Structure assignment. */
 956
 957         big_len = 0;
 958         big_buf = NULL;
 959         val.flags = key.flags = 0;
 960         while (temp_pagep != NULL) {
 961                 for (n = 0; n < (db_indx_t)H_NUMPAIRS(temp_pagep); n++) {
 962                         if ((ret =
 963                             __db_ret(hashp->dbp, temp_pagep, H_KEYINDEX(n),
 964                             &key, &big_buf, &big_len)) != 0)
 965                                 goto err;
 966
 967                         if (__ham_call_hash(hashp, key.data, key.size)
 968                             == obucket)
 969                                 pp = &old_pagep;
 970                         else
 971                                 pp = &new_pagep;
 972
 973                         /*
 974                          * Figure out how many bytes we need on the new
 975                          * page to store the key/data pair.
 976                          */
 977
 978                         len = LEN_HITEM(temp_pagep, hashp->hdr->pagesize,
 979                             H_DATAINDEX(n)) +
 980                             LEN_HITEM(temp_pagep, hashp->hdr->pagesize,
 981                             H_KEYINDEX(n)) +
 982                             2 * sizeof(db_indx_t);
 983
 984                         if (P_FREESPACE(*pp) < len) {
 985                                 if (DB_LOGGING(hashp->dbp)) {
 986                                         page_dbt.size = hashp->hdr->pagesize;
 987                                         page_dbt.data = *pp;
 988                                         if ((ret = __ham_splitdata_log(
 989                                             dbenv->lg_info,
 990                                             (DB_TXN *)hashp->dbp->txn,
 991                                             &new_lsn, 0,
 992                                             hashp->dbp->log_fileid, SPLITNEW,
 993                                             PGNO(*pp), &page_dbt,
 994                                             &LSN(*pp))) != 0)
 995                                                 goto err;
 996                                         LSN(*pp) = new_lsn;
 997                                 }
 998                                 if ((ret = __ham_add_ovflpage(hashp,
 999                                     *pp, 1, pp)) != 0)
1000                                         goto err;
1001                         }
1002                         __ham_copy_item(hashp, temp_pagep, H_KEYINDEX(n), *pp);
1003                         __ham_copy_item(hashp, temp_pagep, H_DATAINDEX(n), *pp);
1004                 }
1005                 next_pgno = NEXT_PGNO(temp_pagep);
1006
1007                 /* Clear temp_page; if it's a link overflow page, free it. */
1008                 if (PGNO(temp_pagep) != bucket_pgno && (ret =
1009                     __ham_del_page(hashp->dbp, temp_pagep)) != 0)
1010                         goto err;
1011
1012                 if (next_pgno == PGNO_INVALID)
1013                         temp_pagep = NULL;
1014                 else if ((ret =
1015                     __ham_get_page(hashp->dbp, next_pgno, &temp_pagep)) != 0)
1016                         goto err;
1017
1018                 if (temp_pagep != NULL && DB_LOGGING(hashp->dbp)) {
1019                         page_dbt.size = hashp->hdr->pagesize;
1020                         page_dbt.data = temp_pagep;
1021                         if ((ret = __ham_splitdata_log(dbenv->lg_info,
1022                             (DB_TXN *)hashp->dbp->txn, &new_lsn, 0,
1023                             hashp->dbp->log_fileid, SPLITOLD, PGNO(temp_pagep),
1024                             &page_dbt, &LSN(temp_pagep))) != 0)
1025                                 goto err;
1026                         LSN(temp_pagep) = new_lsn;
1027                 }
1028         }
1029         if (big_buf != NULL)
1030                 free(big_buf);
1031
1032         /*
1033          * If the original bucket spanned multiple pages, then we've got
1034          * a pointer to a page that used to be on the bucket chain.  It
1035          * should be deleted.
1036          */
1037         if (temp_pagep != NULL && PGNO(temp_pagep) != bucket_pgno &&
1038             (ret = __ham_del_page(hashp->dbp, temp_pagep)) != 0)
1039                 goto err;
1040
1041         /*
1042          * Write new buckets out.
1043          */
1044         if (DB_LOGGING(hashp->dbp)) {
1045                 page_dbt.size = hashp->hdr->pagesize;
1046                 page_dbt.data = old_pagep;
1047                 if ((ret = __ham_splitdata_log(dbenv->lg_info,
1048                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0,
1049                     hashp->dbp->log_fileid, SPLITNEW, PGNO(old_pagep),
1050                     &page_dbt, &LSN(old_pagep))) != 0)
1051                         goto err;
1052                 LSN(old_pagep) = new_lsn;
1053
1054                 page_dbt.data = new_pagep;
1055                 if ((ret = __ham_splitdata_log(dbenv->lg_info,
1056                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0,
1057                     hashp->dbp->log_fileid, SPLITNEW, PGNO(new_pagep),
1058                     &page_dbt, &LSN(new_pagep))) != 0)
1059                         goto err;
1060                 LSN(new_pagep) = new_lsn;
1061         }
1062         ret = __ham_put_page(hashp->dbp, old_pagep, 1);
1063         if ((tret = __ham_put_page(hashp->dbp, new_pagep, 1)) != 0 &&
1064             ret == 0)
1065                 ret = tret;
1066
1067 err:    if (0) {
1068                 if (old_pagep != NULL)
1069                         (void)__ham_put_page(hashp->dbp, old_pagep, 1);
1070                 if (new_pagep != NULL)
1071                         (void)__ham_put_page(hashp->dbp, new_pagep, 1);
1072                 if (temp_pagep != NULL && PGNO(temp_pagep) != bucket_pgno)
1073                         (void)__ham_put_page(hashp->dbp, temp_pagep, 1);
1074         }
1075         return (ret);
1076 }
1077
1078 /*
1079  * Add the given pair to the page.  The page in question may already be
1080  * held (i.e. it was already gotten).  If it is, then the page is passed
1081  * in via the pagep parameter.  On return, pagep will contain the page
1082  * to which we just added something.  This allows us to link overflow
1083  * pages and return the new page having correctly put the last page.
1084  *
1085  * PUBLIC: int __ham_add_el __P((HTAB *, HASH_CURSOR *, const DBT *, const DBT *,
1086  * PUBLIC:     int));
1087  */
1088 int
1089 __ham_add_el(hashp, hcp, key, val, type)
1090         HTAB *hashp;
1091         HASH_CURSOR *hcp;
1092         const DBT *key, *val;
1093         int type;
1094 {
1095         DBT *pkey, *pdata, key_dbt, data_dbt;
1096         DB_LSN new_lsn;
1097         HOFFPAGE doff, koff;
1098         db_pgno_t next_pgno;
1099         u_int32_t data_size, key_size, pairsize;
1100         int do_expand, is_keybig, is_databig, rectype, ret;
1101         int key_type, data_type;
1102
1103         do_expand = 0;
1104
1105         if (hcp->pagep == NULL && (ret = __ham_get_page(hashp->dbp,
1106             hcp->seek_found_page != PGNO_INVALID ?  hcp->seek_found_page :
1107             hcp->pgno, &hcp->pagep)) != 0)
1108                 return (ret);
1109
1110         key_size = HKEYDATA_PSIZE(key->size);
1111         data_size = HKEYDATA_PSIZE(val->size);
1112         is_keybig = ISBIG(hashp, key->size);
1113         is_databig = ISBIG(hashp, val->size);
1114         if (is_keybig)
1115                 key_size = HOFFPAGE_PSIZE;
1116         if (is_databig)
1117                 data_size = HOFFPAGE_PSIZE;
1118
1119         pairsize = key_size + data_size;
1120
1121         /* Advance to first page in chain with room for item. */
1122         while (H_NUMPAIRS(hcp->pagep) && NEXT_PGNO(hcp->pagep) !=
1123             PGNO_INVALID) {
1124                 /*
1125                  * This may not be the end of the chain, but the pair may fit
1126                  * anyway.  Check if it's a bigpair that fits or a regular
1127                  * pair that fits.
1128                  */
1129                 if (P_FREESPACE(hcp->pagep) >= pairsize)
1130                         break;
1131                 next_pgno = NEXT_PGNO(hcp->pagep);
1132                 if ((ret =
1133                     __ham_next_cpage(hashp, hcp, next_pgno, 0, 0)) != 0)
1134                         return (ret);
1135         }
1136
1137         /*
1138          * Check if we need to allocate a new page.
1139          */
1140         if (P_FREESPACE(hcp->pagep) < pairsize) {
1141                 do_expand = 1;
1142                 if ((ret = __ham_add_ovflpage(hashp,
1143                     hcp->pagep, 1, &hcp->pagep)) !=  0)
1144                         return (ret);
1145                 hcp->pgno = PGNO(hcp->pagep);
1146         }
1147
1148         /*
1149          * Update cursor.
1150          */
1151         hcp->bndx = H_NUMPAIRS(hcp->pagep);
1152         F_CLR(hcp, H_DELETED);
1153         if (is_keybig) {
1154                 if ((ret = __db_poff(hashp->dbp,
1155                     key, &koff.pgno, __ham_overflow_page)) != 0)
1156                         return (ret);
1157                 koff.type = H_OFFPAGE;
1158                 koff.tlen = key->size;
1159                 key_dbt.data = &koff;
1160                 key_dbt.size = sizeof(koff);
1161                 pkey = &key_dbt;
1162                 key_type = H_OFFPAGE;
1163         } else {
1164                 pkey = (DBT *)key;
1165                 key_type = H_KEYDATA;
1166         }
1167
1168         if (is_databig) {
1169                 if ((ret = __db_poff(hashp->dbp,
1170                     val, &doff.pgno, __ham_overflow_page)) != 0)
1171                         return (ret);
1172                 doff.type = H_OFFPAGE;
1173                 doff.tlen = val->size;
1174                 data_dbt.data = &doff;
1175                 data_dbt.size = sizeof(doff);
1176                 pdata = &data_dbt;
1177                 data_type = H_OFFPAGE;
1178         } else {
1179                 pdata = (DBT *)val;
1180                 data_type = type;
1181         }
1182
1183         if (DB_LOGGING(hashp->dbp)) {
1184                 rectype = PUTPAIR;
1185                 if (is_databig)
1186                         rectype |= PAIR_DATAMASK;
1187                 if (is_keybig)
1188                         rectype |= PAIR_KEYMASK;
1189
1190                 if ((ret = __ham_insdel_log(hashp->dbp->dbenv->lg_info,
1191                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, rectype,
1192                     hashp->dbp->log_fileid, PGNO(hcp->pagep),
1193                     (u_int32_t)H_NUMPAIRS(hcp->pagep),
1194                     &LSN(hcp->pagep), pkey, pdata)) != 0)
1195                         return (ret);
1196
1197                 /* Move lsn onto page. */
1198                 LSN(hcp->pagep) = new_lsn;      /* Structure assignment. */
1199         }
1200
1201         __ham_putitem(hcp->pagep, pkey, key_type);
1202         __ham_putitem(hcp->pagep, pdata, data_type);
1203
1204         /*
1205          * For splits, we are going to update item_info's page number
1206          * field, so that we can easily return to the same page the
1207          * next time we come in here.  For other operations, this shouldn't
1208          * matter, since odds are this is the last thing that happens before
1209          * we return to the user program.
1210          */
1211         hcp->pgno = PGNO(hcp->pagep);
1212
1213         /*
1214          * XXX Maybe keep incremental numbers here
1215          */
1216         if (!F_ISSET(hashp->dbp, DB_AM_LOCKING))
1217                 hashp->hdr->nelem++;
1218
1219         if (do_expand || (hashp->hdr->ffactor != 0 &&
1220             (u_int32_t)H_NUMPAIRS(hcp->pagep) > hashp->hdr->ffactor))
1221                 F_SET(hcp, H_EXPAND);
1222         return (0);
1223 }
1224
1225
1226 /*
1227  * Special __putitem call used in splitting -- copies one entry to
1228  * another.  Works for all types of hash entries (H_OFFPAGE, H_KEYDATA,
1229  * H_DUPLICATE, H_OFFDUP).  Since we log splits at a high level, we
1230  * do not need to do any logging here.
1231  * PUBLIC: void __ham_copy_item __P((HTAB *, PAGE *, int, PAGE *));
1232  */
1233 void
1234 __ham_copy_item(hashp, src_page, src_ndx, dest_page)
1235         HTAB *hashp;
1236         PAGE *src_page;
1237         int src_ndx;
1238         PAGE *dest_page;
1239 {
1240         u_int32_t len;
1241         void *src, *dest;
1242
1243         /*
1244          * Copy the key and data entries onto this new page.
1245          */
1246         src = P_ENTRY(src_page, src_ndx);
1247
1248         /* Set up space on dest. */
1249         len = LEN_HITEM(src_page, hashp->hdr->pagesize, src_ndx);
1250         HOFFSET(dest_page) -= len;
1251         dest_page->inp[NUM_ENT(dest_page)] = HOFFSET(dest_page);
1252         dest = P_ENTRY(dest_page, NUM_ENT(dest_page));
1253         NUM_ENT(dest_page)++;
1254
1255         memcpy(dest, src, len);
1256 }
1257
1258 /*
1259  *
1260  * Returns:
1261  *      pointer on success
1262  *      NULL on error
1263  *
1264  * PUBLIC: int __ham_add_ovflpage __P((HTAB *, PAGE *, int, PAGE **));
1265  */
1266 int
1267 __ham_add_ovflpage(hashp, pagep, release, pp)
1268         HTAB *hashp;
1269         PAGE *pagep;
1270         int release;
1271         PAGE **pp;
1272 {
1273         DB_ENV *dbenv;
1274         DB_LSN new_lsn;
1275         PAGE *new_pagep;
1276         int ret;
1277
1278         dbenv = hashp->dbp->dbenv;
1279
1280         if ((ret = __ham_overflow_page(hashp->dbp, P_HASH, &new_pagep)) != 0)
1281                 return (ret);
1282
1283         if (DB_LOGGING(hashp->dbp)) {
1284                 if ((ret = __ham_newpage_log(dbenv->lg_info,
1285                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, PUTOVFL,
1286                     hashp->dbp->log_fileid, PGNO(pagep), &LSN(pagep),
1287                     PGNO(new_pagep), &LSN(new_pagep), PGNO_INVALID, NULL)) != 0)
1288                         return (ret);
1289
1290                 /* Move lsn onto page. */
1291                 LSN(pagep) = LSN(new_pagep) = new_lsn;
1292         }
1293         NEXT_PGNO(pagep) = PGNO(new_pagep);
1294         PREV_PGNO(new_pagep) = PGNO(pagep);
1295
1296         if (release)
1297                 ret = __ham_put_page(hashp->dbp, pagep, 1);
1298
1299         hashp->hash_overflows++;
1300         *pp = new_pagep;
1301         return (ret);
1302 }
1303
1304
1305 /*
1306  * PUBLIC: int __ham_new_page __P((HTAB *, u_int32_t, u_int32_t, PAGE **));
1307  */
1308 int
1309 __ham_new_page(hashp, addr, type, pp)
1310         HTAB *hashp;
1311         u_int32_t addr, type;
1312         PAGE **pp;
1313 {
1314         PAGE *pagep;
1315         int ret;
1316
1317         if ((ret = memp_fget(hashp->dbp->mpf,
1318             &addr, DB_MPOOL_CREATE, &pagep)) != 0)
1319                 return (ret);
1320
1321 #ifdef DEBUG_SLOW
1322         account_page(hashp, addr, 1);
1323 #endif
1324         /* This should not be necessary because page-in should do it. */
1325         P_INIT(pagep,
1326             hashp->hdr->pagesize, addr, PGNO_INVALID, PGNO_INVALID, 0, type);
1327
1328         *pp = pagep;
1329         return (0);
1330 }
1331
1332 /*
1333  * PUBLIC: int __ham_del_page __P((DB *, PAGE *));
1334  */
1335 int
1336 __ham_del_page(dbp, pagep)
1337         DB *dbp;
1338         PAGE *pagep;
1339 {
1340         DB_LSN new_lsn;
1341         HTAB *hashp;
1342         int ret;
1343
1344         hashp = (HTAB *)dbp->internal;
1345         ret = 0;
1346         DIRTY_META(hashp, ret);
1347         if (ret != 0) {
1348                 if (ret != EAGAIN)
1349                         __db_err(hashp->dbp->dbenv,
1350                             "free_ovflpage: unable to lock meta data page %s\n",
1351                             strerror(ret));
1352                 /*
1353                  * If we are going to return an error, then we should free
1354                  * the page, so it doesn't stay pinned forever.
1355                  */
1356                 (void)__ham_put_page(hashp->dbp, pagep, 0);
1357                 return (ret);
1358         }
1359
1360         if (DB_LOGGING(hashp->dbp)) {
1361                 if ((ret = __ham_newpgno_log(hashp->dbp->dbenv->lg_info,
1362                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, DELPGNO,
1363                     hashp->dbp->log_fileid, PGNO(pagep), hashp->hdr->last_freed,
1364                     (u_int32_t)TYPE(pagep), NEXT_PGNO(pagep), P_INVALID,
1365                     &LSN(pagep), &hashp->hdr->lsn)) != 0)
1366                         return (ret);
1367
1368                 hashp->hdr->lsn = new_lsn;
1369                 LSN(pagep) = new_lsn;
1370         }
1371
1372 #ifdef DEBUG
1373         {
1374                 db_pgno_t __pgno;
1375                 DB_LSN __lsn;
1376                 __pgno = pagep->pgno;
1377                 __lsn = pagep->lsn;
1378                 memset(pagep, 0xff, dbp->pgsize);
1379                 pagep->pgno = __pgno;
1380                 pagep->lsn = __lsn;
1381         }
1382 #endif
1383         TYPE(pagep) = P_INVALID;
1384         NEXT_PGNO(pagep) = hashp->hdr->last_freed;
1385         hashp->hdr->last_freed = PGNO(pagep);
1386
1387         return (__ham_put_page(hashp->dbp, pagep, 1));
1388 }
1389
1390
1391 /*
1392  * PUBLIC: int __ham_put_page __P((DB *, PAGE *, int32_t));
1393  */
1394 int
1395 __ham_put_page(dbp, pagep, is_dirty)
1396         DB *dbp;
1397         PAGE *pagep;
1398         int32_t is_dirty;
1399 {
1400 #ifdef DEBUG_SLOW
1401         account_page((HTAB *)dbp->cookie,
1402             ((BKT *)((char *)pagep - sizeof(BKT)))->pgno, -1);
1403 #endif
1404         return (memp_fput(dbp->mpf, pagep, (is_dirty ? DB_MPOOL_DIRTY : 0)));
1405 }
1406
1407 /*
1408  * __ham_dirty_page --
1409  *      Mark a page dirty.
1410  *
1411  * PUBLIC: int __ham_dirty_page __P((HTAB *, PAGE *));
1412  */
1413 int
1414 __ham_dirty_page(hashp, pagep)
1415         HTAB *hashp;
1416         PAGE *pagep;
1417 {
1418         return (memp_fset(hashp->dbp->mpf, pagep, DB_MPOOL_DIRTY));
1419 }
1420
1421 /*
1422  * PUBLIC: int __ham_get_page __P((DB *, db_pgno_t, PAGE **));
1423  */
1424 int
1425 __ham_get_page(dbp, addr, pagep)
1426         DB *dbp;
1427         db_pgno_t addr;
1428         PAGE **pagep;
1429 {
1430         int ret;
1431
1432         ret = memp_fget(dbp->mpf, &addr, DB_MPOOL_CREATE, pagep);
1433 #ifdef DEBUG_SLOW
1434         if (*pagep != NULL)
1435                 account_page((HTAB *)dbp->internal, addr, 1);
1436 #endif
1437         return (ret);
1438 }
1439
1440 /*
1441  * PUBLIC: int __ham_overflow_page __P((DB *, u_int32_t, PAGE **));
1442  */
1443 int
1444 __ham_overflow_page(dbp, type, pp)
1445         DB *dbp;
1446         u_int32_t type;
1447         PAGE **pp;
1448 {
1449         DB_LSN *lsnp, new_lsn;
1450         HTAB *hashp;
1451         PAGE *p;
1452         db_pgno_t new_addr, next_free, newalloc_flag;
1453         u_int32_t offset, splitnum;
1454         int ret;
1455
1456         hashp = (HTAB *)dbp->internal;
1457
1458         ret = 0;
1459         DIRTY_META(hashp, ret);
1460         if (ret != 0)
1461                 return (ret);
1462
1463         /*
1464          * This routine is split up into two parts.  First we have
1465          * to figure out the address of the new page that we are
1466          * allocating.  Then we have to log the allocation.  Only
1467          * after the log do we get to complete allocation of the
1468          * new page.
1469          */
1470         new_addr = hashp->hdr->last_freed;
1471         if (new_addr != PGNO_INVALID) {
1472                 if ((ret = __ham_get_page(hashp->dbp, new_addr, &p)) != 0)
1473                         return (ret);
1474                 next_free = NEXT_PGNO(p);
1475                 lsnp = &LSN(p);
1476                 newalloc_flag = 0;
1477         } else {
1478                 splitnum = hashp->hdr->ovfl_point;
1479                 hashp->hdr->spares[splitnum]++;
1480                 offset = hashp->hdr->spares[splitnum] -
1481                     (splitnum ? hashp->hdr->spares[splitnum - 1] : 0);
1482                 new_addr = PGNO_OF(hashp, hashp->hdr->ovfl_point, offset);
1483                 if (new_addr > MAX_PAGES(hashp)) {
1484                         __db_err(hashp->dbp->dbenv, "hash: out of file pages");
1485                         hashp->hdr->spares[splitnum]--;
1486                         return (ENOMEM);
1487                 }
1488                 next_free = PGNO_INVALID;
1489                 p = NULL;
1490                 lsnp = NULL;
1491                 newalloc_flag = 1;
1492         }
1493
1494         if (DB_LOGGING(hashp->dbp)) {
1495                 if ((ret = __ham_newpgno_log(hashp->dbp->dbenv->lg_info,
1496                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, ALLOCPGNO,
1497                     hashp->dbp->log_fileid, new_addr, next_free,
1498                     0, newalloc_flag, type, lsnp, &hashp->hdr->lsn)) != 0)
1499                         return (ret);
1500
1501                 hashp->hdr->lsn = new_lsn;
1502                 if (lsnp != NULL)
1503                         *lsnp = new_lsn;
1504         }
1505
1506         if (p != NULL) {
1507                 /* We just took something off the free list, initialize it. */
1508                 hashp->hdr->last_freed = next_free;
1509                 P_INIT(p, hashp->hdr->pagesize, PGNO(p), PGNO_INVALID,
1510                     PGNO_INVALID, 0, (u_int8_t)type);
1511         } else {
1512                 /* Get the new page. */
1513                 if ((ret = __ham_new_page(hashp, new_addr, type, &p)) != 0)
1514                         return (ret);
1515         }
1516         if (DB_LOGGING(hashp->dbp))
1517                 LSN(p) = new_lsn;
1518
1519         *pp = p;
1520         return (0);
1521 }
1522
1523 #ifdef DEBUG
1524 /*
1525  * PUBLIC: #ifdef DEBUG
1526  * PUBLIC: int bucket_to_page __P((HTAB *, int));
1527  * PUBLIC: #endif
1528  */
1529 int
1530 bucket_to_page(hashp, n)
1531         HTAB *hashp;
1532         int n;
1533 {
1534         int ret_val;
1535
1536         ret_val = n + 1;
1537         if (n != 0)
1538                 ret_val += hashp->hdr->spares[__db_log2(n + 1) - 1];
1539         return (ret_val);
1540 }
1541 #endif
1542
1543
1544 /*
1545  * Create a bunch of overflow pages at the current split point.
1546  * PUBLIC: void __ham_init_ovflpages __P((HTAB *));
1547  */
1548 void
1549 __ham_init_ovflpages(hp)
1550         HTAB *hp;
1551 {
1552         DB_LSN new_lsn;
1553         PAGE *p;
1554         db_pgno_t last_pgno;
1555         u_int32_t i, numpages;
1556
1557         numpages = hp->hdr->ovfl_point + 1;
1558
1559         last_pgno = hp->hdr->last_freed;
1560         if (DB_LOGGING(hp->dbp)) {
1561                 (void)__ham_ovfl_log(hp->dbp->dbenv->lg_info,
1562                     (DB_TXN *)hp->dbp->txn, &new_lsn, 0,
1563                     hp->dbp->log_fileid, PGNO_OF(hp, hp->hdr->ovfl_point, 1),
1564                     numpages, last_pgno, &hp->hdr->lsn);
1565                 hp->hdr->lsn = new_lsn;
1566         } else
1567                 ZERO_LSN(new_lsn);
1568
1569         hp->hdr->spares[hp->hdr->ovfl_point] += numpages;
1570         for (i = numpages; i > 0; i--) {
1571                 if (__ham_new_page(hp,
1572                     PGNO_OF(hp, hp->hdr->ovfl_point, i), P_INVALID, &p) != 0)
1573                         break;
1574                 LSN(p) = new_lsn;
1575                 NEXT_PGNO(p) = last_pgno;
1576                 last_pgno = PGNO(p);
1577                 (void)__ham_put_page(hp->dbp, p, 1);
1578         }
1579         hp->hdr->last_freed = last_pgno;
1580 }
1581
1582 /*
1583  * PUBLIC: int __ham_get_cpage __P((HTAB *, HASH_CURSOR *, db_lockmode_t));
1584  */
1585 int
1586 __ham_get_cpage(hashp, hcp, mode)
1587         HTAB *hashp;
1588         HASH_CURSOR *hcp;
1589         db_lockmode_t mode;
1590 {
1591         int ret;
1592
1593         if (hcp->lock == 0 && F_ISSET(hashp->dbp, DB_AM_LOCKING) &&
1594             (ret = __ham_lock_bucket(hashp->dbp, hcp, mode)) != 0)
1595                 return (ret);
1596
1597         if (hcp->pagep == NULL) {
1598                 if (hcp->pgno == PGNO_INVALID) {
1599                         hcp->pgno = BUCKET_TO_PAGE(hashp, hcp->bucket);
1600                         hcp->bndx = 0;
1601                 }
1602
1603                 if ((ret =
1604                     __ham_get_page(hashp->dbp, hcp->pgno, &hcp->pagep)) != 0)
1605                         return (ret);
1606         }
1607
1608         if (hcp->dpgno != PGNO_INVALID && hcp->dpagep == NULL)
1609                 if ((ret =
1610                     __ham_get_page(hashp->dbp, hcp->dpgno, &hcp->dpagep)) != 0)
1611                         return (ret);
1612         return (0);
1613 }
1614
1615 /*
1616  * Get a new page at the cursor, putting the last page if necessary.
1617  * If the flag is set to H_ISDUP, then we are talking about the
1618  * duplicate page, not the main page.
1619  * PUBLIC: int __ham_next_cpage __P((HTAB *, HASH_CURSOR *, db_pgno_t,
1620  * PUBLIC:     int, int));
1621  */
1622 int
1623 __ham_next_cpage(hashp, hcp, pgno, dirty, flags)
1624         HTAB *hashp;
1625         HASH_CURSOR *hcp;
1626         db_pgno_t pgno;
1627         int dirty;
1628         int flags;
1629 {
1630         PAGE *p;
1631         int ret;
1632
1633         if (flags & H_ISDUP && hcp->dpagep != NULL &&
1634             (ret = __ham_put_page(hashp->dbp, hcp->dpagep, dirty)) != 0)
1635                 return (ret);
1636         else if (!(flags & H_ISDUP) && hcp->pagep != NULL &&
1637             (ret = __ham_put_page(hashp->dbp, hcp->pagep, dirty)) != 0)
1638                 return (ret);
1639
1640         if ((ret = __ham_get_page(hashp->dbp, pgno, &p)) != 0)
1641                 return (ret);
1642
1643         if (flags & H_ISDUP) {
1644                 hcp->dpagep = p;
1645                 hcp->dpgno = pgno;
1646                 hcp->dndx = 0;
1647         } else {
1648                 hcp->pagep = p;
1649                 hcp->pgno = pgno;
1650                 hcp->bndx = 0;
1651         }
1652
1653         return (0);
1654 }
1655
1656 /*
1657  * __ham_lock_bucket --
1658  *      Get the lock on a particular bucket.
1659  */
1660 static int
1661 __ham_lock_bucket(dbp, hcp, mode)
1662         DB *dbp;
1663         HASH_CURSOR *hcp;
1664         db_lockmode_t mode;
1665 {
1666         int ret;
1667
1668         /*
1669          * What a way to trounce on the memory system.  It might be
1670          * worth copying the lk_info into the hashp.
1671          */
1672         ret = 0;
1673         dbp->lock.pgno = (db_pgno_t)(hcp->bucket);
1674         ret = lock_get(dbp->dbenv->lk_info,
1675             dbp->txn == NULL ?  dbp->locker : dbp->txn->txnid, 0,
1676             &dbp->lock_dbt, mode, &hcp->lock);
1677
1678         return (ret < 0 ? EAGAIN : ret);
1679 }
1680
1681 /*
1682  * __ham_dpair --
1683  *      Delete a pair on a page, paying no attention to what the pair
1684  *      represents.  The caller is responsible for freeing up duplicates
1685  *      or offpage entries that might be referenced by this pair.
1686  *
1687  * PUBLIC: void __ham_dpair __P((DB *, PAGE *, u_int32_t));
1688  */
1689 void
1690 __ham_dpair(dbp, p, pndx)
1691         DB *dbp;
1692         PAGE *p;
1693         u_int32_t pndx;
1694 {
1695         db_indx_t delta, n;
1696         u_int8_t *dest, *src;
1697
1698         /*
1699          * Compute "delta", the amount we have to shift all of the
1700          * offsets.  To find the delta, we just need to calculate
1701          * the size of the pair of elements we are removing.
1702          */
1703         delta = H_PAIRSIZE(p, dbp->pgsize, pndx);
1704
1705         /*
1706          * The hard case: we want to remove something other than
1707          * the last item on the page.  We need to shift data and
1708          * offsets down.
1709          */
1710         if ((db_indx_t)pndx != H_NUMPAIRS(p) - 1) {
1711                 /*
1712                  * Move the data: src is the first occupied byte on
1713                  * the page. (Length is delta.)
1714                  */
1715                 src = (u_int8_t *)p + HOFFSET(p);
1716
1717                 /*
1718                  * Destination is delta bytes beyond src.  This might
1719                  * be an overlapping copy, so we have to use memmove.
1720                  */
1721                 dest = src + delta;
1722                 memmove(dest, src, p->inp[H_DATAINDEX(pndx)] - HOFFSET(p));
1723         }
1724
1725         /* Adjust the offsets. */
1726         for (n = (db_indx_t)pndx; n < (db_indx_t)(H_NUMPAIRS(p) - 1); n++) {
1727                 p->inp[H_KEYINDEX(n)] = p->inp[H_KEYINDEX(n+1)] + delta;
1728                 p->inp[H_DATAINDEX(n)] = p->inp[H_DATAINDEX(n+1)] + delta;
1729         }
1730
1731         /* Adjust page metadata. */
1732         HOFFSET(p) = HOFFSET(p) + delta;
1733         NUM_ENT(p) = NUM_ENT(p) - 2;
1734 }
1735
1736 #ifdef DEBUG_SLOW
1737 static void
1738 account_page(hashp, pgno, inout)
1739         HTAB *hashp;
1740         db_pgno_t pgno;
1741         int inout;
1742 {
1743         static struct {
1744                 db_pgno_t pgno;
1745                 int times;
1746         } list[100];
1747         static int last;
1748         int i, j;
1749
1750         if (inout == -1)                        /* XXX: Kluge */
1751                 inout = 0;
1752
1753         /* Find page in list. */
1754         for (i = 0; i < last; i++)
1755                 if (list[i].pgno == pgno)
1756                         break;
1757         /* Not found. */
1758         if (i == last) {
1759                 list[last].times = inout;
1760                 list[last].pgno = pgno;
1761                 last++;
1762         }
1763         list[i].times = inout;
1764         if (list[i].times == 0) {
1765                 for (j = i; j < last; j++)
1766                         list[j] = list[j + 1];
1767                 last--;
1768         }
1769         for (i = 0; i < last; i++, list[i].times++)
1770                 if (list[i].times > 20 && !is_bitmap_pgno(hashp, list[i].pgno))
1771                         (void)fprintf(stderr,
1772                             "Warning: pg %lu has been out for %d times\n",
1773                             (u_long)list[i].pgno, list[i].times);
1774 }
1775 #endif /* DEBUG_SLOW */