pdorrell / keevalbak
watch download tarball
public
Forks0Right
Watchers2Right
Description: Python backup classes to backup files to key/value store, such as Amazon S3
Homepage: http://www.1729.com/software/keevalbak/index.html
Clone URL: git://github.com/pdorrell/keevalbak.git
keevalbak / keevalbak / ThreadedTaskRunner.py
raw normal view history
c140ae08 » Philip Dorrell 2008-06-29 license for ThreadedTaskRunner 1 # Copyright (c) 2008 Philip Dorrell, http://www.1729.com/
2 #
3 # Permission is hereby granted, free of charge, to any person obtaining a copy
4 # of this software and associated documentation files (the "Software"), to deal
5 # in the Software without restriction, including without limitation the rights
6 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7 # copies of the Software, and to permit persons to whom the Software is
8 # furnished to do so, subject to the following conditions:
9 #
10 # The above copyright notice and this permission notice shall be included in
11 # all copies or substantial portions of the Software.
12 #
13 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19 # THE SOFTWARE.
20
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 21 import Queue
22 import threading
23
b147c7f4 » Philip Dorrell 2008-07-03 ThreadedTaskRunner inherits... 24 class TaskRunner(object):
25 """Simple task runner: runs both parts of tasks synchronously"""
26 def __init__(self, checkpointFreq = None):
27 self.checkpointFreq = checkpointFreq
28
29 def runTasksInit(self):
30 pass
31
32 def doUnsynchronizedTasks(self, tasks):
33 for task in tasks:
34 task.doUnsynchronized()
35
36 def runTasks(self, tasks, checkpointTask = None):
37 self.runTasksInit()
38 startIndex = 0
39 numTasks = len(tasks)
40 while startIndex < numTasks:
41 if self.checkpointFreq == None or checkpointTask == None:
42 endIndex = numTasks
43 else:
44 endIndex = min(startIndex+self.checkpointFreq, numTasks)
45 self.doUnsynchronizedTasks (tasks[startIndex:endIndex])
46 for i in range(startIndex, endIndex):
47 tasks[i].doSynchronized()
48 if endIndex < numTasks:
49 print "CHECKPOINT:"
50 if checkpointTask != None:
51 checkpointTask.checkpoint()
52 startIndex = endIndex
53
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 54 class TaskProcessor(threading.Thread):
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 55 def __init__(self, queue, index):
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 56 threading.Thread.__init__(self)
57 self.queue = queue
58 self.index = index
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 59 self.threadLocals = None
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 60
61 def run(self):
62 while True:
63 task = self.queue.get()
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 64 if self.threadLocals == None:
65 self.threadLocals = task.getThreadLocals()
f5d4897a » Philip Dorrell 2008-06-29 dont log thread start/finish 66 #print "Thread %d performing task ..." % self.index
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 67 for key, value in self.threadLocals.iteritems():
68 task.__dict__[key] = value
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 69 task.doUnsynchronized()
70 self.queue.task_done()
f5d4897a » Philip Dorrell 2008-06-29 dont log thread start/finish 71 #print "Thread %d finished performing task ..." % self.index
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 72
b147c7f4 » Philip Dorrell 2008-07-03 ThreadedTaskRunner inherits... 73 class ThreadedTaskRunner(TaskRunner):
74 def __init__(self, checkpointFreq = 10, numThreads = 10):
75 super(ThreadedTaskRunner, self).__init__(checkpointFreq)
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 76 self.queue = Queue.Queue()
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 77 self.processors = [TaskProcessor(self.queue, i) for i in range(numThreads)]
3fd43cb5 » Philip Dorrell 2008-06-29 using ThreadedTaskRunner af... 78 for processor in self.processors:
79 processor.setDaemon(True)
80 processor.start()
b147c7f4 » Philip Dorrell 2008-07-03 ThreadedTaskRunner inherits... 81
82 def runTasksInit(self):
028e041c » Philip Dorrell 2008-07-03 tasks are now responsible f... 83 for processor in self.processors:
84 processor.threadLocals = None
b147c7f4 » Philip Dorrell 2008-07-03 ThreadedTaskRunner inherits... 85
86 def doUnsynchronizedTasks(self, tasks):
ced8efa1 » Philip Dorrell 2008-06-28 threaded task runner 87 for task in tasks:
88 self.queue.put (task)
89 self.queue.join()