<s> """<STR_LIT>""" <EOL> import os <EOL> import unittest <EOL> from test_all import db , test_support , get_new_environment_path , get_new_database_path <EOL> from test_all import verbose <EOL> class DBTxn_distributed ( unittest . TestCase ) : <EOL> num_txns = <NUM_LIT> <EOL> nosync = True <EOL> must_open_db = False <EOL> def _create_env ( self , must_open_db ) : <EOL> self . dbenv = db . DBEnv ( ) <EOL> self . dbenv . set_tx_max ( self . num_txns ) <EOL> self . dbenv . set_lk_max_lockers ( self . num_txns * <NUM_LIT:2> ) <EOL> self . dbenv . set_lk_max_locks ( self . num_txns * <NUM_LIT:2> ) <EOL> self . dbenv . set_lk_max_objects ( self . num_txns * <NUM_LIT:2> ) <EOL> if self . nosync : <EOL> self . dbenv . set_flags ( db . DB_TXN_NOSYNC , True ) <EOL> self . dbenv . open ( self . homeDir , db . DB_CREATE | db . DB_THREAD | <EOL> db . DB_RECOVER | <EOL> db . DB_INIT_TXN | db . DB_INIT_LOG | db . DB_INIT_MPOOL | <EOL> db . DB_INIT_LOCK , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . db = db . DB ( self . dbenv ) <EOL> self . db . set_re_len ( db . DB_GID_SIZE ) <EOL> if must_open_db : <EOL> txn = self . dbenv . txn_begin ( ) <EOL> self . db . open ( self . filename , <EOL> db . DB_QUEUE , db . DB_CREATE | db . DB_THREAD , <NUM_LIT:0> <NUM_LIT> , <EOL> txn = txn ) <EOL> txn . commit ( ) <EOL> def setUp ( self ) : <EOL> self . homeDir = get_new_environment_path ( ) <EOL> self . filename = "<STR_LIT:test>" <EOL> return self . _create_env ( must_open_db = True ) <EOL> def _destroy_env ( self ) : <EOL> if self . nosync or ( db . version ( ) [ : <NUM_LIT:2> ] == ( <NUM_LIT:4> , <NUM_LIT:6> ) ) : <EOL> self . dbenv . log_flush ( ) <EOL> self . db . close ( ) <EOL> self . dbenv . close ( ) <EOL> def tearDown ( self ) : <EOL> self . _destroy_env ( ) <EOL> test_support . rmtree ( self . homeDir ) <EOL> def _recreate_env ( self , must_open_db ) : <EOL> self . _destroy_env ( ) <EOL> self . _create_env ( must_open_db ) <EOL> def test01_distributed_transactions ( self ) : <EOL> txns = set ( ) <EOL> adapt = lambda x : x <EOL> import sys <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> adapt = lambda x : bytes ( x , "<STR_LIT:ascii>" ) <EOL> for i in xrange ( self . num_txns ) : <EOL> txn = self . dbenv . txn_begin ( ) <EOL> gid = "<STR_LIT>" % db . DB_GID_SIZE <EOL> gid = adapt ( gid % i ) <EOL> self . db . put ( i , gid , txn = txn , flags = db . DB_APPEND ) <EOL> txns . add ( gid ) <EOL> txn . prepare ( gid ) <EOL> del txn <EOL> self . _recreate_env ( self . must_open_db ) <EOL> recovered_txns = self . dbenv . txn_recover ( ) <EOL> self . assertEqual ( self . num_txns , len ( recovered_txns ) ) <EOL> for gid , txn in recovered_txns : <EOL> self . assertTrue ( gid in txns ) <EOL> del txn <EOL> del recovered_txns <EOL> self . _recreate_env ( self . must_open_db ) <EOL> recovered_txns = self . dbenv . txn_recover ( ) <EOL> self . assertEqual ( self . num_txns , len ( recovered_txns ) ) <EOL> discard_txns = set ( ) <EOL> committed_txns = set ( ) <EOL> state = <NUM_LIT:0> <EOL> for gid , txn in recovered_txns : <EOL> if state == <NUM_LIT:0> or state == <NUM_LIT:1> : <EOL> committed_txns . add ( gid ) <EOL> txn . commit ( ) <EOL> elif state == <NUM_LIT:2> : <EOL> txn . abort ( ) <EOL> elif state == <NUM_LIT:3> : <EOL> txn . discard ( ) <EOL> discard_txns . add ( gid ) <EOL> state = - <NUM_LIT:1> <EOL> state += <NUM_LIT:1> <EOL> del txn <EOL> del recovered_txns <EOL> self . _recreate_env ( self . must_open_db ) <EOL> recovered_txns = self . dbenv . txn_recover ( ) <EOL> self . assertEqual ( len ( discard_txns ) , len ( recovered_txns ) ) <EOL> for gid , txn in recovered_txns : <EOL> txn . abort ( ) <EOL> del txn <EOL> del recovered_txns <EOL> self . _recreate_env ( must_open_db = True ) <EOL> recovered_txns = self . dbenv . txn_recover ( ) <EOL> self . assertTrue ( len ( recovered_txns ) == <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( committed_txns ) , self . db . stat ( ) [ "<STR_LIT>" ] ) <EOL> class DBTxn_distributedSYNC ( DBTxn_distributed ) : <EOL> nosync = False <EOL> class DBTxn_distributed_must_open_db ( DBTxn_distributed ) : <EOL> must_open_db = True <EOL> class DBTxn_distributedSYNC_must_open_db ( DBTxn_distributed ) : <EOL> nosync = False <EOL> must_open_db = True <EOL> def test_suite ( ) : <EOL> suite = unittest . TestSuite ( ) <EOL> if db . version ( ) >= ( <NUM_LIT:4> , <NUM_LIT:5> ) : <EOL> suite . addTest ( unittest . makeSuite ( DBTxn_distributed ) ) <EOL> suite . addTest ( unittest . makeSuite ( DBTxn_distributedSYNC ) ) <EOL> if db . version ( ) >= ( <NUM_LIT:4> , <NUM_LIT:6> ) : <EOL> suite . addTest ( unittest . makeSuite ( DBTxn_distributed_must_open_db ) ) <EOL> suite . addTest ( unittest . makeSuite ( DBTxn_distributedSYNC_must_open_db ) ) <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
