0 Replies Latest reply on Oct 2, 2013 9:41 AM by DaveOHare01

    CFThread Corruption Issue

    DaveOHare01

      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"/>