@ -668,37 +668,40 @@ open class Storage {
// /
// /
// / T h e ` a s y n c ` v a r i a n t s d o n ' t n e e d t o w o r r y a b o u t t h i s r e e n t r a n c y i s s u e s o i n s t e a d w e r o u t e w e u s e t h o s e f o r a l l o p e r a t i o n s i n s t e a d
// / T h e ` a s y n c ` v a r i a n t s d o n ' t n e e d t o w o r r y a b o u t t h i s r e e n t r a n c y i s s u e s o i n s t e a d w e r o u t e w e u s e t h o s e f o r a l l o p e r a t i o n s i n s t e a d
// / a n d j u s t b l o c k t h e t h r e a d w h e n w e w a n t t o p e r f o r m a s y n c h r o n o u s o p e r a t i o n
// / a n d j u s t b l o c k t h e t h r e a d w h e n w e w a n t t o p e r f o r m a s y n c h r o n o u s o p e r a t i o n
// /
// / * * N o t e : * * W h e n r u n n i n g a s y n c h r o n o u s o p e r a t i o n t h e r e s u l t w i l l b e r e t u r n e d a n d ` a s y n c C o m p l e t i o n ` w i l l n o t b e c a l l e d , a n d
// / v i c e - v e r s a f o r a n a s y n c h r o n o u s o p e r a t i o n
@ discardableResult private static func performOperation < T > (
@ discardableResult private static func performOperation < T > (
_ info : CallInfo ,
_ info : CallInfo ,
_ dependencies : Dependencies ,
_ dependencies : Dependencies ,
_ operation : @ escaping ( Database ) throws -> T ,
_ operation : @ escaping ( Database ) throws -> T ,
_ : ( ( Result < T , Error > ) -> Void ) ? = nil
_ asyn cC ompletion: ( ( Result < T , Error > ) -> Void ) ? = nil
) -> Result < T , Error > {
) -> Result < T , Error > {
// A s e r i a l q u e u e f o r s y n c h r o n i z i n g c o m p l e t i o n u p d a t e s .
// A s e r i a l q u e u e f o r s y n c h r o n i z i n g c o m p l e t i o n u p d a t e s .
let syncQueue = DispatchQueue ( label : " com.session.performOperation.syncQueue " )
let syncQueue = DispatchQueue ( label : " com.session.performOperation.syncQueue " )
var queryDb : Database ?
var queryDb : Database ?
var did Complete : Bool = false
var did Timeout : Bool = false
var finalResult: Result < T , Error > = . failure ( StorageError . invalidQueryResult )
var operationResult: Result < T , Error > ?
let semaphore : DispatchSemaphore ? = ( info . isAsync ? nil : DispatchSemaphore ( value : 0 ) )
let semaphore : DispatchSemaphore ? = ( info . isAsync ? nil : DispatchSemaphore ( value : 0 ) )
let logErrorIfNeeded : ( Result < T , Error > ) -> ( ) = { result in
let logErrorIfNeeded : ( Result < T , Error > ) -> Result < T , Error > = { result in
switch result {
switch result {
case . success : break
case . success : break
case . failure ( let error ) : StorageState . logIfNeeded ( error , isWrite : info . isWrite )
case . failure ( let error ) : StorageState . logIfNeeded ( error , isWrite : info . isWrite )
}
}
return result
}
}
func completeOperation ( with result : Result < T , Error > ) {
func completeOperation ( with result : Result < T , Error > ) {
syncQueue . sync {
syncQueue . sync {
if didComplete { return }
guard ! didTimeout && operationResult = = nil else { return }
didComplete = true
operationResult = result
finalResult = result
semaphore ? . signal ( )
semaphore ? . signal ( )
// F o r a s y n c o p e r a t i o n s , l o g a n d i n v o k e t h e c o m p l e t i o n c l o s u r e .
// F o r a s y n c o p e r a t i o n s , l o g a n d i n v o k e t h e c o m p l e t i o n c l o s u r e .
if info . isAsync {
if info . isAsync {
logErrorIfNeeded ( result )
asyncCompletion ? ( logErrorIfNeeded ( result ) )
completion ? ( result )
}
}
}
}
}
}
@ -749,63 +752,38 @@ open class Storage {
// /
// /
// / T o t r y t o a v o i d t h i s w e h a v e t h e b e l o w c o d e t o t r y t o r e p l i c a t e t h e b e h a v i o u r o f t h e p r o p e r s e m a p h o r e t i m e o u t w h i l e t h e d e b u g g e r
// / T o t r y t o a v o i d t h i s w e h a v e t h e b e l o w c o d e t o t r y t o r e p l i c a t e t h e b e h a v i o u r o f t h e p r o p e r s e m a p h o r e t i m e o u t w h i l e t h e d e b u g g e r
// / i s a t t a c h e d a s t h i s a p p r o a c h d o e s s e e m t o g e t p a u s e d ( o r a t l e a s t o n l y p e r f o r m a s i n g l e i t e r a t i o n p e r d e b u g g e r s t e p )
// / i s a t t a c h e d a s t h i s a p p r o a c h d o e s s e e m t o g e t p a u s e d ( o r a t l e a s t o n l y p e r f o r m a s i n g l e i t e r a t i o n p e r d e b u g g e r s t e p )
var semaphoreResult : DispatchTimeoutResult ?
if let semaphore : DispatchSemaphore = semaphore {
var semaphoreResult : DispatchTimeoutResult
#if DEBUG
if ! isDebuggerAttached ( ) {
semaphoreResult = semaphore ? . wait ( timeout : . now ( ) + . seconds ( Storage . transactionDeadlockTimeoutSeconds ) )
}
else if ! info . isAsync , let semaphore : DispatchSemaphore = semaphore {
let pollQueue : DispatchQueue = DispatchQueue . global ( qos : . userInitiated )
let standardPollInterval : DispatchTimeInterval = . milliseconds ( 100 )
var iterations : Int = 0
let maxIterations : Int = ( ( Storage . transactionDeadlockTimeoutSeconds * 1000 ) / standardPollInterval . milliseconds )
let pollCompletionSemaphore : DispatchSemaphore = DispatchSemaphore ( value : 0 )
// / S t a g g e r t h e s i z e o f t h e ` p o l l I n t e r v a l s ` t o a v o i d h o l d i n g u p t h e t h r e a d i n c a s e t h e q u e r y r e s o l v e s v e r y q u i c k l y ( t h i s
#if DEBUG
// / m e a n s t h e t i m e o u t w i l l o c c u r ~ 5 0 0 m s e a r l y b u t h e l p s p r e v e n t f a l s e m a i n t h r e a d l a g a p p e a r i n g w h e n d e b u g g i n g t h a t w o u l d n ' t
if isDebuggerAttached ( ) {
// / a f f e c t p r o d u c t i o n )
semaphoreResult = debugWait ( semaphore : semaphore , info : info )
let pollIntervals : [ DispatchTimeInterval ] = [
}
. milliseconds ( 5 ) , . milliseconds ( 5 ) , . milliseconds ( 10 ) , . milliseconds ( 10 ) , . milliseconds ( 10 ) ,
else {
standardPollInterval
semaphoreResult = semaphore . wait ( timeout : . now ( ) + . seconds ( Storage . transactionDeadlockTimeoutSeconds ) )
]
}
#else
semaphoreResult = semaphore ? . wait ( timeout : . now ( ) + . seconds ( Storage . transactionDeadlockTimeoutSeconds ) )
#endif
func pollSemaphore ( ) {
// / I f t h e q u e r y t i m e d o u t t h e n w e s h o u l d i n t e r r u p t t h e q u e r y ( d o n ' t w a n t t h e q u e r y t h r e a d t o r e m a i n b l o c k e d w h e n w e ' v e
iterations += 1
// / a l r e a d y h a n d l e d i t a s a f a i l u r e )
if semaphoreResult = = . timedOut {
guard iterations < maxIterations && semaphore . wait ( timeout : . now ( ) ) != . success else {
syncQueue . sync {
pollCompletionSemaphore . signal ( )
didTimeout = true
return
queryDb ? . interrupt ( )
}
let nextInterval : DispatchTimeInterval = pollIntervals [ min ( iterations , pollIntervals . count - 1 ) ]
pollQueue . asyncAfter ( deadline : . now ( ) + nextInterval ) {
pollSemaphore ( )
}
}
}
}
// / P o l l t h e s e m a p h o r e i n a b a c k g r o u n d q u e u e
// / B e f o r e r e t u r n i n g w e n e e d t o w a i t f o r a n y p e n d i n g u p d a t e s o n ` s y n c Q u e u e ` t o b e c o m p l e t e d t o e n s u r e t h a t o b j e c t s
pollQueue . asyncAfter ( deadline : . now ( ) + pollIntervals [ 0 ] ) { pollSemaphore ( ) }
// / d o n ' t g e t i n c o r r e c t l y r e l e a s e d w h i l e t h e y a r e s t i l l b e i n g u s e d
pollCompletionSemaphore . wait ( ) // W a i t i n d e f i n i t e l y f o r t h e t i m e r s e m a p h o r e
syncQueue . sync { }
semaphoreResult = ( iterations >= 50 ? . timedOut : . success )
}
return logErrorIfNeeded ( operationResult ? ? . failure ( StorageError . transactionDeadlockTimeout ) )
#else
semaphoreResult = semaphore ? . wait ( timeout : . now ( ) + . seconds ( Storage . transactionDeadlockTimeoutSeconds ) )
#endif
// / I f t h e q u e r y t i m e d o u t t h e n w e s h o u l d i n t e r r u p t t h e q u e r y ( d o n ' t w a n t t h e q u e r y t h r e a d t o r e m a i n b l o c k e d w h e n w e ' v e
// / a l r e a d y h a n d l e d i t a s a f a i l u r e ) a n d n e e d t o c a l l ` c o m p l e t e O p e r a t i o n ` a s i t w o u l d n ' t h a v e b e e n c a l l e d w i t h i n t h e
// / d b t r a n s a c t i o n y e t
if semaphoreResult = = . timedOut {
syncQueue . sync { queryDb ? . interrupt ( ) }
completeOperation ( with : . failure ( StorageError . transactionDeadlockTimeout ) )
}
}
// / B e f o r e r e t u r n i n g w e n e e d t o w a i t f o r a n y p e n d i n g u p d a t e s o n ` s y n c Q u e u e ` t o b e c o m p l e t e d t o e n s u r e t h a t o b j e c t s
// / F o r t h e ` a s y n c ` o p e r a t i o n t h e r e t u r n e d v a l u e s h o u l d b e i g n o r e d s o j u s t r e t u r n t h e ` i n v a l i d Q u e r y R e s u l t ` e r r o r
// / d o n ' t g e t i n c o r r e c t l y r e l e a s e d w h i l e t h e y a r e s t i l l b e i n g u s e d
return . failure ( StorageError . invalidQueryResult )
syncQueue . sync { }
return finalResult
}
}
private func performPublisherOperation < T > (
private func performPublisherOperation < T > (
@ -834,6 +812,42 @@ open class Storage {
}
}
}
}
private static func debugWait ( semaphore : DispatchSemaphore , info : CallInfo ) -> DispatchTimeoutResult {
let pollQueue : DispatchQueue = DispatchQueue ( label : " com.session.debugWaitTimer. \( UUID ( ) . uuidString ) " )
let standardPollInterval : DispatchTimeInterval = . milliseconds ( 100 )
var iterations : Int = 0
let maxIterations : Int = ( ( Storage . transactionDeadlockTimeoutSeconds * 1000 ) / standardPollInterval . milliseconds )
let pollCompletionSemaphore : DispatchSemaphore = DispatchSemaphore ( value : 0 )
// / S t a g g e r t h e s i z e o f t h e ` p o l l I n t e r v a l s ` t o a v o i d h o l d i n g u p t h e t h r e a d i n c a s e t h e q u e r y r e s o l v e s v e r y q u i c k l y ( t h i s
// / m e a n s t h e t i m e o u t w i l l o c c u r ~ 5 0 0 m s e a r l y b u t h e l p s p r e v e n t f a l s e m a i n t h r e a d l a g a p p e a r i n g w h e n d e b u g g i n g t h a t w o u l d n ' t
// / a f f e c t p r o d u c t i o n )
let pollIntervals : [ DispatchTimeInterval ] = [
. milliseconds ( 5 ) , . milliseconds ( 5 ) , . milliseconds ( 10 ) , . milliseconds ( 10 ) , . milliseconds ( 10 ) ,
standardPollInterval
]
func pollSemaphore ( ) {
iterations += 1
guard iterations < maxIterations && semaphore . wait ( timeout : . now ( ) ) != . success else {
pollCompletionSemaphore . signal ( )
return
}
let nextInterval : DispatchTimeInterval = pollIntervals [ min ( iterations , pollIntervals . count - 1 ) ]
pollQueue . asyncAfter ( deadline : . now ( ) + nextInterval ) {
pollSemaphore ( )
}
}
// / P o l l t h e s e m a p h o r e i n a b a c k g r o u n d q u e u e
pollQueue . asyncAfter ( deadline : . now ( ) + pollIntervals [ 0 ] ) { pollSemaphore ( ) }
pollCompletionSemaphore . wait ( ) // W a i t i n d e f i n i t e l y f o r t h e t i m e r s e m a p h o r e
return ( iterations >= 50 ? . timedOut : . success )
}
// MARK: - F u n c t i o n s
// MARK: - F u n c t i o n s
@ discardableResult public func write < T > (
@ discardableResult public func write < T > (
Morgan Pretty
2 months ago