ag/cpython-source-deps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bsddb
cpython-source-deps/rep/rep_backup.c
Zachary Ware 8f590873d0 Import BSDDB 4.7.25 (as of svn r89086)
2017-09-04 13:40:25 -05:00

2897 lines
79 KiB
C
Raw Permalink Blame History

/*-
* See the file LICENSE for redistribution information.
*
* Copyright (c) 2004,2008 Oracle. All rights reserved.
*
* $Id: rep_backup.c 63573 2008-05-23 21:43:21Z trent.nelson $
*/
#include "db_config.h"
#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/db_am.h"
#include "dbinc/fop.h"
#include "dbinc/lock.h"
#include "dbinc/log.h"
#include "dbinc/mp.h"
#include "dbinc/qam.h"
#include "dbinc/txn.h"
static int __rep_check_uid __P((ENV *, u_int8_t *, u_int8_t *, u_int32_t,
u_int8_t *));
static int __rep_filedone __P((ENV *, DB_THREAD_INFO *ip, int,
REP *, __rep_fileinfo_args *, u_int32_t));
static int __rep_find_dbs __P((ENV *, u_int32_t, u_int8_t **, size_t *,
size_t *, u_int32_t *));
static int __rep_get_fileinfo __P((ENV *, const char *,
const char *, __rep_fileinfo_args *, u_int8_t *, u_int32_t *));
static int __rep_get_file_list __P((ENV *,
DB_FH *, u_int32_t, u_int32_t *, DBT *));
static int __rep_log_setup __P((ENV *,
REP *, u_int32_t, u_int32_t, DB_LSN *));
static int __rep_mpf_open __P((ENV *, DB_MPOOLFILE **,
__rep_fileinfo_args *, u_int32_t));
static int __rep_nextfile __P((ENV *, int, REP *));
static int __rep_page_gap __P((ENV *,
REP *, __rep_fileinfo_args *, u_int32_t));
static int __rep_page_sendpages __P((ENV *, DB_THREAD_INFO *, int,
__rep_control_args *, __rep_fileinfo_args *, DB_MPOOLFILE *, DB *));
static int __rep_queue_filedone __P((ENV *,
DB_THREAD_INFO *, REP *, __rep_fileinfo_args *));
static int __rep_remove_all __P((ENV *, u_int32_t, DBT *));
static int __rep_remove_file __P((ENV *, u_int8_t *, const char *,
u_int32_t, u_int32_t));
static int __rep_remove_logs __P((ENV *));
static int __rep_remove_by_list __P((ENV *, u_int32_t,
u_int8_t *, u_int32_t, u_int32_t));
static int __rep_remove_by_prefix __P((ENV *, const char *, const char *,
size_t, APPNAME));
static int __rep_walk_dir __P((ENV *, const char *, u_int32_t, u_int8_t **,
u_int8_t *, size_t *, size_t *, u_int32_t *));
static int __rep_write_page __P((ENV *,
DB_THREAD_INFO *, REP *, __rep_fileinfo_args *));
/*
* __rep_update_req -
* Process an update_req and send the file information to the client.
*
* PUBLIC: int __rep_update_req __P((ENV *, __rep_control_args *, int));
*/
int
__rep_update_req(env, rp, eid)
ENV *env;
__rep_control_args *rp;
int eid;
{
DBT updbt, vdbt;
DB_LOG *dblp;
DB_LOGC *logc;
DB_LSN lsn;
__rep_update_args u_args;
size_t filelen, filesz, updlen;
u_int32_t filecnt, flag, version;
u_int8_t *buf, *fp;
int ret, t_ret;
/*
* Allocate enough for all currently open files and then some.
* Optimize for the common use of having most databases open.
* Allocate dbentry_cnt * 2 plus an estimated 60 bytes per
* file for the filename/path (or multiplied by 120).
*
* The data we send looks like this:
* __rep_update_args
* __rep_fileinfo_args
* __rep_fileinfo_args
* ...
*/
dblp = env->lg_handle;
logc = NULL;
filecnt = 0;
filelen = 0;
updlen = 0;
filesz = MEGABYTE;
if ((ret = __os_calloc(env, 1, filesz, &buf)) != 0)
return (ret);
/*
* First get our file information. Get in-memory files first
* then get on-disk files.
*/
fp = buf + __REP_UPDATE_SIZE;
if ((ret = __rep_find_dbs(env, rp->rep_version,
&fp, &filesz, &filelen, &filecnt)) != 0)
goto err;
/*
* Now get our first LSN. We send the lsn of the first
* non-archivable log file.
*/
flag = DB_SET;
if ((ret = __log_get_stable_lsn(env, &lsn)) != 0) {
if (ret != DB_NOTFOUND)
goto err;
/*
* If ret is DB_NOTFOUND then there is no checkpoint
* in this log, that is okay, just start at the beginning.
*/
ret = 0;
flag = DB_FIRST;
}
/*
* Now get the version number of the log file of that LSN.
*/
if ((ret = __log_cursor(env, &logc)) != 0)
goto err;
memset(&vdbt, 0, sizeof(vdbt));
/*
* Set our log cursor on the LSN we are sending. Or
* to the first LSN if we have no stable LSN.
*/
if ((ret = __logc_get(logc, &lsn, &vdbt, flag)) != 0) {
/*
* We could be racing a fresh master starting up. If we
* have no log records, assume an initial LSN and current
* log version.
*/
if (ret != DB_NOTFOUND)
goto err;
INIT_LSN(lsn);
version = DB_LOGVERSION;
} else {
if ((ret = __logc_version(logc, &version)) != 0)
goto err;
}
/*
* Package up the update information.
*/
u_args.first_lsn = lsn;
u_args.first_vers = version;
u_args.num_files = filecnt;
if ((ret = __rep_update_marshal(env, rp->rep_version,
&u_args, buf, filesz, &updlen)) != 0)
goto err;
/*
* We have all the file information now. Send it to the client.
*/
DB_INIT_DBT(updbt, buf, filelen + updlen);
LOG_SYSTEM_LOCK(env);
lsn = ((LOG *)dblp->reginfo.primary)->lsn;
LOG_SYSTEM_UNLOCK(env);
(void)__rep_send_message(
env, eid, REP_UPDATE, &lsn, &updbt, 0, 0);
err: __os_free(env, buf);
if (logc != NULL && (t_ret = __logc_close(logc)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
/*
* __rep_find_dbs -
* Walk through all the named files/databases including those in the
* environment or data_dirs and those that in named and in-memory. We
* need to open them, gather the necessary information and then close
* them.
*
* !!!
* The pointer *fp is expected to point into a buffer that may be used for an
* UPDATE message, at an offset equal to the size of __rep_update_args. This
* assumption is relied upon if the buffer is found to be too small and must be
* reallocated.
*/
static int
__rep_find_dbs(env, version, fp, fileszp, filelenp, filecntp)
ENV *env;
u_int32_t version;
u_int8_t **fp;
size_t *fileszp, *filelenp;
u_int32_t *filecntp;
{
DB_ENV *dbenv;
int ret;
char **ddir, *real_dir;
u_int8_t *origfp;
dbenv = env->dbenv;
ret = 0;
real_dir = NULL;
if (dbenv->db_data_dir == NULL) {
/*
* If we don't have a data dir, we have just the
* env home dir.
*/
ret = __rep_walk_dir(env, env->db_home, version, fp, NULL,
fileszp, filelenp, filecntp);
} else {
origfp = *fp;
for (ddir = dbenv->db_data_dir; *ddir != NULL; ++ddir) {
if ((ret = __db_appname(env, DB_APP_NONE,
*ddir, 0, NULL, &real_dir)) != 0)
break;
if ((ret = __rep_walk_dir(env, real_dir, version, fp,
origfp, fileszp, filelenp, filecntp)) != 0)
break;
__os_free(env, real_dir);
real_dir = NULL;
}
}
/* Now, collect any in-memory named databases. */
if (ret == 0)
ret = __rep_walk_dir(env, NULL, version,
fp, NULL, fileszp, filelenp, filecntp);
if (real_dir != NULL)
__os_free(env, real_dir);
return (ret);
}
/*
* __rep_walk_dir --
*
* This is the routine that walks a directory and fills in the structures
* that we use to generate messages to the client telling it what
* files are available. If the directory name is NULL, then we should
* walk the list of in-memory named files.
*/
static int
__rep_walk_dir(env, dir, version, fp, origfp, fileszp, filelenp, filecntp)
ENV *env;
const char *dir;
u_int32_t version;
u_int8_t **fp, *origfp;
size_t *fileszp, *filelenp;
u_int32_t *filecntp;
{
__rep_fileinfo_args tmpfp;
size_t len, offset;
int cnt, first_file, i, ret;
u_int8_t *rfp, uid[DB_FILE_ID_LEN];
char *file, **names, *subdb;
if (dir == NULL) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Walk_dir: Getting info for in-memory named files"));
if ((ret = __memp_inmemlist(env, &names, &cnt)) != 0)
return (ret);
} else {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Walk_dir: Getting info for dir: %s", dir));
if ((ret = __os_dirlist(env, dir, 0, &names, &cnt)) != 0)
return (ret);
}
rfp = NULL;
if (fp != NULL)
rfp = *fp;
RPRINT(env, DB_VERB_REP_SYNC, (env, "Walk_dir: Dir %s has %d files",
(dir == NULL) ? "INMEM" : dir, cnt));
first_file = 1;
for (i = 0; i < cnt; i++) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Walk_dir: File %d name: %s", i, names[i]));
/*
* Skip DB-owned files: __db*, DB_CONFIG, log*
*/
if (strncmp(names[i],
DB_REGION_PREFIX, sizeof(DB_REGION_PREFIX) - 1) == 0)
continue;
if (strncmp(names[i], "DB_CONFIG", 9) == 0)
continue;
if (strncmp(names[i], "log.", 4) == 0)
continue;
/*
* We found a file to process. Check if we need
* to allocate more space.
*/
if (dir == NULL) {
file = NULL;
subdb = names[i];
} else {
file = names[i];
subdb = NULL;
}
if ((ret = __rep_get_fileinfo(env,
file, subdb, &tmpfp, uid, filecntp)) != 0) {
/*
* If we find a file that isn't a database, skip it.
*/
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Walk_dir: File %d %s: returned error %s",
i, names[i], db_strerror(ret)));
ret = 0;
continue;
}
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Walk_dir: File %d (of %d) %s at 0x%lx: pgsize %lu, max_pgno %lu",
tmpfp.filenum, *filecntp, names[i], P_TO_ULONG(rfp),
(u_long)tmpfp.pgsize, (u_long)tmpfp.max_pgno));
/*
* Check if we already have info on this file. Since we're
* walking directories, we only need to check the first
* file to discover if we have a duplicate data_dir.
*/
if (first_file && origfp != NULL) {
/*
* If we have any file info, check if we have this uid.
*/
if (rfp != origfp &&
(ret = __rep_check_uid(env, origfp,
origfp + *filelenp, version, uid)) != 0) {
/*
* If we have this uid. Adjust the file
* count and stop processing this dir.
*/
if (ret == DB_KEYEXIST) {
ret = 0;
(*filecntp)--;
}
goto err;
}
first_file = 0;
}
DB_SET_DBT(tmpfp.info, names[i], strlen(names[i]) + 1);
DB_SET_DBT(tmpfp.uid, uid, DB_FILE_ID_LEN);
retry: ret = __rep_fileinfo_marshal(env, version,
&tmpfp, rfp, *fileszp, &len);
if (ret == ENOMEM) {
offset = (size_t)(rfp - *fp);
*fileszp *= 2;
/*
* Need to account for update info on both sides
* of the allocation.
*/
*fp -= __REP_UPDATE_SIZE;
if ((ret = __os_realloc(env, *fileszp, *fp)) != 0)
break;
*fp += __REP_UPDATE_SIZE;
rfp = *fp + offset;
/*
* Now that we've reallocated the space, try to
* store it again.
*/
goto retry;
}
rfp += len;
*fp = rfp;
*filelenp += len;
}
err:
__os_dirfree(env, names, cnt);
return (ret);
}
/*
* This function is called when we process the first file of any
* new directory for internal init. We walk the list of current
* files to see if we have already processed these files. This
* is to prevent transmitting the same file multiple times if the
* user calls env->set_data_dir on the same directory more than once.
*/
static int
__rep_check_uid(env, origfp, endfp, version, uid)
ENV *env;
u_int32_t version;
u_int8_t *origfp, *endfp, *uid;
{
__rep_fileinfo_args *rfp;
size_t filesz;
u_int8_t *fp, *fuid, *new_fp;
int ret;
ret = 0;
fp = origfp;
rfp = NULL;
/*
* We don't know how many fp's there are, so compute the maximum
* size based on the endfp and the first fp.
*/
filesz = (uintptr_t)endfp - (uintptr_t)origfp;
while (fp <= endfp) {
if ((ret = __rep_fileinfo_unmarshal(env, version,
&rfp, fp, filesz, &new_fp)) != 0) {
__db_errx(env, "rep_check_uid: Could not malloc");
goto err;
}
filesz -= (u_int32_t)(new_fp - fp);
fp = new_fp;
fuid = (u_int8_t *)rfp->uid.data;
if (memcmp(fuid, uid, DB_FILE_ID_LEN) == 0) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Check_uid: Found matching file."));
ret = DB_KEYEXIST;
goto err;
}
__os_free(env, rfp);
rfp = NULL;
}
err:
if (rfp != NULL)
__os_free(env, rfp);
return (ret);
}
static int
__rep_get_fileinfo(env, file, subdb, rfp, uid, filecntp)
ENV *env;
const char *file, *subdb;
__rep_fileinfo_args *rfp;
u_int8_t *uid;
u_int32_t *filecntp;
{
DB *dbp;
DBC *dbc;
DBMETA *dbmeta;
DB_LOCK lk;
DB_MPOOLFILE *mpf;
DB_THREAD_INFO *ip;
PAGE *pagep;
int lorder, ret, t_ret;
dbp = NULL;
dbc = NULL;
pagep = NULL;
mpf = NULL;
LOCK_INIT(lk);
ENV_GET_THREAD_INFO(env, ip);
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
goto err;
if ((ret = __db_open(dbp, ip, NULL, file, subdb, DB_UNKNOWN,
DB_RDONLY | (F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0),
0, PGNO_BASE_MD)) != 0)
goto err;
if ((ret = __db_cursor(dbp, ip, NULL, &dbc, 0)) != 0)
goto err;
if ((ret = __db_lget(
dbc, 0, dbp->meta_pgno, DB_LOCK_READ, 0, &lk)) != 0)
goto err;
if ((ret = __memp_fget(dbp->mpf, &dbp->meta_pgno, ip, dbc->txn,
0, &pagep)) != 0)
goto err;
/*
* We have the meta page. Set up our information.
*/
dbmeta = (DBMETA *)pagep;
rfp->pgno = 0;
/*
* Queue is a special-case. We need to set max_pgno to 0 so that
* the client can compute the pages from the meta-data.
*/
if (dbp->type == DB_QUEUE)
rfp->max_pgno = 0;
else
rfp->max_pgno = dbmeta->last_pgno;
rfp->pgsize = dbp->pgsize;
memcpy(uid, dbp->fileid, DB_FILE_ID_LEN);
rfp->filenum = (*filecntp)++;
rfp->type = (u_int32_t)dbp->type;
rfp->db_flags = dbp->flags;
rfp->finfo_flags = 0;
/*
* Send the lorder of this database.
*/
(void)__db_get_lorder(dbp, &lorder);
if (lorder == 1234)
FLD_SET(rfp->finfo_flags, REPINFO_DB_LITTLEENDIAN);
else
FLD_CLR(rfp->finfo_flags, REPINFO_DB_LITTLEENDIAN);
ret = __memp_fput(dbp->mpf, ip, pagep, dbc->priority);
pagep = NULL;
if ((t_ret = __LPUT(dbc, lk)) != 0 && ret == 0)
ret = t_ret;
if (ret != 0)
goto err;
err:
if ((t_ret = __LPUT(dbc, lk)) != 0 && ret == 0)
ret = t_ret;
if (pagep != NULL && (t_ret =
__memp_fput(mpf, ip, pagep, dbc->priority)) != 0 && ret == 0)
ret = t_ret;
if (dbc != NULL && (t_ret = __dbc_close(dbc)) != 0 && ret == 0)
ret = t_ret;
if (dbp != NULL && (t_ret = __db_close(dbp, NULL, 0)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
/*
* __rep_page_req
* Process a page_req and send the page information to the client.
*
* PUBLIC: int __rep_page_req __P((ENV *,
* PUBLIC: DB_THREAD_INFO *, int, __rep_control_args *, DBT *));
*/
int
__rep_page_req(env, ip, eid, rp, rec)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
__rep_control_args *rp;
DBT *rec;
{
__rep_fileinfo_args *msgfp;
DB_MPOOLFILE *mpf;
DB_REP *db_rep;
REP *rep;
int ret, t_ret;
u_int8_t *next;
db_rep = env->rep_handle;
rep = db_rep->region;
if ((ret = __rep_fileinfo_unmarshal(env, rp->rep_version,
&msgfp, rec->data, rec->size, &next)) != 0)
return (ret);
RPRINT(env, DB_VERB_REP_SYNC,
(env, "page_req: file %d page %lu to %lu",
msgfp->filenum, (u_long)msgfp->pgno, (u_long)msgfp->max_pgno));
/*
* We need to open the file and then send its pages.
* If we cannot open the file, we send REP_FILE_FAIL.
*/
RPRINT(env, DB_VERB_REP_SYNC,
(env, "page_req: Open %d via mpf_open", msgfp->filenum));
if ((ret = __rep_mpf_open(env, &mpf, msgfp, 0)) != 0) {
RPRINT(env, DB_VERB_REP_SYNC,
(env, "page_req: Open %d failed", msgfp->filenum));
if (F_ISSET(rep, REP_F_MASTER))
(void)__rep_send_message(env, eid, REP_FILE_FAIL,
NULL, rec, 0, 0);
else
ret = DB_NOTFOUND;
goto err;
}
ret = __rep_page_sendpages(env, ip, eid, rp, msgfp, mpf, NULL);
t_ret = __memp_fclose(mpf, 0);
if (ret == 0 && t_ret != 0)
ret = t_ret;
err:
__os_free(env, msgfp);
return (ret);
}
static int
__rep_page_sendpages(env, ip, eid, rp, msgfp, mpf, dbp)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
__rep_control_args *rp;
__rep_fileinfo_args *msgfp;
DB_MPOOLFILE *mpf;
DB *dbp;
{
DB *qdbp;
DBC *qdbc;
DBT lockdbt, msgdbt;
DB_LOCK lock;
DB_LOCKER *locker;
DB_LOCK_ILOCK lock_obj;
DB_LOG *dblp;
DB_LSN lsn;
DB_REP *db_rep;
PAGE *pagep;
REP *rep;
REP_BULK bulk;
REP_THROTTLE repth;
db_pgno_t p;
uintptr_t bulkoff;
size_t len, msgsz;
u_int32_t bulkflags, use_bulk;
int opened, ret, t_ret;
u_int8_t *buf;
db_rep = env->rep_handle;
rep = db_rep->region;
locker = NULL;
opened = 0;
t_ret = 0;
qdbp = NULL;
qdbc = NULL;
buf = NULL;
bulk.addr = NULL;
use_bulk = FLD_ISSET(rep->config, REP_C_BULK);
if (msgfp->type == (u_int32_t)DB_QUEUE) {
if (dbp == NULL) {
if ((ret = __db_create_internal(&qdbp, env, 0)) != 0)
goto err;
/*
* We need to check whether this is in-memory so that
* we pass the name correctly as either the file or
* the database name.
*/
if ((ret = __db_open(qdbp, ip, NULL,
FLD_ISSET(msgfp->db_flags, DB_AM_INMEM) ?
NULL : msgfp->info.data,
FLD_ISSET(msgfp->db_flags, DB_AM_INMEM) ?
msgfp->info.data : NULL,
DB_UNKNOWN,
DB_RDONLY | (F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0),
0, PGNO_BASE_MD)) != 0)
goto err;
opened = 1;
} else
qdbp = dbp;
if ((ret = __db_cursor(qdbp, ip, NULL, &qdbc, 0)) != 0)
goto err;
}
msgsz = __REP_FILEINFO_SIZE + DB_FILE_ID_LEN + msgfp->pgsize;
if ((ret = __os_calloc(env, 1, msgsz, &buf)) != 0)
goto err;
memset(&msgdbt, 0, sizeof(msgdbt));
RPRINT(env, DB_VERB_REP_SYNC,
(env, "sendpages: file %d page %lu to %lu",
msgfp->filenum, (u_long)msgfp->pgno, (u_long)msgfp->max_pgno));
memset(&repth, 0, sizeof(repth));
/*
* If we're doing bulk transfer, allocate a bulk buffer to put our
* pages in. We still need to initialize the throttle info
* because if we encounter a page larger than our entire bulk
* buffer, we need to send it as a singleton.
*
* Use a local var so that we don't need to worry if someone else
* turns on/off bulk in the middle of our call here.
*/
if (use_bulk && (ret = __rep_bulk_alloc(env, &bulk, eid,
&bulkoff, &bulkflags, REP_BULK_PAGE)) != 0)
goto err;
REP_SYSTEM_LOCK(env);
repth.gbytes = rep->gbytes;
repth.bytes = rep->bytes;
repth.type = REP_PAGE;
repth.data_dbt = &msgdbt;
REP_SYSTEM_UNLOCK(env);
/*
* Set up locking.
*/
LOCK_INIT(lock);
memset(&lock_obj, 0, sizeof(lock_obj));
if ((ret = __lock_id(env, NULL, &locker)) != 0)
goto err;
memcpy(lock_obj.fileid, mpf->fileid, DB_FILE_ID_LEN);
lock_obj.type = DB_PAGE_LOCK;
memset(&lockdbt, 0, sizeof(lockdbt));
lockdbt.data = &lock_obj;
lockdbt.size = sizeof(lock_obj);
for (p = msgfp->pgno; p <= msgfp->max_pgno; p++) {
/*
* We're not waiting for the lock, if we cannot get
* the lock for this page, skip it. The gap
* code will rerequest it.
*/
lock_obj.pgno = p;
if ((ret = __lock_get(env, locker, DB_LOCK_NOWAIT, &lockdbt,
DB_LOCK_READ, &lock)) != 0) {
/*
* Continue if we couldn't get the lock.
*/
if (ret == DB_LOCK_NOTGRANTED) {
ret = 0;
continue;
}
/*
* Otherwise we have an error.
*/
goto err;
}
if (msgfp->type == (u_int32_t)DB_QUEUE && p != 0)
#ifdef HAVE_QUEUE
ret = __qam_fget(qdbc, &p, DB_MPOOL_CREATE, &pagep);
#else
ret = DB_PAGE_NOTFOUND;
#endif
else
ret = __memp_fget(mpf, &p, ip, NULL,
DB_MPOOL_CREATE, &pagep);
msgfp->pgno = p;
if (ret == DB_PAGE_NOTFOUND) {
ZERO_LSN(lsn);
if (F_ISSET(rep, REP_F_MASTER)) {
ret = 0;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"sendpages: PAGE_FAIL on page %lu",
(u_long)p));
(void)__rep_send_message(env, eid,
REP_PAGE_FAIL, &lsn, &msgdbt, 0, 0);
} else
ret = DB_NOTFOUND;
goto lockerr;
} else if (ret != 0)
goto lockerr;
else
DB_SET_DBT(msgfp->info, pagep, msgfp->pgsize);
len = 0;
/*
* Send along an indication of the byte order of this mpool
* page. Since mpool always keeps pages in the native byte
* order of the local environment, this is simply my
* environment's byte order.
*
* Since pages can be served from a variety of sites when using
* client-to-client synchronization, the receiving client needs
* to know the byte order of each page independently.
*/
if (F_ISSET(env, ENV_LITTLEENDIAN))
FLD_SET(msgfp->finfo_flags, REPINFO_PG_LITTLEENDIAN);
else
FLD_CLR(msgfp->finfo_flags, REPINFO_PG_LITTLEENDIAN);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"sendpages: %lu, page lsn [%lu][%lu]", (u_long)p,
(u_long)pagep->lsn.file, (u_long)pagep->lsn.offset));
ret = __rep_fileinfo_marshal(env, rp->rep_version,
msgfp, buf, msgsz, &len);
if (msgfp->type != (u_int32_t)DB_QUEUE || p == 0)
t_ret = __memp_fput(mpf,
ip, pagep, DB_PRIORITY_UNCHANGED);
#ifdef HAVE_QUEUE
else
/*
* We don't need an #else for HAVE_QUEUE here because if
* we're not compiled with queue, then we're guaranteed
* to have set REP_PAGE_FAIL above.
*/
t_ret = __qam_fput(qdbc, p, pagep, qdbp->priority);
#endif
if (t_ret != 0 && ret == 0)
ret = t_ret;
if ((t_ret = __ENV_LPUT(env, lock)) != 0 && ret == 0)
ret = t_ret;
if (ret != 0)
goto err;
DB_ASSERT(env, len <= msgsz);
DB_SET_DBT(msgdbt, buf, len);
dblp = env->lg_handle;
LOG_SYSTEM_LOCK(env);
repth.lsn = ((LOG *)dblp->reginfo.primary)->lsn;
LOG_SYSTEM_UNLOCK(env);
/*
* If we are configured for bulk, try to send this as a bulk
* request. If not configured, or it is too big for bulk
* then just send normally.
*/
if (use_bulk)
ret = __rep_bulk_message(env, &bulk, &repth,
&repth.lsn, &msgdbt, 0);
if (!use_bulk || ret == DB_REP_BULKOVF)
ret = __rep_send_throttle(env, eid, &repth, 0, 0);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"sendpages: %lu, lsn [%lu][%lu]", (u_long)p,
(u_long)repth.lsn.file, (u_long)repth.lsn.offset));
/*
* If we have REP_PAGE_MORE we need to break this loop.
* Otherwise, with REP_PAGE, we keep going.
*/
if (repth.type == REP_PAGE_MORE || ret != 0) {
/* Ignore send failure, except to break the loop. */
if (ret == DB_REP_UNAVAIL)
ret = 0;
break;
}
}
if (0) {
lockerr: if ((t_ret = __ENV_LPUT(env, lock)) != 0 && ret == 0)
ret = t_ret;
}
err:
/*
* We're done, force out whatever remains in the bulk buffer and
* free it.
*/
if (use_bulk && bulk.addr != NULL &&
(t_ret = __rep_bulk_free(env, &bulk, 0)) != 0 && ret == 0)
ret = t_ret;
if (qdbc != NULL && (t_ret = __dbc_close(qdbc)) != 0 && ret == 0)
ret = t_ret;
if (opened && (t_ret = __db_close(qdbp, NULL, DB_NOSYNC)) != 0 &&
ret == 0)
ret = t_ret;
if (buf != NULL)
__os_free(env, buf);
if (locker != NULL && (t_ret = __lock_id_free(env,
locker)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
/*
* __rep_update_setup
* Process and setup with this file information.
*
* PUBLIC: int __rep_update_setup __P((ENV *, int, __rep_control_args *,
* PUBLIC: DBT *));
*/
int
__rep_update_setup(env, eid, rp, rec)
ENV *env;
int eid;
__rep_control_args *rp;
DBT *rec;
{
DB_LOG *dblp;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
LOG *lp;
REGENV *renv;
REGINFO *infop;
REP *rep;
__rep_update_args *rup;
int ret;
u_int32_t count;
u_int8_t *next;
db_rep = env->rep_handle;
rep = db_rep->region;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
ret = 0;
REP_SYSTEM_LOCK(env);
if (!F_ISSET(rep, REP_F_RECOVER_UPDATE) || IN_ELECTION(rep)) {
REP_SYSTEM_UNLOCK(env);
return (0);
}
F_CLR(rep, REP_F_RECOVER_UPDATE);
/*
* We know we're the first to come in here due to the
* REP_F_RECOVER_UPDATE flag.
*/
F_SET(rep, REP_F_RECOVER_PAGE);
/*
* We should not ever be in internal init with a lease granted.
*/
DB_ASSERT(env,
!IS_USING_LEASES(env) || __rep_islease_granted(env) == 0);
/*
* We do not clear REP_F_READY_* in this code.
* We'll eventually call the normal __rep_verify_match recovery
* code and that will clear all the flags and allow others to
* proceed. We only need to lockout the API here. We do not
* need to lockout other message threads.
*/
if ((ret = __rep_lockout_api(env, rep)) != 0)
goto err;
/*
* We need to update the timestamp and kill any open handles
* on this client. The files are changing completely.
*/
infop = env->reginfo;
renv = infop->primary;
(void)time(&renv->rep_timestamp);
REP_SYSTEM_UNLOCK(env);
MUTEX_LOCK(env, rep->mtx_clientdb);
__os_gettime(env, &lp->rcvd_ts, 1);
lp->wait_ts = rep->request_gap;
ZERO_LSN(lp->ready_lsn);
ZERO_LSN(lp->verify_lsn);
ZERO_LSN(lp->waiting_lsn);
ZERO_LSN(lp->max_wait_lsn);
ZERO_LSN(lp->max_perm_lsn);
if (db_rep->rep_db == NULL)
ret = __rep_client_dbinit(env, 0, REP_DB);
MUTEX_UNLOCK(env, rep->mtx_clientdb);
if (ret != 0)
goto err_nolock;
if ((ret = __rep_update_unmarshal(env, rp->rep_version,
&rup, rec->data, rec->size, &next)) != 0)
goto err_nolock;
/*
* We need to empty out any old log records that might be in the
* temp database.
*/
ENV_GET_THREAD_INFO(env, ip);
if ((ret = __db_truncate(db_rep->rep_db, ip, NULL, &count)) != 0)
goto err_nolock;
/*
* We will remove all logs we have so we need to request
* from the master's beginning.
*/
REP_SYSTEM_LOCK(env);
rep->first_lsn = rup->first_lsn;
rep->first_vers = rup->first_vers;
rep->last_lsn = rp->lsn;
rep->nfiles = rup->num_files;
__os_free(env, rup);
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Update setup for %d files.", rep->nfiles));
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Update setup: First LSN [%lu][%lu].",
(u_long)rep->first_lsn.file, (u_long)rep->first_lsn.offset));
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Update setup: Last LSN [%lu][%lu]",
(u_long)rep->last_lsn.file, (u_long)rep->last_lsn.offset));
if (rep->nfiles > 0) {
rep->infoversion = rp->rep_version;
rep->originfolen = rep->infolen =
rec->size - __REP_UPDATE_SIZE;
if ((ret = __os_calloc(env, 1, rep->infolen,
&rep->originfo)) != 0)
goto err;
memcpy(rep->originfo, next, rep->infolen);
rep->nextinfo = rep->originfo;
}
/*
* We need to remove all logs and databases the client has prior to
* getting pages for current databases on the master.
*/
if ((ret = __rep_remove_all(env, rp->rep_version, rec)) != 0)
goto err;
rep->curfile = 0;
if ((ret = __rep_nextfile(env, eid, rep)) != 0)
goto err;
if (0) {
err_nolock: REP_SYSTEM_LOCK(env);
}
err: /*
* If we get an error, we cannot leave ourselves in the RECOVER_PAGE
* state because we have no file information. That also means undo'ing
* the rep_lockout. We need to move back to the RECOVER_UPDATE stage.
*/
if (ret != 0) {
if (rep->originfo != NULL) {
__os_free(env, rep->originfo);
rep->originfo = NULL;
}
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Update_setup: Error: Clear PAGE, set UPDATE again. %s",
db_strerror(ret)));
F_CLR(rep, REP_F_RECOVER_PAGE | REP_F_READY_API |
REP_F_READY_OP);
F_SET(rep, REP_F_RECOVER_UPDATE);
}
REP_SYSTEM_UNLOCK(env);
return (ret);
}
/*
* Removes all existing logs and databases, at the start of internal init. But
* before we do, write a list of the databases onto the init file, so that in
* case we crash in the middle, we'll know how to resume when we restart.
* Finally, also write into the init file the UPDATE message from the master (in
* the "rec" DBT), which includes the (new) list of databases we intend to
* request copies of (again, so that we know what to do if we crash in the
* middle).
*
* For the sake of simplicity, these database lists are in the form of an UPDATE
* message (since we already have the mechanisms in place), even though strictly
* speaking that contains more information than we really need to store.
*/
static int
__rep_remove_all(env, msg_version, rec)
ENV *env;
u_int32_t msg_version;
DBT *rec;
{
__rep_fileinfo_args *finfo;
__rep_update_args u_args;
DB_FH *fhp;
size_t cnt, filelen, filesz, updlen;
u_int32_t bufsz, filecnt, fvers, mvers, zero;
u_int8_t *buf, *fp, *new_fp, *origfp;
int ret, t_ret;
char *fname;
finfo = NULL;
fname = NULL;
fhp = NULL;
/*
* 1. Get list of databases currently present at this client, which we
* intend to remove.
*/
filelen = 0;
filecnt = 0;
filesz = MEGABYTE;
if ((ret = __os_calloc(env, 1, filesz, &buf)) != 0)
return (ret);
origfp = fp = buf + __REP_UPDATE_SIZE;
filesz -= __REP_UPDATE_SIZE;
if ((ret = __rep_find_dbs(env, DB_REPVERSION,
&fp, &filesz, &filelen, &filecnt)) != 0)
goto out;
ZERO_LSN(u_args.first_lsn);
u_args.first_vers = 0;
u_args.num_files = filecnt;
if ((ret = __rep_update_marshal(env, DB_REPVERSION,
&u_args, buf, filesz, &updlen)) != 0)
goto out;
/*
* 2. Before removing anything, safe-store the database list, so that in
* case we crash before we've removed them all, when we restart we
* can clean up what we were doing.
*
* The original version of the file contains:
* data1 size (4 bytes)
* data1
* data2 size (possibly) (4 bytes)
* data2 (possibly)
*
* As of 4.7 the file has the following form:
* 0 (4 bytes - to indicate a new style file)
* file version (4 bytes)
* data1 version (4 bytes)
* data1 size (4 bytes)
* data1
* data2 version (possibly) (4 bytes)
* data2 size (possibly) (4 bytes)
* data2 (possibly)
*/
if ((ret = __db_appname(
env, DB_APP_NONE, REP_INITNAME, 0, NULL, &fname)) != 0)
goto out;
/* Sanity check that the write size fits into 32 bits. */
DB_ASSERT(env, updlen + filelen == (u_int32_t)(updlen + filelen));
bufsz = (u_int32_t)(updlen + filelen);
/*
* (Short writes aren't possible, so we don't have to verify 'cnt'.)
* This first list is generated internally, so it is always in
* the form of the current message version.
*/
zero = 0;
fvers = REP_INITVERSION;
mvers = DB_REPVERSION;
if ((ret = __os_open(env, fname, 0,
DB_OSO_CREATE | DB_OSO_TRUNC, DB_MODE_600, &fhp)) != 0 ||
(ret = __os_write(env, fhp, &zero, sizeof(zero), &cnt)) != 0 ||
(ret = __os_write(env, fhp, &fvers, sizeof(fvers), &cnt)) != 0 ||
(ret = __os_write(env, fhp, &mvers, sizeof(mvers), &cnt)) != 0 ||
(ret = __os_write(env, fhp, &bufsz, sizeof(bufsz), &cnt)) != 0 ||
(ret = __os_write(env, fhp, buf, bufsz, &cnt)) != 0 ||
(ret = __os_fsync(env, fhp)) != 0) {
__db_err(env, ret, "%s", fname);
goto out;
}
/*
* 3. Go ahead and remove logs and databases. The databases get removed
* according to the list we just finished safe-storing.
*/
if ((ret = __rep_remove_logs(env)) != 0)
goto out;
if ((ret = __rep_closefiles(env, 0)) != 0)
goto out;
fp = origfp;
while (filecnt-- > 0) {
if ((ret = __rep_fileinfo_unmarshal(env, DB_REPVERSION,
&finfo, fp, filesz, &new_fp)) != 0)
goto out;
if ((ret = __rep_remove_file(env, finfo->uid.data,
finfo->info.data, finfo->type, finfo->db_flags)) != 0)
goto out;
filesz -= (u_int32_t)(new_fp - fp);
fp = new_fp;
__os_free(env, finfo);
finfo = NULL;
}
/*
* 4. Safe-store the (new) list of database files we intend to copy from
* the master (again, so that in case we crash before we're finished
* doing so, we'll have enough information to clean up and start over
* again). This list is the list from the master, so it uses
* the message version.
*/
mvers = msg_version;
if ((ret = __os_write(env, fhp, &mvers, sizeof(mvers), &cnt)) != 0 ||
(ret = __os_write(env, fhp,
&rec->size, sizeof(rec->size), &cnt)) != 0 ||
(ret = __os_write(env, fhp, rec->data, rec->size, &cnt)) != 0 ||
(ret = __os_fsync(env, fhp)) != 0) {
__db_err(env, ret, "%s", fname);
goto out;
}
out:
if (fhp != NULL && (t_ret = __os_closehandle(env, fhp)) && ret == 0)
ret = t_ret;
if (fname != NULL)
__os_free(env, fname);
if (finfo != NULL)
__os_free(env, finfo);
__os_free(env, buf);
return (ret);
}
/*
* __rep_remove_logs -
* Remove our logs to prepare for internal init.
*/
static int
__rep_remove_logs(env)
ENV *env;
{
DB_LOG *dblp;
DB_LSN lsn;
LOG *lp;
u_int32_t fnum, lastfile;
int ret;
char *name;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
ret = 0;
/*
* Call memp_sync to flush any pages that might be in the log buffers
* and not on disk before we remove files on disk. If there were no
* dirty pages, the log isn't flushed. Yet the log buffers could still
* be dirty: __log_flush should take care of this rare situation.
*/
if ((ret = __memp_sync_int(env,
NULL, 0, DB_SYNC_CACHE | DB_SYNC_INTERRUPT_OK, NULL, NULL)) != 0)
return (ret);
if ((ret = __log_flush(env, NULL)) != 0)
return (ret);
/*
* Forcibly remove existing log files or reset
* the in-memory log space.
*/
if (lp->db_log_inmemory) {
ZERO_LSN(lsn);
if ((ret = __log_zero(env, &lsn)) != 0)
return (ret);
} else {
lastfile = lp->lsn.file;
for (fnum = 1; fnum <= lastfile; fnum++) {
if ((ret = __log_name(dblp, fnum, &name, NULL, 0)) != 0)
return (ret);
(void)time(&lp->timestamp);
(void)__os_unlink(env, name, 0);
__os_free(env, name);
}
}
return (0);
}
/*
* Removes a file during internal init. Assumes underlying subsystems are
* active; therefore, this can't be used for internal init crash recovery.
*/
static int
__rep_remove_file(env, uid, name, type, flags)
ENV *env;
u_int8_t *uid;
const char *name;
u_int32_t type, flags;
{
/*
* Calling __fop_remove will both purge any matching
* fileid from mpool and unlink it on disk.
*/
#ifdef HAVE_QUEUE
DB *dbp;
int ret;
/*
* Handle queue separately. __fop_remove will not
* remove extent files. Use __qam_remove to remove
* extent files that might exist under this name. Note that
* in-memory queue databases can't have extent files.
*/
if (type == (u_int32_t)DB_QUEUE && !LF_ISSET(DB_AM_INMEM)) {
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
return (ret);
/*
* At present, qam_remove expects the passed-in dbp to have a
* locker allocated, and if not, db_open allocates a locker
* which qam_remove then leaks.
*
* TODO: it would be better to avoid cobbling together this
* sequence of low-level operations, if fileops provided some
* API to allow us to remove a database without write-locking
* its handle.
*/
if ((ret = __lock_id(env, NULL, &dbp->locker)) != 0)
return (ret);
RPRINT(env, DB_VERB_REP_SYNC,
(env, "QAM: Unlink %s via __qam_remove", name));
if ((ret = __qam_remove(dbp, NULL, name, NULL)) != 0) {
RPRINT(env, DB_VERB_REP_SYNC,
(env, "qam_remove returned %d", ret));
(void)__db_close(dbp, NULL, DB_NOSYNC);
return (ret);
}
if ((ret = __db_close(dbp, NULL, DB_NOSYNC)) != 0)
return (ret);
}
#else
COMPQUIET(type, 0);
COMPQUIET(flags, 0);
#endif
/*
* We call fop_remove even if we've called qam_remove.
* That will only have removed extent files. Now
* we need to deal with the actual file itself.
*/
return (__fop_remove(env, NULL, uid, name, DB_APP_DATA, 0));
}
/*
* __rep_bulk_page
* Process a bulk page message.
*
* PUBLIC: int __rep_bulk_page __P((ENV *,
* PUBLIC: DB_THREAD_INFO *, int, __rep_control_args *, DBT *));
*/
int
__rep_bulk_page(env, ip, eid, rp, rec)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
__rep_control_args *rp;
DBT *rec;
{
__rep_control_args tmprp;
__rep_bulk_args b_args;
int ret;
u_int8_t *p, *ep;
/*
* We're going to be modifying the rp LSN contents so make
* our own private copy to play with. We need to set the
* rectype to REP_PAGE because we're calling through __rep_page
* to process each page, and lower functions make decisions
* based on the rectypes (for throttling/gap processing)
*/
memcpy(&tmprp, rp, sizeof(tmprp));
tmprp.rectype = REP_PAGE;
ret = 0;
for (ep = (u_int8_t *)rec->data + rec->size, p = (u_int8_t *)rec->data;
p < ep;) {
/*
* First thing in the buffer is the length. Then the LSN
* of this page, then the page info itself.
*/
if ((ret = __rep_bulk_unmarshal(env,
&b_args, p, rec->size, &p)) != 0)
return (ret);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"rep_bulk_page: Processing LSN [%lu][%lu]",
(u_long)tmprp.lsn.file, (u_long)tmprp.lsn.offset));
RPRINT(env, DB_VERB_REP_SYNC, (env,
"rep_bulk_page: p %#lx ep %#lx pgrec data %#lx, size %lu (%#lx)",
P_TO_ULONG(p), P_TO_ULONG(ep),
P_TO_ULONG(b_args.bulkdata.data),
(u_long)b_args.bulkdata.size,
(u_long)b_args.bulkdata.size));
/*
* Now send the page info DBT to the page processing function.
*/
ret = __rep_page(env, ip, eid, &tmprp, &b_args.bulkdata);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"rep_bulk_page: rep_page ret %d", ret));
/*
* If this set of pages is already done just return.
*/
if (ret != 0) {
if (ret == DB_REP_PAGEDONE)
ret = 0;
break;
}
}
return (ret);
}
/*
* __rep_page
* Process a page message.
*
* PUBLIC: int __rep_page __P((ENV *,
* PUBLIC: DB_THREAD_INFO *, int, __rep_control_args *, DBT *));
*/
int
__rep_page(env, ip, eid, rp, rec)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
__rep_control_args *rp;
DBT *rec;
{
DB_REP *db_rep;
DBT key, data;
REP *rep;
__rep_fileinfo_args *msgfp;
db_recno_t recno;
int ret;
ret = 0;
db_rep = env->rep_handle;
rep = db_rep->region;
if (!F_ISSET(rep, REP_F_RECOVER_PAGE))
return (DB_REP_PAGEDONE);
/*
* If we restarted internal init, it is possible to receive
* an old REP_PAGE message, while we're in the current
* stage of recovering pages. Until we have some sort of
* an init generation number, ignore any message that has
* a message LSN that is before this internal init's first_lsn.
*/
if (LOG_COMPARE(&rp->lsn, &rep->first_lsn) < 0) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE: Old page: msg LSN [%lu][%lu] first_lsn [%lu][%lu]",
(u_long)rp->lsn.file, (u_long)rp->lsn.offset,
(u_long)rep->first_lsn.file,
(u_long)rep->first_lsn.offset));
return (DB_REP_PAGEDONE);
}
if ((ret = __rep_fileinfo_unmarshal(env, rp->rep_version,
&msgfp, rec->data, rec->size, NULL)) != 0)
return (ret);
MUTEX_LOCK(env, rep->mtx_clientdb);
REP_SYSTEM_LOCK(env);
/*
* We should not ever be in internal init with a lease granted.
*/
DB_ASSERT(env,
!IS_USING_LEASES(env) || __rep_islease_granted(env) == 0);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE: Received page %lu from file %d",
(u_long)msgfp->pgno, msgfp->filenum));
/*
* Check if this page is from the file we're expecting.
* This may be an old or delayed page message.
*/
/*
* !!!
* If we allow dbrename/dbremove on the master while a client
* is updating, then we'd have to verify the file's uid here too.
*/
if (msgfp->filenum != rep->curfile) {
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Msg file %d != curfile %d",
msgfp->filenum, rep->curfile));
ret = DB_REP_PAGEDONE;
goto err;
}
/*
* We want to create/open our dbp to the database
* where we'll keep our page information.
*/
if ((ret = __rep_client_dbinit(env, 1, REP_PG)) != 0) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE: Client_dbinit %s", db_strerror(ret)));
goto err;
}
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
recno = (db_recno_t)(msgfp->pgno + 1);
key.data = &recno;
key.ulen = key.size = sizeof(db_recno_t);
key.flags = DB_DBT_USERMEM;
/*
* If we already have this page, then we don't want to bother
* rewriting it into the file. Otherwise, any other error
* we want to return.
*/
ret = __db_put(rep->file_dbp, ip, NULL, &key, &data, DB_NOOVERWRITE);
if (ret == DB_KEYEXIST) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE: Received duplicate page %lu from file %d",
(u_long)msgfp->pgno, msgfp->filenum));
STAT(rep->stat.st_pg_duplicated++);
ret = 0;
goto err;
}
if (ret != 0)
goto err;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE: Write page %lu into mpool", (u_long)msgfp->pgno));
/*
* We put the page in the database file itself.
*/
ret = __rep_write_page(env, ip, rep, msgfp);
if (ret != 0) {
/*
* We got an error storing the page, therefore, we need
* remove this page marker from the page database too.
* !!!
* I'm ignoring errors from the delete because we want to
* return the original error. If we cannot write the page
* and we cannot delete the item we just put, what should
* we do? Panic the env and return DB_RUNRECOVERY?
*/
(void)__db_del(rep->file_dbp, NULL, NULL, &key, 0);
goto err;
}
STAT(rep->stat.st_pg_records++);
rep->npages++;
/*
* Now check the LSN on the page and save it if it is later
* than the one we have.
*/
if (LOG_COMPARE(&rp->lsn, &rep->last_lsn) > 0)
rep->last_lsn = rp->lsn;
/*
* We've successfully written the page. Now we need to see if
* we're done with this file. __rep_filedone will check if we
* have all the pages expected and if so, set up for the next
* file and send out a page request for the next file's pages.
*/
ret = __rep_filedone(env, ip, eid, rep, msgfp, rp->rectype);
err: REP_SYSTEM_UNLOCK(env);
MUTEX_UNLOCK(env, rep->mtx_clientdb);
__os_free(env, msgfp);
return (ret);
}
/*
* __rep_page_fail
* Process a page fail message.
*
* PUBLIC: int __rep_page_fail __P((ENV *,
* PUBLIC: DB_THREAD_INFO *, int, __rep_control_args *, DBT *));
*/
int
__rep_page_fail(env, ip, eid, rp, rec)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
__rep_control_args *rp;
DBT *rec;
{
DB_REP *db_rep;
REP *rep;
__rep_fileinfo_args *msgfp, *rfp;
int ret;
ret = 0;
db_rep = env->rep_handle;
rep = db_rep->region;
if (!F_ISSET(rep, REP_F_RECOVER_PAGE))
return (0);
if ((ret = __rep_fileinfo_unmarshal(env, rp->rep_version,
&msgfp, rec->data, rec->size, NULL)) != 0)
return (ret);
/*
* Check if this page is from the file we're expecting.
* This may be an old or delayed page message.
*/
/*
* !!!
* If we allow dbrename/dbremove on the master while a client
* is updating, then we'd have to verify the file's uid here too.
*/
MUTEX_LOCK(env, rep->mtx_clientdb);
REP_SYSTEM_LOCK(env);
/*
* We should not ever be in internal init with a lease granted.
*/
DB_ASSERT(env,
!IS_USING_LEASES(env) || __rep_islease_granted(env) == 0);
if (msgfp->filenum != rep->curfile) {
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Msg file %d != curfile %d",
msgfp->filenum, rep->curfile));
goto out;
}
rfp = rep->curinfo;
if (rfp->type != (u_int32_t)DB_QUEUE)
--rfp->max_pgno;
else {
/*
* Queue is special. Pages at the beginning of the queue
* may disappear, as well as at the end. Use msgfp->pgno
* to adjust accordingly.
*/
RPRINT(env, DB_VERB_REP_SYNC, (env,
"page_fail: BEFORE page %lu failed. ready %lu, max %lu, npages %d",
(u_long)msgfp->pgno, (u_long)rep->ready_pg,
(u_long)rfp->max_pgno, rep->npages));
if (msgfp->pgno == rfp->max_pgno)
--rfp->max_pgno;
if (msgfp->pgno >= rep->ready_pg) {
rep->ready_pg = msgfp->pgno + 1;
rep->npages = rep->ready_pg;
}
RPRINT(env, DB_VERB_REP_SYNC, (env,
"page_fail: AFTER page %lu failed. ready %lu, max %lu, npages %d",
(u_long)msgfp->pgno, (u_long)rep->ready_pg,
(u_long)rfp->max_pgno, rep->npages));
}
/*
* We've lowered the number of pages expected. It is possible that
* this was the last page we were expecting. Now we need to see if
* we're done with this file. __rep_filedone will check if we have
* all the pages expected and if so, set up for the next file and
* send out a page request for the next file's pages.
*/
ret = __rep_filedone(env, ip, eid, rep, msgfp, REP_PAGE_FAIL);
out:
REP_SYSTEM_UNLOCK(env);
MUTEX_UNLOCK(env, rep->mtx_clientdb);
__os_free(env, msgfp);
return (ret);
}
/*
* __rep_write_page -
* Write this page into a database.
*/
static int
__rep_write_page(env, ip, rep, msgfp)
ENV *env;
DB_THREAD_INFO *ip;
REP *rep;
__rep_fileinfo_args *msgfp;
{
DB db;
DBT pgcookie;
DB_MPOOLFILE *mpf;
DB_PGINFO *pginfo;
__rep_fileinfo_args *rfp;
int ret;
void *dst;
rfp = NULL;
/*
* If this is the first page we're putting in this database, we need
* to create the mpool file. Otherwise call memp_fget to create the
* page in mpool. Then copy the data to the page, and memp_fput the
* page to give it back to mpool.
*
* We need to create the file, removing any existing file and associate
* the correct file ID with the new one.
*/
rfp = rep->curinfo;
if (rep->file_mpf == NULL) {
if (!FLD_ISSET(rfp->db_flags, DB_AM_INMEM)) {
/*
* Recreate the file on disk. We'll be putting
* the data into the file via mpool.
*/
RPRINT(env, DB_VERB_REP_SYNC, (env,
"rep_write_page: Calling fop_create for %s",
(char *)rfp->info.data));
if ((ret = __fop_create(env, NULL, NULL,
rfp->info.data, DB_APP_DATA,
env->db_mode, 0)) != 0)
goto err;
}
if ((ret =
__rep_mpf_open(env, &rep->file_mpf, rep->curinfo,
FLD_ISSET(rfp->db_flags, DB_AM_INMEM) ? DB_CREATE : 0)) != 0)
goto err;
}
/*
* Handle queue specially. If we're a QUEUE database, we need to
* use the __qam_fget/put calls. We need to use rep->queue_dbc for
* that. That dbp is opened after getting the metapage for the
* queue database. Since the meta-page is always in the queue file,
* we'll use the normal path for that first page. After that we
* can assume the dbp is opened.
*/
if (msgfp->type == (u_int32_t)DB_QUEUE && msgfp->pgno != 0) {
#ifdef HAVE_QUEUE
ret = __qam_fget(rep->queue_dbc, &msgfp->pgno,
DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &dst);
#else
/*
* This always returns an error.
*/
ret = __db_no_queue_am(env);
#endif
} else
ret = __memp_fget(rep->file_mpf, &msgfp->pgno, ip, NULL,
DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &dst);
if (ret != 0)
goto err;
/*
* Before writing this page into our local mpool, see if its byte order
* needs to be swapped. When in mpool the page should be in the native
* byte order of our local environment. But the page image we've
* received may be in the opposite order (as indicated in finfo_flags).
*/
if ((F_ISSET(env, ENV_LITTLEENDIAN) &&
!FLD_ISSET(msgfp->finfo_flags, REPINFO_PG_LITTLEENDIAN)) ||
(!F_ISSET(env, ENV_LITTLEENDIAN) &&
FLD_ISSET(msgfp->finfo_flags, REPINFO_PG_LITTLEENDIAN))) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"write_page: Page %d needs to be swapped", msgfp->pgno));
/*
* Set up a dbp to pass into the swap functions. We need
* only a few things: The environment and any special
* dbp flags and some obvious basics like db type and
* pagesize. Those flags were set back in rep_mpf_open
* and are available in the pgcookie set up with the
* mpoolfile associated with this database.
*/
memset(&db, 0, sizeof(db));
db.env = env;
db.type = (DBTYPE)msgfp->type;
db.pgsize = msgfp->pgsize;
mpf = rep->file_mpf;
if ((ret = __memp_get_pgcookie(mpf, &pgcookie)) != 0)
goto err;
pginfo = (DB_PGINFO *)pgcookie.data;
db.flags = pginfo->flags;
if ((ret = __db_pageswap(&db, msgfp->info.data, msgfp->pgsize,
NULL, 1)) != 0)
goto err;
}
memcpy(dst, msgfp->info.data, msgfp->pgsize);
#ifdef HAVE_QUEUE
if (msgfp->type == (u_int32_t)DB_QUEUE && msgfp->pgno != 0)
ret = __qam_fput(rep->queue_dbc,
msgfp->pgno, dst, rep->queue_dbc->priority);
else
#endif
ret = __memp_fput(rep->file_mpf,
ip, dst, rep->file_dbp->priority);
err: return (ret);
}
/*
* __rep_page_gap -
* After we've put the page into the database, we need to check if
* we have a page gap and whether we need to request pages.
*/
static int
__rep_page_gap(env, rep, msgfp, type)
ENV *env;
REP *rep;
__rep_fileinfo_args *msgfp;
u_int32_t type;
{
DBC *dbc;
DBT data, key;
DB_LOG *dblp;
DB_THREAD_INFO *ip;
LOG *lp;
__rep_fileinfo_args *rfp;
db_recno_t recno;
int ret, t_ret;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
ret = 0;
dbc = NULL;
/*
* We've successfully put this page into our file.
* Now we need to account for it and re-request new pages
* if necessary.
*/
/*
* We already hold both the db mutex and rep mutex.
*/
rfp = rep->curinfo;
/*
* Make sure we're still talking about the same file.
* If not, we're done here.
*/
if (rfp->filenum != msgfp->filenum) {
ret = DB_REP_PAGEDONE;
goto err;
}
/*
* We have 3 possible states:
* 1. We receive a page we already have accounted for.
* msg pgno < ready pgno
* 2. We receive a page that is beyond a gap.
* msg pgno > ready pgno
* 3. We receive the page we're expecting next.
* msg pgno == ready pgno
*/
/*
* State 1. This can happen once we put our page record into the
* database, but by the time we acquire the mutex other
* threads have already accounted for this page and moved on.
* We just want to return.
*/
if (msgfp->pgno < rep->ready_pg) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE_GAP: pgno %lu < ready %lu, waiting %lu",
(u_long)msgfp->pgno, (u_long)rep->ready_pg,
(u_long)rep->waiting_pg));
goto err;
}
/*
* State 2. This page is beyond the page we're expecting.
* We need to update waiting_pg if this page is less than
* (earlier) the current waiting_pg. There is nothing
* to do but see if we need to request.
*/
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE_GAP: pgno %lu, max_pg %lu ready %lu, waiting %lu max_wait %lu",
(u_long)msgfp->pgno, (u_long)rfp->max_pgno, (u_long)rep->ready_pg,
(u_long)rep->waiting_pg, (u_long)rep->max_wait_pg));
if (msgfp->pgno > rep->ready_pg) {
if (rep->waiting_pg == PGNO_INVALID ||
msgfp->pgno < rep->waiting_pg)
rep->waiting_pg = msgfp->pgno;
} else {
/*
* We received the page we're expecting.
*/
rep->ready_pg++;
__os_gettime(env, &lp->rcvd_ts, 1);
if (rep->ready_pg == rep->waiting_pg) {
/*
* If we get here we know we just filled a gap.
* Move the cursor to that place and then walk
* forward looking for the next gap, if it exists.
*/
lp->wait_ts = rep->request_gap;
rep->max_wait_pg = PGNO_INVALID;
/*
* We need to walk the recno database looking for the
* next page we need or expect.
*/
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
ENV_GET_THREAD_INFO(env, ip);
if ((ret = __db_cursor(rep->file_dbp, ip, NULL,
&dbc, 0)) != 0)
goto err;
/*
* Set cursor to the first waiting page.
* Page numbers/record numbers are offset by 1.
*/
recno = (db_recno_t)rep->waiting_pg + 1;
key.data = &recno;
key.ulen = key.size = sizeof(db_recno_t);
key.flags = DB_DBT_USERMEM;
/*
* We know that page is there, this should
* find the record.
*/
ret = __dbc_get(dbc, &key, &data, DB_SET);
if (ret != 0)
goto err;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE_GAP: Set cursor for ready %lu, waiting %lu",
(u_long)rep->ready_pg, (u_long)rep->waiting_pg));
}
while (ret == 0 && rep->ready_pg == rep->waiting_pg) {
rep->ready_pg++;
ret = __dbc_get(dbc, &key, &data, DB_NEXT);
/*
* If we get to the end of the list, there are no
* more gaps. Reset waiting_pg.
*/
if (ret == DB_NOTFOUND || ret == DB_KEYEMPTY) {
rep->waiting_pg = PGNO_INVALID;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE_GAP: Next cursor No next - ready %lu, waiting %lu",
(u_long)rep->ready_pg,
(u_long)rep->waiting_pg));
break;
}
/*
* Subtract 1 from waiting_pg because record numbers
* are 1-based and pages are 0-based and we added 1
* into the page number when we put it into the db.
*/
rep->waiting_pg = *(db_pgno_t *)key.data;
rep->waiting_pg--;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"PAGE_GAP: Next cursor ready %lu, waiting %lu",
(u_long)rep->ready_pg, (u_long)rep->waiting_pg));
}
}
/*
* If we filled a gap and now have the entire file, there's
* nothing to do. We're done when ready_pg is > max_pgno
* because ready_pg is larger than the last page we received.
*/
if (rep->ready_pg > rfp->max_pgno)
goto err;
/*
* Check if we need to ask for more pages.
*/
if ((rep->waiting_pg != PGNO_INVALID &&
rep->ready_pg != rep->waiting_pg) || type == REP_PAGE_MORE) {
/*
* We got a page but we may still be waiting for more.
* If we got REP_PAGE_MORE we always want to ask for more.
* We need to set rfp->pgno to the current page number
* we will use to ask for more pages.
*/
if (type == REP_PAGE_MORE)
rfp->pgno = msgfp->pgno;
if ((__rep_check_doreq(env, rep) || type == REP_PAGE_MORE) &&
((ret = __rep_pggap_req(env, rep, rfp,
(type == REP_PAGE_MORE) ? REP_GAP_FORCE : 0)) != 0))
goto err;
} else {
lp->wait_ts = rep->request_gap;
rep->max_wait_pg = PGNO_INVALID;
}
err:
if (dbc != NULL && (t_ret = __dbc_close(dbc)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
/*
* __rep_init_cleanup -
* Clean up internal initialization pieces.
*
* !!!
* Caller must hold client database mutex (mtx_clientdb) and REP_SYSTEM_LOCK.
*
* PUBLIC: int __rep_init_cleanup __P((ENV *, REP *, int));
*/
int
__rep_init_cleanup(env, rep, force)
ENV *env;
REP *rep;
int force;
{
DB *queue_dbp;
DB_LOG *dblp;
LOG *lp;
int cleanup_failure, ret, t_ret;
ret = 0;
/*
* 1. Close up the file data pointer we used.
* 2. Close/reset the page database.
* 3. Close/reset the queue database if we're forcing a cleanup.
* 4. Free current file info.
* 5. If we have all files or need to force, free original file info.
*/
if (rep->file_mpf != NULL) {
ret = __memp_fclose(rep->file_mpf, 0);
rep->file_mpf = NULL;
}
if (rep->file_dbp != NULL) {
t_ret = __db_close(rep->file_dbp, NULL, DB_NOSYNC);
rep->file_dbp = NULL;
if (t_ret != 0 && ret == 0)
ret = t_ret;
}
if (force && rep->queue_dbc != NULL) {
queue_dbp = rep->queue_dbc->dbp;
if ((t_ret = __dbc_close(rep->queue_dbc)) != 0 && ret == 0)
ret = t_ret;
rep->queue_dbc = NULL;
if ((t_ret = __db_close(queue_dbp, NULL, DB_NOSYNC)) != 0 &&
ret == 0)
ret = t_ret;
}
if (rep->curinfo != NULL) {
__os_free(env, rep->curinfo);
rep->curinfo = NULL;
}
if (F_ISSET(rep, REP_F_INTERNAL_INIT_MASK) && force) {
/*
* Clean up files involved in an interrupted internal init.
*
* 1. logs
* a) remove old log files
* b) set up initial log file #1
* 2. database files
* 3. the "init file"
*
* Steps 1 and 2 can be attempted independently. Step 1b is
* dependent on successful completion of 1a. Step 3 must not be
* done if anything fails along the way, because the init file's
* raison d'etre is to show that some files remain to be cleaned
* up.
*/
RPRINT(env, DB_VERB_REP_SYNC,
(env, "clean up interrupted internal init"));
cleanup_failure = 0;
if ((t_ret = __rep_remove_logs(env)) == 0) {
/*
* Since we have no logs, recover by making it look like
* the case when a new client first starts up, namely we
* have nothing but a fresh log file #1. This is a
* little wasteful, since we may soon remove this log
* file again. But that's OK, because this is the
* unusual case of NEWMASTER during internal init, and
* the rest of internal init doubtless dwarfs this.
*/
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
if ((t_ret = __rep_log_setup(env,
rep, 1, DB_LOGVERSION, &lp->ready_lsn)) != 0) {
cleanup_failure = 1;
if (ret == 0)
ret = t_ret;
}
} else {
cleanup_failure = 1;
if (ret == 0)
ret = t_ret;
}
if ((t_ret = __rep_remove_by_list(env, rep->infoversion,
rep->originfo, rep->originfolen, rep->nfiles)) != 0) {
cleanup_failure = 1;
if (ret == 0)
ret = t_ret;
}
if (!cleanup_failure &&
(t_ret = __rep_remove_init_file(env)) != 0) {
if (ret == 0)
ret = t_ret;
}
if (rep->originfo != NULL) {
__os_free(env, rep->originfo);
rep->originfo = NULL;
}
}
return (ret);
}
/*
* __rep_filedone -
* We need to check if we're done with the current file after
* processing the current page. Stat the database to see if
* we have all the pages. If so, we need to clean up/close
* this one, set up for the next one, and ask for its pages,
* or if this is the last file, request the log records and
* move to the REP_RECOVER_LOG state.
*/
static int
__rep_filedone(env, ip, eid, rep, msgfp, type)
ENV *env;
DB_THREAD_INFO *ip;
int eid;
REP *rep;
__rep_fileinfo_args *msgfp;
u_int32_t type;
{
__rep_fileinfo_args *rfp;
int ret;
/*
* We've put our page, now we need to do any gap processing
* that might be needed to re-request pages.
*/
ret = __rep_page_gap(env, rep, msgfp, type);
/*
* The world changed while we were doing gap processing.
* We're done here.
*/
if (ret == DB_REP_PAGEDONE)
return (0);
rfp = rep->curinfo;
/*
* max_pgno is 0-based and npages is 1-based, so we don't have
* all the pages until npages is > max_pgno.
*/
RPRINT(env, DB_VERB_REP_SYNC,
(env, "FILEDONE: have %lu pages. Need %lu.",
(u_long)rep->npages, (u_long)rfp->max_pgno + 1));
if (rep->npages <= rfp->max_pgno)
return (0);
/*
* If we're queue and we think we have all the pages for this file,
* we need to do special queue processing. Queue is handled in
* several stages.
*/
if (rfp->type == (u_int32_t)DB_QUEUE &&
((ret = __rep_queue_filedone(env, ip, rep, rfp)) !=
DB_REP_PAGEDONE))
return (ret);
/*
* We have all the pages for this file. Clean up.
*/
if ((ret = __rep_init_cleanup(env, rep, 0)) != 0)
goto err;
rep->curfile++;
ret = __rep_nextfile(env, eid, rep);
err:
return (ret);
}
/*
* Starts requesting pages for the next file in the list (if any), or if not,
* proceeds to the next stage: requesting logs.
*
* !!!
* Called with REP_SYSTEM_LOCK held.
*/
static int
__rep_nextfile(env, eid, rep)
ENV *env;
int eid;
REP *rep;
{
DBT dbt;
__rep_logreq_args lr_args;
int ret;
u_int8_t *buf, lrbuf[__REP_LOGREQ_SIZE];
size_t len, msgsz;
/*
* Always direct the next request to the master (at least nominally),
* regardless of where the current response came from. The application
* can always still redirect it to another client.
*/
if (rep->master_id != DB_EID_INVALID)
eid = rep->master_id;
if (rep->curfile == rep->nfiles) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"NEXTFILE: have %d files. RECOVER_LOG now", rep->nfiles));
/*
* Move to REP_RECOVER_LOG state.
* Request logs.
*/
/*
* We need to do a sync here so that any later opens
* can find the file and file id. We need to do it
* before we clear REP_F_RECOVER_PAGE so that we do not
* try to flush the log.
*/
if ((ret = __memp_sync_int(env, NULL, 0,
DB_SYNC_CACHE | DB_SYNC_INTERRUPT_OK, NULL, NULL)) != 0)
return (ret);
F_CLR(rep, REP_F_RECOVER_PAGE);
F_SET(rep, REP_F_RECOVER_LOG);
memset(&dbt, 0, sizeof(dbt));
lr_args.endlsn = rep->last_lsn;
if ((ret = __rep_logreq_marshal(env, &lr_args, lrbuf,
__REP_LOGREQ_SIZE, &len)) != 0)
return (ret);
DB_INIT_DBT(dbt, lrbuf, len);
REP_SYSTEM_UNLOCK(env);
if ((ret = __rep_log_setup(env, rep,
rep->first_lsn.file, rep->first_vers, NULL)) != 0)
return (ret);
RPRINT(env, DB_VERB_REP_SYNC, (env,
"NEXTFILE: LOG_REQ from LSN [%lu][%lu] to [%lu][%lu]",
(u_long)rep->first_lsn.file, (u_long)rep->first_lsn.offset,
(u_long)rep->last_lsn.file, (u_long)rep->last_lsn.offset));
(void)__rep_send_message(env, eid,
REP_LOG_REQ, &rep->first_lsn, &dbt,
REPCTL_INIT, DB_REP_ANYWHERE);
REP_SYSTEM_LOCK(env);
return (0);
}
/*
* 4. If not, set curinfo to next file and request its pages.
*/
rep->finfo = rep->nextinfo;
if ((ret = __rep_fileinfo_unmarshal(env, rep->infoversion,
&rep->curinfo, rep->finfo, rep->infolen, &rep->nextinfo)) != 0) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"NEXTINFO: Fileinfo read: %s", db_strerror(ret)));
return (ret);
}
DB_ASSERT(env, rep->curinfo->pgno == 0);
rep->infolen -= (u_int32_t)(rep->nextinfo - rep->finfo);
rep->ready_pg = 0;
rep->npages = 0;
rep->waiting_pg = PGNO_INVALID;
rep->max_wait_pg = PGNO_INVALID;
memset(&dbt, 0, sizeof(dbt));
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Next file %d: pgsize %lu, maxpg %lu", rep->curinfo->filenum,
(u_long)rep->curinfo->pgsize, (u_long)rep->curinfo->max_pgno));
msgsz = __REP_FILEINFO_SIZE +
rep->curinfo->uid.size + rep->curinfo->info.size;
if ((ret = __os_calloc(env, 1, msgsz, &buf)) != 0)
return (ret);
if ((ret = __rep_fileinfo_marshal(env, rep->infoversion,
rep->curinfo, buf, msgsz, &len)) != 0)
return (ret);
DB_INIT_DBT(dbt, buf, len);
(void)__rep_send_message(env, eid, REP_PAGE_REQ,
NULL, &dbt, 0, DB_REP_ANYWHERE);
__os_free(env, buf);
return (0);
}
/*
* __rep_mpf_open -
* Create and open the mpool file for a database.
* Used by both master and client to bring files into mpool.
*/
static int
__rep_mpf_open(env, mpfp, rfp, flags)
ENV *env;
DB_MPOOLFILE **mpfp;
__rep_fileinfo_args *rfp;
u_int32_t flags;
{
DB db;
int ret;
if ((ret = __memp_fcreate(env, mpfp)) != 0)
return (ret);
/*
* We need a dbp to pass into to __env_mpool. Set up
* only the parts that it needs.
*/
db.env = env;
db.type = (DBTYPE)rfp->type;
db.pgsize = rfp->pgsize;
memcpy(db.fileid, rfp->uid.data, DB_FILE_ID_LEN);
db.flags = rfp->db_flags;
/* We need to make sure the dbp isn't marked open. */
F_CLR(&db, DB_AM_OPEN_CALLED);
/*
* The byte order of this database may be different from my local native
* byte order. If so, set the swap bit so that the necessary swapping
* will be done during file I/O.
*/
if ((F_ISSET(env, ENV_LITTLEENDIAN) &&
!FLD_ISSET(rfp->finfo_flags, REPINFO_DB_LITTLEENDIAN)) ||
(!F_ISSET(env, ENV_LITTLEENDIAN) &&
FLD_ISSET(rfp->finfo_flags, REPINFO_DB_LITTLEENDIAN))) {
RPRINT(env, DB_VERB_REP_SYNC, (env,
"rep_mpf_open: Different endian database. Set swap bit."));
F_SET(&db, DB_AM_SWAP);
} else
F_CLR(&db, DB_AM_SWAP);
db.mpf = *mpfp;
if (F_ISSET(&db, DB_AM_INMEM))
(void)__memp_set_flags(db.mpf, DB_MPOOL_NOFILE, 1);
if ((ret = __env_mpool(&db, rfp->info.data, flags)) != 0) {
(void)__memp_fclose(db.mpf, 0);
*mpfp = NULL;
}
return (ret);
}
/*
* __rep_pggap_req -
* Request a page gap. Assumes the caller holds the rep_mutex.
*
* PUBLIC: int __rep_pggap_req __P((ENV *, REP *, __rep_fileinfo_args *,
* PUBLIC: u_int32_t));
*/
int
__rep_pggap_req(env, rep, reqfp, gapflags)
ENV *env;
REP *rep;
__rep_fileinfo_args *reqfp;
u_int32_t gapflags;
{
DBT max_pg_dbt;
__rep_fileinfo_args *tmpfp, t;
size_t len, msgsz;
u_int32_t flags;
int alloc, ret;
u_int8_t *buf;
ret = 0;
alloc = 0;
/*
* There is a window where we have to set REP_RECOVER_PAGE when
* we receive the update information to transition from getting
* file information to getting page information. However, that
* thread does release and then reacquire mutexes. So, we might
* try re-requesting before the original thread can get curinfo
* setup. If curinfo isn't set up there is nothing to do.
*/
if (rep->curinfo == NULL)
return (0);
if (reqfp == NULL) {
if ((ret = __rep_finfo_alloc(env, rep->curinfo, &tmpfp)) != 0)
return (ret);
alloc = 1;
} else {
t = *reqfp;
tmpfp = &t;
}
/*
* If we've never requested this page, then
* request everything between it and the first
* page we have. If we have requested this page
* then only request this record, not the entire gap.
*/
flags = 0;
memset(&max_pg_dbt, 0, sizeof(max_pg_dbt));
/*
* If this is a PAGE_MORE and we're forcing then we want to
* force the request to ask for the next page after this one.
*/
if (FLD_ISSET(gapflags, REP_GAP_FORCE))
tmpfp->pgno++;
else
tmpfp->pgno = rep->ready_pg;
msgsz = __REP_FILEINFO_SIZE +
tmpfp->uid.size + tmpfp->info.size;
if ((ret = __os_calloc(env, 1, msgsz, &buf)) != 0)
goto err;
if (rep->max_wait_pg == PGNO_INVALID ||
FLD_ISSET(gapflags, REP_GAP_FORCE | REP_GAP_REREQUEST)) {
/*
* Request the gap - set max to waiting_pg - 1 or if
* there is no waiting_pg, just ask for one.
*/
if (rep->waiting_pg == PGNO_INVALID) {
if (FLD_ISSET(gapflags,
REP_GAP_FORCE | REP_GAP_REREQUEST))
rep->max_wait_pg = rep->curinfo->max_pgno;
else
rep->max_wait_pg = rep->ready_pg;
} else {
/*
* If we're forcing, and waiting_pg is less than
* the page we want to start this request at, then
* we set max_wait_pg to the max pgno in the file.
*/
if (FLD_ISSET(gapflags, REP_GAP_FORCE) &&
rep->waiting_pg < tmpfp->pgno)
rep->max_wait_pg = rep->curinfo->max_pgno;
else
rep->max_wait_pg = rep->waiting_pg - 1;
}
tmpfp->max_pgno = rep->max_wait_pg;
/*
* Gap requests are "new" and can go anywhere.
*/
if (FLD_ISSET(gapflags, REP_GAP_REREQUEST))
flags = DB_REP_REREQUEST;
else
flags = DB_REP_ANYWHERE;
} else {
/*
* Request 1 page - set max to ready_pg.
*/
rep->max_wait_pg = rep->ready_pg;
tmpfp->max_pgno = rep->ready_pg;
/*
* If we're dropping to singletons, this is a rerequest.
*/
flags = DB_REP_REREQUEST;
}
if (rep->master_id != DB_EID_INVALID) {
STAT(rep->stat.st_pg_requested++);
/*
* We need to request the pages, but we need to get the
* new info into rep->finfo. Assert that the sizes never
* change. The only thing this should do is change
* the pgno field. Everything else remains the same.
*/
if ((ret = __rep_fileinfo_marshal(env, rep->infoversion,
tmpfp, buf, msgsz, &len)) == 0) {
DB_INIT_DBT(max_pg_dbt, buf, len);
DB_ASSERT(env, len == max_pg_dbt.size);
(void)__rep_send_message(env, rep->master_id,
REP_PAGE_REQ, NULL, &max_pg_dbt, 0, flags);
}
} else
(void)__rep_send_message(env, DB_EID_BROADCAST,
REP_MASTER_REQ, NULL, NULL, 0, 0);
__os_free(env, buf);
err:
if (alloc)
__os_free(env, tmpfp);
return (ret);
}
/*
* __rep_finfo_alloc -
* Allocate and initialize a fileinfo structure.
*
* PUBLIC: int __rep_finfo_alloc __P((ENV *, __rep_fileinfo_args *,
* PUBLIC: __rep_fileinfo_args **));
*/
int
__rep_finfo_alloc(env, rfpsrc, rfpp)
ENV *env;
__rep_fileinfo_args *rfpsrc, **rfpp;
{
__rep_fileinfo_args *rfp;
size_t size;
int ret;
void *uidp, *infop;
/*
* Allocate enough for the structure and the two DBT data areas.
*/
size = sizeof(__rep_fileinfo_args) + rfpsrc->uid.size +
rfpsrc->info.size;
if ((ret = __os_malloc(env, size, &rfp)) != 0)
return (ret);
/*
* Copy the structure itself, and then set the DBT data pointers
* to their space and copy the data itself as well.
*/
memcpy(rfp, rfpsrc, sizeof(__rep_fileinfo_args));
uidp = (u_int8_t *)rfp + sizeof(__rep_fileinfo_args);
rfp->uid.data = uidp;
memcpy(uidp, rfpsrc->uid.data, rfpsrc->uid.size);
infop = (u_int8_t *)uidp + rfpsrc->uid.size;
rfp->info.data = infop;
memcpy(infop, rfpsrc->info.data, rfpsrc->info.size);
*rfpp = rfp;
return (ret);
}
/*
* __rep_log_setup -
* We know our first LSN and need to reset the log subsystem
* to get our logs set up for the proper file.
*/
static int
__rep_log_setup(env, rep, file, version, lsnp)
ENV *env;
REP *rep;
u_int32_t file;
u_int32_t version;
DB_LSN *lsnp;
{
DB_LOG *dblp;
DB_LSN lsn;
DB_TXNMGR *mgr;
DB_TXNREGION *region;
LOG *lp;
int ret;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
mgr = env->tx_handle;
region = mgr->reginfo.primary;
/*
* Set up the log starting at the file number of the first LSN we
* need to get from the master.
*/
LOG_SYSTEM_LOCK(env);
if ((ret = __log_newfile(dblp, &lsn, file, version)) == 0 &&
lsnp != NULL)
*lsnp = lsn;
LOG_SYSTEM_UNLOCK(env);
/*
* We reset first_lsn to the lp->lsn. We were given the LSN of
* the checkpoint and we now need the LSN for the beginning of
* the file, which __log_newfile conveniently set up for us
* in lp->lsn.
*/
rep->first_lsn = lp->lsn;
TXN_SYSTEM_LOCK(env);
ZERO_LSN(region->last_ckp);
TXN_SYSTEM_UNLOCK(env);
return (ret);
}
/*
* __rep_queue_filedone -
* Determine if we're really done getting the pages for a queue file.
* Queue is handled in several steps.
* 1. First we get the meta page only.
* 2. We use the meta-page information to figure out first and last
* page numbers (and if queue wraps, first can be > last.
* 3. If first < last, we do a REP_PAGE_REQ for all pages.
* 4. If first > last, we REP_PAGE_REQ from first -> max page number.
* Then we'll ask for page 1 -> last.
*
* This function can return several things:
* DB_REP_PAGEDONE - if we're done with this file.
* 0 - if we're not done with this file.
* error - if we get an error doing some operations.
*
* This function will open a dbp handle to the queue file. This is needed
* by most of the QAM macros. We'll open it on the first pass through
* here and we'll close it whenever we decide we're done.
*/
static int
__rep_queue_filedone(env, ip, rep, rfp)
ENV *env;
DB_THREAD_INFO *ip;
REP *rep;
__rep_fileinfo_args *rfp;
{
#ifndef HAVE_QUEUE
COMPQUIET(ip, NULL);
COMPQUIET(rep, NULL);
COMPQUIET(rfp, NULL);
return (__db_no_queue_am(env));
#else
DB *queue_dbp;
db_pgno_t first, last;
u_int32_t flags;
int empty, ret, t_ret;
ret = 0;
queue_dbp = NULL;
if (rep->queue_dbc == NULL) {
/*
* We need to do a sync here so that the open
* can find the file and file id.
*/
if ((ret = __memp_sync_int(env, NULL, 0,
DB_SYNC_CACHE | DB_SYNC_INTERRUPT_OK, NULL, NULL)) != 0)
goto out;
if ((ret =
__db_create_internal(&queue_dbp, env, 0)) != 0)
goto out;
flags = DB_NO_AUTO_COMMIT |
(F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0);
/*
* We need to check whether this is in-memory so that we pass
* the name correctly as either the file or the database name.
*/
if ((ret = __db_open(queue_dbp, ip, NULL,
FLD_ISSET(rfp->db_flags, DB_AM_INMEM) ? NULL :
rfp->info.data,
FLD_ISSET(rfp->db_flags, DB_AM_INMEM) ? rfp->info.data :
NULL,
DB_QUEUE, flags, 0, PGNO_BASE_MD)) != 0)
goto out;
if ((ret = __db_cursor(queue_dbp,
ip, NULL, &rep->queue_dbc, 0)) != 0)
goto out;
} else
queue_dbp = rep->queue_dbc->dbp;
if ((ret = __queue_pageinfo(queue_dbp,
&first, &last, &empty, 0, 0)) != 0)
goto out;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Queue fileinfo: first %lu, last %lu, empty %d",
(u_long)first, (u_long)last, empty));
/*
* We can be at the end of 3 possible states.
* 1. We have received the meta-page and now need to get the
* rest of the pages in the database.
* 2. We have received from first -> max_pgno. We might be done,
* or we might need to ask for wrapped pages.
* 3. We have received all pages in the file. We're done.
*/
if (rfp->max_pgno == 0) {
/*
* We have just received the meta page. Set up the next
* pages to ask for and check if the file is empty.
*/
if (empty)
goto out;
if (first > last) {
rfp->max_pgno =
QAM_RECNO_PAGE(rep->queue_dbc->dbp, UINT32_MAX);
} else
rfp->max_pgno = last;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Queue fileinfo: First req: first %lu, last %lu",
(u_long)first, (u_long)rfp->max_pgno));
goto req;
} else if (rfp->max_pgno != last) {
/*
* If max_pgno != last that means we're dealing with a
* wrapped situation. Request next batch of pages.
* Set npages to 1 because we already have page 0, the
* meta-page, now we need pages 1-max_pgno.
*/
first = 1;
rfp->max_pgno = last;
RPRINT(env, DB_VERB_REP_SYNC, (env,
"Queue fileinfo: Wrap req: first %lu, last %lu",
(u_long)first, (u_long)last));
req:
/*
* Since we're simulating a "gap" to resend new PAGE_REQ
* for this file, we need to set waiting page to last + 1
* so that we'll ask for all from ready_pg -> last.
*/
rep->npages = first;
rep->ready_pg = first;
rep->waiting_pg = rfp->max_pgno + 1;
rep->max_wait_pg = PGNO_INVALID;
ret = __rep_pggap_req(env, rep, rfp, 0);
return (ret);
}
/*
* max_pgno == last
* If we get here, we have all the pages we need.
* Close the dbp and return.
*/
out:
if (rep->queue_dbc != NULL &&
(t_ret = __dbc_close(rep->queue_dbc)) != 0 && ret == 0)
ret = t_ret;
rep->queue_dbc = NULL;
if (queue_dbp != NULL &&
(t_ret = __db_close(queue_dbp, NULL, DB_NOSYNC)) != 0 && ret == 0)
ret = t_ret;
if (ret == 0)
ret = DB_REP_PAGEDONE;
return (ret);
#endif
}
/*
* PUBLIC: int __rep_remove_init_file __P((ENV *));
*/
int
__rep_remove_init_file(env)
ENV *env;
{
int ret;
char *name;
if ((ret = __db_appname(
env, DB_APP_NONE, REP_INITNAME, 0, NULL, &name)) != 0)
return (ret);
(void)__os_unlink(env, name, 0);
__os_free(env, name);
return (0);
}
/*
* Checks for the existence of the internal init flag file. If it exists, we
* remove all logs and databases, and then remove the flag file. This is
* intended to force the internal init to start over again, and thus affords
* protection against a client crashing during internal init. This function
* must be called before normal recovery in order to be properly effective.
*
* !!!
* This function should only be called during initial set-up of the environment,
* before various subsystems are initialized. It doesn't rely on the
* subsystems' code having been initialized, and it summarily deletes files "out
* from under" them, which might disturb the subsystems if they were up.
*
* PUBLIC: int __rep_reset_init __P((ENV *));
*/
int
__rep_reset_init(env)
ENV *env;
{
DB_FH *fhp;
__rep_update_args *rup;
DBT dbt;
char *allocated_dir, *dir, *init_name;
size_t cnt;
u_int32_t dbtvers, fvers, zero;
u_int8_t *next;
int ret, t_ret;
allocated_dir = NULL;
rup = NULL;
dbt.data = NULL;
if ((ret = __db_appname(
env, DB_APP_NONE, REP_INITNAME, 0, NULL, &init_name)) != 0)
return (ret);
if ((ret = __os_open(
env, init_name, 0, DB_OSO_RDONLY, DB_MODE_600, &fhp)) != 0) {
if (ret == ENOENT)
ret = 0;
goto out;
}
RPRINT(env, DB_VERB_REP_SYNC,
(env, "Cleaning up interrupted internal init"));
/* There are a few possibilities:
* 1. no init file, or less than 1 full file list
* 2. exactly one full file list
* 3. more than one, less then a second full file list
* 4. second file list in full
*
* In cases 2 or 4, we need to remove all logs, and then remove files
* according to the (most recent) file list. (In case 1 or 3, we don't
* have to do anything.)
*
* The __rep_get_file_list function takes care of folding these cases
* into two simple outcomes.
*
* As of 4.7, the first 4 bytes are 0. Read the first 4 bytes now.
* If they are non-zero it means we have an old-style init file.
* Otherwise, pass the file version in to rep_get_file_list.
*/
if ((ret = __os_read(env, fhp, &zero, sizeof(zero), &cnt)) != 0)
goto out;
/*
* If we read successfully, but not enough, then unlink the file.
*/
if (cnt != sizeof(zero))
goto rm;
if (zero != 0) {
/*
* Old style file. We have to set fvers to the 4.6
* version of the file and also rewind the file so
* that __rep_get_file_list can read out the length itself.
*/
if ((ret = __os_seek(env, fhp, 0, 0, 0)) != 0)
goto out;
fvers = REP_INITVERSION_46;
} else if ((ret = __os_read(env,
fhp, &fvers, sizeof(fvers), &cnt)) != 0)
goto out;
else if (cnt != sizeof(fvers))
goto rm;
ret = __rep_get_file_list(env, fhp, fvers, &dbtvers, &dbt);
if ((t_ret = __os_closehandle(env, fhp)) != 0 || ret != 0) {
if (ret == 0)
ret = t_ret;
goto out;
}
if (dbt.data == NULL) {
/*
* The init file did not end with an intact file list. Since we
* never start log/db removal without an intact file list
* sync'ed to the init file, this must mean we don't have any
* partial set of files to clean up. So all we need to do is
* remove the init file.
*/
goto rm;
}
/* Remove all log files. */
if (env->dbenv->db_log_dir == NULL)
dir = env->db_home;
else {
if ((ret = __db_appname(env, DB_APP_NONE,
env->dbenv->db_log_dir, 0, NULL, &dir)) != 0)
goto out;
allocated_dir = dir;
}
if ((ret = __rep_remove_by_prefix(env,
dir, LFPREFIX, sizeof(LFPREFIX)-1, DB_APP_LOG)) != 0)
goto out;
/*
* Remove databases according to the list, and queue extent files by
* searching them out on a walk through the data_dir's.
*/
if ((ret = __rep_update_unmarshal(env, dbtvers,
&rup, dbt.data, dbt.size, &next)) != 0)
goto out;
if ((ret = __rep_remove_by_list(env, dbtvers,
next, dbt.size, rup->num_files)) != 0)
goto out;
/* Here, we've established that the file exists. */
rm: (void)__os_unlink(env, init_name, 0);
out: if (rup != NULL)
__os_free(env, rup);
if (allocated_dir != NULL)
__os_free(env, allocated_dir);
if (dbt.data != NULL)
__os_free(env, dbt.data);
__os_free(env, init_name);
return (ret);
}
/*
* Reads the last fully intact file list from the init file. If the file ends
* with a partial list (or is empty), we're not interested in it. Lack of a
* full file list is indicated by a NULL dbt->data. On success, the list is
* returned in allocated space, which becomes the responsibility of the caller.
*
* The file format is a u_int32_t buffer length, in native format, followed by
* the file list itself, in the same format as in an UPDATE message (though
* many parts of it in this case are meaningless).
*/
static int
__rep_get_file_list(env, fhp, fvers, dbtvers, dbt)
ENV *env;
DB_FH *fhp;
u_int32_t fvers;
u_int32_t *dbtvers;
DBT *dbt;
{
u_int32_t length, mvers;
size_t cnt;
int i, ret;
/* At most 2 file lists: old and new. */
dbt->data = NULL;
mvers = DB_REPVERSION_46;
length = 0;
for (i = 1; i <= 2; i++) {
if (fvers >= REP_INITVERSION_47) {
if ((ret = __os_read(env, fhp, &mvers,
sizeof(mvers), &cnt)) != 0)
goto err;
if (cnt == 0 && dbt->data != NULL)
break;
if (cnt != sizeof(mvers))
goto err;
}
if ((ret = __os_read(env,
fhp, &length, sizeof(length), &cnt)) != 0)
goto err;
/*
* Reaching the end here is fine, if we've been through at least
* once already.
*/
if (cnt == 0 && dbt->data != NULL)
break;
if (cnt != sizeof(length))
goto err;
if ((ret = __os_realloc(env,
(size_t)length, &dbt->data)) != 0)
goto err;
if ((ret = __os_read(
env, fhp, dbt->data, length, &cnt)) != 0 ||
cnt != (size_t)length)
goto err;
}
*dbtvers = mvers;
dbt->size = length;
return (0);
err:
/*
* Note that it's OK to get here with a zero value in 'ret': it means we
* read less than we expected, and dbt->data == NULL indicates to the
* caller that we don't have an intact list.
*/
if (dbt->data != NULL)
__os_free(env, dbt->data);
dbt->data = NULL;
return (ret);
}
/*
* Removes every file in a given directory that matches a given prefix. Notice
* how similar this is to __rep_walk_dir.
*/
static int
__rep_remove_by_prefix(env, dir, prefix, pref_len, appname)
ENV *env;
const char *dir;
const char *prefix;
size_t pref_len;
APPNAME appname; /* What kind of name. */
{
char *namep, **names;
int cnt, i, ret;
if ((ret = __os_dirlist(env, dir, 0, &names, &cnt)) != 0)
return (ret);
for (i = 0; i < cnt; i++) {
if (strncmp(names[i], prefix, pref_len) == 0) {
if ((ret = __db_appname(env,
appname, names[i], 0, NULL, &namep)) != 0)
goto out;
(void)__os_unlink(env, namep, 0);
__os_free(env, namep);
}
}
out: __os_dirfree(env, names, cnt);
return (ret);
}
/*
* Removes database files according to the contents of a list.
*
* This function must support removal either during environment creation, or
* when an internal init is reset in the middle. This means it must work
* regardless of whether underlying subsystems are initialized. However, it may
* assume that databases are not open. That means there is no REP!
*/
static int
__rep_remove_by_list(env, version, filelist, filesz, count)
ENV *env;
u_int32_t version;
u_int8_t *filelist;
u_int32_t filesz;
u_int32_t count;
{
DB_ENV *dbenv;
__rep_fileinfo_args *rfp;
char **ddir, *dir, *namep;
u_int8_t *new_fp;
int ret;
dbenv = env->dbenv;
ret = 0;
rfp = NULL;
while (count-- > 0) {
if ((ret = __rep_fileinfo_unmarshal(env, version,
&rfp, filelist, filesz, &new_fp)) != 0)
goto out;
filesz -= (u_int32_t)(new_fp - filelist);
filelist = new_fp;
if ((ret = __db_appname(env,
DB_APP_DATA, rfp->info.data, 0, NULL, &namep)) != 0)
goto out;
(void)__os_unlink(env, namep, 0);
__os_free(env, namep);
__os_free(env, rfp);
rfp = NULL;
}
/* Notice how similar this code is to __rep_find_dbs. */
if (dbenv->db_data_dir == NULL)
ret = __rep_remove_by_prefix(env, env->db_home,
QUEUE_EXTENT_PREFIX, sizeof(QUEUE_EXTENT_PREFIX) - 1,
DB_APP_DATA);
else {
for (ddir = dbenv->db_data_dir; *ddir != NULL; ++ddir) {
if ((ret = __db_appname(env, DB_APP_NONE,
*ddir, 0, NULL, &dir)) != 0)
break;
ret = __rep_remove_by_prefix(env, dir,
QUEUE_EXTENT_PREFIX, sizeof(QUEUE_EXTENT_PREFIX)-1,
DB_APP_DATA);
__os_free(env, dir);
if (ret != 0)
break;
}
}
out:
if (rfp != NULL)
__os_free(env, rfp);
return (ret);
}
Reference in New Issue View Git Blame Copy Permalink