Ticket #287: p3.diff.txt

File p3.diff.txt, 9.7 KB (added by zooko, at 2010-02-01T03:15:37Z)
Line 
1diff -rN -u old-#928-late-buckets/src/allmydata/immutable/download.py new-#928-late-buckets/src/allmydata/immutable/download.py
2--- old-#928-late-buckets/src/allmydata/immutable/download.py   2010-01-31 20:09:22.000000000 -0700
3+++ new-#928-late-buckets/src/allmydata/immutable/download.py   2010-01-31 20:09:26.000000000 -0700
4@@ -811,6 +811,10 @@
5         # self._responses_received = 0
6         # self._queries_failed = 0
7 
8+        # This is solely for the use of unit tests. It will be triggered when
9+        # we start downloading shares.
10+        self._stage_4_d = defer.Deferred()
11+
12     def pauseProducing(self):
13         if self._paused:
14             return
15@@ -1119,6 +1123,8 @@
16             # this pause, at the end of write, prevents pre-fetch from
17             # happening until the consumer is ready for more data.
18             d.addCallback(self._check_for_pause)
19+
20+        self._stage_4_d.callback(None)
21         return d
22 
23     def _check_for_pause(self, res):
24diff -rN -u old-#928-late-buckets/src/allmydata/test/no_network.py new-#928-late-buckets/src/allmydata/test/no_network.py
25--- old-#928-late-buckets/src/allmydata/test/no_network.py      2010-01-31 20:09:22.000000000 -0700
26+++ new-#928-late-buckets/src/allmydata/test/no_network.py      2010-01-31 20:09:26.000000000 -0700
27@@ -252,12 +252,22 @@
28 
29     def break_server(self, serverid):
30         # mark the given server as broken, so it will throw exceptions when
31-        # asked to hold a share
32+        # asked to hold a share or serve a share
33         self.servers_by_id[serverid].broken = True
34 
35-    def hang_server(self, serverid, until=defer.Deferred()):
36-        # hang the given server until 'until' fires
37-        self.servers_by_id[serverid].hung_until = until
38+    def break_server_xyz(self, serverid):
39+        # mark the given server as broken, so it will throw exceptions when
40+        # asked to hold a share or serve a share
41+        self.servers_by_id[serverid].broken = True
42+
43+    def hang_server(self, serverid):
44+        # hang the given server
45+        assert self.servers_by_id[serverid].hung_until is None
46+        self.servers_by_id[serverid].hung_until = defer.Deferred()
47+
48+    def unhang_server(self, serverid):
49+        # unhang the given server
50+        self.servers_by_id[serverid].hung_until.callback(None)
51 
52 
53 class GridTestMixin:
54diff -rN -u old-#928-late-buckets/src/allmydata/test/test_hung_server.py new-#928-late-buckets/src/allmydata/test/test_hung_server.py
55--- old-#928-late-buckets/src/allmydata/test/test_hung_server.py        2010-01-31 20:09:23.000000000 -0700
56+++ new-#928-late-buckets/src/allmydata/test/test_hung_server.py        2010-01-31 20:09:26.000000000 -0700
57@@ -25,6 +25,10 @@
58         for (id, ss) in servers:
59             self.g.hang_server(id, **kwargs)
60 
61+    def _unhang(self, servers, **kwargs):
62+        for (id, ss) in servers:
63+            self.g.unhang_server(id, **kwargs)
64+
65     def _delete_all_shares_from(self, servers):
66         serverids = [id for (id, ss) in servers]
67         for (i_shnum, i_serverid, i_sharefile) in self.shares:
68@@ -80,35 +84,51 @@
69             d.addCallback(_uploaded_immutable)
70         return d
71 
72-    def _check_download(self):
73+    def _start_download(self):
74         n = self.c0.create_node_from_uri(self.uri)
75         if self.mutable:
76             d = n.download_best_version()
77-            expected_plaintext = mutable_plaintext
78+            stage_4_d = None # currently we aren't doing any tests which require this for mutable files
79+        else:
80+            d = download_to_data(n)
81+            stage_4_d = n._downloader._all_downloads.keys()[0]._stage_4_d
82+        return (d, stage_4_d,)
83+
84+    def _wait_for_data(self, n):
85+        if self.mutable:
86+            d = n.download_best_version()
87         else:
88             d = download_to_data(n)
89-            expected_plaintext = immutable_plaintext
90-        def _got_data(data):
91-            self.failUnlessEqual(data, expected_plaintext)
92-        d.addCallback(_got_data)
93+        return d
94+
95+    def _check(self, resultingdata):
96+        print "zzz: %s" % (resultingdata,)
97+        if self.mutable:
98+            self.failUnlessEqual(resultingdata, mutable_plaintext)
99+        else:
100+            self.failUnlessEqual(resultingdata, immutable_plaintext)
101+
102+    def _download_and_check(self):
103+        d, stage4d = self._start_download()
104+        d.addCallback(self._check)
105         return d
106 
107     def _should_fail_download(self):
108         if self.mutable:
109             return self.shouldFail(UnrecoverableFileError, self.basedir,
110                                    "no recoverable versions",
111-                                   self._check_download)
112+                                   self._download_and_check)
113         else:
114             return self.shouldFail(NotEnoughSharesError, self.basedir,
115                                    "Failed to get enough shareholders",
116-                                   self._check_download)
117+                                   self._download_and_check)
118 
119 
120     def test_10_good_sanity_check(self):
121         d = defer.succeed(None)
122         for mutable in [False, True]:
123             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_sanity_check"))
124-            d.addCallback(lambda ign: self._check_download())
125+            d.addCallback(lambda ign: self._download_and_check())
126         return d
127 
128     def test_10_good_copied_share(self):
129@@ -116,7 +136,7 @@
130         for mutable in [False, True]:
131             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_copied_share"))
132             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
133-            d.addCallback(lambda ign: self._check_download())
134+            d.addCallback(lambda ign: self._download_and_check())
135             return d
136 
137     def test_3_good_7_noshares(self):
138@@ -124,7 +144,7 @@
139         for mutable in [False, True]:
140             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_noshares"))
141             d.addCallback(lambda ign: self._delete_all_shares_from(self.servers[3:]))
142-            d.addCallback(lambda ign: self._check_download())
143+            d.addCallback(lambda ign: self._download_and_check())
144         return d
145 
146     def test_2_good_8_broken_fail(self):
147@@ -149,7 +169,7 @@
148             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_copied_share"))
149             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
150             d.addCallback(lambda ign: self._break(self.servers[2:]))
151-            d.addCallback(lambda ign: self._check_download())
152+            d.addCallback(lambda ign: self._download_and_check())
153         return d
154 
155     def test_2_good_8_broken_duplicate_share_fail(self):
156@@ -168,7 +188,7 @@
157         for mutable in [False]:
158             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_hung"))
159             d.addCallback(lambda ign: self._hang(self.servers[3:]))
160-            d.addCallback(lambda ign: self._check_download())
161+            d.addCallback(lambda ign: self._download_and_check())
162         return d
163 
164     def test_2_good_8_hung_then_1_recovers(self):
165@@ -176,10 +196,10 @@
166         for mutable in [False]:
167             recovered = defer.Deferred()
168             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers"))
169-            d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
170+            d.addCallback(lambda ign: self._hang(self.servers[2:3]))
171             d.addCallback(lambda ign: self._hang(self.servers[3:]))
172-            d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
173-            d.addCallback(lambda ign: self._check_download())
174+            d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
175+            d.addCallback(lambda ign: self._download_and_check())
176         return d
177 
178     def test_2_good_8_hung_then_1_recovers_with_2_shares(self):
179@@ -188,8 +208,34 @@
180             recovered = defer.Deferred()
181             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers_with_2_shares"))
182             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[0:1], self.servers[2]))
183-            d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
184+            d.addCallback(lambda ign: self._hang(self.servers[2:3]))
185             d.addCallback(lambda ign: self._hang(self.servers[3:]))
186-            d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
187-            d.addCallback(lambda ign: self._check_download())
188+            d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
189+            d.addCallback(lambda ign: self._download_and_check())
190+        return d
191+
192+    def test_failover_during_stage_4(self):
193+        # See #287
194+        d = defer.succeed(None)
195+        for mutable in [False]:
196+            recovered = defer.Deferred()
197+            d.addCallback(lambda ign: self._corrupt_share_in(self.servers[2]))
198+            d.addCallback(lambda ign: self._set_up(mutable, "test_failover_during_stage_4"))
199+            d.addCallback(lambda ign: self._hang(self.servers[3:]))
200+            d.addCallback(lambda ign: self._start_download())
201+            def _after_starting_download((doned, started4d)):
202+                print "xxx 0 %s %s" % (doned, started4d,)
203+                def p(x):
204+                    print "xxx1 %s" % (x,)
205+                    return x
206+                started4d.addCallback(p)
207+                started4d.addCallback(lambda ign: self._unhang(self.servers[3:4]))
208+                def p(x):
209+                    print "xxx2 %s" % (x,)
210+                    return x
211+                doned.addCallback(p)
212+                doned.addCallback(self._check)
213+                return doned
214+            d.addCallback(_after_starting_download)
215+
216         return d