Source Code for Module park.core.pool

 1  # This program is public domain 
 2  """ 
 3  Support for job pools. 
 4   
 5  Example 
 6  ======= 
 7   
 8  Using the above, you can set up a workflow which halts when any job on the 
 9  workflow improves past a certain point:: 
10   
11      from park.core.pool import terminate, any_improved 
12   
13      M = mymodel 
14      servers = ['compufans','teragrid','teragrid',local_service] 
15      pool = [fit(M, server=s) for s in servers] 
16      terminate(pool, when=any_improved(pool, 'chisq<20'), server=local_server) 
17   
18  TODO: How do we make the output of terminate be the result of the first 
19  job to 
20   
21  :group Composite conditions: 
22      any_improved, all_improved, any_completed, all_completed 
23   
24  :group Services: 
25      terminate 
26  """ 
27   
28  __all__ = ['terminate', 'any_improved', 'all_improved', 
29             'any_completed', 'all_completed'] 
30  import park.core.service as service 
31   
32 -class TerminateService(service.Service): 
33      version='1.0' 
34 -    def prepare(self, request): 
35          self.jobs = request.jobs 
36 -    def run(self, handler): 
37          for job in jobs: 
38              # TODO: send() needs to accept server, auth, stream 
39              handler.send(message.Abort(), 
40                           server=job['server'], 
41                           auth=job['auth'], 
42                           stream=job['stream']) 
43          return [] 
44   
45 -def terminate(pool, **kw): 
46      """ 
47      Terminate a pool of jobs when condition is satisfied. 
48      Standard job submission keywords are accepted, including:: 
49          when=condition 
50          service=service 
51      See `park.core.service.Proxy.submit` 
52      """ 
53      jobs = [dict(stream=job.id,auth=job.auth,server=job.server) 
54              for job in pool] 
55      request = service.request(TerminateService, jobs=jobs) 
56      job = service.Proxy() 
57      job.submit(request=request, **kw) 
58      return job 
59   
60   
61 -def any_improved(jobs=[], expression=None): 
62      """ 
63      Condition is satisfied when any of a set of jobs has improved to 
64      the point where expression evaluates to True. 
65      """ 
66      assert len(jobs)>0 
67      condition = jobs[0].isimproved(expression) 
68      for job in jobs[1:]: 
69          condition |= job.isimproved(expression) 
70   
71 -def any_completed(jobs=[]): 
72      """ 
73      Condition is satisfied when any of a set of jobs are completed. 
74      """ 
75      assert len(jobs)>0 
76      condition = jobs[0].iscompleted() 
77      for job in jobs[1:]: 
78          condition |= job.iscompleted() 
79   
80 -def all_improved(jobs=[], expression=None): 
81      """ 
82      Condition is satisfied when any of a set of jobs has improved to 
83      the point where expression evaluates to True. 
84      """ 
85      assert len(jobs)>0 
86      condition = jobs[0].isimproved(expression) 
87      for job in jobs[1:]: 
88          condition &= job.isimproved(expression) 
89   
90 -def all_completed(jobs=[]): 
91      """ 
92      Condition is satisfied when any of a set of jobs are completed. 
93      """ 
94      assert len(jobs)>0 
95      condition = jobs[0].iscompleted() 
96      for job in jobs[1:]: 
97          condition &= job.iscompleted() 
98