db_am.c

00001 /*-
00002  * See the file LICENSE for redistribution information.
00003  *
00004  * Copyright (c) 1998, 1999, 2000
00005  * Sleepycat Software. All rights reserved.
00006  */
00007 
00008 #include "config.h"
00009 
00010 #ifndef lint
00011 static const char revid[] = "$Id: db__am_8c-source.html,v 1.1 2008年06月08日 10:17:11 sebdiaz Exp $";
00012 #endif /* not lint */
00013 
00014 #ifndef NO_SYSTEM_INCLUDES
00015 #include <sys/types.h>
00016 
00017 #include <errno.h>
00018 #include <string.h>
00019 #endif
00020 
00021 #include "db_int.h"
00022 #include "db_page.h"
00023 #include "db_shash.h"
00024 #include "btree.h"
00025 #include "hash.h"
00026 #include "qam.h"
00027 #include "lock.h"
00028 #include "mp.h"
00029 #include "txn.h"
00030 #include "db_am.h"
00031 #include "db_ext.h"
00032 
00033 #ifdef DEBUG
00034 #include "WordMonitor.h"
00035 #endif /* DEBUG */
00036 
00037 /*
00038  * CDB___db_cursor --
00039  * Allocate and return a cursor.
00040  *
00041  * PUBLIC: int CDB___db_cursor __P((DB *, DB_TXN *, DBC **, u_int32_t));
00042  */
00043 int
00044 CDB___db_cursor(dbp, txn, dbcp, flags)
00045 DB *dbp;
00046 DB_TXN *txn;
00047 DBC **dbcp;
00048 u_int32_t flags;
00049 {
00050 DB_ENV *dbenv;
00051 DBC *dbc;
00052 db_lockmode_t mode;
00053 u_int32_t op;
00054 int ret;
00055 
00056 dbenv = dbp->dbenv;
00057 
00058 PANIC_CHECK(dbenv);
00059 DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->cursor");
00060 
00061 /* Check for invalid flags. */
00062 if ((ret = CDB___db_cursorchk(dbp, flags, F_ISSET(dbp, DB_AM_RDONLY))) != 0)
00063 return (ret);
00064 
00065 if ((ret =
00066 CDB___db_icursor(dbp, txn, dbp->type, PGNO_INVALID, 0, dbcp)) != 0)
00067 return (ret);
00068 dbc = *dbcp;
00069 
00070 /*
00071  * If this is CDB, do all the locking in the interface, which is
00072  * right here.
00073  */
00074 if (LOCKING(dbenv)) {
00075 op = LF_ISSET(DB_OPFLAGS_MASK);
00076 mode = (op == DB_WRITELOCK) ? DB_LOCK_WRITE :
00077 ((op == DB_WRITECURSOR) ? DB_LOCK_IWRITE : DB_LOCK_READ);
00078 if ((ret = CDB_lock_get(dbenv, dbc->locker, 0,
00079 &dbc->lock_dbt, mode, &dbc->mylock)) != 0) {
00080 (void)CDB___db_c_close(dbc);
00081 return (ret);
00082 }
00083 if (op == DB_WRITECURSOR)
00084 F_SET(dbc, DBC_WRITECURSOR);
00085 if (op == DB_WRITELOCK)
00086 F_SET(dbc, DBC_WRITER);
00087 }
00088 
00089 return (0);
00090 }
00091 
00092 /*
00093  * CDB___db_icursor --
00094  * Internal version of CDB___db_cursor. If dbcp is
00095  * non-NULL it is assumed to point to an area to
00096  * initialize as a cursor.
00097  *
00098  * PUBLIC: int CDB___db_icursor
00099  * PUBLIC: __P((DB *, DB_TXN *, DBTYPE, db_pgno_t, int, DBC **));
00100  */
00101 int
00102 CDB___db_icursor(dbp, txn, dbtype, root, is_opd, dbcp)
00103 DB *dbp;
00104 DB_TXN *txn;
00105 DBTYPE dbtype;
00106 db_pgno_t root;
00107 int is_opd;
00108 DBC **dbcp;
00109 {
00110 DBC *dbc, *adbc;
00111 DBC_INTERNAL *cp;
00112 DB_ENV *dbenv;
00113 int allocated, ret;
00114 
00115 dbenv = dbp->dbenv;
00116 allocated = 0;
00117 
00118 /*
00119  * Take one from the free list if it's available. Take only the
00120  * right type. With off page dups we may have different kinds
00121  * of cursors on the queue for a single database.
00122  */
00123 MUTEX_THREAD_LOCK(dbp->mutexp);
00124 for (dbc = TAILQ_FIRST(&dbp->free_queue);
00125 dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
00126 if (dbtype == dbc->dbtype) {
00127 TAILQ_REMOVE(&dbp->free_queue, dbc, links);
00128 dbc->flags = 0;
00129 break;
00130 }
00131 MUTEX_THREAD_UNLOCK(dbp->mutexp);
00132 
00133 if (dbc == NULL) {
00134 if ((ret = CDB___os_calloc(dbp->dbenv, 1, sizeof(DBC), &dbc)) != 0)
00135 return (ret);
00136 allocated = 1;
00137 dbc->flags = 0;
00138 
00139 dbc->dbp = dbp;
00140 
00141 /* Set up locking information. */
00142 if (LOCKING_ON(dbenv)) {
00143 /*
00144  * If we are not threaded, then there is no need to
00145  * create new locker ids. We know that no one else
00146  * is running concurrently using this DB, so we can
00147  * take a peek at any cursors on the active queue.
00148  */
00149 if (!DB_IS_THREADED(dbp) &&
00150 (adbc = TAILQ_FIRST(&dbp->active_queue)) != NULL)
00151 dbc->lid = adbc->lid;
00152 else
00153 if ((ret = CDB_lock_id(dbenv, &dbc->lid)) != 0)
00154 goto err;
00155 
00156 memcpy(dbc->lock.fileid, dbp->fileid, DB_FILE_ID_LEN);
00157 if (LOCKING(dbenv)) {
00158 dbc->lock_dbt.size = DB_FILE_ID_LEN;
00159 dbc->lock_dbt.data = dbc->lock.fileid;
00160 } else {
00161 dbc->lock.type = DB_PAGE_LOCK;
00162 dbc->lock_dbt.size = sizeof(dbc->lock);
00163 dbc->lock_dbt.data = &dbc->lock;
00164 }
00165 }
00166 /* Init the DBC internal structure. */
00167 switch (dbtype) {
00168 case DB_BTREE:
00169 case DB_RECNO:
00170 if ((ret = CDB___bam_c_init(dbc, dbtype)) != 0)
00171 goto err;
00172 break;
00173 case DB_HASH:
00174 if ((ret = CDB___ham_c_init(dbc)) != 0)
00175 goto err;
00176 break;
00177 case DB_QUEUE:
00178 if ((ret = CDB___qam_c_init(dbc)) != 0)
00179 goto err;
00180 break;
00181 default:
00182 ret = CDB___db_unknown_type(dbp->dbenv,
00183 "CDB___db_icursor", dbtype);
00184 goto err;
00185 }
00186 
00187 cp = dbc->internal;
00188 }
00189 
00190 /* Refresh the DBC structure. */
00191 dbc->dbtype = dbtype;
00192 
00193 if ((dbc->txn = txn) == NULL)
00194 dbc->locker = dbc->lid;
00195 else
00196 dbc->locker = txn->txnid;
00197 
00198 if (is_opd)
00199 F_SET(dbc, DBC_OPD);
00200 if (F_ISSET(dbp, DB_AM_RECOVER))
00201 F_SET(dbc, DBC_RECOVER);
00202 
00203 /* Refresh the DBC internal structure. */
00204 cp = dbc->internal;
00205 cp->opd = NULL;
00206 
00207 cp->indx = 0;
00208 cp->page = NULL;
00209 cp->pgno = PGNO_INVALID;
00210 cp->root = root;
00211 
00212 switch (dbtype) {
00213 case DB_BTREE:
00214 case DB_RECNO:
00215 if ((ret = CDB___bam_c_refresh(dbc)) != 0)
00216 goto err;
00217 break;
00218 case DB_HASH:
00219 case DB_QUEUE:
00220 break;
00221 default:
00222 ret = CDB___db_unknown_type(dbp->dbenv, "CDB___db_icursor", dbp->type);
00223 goto err;
00224 }
00225 
00226 MUTEX_THREAD_LOCK(dbp->mutexp);
00227 TAILQ_INSERT_TAIL(&dbp->active_queue, dbc, links);
00228 F_SET(dbc, DBC_ACTIVE);
00229 MUTEX_THREAD_UNLOCK(dbp->mutexp);
00230 
00231 *dbcp = dbc;
00232 return (0);
00233 
00234 err: if (allocated)
00235 CDB___os_free(dbc, sizeof(*dbc));
00236 return (ret);
00237 }
00238 
00239 #ifdef DEBUG
00240 /*
00241  * CDB___db_cprint --
00242  * Display the current cursor list.
00243  *
00244  * PUBLIC: int CDB___db_cprint __P((DB *));
00245  */
00246 int
00247 CDB___db_cprint(dbp)
00248 DB *dbp;
00249 {
00250 static const FN fn[] = {
00251 { DBC_ACTIVE, "active" },
00252 { DBC_OPD, "off-page-dup" },
00253 { DBC_RECOVER, "recover" },
00254 { DBC_RMW, "read-modify-write" },
00255 { DBC_WRITECURSOR, "write cursor" },
00256 { DBC_WRITEDUP, "internally dup'ed write cursor" },
00257 { DBC_WRITER, "short-term write cursor" },
00258 { 0, NULL }
00259 };
00260 DBC *dbc;
00261 DBC_INTERNAL *cp;
00262 char *s;
00263 
00264 MUTEX_THREAD_LOCK(dbp->mutexp);
00265 for (dbc = TAILQ_FIRST(&dbp->active_queue);
00266 dbc != NULL; dbc = TAILQ_NEXT(dbc, links)) {
00267 switch (dbc->dbtype) {
00268 case DB_BTREE:
00269 s = "btree";
00270 break;
00271 case DB_HASH:
00272 s = "hash";
00273 break;
00274 case DB_RECNO:
00275 s = "recno";
00276 break;
00277 case DB_QUEUE:
00278 s = "queue";
00279 break;
00280 default:
00281 DB_ASSERT(0);
00282 return (1);
00283 }
00284 cp = dbc->internal;
00285 fprintf(stderr, "%s/%#0lx: opd: %#0lx\n",
00286 s, P_TO_ULONG(dbc), P_TO_ULONG(cp->opd));
00287 fprintf(stderr, "\ttxn: %#0lx lid: %lu locker: %lu\n",
00288 P_TO_ULONG(dbc->txn),
00289 (u_long)dbc->lid, (u_long)dbc->locker);
00290 fprintf(stderr, "\troot: %lu page/index: %lu/%lu",
00291 (u_long)cp->root, (u_long)cp->pgno, (u_long)cp->indx);
00292 CDB___db_prflags(dbc->flags, fn, stderr);
00293 fprintf(stderr, "\n");
00294 
00295 if (dbp->type == DB_BTREE)
00296 CDB___bam_cprint(dbc);
00297 }
00298 for (dbc = TAILQ_FIRST(&dbp->free_queue);
00299 dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
00300 fprintf(stderr, "free: %#0lx ", P_TO_ULONG(dbc));
00301 fprintf(stderr, "\n");
00302 MUTEX_THREAD_UNLOCK(dbp->mutexp);
00303 
00304 return (0);
00305 }
00306 #endif /* DEBUG */
00307 
00308 /*
00309  * db_fd --
00310  * Return a file descriptor for flock'ing.
00311  *
00312  * PUBLIC: int CDB___db_fd __P((DB *, int *));
00313  */
00314 int
00315 CDB___db_fd(dbp, fdp)
00316 DB *dbp;
00317 int *fdp;
00318 {
00319 DB_FH *fhp;
00320 int ret;
00321 
00322 PANIC_CHECK(dbp->dbenv);
00323 DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->fd");
00324 
00325 /*
00326  * XXX
00327  * Truly spectacular layering violation.
00328  */
00329 if ((ret = CDB___mp_xxx_fh(dbp->mpf, &fhp)) != 0)
00330 return (ret);
00331 
00332 if (F_ISSET(fhp, DB_FH_VALID)) {
00333 *fdp = fhp->fd;
00334 return (0);
00335 } else {
00336 *fdp = -1;
00337 CDB___db_err(dbp->dbenv, "DB does not have a valid file handle.");
00338 return (ENOENT);
00339 }
00340 }
00341 
00342 /*
00343  * CDB___db_get --
00344  * Return a key/data pair.
00345  *
00346  * PUBLIC: int CDB___db_get __P((DB *, DB_TXN *, DBT *, DBT *, u_int32_t));
00347  */
00348 int
00349 CDB___db_get(dbp, txn, key, data, flags)
00350 DB *dbp;
00351 DB_TXN *txn;
00352 DBT *key, *data;
00353 u_int32_t flags;
00354 {
00355 DBC *dbc;
00356 int ret, t_ret;
00357 
00358 PANIC_CHECK(dbp->dbenv);
00359 DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->get");
00360 
00361 if ((ret = CDB___db_getchk(dbp, key, data, flags)) != 0)
00362 return (ret);
00363 
00364 if ((ret = dbp->cursor(dbp, txn, &dbc, 0)) != 0)
00365 return (ret);
00366 
00367 DEBUG_LREAD(dbc, txn, "CDB___db_get", key, NULL, flags);
00368 
00369 ret = dbc->c_get(dbc, key, data,
00370 flags == 0 || flags == DB_RMW ? flags | DB_SET : flags);
00371 
00372 if ((t_ret = CDB___db_c_close(dbc)) != 0 && ret == 0)
00373 ret = t_ret;
00374 
00375 #ifdef DEBUG
00376 switch(flags) {
00377 case 0:
00378 word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET, 1);
00379 break;
00380 case DB_NEXT:
00381 word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_NEXT, 1);
00382 break;
00383 case DB_SET_RANGE:
00384 word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_SET_RANGE, 1);
00385 break;
00386 default:
00387 word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_OTHER, 1);
00388 break;
00389 }
00390 #endif /* DEBUG */
00391 return (ret);
00392 }
00393 
00394 /*
00395  * CDB___db_put --
00396  * Store a key/data pair.
00397  *
00398  * PUBLIC: int CDB___db_put __P((DB *, DB_TXN *, DBT *, DBT *, u_int32_t));
00399  */
00400 int
00401 CDB___db_put(dbp, txn, key, data, flags)
00402 DB *dbp;
00403 DB_TXN *txn;
00404 DBT *key, *data;
00405 u_int32_t flags;
00406 {
00407 DBC *dbc;
00408 DBT tdata;
00409 int ret, t_ret;
00410 
00411 PANIC_CHECK(dbp->dbenv);
00412 DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->put");
00413 
00414 if ((ret = CDB___db_putchk(dbp, key, data,
00415 flags, F_ISSET(dbp, DB_AM_RDONLY),
00416 F_ISSET(dbp, DB_AM_DUP) || F_ISSET(key, DB_DBT_DUPOK))) != 0)
00417 return (ret);
00418 
00419 if ((ret = dbp->cursor(dbp, txn, &dbc, DB_WRITELOCK)) != 0)
00420 return (ret);
00421 
00422 DEBUG_LWRITE(dbc, txn, "CDB___db_put", key, data, flags);
00423 
00424 if (flags == DB_NOOVERWRITE) {
00425 flags = 0;
00426 /*
00427  * Set DB_DBT_USERMEM, this might be a threaded application and
00428  * the flags checking will catch us. We don't want the actual
00429  * data, so request a partial of length 0.
00430  */
00431 memset(&tdata, 0, sizeof(tdata));
00432 F_SET(&tdata, DB_DBT_USERMEM | DB_DBT_PARTIAL);
00433 
00434 /*
00435  * If we're doing page-level locking, set the read-modify-write
00436  * flag, we're going to overwrite immediately.
00437  */
00438 if ((ret = dbc->c_get(dbc, key, &tdata,
00439 DB_SET | (STD_LOCKING(dbc) ? DB_RMW : 0))) == 0)
00440 ret = DB_KEYEXIST;
00441 else if (ret == DB_NOTFOUND)
00442 ret = 0;
00443 }
00444 if (ret == 0)
00445 ret = dbc->c_put(dbc,
00446 key, data, flags == 0 ? DB_KEYLAST : flags);
00447 
00448 if ((t_ret = CDB___db_c_close(dbc)) != 0 && ret == 0)
00449 ret = t_ret;
00450 
00451 #ifdef DEBUG
00452 word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_PUT, 1);
00453 #endif /* DEBUG */
00454 
00455 return (ret);
00456 }
00457 
00458 /*
00459  * CDB___db_sync --
00460  * Flush the database cache.
00461  *
00462  * PUBLIC: int CDB___db_sync __P((DB *, u_int32_t));
00463  */
00464 int
00465 CDB___db_sync(dbp, flags)
00466 DB *dbp;
00467 u_int32_t flags;
00468 {
00469 int ret, t_ret;
00470 
00471 PANIC_CHECK(dbp->dbenv);
00472 DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->sync");
00473 
00474 if ((ret = CDB___db_syncchk(dbp, flags)) != 0)
00475 return (ret);
00476 
00477 /* Read-only trees never need to be sync'd. */
00478 if (F_ISSET(dbp, DB_AM_RDONLY))
00479 return (0);
00480 
00481 /* If it's a Recno tree, write the backing source text file. */
00482 if (dbp->type == DB_RECNO)
00483 ret = CDB___ram_writeback(dbp);
00484 
00485 /* If the tree was never backed by a database file, we're done. */
00486 if (F_ISSET(dbp, DB_AM_INMEM))
00487 return (0);
00488 
00489 /* Flush any dirty pages from the cache to the backing file. */
00490 if ((t_ret = CDB_memp_fsync(dbp->mpf)) != 0 && ret == 0)
00491 ret = t_ret;
00492 return (ret);
00493 }