CodeBlog.TheG2.NET: Parallel ForEach file processing in IronPython: Get some TPL love in IPY!

I recently have started playing around with the Task Parallel Library (TPL) that will be shipping in .NET 4, which lets you easily spread a workload across multiple cores using a simple Parallel.ForEach statement. I have a few IronPython scripts that I want to convert to start using multiple threads, so I thought I would try using the .NET 3.5 version of the TPL from inside of IronPython. It turns out that the TPL works just fine, but writing thread safe code in IronPython can be tricky. First off, the System.Threading.Interlocked class does not work on IronPython integers because python integers are immutable. That really sucks, because it means that the only way to change a global integer value in a thread safe manner is to use locks or the System.Threading.Monitor class. One interesting workaround is to use a python list instead of an integer. Appending a value to a list is an atomic operation, so instead of calling Interlocked.Add you can just append all the values to an empty list an use sum to return the aggregate value or len() to get the number of items in the list (emulating incrementing a variable). It is a bit of a hack, but sometimes easier that using locks. Also, you have to remember that the print statement is not thread safe, so instead you should use something like System.Text.StringBuilder to buffer your print statements and then print them once you are back to a single thread code section.

Below is a small sample script that I wrote to test processing a large number of text files using IronPython. It will search through all files in a given directory that match a given pattern (ie: *.txt) and return the total number of files and lines. It will also search for a given tolken inside each file and return the total number of matches. This is a task that is very easy to run on multiple cores and is a perfect fit for the Parallel.ForEach method. By default it will search your temporary files folder, which on my workstation had 69 .txt files with a total of 469,286 lines. The single threaded version took about 2.3 seconds to run and the multi-threaded version took between 0.8 and 1.0 seconds. The workload was spread across 8 cores, which caused a 2x improvement in speed, even thought this example is still primarily bound by the drive IO speed. Still it shows how using the TPL can greatly increase performance for common tasks.

"""Uses the Task Parallel Libray (TPL) to processes files using multiple threads.
Please download and install the Reactive Extensions (Rx) for .Net 3.5 from http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx before running this script
Usage: import the file into an existing script or run the following from the IronPython console:
execfile("c:\PathToFile\Filename.py")
"""
import clr
from System import Environment
from System.IO import File,Directory
from System.Diagnostics import Stopwatch
from System.Text import StringBuilder
from System.Threading import Monitor
clr.AddReference('System.Windows.Forms')

strAssembly = "C:\Program Files (x86)\Microsoft Reactive Extensions\Redist\DesktopV2\System.Threading.dll";
if not File.Exists(strAssembly): # Try both 32bit and 64bit file path
    strAssembly = "C:\Program Files\Microsoft Reactive Extensions\Redist\DesktopV2\System.Threading.dll";
    if not File.Exists(strAssembly): # Library does not exist, prompt with error message
        print "TPL assembly not found. Please download and install the Reactive Extensions (Rx) for .Net 3.5 from http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx"
        raise Exception("Task Parallel Libray not found. Please download and install the Reactive Extensions (Rx) for .Net 3.5");
        #Rx and TPL for .NET 3.5 can be downloaded from  http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx

#Load TPL
clr.AddReferenceToFileAndPath(strAssembly)
from System.Threading.Tasks import *

#Method to process files. See test call at bottom of file
def TPLProcessFiles(strTolken, strPath=None, strFilter="*.txt"):
    """Finds all occurences of strTolken in strPath for files that match strFilter.
        strTolken = string tolken used for searching. NOTE: search is case-insensitive and includes partial match (cat will match Category).
        strPath = Path to the directory used for searching. Default is system's temp directory
        strFilter = file match filter for search. Default is *.txt"""
    print "Start TPLProcessFiles('%s', '%s', '%s')" % (strTolken, strPath, strFilter)
    
    if strPath == None:
        strPath = Environment.GetEnvironmentVariable("TEMP")
        print "Using temporary files path: %s" % strPath
    elif not Directory.Exists(strPath):
        raise Exception("Invalid path: " + strPath)
    
    #Setup
    swTime = Stopwatch()
    swTime.Start()
    LogWriter.Reset()
    Results.Reset()
    strFiles = Directory.GetFiles(strPath, strFilter)

#Single thread test code. Useful for when you get an AggregateException_ctor_DefaultMessage exception in the Parallel.ForEach version or for benchmarking
    #for aFile in strFiles:
    #    SearchFile(aFile, strTolken)
    
    #TPL version with multiple threads
    pfeResult = Parallel.ForEach(strFiles, lambda aFile: SearchFile(aFile, strTolken))
    
    print LogWriter.ToString() #Have to use LogWriter class below because print is not a threadsafe function
    strResults = Results.ToString() #Have to store results in a thread safe class
    swTime.Stop()
    print "Finished. Total Files: %d  LogCallCount: %d   Total Time: %0.3fs  Total Cores: %d" % (len(strFiles), LogWriter.GetCallCount(), float(swTime.ElapsedMilliseconds)/1000, Environment.ProcessorCount)
    print "Results - %s" % strResults

#File search method    
def SearchFile(strFilename, strTolken):
    """Method used for searching a file. This will be called from the Parallel.ForEach loop.
        Returns a tuple of (Filename, Total line count, Total tolken count).
        NOTE: Code must be thread safe, so you cannot use the print statement or increment global variables."""
    #LogWriter.WriteLine("Start file: %s searching for %s" % (strFilename, strTolken))
    f = open(strFilename, "r") #open file as readonly
    t = f.read() #get text
    f.close()
    Results.Add(1, t.count("\n"),t.count(strTolken)) #Add to global results

#Class used to store results    
class Results:
    Files = 0 #static members
    Lines = 0
    Tolkens = 0   
    
    @staticmethod
    def Reset(): #NOTE: this is not thread safe, but we will only call this method from a single thread
        Results.Files = 0
        Results.Lines = 0
        Results.Tolkens = 0

@staticmethod
    def Add(intFiles, intLines, intTolkens):
        Monitor.Enter(Results) #Use the System.Threading.Monitor class to ensure addition is thread safe
        Results.Files += intFiles
        Results.Lines += intLines
        Results.Tolkens += intTolkens
        Monitor.Exit(Results) #Finish thread safe code

@staticmethod
    def GetValues():
        return (Results.Files, Results.Lines, Results.Tolkens)
        
    @staticmethod
    def ToString():
        return "Total files: %d  Total lines: %d  Total tolkens: %d" % Results.GetValues()
        
#Log writer class to buffer print statements    
class LogWriter:
    sbLog = StringBuilder() #StringBuilder is thread safe when used as a static member
    CallCount = [] #Integer operations are not threadsafe, but appending to a list is
    
    @staticmethod
    def WriteLine(strMessage): #NOTE: must be thread safe. String 
        LogWriter.sbLog.Append(strMessage+"\r\n")
        LogWriter.CallCount.append(1)
    
    @staticmethod
    def ToString():
        return LogWriter.sbLog.ToString()
    
    @staticmethod
    def Reset():
        LogWriter.sbLog.Length = 0
        LogWriter.intCallCount = []

@staticmethod
    def GetCallCount():
        return sum(LogWriter.CallCount)

if __name__ == "__main__":
    TPLProcessFiles("Error") #Run test if this file is loaded from the IronPython Console