summaryrefslogtreecommitdiffstats
path: root/Lib/bsddb/test/test_replication.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/bsddb/test/test_replication.py')
-rw-r--r--Lib/bsddb/test/test_replication.py444
1 files changed, 444 insertions, 0 deletions
diff --git a/Lib/bsddb/test/test_replication.py b/Lib/bsddb/test/test_replication.py
new file mode 100644
index 0000000..4668194
--- /dev/null
+++ b/Lib/bsddb/test/test_replication.py
@@ -0,0 +1,444 @@
+"""TestCases for distributed transactions.
+"""
+
+import os
+import time
+import unittest
+
+from .test_all import db, test_support, have_threads, verbose, \
+ get_new_environment_path, get_new_database_path
+
+
+#----------------------------------------------------------------------
+
+class DBReplicationManager(unittest.TestCase):
+ import sys
+ if sys.version_info[:3] < (2, 4, 0):
+ def assertTrue(self, expr, msg=None):
+ self.failUnless(expr,msg=msg)
+
+ def setUp(self) :
+ self.homeDirMaster = get_new_environment_path()
+ self.homeDirClient = get_new_environment_path()
+
+ self.dbenvMaster = db.DBEnv()
+ self.dbenvClient = db.DBEnv()
+
+ # Must use "DB_THREAD" because the Replication Manager will
+ # be executed in other threads but will use the same environment.
+ # http://forums.oracle.com/forums/thread.jspa?threadID=645788&tstart=0
+ self.dbenvMaster.open(self.homeDirMaster, db.DB_CREATE | db.DB_INIT_TXN
+ | db.DB_INIT_LOG | db.DB_INIT_MPOOL | db.DB_INIT_LOCK |
+ db.DB_INIT_REP | db.DB_RECOVER | db.DB_THREAD, 0o666)
+ self.dbenvClient.open(self.homeDirClient, db.DB_CREATE | db.DB_INIT_TXN
+ | db.DB_INIT_LOG | db.DB_INIT_MPOOL | db.DB_INIT_LOCK |
+ db.DB_INIT_REP | db.DB_RECOVER | db.DB_THREAD, 0o666)
+
+ self.confirmed_master=self.client_startupdone=False
+ def confirmed_master(a,b,c) :
+ if b==db.DB_EVENT_REP_MASTER :
+ self.confirmed_master=True
+
+ def client_startupdone(a,b,c) :
+ if b==db.DB_EVENT_REP_STARTUPDONE :
+ self.client_startupdone=True
+
+ self.dbenvMaster.set_event_notify(confirmed_master)
+ self.dbenvClient.set_event_notify(client_startupdone)
+
+ #self.dbenvMaster.set_verbose(db.DB_VERB_REPLICATION, True)
+ #self.dbenvMaster.set_verbose(db.DB_VERB_FILEOPS_ALL, True)
+ #self.dbenvClient.set_verbose(db.DB_VERB_REPLICATION, True)
+ #self.dbenvClient.set_verbose(db.DB_VERB_FILEOPS_ALL, True)
+
+ self.dbMaster = self.dbClient = None
+
+
+ def tearDown(self):
+ if self.dbClient :
+ self.dbClient.close()
+ if self.dbMaster :
+ self.dbMaster.close()
+ self.dbenvClient.close()
+ self.dbenvMaster.close()
+ test_support.rmtree(self.homeDirClient)
+ test_support.rmtree(self.homeDirMaster)
+
+ def test01_basic_replication(self) :
+ master_port = test_support.find_unused_port()
+ self.dbenvMaster.repmgr_set_local_site("127.0.0.1", master_port)
+ client_port = test_support.find_unused_port()
+ self.dbenvClient.repmgr_set_local_site("127.0.0.1", client_port)
+ self.dbenvMaster.repmgr_add_remote_site("127.0.0.1", client_port)
+ self.dbenvClient.repmgr_add_remote_site("127.0.0.1", master_port)
+ self.dbenvMaster.rep_set_nsites(2)
+ self.dbenvClient.rep_set_nsites(2)
+ self.dbenvMaster.rep_set_priority(10)
+ self.dbenvClient.rep_set_priority(0)
+
+ self.dbenvMaster.rep_set_timeout(db.DB_REP_CONNECTION_RETRY,100123)
+ self.dbenvClient.rep_set_timeout(db.DB_REP_CONNECTION_RETRY,100321)
+ self.assertEquals(self.dbenvMaster.rep_get_timeout(
+ db.DB_REP_CONNECTION_RETRY), 100123)
+ self.assertEquals(self.dbenvClient.rep_get_timeout(
+ db.DB_REP_CONNECTION_RETRY), 100321)
+
+ self.dbenvMaster.rep_set_timeout(db.DB_REP_ELECTION_TIMEOUT, 100234)
+ self.dbenvClient.rep_set_timeout(db.DB_REP_ELECTION_TIMEOUT, 100432)
+ self.assertEquals(self.dbenvMaster.rep_get_timeout(
+ db.DB_REP_ELECTION_TIMEOUT), 100234)
+ self.assertEquals(self.dbenvClient.rep_get_timeout(
+ db.DB_REP_ELECTION_TIMEOUT), 100432)
+
+ self.dbenvMaster.rep_set_timeout(db.DB_REP_ELECTION_RETRY, 100345)
+ self.dbenvClient.rep_set_timeout(db.DB_REP_ELECTION_RETRY, 100543)
+ self.assertEquals(self.dbenvMaster.rep_get_timeout(
+ db.DB_REP_ELECTION_RETRY), 100345)
+ self.assertEquals(self.dbenvClient.rep_get_timeout(
+ db.DB_REP_ELECTION_RETRY), 100543)
+
+ self.dbenvMaster.repmgr_set_ack_policy(db.DB_REPMGR_ACKS_ALL)
+ self.dbenvClient.repmgr_set_ack_policy(db.DB_REPMGR_ACKS_ALL)
+
+ self.dbenvMaster.repmgr_start(1, db.DB_REP_MASTER);
+ self.dbenvClient.repmgr_start(1, db.DB_REP_CLIENT);
+
+ self.assertEquals(self.dbenvMaster.rep_get_nsites(),2)
+ self.assertEquals(self.dbenvClient.rep_get_nsites(),2)
+ self.assertEquals(self.dbenvMaster.rep_get_priority(),10)
+ self.assertEquals(self.dbenvClient.rep_get_priority(),0)
+ self.assertEquals(self.dbenvMaster.repmgr_get_ack_policy(),
+ db.DB_REPMGR_ACKS_ALL)
+ self.assertEquals(self.dbenvClient.repmgr_get_ack_policy(),
+ db.DB_REPMGR_ACKS_ALL)
+
+ # The timeout is necessary in BDB 4.5, since DB_EVENT_REP_STARTUPDONE
+ # is not generated if the master has no new transactions.
+ # This is solved in BDB 4.6 (#15542).
+ import time
+ timeout = time.time()+10
+ while (time.time()<timeout) and not (self.confirmed_master and self.client_startupdone) :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+
+ d = self.dbenvMaster.repmgr_site_list()
+ self.assertEquals(len(d), 1)
+ self.assertEquals(d[0][0], "127.0.0.1")
+ self.assertEquals(d[0][1], client_port)
+ self.assertTrue((d[0][2]==db.DB_REPMGR_CONNECTED) or \
+ (d[0][2]==db.DB_REPMGR_DISCONNECTED))
+
+ d = self.dbenvClient.repmgr_site_list()
+ self.assertEquals(len(d), 1)
+ self.assertEquals(d[0][0], "127.0.0.1")
+ self.assertEquals(d[0][1], master_port)
+ self.assertTrue((d[0][2]==db.DB_REPMGR_CONNECTED) or \
+ (d[0][2]==db.DB_REPMGR_DISCONNECTED))
+
+ if db.version() >= (4,6) :
+ d = self.dbenvMaster.repmgr_stat(flags=db.DB_STAT_CLEAR);
+ self.assertTrue("msgs_queued" in d)
+
+ self.dbMaster=db.DB(self.dbenvMaster)
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.open("test", db.DB_HASH, db.DB_CREATE, 0o666, txn=txn)
+ txn.commit()
+
+ import time,os.path
+ timeout=time.time()+10
+ while (time.time()<timeout) and \
+ not (os.path.exists(os.path.join(self.homeDirClient,"test"))) :
+ time.sleep(0.01)
+
+ self.dbClient=db.DB(self.dbenvClient)
+ while True :
+ txn=self.dbenvClient.txn_begin()
+ try :
+ self.dbClient.open("test", db.DB_HASH, flags=db.DB_RDONLY,
+ mode=0o666, txn=txn)
+ except db.DBRepHandleDeadError :
+ txn.abort()
+ self.dbClient.close()
+ self.dbClient=db.DB(self.dbenvClient)
+ continue
+
+ txn.commit()
+ break
+
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.put("ABC", "123", txn=txn)
+ txn.commit()
+ import time
+ timeout=time.time()+10
+ v=None
+ while (time.time()<timeout) and (v==None) :
+ txn=self.dbenvClient.txn_begin()
+ v=self.dbClient.get("ABC", txn=txn)
+ txn.commit()
+ if v==None :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+ self.assertEquals("123", v)
+
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.delete("ABC", txn=txn)
+ txn.commit()
+ timeout=time.time()+10
+ while (time.time()<timeout) and (v!=None) :
+ txn=self.dbenvClient.txn_begin()
+ v=self.dbClient.get("ABC", txn=txn)
+ txn.commit()
+ if v==None :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+ self.assertEquals(None, v)
+
+class DBBaseReplication(DBReplicationManager):
+ def setUp(self) :
+ DBReplicationManager.setUp(self)
+ def confirmed_master(a,b,c) :
+ if (b == db.DB_EVENT_REP_MASTER) or (b == db.DB_EVENT_REP_ELECTED) :
+ self.confirmed_master = True
+
+ def client_startupdone(a,b,c) :
+ if b == db.DB_EVENT_REP_STARTUPDONE :
+ self.client_startupdone = True
+
+ self.dbenvMaster.set_event_notify(confirmed_master)
+ self.dbenvClient.set_event_notify(client_startupdone)
+
+ import queue
+ self.m2c = queue.Queue()
+ self.c2m = queue.Queue()
+
+ # There are only two nodes, so we don't need to
+ # do any routing decision
+ def m2c(dbenv, control, rec, lsnp, envid, flags) :
+ self.m2c.put((control, rec))
+
+ def c2m(dbenv, control, rec, lsnp, envid, flags) :
+ self.c2m.put((control, rec))
+
+ self.dbenvMaster.rep_set_transport(13,m2c)
+ self.dbenvMaster.rep_set_priority(10)
+ self.dbenvClient.rep_set_transport(3,c2m)
+ self.dbenvClient.rep_set_priority(0)
+
+ self.assertEquals(self.dbenvMaster.rep_get_priority(),10)
+ self.assertEquals(self.dbenvClient.rep_get_priority(),0)
+
+ #self.dbenvMaster.set_verbose(db.DB_VERB_REPLICATION, True)
+ #self.dbenvMaster.set_verbose(db.DB_VERB_FILEOPS_ALL, True)
+ #self.dbenvClient.set_verbose(db.DB_VERB_REPLICATION, True)
+ #self.dbenvClient.set_verbose(db.DB_VERB_FILEOPS_ALL, True)
+
+ def thread_master() :
+ return self.thread_do(self.dbenvMaster, self.c2m, 3,
+ self.master_doing_election, True)
+
+ def thread_client() :
+ return self.thread_do(self.dbenvClient, self.m2c, 13,
+ self.client_doing_election, False)
+
+ from threading import Thread
+ t_m=Thread(target=thread_master)
+ t_c=Thread(target=thread_client)
+ import sys
+ if sys.version_info[0] < 3 :
+ t_m.setDaemon(True)
+ t_c.setDaemon(True)
+ else :
+ t_m.daemon = True
+ t_c.daemon = True
+
+ self.t_m = t_m
+ self.t_c = t_c
+
+ self.dbMaster = self.dbClient = None
+
+ self.master_doing_election=[False]
+ self.client_doing_election=[False]
+
+
+ def tearDown(self):
+ if self.dbClient :
+ self.dbClient.close()
+ if self.dbMaster :
+ self.dbMaster.close()
+ self.m2c.put(None)
+ self.c2m.put(None)
+ self.t_m.join()
+ self.t_c.join()
+ self.dbenvClient.close()
+ self.dbenvMaster.close()
+ test_support.rmtree(self.homeDirClient)
+ test_support.rmtree(self.homeDirMaster)
+
+ def basic_rep_threading(self) :
+ self.dbenvMaster.rep_start(flags=db.DB_REP_MASTER)
+ self.dbenvClient.rep_start(flags=db.DB_REP_CLIENT)
+
+ def thread_do(env, q, envid, election_status, must_be_master) :
+ while True :
+ v=q.get()
+ if v == None : return
+ env.rep_process_message(v[0], v[1], envid)
+
+ self.thread_do = thread_do
+
+ self.t_m.start()
+ self.t_c.start()
+
+ def test01_basic_replication(self) :
+ self.basic_rep_threading()
+
+ # The timeout is necessary in BDB 4.5, since DB_EVENT_REP_STARTUPDONE
+ # is not generated if the master has no new transactions.
+ # This is solved in BDB 4.6 (#15542).
+ import time
+ timeout = time.time()+10
+ while (time.time()<timeout) and not (self.confirmed_master and
+ self.client_startupdone) :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+
+ self.dbMaster=db.DB(self.dbenvMaster)
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.open("test", db.DB_HASH, db.DB_CREATE, 0o666, txn=txn)
+ txn.commit()
+
+ import time,os.path
+ timeout=time.time()+10
+ while (time.time()<timeout) and \
+ not (os.path.exists(os.path.join(self.homeDirClient,"test"))) :
+ time.sleep(0.01)
+
+ self.dbClient=db.DB(self.dbenvClient)
+ while True :
+ txn=self.dbenvClient.txn_begin()
+ try :
+ self.dbClient.open("test", db.DB_HASH, flags=db.DB_RDONLY,
+ mode=0o666, txn=txn)
+ except db.DBRepHandleDeadError :
+ txn.abort()
+ self.dbClient.close()
+ self.dbClient=db.DB(self.dbenvClient)
+ continue
+
+ txn.commit()
+ break
+
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.put("ABC", "123", txn=txn)
+ txn.commit()
+ import time
+ timeout=time.time()+10
+ v=None
+ while (time.time()<timeout) and (v==None) :
+ txn=self.dbenvClient.txn_begin()
+ v=self.dbClient.get("ABC", txn=txn)
+ txn.commit()
+ if v==None :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+ self.assertEquals("123", v)
+
+ txn=self.dbenvMaster.txn_begin()
+ self.dbMaster.delete("ABC", txn=txn)
+ txn.commit()
+ timeout=time.time()+10
+ while (time.time()<timeout) and (v!=None) :
+ txn=self.dbenvClient.txn_begin()
+ v=self.dbClient.get("ABC", txn=txn)
+ txn.commit()
+ if v==None :
+ time.sleep(0.02)
+ self.assertTrue(time.time()<timeout)
+ self.assertEquals(None, v)
+
+ if db.version() >= (4,7) :
+ def test02_test_request(self) :
+ self.basic_rep_threading()
+ (minimum, maximum) = self.dbenvClient.rep_get_request()
+ self.dbenvClient.rep_set_request(minimum-1, maximum+1)
+ self.assertEqual(self.dbenvClient.rep_get_request(),
+ (minimum-1, maximum+1))
+
+ if db.version() >= (4,6) :
+ def test03_master_election(self) :
+ # Get ready to hold an election
+ #self.dbenvMaster.rep_start(flags=db.DB_REP_MASTER)
+ self.dbenvMaster.rep_start(flags=db.DB_REP_CLIENT)
+ self.dbenvClient.rep_start(flags=db.DB_REP_CLIENT)
+
+ def thread_do(env, q, envid, election_status, must_be_master) :
+ while True :
+ v=q.get()
+ if v == None : return
+ r = env.rep_process_message(v[0],v[1],envid)
+ if must_be_master and self.confirmed_master :
+ self.dbenvMaster.rep_start(flags = db.DB_REP_MASTER)
+ must_be_master = False
+
+ if r[0] == db.DB_REP_HOLDELECTION :
+ def elect() :
+ while True :
+ try :
+ env.rep_elect(2, 1)
+ election_status[0] = False
+ break
+ except db.DBRepUnavailError :
+ pass
+ if not election_status[0] and not self.confirmed_master :
+ from threading import Thread
+ election_status[0] = True
+ t=Thread(target=elect)
+ import sys
+ if sys.version_info[0] < 3 :
+ t.setDaemon(True)
+ else :
+ t.daemon = True
+ t.start()
+
+ self.thread_do = thread_do
+
+ self.t_m.start()
+ self.t_c.start()
+
+ self.dbenvMaster.rep_set_timeout(db.DB_REP_ELECTION_TIMEOUT, 50000)
+ self.dbenvClient.rep_set_timeout(db.DB_REP_ELECTION_TIMEOUT, 50000)
+ self.client_doing_election[0] = True
+ while True :
+ try :
+ self.dbenvClient.rep_elect(2, 1)
+ self.client_doing_election[0] = False
+ break
+ except db.DBRepUnavailError :
+ pass
+
+ self.assertTrue(self.confirmed_master)
+
+#----------------------------------------------------------------------
+
+def test_suite():
+ suite = unittest.TestSuite()
+ if db.version() >= (4, 6) :
+ dbenv = db.DBEnv()
+ try :
+ dbenv.repmgr_get_ack_policy()
+ ReplicationManager_available=True
+ except :
+ ReplicationManager_available=False
+ dbenv.close()
+ del dbenv
+ if ReplicationManager_available :
+ suite.addTest(unittest.makeSuite(DBReplicationManager))
+
+ if have_threads :
+ suite.addTest(unittest.makeSuite(DBBaseReplication))
+
+ return suite
+
+
+if __name__ == '__main__':
+ unittest.main(defaultTest='test_suite')
generated by cgit v0.12 at 2025-11-11 02:46:50 (GMT)