@@ -27,14 +27,27 @@ type LockBasedTxMgr struct {
2727 ledgerid string
2828 db privacyenabledstate.DB
2929 validator validator.Validator
30- batch * privacyenabledstate.UpdateBatch
31- currentBlock * common.Block
32- stateListeners ledger.StateListeners
30+ stateListeners []ledger.StateListener
3331 commitRWLock sync.RWMutex
32+ current * current
33+ }
34+
35+ type current struct {
36+ block * common.Block
37+ batch * privacyenabledstate.UpdateBatch
38+ listeners []ledger.StateListener
39+ }
40+
41+ func (c * current ) blockNum () uint64 {
42+ return c .block .Header .Number
43+ }
44+
45+ func (c * current ) maxTxNumber () uint64 {
46+ return uint64 (len (c .block .Data .Data )) - 1
3447}
3548
3649// NewLockBasedTxMgr constructs a new instance of NewLockBasedTxMgr
37- func NewLockBasedTxMgr (ledgerid string , db privacyenabledstate.DB , stateListeners ledger.StateListeners ) * LockBasedTxMgr {
50+ func NewLockBasedTxMgr (ledgerid string , db privacyenabledstate.DB , stateListeners [] ledger.StateListener ) * LockBasedTxMgr {
3851 db .Open ()
3952 txmgr := & LockBasedTxMgr {ledgerid : ledgerid , db : db , stateListeners : stateListeners }
4053 txmgr .validator = valimpl .NewStatebasedValidator (txmgr , db )
@@ -71,30 +84,28 @@ func (txmgr *LockBasedTxMgr) ValidateAndPrepare(blockAndPvtdata *ledger.BlockAnd
7184 logger .Debugf ("Validating new block with num trans = [%d]" , len (block .Data .Data ))
7285 batch , err := txmgr .validator .ValidateAndPrepareBatch (blockAndPvtdata , doMVCCValidation )
7386 if err != nil {
74- txmgr .clearCache ()
87+ txmgr .reset ()
88+ return err
89+ }
90+ txmgr .current = & current {block : block , batch : batch }
91+ if err := txmgr .invokeNamespaceListeners (); err != nil {
92+ txmgr .reset ()
7593 return err
7694 }
77- txmgr .currentBlock = block
78- txmgr .batch = batch
79- return txmgr .invokeNamespaceListeners (batch )
95+ return nil
8096}
8197
82- func (txmgr * LockBasedTxMgr ) invokeNamespaceListeners (batch * privacyenabledstate.UpdateBatch ) error {
83- namespaces := batch .PubUpdates .GetUpdatedNamespaces ()
84- for _ , namespace := range namespaces {
85- listener := txmgr .stateListeners [namespace ]
86- if listener == nil {
98+ func (txmgr * LockBasedTxMgr ) invokeNamespaceListeners () error {
99+ for _ , listener := range txmgr .stateListeners {
100+ stateUpdatesForListener := extractStateUpdates (txmgr .current .batch , listener .InterestedInNamespaces ())
101+ if len (stateUpdatesForListener ) == 0 {
87102 continue
88103 }
89- logger .Debugf ("Invoking listener for state changes over namespace:%s" , namespace )
90- updatesMap := batch .PubUpdates .GetUpdates (namespace )
91- var kvwrites []* kvrwset.KVWrite
92- for key , versionedValue := range updatesMap {
93- kvwrites = append (kvwrites , & kvrwset.KVWrite {Key : key , IsDelete : versionedValue .Value == nil , Value : versionedValue .Value })
94- }
95- if err := listener .HandleStateUpdates (txmgr .ledgerid , kvwrites ); err != nil {
104+ txmgr .current .listeners = append (txmgr .current .listeners , listener )
105+ if err := listener .HandleStateUpdates (txmgr .ledgerid , stateUpdatesForListener , txmgr .current .blockNum ()); err != nil {
96106 return err
97107 }
108+ logger .Debugf ("Invoking listener for state changes:%s" , listener )
98109 }
99110 return nil
100111}
@@ -106,35 +117,28 @@ func (txmgr *LockBasedTxMgr) Shutdown() {
106117
107118// Commit implements method in interface `txmgmt.TxMgr`
108119func (txmgr * LockBasedTxMgr ) Commit () error {
109- // If statedb implementation needed bulk read optimization, cache might have been populated by
110- // ValidateAndPrepare(). Once the block is validated and committed, populated cache needs to
111- // be cleared.
112- defer txmgr .clearCache ()
113-
120+ defer txmgr .reset ()
114121 logger .Debugf ("Committing updates to state database" )
115122 txmgr .commitRWLock .Lock ()
116123 defer txmgr .commitRWLock .Unlock ()
117124 logger .Debugf ("Write lock acquired for committing updates to state database" )
118- if txmgr .batch == nil {
125+ if txmgr .current == nil {
119126 panic ("validateAndPrepare() method should have been called before calling commit()" )
120127 }
121- defer func () { txmgr .batch = nil }()
122- if err := txmgr .db .ApplyPrivacyAwareUpdates (txmgr .batch ,
123- version .NewHeight (txmgr .currentBlock .Header .Number , uint64 (len (txmgr .currentBlock .Data .Data )- 1 ))); err != nil {
128+ commitHeight := version .NewHeight (txmgr .current .blockNum (), txmgr .current .maxTxNumber ())
129+ if err := txmgr .db .ApplyPrivacyAwareUpdates (txmgr .current .batch , commitHeight ); err != nil {
124130 return err
125131 }
126132 logger .Debugf ("Updates committed to state database" )
127-
133+ // In the case of error state listeners will not recieve this call - instead a peer panic is caused by the ledger upon receiveing
134+ // an error from this function
135+ txmgr .updateStateListeners ()
128136 return nil
129137}
130138
131139// Rollback implements method in interface `txmgmt.TxMgr`
132140func (txmgr * LockBasedTxMgr ) Rollback () {
133- txmgr .batch = nil
134- // If statedb implementation needed bulk read optimization, cache might have been populated by
135- // ValidateAndPrepareBatch(). As the block commit is rollbacked, populated cache needs to
136- // be cleared now.
137- txmgr .clearCache ()
141+ txmgr .reset ()
138142}
139143
140144// clearCache empty the cache maintained by the statedb implementation
@@ -166,3 +170,32 @@ func (txmgr *LockBasedTxMgr) CommitLostBlock(blockAndPvtdata *ledger.BlockAndPvt
166170 logger .Debugf ("Committing block %d to state database" , block .Header .Number )
167171 return txmgr .Commit ()
168172}
173+
174+ func extractStateUpdates (batch * privacyenabledstate.UpdateBatch , namespaces []string ) ledger.StateUpdates {
175+ stateupdates := make (ledger.StateUpdates )
176+ for _ , namespace := range namespaces {
177+ updatesMap := batch .PubUpdates .GetUpdates (namespace )
178+ var kvwrites []* kvrwset.KVWrite
179+ for key , versionedValue := range updatesMap {
180+ kvwrites = append (kvwrites , & kvrwset.KVWrite {Key : key , IsDelete : versionedValue .Value == nil , Value : versionedValue .Value })
181+ if len (kvwrites ) > 0 {
182+ stateupdates [namespace ] = kvwrites
183+ }
184+ }
185+ }
186+ return stateupdates
187+ }
188+
189+ func (txmgr * LockBasedTxMgr ) updateStateListeners () {
190+ for _ , l := range txmgr .current .listeners {
191+ l .StateCommitDone (txmgr .ledgerid )
192+ }
193+ }
194+
195+ func (txmgr * LockBasedTxMgr ) reset () {
196+ txmgr .current = nil
197+ // If statedb implementation needed bulk read optimization, cache might have been populated by
198+ // ValidateAndPrepare(). Once the block is validated and committed, populated cache needs to
199+ // be cleared.
200+ defer txmgr .clearCache ()
201+ }
0 commit comments