!=========================================================================
! Copyright (C) VMware, Inc. 1986-2011.  All Rights Reserved.
!
! $Id: rcq.gs,v 1.15.2.1 2008-04-07 13:33:06 normg Exp $
!
! Superclass Hierarchy:
!   RcQueue, Collection, Object.
!
! class created in idxclasses.topaz
!=========================================================================

! remove all methods
removeallmethods RcQueue
removeallclassmethods RcQueue

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

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

txt := (GsDocText new) details:
'An RcQueue (reduced-conflict queue) is an implementation of a FIFO queue that
 provides significantly reduced concurrency conflicts when used in an
 environment with multiple producers (users that add elements to the queue) and
 a single consumer (a user that removes items from the queue).  Producers are
 guaranteed not to conflict with each other, nor with a single consumer.  An
 RcQueue is implemented as a collection of RcQueueSessionComponents, each of
 which contains the queue elements submitted by a particular session.'.

doc documentClassWith: txt.

txt := (GsDocText new) details:
'An instance of the class RcQueueRemovalSeqNumbers (essentially an
 Array of SmallIntegers) representing the order in which elements are
 to be removed from the queue.'.
doc documentInstVar: #removalSeqNumbers with: txt.

txt := (GsDocText new) details:
'Searching methods should be used with care because they tend to iterate 
 over the whole collection and nullify the concurrency conflict mechanisms
 built into RC classes.'.
doc documentCategory: #Searching with: txt.

self description: doc.
%

! for 31160, increased the default number of users
category: 'Instance Creation'
classmethod: RcQueue
new

"Returns a new RcQueue."

^ (super new: 20) _initialize
%

category: 'Instance Creation'
classmethod: RcQueue
new: initialNumberOfUsers

"Returns a new RcQueue with a size that supports initialNumberOfUsers.  The
 new RcQueue will handle more users, but will have subcomponents created
 for initialNumberOfUsers."

| newOne |
newOne := (super new: initialNumberOfUsers) _initialize.
^ newOne initialize
%

category: 'Enumerating'
method: RcQueue
_collectElems

"This private method is added to support C++ access to the valid elements
 in the RcQueue (the C++ equivalent of do:)."

| resultArray |

resultArray := Array new.
self do: [:each | resultArray add: each].
^resultArray
%
category: 'Comparing'
method: RcQueue
= anRcQueue

"(R) Returns true if all of the following conditions are true:

 1.  The receiver and anRcQueue are of the same class.
 2.  The two collections are of the same size.
 3.  The corresponding elements of the receiver and anRcQueue are equal."

| selfSorted otherSorted |

self == anRcQueue
  ifTrue: [ ^ true ].

self _class == anRcQueue _class
  ifFalse: [ ^ false ].

self size == anRcQueue size
  ifFalse: [ ^ false ].

self size == 0 ifTrue:
    [^true].

selfSorted := self _timeSortedComponents.
otherSorted := anRcQueue _timeSortedComponents.

1 to: selfSorted size do:
    [ :i |
     (selfSorted at: i) value = (otherSorted at: i) value
         ifFalse: [^false]].
^true
%
category: 'Private'
method: RcQueue
_timeSortedComponents
    "Answer a SortedCollection of the receiver's internal queue
    element objects sorted by increasing time."

    | collection |
    collection := SortedCollection sortBlock: [ :a :b | a isOlderThan: b].
    self _doComponents:
        [ :each |
         collection add: each].
    ^collection
%
category: 'Private'
method: RcQueue
_doComponents: aBlock
    "Evaluates aBlock with each of the current element components of
    the RcQueue as the argument. The argument aBlock must be a
    one-argument block. This method does not traverse the queue
    elements in order."

    | sessionComponent firstValid |
    1 to: self maxSessionId do: 
        [ :i |
         sessionComponent := self at: i.
         sessionComponent ~~ nil
             ifTrue: 
                 [firstValid := self _findFirstValidQueueElementForSession: i.
                  firstValid to: sessionComponent size do: 
                      [ :j |
                       aBlock value: (sessionComponent at: j)]]]
%
category: 'Enumerating'
method: RcQueue
do: aBlock

"Evaluates aBlock with each of the current elements of the RcQueue as the
 argument.  The argument aBlock must be a one-argument block.  This method does
 not traverse the queue elements in order."

| sessionComponent firstValid |

1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      firstValid to: sessionComponent size do: [:j |
        aBlock value: (sessionComponent at: j) value].
      ]
    ]
%

category: 'Removing'
method: RcQueue
removeAll

"Removes all entries from the RcQueue, and returns an Array that contains those
 entries, in order.  It is more efficient to use removeAll than to send the
 message remove repeatedly."

| resultArray validSessions firstValidInSession sz
 minElement minSession firstValid sessionComponent anElement |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

sz := self size.
resultArray := Array new: sz.
sz == 0
  ifTrue: [ ^ resultArray ].

validSessions := Array new.
firstValidInSession := Array new.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         firstValidInSession add: firstValid.
         removalSeqNumbers at: i put:
            ((sessionComponent at: (sessionComponent size)) sequenceNumber)
         ]
    ]
  ].
1 to: resultArray size do: [:j |
  minSession := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
    (anElement isOlderThan: minElement) ifTrue: [
           minElement := anElement.
           minSession := i.
         ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      resultArray at: j put: minElement value.
      minElement value: nil.
      firstValidInSession at: minSession put:
                      ((firstValidInSession at: minSession) + 1).
    ]
  ].

sz := System rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifFalse: [ System rcValueCacheAt: #size put: 0 for: self ].

^resultArray
%

category: 'Removing'
method: RcQueue
removeCount: maxToRemove

"Removes entries from the RcQueue, and returns an Array that contains the
 minimum of maxToRemove or the queue size entries, in order.
 It is more efficient to remove multiple entries than to send the
 message remove repeatedly."

| resultArray validSessions validSessionIds firstValidInSession sz
  minElement minSession firstValid sessionComponent anElement |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

resultArray := Array new: ((self size) min: (maxToRemove max: 0)).
validSessions := Array new.
validSessionIds := Array new.
firstValidInSession := Array new.

1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         validSessionIds add: i.
         firstValidInSession add: firstValid.
         ]
    ]
  ].
1 to: resultArray size do: [:j |
  minSession := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
      (anElement isOlderThan: minElement) ifTrue: [
         minElement := anElement.
         minSession := i.
         ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      resultArray at: j put: minElement value.
      minElement value: nil.
      removalSeqNumbers at: (validSessionIds at: minSession)
                        put: (minElement sequenceNumber).
      firstValidInSession at: minSession
                        put: ((firstValidInSession at: minSession) + 1).
    ]
  ].

sz := System rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
  ifFalse: [
    System rcValueCacheAt: #size put: (sz - resultArray size) for: self
  ].

^resultArray
%

category: 'Removing'
method: RcQueue
remove

"Removes the leading element from the receiver and returns that element.
 If the receiver is empty, returns nil."

| minElement minSession firstValid sessionComponent anElement val sz |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

minSession := 0.
minElement := nil.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
        anElement := sessionComponent at: firstValid.
       (anElement isOlderThan: minElement)
          ifTrue: [
             minElement := anElement.
             minSession := i.
           ].
        ]
    ]
  ].
( minSession > 0 _and: [ minElement ~~ nil ] )
   ifTrue: [ | systm |
     removalSeqNumbers at: minSession put: (minElement sequenceNumber).
     val := minElement value.
     minElement value: nil.

     systm := System .
     sz := systm rcValueCacheAt: #size for: self otherwise: nil.
     sz == nil
         ifFalse: [ systm rcValueCacheAt: #size put: (sz - 1) for: self ].

     ^ val
   ]
   ifFalse: [^nil]
%

category: 'Removing'
method: RcQueue
peek

"Returns the leading element from the receiver without removing it.
 If the receiver is empty, returns nil."

| minElement minSession firstValid sessionComponent anElement |


minSession := 0.
minElement := nil.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
        anElement := sessionComponent at: firstValid.
        (anElement isOlderThan: minElement) ifTrue: [
         minElement := anElement.
         minSession := i.
         ].
        ]
    ]
  ].
(minSession > 0)
   ifTrue: [
     ^ minElement value.
   ]
   ifFalse: [^nil]
%

category: 'Accessing'
method: RcQueue
maxSessionId

"Returns the max sessionId that can be used with this RcQueue."
System _addToRcReadSet: self includingAllNodes: true.
^ super size
%

category: 'Accessing'
method: RcQueue
numberInvalid

"Returns the number of entries in the RcQueue which have been removed but
 not yet reclaimed in the session components for sessions that are not active.
 The intent is to determine if it is worth invoking cleanup."

| sz invalidSessions count firstValid |

sz := super size.
invalidSessions := Array new: sz.
1 to: sz do: [:i | invalidSessions at: i put: true].
System currentSessions do: [:each |
  (each < sz) ifTrue: [
     invalidSessions at: each + 1 put: false]
  ].

count := 0.
1 to: sz do: [:i |
  (invalidSessions at: i) ifTrue: [
    firstValid := self _findFirstValidQueueElementForSession: i.
    firstValid ~~ nil
      ifTrue: [ count := count + (firstValid - 1) ]
    ]
  ].
^count
%

category: 'Accessing'
method: RcQueue
_calculateSize

"Returns the number of valid entries in the RcQueue."

| count firstValid systm |

count := 0.
systm := System.
systm _addToRcReadSet: self includingAllNodes: true.
1 to: super size do: [:i |
  firstValid := self _findFirstValidQueueElementForSession: i.
  firstValid ~~ nil
    ifTrue: [ count := count + ((self at: i) size - firstValid + 1) ]
  ].

systm rcValueCacheAt: #size put: count for: self.

^count
%

category: 'Accessing'
method: RcQueue
size

"Returns the number of valid entries in the RcQueue."

| sz |
sz := System rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifTrue: [ sz := self _calculateSize ].
^ sz
%

category: 'Updating'
method: RcQueue
changeMaxSessionId: newMaxSessionId

"Changes the maximum number of sessions for which the RcQueue is configured.
 Modifying the capacity of an RcQueue in this way may cause concurrency conflicts
 with the consumer session, if one is active."

| oldSize firstValid|

oldSize := self maxSessionId.
(oldSize == newMaxSessionId) ifTrue:[ ^self ].

(newMaxSessionId > oldSize)
  ifTrue: [
    removalSeqNumbers size: newMaxSessionId.
    super size: newMaxSessionId.
    oldSize + 1 to: newMaxSessionId do: [:i |
      removalSeqNumbers at: i put: 0.
      ]
    ]
  ifFalse: [
    newMaxSessionId + 1 to: oldSize do: [:i |
      firstValid := self _findFirstValidQueueElementForSession: i.
      firstValid ~~ nil
        ifTrue: [
          (firstValid <= (self at: i) size) ifTrue:
             [ ^ self _errorRcQueueEntriesFound].
          ]
        ].
    super size: newMaxSessionId
    ].
^self
%

category: 'Updating'
method: RcQueue
cleanupQueue

"Removes obsolete entries belonging to inactive sessions.  Can cause
 concurrency conflicts with the consumer."

| maxSessionId invalidSessions firstValid
  sessionComponent sessionComponentSize systm |

maxSessionId := self maxSessionId.
invalidSessions := Array new: maxSessionId.
systm := System.
1 to: maxSessionId do: [:i | invalidSessions at: i put: true].
systm currentSessions do: [:each |
  (each < maxSessionId) ifTrue: [
     invalidSessions at: each + 1 put: false]
  ].

1 to: maxSessionId do: [:i |
  (invalidSessions at: i) ifTrue: [
    sessionComponent := self at: i.
    sessionComponent ~~ nil
      ifTrue: [
        firstValid := self _findFirstValidQueueElementForSession: i.
        sessionComponentSize := sessionComponent size.
        removalSeqNumbers at: i put: 0.
        (firstValid <= sessionComponentSize)
           ifTrue: [
             (firstValid > 1) ifTrue:[
               sessionComponent removeFrom: 1 to: firstValid - 1 .
               ].
             ]
            ifFalse: [ sessionComponent size: 0]
      ]
    ]
  ].
^self
%

category: 'Updating'
method: RcQueue
size: anInteger

"Disallowed.  You cannot change the size of anRcQueue other than to add or
 remove elements.  To change the maximum sessionId that can be used with
 this queue see changeMaxSessionId:."

^self shouldNotImplement: #size:
%

category: 'Testing'
method: RcQueue
isEmpty

"Returns true if the queue is empty, and false otherwise."
System _addToRcReadSet: self includingAllNodes: true.
^self size == 0
%

category: 'Clustering'
method: RcQueue
clusterDepthFirst

"Performs no action, as clustering defeats the conflict-reduction scheme."

^true
%

category: 'Performance Enhancement'
method: RcQueue
cleanupMySession

"Cleans up the entries for my session that have already been removed by
 the consumer.  This method is only needed if a producer adds entries to
 the queue faster than the consumer removes them and then no longer adds
 to the queue.  In this situation, the producer can enhance the performance
 of the consumer by either logging out or periodically executing this
 method and committing."

| sessionId sessionComponent firstValid systm |
systm := System.
sessionId := systm session + 1.
( sessionId > self maxSessionId _or:
[ (sessionComponent := self at: sessionId) == nil ] )
  ifTrue:[ ^ self ].

systm _addToRcReadSet: sessionComponent includingAllNodes: true.
"clean up entries that have been removed"
firstValid := self _findFirstValidQueueElementForSession: sessionId.
(firstValid > 1) ifTrue: [
   sessionComponent removeFrom: 1 to: firstValid - 1 .
   ].
^self
%

category: 'Enumerating'
method: RcQueue
speciesForCollect

"(R) Returns a class, an instance of which should be used as the result of
 collect: or other projections applied to the receiver."

^ Array
%

category: 'Enumerating'
method: RcQueue
speciesForSelect

"(R) Returns a class, an instance of which should be used as the result of
 select: or other projections applied to the receiver."

^ Array
%

category: 'Class Membership'
method: RcQueue
species

"(R) Returns a class similar to, or the same as, the receiver's class which
 can be used for containing derived copies of the receiver."

^ Array
%

category: 'Private'
method: RcQueue
_getSessionComponentFor: offset

"Create an RcQueueSessionComponent for the session with the given offset."

| sessionComponent |
offset > self maxSessionId
  ifTrue: [ self changeMaxSessionId: offset ].

sessionComponent := self _rcAt: offset.
sessionComponent == nil
    ifTrue: [
      sessionComponent := self class varyingConstraint new.
      sessionComponent assignToSegment: self segment.
      self at: offset put: sessionComponent.
      System redoLog addConflictObject: self for: self.
      System redoLog addConflictObject: removalSeqNumbers for: self.
    ].
^ sessionComponent
%

category: 'Adding'
method: RcQueue
add: aValue

"Adds aValue to the RcQueue, returns the receiver."

| sessionId queueElement sessionComponent sessCompSize firstValid lastSeqNumber sz systm |
systm := System .
sessionId := systm session + 1.
sessionComponent := self _getSessionComponentFor: sessionId.
sessCompSize := sessionComponent size.
(sessCompSize > 0)
  ifTrue: [
    queueElement := sessionComponent at: sessCompSize.
    lastSeqNumber := queueElement sequenceNumber.
    systm _addToRcReadSet: queueElement includingAllNodes: false.
  ]
  ifFalse: [lastSeqNumber := removalSeqNumbers at: sessionId].

queueElement := sessionComponent class speciesOfElements 
                newWithValue: aValue sequenceNumber: lastSeqNumber + 1 inSegmentOf: self .
systm _addToRcReadSet: sessionComponent includingAllNodes: true.
sessionComponent at: (sessionComponent size + 1) put: queueElement.

"clean up entries that have been removed"
firstValid := self _findFirstValidQueueElementForSession: sessionId.
(firstValid > 1) ifTrue: [
   sessionComponent removeFrom: 1 to: firstValid - 1 .
   ].

sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifFalse: [ systm rcValueCacheAt: #size put: (sz + 1) for: self ].

^aValue
%

category: 'Private'
method: RcQueue
_findFirstValidQueueElementForSession: aSessionId

"Returns the index of the first valid queue element for a session.

 The first valid queue element is the one whose sequence number is the smallest
 value that is greater than the removal sequence number of session.

 The argument aSessionId is an instance of SmallInteger that has a value that
 is one greater than the session number (which is returned by the
 System | session method)."

<primitive: 638>
aSessionId _validateClass: SmallInteger .
^self _primitiveFailed: #_findFirstValidQueueElementForSession: .
%

category: 'Initializing'
method: RcQueue
_initialize

"Creates and initializes the components of the RcQueue.  The new method
 automatically initializes a new instance of RcQueue, so there is rarely any
 need for a user application to call initialize."
| seg |
seg := self segment.
removalSeqNumbers := RcQueueRemovalSeqNumbers new: super size.
removalSeqNumbers assignToSegment: seg.
System _addToRcReadSet: self includingAllNodes: true.
1 to: super size do: [:i | | newOne |
  removalSeqNumbers at: i put: 0.
  newOne := self class varyingConstraint new.
  newOne assignToSegment: seg.
  self at: i put: newOne.
].
%

category: 'Error Handling'
method: RcQueue
_errorRcQueueEntriesFound

"In attempting to reset the maximum session ID for the queue to a smaller value,
 an entry was found that was added by a session whose session ID is larger than
 the value currently being set.  Remove and save all entries in the queue.  Then
 change the maximum session ID and add the saved entries back into the queue."

^ self _error: #rtErrRcQueueEntriesFound
%

category: 'Private'
method: RcQueue
_resolveRcConflictsWith: conflictObjects

"A logical write-write conflict has occurred on the receiver.  The objects that
 had the actual physical write-write conflicts are in the conflictObjects
 Array.  Returns whether the conflicts could be successfully resolved."

| offset sessionComponent beginX endX sz |
" If no objects experienced physical conflict, then just returns "
conflictObjects isEmpty
  ifTrue: [ ^ true ].

" determine if this transaction is a consumer and a transaction that
committed in the interim was a consumer "

self _performedRemoval
  ifTrue: [ " this session is a consumer "

    " get a copy before the first removal in the transaction occurred "
    beginX := self _removalSeqNumbersAtBeginningOfTransaction.

    " save a copy of the state at the end of the transaction "
    endX := removalSeqNumbers copy.
    " refresh the view to see other transaction's changes "
    removalSeqNumbers _selectiveAbort.

    " compare removal numbers to see if any changed "
    1 to: (removalSeqNumbers size min: beginX size) do: [ :i |
      (beginX at: i) = (removalSeqNumbers at: i)
        ifFalse: [
          " replace removalSeqNumbers to get a consistent state of the queue "
          removalSeqNumbers := endX.
          ^ false
        ]
    ]
  ].

" at this point, we've failed if two consumers conflicted "

" see if just conflicted on removalSeqNumbers "
( (conflictObjects size = 1) _and:
[ (conflictObjects at: 1) == removalSeqNumbers ] )
  ifTrue: [
    " if endX is nil, we did not abort removalSeqNumbers above "
    endX == nil
      ifTrue: [
        " save a copy of the state at the end of the transaction "
        endX := removalSeqNumbers copy.
        " refresh the view to see other transaction's changes "
        removalSeqNumbers _selectiveAbort.
      ].

    " compare the sizes of removalSeqNumbers before and after the abort "

    endX size > removalSeqNumbers size
      ifTrue: [
        " other session consumed and committed,
          this session is a producer who grew the root object "

        sz := removalSeqNumbers size.
        " grow the removalSeqNumbers Array and initialize with zeros "
        removalSeqNumbers size: endX size.
        sz + 1 to: endX size do: [ :i | removalSeqNumbers at: i put: 0 ].

        ^ true
      ].

    endX size < removalSeqNumbers size
      ifTrue: [
        " other session was a producer who grew the root and committed,
          this session consumed "

        " overlay my changes to removalSeqNumbers "
        1 to: endX size do: [ :i | removalSeqNumbers at: i put: (endX at: i) ].

        ^ true
      ]
  ].

" if had a physical conflict on the root "
(conflictObjects includes: self)
    ifTrue: [
        " keep reference to this session's components "
        offset := System session + 1.
        sessionComponent := self _rcAt: offset.

        " abort the root and the removalSeqNumbers "
        super _selectiveAbort.
        removalSeqNumbers _selectiveAbort.

        " check if need to grow the root object "
        offset > self maxSessionId
            ifTrue: [ self changeMaxSessionId: offset ].

        " now re-insert the session element "
        self at: offset put: sessionComponent.

        ^ true
    ].

^ true
%

category: 'Initialization'
method: RcQueue
initialize

"Create subcomponents for all available session IDs.  This can avoid
 initial concurrency conflict when many sessions add an object to the
 RcQueue for the first time."

1 to: super _basicSize do: [ :i |
    self _getSessionComponentFor: i 
]
%

category: 'Private'
method: RcQueue
_logRemoval

"Log the fact that a removal occurred (used in _resolveRcConflicts:)."
| systm |

systm := System.
systm redoLog addConflictObject: removalSeqNumbers for: self.

(systm redoLog redoObjects at: self otherwise: nil) == nil
  ifTrue: [ systm redoLog redoObjects at: self put: removalSeqNumbers copy ]
%

category: 'Private'
method: RcQueue
_performedRemoval

"Returns whether the receiver performed a removal during the current
 transaction."

^ (System redoLog redoObjects at: self otherwise: nil) ~~ nil
%

category: 'Private'
method: RcQueue
_removalSeqNumbersAtBeginningOfTransaction

"Returns a copy of removalSeqNumbers that was stored before the first
 removal occurred."

^ System redoLog redoObjects at: self otherwise: nil
%

category: 'Private'
method: RcQueue
_selectiveAbort

"In addition to aborting the RcQueue, selectively aborts the session components
 and removal sequence numbers.  Returns the receiver."

| sessionComponent |

super _selectiveAbort.
removalSeqNumbers _selectiveAbort.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      sessionComponent _selectiveAbort.
      1 to: sessionComponent size do: [ :j |
        (sessionComponent at: j) _selectiveAbort
      ]
    ].
].
%

category: 'Private'
method: RcQueue
_refreshAfterCommitFailure

"Returns whether the receiver should be selectively aborted when there is a
 failed attempt to commit.  Only special Reduced Objects should answer true."

^ true
%

category: 'Locking Support'
method: RcQueue
_lockableValues

"Returns a kind of object usable as an argument to _lockAll: primitives.
 This method returns all the elements logically contained in the receiver."

^ self _collectElems
%

category: 'Error Handling'
method: RcQueue
_errorIndexOutOfRange: anIndex

"Sends an error message indicating that anIndex was outside legal limits
 for the receiver."

^ self _error: #objErrBadOffsetIncomplete args: #[anIndex - 1]
%

category: 'Private'
method: RcQueue
_validateRcConflictsWith: conflictObjects

" There are four possible cases that we can analyze by looking
the write sets:

1. this transaction writes the removalSeqNum
   another transaction writes the removalSeqNum and the root
In this case we only conflict on removalSeqNum

2. this transaction writes the removalSeqNum
   another transaction writes the removalSeqNum
In this case we only conflict on removalSeqNum

3. this transaction writes the removalSeqNum and the root
   another transaction writes the removalSeqNum and the root
In this case we conflict on both

4. this transaction writes the removalSeqNum and the root
   another transaction writes the removalSeqNum
In this case we only conflict on removalSeqNum

When this transaction writes the removalSeqNum and the root,
we performed an add:, and we can also tell if we performed a removal.

When the other transaction writes the removalSeqNum and the root,
he either performed just an add:, or an add: and a removal:,
and we cannot determine which without doing a selective abort.
"

" If no objects experienced physical conflict, then just returns "
conflictObjects isEmpty
  ifTrue: [ ^ true ].

" if conflicted on both the removalSeqNum and the root (case 3) or
we wrote the root (case 4) "
(conflictObjects size > 1 _or:
[ (System _testIf: self isIn: 9) ])
  ifTrue: [ " see if we also performed a removal "
    ^ self _notPerformedRemoval
  ].

" case 1 and 2 return false (we just cannot tell) "
^ false
%

category: 'private - extensions'
method: RcQueue
_addAll: aCollection 

"Adds an array of objects to the RcQueue, returns the receiver.  
It is much more efficient to add many elements with this method
than calling add: multiple times."

| sessionId queueElement sessionComponent sessCompSize firstValid
  lastSeqNumber sz clss systm |

systm := System .
sessionId := systm session + 1.
sessionComponent := self _getSessionComponentFor: sessionId.
sessCompSize := sessionComponent size.
(sessCompSize > 0)
  ifTrue: [
    queueElement := sessionComponent at: sessCompSize.
    lastSeqNumber := queueElement sequenceNumber.
    systm _addToRcReadSet: queueElement includingAllNodes: false.
  ]
  ifFalse: [lastSeqNumber := removalSeqNumbers at: sessionId].
clss := sessionComponent class speciesOfElements.
aCollection do:[:n| 
  queueElement := clss newWithValue: n sequenceNumber: lastSeqNumber + 1
	               inSegmentOf: self .
  sessionComponent at: (sessionComponent size + 1) put: queueElement.
  lastSeqNumber := lastSeqNumber + 1.
].

systm _addToRcReadSet: sessionComponent includingAllNodes: true.

"clean up entries that have been removed"
firstValid := self _findFirstValidQueueElementForSession: sessionId.
(firstValid > 1) ifTrue: [
  sessionComponent removeFrom: 1 to: firstValid - 1 .
  ].

sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
  ifFalse: [systm rcValueCacheAt: #size put: (sz + aCollection size) for: self ].

^aCollection

%
category: 'private - extensions'
method: RcQueue
_fastAddAll: anArray

"Adds an array of objects to the RcQueue, returns the receiver.  
It is much more efficient to add many elements with this method
than calling add: multiple times.  This method will perform slightly 
better than addAll: when the collection to be added is large."

| sessionId queueElement sessionComponent sessCompSize firstValid
  lastSeqNumber sz clss systm |

anArray _validateClass: Array.
systm := System .
sessionId := systm session + 1.
sessionComponent := self _getSessionComponentFor: sessionId.
sessCompSize := sessionComponent size.
(sessCompSize > 0)
  ifTrue: [
    queueElement := sessionComponent at: sessCompSize.
    lastSeqNumber := queueElement sequenceNumber.
    systm _addToRcReadSet: queueElement includingAllNodes: false.
  ]
  ifFalse: [lastSeqNumber := removalSeqNumbers at: sessionId].
clss := sessionComponent class speciesOfElements.
1 to: anArray size do:[:n| 
  queueElement := clss newWithValue: n sequenceNumber: lastSeqNumber + 1
                       inSegmentOf: self .
  sessionComponent at: (sessionComponent size + 1) put: queueElement.
  lastSeqNumber := lastSeqNumber + 1.
].

systm _addToRcReadSet: sessionComponent includingAllNodes: true.
"clean up entries that have been removed"
firstValid := self _findFirstValidQueueElementForSession: sessionId.
(firstValid > 1) ifTrue: [
   sessionComponent removeFrom: 1 to: firstValid - 1 .
   ].

sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifFalse: [ systm rcValueCacheAt: #size put: (sz + anArray size) for: self ].
^anArray

%
category: 'private - extensions'
method: RcQueue
_numberInvalidForMySession

"Returns the number of entries in the RcQueue which have been removed
but
 not yet reclaimed in for my session.  
The intent is to determine if it is worth invoking cleanup using
cleanupMySession."

| firstValid |

firstValid := self _findFirstValidQueueElementForSession: (System session + 1).
firstValid == nil
  ifTrue:[ ^0 ]
  ifFalse:[ ^firstValid - 1 ]
%

category: 'private - extensions'
method: RcQueue
_removeAllInto: anArray

"Removes all entries from the RcQueue and put them into anArray in the
correct order."

| resultArray validSessions firstValidInSession sz
 minElement minSession firstValid sessionComponent
 anElement systm |


" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

anArray _validateClass: Array.

sz := self size.
anArray size: 0.
resultArray := anArray size: sz.
sz == 0
  ifTrue: [ ^ resultArray ].

validSessions := Array new.
firstValidInSession := Array new.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         firstValidInSession add: firstValid.
         removalSeqNumbers at: i put:
            ((sessionComponent at: (sessionComponent size)) sequenceNumber)
         ]
    ]
  ].
1 to: resultArray size do: [:j |
  minSession := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
     (anElement isOlderThan: minElement) ifTrue: [
           minElement := anElement.
           minSession := i.
         ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      resultArray at: j put: minElement value.
      minElement value: nil.
      firstValidInSession at: minSession put:
                      ((firstValidInSession at: minSession) + 1).
    ]
  ].

systm := System .
sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifFalse: [ systm rcValueCacheAt: #size put: 0 for: self ].

^resultArray

%
category: 'private - extensions'
method: RcQueue
_removeCount: maxToRemove into: anArray

"Removes entries from the RcQueue, and returns the argument, anArray,
 which contains the  minimum of maxToRemove or the queue size entries, 
 in order.  It is more efficient to remove multiple entries than to send the
 message remove repeatedly."

| resultArray validSessions validSessionIds firstValidInSession sz
  theSize minElement minSession firstValid sessionComponent
  anElement systm |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

anArray _validateClass: Array.

anArray size: 0.
theSize := self size min: (maxToRemove max: 0).
resultArray := anArray size: theSize.

validSessions := Array new.
validSessionIds := Array new.
firstValidInSession := Array new.

1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         validSessionIds add: i.
         firstValidInSession add: firstValid.
         ]
    ]
  ].
1 to: resultArray size do: [:j |
  minSession := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
      (anElement isOlderThan: minElement)
          ifTrue: [
             minElement := anElement.
             minSession := i.
           ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      resultArray at: j put: minElement value.
      minElement value: nil.
      removalSeqNumbers at: (validSessionIds at: minSession)
                        put: (minElement sequenceNumber).
      firstValidInSession at: minSession
                        put: ((firstValidInSession at: minSession) + 1).
    ]
  ].

systm := System .
sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
  ifFalse: [
    systm rcValueCacheAt: #size put: (sz - resultArray size) for: self
  ].

^resultArray

%
category: 'private - extensions'
method: RcQueue
_statistics
"Return a SymbolKeyValueDictionary containing statistics
on the RcQueue as follows:
	#largestSessionComponent - the largest session component
	#largestSessionComponentSize - size of the largest session component
	#largestSessionComponentPhysicalSize - physical size of the largest session component, in bytes
	#totalNumberInvalid - total number of invalid entries, including those belonging to active sessions
	#numberInvalid - number of invalid entries which belong to inactive sessions.
	#numberOfSessions - number of concurrent sessions for which the queue is configured 
	#queueSize - number of entries in the queue"

	| largest largestSize result invalid firstValid |
	largest := nil.
	largestSize := 0.
	invalid := 0.
        System _addToRcReadSet: self includingAllNodes: true.
	1 to: super size do: [:i | |s e|
  		e := self _rcAt: i.
		s := e size.
		s > largestSize
			ifTrue:[largest := e.
					largestSize := s.
			].
		firstValid := self _findFirstValidQueueElementForSession: i.
		(firstValid > 1) ifTrue: [
  			invalid :=  invalid + firstValid - 1 .
  		 ].

	].
	largest == nil
		ifTrue:[^nil].
	result := SymbolKeyValueDictionary new.
	^result add: (#largestSessionComponent -> largest);
		   add: (#largestSessionComponentSize -> largestSize);
		   add: (#largestSessionComponentPhysicalSize -> largest physicalSize);
		   add: (#totalNumberInvalid -> invalid);
		   add: (#numberInvalid -> self numberInvalid);
		   add: (#numberOfSessions -> self maxSessionId);
		   add: (#queueSize -> self size); yourself.
	
%
category: 'private - extensions'
method: RcQueue
_totalNumberInvalid

"Returns the total number of entries in the RcQueue which have been
removed but not yet reclaimed in all session components.  Unlike the numberInvalid
method, this method includes invalid entries for both active and inactive
sessions.  The intent is to determine if it is worth invoking cleanup."

| sz count firstValid |

sz := super size.
count := 0.
1 to: sz do: [:i |
    firstValid := self _findFirstValidQueueElementForSession: i.
    firstValid ~~ nil
      ifTrue: [ count := count + (firstValid - 1) ]
  
  ].
^count
%

category: 'private - extensions'
method: RcQueue
_unsafeCleanupQueue

"Removes all obsolete entries.  Unlike the cleanupQueue method, this method
removes all obsolete entries, including those belonging to active
sessions.  This method will cause concurrency conflicts with the consumer and any
feeders  and should only be used while no sessions are adding to or removing from
the queue."

| maxSessionId  firstValid
  sessionComponent sessionComponentSize |

maxSessionId := self maxSessionId.

1 to: maxSessionId do: [:i |
    sessionComponent := self at: i.
    sessionComponent ~~ nil
      ifTrue: [
        firstValid := self _findFirstValidQueueElementForSession: i.
        sessionComponentSize := sessionComponent size.
        removalSeqNumbers at: i put: 0.
        (firstValid <= sessionComponentSize)
           ifTrue: [
             (firstValid > 1) ifTrue:[
               sessionComponent removeFrom: 1 to: firstValid - 1 .
               ].
             ]
            ifFalse: [ sessionComponent size: 0]
      ]
   
  ].
^self

%
category: 'private - extensions'
method: RcQueue
_unsafeResetQueue
	"Reinitialize the internal RcQueue structures.
	WARNING: this method removes all elements from the queue!
	USE AT YOUR OWN PERIL!"

	self _initialize.
	self initialize.
	^self.
%

category: 'Updating'
method: RcQueue
changeToSegment: segment

"Assigns the receiver and subcomponents to the given segment."

| sessionComponent |
self assignToSegment: segment.
removalSeqNumbers assignToSegment: segment.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      sessionComponent assignToSegment: segment.
      1 to: sessionComponent size do: [ :j |
        (sessionComponent at: j) assignToSegment: segment
      ]
    ]
].
%

category: 'Removing'
method: RcQueue
removeIntoArray: anArray

"Removes the leading element from the receiver and stores the element in 
 anArray at index 1.  A SmallInteger representing the session ID of the 
 session that added the element to the queue is also stored in anArray
 at index 2.

 If the receiver is empty, returns nil."

| minElement minSession firstValid sessionComponent anElement val sz |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

minSession := 0.
minElement := nil.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
        anElement := sessionComponent at: firstValid.
       (anElement isOlderThan: minElement)
          ifTrue: [
             minElement := anElement.
             minSession := i.
           ].
        ]
    ]
  ].
( minSession > 0 _and: [ minElement ~~ nil ] )
   ifTrue: [ | systm |
     removalSeqNumbers at: minSession put: (minElement sequenceNumber).
     val := minElement value.
     minElement value: nil.
     anArray at: 1 put: val.
     anArray at: 2 put: (minSession - 1).
     systm := System .
     sz := systm rcValueCacheAt: #size for: self otherwise: nil.
     sz == nil
         ifFalse: [ systm rcValueCacheAt: #size put: (sz - 1) for: self ].

     ^anArray .
   ]
   ifFalse: [^nil]
%

category: 'Removing'
method: RcQueue
removeIntoArray

"Removes the leading element from the receiver and stores the element in 
 a new Array at index 1.  A SmallInteger representing the session ID of the 
 session that added the element to the queue is also stored in the new Array
 at index 2.

 If the receiver is empty, returns nil."

^ self removeIntoArray: (Array new: 2) .
%

category: 'Removing'
method: RcQueue
removeObject: anObject addedBySessionId: sessionId

"Removes the given object from the receiver assuming it is the oldest
 object added by the given session Id.  Returns the object or nil if 
 the object was not found or the receiver is empty."

| firstValid sessionComponent anElement val sessId sz systm |

 " adjust sessionId, check range"
 sessId := sessionId + 1.
 ( (sessId < 1 ) _or: [ sessId > self _basicSize ] ) 
   ifTrue: [ ^ nil ].

 " fetch sessionComponent "
 sessionComponent := self at: sessId.
 ( sessionComponent == nil ) 
   ifTrue:[ ^ nil ].

 " find appropriate element, and from that 
   the value, with appropriate range checks "
 firstValid := self _findFirstValidQueueElementForSession: sessId.
 ( firstValid > sessionComponent size ) 
   ifTrue: [ ^ nil ].
 anElement := sessionComponent at: firstValid.
 ( anElement == nil ) 
   ifTrue:[ ^ nil ].
 val := anElement value.
 ( val == anObject ) 
   ifFalse: [ ^ nil ].

" log the fact that a removal occurred (used in _resolveRcConflicts:) " 
 self _logRemoval.

 " handle details of removing element from rcqueue "
 removalSeqNumbers at: sessId put: anElement sequenceNumber.
 anElement value: nil.
 systm := System .
 sz := systm rcValueCacheAt: #size for: self otherwise: nil.
 ( sz == nil ) 
   ifFalse: [ systm rcValueCacheAt: #size put: (sz - 1) for: self ].

 ^ val
%

category: 'Copying'
method: RcQueue
copy

"Copies the session components and sesssion entries and answers
 a copy of the receiver."

| result |

result := super copy.
result removalSeqNumbers: removalSeqNumbers copy.
1 to: self _basicSize do:[:n| | eachSessionComp |
	eachSessionComp := self _at: n.
	result _at: n put: eachSessionComp copy.
].
^ result
%

category: 'Updating'
method: RcQueue
removalSeqNumbers: newValue

"Modify the value of the instance variable 'removalSeqNumbers'."
removalSeqNumbers := newValue
%

category: 'Accessing'
method: RcQueue
removalSeqNumbers

"Return the value of the instance variable 'removalSeqNumbers'."
^removalSeqNumbers
%

category: 'Removing'
method: RcQueue
removeCountAsPairs: maxToRemove intoArray: anArray

"Removes entries from the RcQueue, and returns an Array that contains 2 elements
 per entry removed:  the object removed and the session ID of the session that
 added the element."
  

| validSessions validSessionIds firstValidInSession sz numRemoved
  minElement minSessionIdx firstValid sessionComponent anElement systm |

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

anArray size: (2 * ((self size) min: (maxToRemove max: 0))).
validSessions := Array new.
validSessionIds := Array new.
firstValidInSession := Array new.
numRemoved := 0.

1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         validSessionIds add: i.
         firstValidInSession add: firstValid.
         ]
    ]
  ].
1 to: anArray size by: 2 do: [:j |
  minSessionIdx := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
      (anElement isOlderThan: minElement) ifTrue: [
         minElement := anElement.
         minSessionIdx := i.
         ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      | minSessionId |
      minSessionId := validSessionIds at: minSessionIdx.
      anArray at: j put: minElement value.
      anArray at: (j + 1) put: minSessionId - 1.
      numRemoved := numRemoved + 1.
      minElement value: nil.
      removalSeqNumbers at: minSessionId
                        put: (minElement sequenceNumber).
      firstValidInSession at: minSessionIdx
                        put: ((firstValidInSession at: minSessionIdx) + 1).
    ]
  ].
systm := System .
sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
  ifFalse: [
    systm rcValueCacheAt: #size put: (sz - numRemoved) for: self
  ].

^anArray
%

category: 'Removing'
method: RcQueue
removeAllAsPairsInto: anArray

"Removes all entries from the RcQueue, and returns anArray which contains
 entries in pairs.  The odd numbered entries are the objects removed from the queue,
 in order.  The even numbered entries are the session IDs of the sessions that added
 the preceeding entry."

| validSessions firstValidInSession sz minElement minSessionIdx minSession
  firstValid sessionComponent anElement validSessionIds systm |

sz := self size.
anArray size: (sz * 2).
sz == 0
  ifTrue: [ ^ anArray ].

" log the fact that a removal occurred (used in _resolveRcConflicts:) "
self _logRemoval.

validSessions := Array new.
validSessionIds := Array new.
firstValidInSession := Array new.
1 to: self maxSessionId do: [:i |
  sessionComponent := self at: i.
  sessionComponent ~~ nil
    ifTrue: [
      firstValid := self _findFirstValidQueueElementForSession: i.
      (firstValid <= sessionComponent size) ifTrue: [
         validSessions add: sessionComponent.
         firstValidInSession add: firstValid.
         validSessionIds add: i.
         removalSeqNumbers at: i put:
            ((sessionComponent at: (sessionComponent size)) sequenceNumber)
         ]
    ]
  ].
1 to: anArray size by: 2 do: [:j |
  minSessionIdx := 0.
  minElement := nil.
  1 to: validSessions size do: [:i |
    sessionComponent := validSessions at: i.
    firstValid := firstValidInSession at: i.
    (firstValid <= sessionComponent size) ifTrue: [
      anElement := sessionComponent at: firstValid.
    (anElement isOlderThan: minElement) ifTrue: [
           minElement := anElement.
           minSessionIdx := i.
           minSession := validSessionIds at: i.
         ].
      ]
    ].
  minElement ~~ nil
    ifTrue: [
      anArray at: j put: minElement value.
      anArray at: (j + 1) put: (minSession - 1).
      minElement value: nil.
      firstValidInSession at: minSessionIdx put:
                      ((firstValidInSession at: minSessionIdx) + 1).
    ]
  ].

systm := System .
sz := systm rcValueCacheAt: #size for: self otherwise: nil.
sz == nil
    ifFalse: [ systm rcValueCacheAt: #size put: 0 for: self ].

^anArray
%

category: 'Private'
method: RcQueue
_notPerformedRemoval

"Returns whether the receiver did not perform a removal during the current
 transaction."

^ (System redoLog redoObjects at: self otherwise: nil) == nil
%
