/*- * See the file LICENSE for redistribution information. * * Copyright (c) 2001-2005 * Sleepycat Software. All rights reserved. * * $Id: rep_method.c,v 12.18 2005/11/08 03:25:13 bostic Exp $ */ #include "db_config.h" #ifndef NO_SYSTEM_INCLUDES #include #include #include #endif #include "db_int.h" #include "dbinc/db_page.h" #include "dbinc/btree.h" #include "dbinc/log.h" #include "dbinc/txn.h" static int __rep_abort_prepared __P((DB_ENV *)); static int __rep_bt_cmp __P((DB *, const DBT *, const DBT *)); static void __rep_config_map __P((DB_ENV *, u_int32_t *, u_int32_t *)); static int __rep_restore_prepared __P((DB_ENV *)); /* * __rep_open -- * Replication-specific initialization of the DB_ENV structure. * * PUBLIC: int __rep_open __P((DB_ENV *)); */ int __rep_open(dbenv) DB_ENV *dbenv; { DB_REP *db_rep; int ret; if ((ret = __os_calloc(dbenv, 1, sizeof(DB_REP), &db_rep)) != 0) return (ret); dbenv->rep_handle = db_rep; ret = __rep_region_init(dbenv); return (ret); } /* * __rep_get_config -- * Configure the replication subsystem. * * PUBLIC: int __rep_get_config __P((DB_ENV *, u_int32_t, int *)); */ int __rep_get_config(dbenv, which, onp) DB_ENV *dbenv; u_int32_t which; int *onp; { DB_REP *db_rep; REP *rep; u_int32_t mapped; #undef OK_FLAGS #define OK_FLAGS \ (DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT | DB_REP_CONF_NOAUTOINIT \ | DB_REP_CONF_NOWAIT) PANIC_CHECK(dbenv); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_get_config", DB_INIT_REP); if (FLD_ISSET(which, ~OK_FLAGS)) return (__db_ferr(dbenv, "DB_ENV->rep_get_config", 0)); db_rep = dbenv->rep_handle; rep = db_rep->region; mapped = 0; __rep_config_map(dbenv, &which, &mapped); if (FLD_ISSET(rep->config, mapped)) *onp = 1; else *onp = 0; return (0); } /* * __rep_set_config -- * Configure the replication subsystem. * * PUBLIC: int __rep_set_config __P((DB_ENV *, u_int32_t, int)); */ int __rep_set_config(dbenv, which, on) DB_ENV *dbenv; u_int32_t which; int on; { DB_LOG *dblp; DB_REP *db_rep; LOG *lp; REP *rep; REP_BULK bulk; int ret; u_int32_t mapped, orig; #undef OK_FLAGS #define OK_FLAGS \ (DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT | DB_REP_CONF_NOAUTOINIT \ | DB_REP_CONF_NOWAIT) PANIC_CHECK(dbenv); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_config", DB_INIT_REP); if (FLD_ISSET(which, ~OK_FLAGS)) return (__db_ferr(dbenv, "DB_ENV->rep_set_config", 0)); dblp = dbenv->lg_handle; lp = dblp->reginfo.primary; db_rep = dbenv->rep_handle; rep = db_rep->region; mapped = ret = 0; __rep_config_map(dbenv, &which, &mapped); MUTEX_LOCK(dbenv, rep->mtx_clientdb); REP_SYSTEM_LOCK(dbenv); orig = rep->config; if (on) FLD_SET(rep->config, mapped); else FLD_CLR(rep->config, mapped); /* * Bulk transfer requires special processing if it is getting * toggled. */ if (FLD_ISSET(rep->config, REP_C_BULK) && !FLD_ISSET(orig, REP_C_BULK)) db_rep->bulk = R_ADDR(&dblp->reginfo, lp->bulk_buf); REP_SYSTEM_UNLOCK(dbenv); /* * If turning bulk off and it was on, send out whatever is in the * buffer already. */ if (FLD_ISSET(orig, REP_C_BULK) && !FLD_ISSET(rep->config, REP_C_BULK) && lp->bulk_off != 0) { memset(&bulk, 0, sizeof(bulk)); if (db_rep->bulk == NULL) bulk.addr = R_ADDR(&dblp->reginfo, lp->bulk_buf); else bulk.addr = db_rep->bulk; bulk.offp = &lp->bulk_off; bulk.len = lp->bulk_len; bulk.type = REP_BULK_LOG; bulk.eid = DB_EID_BROADCAST; bulk.flagsp = &lp->bulk_flags; ret = __rep_send_bulk(dbenv, &bulk, 0); } MUTEX_UNLOCK(dbenv, rep->mtx_clientdb); return (ret); } static void __rep_config_map(dbenv, inflagsp, outflagsp) DB_ENV *dbenv; u_int32_t *inflagsp, *outflagsp; { COMPQUIET(dbenv, NULL); if (FLD_ISSET(*inflagsp, DB_REP_CONF_BULK)) { FLD_SET(*outflagsp, REP_C_BULK); FLD_CLR(*inflagsp, DB_REP_CONF_BULK); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_DELAYCLIENT)) { FLD_SET(*outflagsp, REP_C_DELAYCLIENT); FLD_CLR(*inflagsp, DB_REP_CONF_DELAYCLIENT); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_NOAUTOINIT)) { FLD_SET(*outflagsp, REP_C_NOAUTOINIT); FLD_CLR(*inflagsp, DB_REP_CONF_NOAUTOINIT); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_NOWAIT)) { FLD_SET(*outflagsp, REP_C_NOWAIT); FLD_CLR(*inflagsp, DB_REP_CONF_NOWAIT); } } /* * __rep_start -- * Become a master or client, and start sending messages to participate * in the replication environment. Must be called after the environment * is open. * * We must protect rep_start, which may change the world, with the rest * of the DB library. Each API interface will count itself as it enters * the library. Rep_start checks the following: * * rep->msg_th - this is the count of threads currently in rep_process_message * rep->start_th - this is set if a thread is in rep_start. * rep->handle_cnt - number of threads actively using a dbp in library. * rep->txn_cnt - number of active txns. * REP_F_READY - Replication flag that indicates that we wish to run * recovery, and want to prohibit new transactions from entering and cause * existing ones to return immediately (with a DB_LOCK_DEADLOCK error). * * There is also the renv->rep_timestamp which is updated whenever significant * events (i.e., new masters, log rollback, etc). Upon creation, a handle * is associated with the current timestamp. Each time a handle enters the * library it must check if the handle timestamp is the same as the one * stored in the replication region. This prevents the use of handles on * clients that reference non-existent files whose creation was backed out * during a synchronizing recovery. * * PUBLIC: int __rep_start __P((DB_ENV *, DBT *, u_int32_t)); */ int __rep_start(dbenv, dbt, flags) DB_ENV *dbenv; DBT *dbt; u_int32_t flags; { DB_LOG *dblp; DB_LSN lsn; DB_REP *db_rep; REP *rep; u_int32_t repflags; int announce, init_db, redo_prepared, ret, role_chg; int sleep_cnt, t_ret; #ifdef DIAGNOSTIC DB_MSGBUF mb; #endif PANIC_CHECK(dbenv); ENV_ILLEGAL_BEFORE_OPEN(dbenv, "DB_ENV->rep_start"); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_start", DB_INIT_REP); db_rep = dbenv->rep_handle; rep = db_rep->region; if ((ret = __db_fchk(dbenv, "DB_ENV->rep_start", flags, DB_REP_CLIENT | DB_REP_MASTER)) != 0) return (ret); /* Exactly one of CLIENT and MASTER must be specified. */ if ((ret = __db_fcchk(dbenv, "DB_ENV->rep_start", flags, DB_REP_CLIENT, DB_REP_MASTER)) != 0) return (ret); if (!LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER)) { __db_err(dbenv, "DB_ENV->rep_start: replication mode must be specified"); return (EINVAL); } /* We need a transport function. */ if (dbenv->rep_send == NULL) { __db_err(dbenv, "DB_ENV->set_rep_transport must be called before DB_ENV->rep_start"); return (EINVAL); } /* * If we are about to become (or stay) a master. Let's flush the log * to close any potential holes that might happen when upgrading from * client to master status. */ if (LF_ISSET(DB_REP_MASTER) && (ret = __log_flush(dbenv, NULL)) != 0) return (ret); REP_SYSTEM_LOCK(dbenv); /* * We only need one thread to start-up replication, so if * there is another thread in rep_start, we'll let it finish * its work and have this thread simply return. */ if (rep->start_th != 0) { /* * There is already someone in rep_start. Return. */ RPRINT(dbenv, rep, (dbenv, &mb, "Thread already in rep_start")); goto err; } else rep->start_th = 1; role_chg = (!F_ISSET(rep, REP_F_MASTER) && LF_ISSET(DB_REP_MASTER)) || (!F_ISSET(rep, REP_F_CLIENT) && LF_ISSET(DB_REP_CLIENT)); /* * Wait for any active txns or mpool ops to complete, and * prevent any new ones from occurring, only if we're * changing roles. If we are not changing roles, then we * only need to coordinate with msg_th. */ if (role_chg) { if ((ret = __rep_lockout(dbenv, rep, 0)) != 0) goto errunlock; } else { for (sleep_cnt = 0; rep->msg_th != 0;) { if (++sleep_cnt % 60 == 0) __db_err(dbenv, "DB_ENV->rep_start waiting %d minutes for replication message thread", sleep_cnt / 60); REP_SYSTEM_UNLOCK(dbenv); __os_sleep(dbenv, 1, 0); REP_SYSTEM_LOCK(dbenv); } } if (rep->eid == DB_EID_INVALID) rep->eid = dbenv->rep_eid; if (LF_ISSET(DB_REP_MASTER)) { if (role_chg) { /* * If we're upgrading from having been a client, * preclose, so that we close our temporary database * and any files we opened while doing a rep_apply. * If we don't we can infinitely leak file ids if * the master crashed with files open (the likely * case). If we don't close them we can run into * problems if we try to remove that file or long * running applications end up with an unbounded * number of used fileids, each getting written * on checkpoint. Just close them. */ if ((ret = __rep_preclose(dbenv)) != 0) goto errunlock; } redo_prepared = 0; if (!F_ISSET(rep, REP_F_MASTER)) { /* Master is not yet set. */ if (role_chg) { if (rep->w_gen > rep->recover_gen) rep->gen = ++rep->w_gen; else if (rep->gen > rep->recover_gen) rep->gen++; else rep->gen = rep->recover_gen + 1; /* * There could have been any number of failed * elections, so jump the gen if we need to now. */ if (rep->egen > rep->gen) rep->gen = rep->egen; redo_prepared = 1; } else if (rep->gen == 0) rep->gen = rep->recover_gen + 1; if (F_ISSET(rep, REP_F_MASTERELECT)) { __rep_elect_done(dbenv, rep); F_CLR(rep, REP_F_MASTERELECT); } if (rep->egen <= rep->gen) rep->egen = rep->gen + 1; RPRINT(dbenv, rep, (dbenv, &mb, "New master gen %lu, egen %lu", (u_long)rep->gen, (u_long)rep->egen)); } rep->master_id = rep->eid; /* * Note, setting flags below implicitly clears out * REP_F_NOARCHIVE, REP_F_INIT and REP_F_READY. */ rep->flags = REP_F_MASTER; rep->start_th = 0; REP_SYSTEM_UNLOCK(dbenv); dblp = (DB_LOG *)dbenv->lg_handle; LOG_SYSTEM_LOCK(dbenv); lsn = ((LOG *)dblp->reginfo.primary)->lsn; LOG_SYSTEM_UNLOCK(dbenv); /* * Send the NEWMASTER message first so that clients know * subsequent messages are coming from the right master. * We need to perform all actions below no master what * regarding errors. */ (void)__rep_send_message(dbenv, DB_EID_BROADCAST, REP_NEWMASTER, &lsn, NULL, 0, 0); ret = 0; if (role_chg) { ret = __txn_reset(dbenv); REP_SYSTEM_LOCK(dbenv); F_CLR(rep, REP_F_READY); rep->in_recovery = 0; REP_SYSTEM_UNLOCK(dbenv); } /* * Take a transaction checkpoint so that our new generation * number get written to the log. */ if ((t_ret = __txn_checkpoint(dbenv, 0, 0, DB_FORCE)) != 0 && ret == 0) ret = t_ret; if (redo_prepared && (t_ret = __rep_restore_prepared(dbenv)) != 0 && ret == 0) ret = t_ret; } else { init_db = 0; announce = role_chg || rep->master_id == DB_EID_INVALID; /* * If we're changing roles from master to client or if * we never were any role at all, we need to init the db. */ if (role_chg || !F_ISSET(rep, REP_F_CLIENT)) { rep->master_id = DB_EID_INVALID; init_db = 1; } /* Zero out everything except recovery and tally flags. */ repflags = F_ISSET(rep, REP_F_NOARCHIVE | REP_F_RECOVER_MASK | REP_F_TALLY); FLD_SET(repflags, REP_F_CLIENT); rep->flags = repflags; REP_SYSTEM_UNLOCK(dbenv); /* * Abort any prepared transactions that were restored * by recovery. We won't be able to create any txns of * our own until they're resolved, but we can't resolve * them ourselves; the master has to. If any get * resolved as commits, we'll redo them when commit * records come in. Aborts will simply be ignored. */ if ((ret = __rep_abort_prepared(dbenv)) != 0) goto errlock; MUTEX_LOCK(dbenv, rep->mtx_clientdb); ret = __rep_client_dbinit(dbenv, init_db, REP_DB); MUTEX_UNLOCK(dbenv, rep->mtx_clientdb); if (ret != 0) goto errlock; REP_SYSTEM_LOCK(dbenv); rep->start_th = 0; if (role_chg) { F_CLR(rep, REP_F_READY); rep->in_recovery = 0; } REP_SYSTEM_UNLOCK(dbenv); /* * If this client created a newly replicated environment, * then announce the existence of this client. The master * should respond with a message that will tell this client * the current generation number and the current LSN. This * will allow the client to either perform recovery or * simply join in. */ if (announce) (void)__rep_send_message(dbenv, DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0); else (void)__rep_send_message(dbenv, DB_EID_BROADCAST, REP_ALIVE_REQ, NULL, NULL, 0, 0); } if (0) { /* * We have separate labels for errors. If we're returning an * error before we've set start_th, we use 'err'. If * we are erroring while holding the region mutex, then we use * 'errunlock' label. If we're erroring without holding the rep * mutex we must use 'errlock'. */ errlock: REP_SYSTEM_LOCK(dbenv); errunlock: rep->start_th = 0; if (role_chg) { F_CLR(rep, REP_F_READY); rep->in_recovery = 0; } err: REP_SYSTEM_UNLOCK(dbenv); } return (ret); } /* * __rep_client_dbinit -- * * Initialize the LSN database on the client side. This is called from the * client initialization code. The startup flag value indicates if * this is the first thread/process starting up and therefore should create * the LSN database. This routine must be called once by each process acting * as a client. * * Assumes caller holds appropriate mutex. * * PUBLIC: int __rep_client_dbinit __P((DB_ENV *, int, repdb_t)); */ int __rep_client_dbinit(dbenv, startup, which) DB_ENV *dbenv; int startup; repdb_t which; { DB_REP *db_rep; DB *dbp, **rdbpp; REP *rep; int ret, t_ret; u_int32_t flags; const char *name; PANIC_CHECK(dbenv); db_rep = dbenv->rep_handle; rep = db_rep->region; dbp = NULL; #define REPDBNAME "__db.rep.db" #define REPPAGENAME "__db.reppg.db" if (which == REP_DB) { name = REPDBNAME; rdbpp = &db_rep->rep_db; } else { name = REPPAGENAME; rdbpp = &rep->file_dbp; } /* Check if this has already been called on this environment. */ if (*rdbpp != NULL) return (0); if (startup) { if ((ret = db_create(&dbp, dbenv, 0)) != 0) goto err; /* * Ignore errors, because if the file doesn't exist, this * is perfectly OK. */ (void)__db_remove(dbp, NULL, name, NULL, DB_FORCE); } if ((ret = db_create(&dbp, dbenv, 0)) != 0) goto err; if (which == REP_DB && (ret = __bam_set_bt_compare(dbp, __rep_bt_cmp)) != 0) goto err; /* Allow writes to this database on a client. */ F_SET(dbp, DB_AM_CL_WRITER); flags = DB_NO_AUTO_COMMIT | (startup ? DB_CREATE : 0) | (F_ISSET(dbenv, DB_ENV_THREAD) ? DB_THREAD : 0); if ((ret = __db_open(dbp, NULL, name, NULL, (which == REP_DB ? DB_BTREE : DB_RECNO), flags, 0, PGNO_BASE_MD)) != 0) goto err; *rdbpp= dbp; if (0) { err: if (dbp != NULL && (t_ret = __db_close(dbp, NULL, DB_NOSYNC)) != 0 && ret == 0) ret = t_ret; *rdbpp = NULL; } return (ret); } /* * __rep_bt_cmp -- * * Comparison function for the LSN table. We use the entire control * structure as a key (for simplicity, so we don't have to merge the * other fields in the control with the data field), but really only * care about the LSNs. */ static int __rep_bt_cmp(dbp, dbt1, dbt2) DB *dbp; const DBT *dbt1, *dbt2; { DB_LSN lsn1, lsn2; REP_CONTROL *rp1, *rp2; COMPQUIET(dbp, NULL); rp1 = dbt1->data; rp2 = dbt2->data; (void)__ua_memcpy(&lsn1, &rp1->lsn, sizeof(DB_LSN)); (void)__ua_memcpy(&lsn2, &rp2->lsn, sizeof(DB_LSN)); if (lsn1.file > lsn2.file) return (1); if (lsn1.file < lsn2.file) return (-1); if (lsn1.offset > lsn2.offset) return (1); if (lsn1.offset < lsn2.offset) return (-1); return (0); } /* * __rep_abort_prepared -- * Abort any prepared transactions that recovery restored. * * This is used by clients that have just run recovery, since * they cannot/should not call txn_recover and handle prepared transactions * themselves. */ static int __rep_abort_prepared(dbenv) DB_ENV *dbenv; { #define PREPLISTSIZE 50 DB_PREPLIST prep[PREPLISTSIZE], *p; DB_TXNMGR *mgr; DB_TXNREGION *region; int do_aborts, ret; long count, i; u_int32_t op; mgr = dbenv->tx_handle; region = mgr->reginfo.primary; do_aborts = 0; TXN_SYSTEM_LOCK(dbenv); if (region->stat.st_nrestores != 0) do_aborts = 1; TXN_SYSTEM_UNLOCK(dbenv); if (do_aborts) { op = DB_FIRST; do { if ((ret = __txn_recover(dbenv, prep, PREPLISTSIZE, &count, op)) != 0) return (ret); for (i = 0; i < count; i++) { p = &prep[i]; if ((ret = __txn_abort(p->txn)) != 0) return (ret); } op = DB_NEXT; } while (count == PREPLISTSIZE); } return (0); } /* * __rep_restore_prepared -- * Restore to a prepared state any prepared but not yet committed * transactions. * * This performs, in effect, a "mini-recovery"; it is called from * __rep_start by newly upgraded masters. There may be transactions that an * old master prepared but did not resolve, which we need to restore to an * active state. */ static int __rep_restore_prepared(dbenv) DB_ENV *dbenv; { DB_LOGC *logc; DB_LSN ckp_lsn, lsn; DB_TXNHEAD *txninfo; DBT rec; __txn_ckp_args *ckp_args; __txn_regop_args *regop_args; __txn_xa_regop_args *prep_args; int ret, t_ret; u_int32_t hi_txn, low_txn, rectype, status; txninfo = NULL; ckp_args = NULL; prep_args = NULL; regop_args = NULL; ZERO_LSN(ckp_lsn); ZERO_LSN(lsn); if ((ret = __log_cursor(dbenv, &logc)) != 0) return (ret); /* * We need to consider the set of records between the most recent * checkpoint LSN and the end of the log; any txn in that * range, and only txns in that range, could still have been * active, and thus prepared but not yet committed (PBNYC), * when the old master died. * * Find the most recent checkpoint LSN, and get the record there. * If there is no checkpoint in the log, start off by getting * the very first record in the log instead. */ memset(&rec, 0, sizeof(DBT)); if ((ret = __txn_getckp(dbenv, &lsn)) == 0) { if ((ret = __log_c_get(logc, &lsn, &rec, DB_SET)) != 0) { __db_err(dbenv, "Checkpoint record at LSN [%lu][%lu] not found", (u_long)lsn.file, (u_long)lsn.offset); goto err; } if ((ret = __txn_ckp_read(dbenv, rec.data, &ckp_args)) != 0) { __db_err(dbenv, "Invalid checkpoint record at [%lu][%lu]", (u_long)lsn.file, (u_long)lsn.offset); goto err; } ckp_lsn = ckp_args->ckp_lsn; __os_free(dbenv, ckp_args); if ((ret = __log_c_get(logc, &ckp_lsn, &rec, DB_SET)) != 0) { __db_err(dbenv, "Checkpoint LSN record [%lu][%lu] not found", (u_long)ckp_lsn.file, (u_long)ckp_lsn.offset); goto err; } } else if ((ret = __log_c_get(logc, &lsn, &rec, DB_FIRST)) != 0) { if (ret == DB_NOTFOUND) { /* An empty log means no PBNYC txns. */ ret = 0; goto done; } __db_err(dbenv, "Attempt to get first log record failed"); goto err; } /* * We use the same txnlist infrastructure that recovery does; * it demands an estimate of the high and low txnids for * initialization. * * First, the low txnid. */ do { /* txnid is after rectype, which is a u_int32. */ memcpy(&low_txn, (u_int8_t *)rec.data + sizeof(u_int32_t), sizeof(low_txn)); if (low_txn != 0) break; } while ((ret = __log_c_get(logc, &lsn, &rec, DB_NEXT)) == 0); /* If there are no txns, there are no PBNYC txns. */ if (ret == DB_NOTFOUND) { ret = 0; goto done; } else if (ret != 0) goto err; /* Now, the high txnid. */ if ((ret = __log_c_get(logc, &lsn, &rec, DB_LAST)) != 0) { /* * Note that DB_NOTFOUND is unacceptable here because we * had to have looked at some log record to get this far. */ __db_err(dbenv, "Final log record not found"); goto err; } do { /* txnid is after rectype, which is a u_int32. */ memcpy(&hi_txn, (u_int8_t *)rec.data + sizeof(u_int32_t), sizeof(hi_txn)); if (hi_txn != 0) break; } while ((ret = __log_c_get(logc, &lsn, &rec, DB_PREV)) == 0); if (ret == DB_NOTFOUND) { ret = 0; goto done; } else if (ret != 0) goto err; /* We have a high and low txnid. Initialise the txn list. */ if ((ret = __db_txnlist_init(dbenv, low_txn, hi_txn, NULL, &txninfo)) != 0) goto err; /* * Now, walk backward from the end of the log to ckp_lsn. Any * prepares that we hit without first hitting a commit or * abort belong to PBNYC txns, and we need to apply them and * restore them to a prepared state. * * Note that we wind up applying transactions out of order. * Since all PBNYC txns still held locks on the old master and * were isolated, this should be safe. */ for (ret = __log_c_get(logc, &lsn, &rec, DB_LAST); ret == 0 && log_compare(&lsn, &ckp_lsn) > 0; ret = __log_c_get(logc, &lsn, &rec, DB_PREV)) { memcpy(&rectype, rec.data, sizeof(rectype)); switch (rectype) { case DB___txn_regop: /* * It's a commit or abort--but we don't care * which! Just add it to the list of txns * that are resolved. */ if ((ret = __txn_regop_read(dbenv, rec.data, ®op_args)) != 0) goto err; ret = __db_txnlist_find(dbenv, txninfo, regop_args->txnid->txnid, &status); if (ret == DB_NOTFOUND) ret = __db_txnlist_add(dbenv, txninfo, regop_args->txnid->txnid, regop_args->opcode, &lsn); else if (ret != 0) goto err; __os_free(dbenv, regop_args); break; case DB___txn_xa_regop: /* * It's a prepare. If its not aborted and * we haven't put the txn on our list yet, it * hasn't been resolved, so apply and restore it. */ if ((ret = __txn_xa_regop_read(dbenv, rec.data, &prep_args)) != 0) goto err; ret = __db_txnlist_find(dbenv, txninfo, prep_args->txnid->txnid, &status); if (ret == DB_NOTFOUND) { if (prep_args->opcode == TXN_ABORT) ret = __db_txnlist_add(dbenv, txninfo, prep_args->txnid->txnid, prep_args->opcode, &lsn); else if ((ret = __rep_process_txn(dbenv, &rec)) == 0) ret = __txn_restore_txn(dbenv, &lsn, prep_args); } else if (ret != 0) goto err; __os_free(dbenv, prep_args); break; default: continue; } } /* It's not an error to have hit the beginning of the log. */ if (ret == DB_NOTFOUND) ret = 0; done: err: t_ret = __log_c_close(logc); if (txninfo != NULL) __db_txnlist_end(dbenv, txninfo); return (ret == 0 ? t_ret : ret); } /* * PUBLIC: int __rep_get_limit __P((DB_ENV *, u_int32_t *, u_int32_t *)); */ int __rep_get_limit(dbenv, gbytesp, bytesp) DB_ENV *dbenv; u_int32_t *gbytesp, *bytesp; { DB_REP *db_rep; REP *rep; PANIC_CHECK(dbenv); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_get_limit", DB_INIT_REP); if (!REP_ON(dbenv)) { __db_err(dbenv, "DB_ENV->get_rep_limit: database environment not properly initialized"); return (__db_panic(dbenv, EINVAL)); } db_rep = dbenv->rep_handle; rep = db_rep->region; if (gbytesp != NULL) *gbytesp = rep->gbytes; if (bytesp != NULL) *bytesp = rep->bytes; return (0); } /* * __rep_set_limit -- * Set a limit on the amount of data that will be sent during a single * invocation of __rep_process_message. * * PUBLIC: int __rep_set_limit __P((DB_ENV *, u_int32_t, u_int32_t)); */ int __rep_set_limit(dbenv, gbytes, bytes) DB_ENV *dbenv; u_int32_t gbytes, bytes; { DB_REP *db_rep; REP *rep; PANIC_CHECK(dbenv); ENV_ILLEGAL_BEFORE_OPEN(dbenv, "DB_ENV->rep_set_limit"); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_set_limit", DB_INIT_REP); if (!REP_ON(dbenv)) { __db_err(dbenv, "DB_ENV->set_rep_limit: database environment not properly initialized"); return (__db_panic(dbenv, EINVAL)); } db_rep = dbenv->rep_handle; rep = db_rep->region; REP_SYSTEM_LOCK(dbenv); if (bytes > GIGABYTE) { gbytes += bytes / GIGABYTE; bytes = bytes % GIGABYTE; } rep->gbytes = gbytes; rep->bytes = bytes; REP_SYSTEM_UNLOCK(dbenv); return (0); } /* * __rep_set_request -- * Set the minimum and maximum number of log records that we wait * before retransmitting. * * !!! * UNDOCUMENTED. * * PUBLIC: int __rep_set_request __P((DB_ENV *, u_int32_t, u_int32_t)); */ int __rep_set_request(dbenv, min, max) DB_ENV *dbenv; u_int32_t min, max; { LOG *lp; DB_LOG *dblp; DB_REP *db_rep; REP *rep; PANIC_CHECK(dbenv); ENV_ILLEGAL_BEFORE_OPEN(dbenv, "DB_ENV->rep_set_request"); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_set_request", DB_INIT_REP); if (!REP_ON(dbenv)) { __db_err(dbenv, "DB_ENV->set_rep_request: database environment not properly initialized"); return (__db_panic(dbenv, EINVAL)); } db_rep = dbenv->rep_handle; rep = db_rep->region; /* * We acquire the mtx_region or mtx_clientdb mutexes as needed. */ REP_SYSTEM_LOCK(dbenv); rep->request_gap = min; rep->max_gap = max; REP_SYSTEM_UNLOCK(dbenv); MUTEX_LOCK(dbenv, rep->mtx_clientdb); dblp = dbenv->lg_handle; if (dblp != NULL && (lp = dblp->reginfo.primary) != NULL) { lp->wait_recs = 0; lp->rcvd_recs = 0; } MUTEX_UNLOCK(dbenv, rep->mtx_clientdb); return (0); } /* * __rep_set_transport -- * Set the transport function for replication. * * PUBLIC: int __rep_set_rep_transport __P((DB_ENV *, int, * PUBLIC: int (*)(DB_ENV *, const DBT *, const DBT *, const DB_LSN *, * PUBLIC: int, u_int32_t))); */ int __rep_set_rep_transport(dbenv, eid, f_send) DB_ENV *dbenv; int eid; int (*f_send) __P((DB_ENV *, const DBT *, const DBT *, const DB_LSN *, int, u_int32_t)); { PANIC_CHECK(dbenv); if (f_send == NULL) { __db_err(dbenv, "DB_ENV->set_rep_transport: no send function specified"); return (EINVAL); } if (eid < 0) { __db_err(dbenv, "DB_ENV->set_rep_transport: eid must be greater than or equal to 0"); return (EINVAL); } dbenv->rep_send = f_send; dbenv->rep_eid = eid; return (0); } /* * __rep_flush -- * Re-push the last log record to all clients, in case they've lost * messages and don't know it. * * PUBLIC: int __rep_flush __P((DB_ENV *)); */ int __rep_flush(dbenv) DB_ENV *dbenv; { DBT rec; DB_LOGC *logc; DB_LSN lsn; int ret, t_ret; PANIC_CHECK(dbenv); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_flush", DB_INIT_REP); if ((ret = __log_cursor(dbenv, &logc)) != 0) return (ret); memset(&rec, 0, sizeof(rec)); memset(&lsn, 0, sizeof(lsn)); if ((ret = __log_c_get(logc, &lsn, &rec, DB_LAST)) != 0) goto err; (void)__rep_send_message(dbenv, DB_EID_BROADCAST, REP_LOG, &lsn, &rec, 0, 0); err: if ((t_ret = __log_c_close(logc)) != 0 && ret == 0) ret = t_ret; return (ret); } /* * __rep_sync -- * Force a synchronization to occur between this client and the master. * This is the other half of configuring DELAYCLIENT. * * PUBLIC: int __rep_sync __P((DB_ENV *, u_int32_t)); */ int __rep_sync(dbenv, flags) DB_ENV *dbenv; u_int32_t flags; { DB_LOG *dblp; DB_LSN lsn; DB_REP *db_rep; LOG *lp; REP *rep; int master; u_int32_t type; COMPQUIET(flags, 0); PANIC_CHECK(dbenv); ENV_REQUIRES_CONFIG(dbenv, dbenv->rep_handle, "rep_sync", DB_INIT_REP); dblp = dbenv->lg_handle; lp = dblp->reginfo.primary; db_rep = dbenv->rep_handle; rep = db_rep->region; /* * Simple cases. If we're not in the DELAY state we have nothing * to do. If we don't know who the master is, send a MASTER_REQ. */ MUTEX_LOCK(dbenv, rep->mtx_clientdb); lsn = lp->verify_lsn; MUTEX_UNLOCK(dbenv, rep->mtx_clientdb); REP_SYSTEM_LOCK(dbenv); master = rep->master_id; if (master == DB_EID_INVALID) { REP_SYSTEM_UNLOCK(dbenv); (void)__rep_send_message(dbenv, DB_EID_BROADCAST, REP_MASTER_REQ, NULL, NULL, 0, 0); return (0); } /* * We want to hold the rep mutex to test and then clear the * DELAY flag. Racing threads in here could otherwise result * in dual data streams. */ if (!F_ISSET(rep, REP_F_DELAY)) { REP_SYSTEM_UNLOCK(dbenv); return (0); } /* * If we get here, we clear the delay flag and kick off a * synchronization. From this point forward, we will * synchronize until the next time the master changes. */ F_CLR(rep, REP_F_DELAY); REP_SYSTEM_UNLOCK(dbenv); /* * When we set REP_F_DELAY, we set verify_lsn to the real verify * lsn if we need to verify, or we zeroed it out if this is a client * that needs to sync up from the beginning. So, send the type * of message now that __rep_new_master delayed sending. */ if (IS_ZERO_LSN(lsn)) type = REP_ALL_REQ; else type = REP_VERIFY_REQ; (void)__rep_send_message(dbenv, master, type, &lsn, NULL, 0, DB_REP_ANYWHERE); return (0); }