Using threading module with Arcpy: Manager/Worker architecture

Question

I asked this question on Stack Overflow on how to setup a manager/worker scenario.

But I am getting some error when trying to use threading on my Python install with ArcGIS 10 (just the python install from ArcGIS 10.0).

Is anyone using threading successfully in arcpy.

My end goal is to be able to poll an arcgisjobs directory on ArcGIS Server 10.0 and call worker processes for each new folder. Each new process should be called and set to go and return control back to the manager to call more processes before previous ones are finished.

Answer 1

##################### Developed by: Ken Carrier ######################
#### 01/26/2012
#### Use Multiprocessing with an UpdateCursor
#### Check Values in attribute table that point to files on the network
#### If the file path exist mark field as TRUE If not FALSE
######################################################################
import arcpy, multiprocessing, time
from arcpy import env
from multiprocessing import Pool
# Overwrites values if they already exist.
env.overwriteOutput = True
# Specify the object you will be working with.
inputFC = r"C:\temp\PythonScripts\Multiprocessing\IfExists\DC@GISADMIN@MCESDB@GIS.sde\GIS.GISADMIN.tmpTest"
# Limit the fields used in the cursor, by Default all field are used.
# If all fields are used it will slow down the process.
# "NET_PATH"=0, "EXST" = 1, "OBJECTID" = 2 'Notice whenever UCFields is called it is  followed by [], the number
# corresponds to a field in the string array below.
UCFields = ["NET_PATH", "EXST", "OBJECTID"]
def FindValues(SQLQuery):
global inputFC
global UCFields
# Uncomment the line below to see the string that is being passed to the function.
## arcpy.AddMessage("SQL query being passed: " + SQLQuery)
# Prepare UpdateCursor
# inputFC = the object you will be working with.
# SQLQuery = a range of ObjectID's chunked out equally among processors or user defined processes.
# .join(UCFields) = limits the amount of fields seen by the cursor, limiting the fields increases performance.
# "OBJECTID A" = sorts the ObjectID field in Ascending Order.
rows = arcpy.UpdateCursor(inputFC, SQLQuery, "", "; ".join(UCFields), "OBJECTID A")

for row in rows:

if arcpy.Exists(row.getValue(UCFields[0])):
# Uncomment the line below to see the OBJECTID that is being processed
##arcpy.AddMessage(row.getValue(UCFields[2]))
# Process: Update the value to True if the item exists
row.setValue(UCFields[1], "TRUE")
rows.updateRow(row)

else:

# Uncomment the line below to see the OBJECTID that is being processed
##arcpy.AddMessage(row.getValue(UCFields[2]))

# Process: Update the value to False if the item does NOT exists
row.setValue(UCFields[1], "FALSE")
rows.updateRow(row)
# Process: Delete objects from memory
del rows, row

if __name__ == "__main__":
# Record the start time
arcpy.AddMessage("Started batch UpdateCursor job at: " + time.ctime())
# Uncomment the line below to evaluate the number of processors on the machine.
# This will be a multiple of available CPU cores multiplied by 3 minus 1
##partCount = int(os.environ["NUMBER_OF_PROCESSORS"]) * 2 - 1
# This is a user defined number of process you wish to use to split up the work
# This means the "the number of rows will be divided by 20 and that there will be 20   SQL queries for ranges
# of ObjectID's. You should see 20 python.exe processes in Task Manager.
partCount = 20

# Count the input table rows
#intputFC = r"C:\temp\PythonScripts\Multiprocessing\IfExists\DC@GISADMIN@MCESDB@GIS.sde\GIS.GISADMIN.tmpTest"
result = arcpy.GetCount_management(inputFC)
count = int(result.getOutput(0))
arcpy.AddMessage("Table: " + inputFC + " has rowcount " + str(count))
# Calculate the intervals suitable for a query ID >= <n> and ID < <n>
# based on the amount of rows in input table
intervals = range(1,count,int(count/partCount))
if len(intervals) > partCount:
intervals[-1] = count + 1
else:
intervals.append(count + 1)
# Create the queries for each table part.
# Get the OID,OBJECTID column name for the object you will be working with
desc = arcpy.Describe(inputFC)
OIDFieldName = desc.OIDFieldName
# Create a list of queries for OID ranges, as the queries are built they will be 
# appended to this list
queryList = []

for i in range(len(intervals)-1):
queryList.append(OIDFieldName + " >= " + str(intervals[i]) + " AND " + OIDFieldName + " <= " + str(intervals[i+1]))
arcpy.AddMessage("Built query: " + queryList[i])

# Create a list which results will be appended to.
results = list();
# Tells python to create 20 processes
pool = Pool(processes=partCount);

# Send each query of OID ranges to the worker function as a seperate process.
for q in queryList:
results.append(pool.apply_async(FindValues, args = (q, )))
pool.close();
pool.join();
# Record the end time
arcpy.AddMessage("End batch UpdateCursor job at: " + time.ctime()

Answer 2

I have not been able to get multiprocessing to work perfectly. It generally works, but i still get some occasional dying threads from access collisions no matter how much I isolate the workspaces. My solution has been to have the manager track completion closely, and to rebuild workers when tasks do not return. I do not have access to the code right now, but I will grab it tomorrow and post it up. I think there are some improvements in the next version specific to multiprocessing.

Answer 3

Multiprocessing module sometimes dont work properly that i want to advice you Python multithreaded job distribution module which name is WorkerPool. it is very good with coding. i have used it with some image processing for a long time and i have never get any error with it..

The workerpool module is a simple framework for easily distributing jobs into multiple worker threads.

This module facilitates distributing simple operations into jobs that are sent to worker threads, maintained by a pool object.

It consists of these components:

Jobs - single units of work that need to be performed. Workers - workers grab jobs from a queue and run them. Worker pool - keeps track of workers and the job queue.

example was taken from project site which i always use.

# download2.py - Download many URLs using multiple threads.

import os
import urllib
import workerpool

class DownloadJob(workerpool.Job):
    "Job for downloading a given URL."
    def __init__(self, url):
        self.url = url # The url we'll need to download when the job runs
    def run(self):
        save_to = os.path.basename(self.url)
        urllib.urlretrieve(self.url, save_to)

# Initialize a pool, 5 threads in this case
pool = workerpool.WorkerPool(size=5)

# Loop over urls.txt and create a job to download the URL on each line
for url in open("urls.txt"):
    job = DownloadJob(url.strip())
    pool.put(job)

# Send shutdown jobs to all threads, and wait until all the jobs have been completed
pool.shutdown()
pool.wait()

i hope it help you...