I've recently starting using cfthread on one of my CF 9 servers. I've seen some wildly odd behaviour from the threads and am wondering if this is a known issue.
I am grabbing a set of data which is 1200 rows deep. I loop over that data 12 times, grab 100 rows each time and create a unique thread. If I don't artificially add delay the query distribution between the threads is random and weighted heavily towards the last thread created. Even after adding 5-6 seconds of delay between spawning new threads – the distribution is not level. The ONLY time I can get 100 rows per thread is when I insert a cfdump of the query (adding a LOT of additional processing time).
Im writing the thread names into my table for debug purposes - so I can get an exact count of which thread is processing which numbers.
Here is the distribution after adding 5-6 seconds of delay for a set of 100:
threadtest COUNT(`threadtest`)
MigThread1 21
MigThread2 51
MigThread3 28
-- as you can see, though the data set passed into each thread is distinct, is seems to be handed to the threads in a random way -- as if the thread attributes and scopes were immediately corrupted. Oddly, I always get the number of rows processed that I want - just not distributed between the threads as I expect.
Here is the code:
<!---get the data --->
<cfquery name="getTNs" datasource="xxxx">
SELECT SubKey, TN, LRN, MAC, CLLI, StagingFallout, FinalFallout
FROM MigValidationNew
WHERE JobKey = '#JobKey#' AND StagingFallout <> '1'
</cfquery>
<!--- Build a new dataset so the threads are better controlled --->
<cfset SortedTNs = querynew("IndexCounter, TN, SubKey")>
<cfset RowCounter = '1'>
<cfoutput query="getTNs">
<cfset QueryAddRow(SortedTNs)>
<cfset QuerySetCell(SortedTNs,"IndexCounter",#RowCounter#)>
<cfset QuerySetCell(SortedTNs,"TN",#TN#)>
<cfset QuerySetCell(SortedTNs,"SubKey",#SubKey#)>
<cfset RowCounter = #RowCounter# + 1>
</cfoutput>
<!--- Set the number of threads by dividing the total TNs returned by n --->
<cfset numGroups = 50>
<cfset MigrationRounds = Ceiling(RowCounter / numGroups )>
<!--- Loop over the record set, creating n threads each with n TNs --->
<cfloop index="loopindex" from="1" to="#MigrationRounds#">
<!--- Ensure a unique thread name --->
<cfset threadName = "MigThread#loopindex#">
<cfif #loopindex# EQ 1>
<cfset startNumber = #loopindex#>
<cfelse>
<cfset startNumber = ((#loopindex# - 1) * numGroups) + 1>
</cfif>
<cfset endNumber = #loopindex# * numGroups>
<cfquery name="getSubset" dbtype="query">
SELECT IndexCounter, TN, SubKey
FROM SortedTNs
WHERE IndexCounter >= #startNumber# AND IndexCounter <= #endNumber#
</cfquery>
<cfoutput><p>#threadName# St: #startNumber# En: #endNumber# RecordCount is #getSubset.RecordCount#</p></cfoutput>
<cfthread
name="#threadName#"
action="run"
priority="HIGH">
<cfoutput query="getSubset">
<!--- Update the database with the thread name--->
<cfquery name="updateTN" datasource="xxx">
Update MigValidationNew
SET threadtest = '#threadName#'
WHERE SubKey = '#SubKey#'
</cfquery>
</cfoutput>
</cfthread>
<!--- Add delay between the threads to limit the corruption--->
<cfloop index="InternalIdx" from="1" to="3000000" step="1"> </cfloop>
<p>Next Thread!</p>
</cfloop>
<!--- Rejoin the threads for the display --->
<cfthread action="join" timeout="299999"/>
AdChoices