! ========================================================================
! Copyright (C) VMware, Inc. 1991-2011.  All Rights Reserved
!
! $Id: GsWork.gs,v 1.15 2008-01-09 22:50:07 stever Exp $
!
! ========================================================================
!
doit
Object subclass: 'GsWorkUnit'
  instVarNames: #( executeBlock withValue executed result
                    error stackReport )
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
doit
Object subclass: 'GsWorkAbstract'
  instVarNames: #( name)
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
doit
GsWorkAbstract subclass: 'GsWorkQueue'
  instVarNames: #( workQ responseQ)
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
doit
GsWorkAbstract subclass: 'GsWorkCommon'
  instVarNames: #( queue)
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
doit
GsWorkCommon subclass: 'GsWorkServer'
  instVarNames: #( continueService)
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
doit
| class |
class := GsWorkCommon subclass: 'GsWorkMaster'
  instVarNames: #(outStandingWorkUnits buffer )
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: true.
class makeInstancesNonPersistent.
class immediateInvariant.
^class
%
doit
Error subclass: 'GsWorkError'
  instVarNames: #()
  classVars: #()
  classInstVars: #()
  poolDictionaries: #[]
  inDictionary: UserGlobals
  constraints: #[  ]
  instancesInvariant: false
  isModifiable: false

%
!========================================================================
! class GsWorkUnit
!
! Remove existing behavior from GsWorkUnit
doit
GsWorkUnit removeAllMethods.
GsWorkUnit class removeAllMethods.
%
! ------------------- Class methods for GsWorkUnit
category: 'Instance Creation'
classmethod: GsWorkUnit
execute: aBlock

  | new |
new := self new.
new execute: aBlock.
^new
%
category: 'Instance Creation'
classmethod: GsWorkUnit
execute: aBlock with: aValue

  | new |
new := self new.
new execute: aBlock.
new withValue: aValue.
^new
%
! ------------------- Instance methods for GsWorkUnit
category: 'Accessing'
method: GsWorkUnit
error
"Boolean. true if an error occurred while evaluating the block."

^error
%

category: 'Services'
method: GsWorkUnit
executeWithError
	"don't catch errors"
  | theBlock |
((executed == false) _and: [executeBlock ~~ nil])
  ifTrue: [
    theBlock := executeBlock.
    withValue ~~ nil
       ifTrue: [
         theBlock := [ executeBlock value: withValue ].
       ].
    result := theBlock value.
    executed := true.
  ].
%

category: 'Services'
method: GsWorkUnit
execute

  | theBlock |
((executed == false) _and: [executeBlock ~~ nil])
  ifTrue: [
    theBlock := executeBlock.
    withValue ~~ nil
       ifTrue: [
         theBlock := [ executeBlock value: withValue ].
       ].
    result := theBlock 
                  on: Error 
                  do: [:ex | 
                    | args |
                    error := true.
                    stackReport := GsProcess stackReportToLevel: 2000.
                    "An error may be expected ... or may not"
		    GsFile gciLogServer:'---- Execute Error: ' , ex gsNumber printString .
		    GsFile gciLogServer:'----  Session: ' , System session printString .
		    GsFile gciLogServer:'----  Desc: ' , ex description .
                    args := ex gsArguments.
                    1 to: args size do: [:i |
		      GsFile gciLogServer:'----  Args[', i printString, ']: ' , (args at: i) printString.
                      ].
		    GsFile gciLogServer:'---- Stack: '.
		    GsFile gciLogServer:'----        ' , stackReport .
                    ex.
                  ].
    executed := true.
  ].
%
category: 'Accessing'
method: GsWorkUnit
withValue: aValue

withValue := aValue
%
category: 'Initialization'
method: GsWorkUnit
execute: aBlock

self executeBlock: aBlock.
%
category: 'Accessing'
method: GsWorkUnit
executeBlock: aBlock

executeBlock := aBlock.
error := executed := false.
result := nil.
withValue := nil.
%
category: 'Accessing'
method: GsWorkUnit
executed

^executed
%
category: 'Accessing'
method: GsWorkUnit
result

^result
%
category: 'Accessing'
method: GsWorkUnit
stackReport

^stackReport
%
!========================================================================
! class GsWorkAbstract
!
! Remove existing behavior from GsWorkAbstract
doit
GsWorkAbstract removeAllMethods.
GsWorkAbstract class removeAllMethods.
%
! ------------------- Class methods for GsWorkAbstract
! ------------------- Instance methods for GsWorkAbstract
category: 'Private'
method: GsWorkAbstract
abort

System abortTransaction
%
category: 'Private'
method: GsWorkAbstract
commit

System commitTransaction ifFalse: [self signalError: 'Commit failure'].
^true
%
category: 'Accessing'
method: GsWorkAbstract
name

^name
%
category: 'Accessing'
method: GsWorkAbstract
name: aString

name := aString
%
category: 'Testing'
method: GsWorkAbstract
queueExists: qName

  self abort.
  self userGlobals at: qName asSymbol ifAbsent: [ ^false ].
  ^true
%
category: 'Private'
method: GsWorkAbstract
signalError: errorString

  | ex |
ex := GsWorkError new messageText: errorString; yourself.
^ex signal.
%
category: 'Private'
method: GsWorkAbstract
userGlobals

 ^GsSession currentSession objectNamed: #UserGlobals
%
!========================================================================
! class GsWorkQueue
!
! Remove existing behavior from GsWorkQueue
doit
GsWorkQueue removeAllMethods.
GsWorkQueue class removeAllMethods.
%
! ------------------- Class methods for GsWorkQueue
category: 'Instance Creation'
classmethod: GsWorkQueue
create: qName

| new |
new := self new.
new create: qName.
^new
%
! ------------------- Instance methods for GsWorkQueue
category: 'Services'
method: GsWorkQueue
addToResponseQ: aGsWorkUnit for: client

  self abort.
  responseQ add: aGsWorkUnit.
  self commit
%
category: 'Services'
method: GsWorkQueue
addAllToResponseQ: aCollection for: client

  self abort.
  responseQ addAll: aCollection.
  self commit
%
category: 'Services'
method: GsWorkQueue
addToWorkQ: aGsWorkUnit for: client

  [| response |
    response := self _add: aGsWorkUnit to: workQ.
    client shouldRetryAdd: response] whileTrue: [
      System _sleepMs: 10.
  ].
%
category: 'Services'
method: GsWorkQueue
addAllToWorkQ: aCollection for: client

  [| response |
    response := self _addAll: aCollection to: workQ.
    client shouldRetryAdd: response] whileTrue: [
      System _sleepMs: 10.
  ].
%
category: 'Private'
method: GsWorkQueue
workQ

^workQ
%
category: 'Private'
method: GsWorkQueue
responseQ

^responseQ
%
category: 'Private'
method: GsWorkQueue
commitAndReleaseLocks

System commitAndReleaseLocks ifFalse: [self signalError: 'CommitAndReleaseLocks failure'].
^true
%
category: 'Initialization'
method: GsWorkQueue
create: qName

(self queueExists: qName)
  ifTrue: [
    ^self signalError: 'The GsWorkQueue: ', qName, ' already exists.'.
  ].
self userGlobals at: qName asSymbol put: self.
workQ := OrderedCollection new.
responseQ := RcQueue new:100.
self name: qName.
self commit
%
category: 'Services'
method: GsWorkQueue
getAllResponsesFor: client limit: limit

  | responses |
self abort.
responses := responseQ removeCount: limit.
self commit.
^responses
%
category: 'Services'
method: GsWorkQueue
getResponseFor: client

  | response |
self abort.
response := responseQ remove.
self commit.
^response
%
category: 'Services'
method: GsWorkQueue
getWorkFor: client limit: limit

 | response |
 response := self _removeFrom: workQ  limit: limit.
 [client shouldRetryRemove: response] whileTrue: [
   System _sleepMs: 10.
   response := self _removeFrom: workQ limit: limit.
 ].
 ^response
%
category: 'Services'
method: GsWorkQueue
getWorkFor: client

 | response |
 response := self _removeFrom: workQ.
 [client shouldRetryRemove: response] whileTrue: [
   System _sleepMs: 10.
   response := self _removeFrom: workQ.
 ].
 ^response
%
category: 'Testing'
method: GsWorkQueue
isValid

^(self userGlobals at: self name asSymbol) ~~ nil
%
category: 'Services'
method: GsWorkQueue
teardown

  self userGlobals removeKey: self name asSymbol ifAbsent: [].
  workQ := nil.
  responseQ := nil.
  self name: nil.
  self commit
%
category: 'Services'
method: GsWorkQueue
_add: aGsWorkUnit to: queue

self abort.
System writeLock: queue 
  ifDenied: [ ^#retry ] 
  ifChanged: [
    "Got the lock, but the queue is dirty, so abort and continue" 
    self abort 
  ].
  [
    queue addLast: aGsWorkUnit.
  ] ensure: [
    self commitAndReleaseLocks 
      ifFalse: [ ^self signalError: 'Commit failed after adding to queue. ' ].
  ].
%
category: 'Services'
method: GsWorkQueue
_addAll: aCollection to: queue

self abort.
System writeLock: queue 
  ifDenied: [ ^#retry ] 
  ifChanged: [
    "Got the lock, but the queue is dirty, so abort and continue" 
    self abort 
  ].
  [
    aCollection do: [:aGsWorkUnit |
      queue addLast: aGsWorkUnit.
    ].
  ] ensure: [
    self commitAndReleaseLocks 
      ifFalse: [ ^self signalError: 'Commit failed after adding to queue. ' ].
  ].
%
category: 'Services'
method: GsWorkQueue
_removeFrom: queue limit: limit

self abort.
queue size == 0
  ifTrue: [ ^nil ].
System writeLock: queue 
  ifDenied: [ ^#retry ] 
  ifChanged: [
    "Got the lock, but the queue is dirty, so abort and continue" 
    self abort 
  ].
[
  | ar |
  ar := Array new.
  limit timesRepeat: [
    queue size == 0 ifTrue: [ ^ar ].
    ar add: queue removeFirst.
  ].
  ^ ar
] ensure: [
    self commitAndReleaseLocks 
      ifFalse: [ ^self signalError: 'Commit failed after removing from queue. ' ].
].
%
category: 'Services'
method: GsWorkQueue
_removeFrom: queue

self abort.
queue size == 0
  ifTrue: [ ^nil ].
System writeLock: queue 
  ifDenied: [ ^#retry ] 
  ifChanged: [
    "Got the lock, but the queue is dirty, so abort and continue" 
    self abort 
  ].
[
    queue size == 0 ifTrue: [ ^nil ].
    ^queue removeFirst.
] ensure: [
    self commitAndReleaseLocks 
      ifFalse: [ ^self signalError: 'Commit failed after removing from queue. ' ].
].
%
category: 'Testing'
method: GsWorkQueue
isWorkQEmpty

 ^workQ isEmpty
%
category: 'Testing'
method: GsWorkQueue
isResponseQEmpty

 ^responseQ isEmpty
%
category: 'Testing'
method: GsWorkQueue
isEmpty

 ^(self isResponseQEmpty _and: [self isWorkQEmpty])
%
!========================================================================
! class GsWorkCommon
!
! Remove existing behavior from GsWorkCommon
doit
GsWorkCommon removeAllMethods.
GsWorkCommon class removeAllMethods.
%
! ------------------- Class methods for GsWorkCommon
category: 'Instance Creation'
classmethod: GsWorkCommon
attach: qName

| new |
new := self new.
new attach: qName.
^new
%
! ------------------- Instance methods for GsWorkCommon
category: 'Initialization'
method: GsWorkCommon
attach: qName

(self queueExists: qName)
  ifFalse: [
    ^self signalError: 'The GsWorkQueue: ', qName, ' does not exists.'.
  ].
  queue := self userGlobals at: qName asSymbol.
  self name: qName
%
category: 'Accessing'
method: GsWorkCommon
queue

  ^queue
%
category: 'Testing'
method: GsWorkCommon
shouldRetry: response

^response == #retry
%
category: 'Testing'
method: GsWorkCommon
shouldRetryAdd: response

^self shouldRetry: response
%
category: 'Testing'
method: GsWorkCommon
shouldRetryRemove: response

^self shouldRetry: response
%
!========================================================================
! class GsWorkServer
!
! Remove existing behavior from GsWorkServer
doit
GsWorkServer removeAllMethods.
GsWorkServer class removeAllMethods.
%
! ------------------- Class methods for GsWorkServer
! ------------------- Instance methods for GsWorkServer
category: 'Initialization'
method: GsWorkServer
attach: qName

continueService := true.
super attach: qName.
%
category: 'Accessing'
method: GsWorkServer
continueService

^continueService
%
category: 'Accessing'
method: GsWorkServer
continueService: aBoolean

continueService := aBoolean
%
category: 'Services'
method: GsWorkServer
finish: aGsWorkUnit

self queue addToResponseQ: aGsWorkUnit for: self
%
category: 'Services'
method: GsWorkServer
finishAll: aCollection

[ self queue addAllToResponseQ: aCollection for: self ]
  on: Error
  do: [:ex |
        | args stackReport |
        stackReport := GsProcess stackReportToLevel: 2000.
        GsFile gciLogServer:'---- finishAll Error: ' , ex gsNumber printString .
        GsFile gciLogServer:'----  Session: ' , System session printString .
        GsFile gciLogServer:'----        ' , ex description .
        args := ex gsArguments.
        1 to: args size do: [:i |
          GsFile gciLogServer:'----  Args[', i printString, ']: ' , (args at: i) printString.
          ].
        GsFile gciLogServer:'---- Stack: '.
        GsFile gciLogServer:'----        ' , stackReport .
        ex pass.
  ].
%
category: 'Services'
method: GsWorkServer
getWork

^self queue getWorkFor: self
%
category: 'Services'
method: GsWorkServer
serviceLoop

self serviceLoop: [
  self queue isValid ifFalse: [continueService := false].
  self continueService].
%
category: 'Services'
method: GsWorkServer
serviceLoopWithLimit: limit

self serviceLoop: 
  [
    self queue isValid ifFalse: [continueService := false].
    self continueService
  ]
  limit: limit.
%
category: 'Services'
method: GsWorkServer
serviceLoop: whileTrueBlock

GsFile gciLogServer: 'Starting service loop (workQ: ', self queue workQ asOop printString, ') (responseQ: ', self queue responseQ asOop printString, ')'.
whileTrueBlock whileTrue: [
  | work |
  work := self getWork.
  (work == nil) 
    ifTrue: [
      System _sleepMs: 1000.
    ]
    ifFalse: [
      GsFile gciLogServer: 'Begin execute (', System session printString, '): ', self asOop printString, ' --> ', work asOop printString.
      work execute.
      GsFile gciLogServer: 'Done execute'.
      self commit.
      GsFile gciLogServer: 'Done commit'.
      self finish: work.
      GsFile gciLogServer: 'Done finish'.
    ].
  ].
GsFile gciLogServer: 'Ending service loop (', System session printString, ')'.
%
category: 'Services'
method: GsWorkServer
serviceLoop: whileTrueBlock limit: limit

GsFile gciLogServer: 'Starting service loop (workQ: ', self queue workQ asOop printString, ') (responseQ: ', self queue responseQ asOop printString, ')'.
whileTrueBlock whileTrue: [
  | workArray |
  workArray := self queue getWorkFor: self limit: limit.
  (workArray == nil) 
    ifTrue: [
      System _sleepMs: 1000.
    ]
    ifFalse: [
      GsFile gciLogServer: 'Begin execute (', System session printString, ') limit: ', limit printString, ' oop: ', self asOop printString.
      workArray do: [:work |
        GsFile gciLogServer: '  Execute: ', work asOop printString.
        work execute.
        GsFile gciLogServer: 'Done execute'.
        self commit.
        GsFile gciLogServer: 'Done commit'.
      ].
      self finishAll: workArray.
      GsFile gciLogServer: 'Done finishAll'.
    ].
  ].
GsFile gciLogServer: 'Ending service loop (', System session printString, ')'.
%
!========================================================================
! class GsWorkMaster
!
! Remove existing behavior from GsWorkMaster
doit
GsWorkMaster removeAllMethods.
GsWorkMaster class removeAllMethods.
%
! ------------------- Class methods for GsWorkMaster
! ------------------- Instance methods for GsWorkMaster
category: 'Initialization'
method: GsWorkMaster
attach: qName

(self queueExists: qName)
  ifFalse: [
    GsWorkQueue create: qName.
  ].
super attach: qName.
outStandingWorkUnits := 0.
buffer := OrderedCollection new.
%
category: 'Services'
method: GsWorkMaster
getAllResponses: limit

  | ans |
ans := self queue getAllResponsesFor: self limit: limit.
outStandingWorkUnits := outStandingWorkUnits - ans size.
^ans
%
category: 'Services'
method: GsWorkMaster
getResponse

  | ans |
ans := self queue getResponseFor: self.
ans ~~ nil ifTrue: [ outStandingWorkUnits := outStandingWorkUnits - 1 ].
^ans
%
category: 'Services'
method: GsWorkMaster
flushBuffer

self queue addAllToWorkQ: buffer for: self.
outStandingWorkUnits := outStandingWorkUnits + buffer size.
buffer := OrderedCollection new.
%
category: 'Accessing'
method: GsWorkMaster
outStandingWorkUnits

^outStandingWorkUnits
%
category: 'Services'
method: GsWorkMaster
buffer: aGsWorkUnit limit: limit

buffer addLast: aGsWorkUnit.
buffer size >= limit 
  ifTrue: [
    self queue addAllToWorkQ: buffer for: self.
    outStandingWorkUnits := outStandingWorkUnits + buffer size.
    buffer := OrderedCollection new.
  ].
%
category: 'Services'
method: GsWorkMaster
schedule: aGsWorkUnit

self queue addToWorkQ: aGsWorkUnit for: self.
outStandingWorkUnits := outStandingWorkUnits + 1.
%
category: 'Services'
method: GsWorkMaster
teardown

self queue teardown.
queue := nil.
%
category: 'Testing'
method: GsWorkMaster
isEmpty

 ^(self queue isEmpty _and: [self outStandingWorkUnits == 0])
%
!========================================================================
! class GsWorkError
!
! Remove existing behavior from GsWorkError
doit
GsWorkError removeAllMethods.
GsWorkError class removeAllMethods.
%
! ------------------- Class methods for GsWorkError
! ------------------- Instance methods for GsWorkError
