git.sur5r.net Git - openldap/blob - servers/slapd/back-ldbm/idl.c

   1 /* idl.c - ldap id list handling routines */
   2 /* $OpenLDAP$ */
   3 /* This work is part of OpenLDAP Software <http://www.openldap.org/>.
   4  *
   5  * Copyright 1998-2004 The OpenLDAP Foundation.
   6  * All rights reserved.
   7  *
   8  * Redistribution and use in source and binary forms, with or without
   9  * modification, are permitted only as authorized by the OpenLDAP
  10  * Public License.
  11  *
  12  * A copy of this license is available in the file LICENSE in the
  13  * top-level directory of the distribution or, alternatively, at
  14  * <http://www.OpenLDAP.org/license.html>.
  15  */
  16
  17 #include "portable.h"
  18
  19 #include <stdio.h>
  20
  21 #include <ac/string.h>
  22 #include <ac/socket.h>
  23
  24 #include "slap.h"
  25 #include "back-ldbm.h"
  26
  27 static ID_BLOCK* idl_dup( ID_BLOCK *idl );
  28
  29 static void cont_alloc( Datum *cont, Datum *key )
  30 {
  31         ldbm_datum_init( *cont );
  32         cont->dsize = 1 + sizeof(ID) + key->dsize;
  33         cont->dptr = ch_malloc( cont->dsize );
  34
  35         * (unsigned char *) cont->dptr = SLAP_INDEX_CONT_PREFIX;
  36
  37         AC_MEMCPY( &((unsigned char *)cont->dptr)[1 + sizeof(ID)],
  38                 key->dptr, key->dsize );
  39 }
  40
  41 static void cont_id( Datum *cont, ID id )
  42 {
  43         unsigned int i;
  44
  45         for( i=1; i <= sizeof(id); i++) {
  46                 ((unsigned char *)cont->dptr)[i] = (unsigned char)(id & 0xFF);
  47                 id >>= 8;
  48         }
  49
  50 }
  51
  52 static void cont_free( Datum *cont )
  53 {
  54         ch_free( cont->dptr );
  55 }
  56
  57 #ifdef LDBM_DEBUG_IDL
  58 static void idl_check(ID_BLOCK *idl)
  59 {
  60         int i, max;
  61         ID_BLOCK last;
  62
  63         if( ID_BLOCK_ALLIDS(idl) )
  64         {
  65                 return;
  66         }
  67 #ifndef USE_INDIRECT_NIDS
  68         if( ID_BLOCK_INDIRECT(idl) )
  69         {
  70                 for ( max = 0; !ID_BLOCK_NOID(idl, max); max++ ) ;
  71         } else
  72 #endif
  73         {
  74                 max = ID_BLOCK_NIDS(idl);
  75         }
  76         if ( max <= 1 )
  77         {
  78                 return;
  79         }
  80
  81         for( last = ID_BLOCK_ID(idl, 0), i = 1;
  82                 i < max;
  83                 last = ID_BLOCK_ID(idl, i), i++ )
  84         {
  85                 assert (last < ID_BLOCK_ID(idl, i) );
  86         }
  87 }
  88 #endif
  89
  90 /* Allocate an ID_BLOCK with room for nids ids */
  91 ID_BLOCK *
  92 idl_alloc( unsigned int nids )
  93 {
  94         ID_BLOCK        *new;
  95
  96         /* nmax + nids + space for the ids */
  97         new = (ID_BLOCK *) ch_calloc( (ID_BLOCK_IDS_OFFSET + nids), sizeof(ID) );
  98         ID_BLOCK_NMAX(new) = nids;
  99         ID_BLOCK_NIDS(new) = 0;
 100
 101         return( new );
 102 }
 103
 104
 105 /* Allocate an empty ALLIDS ID_BLOCK */
 106 ID_BLOCK        *
 107 idl_allids( Backend *be )
 108 {
 109         ID_BLOCK        *idl;
 110         ID              id;
 111
 112         idl = idl_alloc( 0 );
 113         ID_BLOCK_NMAX(idl) = ID_BLOCK_ALLIDS_VALUE;
 114         if ( next_id_get( be, &id ) ) {
 115                 return NULL;
 116         }
 117         ID_BLOCK_NIDS(idl) = id;
 118
 119         return( idl );
 120 }
 121
 122 /* Free an ID_BLOCK */
 123 void
 124 idl_free( ID_BLOCK *idl )
 125 {
 126         if ( idl == NULL ) {
 127                 Debug( LDAP_DEBUG_TRACE,
 128                         "idl_free: called with NULL pointer\n",
 129                         0, 0, 0 );
 130
 131                 return;
 132         }
 133
 134         free( (char *) idl );
 135 }
 136
 137
 138 /* Fetch an single ID_BLOCK from the cache */
 139 static ID_BLOCK *
 140 idl_fetch_one(
 141     Backend             *be,
 142     DBCache     *db,
 143     Datum               key
 144 )
 145 {
 146         Datum   data;
 147         ID_BLOCK        *idl;
 148
 149         /* Debug( LDAP_DEBUG_TRACE, "=> idl_fetch_one\n", 0, 0, 0 ); */
 150
 151         data = ldbm_cache_fetch( db, key );
 152
 153         if( data.dptr == NULL ) {
 154                 return NULL;
 155         }
 156
 157         idl = (ID_BLOCK *) data.dptr;
 158         if ( ID_BLOCK_ALLIDS(idl) ) {
 159                 /* make sure we have the current value of highest id */
 160                 idl = idl_allids( be );
 161         } else {
 162                 idl = idl_dup((ID_BLOCK *) data.dptr);
 163         }
 164
 165         ldbm_datum_free( db->dbc_db, data );
 166
 167         return idl;
 168 }
 169
 170
 171 /* Fetch a set of ID_BLOCKs from the cache
 172  *      if not INDIRECT
 173  *              if block return is an ALLIDS block,
 174  *                      return an new ALLIDS block
 175  *              otherwise
 176  *                      return block
 177  *      construct super block from all blocks referenced by INDIRECT block
 178  *      return super block
 179  */
 180 ID_BLOCK *
 181 idl_fetch(
 182     Backend             *be,
 183     DBCache     *db,
 184     Datum               key
 185 )
 186 {
 187         Datum   data;
 188         ID_BLOCK        *idl;
 189         ID_BLOCK        **tmp;
 190         unsigned        i, nids, nblocks;
 191
 192         idl = idl_fetch_one( be, db, key );
 193
 194         if ( idl == NULL ) {
 195                 return NULL;
 196         }
 197
 198         if ( ID_BLOCK_ALLIDS(idl) ) {
 199                 /* all ids block */
 200                 return( idl );
 201         }
 202
 203         if ( ! ID_BLOCK_INDIRECT( idl ) ) {
 204                 /* regular block */
 205                 return( idl );
 206         }
 207
 208         /*
 209          * this is an indirect block which points to other blocks.
 210          * we need to read in all the blocks it points to and construct
 211          * a big id list containing all the ids, which we will return.
 212          */
 213
 214 #ifndef USE_INDIRECT_NIDS
 215         /* count the number of blocks & allocate space for pointers to them */
 216         for ( nblocks = 0; !ID_BLOCK_NOID(idl, nblocks); nblocks++ )
 217                 ;       /* NULL */
 218 #else
 219         nblocks = ID_BLOCK_NIDS(idl);
 220 #endif
 221         tmp = (ID_BLOCK **) ch_malloc( nblocks * sizeof(ID_BLOCK *) );
 222
 223         /* read in all the blocks */
 224         cont_alloc( &data, &key );
 225         nids = 0;
 226         for ( i = 0; i < nblocks; i++ ) {
 227                 cont_id( &data, ID_BLOCK_ID(idl, i) );
 228
 229                 if ( (tmp[i] = idl_fetch_one( be, db, data )) == NULL ) {
 230                         Debug( LDAP_DEBUG_ANY,
 231                             "idl_fetch: one returned NULL\n", 0, 0, 0 );
 232
 233                         continue;
 234                 }
 235
 236                 nids += ID_BLOCK_NIDS(tmp[i]);
 237         }
 238         cont_free( &data );
 239         idl_free( idl );
 240
 241         /* allocate space for the big block */
 242         idl = idl_alloc( nids );
 243         ID_BLOCK_NIDS(idl) = nids;
 244         nids = 0;
 245
 246         /* copy in all the ids from the component blocks */
 247         for ( i = 0; i < nblocks; i++ ) {
 248                 if ( tmp[i] == NULL ) {
 249                         continue;
 250                 }
 251
 252                 AC_MEMCPY(
 253                         (char *) &ID_BLOCK_ID(idl, nids),
 254                         (char *) &ID_BLOCK_ID(tmp[i], 0),
 255                         ID_BLOCK_NIDS(tmp[i]) * sizeof(ID) );
 256                 nids += ID_BLOCK_NIDS(tmp[i]);
 257
 258                 idl_free( tmp[i] );
 259         }
 260         free( (char *) tmp );
 261
 262         assert( ID_BLOCK_NIDS(idl) == nids );
 263
 264 #ifdef LDBM_DEBUG_IDL
 265         idl_check(idl);
 266 #endif
 267
 268         Debug( LDAP_DEBUG_TRACE, "<= idl_fetch %ld ids (%ld max)\n",
 269                ID_BLOCK_NIDS(idl), ID_BLOCK_NMAXN(idl), 0 );
 270
 271         return( idl );
 272 }
 273
 274
 275 /* store a single block */
 276 static int
 277 idl_store(
 278     Backend             *be,
 279     DBCache     *db,
 280     Datum               key,
 281     ID_BLOCK            *idl
 282 )
 283 {
 284         int     rc, flags;
 285         Datum   data;
 286
 287 #ifdef LDBM_DEBUG_IDL
 288         idl_check(idl);
 289 #endif
 290
 291         ldbm_datum_init( data );
 292
 293         /* Debug( LDAP_DEBUG_TRACE, "=> idl_store\n", 0, 0, 0 ); */
 294
 295         data.dptr = (char *) idl;
 296         data.dsize = (ID_BLOCK_IDS_OFFSET + ID_BLOCK_NMAXN(idl)) * sizeof(ID);
 297
 298         flags = LDBM_REPLACE;
 299         rc = ldbm_cache_store( db, key, data, flags );
 300
 301         /* Debug( LDAP_DEBUG_TRACE, "<= idl_store %d\n", rc, 0, 0 ); */
 302         return( rc );
 303 }
 304
 305 /* Binary search for id in block, return index
 306  *    an index is always returned, even with no match. If no
 307  * match, the returned index is the insertion point.
 308  */
 309 static unsigned int
 310 idl_find(
 311     ID_BLOCK    *b,
 312     ID          id
 313 )
 314 {
 315         int lo=0, hi=ID_BLOCK_NIDS(b)-1, nr=0;
 316
 317         for (;lo<=hi;)
 318         {
 319             nr = ( lo + hi ) / 2;
 320             if (ID_BLOCK_ID(b, nr) == id)
 321                 break;
 322             if (ID_BLOCK_ID(b, nr) > id)
 323                 hi = nr - 1;
 324             else
 325                 lo = nr + 1;
 326         }
 327         return nr;
 328 }
 329
 330 /* split the block at id
 331  *      locate ID greater than or equal to id.
 332  */
 333 static void
 334 idl_split_block(
 335     ID_BLOCK    *b,
 336     ID          id,
 337     ID_BLOCK    **right,
 338     ID_BLOCK    **left
 339 )
 340 {
 341         unsigned int    nr, nl;
 342
 343         /* find where to split the block */
 344         nr = idl_find(b, id);
 345         if ( ID_BLOCK_ID(b,nr) < id )
 346                 nr++;
 347
 348         nl = ID_BLOCK_NIDS(b) - nr;
 349
 350         *right = idl_alloc( nr == 0 ? 1 : nr );
 351         *left = idl_alloc( nl + (nr == 0 ? 0 : 1));
 352
 353         /*
 354          * everything before the id being inserted in the first block
 355          * unless there is nothing, in which case the id being inserted
 356          * goes there.
 357          */
 358         if ( nr == 0 ) {
 359                 ID_BLOCK_NIDS(*right) = 1;
 360                 ID_BLOCK_ID(*right, 0) = id;
 361         } else {
 362                 AC_MEMCPY(
 363                         (char *) &ID_BLOCK_ID(*right, 0),
 364                         (char *) &ID_BLOCK_ID(b, 0),
 365                         nr * sizeof(ID) );
 366                 ID_BLOCK_NIDS(*right) = nr;
 367                 ID_BLOCK_ID(*left, 0) = id;
 368         }
 369
 370         /* the id being inserted & everything after in the second block */
 371         AC_MEMCPY(
 372                 (char *) &ID_BLOCK_ID(*left, (nr == 0 ? 0 : 1)),
 373             (char *) &ID_BLOCK_ID(b, nr),
 374                 nl * sizeof(ID) );
 375         ID_BLOCK_NIDS(*left) = nl + (nr == 0 ? 0 : 1);
 376
 377 #ifdef LDBM_DEBUG_IDL
 378         idl_check(*right);
 379         idl_check(*left);
 380 #endif
 381 }
 382
 383
 384 /*
 385  * idl_change_first - called when an indirect block's first key has
 386  * changed, meaning it needs to be stored under a new key, and the
 387  * header block pointing to it needs updating.
 388  */
 389 static int
 390 idl_change_first(
 391     Backend             *be,
 392     DBCache     *db,
 393     Datum               hkey,           /* header block key     */
 394     ID_BLOCK            *h,             /* header block         */
 395     int                 pos,            /* pos in h to update   */
 396     Datum               bkey,           /* data block key       */
 397     ID_BLOCK            *b              /* data block           */
 398 )
 399 {
 400         int     rc;
 401
 402         /* Debug( LDAP_DEBUG_TRACE, "=> idl_change_first\n", 0, 0, 0 ); */
 403
 404         /* delete old key block */
 405         if ( (rc = ldbm_cache_delete( db, bkey )) != 0 ) {
 406                 Debug( LDAP_DEBUG_ANY,
 407                     "idl_change_first: ldbm_cache_delete returned %d\n",
 408                         rc, 0, 0 );
 409
 410                 return( rc );
 411         }
 412
 413         /* write block with new key */
 414         cont_id( &bkey, ID_BLOCK_ID(b, 0) );
 415
 416         if ( (rc = idl_store( be, db, bkey, b )) != 0 ) {
 417                 Debug( LDAP_DEBUG_ANY,
 418                     "idl_change_first: idl_store returned %d\n", rc, 0, 0 );
 419
 420                 return( rc );
 421         }
 422
 423         /* update + write indirect header block */
 424         ID_BLOCK_ID(h, pos) = ID_BLOCK_ID(b, 0);
 425         if ( (rc = idl_store( be, db, hkey, h )) != 0 ) {
 426                 Debug( LDAP_DEBUG_ANY,
 427                     "idl_change_first: idl_store returned %d\n", rc, 0, 0 );
 428
 429                 return( rc );
 430         }
 431
 432         return( 0 );
 433 }
 434
 435
 436 int
 437 idl_insert_key(
 438     Backend             *be,
 439     DBCache     *db,
 440     Datum               key,
 441     ID                  id
 442 )
 443 {
 444         int     i, j, first, rc = 0;
 445         ID_BLOCK        *idl, *tmp, *tmp2, *tmp3;
 446         Datum   k2;
 447
 448         if ( (idl = idl_fetch_one( be, db, key )) == NULL ) {
 449                 idl = idl_alloc( 1 );
 450                 ID_BLOCK_ID(idl, ID_BLOCK_NIDS(idl)++) = id;
 451                 rc = idl_store( be, db, key, idl );
 452
 453                 idl_free( idl );
 454                 return( rc );
 455         }
 456
 457         if ( ID_BLOCK_ALLIDS( idl ) ) {
 458                 /* ALLIDS */
 459                 idl_free( idl );
 460                 return 0;
 461         }
 462
 463         if ( ! ID_BLOCK_INDIRECT( idl ) ) {
 464                 /* regular block */
 465                 switch ( idl_insert( &idl, id, db->dbc_maxids ) ) {
 466                 case 0:         /* id inserted - store the updated block */
 467                 case 1:
 468                         rc = idl_store( be, db, key, idl );
 469                         break;
 470
 471                 case 2:         /* id already there - nothing to do */
 472                         rc = 0;
 473                         break;
 474
 475                 case 3:         /* id not inserted - block must be split */
 476                         /* check threshold for marking this an all-id block */
 477                         if ( db->dbc_maxindirect < 2 ) {
 478                                 idl_free( idl );
 479                                 idl = idl_allids( be );
 480                                 rc = idl_store( be, db, key, idl );
 481                                 break;
 482                         }
 483
 484                         idl_split_block( idl, id, &tmp, &tmp2 );
 485                         idl_free( idl );
 486
 487                         /* create the header indirect block */
 488 #ifndef USE_INDIRECT_NIDS
 489                         idl = idl_alloc( 3 );
 490                         ID_BLOCK_NMAX(idl) = 3;
 491                         ID_BLOCK_NIDS(idl) = ID_BLOCK_INDIRECT_VALUE;
 492                         ID_BLOCK_ID(idl, 0) = ID_BLOCK_ID(tmp, 0);
 493                         ID_BLOCK_ID(idl, 1) = ID_BLOCK_ID(tmp2, 0);
 494                         ID_BLOCK_ID(idl, 2) = NOID;
 495 #else
 496                         idl = idl_alloc( 2 );
 497                         ID_BLOCK_NMAX(idl) = 2 | ID_BLOCK_INDIRECT_VALUE;
 498                         ID_BLOCK_NIDS(idl) = 2;
 499                         ID_BLOCK_ID(idl, 0) = ID_BLOCK_ID(tmp, 0);
 500                         ID_BLOCK_ID(idl, 1) = ID_BLOCK_ID(tmp2, 0);
 501 #endif
 502
 503                         /* store it */
 504                         rc = idl_store( be, db, key, idl );
 505
 506                         cont_alloc( &k2, &key );
 507                         cont_id( &k2, ID_BLOCK_ID(tmp, 0) );
 508
 509                         rc = idl_store( be, db, k2, tmp );
 510
 511                         cont_id( &k2, ID_BLOCK_ID(tmp2, 0) );
 512                         rc = idl_store( be, db, k2, tmp2 );
 513
 514                         cont_free( &k2 );
 515
 516                         idl_free( tmp );
 517                         idl_free( tmp2 );
 518                         break;
 519                 }
 520
 521                 idl_free( idl );
 522                 return( rc );
 523         }
 524
 525         /*
 526          * this is an indirect block which points to other blocks.
 527          * we need to read in the block into which the id should be
 528          * inserted, then insert the id and store the block.  we might
 529          * have to split the block if it is full, which means we also
 530          * need to write a new "header" block.
 531          */
 532
 533 #ifndef USE_INDIRECT_NIDS
 534         /* select the block to try inserting into *//* XXX linear search XXX */
 535         for ( i = 0; !ID_BLOCK_NOID(idl, i) && id >= ID_BLOCK_ID(idl, i); i++ )
 536                 ;       /* NULL */
 537 #else
 538         i = idl_find(idl, id);
 539         if (ID_BLOCK_ID(idl, i) <= id)
 540                 i++;
 541 #endif
 542         if ( i != 0 ) {
 543                 i--;
 544                 first = 0;
 545         } else {
 546                 first = 1;
 547         }
 548
 549         /* At this point, the following condition must be true:
 550          * ID_BLOCK_ID(idl, i) <= id && id < ID_BLOCK_ID(idl, i+1)
 551          * except when i is the first or the last block.
 552          */
 553
 554         /* get the block */
 555         cont_alloc( &k2, &key );
 556         cont_id( &k2, ID_BLOCK_ID(idl, i) );
 557
 558         if ( (tmp = idl_fetch_one( be, db, k2 )) == NULL ) {
 559                 Debug( LDAP_DEBUG_ANY, "idl_insert_key: nonexistent continuation block\n",
 560                     0, 0, 0 );
 561
 562                 cont_free( &k2 );
 563                 idl_free( idl );
 564                 return( -1 );
 565         }
 566
 567         /* insert the id */
 568         switch ( idl_insert( &tmp, id, db->dbc_maxids ) ) {
 569         case 0:         /* id inserted ok */
 570                 if ( (rc = idl_store( be, db, k2, tmp )) != 0 ) {
 571                         Debug( LDAP_DEBUG_ANY,
 572                             "idl_insert_key: idl_store returned %d\n", rc, 0, 0 );
 573
 574                 }
 575                 break;
 576
 577         case 1:         /* id inserted - first id in block has changed */
 578                 /*
 579                  * key for this block has changed, so we have to
 580                  * write the block under the new key, delete the
 581                  * old key block + update and write the indirect
 582                  * header block.
 583                  */
 584
 585                 rc = idl_change_first( be, db, key, idl, i, k2, tmp );
 586                 break;
 587
 588         case 2:         /* id not inserted - already there, do nothing */
 589                 rc = 0;
 590                 break;
 591
 592         case 3:         /* id not inserted - block is full */
 593                 /*
 594                  * first, see if it will fit in the next block,
 595                  * without splitting, unless we're trying to insert
 596                  * into the beginning of the first block.
 597                  */
 598
 599 #ifndef USE_INDIRECT_NIDS
 600                 /* is there a next block? */
 601                 if ( !first && !ID_BLOCK_NOID(idl, i + 1) ) {
 602 #else
 603                 if ( !first && (unsigned long)(i + 1) < ID_BLOCK_NIDS(idl) ) {
 604 #endif
 605                         Datum k3;
 606                         /* read it in */
 607                         cont_alloc( &k3, &key );
 608                         cont_id( &k3, ID_BLOCK_ID(idl, i + 1) );
 609                         if ( (tmp2 = idl_fetch_one( be, db, k3 )) == NULL ) {
 610                                 Debug( LDAP_DEBUG_ANY,
 611                                     "idl_insert_key: idl_fetch_one returned NULL\n",
 612                                     0, 0, 0 );
 613
 614                                 /* split the original block */
 615                                 cont_free( &k3 );
 616                                 goto split;
 617                         }
 618
 619                         /* If the new id is less than the last id in the
 620                          * current block, it must not be put into the next
 621                          * block. Push the last id of the current block
 622                          * into the next block instead.
 623                          */
 624                         if (id < ID_BLOCK_ID(tmp, ID_BLOCK_NIDS(tmp) - 1)) {
 625                             ID id2 = ID_BLOCK_ID(tmp, ID_BLOCK_NIDS(tmp) - 1);
 626
 627                             --ID_BLOCK_NIDS(tmp);
 628                             /* This must succeed since we just popped one
 629                              * ID off the end of it.
 630                              */
 631                             rc = idl_insert( &tmp, id, db->dbc_maxids );
 632
 633                             if ( (rc = idl_store( be, db, k2, tmp )) != 0 ) {
 634                                 Debug( LDAP_DEBUG_ANY,
 635                             "idl_insert_key: idl_store returned %d\n", rc, 0, 0 );
 636
 637                             }
 638
 639                             id = id2;
 640                             /* This new id will necessarily be inserted
 641                              * as the first id of the next block by the
 642                              * following switch() statement.
 643                              */
 644                         }
 645
 646                         switch ( (rc = idl_insert( &tmp2, id,
 647                             db->dbc_maxids )) ) {
 648                         case 1:         /* id inserted first in block */
 649                                 rc = idl_change_first( be, db, key, idl,
 650                                     i + 1, k3, tmp2 );
 651                                 /* FALL */
 652
 653                         case 2:         /* id already there - how? */
 654                         case 0:         /* id inserted: this can never be
 655                                          * the result of idl_insert, because
 656                                          * we guaranteed that idl_change_first
 657                                          * will always be called.
 658                                          */
 659                                 if ( rc == 2 ) {
 660                                         Debug( LDAP_DEBUG_ANY,
 661                                             "idl_insert_key: id %ld already in next block\n",
 662                                             id, 0, 0 );
 663
 664                                 }
 665
 666                                 idl_free( tmp );
 667                                 idl_free( tmp2 );
 668                                 cont_free( &k3 );
 669                                 cont_free( &k2 );
 670                                 idl_free( idl );
 671                                 return( 0 );
 672
 673                         case 3:         /* split the original block */
 674                                 break;
 675                         }
 676
 677                         idl_free( tmp2 );
 678                         cont_free( &k3 );
 679                 }
 680
 681 split:
 682                 /*
 683                  * must split the block, write both new blocks + update
 684                  * and write the indirect header block.
 685                  */
 686
 687                 rc = 0; /* optimistic */
 688
 689
 690 #ifndef USE_INDIRECT_NIDS
 691                 /* count how many indirect blocks *//* XXX linear count XXX */
 692                 for ( j = 0; !ID_BLOCK_NOID(idl, j); j++ )
 693                         ;       /* NULL */
 694 #else
 695                 j = ID_BLOCK_NIDS(idl);
 696 #endif
 697
 698                 /* check it against all-id thresholed */
 699                 if ( j + 1 > db->dbc_maxindirect ) {
 700                         /*
 701                          * we've passed the all-id threshold, meaning
 702                          * that this set of blocks should be replaced
 703                          * by a single "all-id" block.  our job: delete
 704                          * all the indirect blocks, and replace the header
 705                          * block by an all-id block.
 706                          */
 707
 708                         /* delete all indirect blocks */
 709 #ifndef USE_INDIRECT_NIDS
 710                         for ( j = 0; !ID_BLOCK_NOID(idl, j); j++ ) {
 711 #else
 712                         for ( j = 0; (unsigned long) j < ID_BLOCK_NIDS(idl); j++ ) {
 713 #endif
 714                                 cont_id( &k2, ID_BLOCK_ID(idl, j) );
 715
 716                                 rc = ldbm_cache_delete( db, k2 );
 717                         }
 718
 719                         /* store allid block in place of header block */
 720                         idl_free( idl );
 721                         idl = idl_allids( be );
 722                         rc = idl_store( be, db, key, idl );
 723
 724                         cont_free( &k2 );
 725                         idl_free( idl );
 726                         idl_free( tmp );
 727                         return( rc );
 728                 }
 729
 730                 idl_split_block( tmp, id, &tmp2, &tmp3 );
 731                 idl_free( tmp );
 732
 733                 /* create a new updated indirect header block */
 734                 tmp = idl_alloc( ID_BLOCK_NMAXN(idl) + 1 );
 735 #ifndef USE_INDIRECT_NIDS
 736                 ID_BLOCK_NIDS(tmp) = ID_BLOCK_INDIRECT_VALUE;
 737 #else
 738                 ID_BLOCK_NMAX(tmp) |= ID_BLOCK_INDIRECT_VALUE;
 739 #endif
 740                 /* everything up to the split block */
 741                 AC_MEMCPY(
 742                         (char *) &ID_BLOCK_ID(tmp, 0),
 743                         (char *) &ID_BLOCK_ID(idl, 0),
 744                     i * sizeof(ID) );
 745                 /* the two new blocks */
 746                 ID_BLOCK_ID(tmp, i) = ID_BLOCK_ID(tmp2, 0);
 747                 ID_BLOCK_ID(tmp, i + 1) = ID_BLOCK_ID(tmp3, 0);
 748                 /* everything after the split block */
 749 #ifndef USE_INDIRECT_NIDS
 750                 AC_MEMCPY(
 751                         (char *) &ID_BLOCK_ID(tmp, i + 2),
 752                         (char *) &ID_BLOCK_ID(idl, i + 1),
 753                         (ID_BLOCK_NMAXN(idl) - i - 1) * sizeof(ID) );
 754 #else
 755                 AC_MEMCPY(
 756                         (char *) &ID_BLOCK_ID(tmp, i + 2),
 757                         (char *) &ID_BLOCK_ID(idl, i + 1),
 758                         (ID_BLOCK_NIDS(idl) - i - 1) * sizeof(ID) );
 759                 ID_BLOCK_NIDS(tmp) = ID_BLOCK_NIDS(idl) + 1;
 760 #endif
 761
 762                 /* store the header block */
 763                 rc = idl_store( be, db, key, tmp );
 764
 765                 /* store the first id block */
 766                 cont_id( &k2, ID_BLOCK_ID(tmp2, 0) );
 767                 rc = idl_store( be, db, k2, tmp2 );
 768
 769                 /* store the second id block */
 770                 cont_id( &k2, ID_BLOCK_ID(tmp3, 0) );
 771                 rc = idl_store( be, db, k2, tmp3 );
 772
 773                 idl_free( tmp2 );
 774                 idl_free( tmp3 );
 775                 break;
 776         }
 777
 778         cont_free( &k2 );
 779         idl_free( tmp );
 780         idl_free( idl );
 781         return( rc );
 782 }
 783
 784
 785 /*
 786  * idl_insert - insert an id into an id list.
 787  *
 788  *      returns
 789  *              0       id inserted
 790  *              1       id inserted, first id in block has changed
 791  *              2       id not inserted, already there
 792  *              3       id not inserted, block must be split
 793  */
 794 int
 795 idl_insert( ID_BLOCK **idl, ID id, unsigned int maxids )
 796 {
 797         unsigned int    i;
 798
 799         if ( ID_BLOCK_ALLIDS( *idl ) ) {
 800                 return( 2 );    /* already there */
 801         }
 802
 803         /* is it already there? */
 804         i = idl_find(*idl, id);
 805         if ( ID_BLOCK_ID(*idl, i) == id ) {
 806                 return( 2 );    /* already there */
 807         }
 808         if ( ID_BLOCK_NIDS(*idl) && ID_BLOCK_ID(*idl, i) < id )
 809                 i++;
 810
 811         /* do we need to make room for it? */
 812         if ( ID_BLOCK_NIDS(*idl) == ID_BLOCK_NMAXN(*idl) ) {
 813                 /* make room or indicate block needs splitting */
 814                 if ( ID_BLOCK_NMAXN(*idl) >= maxids ) {
 815                         return( 3 );    /* block needs splitting */
 816                 }
 817
 818                 ID_BLOCK_NMAX(*idl) *= 2;
 819                 if ( ID_BLOCK_NMAXN(*idl) > maxids ) {
 820                         ID_BLOCK_NMAX(*idl) = maxids;
 821                 }
 822                 *idl = (ID_BLOCK *) ch_realloc( (char *) *idl,
 823                     (ID_BLOCK_NMAXN(*idl) + ID_BLOCK_IDS_OFFSET) * sizeof(ID) );
 824         }
 825
 826         /* make a slot for the new id */
 827         AC_MEMCPY( &ID_BLOCK_ID(*idl, i+1), &ID_BLOCK_ID(*idl, i),
 828                     (ID_BLOCK_NIDS(*idl) - i) * sizeof(ID) );
 829
 830         ID_BLOCK_ID(*idl, i) = id;
 831         ID_BLOCK_NIDS(*idl)++;
 832         (void) memset(
 833                 (char *) &ID_BLOCK_ID((*idl), ID_BLOCK_NIDS(*idl)),
 834                 '\0',
 835             (ID_BLOCK_NMAXN(*idl) - ID_BLOCK_NIDS(*idl)) * sizeof(ID) );
 836
 837 #ifdef LDBM_DEBUG_IDL
 838         idl_check(*idl);
 839 #endif
 840
 841         return( i == 0 ? 1 : 0 );       /* inserted - first id changed or not */
 842 }
 843
 844
 845 int
 846 idl_delete_key (
 847         Backend         *be,
 848         DBCache  *db,
 849         Datum           key,
 850         ID              id
 851 )
 852 {
 853         Datum  data;
 854         ID_BLOCK *idl;
 855         unsigned i;
 856         int j, nids;
 857
 858         if ( (idl = idl_fetch_one( be, db, key ) ) == NULL )
 859         {
 860                 /* It wasn't found.  Hmm... */
 861                 return -1;
 862         }
 863
 864         if ( ID_BLOCK_ALLIDS( idl ) ) {
 865                 idl_free( idl );
 866                 return 0;
 867         }
 868
 869         if ( ! ID_BLOCK_INDIRECT( idl ) ) {
 870                 i = idl_find(idl, id);
 871                 if ( ID_BLOCK_ID(idl, i) == id ) {
 872                         if( --ID_BLOCK_NIDS(idl) == 0 ) {
 873                                 ldbm_cache_delete( db, key );
 874
 875                         } else {
 876                                 AC_MEMCPY(
 877                                         &ID_BLOCK_ID(idl, i),
 878                                         &ID_BLOCK_ID(idl, i+1),
 879                                         (ID_BLOCK_NIDS(idl)-i) * sizeof(ID) );
 880
 881                                 ID_BLOCK_ID(idl, ID_BLOCK_NIDS(idl)) = NOID;
 882
 883                                 idl_store( be, db, key, idl );
 884                         }
 885
 886                         idl_free( idl );
 887                         return 0;
 888                 }
 889                 /*  We didn't find the ID.  Hmmm... */
 890                 idl_free( idl );
 891                 return -1;
 892         }
 893
 894         /* We have to go through an indirect block and find the ID
 895            in the list of IDL's
 896            */
 897         cont_alloc( &data, &key );
 898 #ifndef USE_INDIRECT_NIDS
 899         for ( nids = 0; !ID_BLOCK_NOID(idl, nids); nids++ ) {
 900                 ;       /* Empty */
 901         }
 902
 903         for ( j = 0; j<nids; j++ )
 904 #else
 905         nids = ID_BLOCK_NIDS(idl);
 906         j = idl_find(idl, id);
 907         if ( ID_BLOCK_ID(idl, j) > id ) j--;
 908         for (; j>=0; j = -1 ) /* execute once */
 909 #endif
 910         {
 911                 ID_BLOCK *tmp;
 912                 cont_id( &data, ID_BLOCK_ID(idl, j) );
 913
 914                 if ( (tmp = idl_fetch_one( be, db, data )) == NULL ) {
 915                         Debug( LDAP_DEBUG_ANY,
 916                             "idl_delete_key: idl_fetch of returned NULL\n", 0, 0, 0 );
 917
 918                         continue;
 919                 }
 920                 /*
 921                    Now try to find the ID in tmp
 922                 */
 923
 924                 i = idl_find(tmp, id);
 925                 if ( ID_BLOCK_ID(tmp, i) == id )
 926                 {
 927                         AC_MEMCPY(
 928                                 &ID_BLOCK_ID(tmp, i),
 929                                 &ID_BLOCK_ID(tmp, i+1),
 930                                 (ID_BLOCK_NIDS(tmp)-(i+1)) * sizeof(ID));
 931                         ID_BLOCK_ID(tmp, ID_BLOCK_NIDS(tmp)-1 ) = NOID;
 932                         ID_BLOCK_NIDS(tmp)--;
 933
 934                         if ( ID_BLOCK_NIDS(tmp) ) {
 935                                 idl_store ( be, db, data, tmp );
 936
 937                         } else {
 938                                 ldbm_cache_delete( db, data );
 939                                 AC_MEMCPY(
 940                                         &ID_BLOCK_ID(idl, j),
 941                                         &ID_BLOCK_ID(idl, j+1),
 942                                         (nids-(j+1)) * sizeof(ID));
 943                                 ID_BLOCK_ID(idl, nids-1) = NOID;
 944                                 nids--;
 945 #ifdef USE_INDIRECT_NIDS
 946                                 ID_BLOCK_NIDS(idl)--;
 947 #endif
 948                                 if ( ! nids )
 949                                         ldbm_cache_delete( db, key );
 950                                 else
 951                                         idl_store( be, db, key, idl );
 952                         }
 953                         idl_free( tmp );
 954                         cont_free( &data );
 955                         idl_free( idl );
 956                         return 0;
 957                 }
 958                 idl_free( tmp );
 959         }
 960
 961         cont_free( &data );
 962         idl_free( idl );
 963         return -1;
 964 }
 965
 966
 967 /* return a duplicate of a single ID_BLOCK */
 968 static ID_BLOCK *
 969 idl_dup( ID_BLOCK *idl )
 970 {
 971         ID_BLOCK        *new;
 972
 973         if ( idl == NULL ) {
 974                 return( NULL );
 975         }
 976
 977         new = idl_alloc( ID_BLOCK_NMAXN(idl) );
 978
 979         AC_MEMCPY(
 980                 (char *) new,
 981                 (char *) idl,
 982                 (ID_BLOCK_NMAXN(idl) + ID_BLOCK_IDS_OFFSET) * sizeof(ID) );
 983
 984 #ifdef LDBM_DEBUG_IDL
 985         idl_check(new);
 986 #endif
 987
 988         return( new );
 989 }
 990
 991
 992 /* return the smaller ID_BLOCK */
 993 static ID_BLOCK *
 994 idl_min( ID_BLOCK *a, ID_BLOCK *b )
 995 {
 996         return( ID_BLOCK_NIDS(a) > ID_BLOCK_NIDS(b) ? b : a );
 997 }
 998
 999
1000 /*
1001  * idl_intersection - return a intersection b
1002  */
1003 ID_BLOCK *
1004 idl_intersection(
1005     Backend     *be,
1006     ID_BLOCK    *a,
1007     ID_BLOCK    *b
1008 )
1009 {
1010         unsigned int    ai, bi, ni;
1011         ID_BLOCK                *n;
1012
1013         if ( a == NULL || b == NULL ) {
1014                 return( NULL );
1015         }
1016         if ( ID_BLOCK_ALLIDS( a ) ) {
1017                 return( idl_dup( b ) );
1018         }
1019         if ( ID_BLOCK_ALLIDS( b ) ) {
1020                 return( idl_dup( a ) );
1021         }
1022         if ( ID_BLOCK_NIDS(a) == 0 || ID_BLOCK_NIDS(b) == 0 ) {
1023                 return( NULL );
1024         }
1025
1026         n = idl_dup( idl_min( a, b ) );
1027
1028 #ifdef LDBM_DEBUG_IDL
1029         idl_check(a);
1030         idl_check(b);
1031 #endif
1032
1033         for ( ni = 0, ai = 0, bi = 0; ; ) {
1034                 if ( ID_BLOCK_ID(b, bi) == ID_BLOCK_ID(a, ai) ) {
1035                         ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai);
1036                         ai++;
1037                         bi++;
1038                         if ( ai >= ID_BLOCK_NIDS(a) || bi >= ID_BLOCK_NIDS(b) )
1039                                 break;
1040                 } else if ( ID_BLOCK_ID(a, ai) < ID_BLOCK_ID(b, bi) ) {
1041                         ai++;
1042                         if ( ai >= ID_BLOCK_NIDS(a) )
1043                                 break;
1044                 } else {
1045                         bi++;
1046                         if ( bi >= ID_BLOCK_NIDS(b) )
1047                                 break;
1048                 }
1049         }
1050
1051         if ( ni == 0 ) {
1052                 idl_free( n );
1053                 return( NULL );
1054         }
1055         ID_BLOCK_NIDS(n) = ni;
1056
1057 #ifdef LDBM_DEBUG_IDL
1058         idl_check(n);
1059 #endif
1060
1061         return( n );
1062 }
1063
1064
1065 /*
1066  * idl_union - return a union b
1067  */
1068 ID_BLOCK *
1069 idl_union(
1070     Backend     *be,
1071     ID_BLOCK    *a,
1072     ID_BLOCK    *b
1073 )
1074 {
1075         unsigned int    ai, bi, ni;
1076         ID_BLOCK                *n;
1077
1078         if ( a == NULL ) {
1079                 return( idl_dup( b ) );
1080         }
1081         if ( b == NULL ) {
1082                 return( idl_dup( a ) );
1083         }
1084         if ( ID_BLOCK_ALLIDS( a ) || ID_BLOCK_ALLIDS( b ) ) {
1085                 return( idl_allids( be ) );
1086         }
1087
1088 #ifdef LDBM_DEBUG_IDL
1089         idl_check(a);
1090         idl_check(b);
1091 #endif
1092
1093         if ( ID_BLOCK_NIDS(b) < ID_BLOCK_NIDS(a) ) {
1094                 n = a;
1095                 a = b;
1096                 b = n;
1097         }
1098
1099         n = idl_alloc( ID_BLOCK_NIDS(a) + ID_BLOCK_NIDS(b) );
1100
1101         for ( ni = 0, ai = 0, bi = 0;
1102                 ai < ID_BLOCK_NIDS(a) && bi < ID_BLOCK_NIDS(b);
1103                 )
1104         {
1105                 if ( ID_BLOCK_ID(a, ai) < ID_BLOCK_ID(b, bi) ) {
1106                         ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai++);
1107
1108                 } else if ( ID_BLOCK_ID(b, bi) < ID_BLOCK_ID(a, ai) ) {
1109                         ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(b, bi++);
1110
1111                 } else {
1112                         ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai);
1113                         ai++, bi++;
1114                 }
1115         }
1116
1117         for ( ; ai < ID_BLOCK_NIDS(a); ai++ ) {
1118                 ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai);
1119         }
1120         for ( ; bi < ID_BLOCK_NIDS(b); bi++ ) {
1121                 ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(b, bi);
1122         }
1123         ID_BLOCK_NIDS(n) = ni;
1124
1125 #ifdef LDBM_DEBUG_IDL
1126         idl_check(n);
1127 #endif
1128
1129         return( n );
1130 }
1131
1132
1133 /*
1134  * idl_notin - return a intersection ~b (or a minus b)
1135  */
1136 ID_BLOCK *
1137 idl_notin(
1138     Backend     *be,
1139     ID_BLOCK    *a,
1140     ID_BLOCK    *b
1141 )
1142 {
1143         unsigned int    ni, ai, bi;
1144         ID_BLOCK                *n;
1145
1146         if ( a == NULL ) {
1147                 return( NULL );
1148         }
1149         if ( b == NULL || ID_BLOCK_ALLIDS( b )) {
1150                 return( idl_dup( a ) );
1151         }
1152
1153         if ( ID_BLOCK_ALLIDS( a ) ) {
1154                 n = idl_alloc( SLAPD_LDBM_MIN_MAXIDS );
1155                 ni = 0;
1156
1157                 for ( ai = 1, bi = 0;
1158                         ai < ID_BLOCK_NIDS(a) && ni < ID_BLOCK_NMAXN(n) && bi < ID_BLOCK_NMAXN(b);
1159                         ai++ )
1160                 {
1161                         if ( ID_BLOCK_ID(b, bi) == ai ) {
1162                                 bi++;
1163                         } else {
1164                                 ID_BLOCK_ID(n, ni++) = ai;
1165                         }
1166                 }
1167
1168                 for ( ; ai < ID_BLOCK_NIDS(a) && ni < ID_BLOCK_NMAXN(n); ai++ ) {
1169                         ID_BLOCK_ID(n, ni++) = ai;
1170                 }
1171
1172                 if ( ni == ID_BLOCK_NMAXN(n) ) {
1173                         idl_free( n );
1174                         return( idl_allids( be ) );
1175                 } else {
1176                         ID_BLOCK_NIDS(n) = ni;
1177                         return( n );
1178                 }
1179         }
1180
1181         n = idl_dup( a );
1182
1183         ni = 0;
1184         for ( ai = 0, bi = 0; ai < ID_BLOCK_NIDS(a); ai++ ) {
1185                 for ( ;
1186                         bi < ID_BLOCK_NIDS(b) && ID_BLOCK_ID(b, bi) < ID_BLOCK_ID(a, ai);
1187                     bi++ )
1188                 {
1189                         ;       /* NULL */
1190                 }
1191
1192                 if ( bi == ID_BLOCK_NIDS(b) ) {
1193                         break;
1194                 }
1195
1196                 if ( ID_BLOCK_ID(b, bi) != ID_BLOCK_ID(a, ai) ) {
1197                         ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai);
1198                 }
1199         }
1200
1201         for ( ; ai < ID_BLOCK_NIDS(a); ai++ ) {
1202                 ID_BLOCK_ID(n, ni++) = ID_BLOCK_ID(a, ai);
1203         }
1204         ID_BLOCK_NIDS(n) = ni;
1205
1206 #ifdef LDBM_DEBUG_IDL
1207         idl_check(n);
1208 #endif
1209
1210         return( n );
1211 }
1212
1213 /*      return the first ID in the block
1214  *      if ALLIDS block
1215  *              NIDS > 1 return 1
1216  *              otherwise return NOID
1217  *      otherwise return first ID
1218  *
1219  *      cursor is set to 1
1220  */
1221 ID
1222 idl_firstid( ID_BLOCK *idl, ID *cursor )
1223 {
1224         *cursor = 1;
1225
1226         if ( idl == NULL || ID_BLOCK_NIDS(idl) == 0 ) {
1227                 return( NOID );
1228         }
1229
1230         if ( ID_BLOCK_ALLIDS( idl ) ) {
1231                 return( ID_BLOCK_NIDS(idl) > 1 ? 1 : NOID );
1232         }
1233
1234         return( ID_BLOCK_ID(idl, 0) );
1235 }
1236
1237 /*      return next ID
1238  *      if ALLIDS block, cursor is id.
1239  *              increment id
1240  *              if id < NIDS return id
1241  *              otherwise NOID.
1242  *      otherwise cursor is index into block
1243  *              if index < nids
1244  *                      return id at index then increment
1245  */
1246 ID
1247 idl_nextid( ID_BLOCK *idl, ID *cursor )
1248 {
1249         if ( ID_BLOCK_ALLIDS( idl ) ) {
1250                 if( ++(*cursor) < ID_BLOCK_NIDS(idl) ) {
1251                         return *cursor;
1252                 } else {
1253                         return NOID;
1254                 }
1255         }
1256
1257         if ( *cursor < ID_BLOCK_NIDS(idl) ) {
1258                 return( ID_BLOCK_ID(idl, (*cursor)++) );
1259         }
1260
1261         return( NOID );
1262 }