! ========================================================================
! Copyright (C) by VMware, Inc. 1991-2011.  All Rights Reserved
!
! $Id: gscommitlist.gs,v 1.7 2008-01-09 22:50:11 stever Exp $
!
! ========================================================================

expectvalue %String
run
Array _newKernelSubclass: #GsCommitList
  instVarNames: #(#localCoordinator #voteResults #commitResults)
  classVars: #()
  classInstVars: #()
  poolDictionaries: #()
  inDictionary: Globals
  constraints: #()
  instancesInvariant: false
  isModifiable: false
  reservedOop: 881
%

!========================================================================
! remove all methods
removeallmethods GsCommitList
removeallclassmethods GsCommitList

category: 'For Documentation Installation only'
classmethod: GsCommitList
installDocumentation

| doc txt |
doc := GsClassDocumentation _newForPrivateGsClass: self.

txt := (GsDocText new) details: ''.
doc documentClassWith: txt.

self description: doc.
%

! ------------------- Class methods for GsCommitList
! ------------------- Instance methods for GsCommitList
category: 'Commit'
method: GsCommitList
_errorFailedCommitAfterRemoteCommits

"During commit, this transaction failed to commit after all remote sessions had
 been told to commit and done so successfully."

self _error: #rtErrLocalSessionFailedCommit args: #()
%

category: 'Commit'
method: GsCommitList
_errorSessionCouldNotCommit: session

"During commit, a session could not commit after it had voted that it could
 (and all others had voted positively)."

self _error: #rtErrRemoteSessionFailedCommit args: #[ session _publicName ]
%

category: 'Commit'
method: GsCommitList
_errorSessionTimedOut: session

"During commit, a non-blocking call timed out for the given session."

System rcValueCacheAt: #'Synchronized-Commit' 
  put: (session _publicName , ' non-blocking operation timed out')
  for: System.
"
self _error: #rtErrGsCommitListNBTimeout args: #[ session _publicName ]
"
%

category: 'Commit'
method: GsCommitList
_sessionVotedCouldNotCommit: session

"During commit, a session voted that it could not commit.  Writes an entry in
 temporary session state so that transactionConflicts will show this failure."

System rcValueCacheAt: #'Synchronized-Commit' 
  put: (session _publicName , ' voted negative')
  for: System
%

category: 'Logging'
method: GsCommitList
_writeCommitDoneRecord

"Writes a record to the transaction log indicating that the transaction has been
 committed."

%

category: 'Logging'
method: GsCommitList
_writeIntentToCommitRecord: sessions

"Writes a record to the transaction log indicating that the transaction is
 considered committed."

%

category: 'Commit'
method: GsCommitList
abort

"Aborts all sessions in the receiver."

| sessions |

self isEmpty
  ifTrue: [
    ^ localCoordinator == nil
      ifTrue: [ System _localAbort ]
      ifFalse: [ localCoordinator _abort ]
  ].

sessions := Array withAll: self.
" tell each remote session to abort "
1 to: self size do: [ :i |
  (sessions at: i) _abort
].

^ localCoordinator == nil
  ifTrue: [ System _localAbort ]
  ifFalse: [ localCoordinator _abort ]
%

category: 'Commit'
method: GsCommitList
useNonBlocking

"Return whether the receiver can use the non-blocking operations to perform
synchronized commits."

1 to: self size do: [ :i |
  (self at: i) _isNonBlocking
    ifFalse: [ ^ false ]
].
^ true
%

category: 'Commit'
method: GsCommitList
commit: commitMode

"Commits the current transaction similar to two-phase commit.  Returns true if
 this transaction and all spawned sessions committed successfully."

^ self useNonBlocking
  ifTrue: [ self nbCommit: commitMode ]
  ifFalse: [ self _commit: commitMode ]
%

category: 'Commit'
method: GsCommitList
_commit: commitMode

"Commits the current transaction similar to two-phase commit.  Returns true if
 this transaction and all spawned sessions committed successfully.  Uses
 blocking operations."

| sz sessions sess writingSessions localCommitResult localVoteResult |

sz := self size.
voteResults := Array new: sz.
commitResults := Array new: sz.

self isEmpty
  ifTrue: [
    ^ localCoordinator == nil
      ifTrue: [ System _localCommit: commitMode ]
      ifFalse: [ localCoordinator _commit: commitMode ]
  ].

" first see if the localCoordinator can commit "
localVoteResult := localCoordinator == nil
  ifTrue: [ System _prepareToCommit ]
  ifFalse: [ localCoordinator _voteToCommit ~~ 2 ].

localVoteResult
  ifFalse: [ ^ false ].

writingSessions := Array new: sz.
" ask each remote session to vote "
1 to: sz do: [ :i | | voteResult |
  sess := self at: i.
  voteResults at: i put: (voteResult := sess _voteToCommit).
  voteResult == 0
    ifTrue: [ " remote session was read-only "
      writingSessions at: i put: false.
    ]
    ifFalse: [ 
      writingSessions at: i put: true.
      voteResult == 2
        ifTrue: [
          self _sessionVotedCouldNotCommit: sess.
          ^ false
        ]
    ].
].

" at this point, all sessions voted 'yes' (or were read-only) "

" write log record that we plan to commit "
self _writeIntentToCommitRecord: writingSessions.

localCommitResult := localCoordinator == nil
  ifTrue: [ System _localCommit: commitMode ]
  ifFalse: [ localCoordinator _commit: commitMode ].

localCommitResult
  ifTrue: [
    sessions := Array withAll: self.

    " tell each writing remote session to commit "
    1 to: sz do: [ :i | | commitResult |
      sess := sessions at: i.
      (writingSessions at: i)
        ifTrue: [
          commitResult := sess _commit: commitMode.
          commitResults at: i put: commitResult.
          commitResult
            ifFalse: [ " raising this error is not necessary when 2PC works "
              " xxx : when 2PC is implemented, continue to commit others, then raise error "
              self _errorSessionCouldNotCommit: sess.
              ^ false
            ]
        ]
        " tell read-only sessions to abort "
        ifFalse: [ sess _abort ]
    ]
  ]
  ifFalse: [ " raising this error is not necessary when 2PC works "
    self _errorFailedCommitAfterRemoteCommits.
    ^ false
  ].

self _writeCommitDoneRecord.

^ true
%

category: 'Accessing'
method: GsCommitList
localCoordinator

"Returns the value of the instance variable 'localCoordinator'."

^ localCoordinator
%

category: 'Updating'
method: GsCommitList
localCoordinator: newValue

"Modifies the value of the instance variable 'localCoordinator'."

localCoordinator := newValue
%

category: 'Accessing'
method: GsCommitList
voteResults

"Returns the value of the instance variable 'voteResults'."

^ voteResults
%

category: 'Accessing'
method: GsCommitList
commitResults

"Returns the value of the instance variable 'commitResults'."

^ commitResults
%

category: 'Commit'
method: GsCommitList
nbCommit: commitMode

"Commits the current transaction similar to two-phase commit.  Returns true if
 this transaction and all spawned sessions committed successfully.  Uses
 non-blocking operations."

| sz sessions sess writingSessions localCommitResult localVoteResult
startTime endTimes availSessions unAvailSessions timedOut systm |

sz := self size.
voteResults := Array new: sz.
commitResults := Array new: sz.

self isEmpty
  ifTrue: [
    ^ localCoordinator == nil
      ifTrue: [ System _localCommit: commitMode ]
      ifFalse: [ localCoordinator _commit: commitMode ]
  ].

" quick check to see if can commit locally "
systm := System .
localVoteResult := systm _validateTransaction.
( localVoteResult == 0 _or: [ localVoteResult == 1 ])
  ifFalse: [ ^ false ].

" broadcast the request for each remote session to vote "
1 to: sz do: [ :i | (self at: i) _nbVoteToCommit ].

" calculate end times for each session "
startTime := systm timeGmt95.
endTimes := Array new: sz.
availSessions := Array new: sz .
1 to: sz do: [ :i |
  endTimes at: i put: startTime + (self at: i) _nbTimeout.
  availSessions at: i put: i.
].

" see if the local session can commit "
localVoteResult := localCoordinator == nil
  ifTrue: [ systm _prepareToCommit ]
  ifFalse: [ localCoordinator _voteToCommit ~~ 2 ].

localVoteResult
  ifFalse: [ " cancel sessions requests to vote "
    1 to: sz do: [ :i | (self at: i) _nbCancel ].
    ^ false
  ].

unAvailSessions := Array new .
writingSessions := Array new: sz.
[ availSessions isEmpty ] whileFalse: [ | voteResult sessIndex |
  unAvailSessions size: 0.
  1 to: availSessions size do: [ :i |
    sessIndex := availSessions at: i.
    sess := self at: sessIndex.
    sess _nbEnd == 1
      ifTrue: [ " done "
        unAvailSessions add: sessIndex.
        voteResult := sess _nbVoteResult.
        voteResults at: sessIndex put: voteResult.
        " record whether session wrote anything "
        writingSessions at: sessIndex put: (voteResult ~~ 0).
      ]
      ifFalse: [
        " check for timeout "
        systm timeGmt95 > (endTimes at: sessIndex)
          ifTrue: [
            sess _nbCancel.
            unAvailSessions add: sessIndex.
            voteResults at: sessIndex put: 2.
            timedOut := sess.
          ]
      ]
  ].
  availSessions removeAll: unAvailSessions.

  " if any voted no, stop checking for results "
  (voteResults includesIdentical: 2)
    ifTrue: [
      1 to: availSessions size do: [ :i |
        (self at: (availSessions at: i)) _nbCancel
      ].
      availSessions size: 0
    ].
].

timedOut ~~ nil
  ifTrue: [ 
    self _errorSessionTimedOut: timedOut.
    ^ false
  ].

" see if any voted negative "
(voteResults includesIdentical: 2)
  ifTrue: [
    self _sessionVotedCouldNotCommit: (self at: (voteResults indexOf: 2)).
    ^ false
  ].

" at this point, all sessions voted 'yes' (or were read-only) "

" write log record that we plan to commit "
self _writeIntentToCommitRecord: writingSessions.

sessions := Array withAll: self.

" broadcast the commit to each remote session that wrote something "
1 to: sz do: [ :i |
  sess := sessions at: i.
  (writingSessions at: i)
    ifTrue: [ sess _nbCommit: commitMode ]
    ifFalse: [ sess _nbAbort ]
].

" calculate end times for each session "
startTime := systm timeGmt95.
availSessions size: sz .
1 to: sz do: [ :i |
  endTimes at: i put: startTime + (sessions at: i) _nbTimeout.
  availSessions at: i put: i.
].

localCommitResult := localCoordinator == nil
  ifTrue: [ systm _localCommit: commitMode ]
  ifFalse: [ localCoordinator _commit: commitMode ].

localCommitResult
  ifFalse: [ " raising this error is not necessary when 2PC works "
    " cancel non-blocking commit "
    1 to: sz do: [ :i |
      (writingSessions at: i)
        ifTrue: [ (sessions at: i) _nbCancel ]
    ].
    self _errorFailedCommitAfterRemoteCommits.
    ^ false
  ].

[ availSessions isEmpty ] whileFalse: [ | commitResult sessIndex |
  unAvailSessions size: 0.
  1 to: availSessions size do: [ :i |
    sessIndex := availSessions at: i.
    sess := sessions at: sessIndex.
    sess _nbEnd == 1
      ifTrue: [ " done "
        unAvailSessions add: sessIndex.
        commitResult := sess _nbCommitResult.
        commitResults at: sessIndex put: commitResult.
      ]
      ifFalse: [
        " check for timeout "
        systm timeGmt95 > (endTimes at: sessIndex)
          ifTrue: [
            sess _nbCancel.
            unAvailSessions add: sessIndex.
            commitResults at: sessIndex put: false.
            timedOut := sess.
          ]
      ]
  ].
  availSessions removeAll: unAvailSessions.
    
  " if any failed to commit, stop checking for results "
  (commitResults includesIdentical: false)
    ifTrue: [
      1 to: availSessions size do: [ :i |
        (self at: (availSessions at: i)) _nbCancel
      ].
      availSessions size: 0
    ]
].

timedOut ~~ nil
  ifTrue: [ 
    self _errorSessionTimedOut: timedOut.
    self _errorSessionCouldNotCommit: timedOut.
    ^ false
  ].

" see if any failed to commit "
(commitResults includesIdentical: false)
  ifTrue: [
    self _errorSessionCouldNotCommit:
      (self at: (commitResults indexOf: false)).
    ^ false
  ].

self _writeCommitDoneRecord.

^ true
%

category: 'Commit'
method: GsCommitList
beginTransaction

"Starts a new transaction for the local session, and any remote
sessions."

| sessions |

self isEmpty
  ifTrue: [
    ^ localCoordinator == nil
      ifTrue: [ System _localBeginTransaction ]
      ifFalse: [ localCoordinator _beginTransaction ]
  ].

sessions := Array withAll: self.
" tell each remote session to change "
1 to: self size do: [ :i |
  (sessions at: i) _beginTransaction
].

^ localCoordinator == nil
  ifTrue: [ System _localBeginTransaction ]
  ifFalse: [ localCoordinator _beginTransaction ]
%

category: 'Commit'
method: GsCommitList
transactionMode: newMode

"Sets a new transaction mode for the local session and any remote sessions,
and exits the previous mode by aborting the current transaction.
Valid arguments are #autoBegin, #manualBegin and #transactionless."

| sessions |

self isEmpty
  ifTrue: [
    ^ localCoordinator == nil
      ifTrue: [ System _localTransactionMode: newMode ]
      ifFalse: [ localCoordinator _transactionMode: newMode ]
  ].

sessions := Array withAll: self.
" tell each remote session to change "
1 to: self size do: [ :i |
  (sessions at: i) _transactionMode: newMode
].

^ localCoordinator == nil
  ifTrue: [ System _localTransactionMode: newMode ]
  ifFalse: [ localCoordinator _transactionMode: newMode ]
%

