diff options
| author | unknown <ram@mysql.r18.ru> | 2002-10-30 15:57:05 +0400 |
|---|---|---|
| committer | unknown <ram@mysql.r18.ru> | 2002-10-30 15:57:05 +0400 |
| commit | 155e78f014de1a2e259ae5119f4621fbb210a784 (patch) | |
| tree | 6881a3cca88bea0bb9eeffd5aae34be437152786 /bdb/rep | |
| parent | b8798d25ab71436bf690ee8ae48285a655c5487e (diff) | |
| download | mariadb-git-155e78f014de1a2e259ae5119f4621fbb210a784.tar.gz | |
BDB 4.1.24
BitKeeper/deleted/.del-ex_access.wpj~3df6ae8c99bf7c5f:
Delete: bdb/build_vxworks/ex_access/ex_access.wpj
BitKeeper/deleted/.del-ex_btrec.wpj~a7622f1c6f432dc6:
Delete: bdb/build_vxworks/ex_btrec/ex_btrec.wpj
BitKeeper/deleted/.del-ex_dbclient.wpj~7345440f3b204cdd:
Delete: bdb/build_vxworks/ex_dbclient/ex_dbclient.wpj
BitKeeper/deleted/.del-ex_env.wpj~fbe1ab10b04e8b74:
Delete: bdb/build_vxworks/ex_env/ex_env.wpj
BitKeeper/deleted/.del-ex_mpool.wpj~4479cfd5c45f327d:
Delete: bdb/build_vxworks/ex_mpool/ex_mpool.wpj
BitKeeper/deleted/.del-ex_tpcb.wpj~f78093006e14bf41:
Delete: bdb/build_vxworks/ex_tpcb/ex_tpcb.wpj
BitKeeper/deleted/.del-db_buildall.dsp~bd749ff6da11682:
Delete: bdb/build_win32/db_buildall.dsp
BitKeeper/deleted/.del-cxx_app.cpp~ad8df8e0791011ed:
Delete: bdb/cxx/cxx_app.cpp
BitKeeper/deleted/.del-cxx_log.cpp~a50ff3118fe06952:
Delete: bdb/cxx/cxx_log.cpp
BitKeeper/deleted/.del-cxx_table.cpp~ecd751e79b055556:
Delete: bdb/cxx/cxx_table.cpp
BitKeeper/deleted/.del-namemap.txt~796a3acd3885d8fd:
Delete: bdb/cxx/namemap.txt
BitKeeper/deleted/.del-Design.fileop~3ca4da68f1727373:
Delete: bdb/db/Design.fileop
BitKeeper/deleted/.del-db185_int.h~61bee3736e7959ef:
Delete: bdb/db185/db185_int.h
BitKeeper/deleted/.del-acconfig.h~411e8854d67ad8b5:
Delete: bdb/dist/acconfig.h
BitKeeper/deleted/.del-mutex.m4~a13383cde18a64e1:
Delete: bdb/dist/aclocal/mutex.m4
BitKeeper/deleted/.del-options.m4~b9d0ca637213750a:
Delete: bdb/dist/aclocal/options.m4
BitKeeper/deleted/.del-programs.m4~3ce7890b47732b30:
Delete: bdb/dist/aclocal/programs.m4
BitKeeper/deleted/.del-tcl.m4~f944e2db93c3b6db:
Delete: bdb/dist/aclocal/tcl.m4
BitKeeper/deleted/.del-types.m4~59cae158c9a32cff:
Delete: bdb/dist/aclocal/types.m4
BitKeeper/deleted/.del-script~d38f6d3a4f159cb4:
Delete: bdb/dist/build/script
BitKeeper/deleted/.del-configure.in~ac795a92c8fe049c:
Delete: bdb/dist/configure.in
BitKeeper/deleted/.del-ltconfig~66bbd007d8024af:
Delete: bdb/dist/ltconfig
BitKeeper/deleted/.del-rec_ctemp~a28554362534f00a:
Delete: bdb/dist/rec_ctemp
BitKeeper/deleted/.del-s_tcl~2ffe4326459fcd9f:
Delete: bdb/dist/s_tcl
BitKeeper/deleted/.del-.IGNORE_ME~d8148b08fa7d5d15:
Delete: bdb/dist/template/.IGNORE_ME
BitKeeper/deleted/.del-btree.h~179f2aefec1753d:
Delete: bdb/include/btree.h
BitKeeper/deleted/.del-cxx_int.h~6b649c04766508f8:
Delete: bdb/include/cxx_int.h
BitKeeper/deleted/.del-db.src~6b433ae615b16a8d:
Delete: bdb/include/db.src
BitKeeper/deleted/.del-db_185.h~ad8b373d9391d35c:
Delete: bdb/include/db_185.h
BitKeeper/deleted/.del-db_am.h~a714912b6b75932f:
Delete: bdb/include/db_am.h
BitKeeper/deleted/.del-db_cxx.h~fcafadf45f5d19e9:
Delete: bdb/include/db_cxx.h
BitKeeper/deleted/.del-db_dispatch.h~6844f20f7eb46904:
Delete: bdb/include/db_dispatch.h
BitKeeper/deleted/.del-db_int.src~419a3f48b6a01da7:
Delete: bdb/include/db_int.src
BitKeeper/deleted/.del-db_join.h~76f9747a42c3399a:
Delete: bdb/include/db_join.h
BitKeeper/deleted/.del-db_page.h~e302ca3a4db3abdc:
Delete: bdb/include/db_page.h
BitKeeper/deleted/.del-db_server_int.h~e1d20b6ba3bca1ab:
Delete: bdb/include/db_server_int.h
BitKeeper/deleted/.del-db_shash.h~5fbf2d696fac90f3:
Delete: bdb/include/db_shash.h
BitKeeper/deleted/.del-db_swap.h~1e60887550864a59:
Delete: bdb/include/db_swap.h
BitKeeper/deleted/.del-db_upgrade.h~c644eee73701fc8d:
Delete: bdb/include/db_upgrade.h
BitKeeper/deleted/.del-db_verify.h~b8d6c297c61f342e:
Delete: bdb/include/db_verify.h
BitKeeper/deleted/.del-debug.h~dc2b4f2cf27ccebc:
Delete: bdb/include/debug.h
BitKeeper/deleted/.del-hash.h~2aaa548b28882dfb:
Delete: bdb/include/hash.h
BitKeeper/deleted/.del-lock.h~a761c1b7de57b77f:
Delete: bdb/include/lock.h
BitKeeper/deleted/.del-log.h~ff20184238e35e4d:
Delete: bdb/include/log.h
BitKeeper/deleted/.del-mp.h~7e317597622f3411:
Delete: bdb/include/mp.h
BitKeeper/deleted/.del-mutex.h~d3ae7a2977a68137:
Delete: bdb/include/mutex.h
BitKeeper/deleted/.del-os.h~91867cc8757cd0e3:
Delete: bdb/include/os.h
BitKeeper/deleted/.del-os_jump.h~e1b939fa5151d4be:
Delete: bdb/include/os_jump.h
BitKeeper/deleted/.del-qam.h~6fad0c1b5723d597:
Delete: bdb/include/qam.h
BitKeeper/deleted/.del-queue.h~4c72c0826c123d5:
Delete: bdb/include/queue.h
BitKeeper/deleted/.del-region.h~513fe04d977ca0fc:
Delete: bdb/include/region.h
BitKeeper/deleted/.del-shqueue.h~525fc3e6c2025c36:
Delete: bdb/include/shqueue.h
BitKeeper/deleted/.del-tcl_db.h~c536fd61a844f23f:
Delete: bdb/include/tcl_db.h
BitKeeper/deleted/.del-txn.h~c8d94b221ec147e4:
Delete: bdb/include/txn.h
BitKeeper/deleted/.del-xa.h~ecc466493aae9d9a:
Delete: bdb/include/xa.h
BitKeeper/deleted/.del-DbRecoveryInit.java~756b52601a0b9023:
Delete: bdb/java/src/com/sleepycat/db/DbRecoveryInit.java
BitKeeper/deleted/.del-DbTxnRecover.java~74607cba7ab89d6d:
Delete: bdb/java/src/com/sleepycat/db/DbTxnRecover.java
BitKeeper/deleted/.del-lock_conflict.c~fc5e0f14cf597a2b:
Delete: bdb/lock/lock_conflict.c
BitKeeper/deleted/.del-log.src~53ac9e7b5cb023f2:
Delete: bdb/log/log.src
BitKeeper/deleted/.del-log_findckp.c~24287f008916e81f:
Delete: bdb/log/log_findckp.c
BitKeeper/deleted/.del-log_rec.c~d51711f2cac09297:
Delete: bdb/log/log_rec.c
BitKeeper/deleted/.del-log_register.c~b40bb4efac75ca15:
Delete: bdb/log/log_register.c
BitKeeper/deleted/.del-Design~b3d0f179f2767b:
Delete: bdb/mp/Design
BitKeeper/deleted/.del-os_finit.c~95dbefc6fe79b26c:
Delete: bdb/os/os_finit.c
BitKeeper/deleted/.del-os_abs.c~df95d1e7db81924:
Delete: bdb/os_vxworks/os_abs.c
BitKeeper/deleted/.del-os_finit.c~803b484bdb9d0122:
Delete: bdb/os_vxworks/os_finit.c
BitKeeper/deleted/.del-os_map.c~3a6d7926398b76d3:
Delete: bdb/os_vxworks/os_map.c
BitKeeper/deleted/.del-os_finit.c~19a227c6d3c78ad:
Delete: bdb/os_win32/os_finit.c
BitKeeper/deleted/.del-log-corruption.patch~1cf2ecc7c6408d5d:
Delete: bdb/patches/log-corruption.patch
BitKeeper/deleted/.del-Btree.pm~af6d0c5eaed4a98e:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB/Btree.pm
BitKeeper/deleted/.del-BerkeleyDB.pm~7244036d4482643:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB.pm
BitKeeper/deleted/.del-BerkeleyDB.pod~e7b18fd6132448e3:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB.pod
BitKeeper/deleted/.del-Hash.pm~10292a26c06a5c95:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB/Hash.pm
BitKeeper/deleted/.del-BerkeleyDB.pod.P~79f76a1495eda203:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB.pod.P
BitKeeper/deleted/.del-BerkeleyDB.xs~80c99afbd98e392c:
Delete: bdb/perl.BerkeleyDB/BerkeleyDB.xs
BitKeeper/deleted/.del-Changes~729c1891efa60de9:
Delete: bdb/perl.BerkeleyDB/Changes
BitKeeper/deleted/.del-MANIFEST~63a1e34aecf157a0:
Delete: bdb/perl.BerkeleyDB/MANIFEST
BitKeeper/deleted/.del-Makefile.PL~c68797707d8df87a:
Delete: bdb/perl.BerkeleyDB/Makefile.PL
BitKeeper/deleted/.del-README~5f2f579b1a241407:
Delete: bdb/perl.BerkeleyDB/README
BitKeeper/deleted/.del-Todo~dca3c66c193adda9:
Delete: bdb/perl.BerkeleyDB/Todo
BitKeeper/deleted/.del-config.in~ae81681e450e0999:
Delete: bdb/perl.BerkeleyDB/config.in
BitKeeper/deleted/.del-dbinfo~28ad67d83be4f68e:
Delete: bdb/perl.BerkeleyDB/dbinfo
BitKeeper/deleted/.del-mkconsts~543ab60669c7a04e:
Delete: bdb/perl.BerkeleyDB/mkconsts
BitKeeper/deleted/.del-mkpod~182c0ca54e439afb:
Delete: bdb/perl.BerkeleyDB/mkpod
BitKeeper/deleted/.del-5.004~e008cb5a48805543:
Delete: bdb/perl.BerkeleyDB/patches/5.004
BitKeeper/deleted/.del-irix_6_5.pl~61662bb08afcdec8:
Delete: bdb/perl.BerkeleyDB/hints/irix_6_5.pl
BitKeeper/deleted/.del-solaris.pl~6771e7182394e152:
Delete: bdb/perl.BerkeleyDB/hints/solaris.pl
BitKeeper/deleted/.del-typemap~783b8f5295b05f3d:
Delete: bdb/perl.BerkeleyDB/typemap
BitKeeper/deleted/.del-5.004_01~6081ce2fff7b0bc:
Delete: bdb/perl.BerkeleyDB/patches/5.004_01
BitKeeper/deleted/.del-5.004_02~87214eac35ad9e6:
Delete: bdb/perl.BerkeleyDB/patches/5.004_02
BitKeeper/deleted/.del-5.004_03~9a672becec7cb40f:
Delete: bdb/perl.BerkeleyDB/patches/5.004_03
BitKeeper/deleted/.del-5.004_04~e326cb51af09d154:
Delete: bdb/perl.BerkeleyDB/patches/5.004_04
BitKeeper/deleted/.del-5.004_05~7ab457a1e41a92fe:
Delete: bdb/perl.BerkeleyDB/patches/5.004_05
BitKeeper/deleted/.del-5.005~f9e2d59b5964cd4b:
Delete: bdb/perl.BerkeleyDB/patches/5.005
BitKeeper/deleted/.del-5.005_01~3eb9fb7b5842ea8e:
Delete: bdb/perl.BerkeleyDB/patches/5.005_01
BitKeeper/deleted/.del-5.005_02~67477ce0bef717cb:
Delete: bdb/perl.BerkeleyDB/patches/5.005_02
BitKeeper/deleted/.del-5.005_03~c4c29a1fb21e290a:
Delete: bdb/perl.BerkeleyDB/patches/5.005_03
BitKeeper/deleted/.del-5.6.0~e1fb9897d124ee22:
Delete: bdb/perl.BerkeleyDB/patches/5.6.0
BitKeeper/deleted/.del-btree.t~e4a1a3c675ddc406:
Delete: bdb/perl.BerkeleyDB/t/btree.t
BitKeeper/deleted/.del-db-3.0.t~d2c60991d84558f2:
Delete: bdb/perl.BerkeleyDB/t/db-3.0.t
BitKeeper/deleted/.del-db-3.1.t~6ee88cd13f55e018:
Delete: bdb/perl.BerkeleyDB/t/db-3.1.t
BitKeeper/deleted/.del-db-3.2.t~f73b6461f98fd1cf:
Delete: bdb/perl.BerkeleyDB/t/db-3.2.t
BitKeeper/deleted/.del-destroy.t~cc6a2ae1980a2ecd:
Delete: bdb/perl.BerkeleyDB/t/destroy.t
BitKeeper/deleted/.del-env.t~a8604a4499c4bd07:
Delete: bdb/perl.BerkeleyDB/t/env.t
BitKeeper/deleted/.del-examples.t~2571b77c3cc75574:
Delete: bdb/perl.BerkeleyDB/t/examples.t
BitKeeper/deleted/.del-examples.t.T~8228bdd75ac78b88:
Delete: bdb/perl.BerkeleyDB/t/examples.t.T
BitKeeper/deleted/.del-examples3.t.T~66a186897a87026d:
Delete: bdb/perl.BerkeleyDB/t/examples3.t.T
BitKeeper/deleted/.del-examples3.t~fe3822ba2f2d7f83:
Delete: bdb/perl.BerkeleyDB/t/examples3.t
BitKeeper/deleted/.del-filter.t~f87b045c1b708637:
Delete: bdb/perl.BerkeleyDB/t/filter.t
BitKeeper/deleted/.del-hash.t~616bfb4d644de3a3:
Delete: bdb/perl.BerkeleyDB/t/hash.t
BitKeeper/deleted/.del-join.t~29fc39f74a83ca22:
Delete: bdb/perl.BerkeleyDB/t/join.t
BitKeeper/deleted/.del-mldbm.t~31f5015341eea040:
Delete: bdb/perl.BerkeleyDB/t/mldbm.t
BitKeeper/deleted/.del-queue.t~8f338034ce44a641:
Delete: bdb/perl.BerkeleyDB/t/queue.t
BitKeeper/deleted/.del-recno.t~d4ddbd3743add63e:
Delete: bdb/perl.BerkeleyDB/t/recno.t
BitKeeper/deleted/.del-strict.t~6885cdd2ea71ca2d:
Delete: bdb/perl.BerkeleyDB/t/strict.t
BitKeeper/deleted/.del-subdb.t~aab62a5d5864c603:
Delete: bdb/perl.BerkeleyDB/t/subdb.t
BitKeeper/deleted/.del-txn.t~65033b8558ae1216:
Delete: bdb/perl.BerkeleyDB/t/txn.t
BitKeeper/deleted/.del-unknown.t~f3710458682665e1:
Delete: bdb/perl.BerkeleyDB/t/unknown.t
BitKeeper/deleted/.del-Changes~436f74a5c414c65b:
Delete: bdb/perl.DB_File/Changes
BitKeeper/deleted/.del-DB_File.pm~ae0951c6c7665a82:
Delete: bdb/perl.DB_File/DB_File.pm
BitKeeper/deleted/.del-DB_File.xs~89e49a0b5556f1d8:
Delete: bdb/perl.DB_File/DB_File.xs
BitKeeper/deleted/.del-DB_File_BS~290fad5dbbb87069:
Delete: bdb/perl.DB_File/DB_File_BS
BitKeeper/deleted/.del-MANIFEST~90ee581572bdd4ac:
Delete: bdb/perl.DB_File/MANIFEST
BitKeeper/deleted/.del-Makefile.PL~ac0567bb5a377e38:
Delete: bdb/perl.DB_File/Makefile.PL
BitKeeper/deleted/.del-README~77e924a5a9bae6b3:
Delete: bdb/perl.DB_File/README
BitKeeper/deleted/.del-config.in~ab4c2792b86a810b:
Delete: bdb/perl.DB_File/config.in
BitKeeper/deleted/.del-dbinfo~461c43b30fab2cb:
Delete: bdb/perl.DB_File/dbinfo
BitKeeper/deleted/.del-dynixptx.pl~50dcddfae25d17e9:
Delete: bdb/perl.DB_File/hints/dynixptx.pl
BitKeeper/deleted/.del-typemap~55cffb3288a9e587:
Delete: bdb/perl.DB_File/typemap
BitKeeper/deleted/.del-version.c~a4df0e646f8b3975:
Delete: bdb/perl.DB_File/version.c
BitKeeper/deleted/.del-5.004_01~d6830d0082702af7:
Delete: bdb/perl.DB_File/patches/5.004_01
BitKeeper/deleted/.del-5.004_02~78b082dc80c91031:
Delete: bdb/perl.DB_File/patches/5.004_02
BitKeeper/deleted/.del-5.004~4411ec2e3c9e008b:
Delete: bdb/perl.DB_File/patches/5.004
BitKeeper/deleted/.del-sco.pl~1e795fe14fe4dcfe:
Delete: bdb/perl.DB_File/hints/sco.pl
BitKeeper/deleted/.del-5.004_03~33f274648b160d95:
Delete: bdb/perl.DB_File/patches/5.004_03
BitKeeper/deleted/.del-5.004_04~8f3d1b3cf18bb20a:
Delete: bdb/perl.DB_File/patches/5.004_04
BitKeeper/deleted/.del-5.004_05~9c0f02e7331e142:
Delete: bdb/perl.DB_File/patches/5.004_05
BitKeeper/deleted/.del-5.005~c2108cb2e3c8d951:
Delete: bdb/perl.DB_File/patches/5.005
BitKeeper/deleted/.del-5.005_01~3b45e9673afc4cfa:
Delete: bdb/perl.DB_File/patches/5.005_01
BitKeeper/deleted/.del-5.005_02~9fe5766bb02a4522:
Delete: bdb/perl.DB_File/patches/5.005_02
BitKeeper/deleted/.del-5.005_03~ffa1c38c19ae72ea:
Delete: bdb/perl.DB_File/patches/5.005_03
BitKeeper/deleted/.del-5.6.0~373be3a5ce47be85:
Delete: bdb/perl.DB_File/patches/5.6.0
BitKeeper/deleted/.del-db-btree.t~3231595a1c241eb3:
Delete: bdb/perl.DB_File/t/db-btree.t
BitKeeper/deleted/.del-db-hash.t~7c4ad0c795c7fad2:
Delete: bdb/perl.DB_File/t/db-hash.t
BitKeeper/deleted/.del-db-recno.t~6c2d3d80b9ba4a50:
Delete: bdb/perl.DB_File/t/db-recno.t
BitKeeper/deleted/.del-db_server.sed~cdb00ebcd48a64e2:
Delete: bdb/rpc_server/db_server.sed
BitKeeper/deleted/.del-db_server_proc.c~d46c8f409c3747f4:
Delete: bdb/rpc_server/db_server_proc.c
BitKeeper/deleted/.del-db_server_svc.sed~3f5e59f334fa4607:
Delete: bdb/rpc_server/db_server_svc.sed
BitKeeper/deleted/.del-db_server_util.c~a809f3a4629acda:
Delete: bdb/rpc_server/db_server_util.c
BitKeeper/deleted/.del-log.tcl~ff1b41f1355b97d7:
Delete: bdb/test/log.tcl
BitKeeper/deleted/.del-mpool.tcl~b0df4dc1b04db26c:
Delete: bdb/test/mpool.tcl
BitKeeper/deleted/.del-mutex.tcl~52fd5c73a150565:
Delete: bdb/test/mutex.tcl
BitKeeper/deleted/.del-txn.tcl~c4ff071550b5446e:
Delete: bdb/test/txn.tcl
BitKeeper/deleted/.del-README~e800a12a5392010a:
Delete: bdb/test/upgrade/README
BitKeeper/deleted/.del-pack-2.6.6.pl~89d5076d758d3e98:
Delete: bdb/test/upgrade/generate-2.X/pack-2.6.6.pl
BitKeeper/deleted/.del-test-2.6.patch~4a52dc83d447547b:
Delete: bdb/test/upgrade/generate-2.X/test-2.6.patch
Diffstat (limited to 'bdb/rep')
| -rw-r--r-- | bdb/rep/rep_method.c | 1144 | ||||
| -rw-r--r-- | bdb/rep/rep_record.c | 1510 | ||||
| -rw-r--r-- | bdb/rep/rep_region.c | 187 | ||||
| -rw-r--r-- | bdb/rep/rep_util.c | 867 |
4 files changed, 3708 insertions, 0 deletions
diff --git a/bdb/rep/rep_method.c b/bdb/rep/rep_method.c new file mode 100644 index 00000000000..6773a537f4f --- /dev/null +++ b/bdb/rep/rep_method.c @@ -0,0 +1,1144 @@ +/*- + * See the file LICENSE for redistribution information. + * + * Copyright (c) 2001-2002 + * Sleepycat Software. All rights reserved. + */ +#include "db_config.h" + +#ifndef lint +static const char revid[] = "$Id: rep_method.c,v 1.78 2002/09/10 12:58:07 bostic Exp $"; +#endif /* not lint */ + +#ifndef NO_SYSTEM_INCLUDES +#include <sys/types.h> + +#ifdef HAVE_RPC +#include <rpc/rpc.h> +#endif + +#include <stdlib.h> +#include <string.h> +#include <unistd.h> +#endif + +#include "db_int.h" +#include "dbinc/db_page.h" +#include "dbinc/db_am.h" +#include "dbinc/log.h" +#include "dbinc/rep.h" +#include "dbinc/txn.h" + +#ifdef HAVE_RPC +#include "dbinc_auto/db_server.h" +#include "dbinc_auto/rpc_client_ext.h" +#endif + +static int __rep_abort_prepared __P((DB_ENV *)); +static int __rep_bt_cmp __P((DB *, const DBT *, const DBT *)); +static int __rep_client_dbinit __P((DB_ENV *, int)); +static int __rep_elect __P((DB_ENV *, int, int, u_int32_t, int *)); +static int __rep_elect_init __P((DB_ENV *, DB_LSN *, int, int, int, int *)); +static int __rep_flush __P((DB_ENV *)); +static int __rep_restore_prepared __P((DB_ENV *)); +static int __rep_set_limit __P((DB_ENV *, u_int32_t, u_int32_t)); +static int __rep_set_request __P((DB_ENV *, u_int32_t, u_int32_t)); +static int __rep_set_rep_transport __P((DB_ENV *, int, + int (*)(DB_ENV *, const DBT *, const DBT *, int, u_int32_t))); +static int __rep_start __P((DB_ENV *, DBT *, u_int32_t)); +static int __rep_stat __P((DB_ENV *, DB_REP_STAT **, u_int32_t)); +static int __rep_wait __P((DB_ENV *, u_int32_t, int *, u_int32_t)); + +/* + * __rep_dbenv_create -- + * Replication-specific initialization of the DB_ENV structure. + * + * PUBLIC: int __rep_dbenv_create __P((DB_ENV *)); + */ +int +__rep_dbenv_create(dbenv) + DB_ENV *dbenv; +{ + DB_REP *db_rep; + int ret; + +#ifdef HAVE_RPC + if (F_ISSET(dbenv, DB_ENV_RPCCLIENT)) { + COMPQUIET(db_rep, NULL); + COMPQUIET(ret, 0); + dbenv->rep_elect = __dbcl_rep_elect; + dbenv->rep_flush = __dbcl_rep_flush; + dbenv->rep_process_message = __dbcl_rep_process_message; + dbenv->rep_start = __dbcl_rep_start; + dbenv->rep_stat = __dbcl_rep_stat; + dbenv->set_rep_limit = __dbcl_rep_set_limit; + dbenv->set_rep_request = __dbcl_rep_set_request; + dbenv->set_rep_transport = __dbcl_rep_set_rep_transport; + + } else +#endif + { + dbenv->rep_elect = __rep_elect; + dbenv->rep_flush = __rep_flush; + dbenv->rep_process_message = __rep_process_message; + dbenv->rep_start = __rep_start; + dbenv->rep_stat = __rep_stat; + dbenv->set_rep_limit = __rep_set_limit; + dbenv->set_rep_request = __rep_set_request; + dbenv->set_rep_transport = __rep_set_rep_transport; + /* + * !!! + * Our caller has not yet had the opportunity to reset the panic + * state or turn off mutex locking, and so we can neither check + * the panic state or acquire a mutex in the DB_ENV create path. + */ + + if ((ret = __os_calloc(dbenv, 1, sizeof(DB_REP), &db_rep)) != 0) + return (ret); + dbenv->rep_handle = db_rep; + + /* Initialize the per-process replication structure. */ + db_rep->rep_send = NULL; + } + + return (0); +} + +/* + * __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. + */ +static 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; + int announce, init_db, redo_prepared, ret; + + PANIC_CHECK(dbenv); + ENV_ILLEGAL_BEFORE_OPEN(dbenv, "rep_start"); + ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_stat", DB_INIT_TXN); + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + + if ((ret = __db_fchk(dbenv, "DB_ENV->rep_start", flags, + DB_REP_CLIENT | DB_REP_LOGSONLY | 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_REP_LOGSONLY)) { + __db_err(dbenv, + "DB_ENV->rep_start: replication mode must be specified"); + return (EINVAL); + } + + /* Masters can't be logs-only. */ + if ((ret = __db_fcchk(dbenv, + "DB_ENV->rep_start", flags, DB_REP_LOGSONLY, DB_REP_MASTER)) != 0) + return (ret); + + /* We need a transport function. */ + if (db_rep->rep_send == NULL) { + __db_err(dbenv, + "DB_ENV->set_rep_transport must be called before DB_ENV->rep_start"); + return (EINVAL); + } + + /* We'd better not have any logged files open if we are a client. */ + if (LF_ISSET(DB_REP_CLIENT) && (ret = __dbreg_nofiles(dbenv)) != 0) { + __db_err(dbenv, "DB_ENV->rep_start called with open files"); + return (ret); + } + + MUTEX_LOCK(dbenv, db_rep->mutexp); + if (rep->eid == DB_EID_INVALID) + rep->eid = dbenv->rep_eid; + + if (LF_ISSET(DB_REP_MASTER)) { + if (F_ISSET(dbenv, DB_ENV_REP_CLIENT)) { + /* + * If we're upgrading from having been a client, + * preclose, so that we close our temporary database. + * + * Do not close files that we may have opened while + * doing a rep_apply; they'll get closed when we + * finally close the environment, but for now, leave + * them open, as we don't want to recycle their + * fileids, and we may need the handles again if + * we become a client and the original master + * that opened them becomes a master again. + */ + if ((ret = __rep_preclose(dbenv, 0)) != 0) + return (ret); + + /* + * Now write a __txn_recycle record so that + * clients don't get confused with our txnids + * and txnids of previous masters. + */ + F_CLR(dbenv, DB_ENV_REP_CLIENT); + if ((ret = __txn_reset(dbenv)) != 0) + return (ret); + } + + redo_prepared = 0; + if (!F_ISSET(rep, REP_F_MASTER)) { + /* Master is not yet set. */ + if (F_ISSET(rep, REP_ISCLIENT)) { + F_CLR(rep, REP_ISCLIENT); + rep->gen = ++rep->w_gen; + redo_prepared = 1; + } else if (rep->gen == 0) + rep->gen = 1; + } + + F_SET(rep, REP_F_MASTER); + F_SET(dbenv, DB_ENV_REP_MASTER); + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + dblp = (DB_LOG *)dbenv->lg_handle; + R_LOCK(dbenv, &dblp->reginfo); + lsn = ((LOG *)dblp->reginfo.primary)->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + + /* + * Send the NEWMASTER message, then restore prepared txns + * if and only if we just upgraded from being a client. + */ + if ((ret = __rep_send_message(dbenv, + DB_EID_BROADCAST, REP_NEWMASTER, &lsn, NULL, 0)) == 0 && + redo_prepared) + ret = __rep_restore_prepared(dbenv); + } else { + F_CLR(dbenv, DB_ENV_REP_MASTER); + F_SET(dbenv, DB_ENV_REP_CLIENT); + if (LF_ISSET(DB_REP_LOGSONLY)) + F_SET(dbenv, DB_ENV_REP_LOGSONLY); + + announce = !F_ISSET(rep, REP_ISCLIENT) || + rep->master_id == DB_EID_INVALID; + init_db = 0; + if (!F_ISSET(rep, REP_ISCLIENT)) { + F_CLR(rep, REP_F_MASTER); + if (LF_ISSET(DB_REP_LOGSONLY)) + F_SET(rep, REP_F_LOGSONLY); + else + F_SET(rep, REP_F_UPGRADE); + + /* + * We initialize the client's generation number to 0. + * Upon startup, it looks for a master and updates the + * generation number as necessary, exactly as it does + * during normal operation and a master failure. + */ + rep->gen = 0; + rep->master_id = DB_EID_INVALID; + init_db = 1; + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + /* + * 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) + return (ret); + + if ((ret = __rep_client_dbinit(dbenv, init_db)) != 0) + return (ret); + + /* + * 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) + ret = __rep_send_message(dbenv, + DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0); + } + 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. + */ +static int +__rep_client_dbinit(dbenv, startup) + DB_ENV *dbenv; + int startup; +{ + DB_REP *db_rep; + DB *dbp; + int ret, t_ret; + u_int32_t flags; + + PANIC_CHECK(dbenv); + db_rep = dbenv->rep_handle; + dbp = NULL; + +#define REPDBNAME "__db.rep.db" + + /* Check if this has already been called on this environment. */ + if (db_rep->rep_db != NULL) + return (0); + + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + + 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)dbp->remove(dbp, REPDBNAME, NULL, 0); + } + + if ((ret = db_create(&dbp, dbenv, 0)) != 0) + goto err; + if ((ret = dbp->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 = (F_ISSET(dbenv, DB_ENV_THREAD) ? DB_THREAD : 0) | + (startup ? DB_CREATE : 0); + if ((ret = dbp->open(dbp, NULL, + "__db.rep.db", NULL, DB_BTREE, flags, 0)) != 0) + goto err; + + db_rep->rep_db = dbp; + + if (0) { +err: if (dbp != NULL && + (t_ret = dbp->close(dbp, DB_NOSYNC)) != 0 && ret == 0) + ret = t_ret; + db_rep->rep_db = NULL; + } + + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + + 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; + + __ua_memcpy(&lsn1, &rp1->lsn, sizeof(DB_LSN)); + __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; + R_LOCK(dbenv, &mgr->reginfo); + if (region->stat.st_nrestores != 0) + do_aborts = 1; + R_UNLOCK(dbenv, &mgr->reginfo); + + if (do_aborts) { + op = DB_FIRST; + do { + if ((ret = dbenv->txn_recover(dbenv, + prep, PREPLISTSIZE, &count, op)) != 0) + return (ret); + for (i = 0; i < count; i++) { + p = &prep[i]; + if ((ret = p->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; + 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; + void *txninfo; + + txninfo = NULL; + ckp_args = NULL; + prep_args = NULL; + regop_args = NULL; + ZERO_LSN(ckp_lsn); + ZERO_LSN(lsn); + + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 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 = logc->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 = logc->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 = logc->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 = logc->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 = logc->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 = logc->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 = logc->get(logc, &lsn, &rec, DB_LAST); + ret == 0 && log_compare(&lsn, &ckp_lsn) > 0; + ret = logc->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); + if (ret == DB_NOTFOUND) + ret = __db_txnlist_add(dbenv, txninfo, + regop_args->txnid->txnid, + regop_args->opcode, &lsn); + __os_free(dbenv, regop_args); + break; + case DB___txn_xa_regop: + /* + * It's a prepare. If 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); + if (ret == DB_NOTFOUND) + if ((ret = __rep_process_txn(dbenv, &rec)) == 0) + ret = __txn_restore_txn(dbenv, + &lsn, prep_args); + __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 = logc->close(logc, 0); + + if (txninfo != NULL) + __db_txnlist_end(dbenv, txninfo); + + return (ret == 0 ? t_ret : ret); +} + +/* + * __rep_set_limit -- + * Set a limit on the amount of data that will be sent during a single + * invocation of __rep_process_message. + */ +static int +__rep_set_limit(dbenv, gbytes, bytes) + DB_ENV *dbenv; + u_int32_t gbytes; + u_int32_t bytes; +{ + DB_REP *db_rep; + REP *rep; + + PANIC_CHECK(dbenv); + + if ((db_rep = dbenv->rep_handle) == NULL) { + __db_err(dbenv, + "DB_ENV->set_rep_limit: database environment not properly initialized"); + return (__db_panic(dbenv, EINVAL)); + } + rep = db_rep->region; + MUTEX_LOCK(dbenv, db_rep->mutexp); + if (bytes > GIGABYTE) { + gbytes += bytes / GIGABYTE; + bytes = bytes % GIGABYTE; + } + rep->gbytes = gbytes; + rep->bytes = bytes; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + return (0); +} + +/* + * __rep_set_request -- + * Set the minimum and maximum number of log records that we wait + * before retransmitting. + * UNDOCUMENTED. + */ +static int +__rep_set_request(dbenv, min, max) + DB_ENV *dbenv; + u_int32_t min; + u_int32_t max; +{ + LOG *lp; + DB_LOG *dblp; + DB_REP *db_rep; + REP *rep; + + PANIC_CHECK(dbenv); + + if ((db_rep = dbenv->rep_handle) == NULL) { + __db_err(dbenv, + "DB_ENV->set_rep_request: database environment not properly initialized"); + return (__db_panic(dbenv, EINVAL)); + } + rep = db_rep->region; + MUTEX_LOCK(dbenv, db_rep->mutexp); + rep->request_gap = min; + rep->max_gap = max; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + dblp = dbenv->lg_handle; + if (dblp != NULL && (lp = dblp->reginfo.primary) != NULL) { + R_LOCK(dbenv, &dblp->reginfo); + lp->wait_recs = 0; + lp->rcvd_recs = 0; + R_UNLOCK(dbenv, &dblp->reginfo); + } + + return (0); +} + +/* + * __rep_set_transport -- + * Set the transport function for replication. + */ +static int +__rep_set_rep_transport(dbenv, eid, f_send) + DB_ENV *dbenv; + int eid; + int (*f_send) __P((DB_ENV *, const DBT *, const DBT *, int, u_int32_t)); +{ + DB_REP *db_rep; + + PANIC_CHECK(dbenv); + + if ((db_rep = dbenv->rep_handle) == NULL) { + __db_err(dbenv, + "DB_ENV->set_rep_transport: database environment not properly initialized"); + return (__db_panic(dbenv, EINVAL)); + } + + 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); + } + + db_rep->rep_send = f_send; + + dbenv->rep_eid = eid; + return (0); +} + +/* + * __rep_elect -- + * Called after master failure to hold/participate in an election for + * a new master. + */ +static int +__rep_elect(dbenv, nsites, priority, timeout, eidp) + DB_ENV *dbenv; + int nsites, priority; + u_int32_t timeout; + int *eidp; +{ + DB_LOG *dblp; + DB_LSN lsn; + DB_REP *db_rep; + REP *rep; + int in_progress, ret, send_vote, tiebreaker; + u_int32_t pid, sec, usec; + + PANIC_CHECK(dbenv); + ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_elect", DB_INIT_TXN); + + /* Error checking. */ + if (nsites <= 0) { + __db_err(dbenv, + "DB_ENV->rep_elect: nsites must be greater than 0"); + return (EINVAL); + } + if (priority < 0) { + __db_err(dbenv, + "DB_ENV->rep_elect: priority may not be negative"); + return (EINVAL); + } + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + dblp = dbenv->lg_handle; + + R_LOCK(dbenv, &dblp->reginfo); + lsn = ((LOG *)dblp->reginfo.primary)->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + + /* Generate a randomized tiebreaker value. */ + __os_id(&pid); + if ((ret = __os_clock(dbenv, &sec, &usec)) != 0) + return (ret); + tiebreaker = pid ^ sec ^ usec ^ (u_int)rand() ^ P_TO_UINT32(&pid); + + if ((ret = __rep_elect_init(dbenv, + &lsn, nsites, priority, tiebreaker, &in_progress)) != 0) { + if (ret == DB_REP_NEWMASTER) { + ret = 0; + *eidp = dbenv->rep_eid; + } + return (ret); + } + + if (!in_progress) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "Beginning an election"); +#endif + if ((ret = __rep_send_message(dbenv, + DB_EID_BROADCAST, REP_ELECT, NULL, NULL, 0)) != 0) + goto err; + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTSEND, ret, NULL); + } + + /* Now send vote */ + if ((ret = + __rep_send_vote(dbenv, &lsn, nsites, priority, tiebreaker)) != 0) + goto err; + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTVOTE1, ret, NULL); + + ret = __rep_wait(dbenv, timeout, eidp, REP_F_EPHASE1); + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTWAIT1, ret, NULL); + switch (ret) { + case 0: + /* Check if election complete or phase complete. */ + if (*eidp != DB_EID_INVALID) + return (0); + goto phase2; + case DB_TIMEOUT: + break; + default: + goto err; + } + /* + * If we got here, we haven't heard from everyone, but we've + * run out of time, so it's time to decide if we have enough + * votes to pick a winner and if so, to send out a vote to + * the winner. + */ + MUTEX_LOCK(dbenv, db_rep->mutexp); + send_vote = DB_EID_INVALID; + if (rep->sites > rep->nsites / 2) { + /* We think we've seen enough to cast a vote. */ + send_vote = rep->winner; + if (rep->winner == rep->eid) + rep->votes++; + F_CLR(rep, REP_F_EPHASE1); + F_SET(rep, REP_F_EPHASE2); + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + if (send_vote == DB_EID_INVALID) { + /* We do not have enough votes to elect. */ +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, + "Not enough votes to elect: received %d of %d", + rep->sites, rep->nsites); +#endif + ret = DB_REP_UNAVAIL; + goto err; + + } +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION) && + send_vote != rep->eid) + __db_err(dbenv, "Sending vote"); +#endif + + if (send_vote != rep->eid && (ret = __rep_send_message(dbenv, + send_vote, REP_VOTE2, NULL, NULL, 0)) != 0) + goto err; + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTVOTE2, ret, NULL); + +phase2: ret = __rep_wait(dbenv, timeout, eidp, REP_F_EPHASE2); + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTWAIT2, ret, NULL); + switch (ret) { + case 0: + return (0); + case DB_TIMEOUT: + ret = DB_REP_UNAVAIL; + break; + default: + goto err; + } + +DB_TEST_RECOVERY_LABEL +err: MUTEX_LOCK(dbenv, db_rep->mutexp); + ELECTION_DONE(rep); + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "Ended election with %d", ret); +#endif + return (ret); +} + +/* + * __rep_elect_init + * Initialize an election. Sets beginp non-zero if the election is + * already in progress; makes it 0 otherwise. + */ +static int +__rep_elect_init(dbenv, lsnp, nsites, priority, tiebreaker, beginp) + DB_ENV *dbenv; + DB_LSN *lsnp; + int nsites, priority, tiebreaker, *beginp; +{ + DB_REP *db_rep; + REP *rep; + int ret, *tally; + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + + ret = 0; + + /* We may miscount, as we don't hold the replication mutex here. */ + rep->stat.st_elections++; + + /* If we are already a master; simply broadcast that fact and return. */ + if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) { + (void)__rep_send_message(dbenv, + DB_EID_BROADCAST, REP_NEWMASTER, lsnp, NULL, 0); + rep->stat.st_elections_won++; + return (DB_REP_NEWMASTER); + } + + MUTEX_LOCK(dbenv, db_rep->mutexp); + *beginp = IN_ELECTION(rep); + if (!*beginp) { + /* + * Make sure that we always initialize all the election fields + * before putting ourselves in an election state. That means + * issuing calls that can fail (allocation) before setting all + * the variables. + */ + if (nsites > rep->asites && + (ret = __rep_grow_sites(dbenv, nsites)) != 0) + goto err; + DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTINIT, ret, NULL); + rep->nsites = nsites; + rep->priority = priority; + rep->votes = 0; + rep->master_id = DB_EID_INVALID; + F_SET(rep, REP_F_EPHASE1); + + /* We have always heard from ourselves. */ + rep->sites = 1; + tally = R_ADDR((REGINFO *)dbenv->reginfo, rep->tally_off); + tally[0] = rep->eid; + + if (priority != 0) { + /* Make ourselves the winner to start. */ + rep->winner = rep->eid; + rep->w_priority = priority; + rep->w_gen = rep->gen; + rep->w_lsn = *lsnp; + rep->w_tiebreaker = tiebreaker; + } else { + rep->winner = DB_EID_INVALID; + rep->w_priority = 0; + rep->w_gen = 0; + ZERO_LSN(rep->w_lsn); + rep->w_tiebreaker = 0; + } + } +DB_TEST_RECOVERY_LABEL +err: MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (ret); +} + +static int +__rep_wait(dbenv, timeout, eidp, flags) + DB_ENV *dbenv; + u_int32_t timeout; + int *eidp; + u_int32_t flags; +{ + DB_REP *db_rep; + REP *rep; + int done, ret; + u_int32_t sleeptime; + + done = 0; + db_rep = dbenv->rep_handle; + rep = db_rep->region; + + /* + * The user specifies an overall timeout function, but checking + * is cheap and the timeout may be a generous upper bound. + * Sleep repeatedly for the smaller of .5s and timeout/10. + */ + sleeptime = (timeout > 5000000) ? 500000 : timeout / 10; + if (sleeptime == 0) + sleeptime++; + while (timeout > 0) { + if ((ret = __os_sleep(dbenv, 0, sleeptime)) != 0) + return (ret); + MUTEX_LOCK(dbenv, db_rep->mutexp); + done = !F_ISSET(rep, flags) && rep->master_id != DB_EID_INVALID; + + *eidp = rep->master_id; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + if (done) + return (0); + + if (timeout > sleeptime) + timeout -= sleeptime; + else + timeout = 0; + } + return (DB_TIMEOUT); +} + +/* + * __rep_flush -- + * Re-push the last log record to all clients, in case they've lost + * messages and don't know it. + */ +static 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->tx_handle, "rep_stat", DB_INIT_TXN); + + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + + memset(&rec, 0, sizeof(rec)); + memset(&lsn, 0, sizeof(lsn)); + + if ((ret = logc->get(logc, &lsn, &rec, DB_LAST)) != 0) + goto err; + + ret = __rep_send_message(dbenv, + DB_EID_BROADCAST, REP_LOG, &lsn, &rec, 0); + +err: if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); +} + +/* + * __rep_stat -- + * Fetch replication statistics. + */ +static int +__rep_stat(dbenv, statp, flags) + DB_ENV *dbenv; + DB_REP_STAT **statp; + u_int32_t flags; +{ + DB_LOG *dblp; + DB_REP *db_rep; + DB_REP_STAT *stats; + LOG *lp; + REP *rep; + u_int32_t queued; + int ret; + + PANIC_CHECK(dbenv); + ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_stat", DB_INIT_TXN); + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + + *statp = NULL; + if ((ret = __db_fchk(dbenv, + "DB_ENV->rep_stat", flags, DB_STAT_CLEAR)) != 0) + return (ret); + + /* Allocate a stat struct to return to the user. */ + if ((ret = __os_umalloc(dbenv, sizeof(DB_REP_STAT), &stats)) != 0) + return (ret); + + MUTEX_LOCK(dbenv, db_rep->mutexp); + memcpy(stats, &rep->stat, sizeof(*stats)); + + /* Copy out election stats. */ + if (IN_ELECTION(rep)) { + if (F_ISSET(rep, REP_F_EPHASE1)) + stats->st_election_status = 1; + else if (F_ISSET(rep, REP_F_EPHASE2)) + stats->st_election_status = 2; + + stats->st_election_nsites = rep->sites; + stats->st_election_cur_winner = rep->winner; + stats->st_election_priority = rep->w_priority; + stats->st_election_gen = rep->w_gen; + stats->st_election_lsn = rep->w_lsn; + stats->st_election_votes = rep->votes; + stats->st_election_tiebreaker = rep->w_tiebreaker; + } + + /* Copy out other info that's protected by the rep mutex. */ + stats->st_env_id = rep->eid; + stats->st_env_priority = rep->priority; + stats->st_nsites = rep->nsites; + stats->st_master = rep->master_id; + stats->st_gen = rep->gen; + + if (F_ISSET(rep, REP_F_MASTER)) + stats->st_status = DB_REP_MASTER; + else if (F_ISSET(rep, REP_F_LOGSONLY)) + stats->st_status = DB_REP_LOGSONLY; + else if (F_ISSET(rep, REP_F_UPGRADE)) + stats->st_status = DB_REP_CLIENT; + else + stats->st_status = 0; + + if (LF_ISSET(DB_STAT_CLEAR)) { + queued = rep->stat.st_log_queued; + memset(&rep->stat, 0, sizeof(rep->stat)); + rep->stat.st_log_queued = rep->stat.st_log_queued_total = + rep->stat.st_log_queued_max = queued; + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + /* + * Log-related replication info is stored in the log system and + * protected by the log region lock. + */ + R_LOCK(dbenv, &dblp->reginfo); + if (F_ISSET(rep, REP_ISCLIENT)) { + stats->st_next_lsn = lp->ready_lsn; + stats->st_waiting_lsn = lp->waiting_lsn; + } else { + if (F_ISSET(rep, REP_F_MASTER)) + stats->st_next_lsn = lp->lsn; + else + ZERO_LSN(stats->st_next_lsn); + ZERO_LSN(stats->st_waiting_lsn); + } + R_UNLOCK(dbenv, &dblp->reginfo); + + *statp = stats; + return (0); +} diff --git a/bdb/rep/rep_record.c b/bdb/rep/rep_record.c new file mode 100644 index 00000000000..e725008786c --- /dev/null +++ b/bdb/rep/rep_record.c @@ -0,0 +1,1510 @@ +/*- + * See the file LICENSE for redistribution information. + * + * Copyright (c) 2001-2002 + * Sleepycat Software. All rights reserved. + */ + +#include "db_config.h" + +#ifndef lint +static const char revid[] = "$Id: rep_record.c,v 1.111 2002/09/11 19:39:11 bostic Exp $"; +#endif /* not lint */ + +#ifndef NO_SYSTEM_INCLUDES +#include <stdlib.h> +#include <string.h> +#endif + +#include "db_int.h" +#include "dbinc/db_page.h" +#include "dbinc/db_am.h" +#include "dbinc/log.h" +#include "dbinc/rep.h" +#include "dbinc/txn.h" + +static int __rep_apply __P((DB_ENV *, REP_CONTROL *, DBT *)); +static int __rep_collect_txn __P((DB_ENV *, DB_LSN *, LSN_COLLECTION *)); +static int __rep_lsn_cmp __P((const void *, const void *)); +static int __rep_newfile __P((DB_ENV *, REP_CONTROL *, DBT *, DB_LSN *)); + +#define IS_SIMPLE(R) ((R) != DB___txn_regop && \ + (R) != DB___txn_ckp && (R) != DB___dbreg_register) + +/* + * __rep_process_message -- + * + * This routine takes an incoming message and processes it. + * + * control: contains the control fields from the record + * rec: contains the actual record + * eidp: contains the machine id of the sender of the message; + * in the case of a DB_NEWMASTER message, returns the eid + * of the new master. + * + * PUBLIC: int __rep_process_message __P((DB_ENV *, DBT *, DBT *, int *)); + */ +int +__rep_process_message(dbenv, control, rec, eidp) + DB_ENV *dbenv; + DBT *control, *rec; + int *eidp; +{ + DB_LOG *dblp; + DB_LOGC *logc; + DB_LSN init_lsn, lsn, newfilelsn, oldfilelsn; + DB_REP *db_rep; + DBT *d, data_dbt, lsndbt, mylog; + LOG *lp; + REP *rep; + REP_CONTROL *rp; + REP_VOTE_INFO *vi; + u_int32_t bytes, gen, gbytes, type, unused; + int check_limit, cmp, done, do_req, i; + int master, old, recovering, ret, t_ret, *tally; + + PANIC_CHECK(dbenv); + ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_stat", DB_INIT_TXN); + + /* Control argument must be non-Null. */ + if (control == NULL || control->size == 0) { + __db_err(dbenv, + "DB_ENV->rep_process_message: control argument must be specified"); + return (EINVAL); + } + + ret = 0; + db_rep = dbenv->rep_handle; + rep = db_rep->region; + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + + MUTEX_LOCK(dbenv, db_rep->mutexp); + gen = rep->gen; + recovering = F_ISSET(rep, REP_F_RECOVER); + + rep->stat.st_msgs_processed++; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + rp = (REP_CONTROL *)control->data; + +#if 0 + __rep_print_message(dbenv, *eidp, rp, "rep_process_message"); +#endif + + /* Complain if we see an improper version number. */ + if (rp->rep_version != DB_REPVERSION) { + __db_err(dbenv, + "unexpected replication message version %d, expected %d", + rp->rep_version, DB_REPVERSION); + return (EINVAL); + } + if (rp->log_version != DB_LOGVERSION) { + __db_err(dbenv, + "unexpected log record version %d, expected %d", + rp->log_version, DB_LOGVERSION); + return (EINVAL); + } + + /* + * Check for generation number matching. Ignore any old messages + * except requests that are indicative of a new client that needs + * to get in sync. + */ + if (rp->gen < gen && rp->rectype != REP_ALIVE_REQ && + rp->rectype != REP_NEWCLIENT && rp->rectype != REP_MASTER_REQ) { + /* + * We don't hold the rep mutex, and could miscount if we race. + */ + rep->stat.st_msgs_badgen++; + return (0); + } + if (rp->gen > gen && rp->rectype != REP_ALIVE && + rp->rectype != REP_NEWMASTER) + return (__rep_send_message(dbenv, + DB_EID_BROADCAST, REP_MASTER_REQ, NULL, NULL, 0)); + + /* + * We need to check if we're in recovery and if we are + * then we need to ignore any messages except VERIFY, VOTE, + * ELECT (the master might fail while we are recovering), and + * ALIVE_REQ. + */ + if (recovering) + switch(rp->rectype) { + case REP_ALIVE: + case REP_ALIVE_REQ: + case REP_ELECT: + case REP_NEWCLIENT: + case REP_NEWMASTER: + case REP_NEWSITE: + case REP_VERIFY: + R_LOCK(dbenv, &dblp->reginfo); + cmp = log_compare(&lp->verify_lsn, &rp->lsn); + R_UNLOCK(dbenv, &dblp->reginfo); + if (cmp != 0) + goto skip; + /* FALLTHROUGH */ + case REP_VOTE1: + case REP_VOTE2: + break; + default: +skip: /* + * We don't hold the rep mutex, and could + * miscount if we race. + */ + rep->stat.st_msgs_recover++; + + /* Check for need to retransmit. */ + R_LOCK(dbenv, &dblp->reginfo); + do_req = *eidp == rep->master_id && + ++lp->rcvd_recs >= lp->wait_recs; + if (do_req) { + lp->wait_recs *= 2; + if (lp->wait_recs + rep->max_gap) + lp->wait_recs = rep->max_gap; + lp->rcvd_recs = 0; + lsn = lp->verify_lsn; + } + R_UNLOCK(dbenv, &dblp->reginfo); + if (do_req) + ret = __rep_send_message(dbenv, *eidp, + REP_VERIFY_REQ, &lsn, NULL, 0); + + return (ret); + } + + switch(rp->rectype) { + case REP_ALIVE: + ANYSITE(dbenv); + if (rp->gen > gen && rp->flags) + return (__rep_new_master(dbenv, rp, *eidp)); + break; + case REP_ALIVE_REQ: + ANYSITE(dbenv); + dblp = dbenv->lg_handle; + R_LOCK(dbenv, &dblp->reginfo); + lsn = ((LOG *)dblp->reginfo.primary)->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + return (__rep_send_message(dbenv, + *eidp, REP_ALIVE, &lsn, NULL, + F_ISSET(dbenv, DB_ENV_REP_MASTER) ? 1 : 0)); + case REP_ALL_REQ: + MASTER_ONLY(dbenv); + gbytes = bytes = 0; + MUTEX_LOCK(dbenv, db_rep->mutexp); + gbytes = rep->gbytes; + bytes = rep->bytes; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + check_limit = gbytes != 0 || bytes != 0; + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + memset(&data_dbt, 0, sizeof(data_dbt)); + oldfilelsn = lsn = rp->lsn; + type = REP_LOG; + for (ret = logc->get(logc, &rp->lsn, &data_dbt, DB_SET); + ret == 0 && type == REP_LOG; + ret = logc->get(logc, &lsn, &data_dbt, DB_NEXT)) { + /* + * lsn.offset will only be 0 if this is the + * beginning of the log; DB_SET, but not DB_NEXT, + * can set the log cursor to [n][0]. + */ + if (lsn.offset == 0) + ret = __rep_send_message(dbenv, *eidp, + REP_NEWFILE, &lsn, NULL, 0); + else { + /* + * DB_NEXT will never run into offsets + * of 0; thus, when a log file changes, + * we'll have a real log record with + * some lsn [n][m], and we'll also want to send + * a NEWFILE message with lsn [n][0]. + * So that the client can detect gaps, + * send in the rec parameter the + * last LSN in the old file. + */ + if (lsn.file != oldfilelsn.file) { + newfilelsn.file = lsn.file; + newfilelsn.offset = 0; + + memset(&lsndbt, 0, sizeof(DBT)); + lsndbt.size = sizeof(DB_LSN); + lsndbt.data = &oldfilelsn; + + if ((ret = __rep_send_message(dbenv, + *eidp, REP_NEWFILE, &newfilelsn, + &lsndbt, 0)) != 0) + break; + } + if (check_limit) { + /* + * data_dbt.size is only the size of + * the log record; it doesn't count + * the size of the control structure. + * Factor that in as well so we're + * not off by a lot if our log + * records are small. + */ + while (bytes < data_dbt.size + + sizeof(REP_CONTROL)) { + if (gbytes > 0) { + bytes += GIGABYTE; + --gbytes; + continue; + } + /* + * We don't hold the rep mutex, + * and may miscount. + */ + rep->stat.st_nthrottles++; + type = REP_LOG_MORE; + goto send; + } + bytes -= (data_dbt.size + + sizeof(REP_CONTROL)); + } +send: ret = __rep_send_message(dbenv, *eidp, + type, &lsn, &data_dbt, 0); + } + + /* + * In case we're about to change files and need it + * for a NEWFILE message, save the current LSN. + */ + oldfilelsn = lsn; + } + + if (ret == DB_NOTFOUND) + ret = 0; + if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); + case REP_ELECT: + if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) { + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + MUTEX_LOCK(dbenv, db_rep->mutexp); + rep->gen++; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (__rep_send_message(dbenv, + *eidp, REP_NEWMASTER, &lsn, NULL, 0)); + } + MUTEX_LOCK(dbenv, db_rep->mutexp); + ret = IN_ELECTION(rep) ? 0 : DB_REP_HOLDELECTION; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (ret); +#ifdef NOTYET + case REP_FILE: /* TODO */ + CLIENT_ONLY(dbenv); + break; + case REP_FILE_REQ: + MASTER_ONLY(dbenv); + return (__rep_send_file(dbenv, rec, *eidp)); + break; +#endif + case REP_LOG: + case REP_LOG_MORE: + CLIENT_ONLY(dbenv); + if ((ret = __rep_apply(dbenv, rp, rec)) != 0) + return (ret); + if (rp->rectype == REP_LOG_MORE) { + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + master = rep->master_id; + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + ret = __rep_send_message(dbenv, master, + REP_ALL_REQ, &lsn, NULL, 0); + } + return (ret); + case REP_LOG_REQ: + MASTER_ONLY(dbenv); + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + memset(&data_dbt, 0, sizeof(data_dbt)); + lsn = rp->lsn; + + /* + * There are three different cases here. + * 1. We asked for a particular LSN and got it. + * 2. We asked for an LSN of X,0 which is invalid and got the + * first log record in a particular file. + * 3. We asked for an LSN and it's not found because it is + * beyond the end of a log file and we need a NEWFILE msg. + */ + ret = logc->get(logc, &rp->lsn, &data_dbt, DB_SET); + cmp = log_compare(&lsn, &rp->lsn); + + if (ret == 0 && cmp == 0) /* Case 1 */ + ret = __rep_send_message(dbenv, *eidp, + REP_LOG, &rp->lsn, &data_dbt, 0); + else if (ret == DB_NOTFOUND || + (ret == 0 && cmp < 0 && rp->lsn.offset == 0)) + /* Cases 2 and 3: Send a NEWFILE message. */ + ret = __rep_send_message(dbenv, *eidp, + REP_NEWFILE, &lsn, NULL, 0); + + if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); + case REP_NEWSITE: + /* We don't hold the rep mutex, and may miscount. */ + rep->stat.st_newsites++; + + /* This is a rebroadcast; simply tell the application. */ + if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) { + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + (void)__rep_send_message(dbenv, + *eidp, REP_NEWMASTER, &lsn, NULL, 0); + } + return (DB_REP_NEWSITE); + case REP_NEWCLIENT: + /* + * This message was received and should have resulted in the + * application entering the machine ID in its machine table. + * We respond to this with an ALIVE to send relevant information + * to the new client. But first, broadcast the new client's + * record to all the clients. + */ + if ((ret = __rep_send_message(dbenv, + DB_EID_BROADCAST, REP_NEWSITE, &rp->lsn, rec, 0)) != 0) + return (ret); + + if (F_ISSET(dbenv, DB_ENV_REP_CLIENT)) + return (0); + + /* FALLTHROUGH */ + case REP_MASTER_REQ: + ANYSITE(dbenv); + if (F_ISSET(dbenv, DB_ENV_REP_CLIENT)) + return (0); + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + return (__rep_send_message(dbenv, + *eidp, REP_NEWMASTER, &lsn, NULL, 0)); + case REP_NEWFILE: + CLIENT_ONLY(dbenv); + return (__rep_apply(dbenv, rp, rec)); + case REP_NEWMASTER: + ANYSITE(dbenv); + if (F_ISSET(dbenv, DB_ENV_REP_MASTER) && + *eidp != dbenv->rep_eid) { + /* We don't hold the rep mutex, and may miscount. */ + rep->stat.st_dupmasters++; + return (DB_REP_DUPMASTER); + } + return (__rep_new_master(dbenv, rp, *eidp)); + case REP_PAGE: /* TODO */ + CLIENT_ONLY(dbenv); + break; + case REP_PAGE_REQ: /* TODO */ + MASTER_ONLY(dbenv); + break; + case REP_PLIST: /* TODO */ + CLIENT_ONLY(dbenv); + break; + case REP_PLIST_REQ: /* TODO */ + MASTER_ONLY(dbenv); + break; + case REP_VERIFY: + CLIENT_ONLY(dbenv); + DB_ASSERT((F_ISSET(rep, REP_F_RECOVER) && + !IS_ZERO_LSN(lp->verify_lsn)) || + (!F_ISSET(rep, REP_F_RECOVER) && + IS_ZERO_LSN(lp->verify_lsn))); + if (IS_ZERO_LSN(lp->verify_lsn)) + return (0); + + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + memset(&mylog, 0, sizeof(mylog)); + if ((ret = logc->get(logc, &rp->lsn, &mylog, DB_SET)) != 0) + goto rep_verify_err; + if (mylog.size == rec->size && + memcmp(mylog.data, rec->data, rec->size) == 0) { + /* + * If we're a logs-only client, we can simply truncate + * the log to the point where it last agreed with the + * master's; otherwise, recover to that point. + */ + R_LOCK(dbenv, &dblp->reginfo); + ZERO_LSN(lp->verify_lsn); + R_UNLOCK(dbenv, &dblp->reginfo); + if (F_ISSET(dbenv, DB_ENV_REP_LOGSONLY)) { + INIT_LSN(init_lsn); + if ((ret = dbenv->log_flush(dbenv, + &rp->lsn)) != 0 || + (ret = __log_vtruncate(dbenv, + &rp->lsn, &init_lsn)) != 0) + goto rep_verify_err; + } else if ((ret = __db_apprec(dbenv, &rp->lsn, 0)) != 0) + goto rep_verify_err; + + /* + * The log has been truncated (either by __db_apprec or + * directly). We want to make sure we're waiting for + * the LSN at the new end-of-log, not some later point. + */ + R_LOCK(dbenv, &dblp->reginfo); + lp->ready_lsn = lp->lsn; + ZERO_LSN(lp->waiting_lsn); + R_UNLOCK(dbenv, &dblp->reginfo); + + /* + * Discard any log records we have queued; we're + * about to re-request them, and can't trust the + * ones in the queue. + */ + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + if ((ret = db_rep->rep_db->truncate(db_rep->rep_db, + NULL, &unused, 0)) != 0) { + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + goto rep_verify_err; + } + rep->stat.st_log_queued = 0; + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + + MUTEX_LOCK(dbenv, db_rep->mutexp); + F_CLR(rep, REP_F_RECOVER); + + /* + * If the master_id is invalid, this means that since + * the last record was sent, somebody declared an + * election and we may not have a master to request + * things of. + * + * This is not an error; when we find a new master, + * we'll re-negotiate where the end of the log is and + * try to bring ourselves up to date again anyway. + */ + if ((master = rep->master_id) == DB_EID_INVALID) { + DB_ASSERT(IN_ELECTION(rep)); + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + ret = 0; + } else { + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + ret = __rep_send_message(dbenv, master, + REP_ALL_REQ, &rp->lsn, NULL, 0); + } + } else if ((ret = + logc->get(logc, &lsn, &mylog, DB_PREV)) == 0) { + R_LOCK(dbenv, &dblp->reginfo); + lp->verify_lsn = lsn; + lp->rcvd_recs = 0; + lp->wait_recs = rep->request_gap; + R_UNLOCK(dbenv, &dblp->reginfo); + ret = __rep_send_message(dbenv, + *eidp, REP_VERIFY_REQ, &lsn, NULL, 0); + } + +rep_verify_err: if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); + case REP_VERIFY_FAIL: + rep->stat.st_outdated++; + return (DB_REP_OUTDATED); + case REP_VERIFY_REQ: + MASTER_ONLY(dbenv); + type = REP_VERIFY; + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + d = &data_dbt; + memset(d, 0, sizeof(data_dbt)); + F_SET(logc, DB_LOG_SILENT_ERR); + ret = logc->get(logc, &rp->lsn, d, DB_SET); + /* + * If the LSN was invalid, then we might get a not + * found, we might get an EIO, we could get anything. + * If we get a DB_NOTFOUND, then there is a chance that + * the LSN comes before the first file present in which + * case we need to return a fail so that the client can return + * a DB_OUTDATED. + */ + if (ret == DB_NOTFOUND && + __log_is_outdated(dbenv, rp->lsn.file, &old) == 0 && + old != 0) + type = REP_VERIFY_FAIL; + + if (ret != 0) + d = NULL; + + ret = __rep_send_message(dbenv, *eidp, type, &rp->lsn, d, 0); + if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); + case REP_VOTE1: + if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "Master received vote"); +#endif + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + return (__rep_send_message(dbenv, + *eidp, REP_NEWMASTER, &lsn, NULL, 0)); + } + + vi = (REP_VOTE_INFO *)rec->data; + MUTEX_LOCK(dbenv, db_rep->mutexp); + + /* + * If you get a vote and you're not in an election, simply + * return an indicator to hold an election which will trigger + * this site to send its vote again. + */ + if (!IN_ELECTION(rep)) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, + "Not in election, but received vote1"); +#endif + ret = DB_REP_HOLDELECTION; + goto unlock; + } + + if (F_ISSET(rep, REP_F_EPHASE2)) + goto unlock; + + /* Check if this site knows about more sites than we do. */ + if (vi->nsites > rep->nsites) + rep->nsites = vi->nsites; + + /* Check if we've heard from this site already. */ + tally = R_ADDR((REGINFO *)dbenv->reginfo, rep->tally_off); + for (i = 0; i < rep->sites; i++) { + if (tally[i] == *eidp) + /* Duplicate vote. */ + goto unlock; + } + + /* + * We are keeping vote, let's see if that changes our count of + * the number of sites. + */ + if (rep->sites + 1 > rep->nsites) + rep->nsites = rep->sites + 1; + if (rep->nsites > rep->asites && + (ret = __rep_grow_sites(dbenv, rep->nsites)) != 0) + goto unlock; + + tally[rep->sites] = *eidp; + rep->sites++; + + /* + * Change winners if the incoming record has a higher + * priority, or an equal priority but a larger LSN, or + * an equal priority and LSN but higher "tiebreaker" value. + */ +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) { + __db_err(dbenv, + "%s(eid)%d (pri)%d (gen)%d (sites)%d [%d,%d]", + "Existing vote: ", + rep->winner, rep->w_priority, rep->w_gen, + rep->sites, rep->w_lsn.file, rep->w_lsn.offset); + __db_err(dbenv, + "Incoming vote: (eid)%d (pri)%d (gen)%d [%d,%d]", + *eidp, vi->priority, rp->gen, rp->lsn.file, + rp->lsn.offset); + } +#endif + cmp = log_compare(&rp->lsn, &rep->w_lsn); + if (vi->priority > rep->w_priority || + (vi->priority != 0 && vi->priority == rep->w_priority && + (cmp > 0 || + (cmp == 0 && vi->tiebreaker > rep->w_tiebreaker)))) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "Accepting new vote"); +#endif + rep->winner = *eidp; + rep->w_priority = vi->priority; + rep->w_lsn = rp->lsn; + rep->w_gen = rp->gen; + } + master = rep->winner; + lsn = rep->w_lsn; + done = rep->sites == rep->nsites && rep->w_priority != 0; + if (done) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) { + __db_err(dbenv, "Phase1 election done"); + __db_err(dbenv, "Voting for %d%s", + master, master == rep->eid ? "(self)" : ""); + } +#endif + F_CLR(rep, REP_F_EPHASE1); + F_SET(rep, REP_F_EPHASE2); + } + + if (done && master == rep->eid) { + rep->votes++; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (0); + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + /* Vote for someone else. */ + if (done) + return (__rep_send_message(dbenv, + master, REP_VOTE2, NULL, NULL, 0)); + + /* Election is still going on. */ + break; + case REP_VOTE2: +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "We received a vote%s", + F_ISSET(dbenv, DB_ENV_REP_MASTER) ? + " (master)" : ""); +#endif + if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) { + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + rep->stat.st_elections_won++; + return (__rep_send_message(dbenv, + *eidp, REP_NEWMASTER, &lsn, NULL, 0)); + } + + MUTEX_LOCK(dbenv, db_rep->mutexp); + + /* If we have priority 0, we should never get a vote. */ + DB_ASSERT(rep->priority != 0); + + if (!IN_ELECTION(rep)) { +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "Not in election, got vote"); +#endif + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (DB_REP_HOLDELECTION); + } + /* avoid counting duplicates. */ + rep->votes++; + done = rep->votes > rep->nsites / 2; + if (done) { + rep->master_id = rep->eid; + rep->gen = rep->w_gen + 1; + ELECTION_DONE(rep); + F_CLR(rep, REP_F_UPGRADE); + F_SET(rep, REP_F_MASTER); + *eidp = rep->master_id; +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, + "Got enough votes to win; election done; winner is %d", + rep->master_id); +#endif + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + if (done) { + R_LOCK(dbenv, &dblp->reginfo); + lsn = lp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + + /* Declare me the winner. */ +#ifdef DIAGNOSTIC + if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION)) + __db_err(dbenv, "I won, sending NEWMASTER"); +#endif + rep->stat.st_elections_won++; + if ((ret = __rep_send_message(dbenv, DB_EID_BROADCAST, + REP_NEWMASTER, &lsn, NULL, 0)) != 0) + break; + return (DB_REP_NEWMASTER); + } + break; + default: + __db_err(dbenv, + "DB_ENV->rep_process_message: unknown replication message: type %lu", + (u_long)rp->rectype); + return (EINVAL); + } + + return (0); + +unlock: MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (ret); +} + +/* + * __rep_apply -- + * + * Handle incoming log records on a client, applying when possible and + * entering into the bookkeeping table otherwise. This is the guts of + * the routine that handles the state machine that describes how we + * process and manage incoming log records. + */ +static int +__rep_apply(dbenv, rp, rec) + DB_ENV *dbenv; + REP_CONTROL *rp; + DBT *rec; +{ + __dbreg_register_args dbreg_args; + __txn_ckp_args ckp_args; + DB_REP *db_rep; + DBT control_dbt, key_dbt, lsn_dbt, nextrec_dbt, rec_dbt; + DB *dbp; + DBC *dbc; + DB_LOG *dblp; + DB_LSN ckp_lsn, lsn, newfile_lsn, next_lsn, waiting_lsn; + LOG *lp; + REP *rep; + REP_CONTROL lsn_rc; + u_int32_t rectype, txnid; + int cmp, do_req, eid, have_mutex, ret, t_ret; + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + dbp = db_rep->rep_db; + dbc = NULL; + have_mutex = ret = 0; + memset(&control_dbt, 0, sizeof(control_dbt)); + memset(&rec_dbt, 0, sizeof(rec_dbt)); + + /* + * If this is a log record and it's the next one in line, simply + * write it to the log. If it's a "normal" log record, i.e., not + * a COMMIT or CHECKPOINT or something that needs immediate processing, + * just return. If it's a COMMIT, CHECKPOINT or LOG_REGISTER (i.e., + * not SIMPLE), handle it now. If it's a NEWFILE record, then we + * have to be prepared to deal with a logfile change. + */ + dblp = dbenv->lg_handle; + R_LOCK(dbenv, &dblp->reginfo); + lp = dblp->reginfo.primary; + cmp = log_compare(&rp->lsn, &lp->ready_lsn); + + /* + * This is written to assume that you don't end up with a lot of + * records after a hole. That is, it optimizes for the case where + * there is only a record or two after a hole. If you have a lot + * of records after a hole, what you'd really want to do is write + * all of them and then process all the commits, checkpoints, etc. + * together. That is more complicated processing that we can add + * later if necessary. + * + * That said, I really don't want to do db operations holding the + * log mutex, so the synchronization here is tricky. + */ + if (cmp == 0) { + /* We got the log record that we are expecting. */ + if (rp->rectype == REP_NEWFILE) { +newfile: ret = __rep_newfile(dbenv, rp, rec, &lp->ready_lsn); + + /* Make this evaluate to a simple rectype. */ + rectype = 0; + } else { + DB_ASSERT(log_compare(&rp->lsn, &lp->lsn) == 0); + ret = __log_rep_put(dbenv, &rp->lsn, rec); + lp->ready_lsn = lp->lsn; + memcpy(&rectype, rec->data, sizeof(rectype)); + if (ret == 0) + /* + * We may miscount if we race, since we + * don't currently hold the rep mutex. + */ + rep->stat.st_log_records++; + } + while (ret == 0 && IS_SIMPLE(rectype) && + log_compare(&lp->ready_lsn, &lp->waiting_lsn) == 0) { + /* + * We just filled in a gap in the log record stream. + * Write subsequent records to the log. + */ +gap_check: lp->wait_recs = 0; + lp->rcvd_recs = 0; + R_UNLOCK(dbenv, &dblp->reginfo); + if (have_mutex == 0) { + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + have_mutex = 1; + } + if (dbc == NULL && + (ret = dbp->cursor(dbp, NULL, &dbc, 0)) != 0) + goto err; + + /* The DBTs need to persist through another call. */ + F_SET(&control_dbt, DB_DBT_REALLOC); + F_SET(&rec_dbt, DB_DBT_REALLOC); + if ((ret = dbc->c_get(dbc, + &control_dbt, &rec_dbt, DB_RMW | DB_FIRST)) != 0) + goto err; + + rp = (REP_CONTROL *)control_dbt.data; + rec = &rec_dbt; + memcpy(&rectype, rec->data, sizeof(rectype)); + R_LOCK(dbenv, &dblp->reginfo); + /* + * We need to check again, because it's possible that + * some other thread of control changed the waiting_lsn + * or removed that record from the database. + */ + if (log_compare(&lp->ready_lsn, &rp->lsn) == 0) { + if (rp->rectype != REP_NEWFILE) { + DB_ASSERT(log_compare + (&rp->lsn, &lp->lsn) == 0); + ret = __log_rep_put(dbenv, + &rp->lsn, rec); + lp->ready_lsn = lp->lsn; + + /* + * We may miscount if we race, since we + * don't currently hold the rep mutex. + */ + if (ret == 0) + rep->stat.st_log_records++; + } else { + ret = __rep_newfile(dbenv, + rp, rec, &lp->ready_lsn); + rectype = 0; + } + waiting_lsn = lp->waiting_lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + if ((ret = dbc->c_del(dbc, 0)) != 0) + goto err; + + /* + * We may miscount, as we don't hold the rep + * mutex. + */ + --rep->stat.st_log_queued; + + /* + * Update waiting_lsn. We need to move it + * forward to the LSN of the next record + * in the queue. + */ + memset(&lsn_dbt, 0, sizeof(lsn_dbt)); + F_SET(&lsn_dbt, DB_DBT_USERMEM); + lsn_dbt.data = &lsn_rc; + lsn_dbt.ulen = sizeof(lsn_rc); + memset(&lsn_rc, 0, sizeof(lsn_rc)); + + /* + * If the next item in the database is a log + * record--the common case--we're not + * interested in its contents, just in its LSN. + * If it's a newfile message, though, the + * data field may be the LSN of the last + * record in the old file, and we need to use + * that to determine whether or not there's + * a gap. + * + * Optimize both these cases by doing a partial + * get of the data item. If it's a newfile + * record, we'll get the whole LSN, and if + * it's not, we won't waste time allocating. + */ + memset(&nextrec_dbt, 0, sizeof(nextrec_dbt)); + F_SET(&nextrec_dbt, + DB_DBT_USERMEM | DB_DBT_PARTIAL); + nextrec_dbt.ulen = + nextrec_dbt.dlen = sizeof(newfile_lsn); + ZERO_LSN(newfile_lsn); + nextrec_dbt.data = &newfile_lsn; + + ret = dbc->c_get(dbc, + &lsn_dbt, &nextrec_dbt, DB_NEXT); + if (ret != DB_NOTFOUND && ret != 0) + goto err; + + R_LOCK(dbenv, &dblp->reginfo); + if (ret == DB_NOTFOUND) { + /* + * Do a quick double-check to make + * sure waiting_lsn hasn't changed. + * It's possible that between the + * DB_NOTFOUND return and the R_LOCK, + * some record was added to the + * database, and we don't want to lose + * sight of the fact that it's there. + */ + if (log_compare(&waiting_lsn, + &lp->waiting_lsn) == 0) + ZERO_LSN( + lp->waiting_lsn); + + /* + * Whether or not the current record is + * simple, there's no next one, and + * therefore we haven't got anything + * else to do right now. Break out. + */ + break; + } + + DB_ASSERT(lsn_dbt.size == sizeof(lsn_rc)); + + /* + * NEWFILE records have somewhat convoluted + * semantics, so there are five cases + * pertaining to what the newly-gotten record + * is and what we want to do about it. + * + * 1) This isn't a NEWFILE record. Advance + * waiting_lsn and proceed. + * + * 2) NEWFILE, no LSN stored as the datum, + * lsn_rc.lsn == ready_lsn. The NEWFILE + * record is next, so set waiting_lsn = + * ready_lsn. + * + * 3) NEWFILE, no LSN stored as the datum, but + * lsn_rc.lsn > ready_lsn. There's still a + * gap; set waiting_lsn = lsn_rc.lsn. + * + * 4) NEWFILE, newfile_lsn in datum, and it's < + * ready_lsn. (If the datum is non-empty, + * it's the LSN of the last record in a log + * file, not the end of the log, and + * lsn_rc.lsn is the LSN of the start of + * the new file--we didn't have the end of + * the old log handy when we sent the + * record.) No gap--we're ready to + * proceed. Set both waiting and ready_lsn + * to lsn_rc.lsn. + * + * 5) NEWFILE, newfile_lsn in datum, and it's >= + * ready_lsn. We're still missing at + * least one record; set waiting_lsn, + * but not ready_lsn, to lsn_rc.lsn. + */ + if (lsn_rc.rectype == REP_NEWFILE && + nextrec_dbt.size > 0 && log_compare( + &newfile_lsn, &lp->ready_lsn) < 0) + /* Case 4. */ + lp->ready_lsn = + lp->waiting_lsn = lsn_rc.lsn; + else { + /* Cases 1, 2, 3, and 5. */ + DB_ASSERT(log_compare(&lsn_rc.lsn, + &lp->ready_lsn) >= 0); + lp->waiting_lsn = lsn_rc.lsn; + } + + /* + * If the current rectype is simple, we're + * done with it, and we should check and see + * whether the next record queued is the next + * one we're ready for. This is just the loop + * condition, so we continue. + * + * Otherwise, we need to break out of this loop + * and process this record first. + */ + if (!IS_SIMPLE(rectype)) + break; + } + } + + /* + * Check if we're at a gap in the table and if so, whether we + * need to ask for any records. + */ + do_req = 0; + if (!IS_ZERO_LSN(lp->waiting_lsn) && + log_compare(&lp->ready_lsn, &lp->waiting_lsn) != 0) { + next_lsn = lp->ready_lsn; + do_req = ++lp->rcvd_recs >= lp->wait_recs; + if (do_req) { + lp->wait_recs = rep->request_gap; + lp->rcvd_recs = 0; + } + } + + R_UNLOCK(dbenv, &dblp->reginfo); + if (dbc != NULL) { + if ((ret = dbc->c_close(dbc)) != 0) + goto err; + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + have_mutex = 0; + } + dbc = NULL; + + if (do_req) { + MUTEX_LOCK(dbenv, db_rep->mutexp); + eid = db_rep->region->master_id; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + if (eid != DB_EID_INVALID) { + rep->stat.st_log_requested++; + if ((ret = __rep_send_message(dbenv, + eid, REP_LOG_REQ, &next_lsn, NULL, 0)) != 0) + goto err; + } + } + } else if (cmp > 0) { + /* + * The LSN is higher than the one we were waiting for. + * If it is a NEWFILE message, this may not mean that + * there's a gap; in some cases, NEWFILE messages contain + * the LSN of the beginning of the new file instead + * of the end of the old. + * + * In these cases, the rec DBT will contain the last LSN + * of the old file, so we can tell whether there's a gap. + */ + if (rp->rectype == REP_NEWFILE && + rp->lsn.file == lp->ready_lsn.file + 1 && + rp->lsn.offset == 0) { + DB_ASSERT(rec != NULL && rec->data != NULL && + rec->size == sizeof(DB_LSN)); + memcpy(&lsn, rec->data, sizeof(DB_LSN)); + if (log_compare(&lp->ready_lsn, &lsn) > 0) + /* + * The last LSN in the old file is smaller + * than the one we're expecting, so there's + * no gap--the one we're expecting just + * doesn't exist. + */ + goto newfile; + } + + /* + * This record isn't in sequence; add it to the table and + * update waiting_lsn if necessary. + */ + memset(&key_dbt, 0, sizeof(key_dbt)); + key_dbt.data = rp; + key_dbt.size = sizeof(*rp); + next_lsn = lp->lsn; + do_req = 0; + if (lp->wait_recs == 0) { + /* + * This is a new gap. Initialize the number of + * records that we should wait before requesting + * that it be resent. We grab the limits out of + * the rep without the mutex. + */ + lp->wait_recs = rep->request_gap; + lp->rcvd_recs = 0; + } + + if (++lp->rcvd_recs >= lp->wait_recs) { + /* + * If we've waited long enough, request the record + * and double the wait interval. + */ + do_req = 1; + lp->wait_recs <<= 1; + lp->rcvd_recs = 0; + if (lp->wait_recs > rep->max_gap) + lp->wait_recs = rep->max_gap; + } + R_UNLOCK(dbenv, &dblp->reginfo); + + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + ret = dbp->put(dbp, NULL, &key_dbt, rec, 0); + rep->stat.st_log_queued++; + rep->stat.st_log_queued_total++; + if (rep->stat.st_log_queued_max < rep->stat.st_log_queued) + rep->stat.st_log_queued_max = rep->stat.st_log_queued; + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + + if (ret != 0) + return (ret); + + R_LOCK(dbenv, &dblp->reginfo); + if (IS_ZERO_LSN(lp->waiting_lsn) || + log_compare(&rp->lsn, &lp->waiting_lsn) < 0) + lp->waiting_lsn = rp->lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + + if (do_req) { + /* Request the LSN we are still waiting for. */ + MUTEX_LOCK(dbenv, db_rep->mutexp); + + /* May as well do this after we grab the mutex. */ + eid = db_rep->region->master_id; + + /* + * If the master_id is invalid, this means that since + * the last record was sent, somebody declared an + * election and we may not have a master to request + * things of. + * + * This is not an error; when we find a new master, + * we'll re-negotiate where the end of the log is and + * try to to bring ourselves up to date again anyway. + */ + if (eid != DB_EID_INVALID) { + rep->stat.st_log_requested++; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + ret = __rep_send_message(dbenv, + eid, REP_LOG_REQ, &next_lsn, NULL, 0); + } else + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + } + return (ret); + } else { + R_UNLOCK(dbenv, &dblp->reginfo); + + /* + * We may miscount if we race, since we + * don't currently hold the rep mutex. + */ + rep->stat.st_log_duplicated++; + } + if (ret != 0 || cmp < 0 || (cmp == 0 && IS_SIMPLE(rectype))) + goto done; + + /* + * If we got here, then we've got a log record in rp and rec that + * we need to process. + */ + switch(rectype) { + case DB___dbreg_register: + /* + * DB opens occur in the context of a transaction, so we can + * simply handle them when we process the transaction. Closes, + * however, are not transaction-protected, so we have to + * handle them here. + * + * Note that it should be unsafe for the master to do a close + * of a file that was opened in an active transaction, so we + * should be guaranteed to get the ordering right. + */ + memcpy(&txnid, (u_int8_t *)rec->data + + ((u_int8_t *)&dbreg_args.txnid - (u_int8_t *)&dbreg_args), + sizeof(u_int32_t)); + if (txnid == TXN_INVALID && + !F_ISSET(dbenv, DB_ENV_REP_LOGSONLY)) + ret = __db_dispatch(dbenv, dbenv->recover_dtab, + dbenv->recover_dtab_size, rec, &rp->lsn, + DB_TXN_APPLY, NULL); + break; + case DB___txn_ckp: + /* Sync the memory pool. */ + memcpy(&ckp_lsn, (u_int8_t *)rec->data + + ((u_int8_t *)&ckp_args.ckp_lsn - (u_int8_t *)&ckp_args), + sizeof(DB_LSN)); + if (!F_ISSET(dbenv, DB_ENV_REP_LOGSONLY)) + ret = dbenv->memp_sync(dbenv, &ckp_lsn); + else + /* + * We ought to make sure the logs on a logs-only + * replica get flushed now and again. + */ + ret = dbenv->log_flush(dbenv, &ckp_lsn); + break; + case DB___txn_regop: + if (!F_ISSET(dbenv, DB_ENV_REP_LOGSONLY)) + do { + /* + * If an application is doing app-specific + * recovery and acquires locks while applying + * a transaction, it can deadlock. Any other + * locks held by this thread should have been + * discarded in the __rep_process_txn error + * path, so if we simply retry, we should + * eventually succeed. + */ + ret = __rep_process_txn(dbenv, rec); + } while (ret == DB_LOCK_DEADLOCK); + break; + default: + goto err; + } + + /* Check if we need to go back into the table. */ + if (ret == 0) { + R_LOCK(dbenv, &dblp->reginfo); + if (log_compare(&lp->ready_lsn, &lp->waiting_lsn) == 0) + goto gap_check; + R_UNLOCK(dbenv, &dblp->reginfo); + } + +done: +err: if (dbc != NULL && (t_ret = dbc->c_close(dbc)) != 0 && ret == 0) + ret = t_ret; + if (have_mutex) + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + + if (control_dbt.data != NULL) + __os_ufree(dbenv, control_dbt.data); + if (rec_dbt.data != NULL) + __os_ufree(dbenv, rec_dbt.data); + + return (ret); +} + +/* + * __rep_process_txn -- + * + * This is the routine that actually gets a transaction ready for + * processing. + * + * PUBLIC: int __rep_process_txn __P((DB_ENV *, DBT *)); + */ +int +__rep_process_txn(dbenv, rec) + DB_ENV *dbenv; + DBT *rec; +{ + DBT data_dbt; + DB_LOCKREQ req, *lvp; + DB_LOGC *logc; + DB_LSN prev_lsn, *lsnp; + DB_REP *db_rep; + LSN_COLLECTION lc; + REP *rep; + __txn_regop_args *txn_args; + __txn_xa_regop_args *prep_args; + u_int32_t lockid, op, rectype; + int i, ret, t_ret; + int (**dtab)__P((DB_ENV *, DBT *, DB_LSN *, db_recops, void *)); + size_t dtabsize; + void *txninfo; + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + + logc = NULL; + txninfo = NULL; + memset(&data_dbt, 0, sizeof(data_dbt)); + if (F_ISSET(dbenv, DB_ENV_THREAD)) + F_SET(&data_dbt, DB_DBT_REALLOC); + + /* + * There are two phases: First, we have to traverse + * backwards through the log records gathering the list + * of all LSNs in the transaction. Once we have this information, + * we can loop through, acquire the locks we need for each record, + * and then apply it. + */ + dtab = NULL; + + /* + * We may be passed a prepare (if we're restoring a prepare + * on upgrade) instead of a commit (the common case). + * Check which and behave appropriately. + */ + memcpy(&rectype, rec->data, sizeof(rectype)); + memset(&lc, 0, sizeof(lc)); + if (rectype == DB___txn_regop) { + /* + * We're the end of a transaction. Make sure this is + * really a commit and not an abort! + */ + if ((ret = __txn_regop_read(dbenv, rec->data, &txn_args)) != 0) + return (ret); + op = txn_args->opcode; + prev_lsn = txn_args->prev_lsn; + __os_free(dbenv, txn_args); + if (op != TXN_COMMIT) + return (0); + } else { + /* We're a prepare. */ + DB_ASSERT(rectype == DB___txn_xa_regop); + + if ((ret = + __txn_xa_regop_read(dbenv, rec->data, &prep_args)) != 0) + return (ret); + prev_lsn = prep_args->prev_lsn; + __os_free(dbenv, prep_args); + } + + /* Phase 1. Get a list of the LSNs in this transaction, and sort it. */ + if ((ret = __rep_collect_txn(dbenv, &prev_lsn, &lc)) != 0) + return (ret); + qsort(lc.array, lc.nlsns, sizeof(DB_LSN), __rep_lsn_cmp); + + if ((ret = dbenv->lock_id(dbenv, &lockid)) != 0) + goto err; + + /* Initialize the getpgno dispatch table. */ + if ((ret = __rep_lockpgno_init(dbenv, &dtab, &dtabsize)) != 0) + goto err; + + /* + * The set of records for a transaction may include dbreg_register + * records. Create a txnlist so that they can keep track of file + * state between records. + */ + if ((ret = __db_txnlist_init(dbenv, 0, 0, NULL, &txninfo)) != 0) + goto err; + + /* Phase 2: Apply updates. */ + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + goto err; + for (lsnp = &lc.array[0], i = 0; i < lc.nlsns; i++, lsnp++) { + if ((ret = __rep_lockpages(dbenv, + dtab, dtabsize, lsnp, NULL, NULL, lockid)) != 0) + goto err; + if ((ret = logc->get(logc, lsnp, &data_dbt, DB_SET)) != 0) + goto err; + if ((ret = __db_dispatch(dbenv, dbenv->recover_dtab, + dbenv->recover_dtab_size, &data_dbt, lsnp, + DB_TXN_APPLY, txninfo)) != 0) + goto err; + } + +err: memset(&req, 0, sizeof(req)); + req.op = DB_LOCK_PUT_ALL; + if ((t_ret = dbenv->lock_vec(dbenv, lockid, + DB_LOCK_FREE_LOCKER, &req, 1, &lvp)) != 0 && ret == 0) + ret = t_ret; + + if (lc.nalloc != 0) + __os_free(dbenv, lc.array); + + if ((t_ret = + dbenv->lock_id_free(dbenv, lockid)) != 0 && ret == 0) + ret = t_ret; + + if (logc != NULL && (t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + + if (txninfo != NULL) + __db_txnlist_end(dbenv, txninfo); + + if (F_ISSET(&data_dbt, DB_DBT_REALLOC) && data_dbt.data != NULL) + __os_ufree(dbenv, data_dbt.data); + + if (dtab != NULL) + __os_free(dbenv, dtab); + + if (ret == 0) + /* + * We don't hold the rep mutex, and could miscount if we race. + */ + rep->stat.st_txns_applied++; + + return (ret); +} + +/* + * __rep_collect_txn + * Recursive function that will let us visit every entry in a transaction + * chain including all child transactions so that we can then apply + * the entire transaction family at once. + */ +static int +__rep_collect_txn(dbenv, lsnp, lc) + DB_ENV *dbenv; + DB_LSN *lsnp; + LSN_COLLECTION *lc; +{ + __txn_child_args *argp; + DB_LOGC *logc; + DB_LSN c_lsn; + DBT data; + u_int32_t rectype; + int nalloc, ret, t_ret; + + memset(&data, 0, sizeof(data)); + F_SET(&data, DB_DBT_REALLOC); + + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + + while (!IS_ZERO_LSN(*lsnp) && + (ret = logc->get(logc, lsnp, &data, DB_SET)) == 0) { + memcpy(&rectype, data.data, sizeof(rectype)); + if (rectype == DB___txn_child) { + if ((ret = __txn_child_read(dbenv, + data.data, &argp)) != 0) + goto err; + c_lsn = argp->c_lsn; + *lsnp = argp->prev_lsn; + __os_free(dbenv, argp); + ret = __rep_collect_txn(dbenv, &c_lsn, lc); + } else { + if (lc->nalloc < lc->nlsns + 1) { + nalloc = lc->nalloc == 0 ? 20 : lc->nalloc * 2; + if ((ret = __os_realloc(dbenv, + nalloc * sizeof(DB_LSN), &lc->array)) != 0) + goto err; + lc->nalloc = nalloc; + } + lc->array[lc->nlsns++] = *lsnp; + + /* + * Explicitly copy the previous lsn. The record + * starts with a u_int32_t record type, a u_int32_t + * txn id, and then the DB_LSN (prev_lsn) that we + * want. We copy explicitly because we have no idea + * what kind of record this is. + */ + memcpy(lsnp, (u_int8_t *)data.data + + sizeof(u_int32_t) + sizeof(u_int32_t), + sizeof(DB_LSN)); + } + + if (ret != 0) + goto err; + } + +err: if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + if (data.data != NULL) + __os_ufree(dbenv, data.data); + return (ret); +} + +/* + * __rep_lsn_cmp -- + * qsort-type-compatible wrapper for log_compare. + */ +static int +__rep_lsn_cmp(lsn1, lsn2) + const void *lsn1, *lsn2; +{ + + return (log_compare((DB_LSN *)lsn1, (DB_LSN *)lsn2)); +} + +/* + * __rep_newfile -- + * NEWFILE messages can contain either the last LSN of the old file + * or the first LSN of the new one, depending on which we have available + * when the message is sent. When applying a NEWFILE message, make sure + * we haven't already swapped files, as it's possible (given the right sequence + * of out-of-order messages) to wind up with a NEWFILE message of each + * variety, and __rep_apply won't detect the two as duplicates of each other. + */ +static int +__rep_newfile(dbenv, rc, msgdbt, lsnp) + DB_ENV *dbenv; + REP_CONTROL *rc; + DBT *msgdbt; + DB_LSN *lsnp; +{ + DB_LOG *dblp; + LOG *lp; + u_int32_t newfile; + + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + + /* + * A NEWFILE message containing the old file's LSN will be + * accompanied by a NULL rec DBT; one containing the new one's LSN + * will need to supply the last record in the old file by + * sending it in the rec DBT. + */ + if (msgdbt == NULL || msgdbt->size == 0) + newfile = rc->lsn.file + 1; + else + newfile = rc->lsn.file; + + if (newfile > lp->lsn.file) + return (__log_newfile(dblp, lsnp)); + else { + /* We've already applied this NEWFILE. Just ignore it. */ + *lsnp = lp->lsn; + return (0); + } +} diff --git a/bdb/rep/rep_region.c b/bdb/rep/rep_region.c new file mode 100644 index 00000000000..1ac3fb8a20c --- /dev/null +++ b/bdb/rep/rep_region.c @@ -0,0 +1,187 @@ +/*- + * See the file LICENSE for redistribution information. + * + * Copyright (c) 2001-2002 + * Sleepycat Software. All rights reserved. + */ +#include "db_config.h" + +#ifndef lint +static const char revid[] = "$Id: rep_region.c,v 1.29 2002/08/06 04:50:36 bostic Exp $"; +#endif /* not lint */ + +#ifndef NO_SYSTEM_INCLUDES +#endif + +#include <string.h> + +#include "db_int.h" +#include "dbinc/rep.h" +#include "dbinc/log.h" + +/* + * __rep_region_init -- + * Initialize the shared memory state for the replication system. + * + * PUBLIC: int __rep_region_init __P((DB_ENV *)); + */ +int +__rep_region_init(dbenv) + DB_ENV *dbenv; +{ + REGENV *renv; + REGINFO *infop; + DB_MUTEX *db_mutexp; + DB_REP *db_rep; + REP *rep; + int ret; + + db_rep = dbenv->rep_handle; + infop = dbenv->reginfo; + renv = infop->primary; + ret = 0; + + MUTEX_LOCK(dbenv, &renv->mutex); + if (renv->rep_off == INVALID_ROFF) { + /* Must create the region. */ + if ((ret = __db_shalloc(infop->addr, + sizeof(REP), MUTEX_ALIGN, &rep)) != 0) + goto err; + memset(rep, 0, sizeof(*rep)); + rep->tally_off = INVALID_ROFF; + renv->rep_off = R_OFFSET(infop, rep); + + if ((ret = __db_mutex_setup(dbenv, infop, &rep->mutex, + MUTEX_NO_RECORD)) != 0) + goto err; + + /* + * We must create a place for the db_mutex separately; + * mutexes have to be aligned to MUTEX_ALIGN, and the only way + * to guarantee that is to make sure they're at the beginning + * of a shalloc'ed chunk. + */ + if ((ret = __db_shalloc(infop->addr, sizeof(DB_MUTEX), + MUTEX_ALIGN, &db_mutexp)) != 0) + goto err; + rep->db_mutex_off = R_OFFSET(infop, db_mutexp); + + /* + * Because we have no way to prevent deadlocks and cannot log + * changes made to it, we single-thread access to the client + * bookkeeping database. This is suboptimal, but it only gets + * accessed when messages arrive out-of-order, so it should + * stay small and not be used in a high-performance app. + */ + if ((ret = __db_mutex_setup(dbenv, infop, db_mutexp, + MUTEX_NO_RECORD)) != 0) + goto err; + + /* We have the region; fill in the values. */ + rep->eid = DB_EID_INVALID; + rep->master_id = DB_EID_INVALID; + rep->gen = 0; + + /* + * Set default values for the min and max log records that we + * wait before requesting a missing log record. + */ + rep->request_gap = DB_REP_REQUEST_GAP; + rep->max_gap = DB_REP_MAX_GAP; + } else + rep = R_ADDR(infop, renv->rep_off); + MUTEX_UNLOCK(dbenv, &renv->mutex); + + db_rep->mutexp = &rep->mutex; + db_rep->db_mutexp = R_ADDR(infop, rep->db_mutex_off); + db_rep->region = rep; + + return (0); + +err: MUTEX_UNLOCK(dbenv, &renv->mutex); + return (ret); +} + +/* + * __rep_region_destroy -- + * Destroy any system resources allocated in the replication region. + * + * PUBLIC: int __rep_region_destroy __P((DB_ENV *)); + */ +int +__rep_region_destroy(dbenv) + DB_ENV *dbenv; +{ + DB_REP *db_rep; + int ret, t_ret; + + ret = t_ret = 0; + db_rep = (DB_REP *)dbenv->rep_handle; + + if (db_rep != NULL) { + if (db_rep->mutexp != NULL) + ret = __db_mutex_destroy(db_rep->mutexp); + if (db_rep->db_mutexp != NULL) + t_ret = __db_mutex_destroy(db_rep->db_mutexp); + } + + return (ret == 0 ? t_ret : ret); +} + +/* + * __rep_dbenv_close -- + * Replication-specific destruction of the DB_ENV structure. + * + * PUBLIC: int __rep_dbenv_close __P((DB_ENV *)); + */ +int +__rep_dbenv_close(dbenv) + DB_ENV *dbenv; +{ + DB_REP *db_rep; + + db_rep = (DB_REP *)dbenv->rep_handle; + + if (db_rep != NULL) { + __os_free(dbenv, db_rep); + dbenv->rep_handle = NULL; + } + + return (0); +} + +/* + * __rep_preclose -- + * If we are a client, shut down our client database and, if we're + * actually closing the environment, close all databases we've opened + * while applying messages. + * + * PUBLIC: int __rep_preclose __P((DB_ENV *, int)); + */ +int +__rep_preclose(dbenv, do_closefiles) + DB_ENV *dbenv; + int do_closefiles; +{ + DB *dbp; + DB_REP *db_rep; + int ret, t_ret; + + ret = t_ret = 0; + + /* If replication is not initialized, we have nothing to do. */ + if ((db_rep = (DB_REP *)dbenv->rep_handle) == NULL) + return (0); + + if ((dbp = db_rep->rep_db) != NULL) { + MUTEX_LOCK(dbenv, db_rep->db_mutexp); + ret = dbp->close(dbp, 0); + db_rep->rep_db = NULL; + MUTEX_UNLOCK(dbenv, db_rep->db_mutexp); + } + + if (do_closefiles) + t_ret = __dbreg_close_files(dbenv); + + return (ret == 0 ? t_ret : ret); +} diff --git a/bdb/rep/rep_util.c b/bdb/rep/rep_util.c new file mode 100644 index 00000000000..9c99d33ed4a --- /dev/null +++ b/bdb/rep/rep_util.c @@ -0,0 +1,867 @@ +/*- + * See the file LICENSE for redistribution information. + * + * Copyright (c) 2001-2002 + * Sleepycat Software. All rights reserved. + */ + +#include "db_config.h" + +#ifndef lint +static const char revid[] = "$Id: rep_util.c,v 1.51 2002/09/05 02:30:00 margo Exp $"; +#endif /* not lint */ + +#ifndef NO_SYSTEM_INCLUDES +#include <stdlib.h> +#include <string.h> +#endif + +#include "db_int.h" +#include "dbinc/db_page.h" +#include "dbinc/btree.h" +#include "dbinc/fop.h" +#include "dbinc/hash.h" +#include "dbinc/log.h" +#include "dbinc/qam.h" +#include "dbinc/rep.h" +#include "dbinc/txn.h" + +/* + * rep_util.c: + * Miscellaneous replication-related utility functions, including + * those called by other subsystems. + */ +static int __rep_cmp_bylsn __P((const void *, const void *)); +static int __rep_cmp_bypage __P((const void *, const void *)); + +#ifdef REP_DIAGNOSTIC +static void __rep_print_logmsg __P((DB_ENV *, const DBT *, DB_LSN *)); +#endif + +/* + * __rep_check_alloc -- + * Make sure the array of TXN_REC entries is of at least size n. + * (This function is called by the __*_getpgnos() functions in + * *.src.) + * + * PUBLIC: int __rep_check_alloc __P((DB_ENV *, TXN_RECS *, int)); + */ +int +__rep_check_alloc(dbenv, r, n) + DB_ENV *dbenv; + TXN_RECS *r; + int n; +{ + int nalloc, ret; + + while (r->nalloc < r->npages + n) { + nalloc = r->nalloc == 0 ? 20 : r->nalloc * 2; + + if ((ret = __os_realloc(dbenv, nalloc * sizeof(LSN_PAGE), + &r->array)) != 0) + return (ret); + + r->nalloc = nalloc; + } + + return (0); +} + +/* + * __rep_send_message -- + * This is a wrapper for sending a message. It takes care of constructing + * the REP_CONTROL structure and calling the user's specified send function. + * + * PUBLIC: int __rep_send_message __P((DB_ENV *, int, + * PUBLIC: u_int32_t, DB_LSN *, const DBT *, u_int32_t)); + */ +int +__rep_send_message(dbenv, eid, rtype, lsnp, dbtp, flags) + DB_ENV *dbenv; + int eid; + u_int32_t rtype; + DB_LSN *lsnp; + const DBT *dbtp; + u_int32_t flags; +{ + DB_REP *db_rep; + REP *rep; + DBT cdbt, scrap_dbt; + REP_CONTROL cntrl; + u_int32_t send_flags; + int ret; + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + + /* Set up control structure. */ + memset(&cntrl, 0, sizeof(cntrl)); + if (lsnp == NULL) + ZERO_LSN(cntrl.lsn); + else + cntrl.lsn = *lsnp; + cntrl.rectype = rtype; + cntrl.flags = flags; + cntrl.rep_version = DB_REPVERSION; + cntrl.log_version = DB_LOGVERSION; + MUTEX_LOCK(dbenv, db_rep->mutexp); + cntrl.gen = rep->gen; + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + memset(&cdbt, 0, sizeof(cdbt)); + cdbt.data = &cntrl; + cdbt.size = sizeof(cntrl); + + /* Don't assume the send function will be tolerant of NULL records. */ + if (dbtp == NULL) { + memset(&scrap_dbt, 0, sizeof(DBT)); + dbtp = &scrap_dbt; + } + + send_flags = (LF_ISSET(DB_PERMANENT) ? DB_REP_PERMANENT : 0); + +#if 0 + __rep_print_message(dbenv, eid, &cntrl, "rep_send_message"); +#endif +#ifdef REP_DIAGNOSTIC + if (rtype == REP_LOG) + __rep_print_logmsg(dbenv, dbtp, lsnp); +#endif + ret = db_rep->rep_send(dbenv, &cdbt, dbtp, eid, send_flags); + + /* + * We don't hold the rep lock, so this could miscount if we race. + * I don't think it's worth grabbing the mutex for that bit of + * extra accuracy. + */ + if (ret == 0) + rep->stat.st_msgs_sent++; + else + rep->stat.st_msgs_send_failures++; + + return (ret); +} + +#ifdef REP_DIAGNOSTIC + +/* + * __rep_print_logmsg -- + * This is a debugging routine for printing out log records that + * we are about to transmit to a client. + */ + +static void +__rep_print_logmsg(dbenv, logdbt, lsnp) + DB_ENV *dbenv; + const DBT *logdbt; + DB_LSN *lsnp; +{ + /* Static structures to hold the printing functions. */ + static int (**ptab)__P((DB_ENV *, + DBT *, DB_LSN *, db_recops, void *)) = NULL; + size_t ptabsize = 0; + + if (ptabsize == 0) { + /* Initialize the table. */ + (void)__bam_init_print(dbenv, &ptab, &ptabsize); + (void)__crdel_init_print(dbenv, &ptab, &ptabsize); + (void)__db_init_print(dbenv, &ptab, &ptabsize); + (void)__dbreg_init_print(dbenv, &ptab, &ptabsize); + (void)__fop_init_print(dbenv, &ptab, &ptabsize); + (void)__qam_init_print(dbenv, &ptab, &ptabsize); + (void)__ham_init_print(dbenv, &ptab, &ptabsize); + (void)__txn_init_print(dbenv, &ptab, &ptabsize); + } + + (void)__db_dispatch(dbenv, + ptab, ptabsize, (DBT *)logdbt, lsnp, DB_TXN_PRINT, NULL); +} + +#endif +/* + * __rep_new_master -- + * Called after a master election to sync back up with a new master. + * It's possible that we already know of this new master in which case + * we don't need to do anything. + * + * This is written assuming that this message came from the master; we + * need to enforce that in __rep_process_record, but right now, we have + * no way to identify the master. + * + * PUBLIC: int __rep_new_master __P((DB_ENV *, REP_CONTROL *, int)); + */ +int +__rep_new_master(dbenv, cntrl, eid) + DB_ENV *dbenv; + REP_CONTROL *cntrl; + int eid; +{ + DB_LOG *dblp; + DB_LOGC *logc; + DB_LSN last_lsn, lsn; + DB_REP *db_rep; + DBT dbt; + LOG *lp; + REP *rep; + int change, ret, t_ret; + + db_rep = dbenv->rep_handle; + rep = db_rep->region; + MUTEX_LOCK(dbenv, db_rep->mutexp); + ELECTION_DONE(rep); + change = rep->gen != cntrl->gen || rep->master_id != eid; + if (change) { + rep->gen = cntrl->gen; + rep->master_id = eid; + F_SET(rep, REP_F_RECOVER); + rep->stat.st_master_changes++; + } + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + if (!change) + return (0); + + /* + * If the master changed, we need to start the process of + * figuring out what our last valid log record is. However, + * if both the master and we agree that the max LSN is 0,0, + * then there is no recovery to be done. If we are at 0 and + * the master is not, then we just need to request all the log + * records from the master. + */ + dblp = dbenv->lg_handle; + lp = dblp->reginfo.primary; + R_LOCK(dbenv, &dblp->reginfo); + last_lsn = lsn = lp->lsn; + if (last_lsn.offset > sizeof(LOGP)) + last_lsn.offset -= lp->len; + R_UNLOCK(dbenv, &dblp->reginfo); + if (IS_INIT_LSN(lsn) || IS_ZERO_LSN(lsn)) { +empty: MUTEX_LOCK(dbenv, db_rep->mutexp); + F_CLR(rep, REP_F_RECOVER); + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + + if (IS_INIT_LSN(cntrl->lsn)) + ret = 0; + else + ret = __rep_send_message(dbenv, rep->master_id, + REP_ALL_REQ, &lsn, NULL, 0); + + if (ret == 0) + ret = DB_REP_NEWMASTER; + return (ret); + } else if (last_lsn.offset <= sizeof(LOGP)) { + /* + * We have just changed log files and need to set lastlsn + * to the last record in the previous log files. + */ + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + return (ret); + memset(&dbt, 0, sizeof(dbt)); + ret = logc->get(logc, &last_lsn, &dbt, DB_LAST); + if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + if (ret == DB_NOTFOUND) + goto empty; + if (ret != 0) + return (ret); + } + + R_LOCK(dbenv, &dblp->reginfo); + lp->verify_lsn = last_lsn; + R_UNLOCK(dbenv, &dblp->reginfo); + if ((ret = __rep_send_message(dbenv, + eid, REP_VERIFY_REQ, &last_lsn, NULL, 0)) != 0) + return (ret); + + return (DB_REP_NEWMASTER); +} + +/* + * __rep_lockpgno_init + * Create a dispatch table for acquiring locks on each log record. + * + * PUBLIC: int __rep_lockpgno_init __P((DB_ENV *, + * PUBLIC: int (***)(DB_ENV *, DBT *, DB_LSN *, db_recops, void *), + * PUBLIC: size_t *)); + */ +int +__rep_lockpgno_init(dbenv, dtabp, dtabsizep) + DB_ENV *dbenv; + int (***dtabp)__P((DB_ENV *, DBT *, DB_LSN *, db_recops, void *)); + size_t *dtabsizep; +{ + int ret; + + /* Initialize dispatch table. */ + *dtabsizep = 0; + *dtabp = NULL; + if ((ret = __bam_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __crdel_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __db_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __dbreg_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __fop_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __qam_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __ham_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0 || + (ret = __txn_init_getpgnos(dbenv, dtabp, dtabsizep)) != 0) + return (ret); + + return (0); +} + +/* + * __rep_unlockpages -- + * Unlock the pages locked in __rep_lockpages. + * + * PUBLIC: int __rep_unlockpages __P((DB_ENV *, u_int32_t)); + */ +int +__rep_unlockpages(dbenv, lid) + DB_ENV *dbenv; + u_int32_t lid; +{ + DB_LOCKREQ req, *lvp; + + req.op = DB_LOCK_PUT_ALL; + return (dbenv->lock_vec(dbenv, lid, 0, &req, 1, &lvp)); +} + +/* + * __rep_lockpages -- + * Called to gather and lock pages in preparation for both + * single transaction apply as well as client synchronization + * with a new master. A non-NULL key_lsn means that we're locking + * in order to apply a single log record during client recovery + * to the joint LSN. A non-NULL max_lsn means that we are applying + * a transaction whose commit is at max_lsn. + * + * PUBLIC: int __rep_lockpages __P((DB_ENV *, + * PUBLIC: int (**)(DB_ENV *, DBT *, DB_LSN *, db_recops, void *), + * PUBLIC: size_t, DB_LSN *, DB_LSN *, TXN_RECS *, u_int32_t)); + */ +int +__rep_lockpages(dbenv, dtab, dtabsize, key_lsn, max_lsn, recs, lid) + DB_ENV *dbenv; + int (**dtab)__P((DB_ENV *, DBT *, DB_LSN *, db_recops, void *)); + size_t dtabsize; + DB_LSN *key_lsn, *max_lsn; + TXN_RECS *recs; + u_int32_t lid; +{ + DBT data_dbt, lo; + DB_LOCK l; + DB_LOCKREQ *lvp; + DB_LOGC *logc; + DB_LSN tmp_lsn; + TXN_RECS tmp, *t; + db_pgno_t cur_pgno; + linfo_t locks; + int i, ret, t_ret, unique; + u_int32_t cur_fid; + + /* + * There are two phases: First, we have to traverse backwards through + * the log records gathering the list of all the pages accessed. Once + * we have this information we can acquire all the locks we need. + */ + + /* Initialization */ + memset(&locks, 0, sizeof(locks)); + ret = 0; + + t = recs != NULL ? recs : &tmp; + t->npages = t->nalloc = 0; + t->array = NULL; + + /* + * We've got to be in one mode or the other; else life will either + * be excessively boring or overly exciting. + */ + DB_ASSERT(key_lsn != NULL || max_lsn != NULL); + DB_ASSERT(key_lsn == NULL || max_lsn == NULL); + + /* + * Phase 1: Fill in the pgno array. + */ + memset(&data_dbt, 0, sizeof(data_dbt)); + if (F_ISSET(dbenv, DB_ENV_THREAD)) + F_SET(&data_dbt, DB_DBT_REALLOC); + + /* Single transaction apply. */ + if (max_lsn != NULL) { + DB_ASSERT(0); /* XXX */ + /* + tmp_lsn = *max_lsn; + if ((ret = __rep_apply_thread(dbenv, dtab, dtabsize, + &data_dbt, &tmp_lsn, t)) != 0) + goto err; + */ + } + + /* In recovery. */ + if (key_lsn != NULL) { + if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0) + goto err; + ret = logc->get(logc, key_lsn, &data_dbt, DB_SET); + + /* Save lsn values, since dispatch functions can change them. */ + tmp_lsn = *key_lsn; + ret = __db_dispatch(dbenv, + dtab, dtabsize, &data_dbt, &tmp_lsn, DB_TXN_GETPGNOS, t); + + if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0) + ret = t_ret; + + /* + * If ret == DB_DELETED, this record refers to a temporary + * file and there's nothing to apply. + */ + if (ret == DB_DELETED) { + ret = 0; + goto out; + } else if (ret != 0) + goto err; + } + + if (t->npages == 0) + goto out; + + /* Phase 2: Write lock all the pages. */ + + /* Sort the entries in the array by page number. */ + qsort(t->array, t->npages, sizeof(LSN_PAGE), __rep_cmp_bypage); + + /* Count the number of unique pages. */ + cur_fid = DB_LOGFILEID_INVALID; + cur_pgno = PGNO_INVALID; + unique = 0; + for (i = 0; i < t->npages; i++) { + if (F_ISSET(&t->array[i], LSN_PAGE_NOLOCK)) + continue; + if (t->array[i].pgdesc.pgno != cur_pgno || + t->array[i].fid != cur_fid) { + cur_pgno = t->array[i].pgdesc.pgno; + cur_fid = t->array[i].fid; + unique++; + } + } + + if (unique == 0) + goto out; + + /* Handle single lock case specially, else allocate space for locks. */ + if (unique == 1) { + memset(&lo, 0, sizeof(lo)); + lo.data = &t->array[0].pgdesc; + lo.size = sizeof(t->array[0].pgdesc); + ret = dbenv->lock_get(dbenv, lid, 0, &lo, DB_LOCK_WRITE, &l); + goto out2; + } + + /* Multi-lock case. */ + locks.n = unique; + if ((ret = __os_calloc(dbenv, + unique, sizeof(DB_LOCKREQ), &locks.reqs)) != 0) + goto err; + if ((ret = __os_calloc(dbenv, unique, sizeof(DBT), &locks.objs)) != 0) + goto err; + + unique = 0; + cur_fid = DB_LOGFILEID_INVALID; + cur_pgno = PGNO_INVALID; + for (i = 0; i < t->npages; i++) { + if (F_ISSET(&t->array[i], LSN_PAGE_NOLOCK)) + continue; + if (t->array[i].pgdesc.pgno != cur_pgno || + t->array[i].fid != cur_fid) { + cur_pgno = t->array[i].pgdesc.pgno; + cur_fid = t->array[i].fid; + locks.reqs[unique].op = DB_LOCK_GET; + locks.reqs[unique].mode = DB_LOCK_WRITE; + locks.reqs[unique].obj = &locks.objs[unique]; + locks.objs[unique].data = &t->array[i].pgdesc; + locks.objs[unique].size = sizeof(t->array[i].pgdesc); + unique++; + } + } + + /* Finally, get the locks. */ + if ((ret = + dbenv->lock_vec(dbenv, lid, 0, locks.reqs, unique, &lvp)) != 0) { + /* + * If we were unsuccessful, unlock any locks we acquired before + * the error and return the original error value. + */ + (void)__rep_unlockpages(dbenv, lid); + } + +err: +out: if (locks.objs != NULL) + __os_free(dbenv, locks.objs); + if (locks.reqs != NULL) + __os_free(dbenv, locks.reqs); + + /* + * Before we return, sort by LSN so that we apply records in the + * right order. + */ + qsort(t->array, t->npages, sizeof(LSN_PAGE), __rep_cmp_bylsn); + +out2: if ((ret != 0 || recs == NULL) && t->nalloc != 0) { + __os_free(dbenv, t->array); + t->array = NULL; + t->npages = t->nalloc = 0; + } + + if (F_ISSET(&data_dbt, DB_DBT_REALLOC) && data_dbt.data != NULL) + __os_ufree(dbenv, data_dbt.data); + + return (ret); +} + +/* + * __rep_cmp_bypage and __rep_cmp_bylsn -- + * Sort functions for qsort. "bypage" sorts first by page numbers and + * then by the LSN. "bylsn" sorts first by the LSN, then by page numbers. + */ +static int +__rep_cmp_bypage(a, b) + const void *a, *b; +{ + LSN_PAGE *ap, *bp; + + ap = (LSN_PAGE *)a; + bp = (LSN_PAGE *)b; + + if (ap->fid < bp->fid) + return (-1); + + if (ap->fid > bp->fid) + return (1); + + if (ap->pgdesc.pgno < bp->pgdesc.pgno) + return (-1); + + if (ap->pgdesc.pgno > bp->pgdesc.pgno) + return (1); + + if (ap->lsn.file < bp->lsn.file) + return (-1); + + if (ap->lsn.file > bp->lsn.file) + return (1); + + if (ap->lsn.offset < bp->lsn.offset) + return (-1); + + if (ap->lsn.offset > bp->lsn.offset) + return (1); + + return (0); +} + +static int +__rep_cmp_bylsn(a, b) + const void *a, *b; +{ + LSN_PAGE *ap, *bp; + + ap = (LSN_PAGE *)a; + bp = (LSN_PAGE *)b; + + if (ap->lsn.file < bp->lsn.file) + return (-1); + + if (ap->lsn.file > bp->lsn.file) + return (1); + + if (ap->lsn.offset < bp->lsn.offset) + return (-1); + + if (ap->lsn.offset > bp->lsn.offset) + return (1); + + if (ap->fid < bp->fid) + return (-1); + + if (ap->fid > bp->fid) + return (1); + + if (ap->pgdesc.pgno < bp->pgdesc.pgno) + return (-1); + + if (ap->pgdesc.pgno > bp->pgdesc.pgno) + return (1); + + return (0); +} + +/* + * __rep_is_client + * Used by other subsystems to figure out if this is a replication + * client sites. + * + * PUBLIC: int __rep_is_client __P((DB_ENV *)); + */ +int +__rep_is_client(dbenv) + DB_ENV *dbenv; +{ + DB_REP *db_rep; + REP *rep; + int ret; + + if ((db_rep = dbenv->rep_handle) == NULL) + return (0); + rep = db_rep->region; + + MUTEX_LOCK(dbenv, db_rep->mutexp); + ret = F_ISSET(rep, REP_F_UPGRADE | REP_F_LOGSONLY); + MUTEX_UNLOCK(dbenv, db_rep->mutexp); + return (ret); +} + +/* + * __rep_send_vote + * Send this site's vote for the election. + * + * PUBLIC: int __rep_send_vote __P((DB_ENV *, DB_LSN *, int, int, int)); + */ +int +__rep_send_vote(dbenv, lsnp, nsites, pri, tiebreaker) + DB_ENV *dbenv; + DB_LSN *lsnp; + int nsites, pri, tiebreaker; +{ + DBT vote_dbt; + REP_VOTE_INFO vi; + + memset(&vi, 0, sizeof(vi)); + + vi.priority = pri; + vi.nsites = nsites; + vi.tiebreaker = tiebreaker; + + memset(&vote_dbt, 0, sizeof(vote_dbt)); + vote_dbt.data = &vi; + vote_dbt.size = sizeof(vi); + + return (__rep_send_message(dbenv, + DB_EID_BROADCAST, REP_VOTE1, lsnp, &vote_dbt, 0)); +} + +/* + * __rep_grow_sites -- + * Called to allocate more space in the election tally information. + * Called with the rep mutex held. We need to call the region mutex, so + * we need to make sure that we *never* acquire those mutexes in the + * opposite order. + * + * PUBLIC: int __rep_grow_sites __P((DB_ENV *dbenv, int nsites)); + */ +int +__rep_grow_sites(dbenv, nsites) + DB_ENV *dbenv; + int nsites; +{ + REGENV *renv; + REGINFO *infop; + REP *rep; + int nalloc, ret, *tally; + + rep = ((DB_REP *)dbenv->rep_handle)->region; + + /* + * Allocate either twice the current allocation or nsites, + * whichever is more. + */ + + nalloc = 2 * rep->asites; + if (nalloc < nsites) + nalloc = nsites; + + infop = dbenv->reginfo; + renv = infop->primary; + MUTEX_LOCK(dbenv, &renv->mutex); + if ((ret = __db_shalloc(infop->addr, + sizeof(nalloc * sizeof(int)), sizeof(int), &tally)) == 0) { + if (rep->tally_off != INVALID_ROFF) + __db_shalloc_free(infop->addr, + R_ADDR(infop, rep->tally_off)); + rep->asites = nalloc; + rep->nsites = nsites; + rep->tally_off = R_OFFSET(infop, tally); + } + MUTEX_UNLOCK(dbenv, &renv->mutex); + return (ret); +} + +#ifdef NOTYET +static int __rep_send_file __P((DB_ENV *, DBT *, u_int32_t)); +/* + * __rep_send_file -- + * Send an entire file, one block at a time. + */ +static int +__rep_send_file(dbenv, rec, eid) + DB_ENV *dbenv; + DBT *rec; + u_int32_t eid; +{ + DB *dbp; + DB_LOCK lk; + DB_MPOOLFILE *mpf; + DBC *dbc; + DBT rec_dbt; + PAGE *pagep; + db_pgno_t last_pgno, pgno; + int ret, t_ret; + + dbp = NULL; + dbc = NULL; + pagep = NULL; + mpf = NULL; + LOCK_INIT(lk); + + if ((ret = db_create(&dbp, dbenv, 0)) != 0) + goto err; + + if ((ret = dbp->open(dbp, rec->data, NULL, DB_UNKNOWN, 0, 0)) != 0) + goto err; + + if ((ret = dbp->cursor(dbp, NULL, &dbc, 0)) != 0) + goto err; + /* + * Force last_pgno to some value that will let us read the meta-dat + * page in the following loop. + */ + memset(&rec_dbt, 0, sizeof(rec_dbt)); + last_pgno = 1; + for (pgno = 0; pgno <= last_pgno; pgno++) { + if ((ret = __db_lget(dbc, 0, pgno, DB_LOCK_READ, 0, &lk)) != 0) + goto err; + + if ((ret = mpf->get(mpf, &pgno, 0, &pagep)) != 0) + goto err; + + if (pgno == 0) + last_pgno = ((DBMETA *)pagep)->last_pgno; + + rec_dbt.data = pagep; + rec_dbt.size = dbp->pgsize; + if ((ret = __rep_send_message(dbenv, eid, + REP_FILE, NULL, &rec_dbt, pgno == last_pgno)) != 0) + goto err; + ret = mpf->put(mpf, pagep, 0); + pagep = NULL; + if (ret != 0) + goto err; + ret = __LPUT(dbc, lk); + LOCK_INIT(lk); + if (ret != 0) + goto err; + } + +err: if (LOCK_ISSET(lk) && (t_ret = __LPUT(dbc, lk)) != 0 && ret == 0) + ret = t_ret; + if (dbc != NULL && (t_ret = dbc->c_close(dbc)) != 0 && ret == 0) + ret = t_ret; + if (pagep != NULL && (t_ret = mpf->put(mpf, pagep, 0)) != 0 && ret == 0) + ret = t_ret; + if (dbp != NULL && (t_ret = dbp->close(dbp, 0)) != 0 && ret == 0) + ret = t_ret; + return (ret); +} +#endif + +#if 0 +/* + * PUBLIC: void __rep_print_message __P((DB_ENV *, int, REP_CONTROL *, char *)); + */ +void +__rep_print_message(dbenv, eid, rp, str) + DB_ENV *dbenv; + int eid; + REP_CONTROL *rp; + char *str; +{ + char *type; + switch (rp->rectype) { + case REP_ALIVE: + type = "alive"; + break; + case REP_ALIVE_REQ: + type = "alive_req"; + break; + case REP_ALL_REQ: + type = "all_req"; + break; + case REP_ELECT: + type = "elect"; + break; + case REP_FILE: + type = "file"; + break; + case REP_FILE_REQ: + type = "file_req"; + break; + case REP_LOG: + type = "log"; + break; + case REP_LOG_MORE: + type = "log_more"; + break; + case REP_LOG_REQ: + type = "log_req"; + break; + case REP_MASTER_REQ: + type = "master_req"; + break; + case REP_NEWCLIENT: + type = "newclient"; + break; + case REP_NEWFILE: + type = "newfile"; + break; + case REP_NEWMASTER: + type = "newmaster"; + break; + case REP_NEWSITE: + type = "newsite"; + break; + case REP_PAGE: + type = "page"; + break; + case REP_PAGE_REQ: + type = "page_req"; + break; + case REP_PLIST: + type = "plist"; + break; + case REP_PLIST_REQ: + type = "plist_req"; + break; + case REP_VERIFY: + type = "verify"; + break; + case REP_VERIFY_FAIL: + type = "verify_fail"; + break; + case REP_VERIFY_REQ: + type = "verify_req"; + break; + case REP_VOTE1: + type = "vote1"; + break; + case REP_VOTE2: + type = "vote2"; + break; + default: + type = "NOTYPE"; + break; + } + printf("%s %s: gen = %d eid %d, type %s, LSN [%u][%u]\n", + dbenv->db_home, str, rp->gen, eid, type, rp->lsn.file, + rp->lsn.offset); +} +#endif |