« back to all changes in this revision
Viewing changes to duplicity/lazy.py
- browse files at revision 1
- compare with another revision
- download diff
- download tarball
- view history from revision 1
- Committer: bescoto
- Date: 2002-10-29 01:49:46 UTC
- Revision ID: vcs-imports@canonical.com-20021029014946-3m4rmm5plom7pl6q
Initial checkin
- files added:
- dist
- duplicity
- testing
added
removed
duplicity/lazy.py
1
# Copyright 2002 Ben Escoto
2
#
3
# This file is part of duplicity.
4
#
5
# duplicity is free software; you can redistribute it and/or modify it
6
# under the terms of the GNU General Public License as published by
7
# the Free Software Foundation, Inc., 675 Mass Ave, Cambridge MA
8
# 02139, USA; either version 2 of the License, or (at your option) any
9
# later version; incorporated herein by reference.
10
11
"""Define some lazy data structures and functions acting on them"""
12
13
from __future__ import generators
14
import os, stat, types, sys
15
from static import *
16
17
18
class Iter:
19
"""Hold static methods for the manipulation of lazy iterators"""
20
21
def filter(predicate, iterator):
22
"""Like filter in a lazy functional programming language"""
23
for i in iterator:
24
if predicate(i): yield i
25
26
def map(function, iterator):
27
"""Like map in a lazy functional programming language"""
28
for i in iterator: yield function(i)
29
30
def foreach(function, iterator):
31
"""Run function on each element in iterator"""
32
for i in iterator: function(i)
33
34
def cat(*iters):
35
"""Lazily concatenate iterators"""
36
for iter in iters:
37
for i in iter: yield i
38
39
def cat2(iter_of_iters):
40
"""Lazily concatenate iterators, iterated by big iterator"""
41
for iter in iter_of_iters:
42
for i in iter: yield i
43
44
def empty(iter):
45
"""True if iterator has length 0"""
46
for i in iter: return None
47
return 1
48
49
def equal(iter1, iter2, verbose = None, operator = lambda x, y: x == y):
50
"""True if iterator 1 has same elements as iterator 2
51
52
Use equality operator, or == if it is unspecified.
53
54
"""
55
for i1 in iter1:
56
try: i2 = iter2.next()
57
except StopIteration:
58
if verbose: print "End when i1 = %s" % (i1,)
59
return None
60
if not operator(i1, i2):
61
if verbose: print "%s not equal to %s" % (i1, i2)
62
return None
63
try: i2 = iter2.next()
64
except StopIteration: return 1
65
if verbose: print "End when i2 = %s" % (i2,)
66
return None
67
68
def Or(iter):
69
"""True if any element in iterator is true. Short circuiting"""
70
i = None
71
for i in iter:
72
if i: return i
73
return i
74
75
def And(iter):
76
"""True if all elements in iterator are true. Short circuiting"""
77
i = 1
78
for i in iter:
79
if not i: return i
80
return i
81
82
def len(iter):
83
"""Return length of iterator"""
84
i = 0
85
while 1:
86
try: iter.next()
87
except StopIteration: return i
88
i = i+1
89
90
def foldr(f, default, iter):
91
"""foldr the "fundamental list recursion operator"?"""
92
try: next = iter.next()
93
except StopIteration: return default
94
return f(next, Iter.foldr(f, default, iter))
95
96
def foldl(f, default, iter):
97
"""the fundamental list iteration operator.."""
98
while 1:
99
try: next = iter.next()
100
except StopIteration: return default
101
default = f(default, next)
102
103
def multiplex(iter, num_of_forks, final_func = None, closing_func = None):
104
"""Split a single iterater into a number of streams
105
106
The return val will be a list with length num_of_forks, each
107
of which will be an iterator like iter. final_func is the
108
function that will be called on each element in iter just as
109
it is being removed from the buffer. closing_func is called
110
when all the streams are finished.
111
112
"""
113
if num_of_forks == 2 and not final_func and not closing_func:
114
im2 = IterMultiplex2(iter)
115
return (im2.yielda(), im2.yieldb())
116
if not final_func: final_func = lambda i: None
117
if not closing_func: closing_func = lambda: None
118
119
# buffer is a list of elements that some iterators need and others
120
# don't
121
buffer = []
122
123
# buffer[forkposition[i]] is the next element yieled by iterator
124
# i. If it is -1, yield from the original iter
125
starting_forkposition = [-1] * num_of_forks
126
forkposition = starting_forkposition[:]
127
called_closing_func = [None]
128
129
def get_next(fork_num):
130
"""Return the next element requested by fork_num"""
131
if forkposition[fork_num] == -1:
132
try: buffer.insert(0, iter.next())
133
except StopIteration:
134
# call closing_func if necessary
135
if (forkposition == starting_forkposition and
136
not called_closing_func[0]):
137
closing_func()
138
called_closing_func[0] = None
139
raise StopIteration
140
for i in range(num_of_forks): forkposition[i] += 1
141
142
return_val = buffer[forkposition[fork_num]]
143
forkposition[fork_num] -= 1
144
145
blen = len(buffer)
146
if not (blen-1) in forkposition:
147
# Last position in buffer no longer needed
148
assert forkposition[fork_num] == blen-2
149
final_func(buffer[blen-1])
150
del buffer[blen-1]
151
return return_val
152
153
def make_iterator(fork_num):
154
while(1): yield get_next(fork_num)
155
156
return tuple(map(make_iterator, range(num_of_forks)))
157
158
MakeStatic(Iter)
159
160
161
class IterMultiplex2:
162
"""Multiplex an iterator into 2 parts
163
164
This is a special optimized case of the Iter.multiplex function,
165
used when there is no closing_func or final_func, and we only want
166
to split it into 2. By profiling, this is a time sensitive class.
167
168
"""
169
def __init__(self, iter):
170
self.a_leading_by = 0 # How many places a is ahead of b
171
self.buffer = []
172
self.iter = iter
173
174
def yielda(self):
175
"""Return first iterator"""
176
buf, iter = self.buffer, self.iter
177
while(1):
178
if self.a_leading_by >= 0: # a is in front, add new element
179
elem = iter.next() # exception will be passed
180
buf.append(elem)
181
else: elem = buf.pop(0) # b is in front, subtract an element
182
self.a_leading_by += 1
183
yield elem
184
185
def yieldb(self):
186
"""Return second iterator"""
187
buf, iter = self.buffer, self.iter
188
while(1):
189
if self.a_leading_by <= 0: # b is in front, add new element
190
elem = iter.next() # exception will be passed
191
buf.append(elem)
192
else: elem = buf.pop(0) # a is in front, subtract an element
193
self.a_leading_by -= 1
194
yield elem
195
196
197
class IterTreeReducer:
198
"""Tree style reducer object for iterator - stolen from rdiff-backup
199
200
The indicies of a RORPIter form a tree type structure. This class
201
can be used on each element of an iter in sequence and the result
202
will be as if the corresponding tree was reduced. This tries to
203
bridge the gap between the tree nature of directories, and the
204
iterator nature of the connection between hosts and the temporal
205
order in which the files are processed.
206
207
This will usually be used by subclassing ITRBranch below and then
208
call the initializer below with the new class.
209
210
"""
211
def __init__(self, branch_class, branch_args):
212
"""ITR initializer"""
213
self.branch_class = branch_class
214
self.branch_args = branch_args
215
self.index = None
216
self.root_branch = branch_class(*branch_args)
217
self.branches = [self.root_branch]
218
219
def finish_branches(self, index):
220
"""Run Finish() on all branches index has passed
221
222
When we pass out of a branch, delete it and process it with
223
the parent. The innermost branches will be the last in the
224
list. Return None if we are out of the entire tree, and 1
225
otherwise.
226
227
"""
228
branches = self.branches
229
while 1:
230
to_be_finished = branches[-1]
231
base_index = to_be_finished.base_index
232
if base_index != index[:len(base_index)]:
233
# out of the tree, finish with to_be_finished
234
to_be_finished.call_end_proc()
235
del branches[-1]
236
if not branches: return None
237
branches[-1].branch_process(to_be_finished)
238
else: return 1
239
240
def add_branch(self):
241
"""Return branch of type self.branch_class, add to branch list"""
242
branch = self.branch_class(*self.branch_args)
243
self.branches.append(branch)
244
return branch
245
246
def process_w_branch(self, index, branch, args):
247
"""Run start_process on latest branch"""
248
robust.check_common_error(branch.on_error,
249
branch.start_process, args)
250
if not branch.caught_exception: branch.start_successful = 1
251
branch.base_index = index
252
253
def Finish(self):
254
"""Call at end of sequence to tie everything up"""
255
while 1:
256
to_be_finished = self.branches.pop()
257
to_be_finished.call_end_proc()
258
if not self.branches: break
259
self.branches[-1].branch_process(to_be_finished)
260
261
def __call__(self, *args):
262
"""Process args, where args[0] is current position in iterator
263
264
Returns true if args successfully processed, false if index is
265
not in the current tree and thus the final result is
266
available.
267
268
Also note below we set self.index after doing the necessary
269
start processing, in case there is a crash in the middle.
270
271
"""
272
index = args[0]
273
if self.index is None:
274
self.process_w_branch(index, self.root_branch, args)
275
self.index = index
276
return 1
277
278
if index <= self.index:
279
log.Log("Warning: oldindex %s >= newindex %s" %
280
(self.index, index), 2)
281
return 1
282
283
if self.finish_branches(index) is None:
284
return None # We are no longer in the main tree
285
last_branch = self.branches[-1]
286
if last_branch.start_successful:
287
if last_branch.can_fast_process(*args):
288
robust.check_common_error(last_branch.on_error,
289
last_branch.fast_process, args)
290
else:
291
branch = self.add_branch()
292
self.process_w_branch(index, branch, args)
293
else: last_branch.log_prev_error(index)
294
295
self.index = index
296
return 1
297
298
299
class ITRBranch:
300
"""Helper class for IterTreeReducer above
301
302
There are five stub functions below: start_process, end_process,
303
branch_process, fast_process, and can_fast_process. A class that
304
subclasses this one will probably fill in these functions to do
305
more.
306
307
"""
308
base_index = index = None
309
finished = None
310
caught_exception = start_successful = None
311
312
def call_end_proc(self):
313
"""Runs the end_process on self, checking for errors"""
314
if self.finished or not self.start_successful:
315
self.caught_exception = 1
316
317
#if self.caught_exception: self.log_prev_error(self.base_index)
318
#else: robust.check_common_error(self.on_error, self.end_process)
319
# Since all end_process does is copy over attributes, might as
320
# well run it even if we did get errors earlier.
321
robust.check_common_error(self.on_error, self.end_process)
322
323
self.finished = 1
324
325
def start_process(self, *args):
326
"""Do some initial processing (stub)"""
327
pass
328
329
def end_process(self):
330
"""Do any final processing before leaving branch (stub)"""
331
pass
332
333
def branch_process(self, branch):
334
"""Process a branch right after it is finished (stub)"""
335
assert branch.finished
336
pass
337
338
def can_fast_process(self, *args):
339
"""True if object can be processed without new branch (stub)"""
340
return None
341
342
def fast_process(self, *args):
343
"""Process args without new child branch (stub)"""
344
pass
345
346
def on_error(self, exc, *args):
347
"""This is run on any exception in start/end-process"""
348
self.caught_exception = 1
349
if args and args[0] and isinstance(args[0], tuple):
350
filename = os.path.join(*args[0])
351
elif self.index: filename = os.path.join(*self.index)
352
else: filename = "."
353
log.Log("Error '%s' processing %s" % (exc, filename), 2)
354
355
def log_prev_error(self, index):
356
"""Call function if no pending exception"""
357
if not index: index_str = "."
358
else: index_str = os.path.join(*index)
359
log.Log("Skipping %s because of previous error" % index_str, 2)
360
361
362
import robust, log
Loggerhead is a web-based interface for Breezy
Version: 2.0.1