Ticket #287: p2.diff.txt

File p2.diff.txt, 9.6 KB (added by zooko, at 2010-02-01T01:56:15Z)
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 18:50:17.000000000 -0700
3+++ new-#928-late-buckets/src/allmydata/immutable/download.py   2010-01-31 18:50:19.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 18:50:17.000000000 -0700
26+++ new-#928-late-buckets/src/allmydata/test/no_network.py      2010-01-31 18:50:19.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 18:50:17.000000000 -0700
56+++ new-#928-late-buckets/src/allmydata/test/test_hung_server.py        2010-01-31 18:50:19.000000000 -0700
57@@ -9,6 +9,7 @@
58 from allmydata.storage.common import storage_index_to_dir
59 from allmydata.test.no_network import GridTestMixin
60 from allmydata.test.common import ShouldFailMixin
61+from allmydata.test.common_util import StallMixin
62 from allmydata.interfaces import NotEnoughSharesError
63 
64 immutable_plaintext = "data" * 10000
65@@ -22,9 +23,15 @@
66             self.g.break_server(id)
67 
68     def _hang(self, servers, **kwargs):
69+        print "YYY 0 %s" % (servers,)
70         for (id, ss) in servers:
71             self.g.hang_server(id, **kwargs)
72 
73+    def _unhang(self, servers, **kwargs):
74+        print "YYY 1 %s" % (servers,)
75+        for (id, ss) in servers:
76+            self.g.unhang_server(id, **kwargs)
77+
78     def _delete_all_shares_from(self, servers):
79         serverids = [id for (id, ss) in servers]
80         for (i_shnum, i_serverid, i_sharefile) in self.shares:
81@@ -80,35 +87,49 @@
82             d.addCallback(_uploaded_immutable)
83         return d
84 
85-    def _check_download(self):
86+    def _start_download(self):
87         n = self.c0.create_node_from_uri(self.uri)
88         if self.mutable:
89             d = n.download_best_version()
90-            expected_plaintext = mutable_plaintext
91         else:
92             d = download_to_data(n)
93-            expected_plaintext = immutable_plaintext
94-        def _got_data(data):
95-            self.failUnlessEqual(data, expected_plaintext)
96-        d.addCallback(_got_data)
97+        ciphertextdownloader = n._downloader._all_downloads.keys()[0]
98+        return (d, ciphertextdownloader._stage_4_d,)
99+
100+    def _wait_for_data(self, n):
101+        if self.mutable:
102+            d = n.download_best_version()
103+        else:
104+            d = download_to_data(n)
105+        return d
106+
107+    def _check(self, resultingdata):
108+        if self.mutable:
109+            self.failUnlessEqual(resultingdata, mutable_plaintext)
110+        else:
111+            self.failUnlessEqual(resultingdata, immutable_plaintext)
112+
113+    def _download_and_check(self):
114+        d, stage4d = self._start_download()
115+        d.addCallback(self._check)
116         return d
117 
118     def _should_fail_download(self):
119         if self.mutable:
120             return self.shouldFail(UnrecoverableFileError, self.basedir,
121                                    "no recoverable versions",
122-                                   self._check_download)
123+                                   self._download_and_check)
124         else:
125             return self.shouldFail(NotEnoughSharesError, self.basedir,
126                                    "Failed to get enough shareholders",
127-                                   self._check_download)
128+                                   self._download_and_check)
129 
130 
131     def test_10_good_sanity_check(self):
132         d = defer.succeed(None)
133         for mutable in [False, True]:
134             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_sanity_check"))
135-            d.addCallback(lambda ign: self._check_download())
136+            d.addCallback(lambda ign: self._download_and_check())
137         return d
138 
139     def test_10_good_copied_share(self):
140@@ -116,7 +137,7 @@
141         for mutable in [False, True]:
142             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_copied_share"))
143             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
144-            d.addCallback(lambda ign: self._check_download())
145+            d.addCallback(lambda ign: self._download_and_check())
146             return d
147 
148     def test_3_good_7_noshares(self):
149@@ -124,7 +145,7 @@
150         for mutable in [False, True]:
151             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_noshares"))
152             d.addCallback(lambda ign: self._delete_all_shares_from(self.servers[3:]))
153-            d.addCallback(lambda ign: self._check_download())
154+            d.addCallback(lambda ign: self._download_and_check())
155         return d
156 
157     def test_2_good_8_broken_fail(self):
158@@ -149,7 +170,7 @@
159             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_copied_share"))
160             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
161             d.addCallback(lambda ign: self._break(self.servers[2:]))
162-            d.addCallback(lambda ign: self._check_download())
163+            d.addCallback(lambda ign: self._download_and_check())
164         return d
165 
166     def test_2_good_8_broken_duplicate_share_fail(self):
167@@ -168,7 +189,7 @@
168         for mutable in [False]:
169             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_hung"))
170             d.addCallback(lambda ign: self._hang(self.servers[3:]))
171-            d.addCallback(lambda ign: self._check_download())
172+            d.addCallback(lambda ign: self._download_and_check())
173         return d
174 
175     def test_2_good_8_hung_then_1_recovers(self):
176@@ -176,10 +197,10 @@
177         for mutable in [False]:
178             recovered = defer.Deferred()
179             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers"))
180-            d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
181+            d.addCallback(lambda ign: self._hang(self.servers[2:3]))
182             d.addCallback(lambda ign: self._hang(self.servers[3:]))
183             d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
184-            d.addCallback(lambda ign: self._check_download())
185+            d.addCallback(lambda ign: self._download_and_check())
186         return d
187 
188     def test_2_good_8_hung_then_1_recovers_with_2_shares(self):
189@@ -188,8 +209,26 @@
190             recovered = defer.Deferred()
191             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers_with_2_shares"))
192             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[0:1], self.servers[2]))
193-            d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
194+            d.addCallback(lambda ign: self._hang(self.servers[2:3]))
195             d.addCallback(lambda ign: self._hang(self.servers[3:]))
196-            d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
197-            d.addCallback(lambda ign: self._check_download())
198+            d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
199+            d.addCallback(lambda ign: self._download_and_check())
200+        return d
201+
202+    def test_failover_during_stage_4(self):
203+        # See #287
204+        d = defer.succeed(None)
205+        for mutable in [False]:
206+            recovered = defer.Deferred()
207+            d.addCallback(lambda ign: self._set_up(mutable, "test_failover_during_stage_4"))
208+            d.addCallback(lambda ign: self._hang(self.servers[3:]))
209+            d.addCallback(lambda ign: self._start_download())
210+            def _after_stage4_starts(ign):
211+                # Now unhang server3
212+                self._unhang(self.servers[3])
213+                # and corrupt a share in server 2
214+                self._corrupt_share_in(self.servers[2])
215+            d.addCallback(lambda doned, started4d: started4d.addCallback(_after_stage4_starts))
216+            d.addCallback(lambda ign: self._check)
217+
218         return d