diff --git a/ModuleConfig.cfc b/ModuleConfig.cfc index 002b4d78..8e8b06a1 100644 --- a/ModuleConfig.cfc +++ b/ModuleConfig.cfc @@ -8,12 +8,14 @@ component { function configure() { settings = { - "defaultGrammar" : "AutoDiscover@qb", - "defaultQueryOptions" : {}, - "preventDuplicateJoins" : true, - "preventLazyLoading" : false, - "refreshOnSaveFallback" : true, - "lazyLoadingViolationCallback" : ( entity, relationName ) => { + "defaultGrammar" : "AutoDiscover@qb", + "defaultQueryOptions" : {}, + "parallelEagerLoadingMaxThreads" : 4, + "parallelEagerLoadingTimeout" : 60000, + "preventDuplicateJoins" : true, + "preventLazyLoading" : false, + "refreshOnSaveFallback" : true, + "lazyLoadingViolationCallback" : ( entity, relationName ) => { throw( type = "QuickLazyLoadingException", message = "Attempted to lazy load the [#arguments.relationName#] relationship on the entity [#arguments.entity.mappingName()#] but lazy loading is disabled. This is usually caused by the N+1 problem and is a sign that you are missing an eager load." diff --git a/models/BaseEntity.cfc b/models/BaseEntity.cfc index 67460aa8..434918b2 100644 --- a/models/BaseEntity.cfc +++ b/models/BaseEntity.cfc @@ -949,11 +949,11 @@ component accessors="true" { overlay = overlay.previous; } - var attributes = []; + var runtimeAttributes = []; for ( var i = newestFirst.len(); i >= 1; i-- ) { - attributes.append( newestFirst[ i ] ); + runtimeAttributes.append( newestFirst[ i ] ); } - return attributes; + return runtimeAttributes; } private void function registerRuntimeAttribute( required struct attribute ) { diff --git a/models/ParallelEagerLoadingContext.cfc b/models/ParallelEagerLoadingContext.cfc new file mode 100644 index 00000000..c993c6e9 --- /dev/null +++ b/models/ParallelEagerLoadingContext.cfc @@ -0,0 +1,24 @@ +/** + * Tracks parallel eager-loading workers without leaking state between threads. + */ +component singleton { + + function init() { + variables.lifecycleEventsSuppressed = createObject( "java", "java.lang.ThreadLocal" ).init(); + return this; + } + + public void function suppressLifecycleEvents() { + variables.lifecycleEventsSuppressed.set( true ); + } + + public void function restoreLifecycleEvents() { + variables.lifecycleEventsSuppressed.remove(); + } + + public boolean function areLifecycleEventsSuppressed() { + var value = variables.lifecycleEventsSuppressed.get(); + return !isNull( value ) && value; + } + +} diff --git a/models/QuickBuilder.cfc b/models/QuickBuilder.cfc index 23983303..5e64b8cc 100644 --- a/models/QuickBuilder.cfc +++ b/models/QuickBuilder.cfc @@ -62,6 +62,27 @@ component accessors="true" transientCache="false" { */ property name="_lazyLoadingViolationCallback" inject="box:setting:lazyLoadingViolationCallback@quick"; + /** + * The maximum number of eager-loading workers that may run at once. + */ + property + name ="_parallelEagerLoadingMaxThreads" + default="4" + inject ="box:setting:parallelEagerLoadingMaxThreads@quick"; + + /** + * The number of milliseconds to wait for a batch of eager-loading workers. + */ + property + name ="_parallelEagerLoadingTimeout" + default="60000" + inject ="box:setting:parallelEagerLoadingTimeout@quick"; + + /** + * Thread-local lifecycle state shared by every QuickBuilder instance. + */ + property name="_parallelEagerLoadingContext" inject="quick.models.ParallelEagerLoadingContext"; + /** * A map of aliases to entities to use when qualifying aliased columns. */ @@ -93,14 +114,17 @@ component accessors="true" transientCache="false" { this.isQuickBuilder = true; function init() { - variables._eagerLoad = []; - variables._globalScopesApplied = false; - variables._globalScopeExcludeAll = false; - variables._asMemento = false; - variables._asQuery = false; - variables._withAliases = false; - variables._entityTransformers = []; - param variables._preventLazyLoading = false; + variables._eagerLoad = []; + variables._parallelEagerLoading = false; + variables._globalScopesApplied = false; + variables._globalScopeExcludeAll = false; + variables._asMemento = false; + variables._asQuery = false; + variables._withAliases = false; + variables._entityTransformers = []; + param variables._parallelEagerLoadingMaxThreads = 4; + param variables._parallelEagerLoadingTimeout = 60000; + param variables._preventLazyLoading = false; if ( !variables.keyExists( "_lazyLoadingViolationCallback" ) || isNull( variables._lazyLoadingViolationCallback ) ) { variables._lazyLoadingViolationCallback = ( entity, relationName ) => { throw( @@ -713,9 +737,11 @@ component accessors="true" transientCache="false" { * @relationName A single relation name or array of relation * names to eager load. * + * @parallel If true, eager loads top-level relationships concurrently. + * * @return QuickBuilder */ - public any function with( required any relationName ) { + public any function with( required any relationName, boolean parallel = false ) { if ( isSimpleValue( arguments.relationName ) && arguments.relationName == "" ) { return this; } @@ -725,6 +751,7 @@ component accessors="true" transientCache="false" { arrayWrap( arguments.relationName ), true ); + variables._parallelEagerLoading = variables._parallelEagerLoading || arguments.parallel; return this; } @@ -806,17 +833,271 @@ component accessors="true" transientCache="false" { } var eagerLoads = denestEagerLoads( variables._eagerLoad ); - for ( var relationName in eagerLoads ) { - arguments.entities = eagerLoadRelation( - relationName, - eagerLoads[ relationName ], - arguments.entities - ); + if ( variables._parallelEagerLoading && eagerLoads.count() > 1 && supportsParallelEagerLoading() ) { + eagerLoadRelationsInParallel( eagerLoads, arguments.entities ); + } else { + for ( var relationName in eagerLoads ) { + arguments.entities = eagerLoadRelation( + relationName, + eagerLoads[ relationName ], + arguments.entities + ); + } } return arguments.entities; } + /** + * Eager loads independent top-level relationships on separate threads. + */ + private void function eagerLoadRelationsInParallel( required struct eagerLoads, required array entities ) { + var relationNames = arguments.eagerLoads.keyArray(); + var maxWorkers = max( 1, int( variables._parallelEagerLoadingMaxThreads ) ); + var timeout = max( 1, int( variables._parallelEagerLoadingTimeout ) ); + var targetEntities = arguments.entities; + var threadResults = createObject( "java", "java.util.concurrent.ConcurrentHashMap" ).init(); + var entityStates = []; + for ( var entity in arguments.entities ) { + entityStates.append( + structKeyExists( entity, "isQuickEntity" ) + ? { + "isQuickEntity" : true, + "mappingName" : entity.mappingName(), + "attributes" : entity.retrieveAttributesData( withNulls = true ) + } + : { + "isQuickEntity" : false, + "value" : duplicate( entity ) + } + ); + } + + for ( var batchStart = 1; batchStart <= relationNames.len(); batchStart += maxWorkers ) { + var batchThreadNames = []; + var batchThreadRelations = {}; + var batchEnd = min( relationNames.len(), batchStart + maxWorkers - 1 ); + for ( var relationIndex = batchStart; relationIndex <= batchEnd; relationIndex++ ) { + var relationName = relationNames[ relationIndex ]; + var threadName = "quick_eager_#replace( createUUID(), "-", "", "all" )#"; + batchThreadNames.append( threadName ); + batchThreadRelations[ threadName ] = relationName; + cfthread( + action = "run", + name = threadName, + threadName = threadName, + relationName = relationName, + eagerLoadConfig = arguments.eagerLoads[ relationName ], + entityStates = entityStates, + results = threadResults + ) { + variables._parallelEagerLoadingContext.suppressLifecycleEvents(); + try { + var workerEntities = []; + for ( var entityState in entityStates ) { + workerEntities.append( + entityState.isQuickEntity + ? hydrateParallelEntityState( entityState ) + : entityState.value + ); + } + var loadedEntities = eagerLoadRelation( + relationName, + eagerLoadConfig, + workerEntities + ); + var relationshipValues = []; + for ( var loadedEntity in loadedEntities ) { + if ( structKeyExists( loadedEntity, "isQuickEntity" ) ) { + if ( isNull( loadedEntity.retrieveRelationship( relationName ) ) ) { + relationshipValues.append( { "type" : "null" } ); + } else { + relationshipValues.append( + serializeParallelValue( loadedEntity.retrieveRelationship( relationName ) ) + ); + } + } else if ( loadedEntity.keyExists( relationName ) ) { + relationshipValues.append( serializeParallelValue( loadedEntity[ relationName ] ) ); + } else { + relationshipValues.append( { "type" : "null" } ); + } + } + results.put( threadName, relationshipValues ); + } finally { + variables._parallelEagerLoadingContext.restoreLifecycleEvents(); + } + } + } + + cfthread( + action = "join", + name = batchThreadNames.toList(), + timeout = timeout + ); + + var failedThread = ""; + var timedOut = false; + for ( var batchThreadName in batchThreadNames ) { + if ( cfthread[ batchThreadName ].status == "TERMINATED" && failedThread == "" ) { + failedThread = batchThreadName; + } + if ( + cfthread[ batchThreadName ].status != "COMPLETED" && cfthread[ batchThreadName ].status != "TERMINATED" + ) { + timedOut = true; + } + } + if ( failedThread != "" || timedOut ) { + terminateParallelEagerLoadingThreads( batchThreadNames ); + } + if ( failedThread != "" ) { + var threadError = cfthread[ failedThread ].error; + throw( + type = "QuickParallelEagerLoadingException", + message = threadError.keyExists( "message" ) ? threadError.message : "A parallel eager-loading thread failed.", + extendedInfo = serializeJSON( threadError ) + ); + } + if ( timedOut ) { + throw( + type = "QuickParallelEagerLoadingTimeout", + message = "Parallel eager loading did not complete within #timeout# milliseconds." + ); + } + + for ( var completedThreadName in batchThreadNames ) { + var completedRelationName = batchThreadRelations[ completedThreadName ]; + var relationshipValues = threadResults.get( completedThreadName ); + for ( var i = 1; i <= targetEntities.len(); i++ ) { + var completedRelationshipValue = deserializeParallelValue( relationshipValues[ i ] ); + if ( structKeyExists( targetEntities[ i ], "isQuickEntity" ) ) { + if ( isNull( completedRelationshipValue ) ) { + targetEntities[ i ].assignRelationship( completedRelationName ); + } else { + targetEntities[ i ].assignRelationship( completedRelationName, completedRelationshipValue ); + } + targetEntities[ i ].fireRelationshipLoaded( completedRelationName ); + } else if ( !isNull( completedRelationshipValue ) ) { + targetEntities[ i ][ completedRelationName ] = completedRelationshipValue; + } + } + } + } + } + + private any function hydrateParallelEntityState( required struct state ) { + return getEntity() + .newEntity( arguments.state.mappingName ) + .assignAttributesData( arguments.state[ "attributes" ] ) + .assignOriginalAttributes( arguments.state[ "attributes" ] ) + .set_loaded( true ); + } + + private void function terminateParallelEagerLoadingThreads( required array threadNames ) { + for ( var threadName in arguments.threadNames ) { + if ( cfthread[ threadName ].status != "COMPLETED" && cfthread[ threadName ].status != "TERMINATED" ) { + cfthread( action = "terminate", name = threadName ); + } + } + } + + /** + * Converts eager-loaded values to CFC-free state for crossing thread boundaries. + */ + private struct function serializeParallelValue( any value ) { + if ( isNull( arguments.value ) ) { + return { "type" : "null" }; + } + if ( isArray( arguments.value ) ) { + var items = []; + for ( var item in arguments.value ) { + items.append( serializeParallelValue( item ) ); + } + return { "type" : "array", "value" : items }; + } + if ( isStruct( arguments.value ) && structKeyExists( arguments.value, "isQuickEntity" ) ) { + var relationships = {}; + for ( var relationshipName in arguments.value.get_relationshipsLoaded().keyArray() ) { + relationships[ relationshipName ] = serializeParallelValue( + arguments.value.retrieveRelationship( relationshipName ) + ); + } + return { + "type" : "entity", + "mappingName" : arguments.value.mappingName(), + "attributes" : arguments.value.retrieveAttributesData( withNulls = true ), + "relationships" : relationships + }; + } + if ( isStruct( arguments.value ) ) { + var values = {}; + for ( var key in arguments.value ) { + values[ key ] = serializeParallelValue( arguments.value[ key ] ); + } + return { "type" : "struct", "value" : values }; + } + return { + "type" : "value", + "value" : arguments.value + }; + } + + /** + * Reconstructs eager-loaded values exported by a worker thread. + */ + private any function deserializeParallelValue( required struct state ) { + switch ( arguments.state.type ) { + case "null": + return javacast( "null", "" ); + case "array": + var items = []; + for ( var item in arguments.state.value ) { + items.append( deserializeParallelValue( item ) ); + } + return items; + case "entity": + var entity = getEntity().newEntity( arguments.state.mappingName ); + try { + entity.hydrate( arguments.state[ "attributes" ] ); + } catch ( MissingHydrationKey missingKey ) { + entity + .assignAttributesData( arguments.state[ "attributes" ] ) + .assignOriginalAttributes( arguments.state[ "attributes" ] ) + .set_loaded( true ); + } + for ( var relationshipName in arguments.state.relationships ) { + var relationshipValue = deserializeParallelValue( + arguments.state.relationships[ relationshipName ] + ); + if ( isNull( relationshipValue ) ) { + entity.assignRelationship( relationshipName ); + } else { + entity.assignRelationship( relationshipName, relationshipValue ); + } + entity.fireRelationshipLoaded( relationshipName ); + } + return entity; + case "struct": + var values = {}; + for ( var key in arguments.state.value ) { + var value = deserializeParallelValue( arguments.state.value[ key ] ); + if ( !isNull( value ) ) { + values[ key ] = value; + } + } + return values; + default: + return arguments.state.value; + } + } + + /** + * Adobe ColdFusion loses CFC private-method resolution inside cfthread. + */ + private boolean function supportsParallelEagerLoading() { + return !server.keyExists( "coldfusion" ) || !findNoCase( "ColdFusion", server.coldfusion.productName ); + } + private struct function denestEagerLoads( required array eagerLoads ) { // this comes in as an array of items which can be: // 1. dot-delimited strings (e.g., "videos.tags") @@ -977,7 +1258,11 @@ component accessors="true" transientCache="false" { ); var loadedRelationshipName = arguments.relationName; for ( var entity in matchedEntities ) { - if ( isStruct( entity ) && structKeyExists( entity, "isQuickEntity" ) ) { + if ( + !variables._parallelEagerLoadingContext.areLifecycleEventsSuppressed() + && isStruct( entity ) + && structKeyExists( entity, "isQuickEntity" ) + ) { entity.fireRelationshipLoaded( loadedRelationshipName ); } } @@ -1913,8 +2198,8 @@ component accessors="true" transientCache="false" { .assignAttributesData( arguments.data ) .assignOriginalAttributes( arguments.data ) .set_preventLazyLoading( variables._preventLazyLoading ) - .set_lazyLoadingViolationCallback( variables._lazyLoadingViolationCallback ) - .markLoaded(); + .set_lazyLoadingViolationCallback( variables._lazyLoadingViolationCallback ); + markLoadedEntity( childEntity ); if ( hasVirtualData ) { childEntity.set_refreshQuery( arguments.refreshQuery ); } @@ -1925,8 +2210,8 @@ component accessors="true" transientCache="false" { .assignAttributesData( arguments.data ) .assignOriginalAttributes( arguments.data ) .set_preventLazyLoading( variables._preventLazyLoading ) - .set_lazyLoadingViolationCallback( variables._lazyLoadingViolationCallback ) - .markLoaded(); + .set_lazyLoadingViolationCallback( variables._lazyLoadingViolationCallback ); + markLoadedEntity( entity ); if ( hasVirtualData ) { entity.set_refreshQuery( arguments.refreshQuery ); } @@ -1934,6 +2219,14 @@ component accessors="true" transientCache="false" { } } + private void function markLoadedEntity( required any entity ) { + if ( variables._parallelEagerLoadingContext.areLifecycleEventsSuppressed() ) { + arguments.entity.set_loaded( true ); + } else { + arguments.entity.markLoaded(); + } + } + /** * Automatically converts the entities found from a query to mementos. * diff --git a/tests/resources/app/models/ParallelLifecycleUser.cfc b/tests/resources/app/models/ParallelLifecycleUser.cfc new file mode 100644 index 00000000..4c685268 --- /dev/null +++ b/tests/resources/app/models/ParallelLifecycleUser.cfc @@ -0,0 +1,26 @@ +component + table ="users" + extends ="quick.models.BaseEntity" + accessors="true" +{ + + property name="id"; + + function postLoad( eventData ) { + param request.parallelLifecyclePostLoads = []; + request.parallelLifecyclePostLoads.append( this ); + } + + function posts() { + return hasMany( "Post", "user_id" ); + } + + function comments() { + return hasMany( "Comment", "user_id" ); + } + + function postsLoaded( entity ) { + arguments.entity.assignRelationship( "loadedByUser", this ); + } + +} diff --git a/tests/specs/integration/BaseEntity/Relationships/EagerLoadingSpec.cfc b/tests/specs/integration/BaseEntity/Relationships/EagerLoadingSpec.cfc index 4d45318a..4e30a1a2 100644 --- a/tests/specs/integration/BaseEntity/Relationships/EagerLoadingSpec.cfc +++ b/tests/specs/integration/BaseEntity/Relationships/EagerLoadingSpec.cfc @@ -16,6 +16,7 @@ component extends="tests.resources.ModuleIntegrationSpec" { describe( "Eager Loading Spec", function() { beforeEach( function() { variables.queries = []; + structDelete( request, "parallelLifecyclePostLoads" ); } ); it( "can eager load a belongs to relationship", function() { @@ -61,6 +62,177 @@ component extends="tests.resources.ModuleIntegrationSpec" { expect( keys ).toHaveLength( 2 ); } ); + it( "can eager load top-level relationships in parallel", function() { + var callingThread = createObject( "java", "java.lang.Thread" ).currentThread().getName(); + var eagerThreads = {}; + var posts = getInstance( "Post" ) + .with( + [ + { + "author" : function( relationship ) { + eagerThreads.author = createObject( "java", "java.lang.Thread" ) + .currentThread() + .getName(); + } + }, + { + "comments" : function( relationship ) { + eagerThreads.comments = createObject( "java", "java.lang.Thread" ) + .currentThread() + .getName(); + } + } + ], + true + ) + .get(); + + expect( posts[ 1 ].getAuthor() ).toBeInstanceOf( "app.models.User" ); + expect( posts[ 1 ].getComments() ).toBeArray(); + expect( eagerThreads ).toHaveKey( "author" ); + expect( eagerThreads ).toHaveKey( "comments" ); + if ( server.keyExists( "coldfusion" ) && findNoCase( "ColdFusion", server.coldfusion.productName ) ) { + expect( eagerThreads.author ).toBe( callingThread ); + expect( eagerThreads.comments ).toBe( callingThread ); + } else { + expect( eagerThreads.author ).notToBe( callingThread ); + expect( eagerThreads.comments ).notToBe( callingThread ); + expect( eagerThreads.author ).notToBe( eagerThreads.comments ); + } + } ); + + it( "keeps a single eager load on the calling thread", function() { + var callingThread = createObject( "java", "java.lang.Thread" ).currentThread().getName(); + var eagerThread = ""; + getInstance( "Post" ) + .with( + { + "author" : function( relationship ) { + eagerThread = createObject( "java", "java.lang.Thread" ).currentThread().getName(); + } + }, + true + ) + .get(); + + expect( eagerThread ).toBe( callingThread ); + } ); + + it( "delivers lifecycle events once on the returned parallel entities", function() { + var users = getInstance( "ParallelLifecycleUser" ) + .where( "id", 1 ) + .with( [ "posts", "comments" ], true ) + .get(); + + expect( users ).toHaveLength( 1 ); + expect( request.parallelLifecyclePostLoads ).toHaveLength( 1 ); + expect( request.parallelLifecyclePostLoads[ 1 ].isSameAs( users[ 1 ] ) ).toBeTrue(); + for ( var post in users[ 1 ].getPosts() ) { + expect( post.retrieveRelationship( "loadedByUser" ).isSameAs( users[ 1 ] ) ).toBeTrue(); + } + } ); + + it( "preserves nested eager loads in parallel relationship graphs", function() { + var post = getInstance( "Post" ) + .where( "post_pk", 1245 ) + .with( [ "author.country", "comments.author" ], true ) + .firstOrFail(); + + expect( post.getAuthor().isRelationshipLoaded( "country" ) ).toBeTrue(); + for ( var comment in post.getComments() ) { + expect( comment.isRelationshipLoaded( "author" ) ).toBeTrue(); + } + } ); + + it( "preserves pivot relationships in parallel relationship graphs", function() { + var post = getInstance( "Post" ) + .where( "post_pk", 1245 ) + .with( [ "tagsAsSubscriptions", "comments" ], true ) + .firstOrFail(); + + for ( var tag in post.getTagsAsSubscriptions() ) { + expect( tag.isRelationshipLoaded( "subscription" ) ).toBeTrue(); + expect( tag.getSubscription() ).toBeInstanceOf( "quick.models.Relationships.Pivot" ); + } + } ); + + it( "supports parallel eager loading for query results", function() { + var posts = getInstance( "Post" ) + .with( [ "author", "comments" ], true ) + .asQuery() + .get(); + + expect( posts[ 1 ] ).toBeStruct(); + expect( posts[ 1 ].author ).toBeStruct(); + expect( posts[ 1 ].comments ).toBeArray(); + expect( posts[ 3 ].author ).toBeStruct().toBeEmpty(); + } ); + + it( "limits the number of concurrent eager-loading workers", function() { + if ( supportsParallelEagerLoadingForTest() ) { + var activeWorkers = createObject( "java", "java.util.concurrent.atomic.AtomicInteger" ).init(); + var maxWorkers = createObject( "java", "java.util.concurrent.atomic.AtomicInteger" ).init(); + var trackWorker = function( relationship ) { + var active = activeWorkers.incrementAndGet(); + while ( active > maxWorkers.get() && !maxWorkers.compareAndSet( maxWorkers.get(), active ) ) { + } + sleep( 25 ); + activeWorkers.decrementAndGet(); + }; + var builder = getInstance( "Post" ).with( + [ + { "author" : trackWorker }, + { "comments" : trackWorker } + ], + true + ); + builder.set_parallelEagerLoadingMaxThreads( 1 ).get(); + + expect( maxWorkers.get() ).toBe( 1 ); + } + } ); + + it( "propagates parallel eager-loading worker failures", function() { + if ( supportsParallelEagerLoadingForTest() ) { + expect( function() { + getInstance( "Post" ) + .with( + [ + { + "author" : function( relationship ) { + throw( type = "ExpectedParallelFailure", message = "worker failed" ); + } + }, + "comments" + ], + true + ) + .get(); + } ).toThrow( type = "QuickParallelEagerLoadingException" ); + } + } ); + + it( "times out and cancels unfinished parallel eager-loading workers", function() { + if ( supportsParallelEagerLoadingForTest() ) { + var builder = getInstance( "Post" ).with( + [ + { + "author" : function( relationship ) { + sleep( 100 ); + } + }, + "comments" + ], + true + ); + builder.set_parallelEagerLoadingTimeout( 1 ); + + expect( function() { + builder.get(); + } ).toThrow( type = "QuickParallelEagerLoadingTimeout", regex = "1 milliseconds" ); + } + } ); + it( "can eager load a belongs to relationship using a composite key", function() { var compositeChildren = getInstance( "CompositeChild" ).with( "parent" ).get(); expect( compositeChildren ).toBeArray(); @@ -869,4 +1041,8 @@ component extends="tests.resources.ModuleIntegrationSpec" { } ); } + private boolean function supportsParallelEagerLoadingForTest() { + return !server.keyExists( "coldfusion" ) || !findNoCase( "ColdFusion", server.coldfusion.productName ); + } + }