Ticket #287: p4a.diff.txt

File p4a.diff.txt, 12.3 KB (added by davidsarah, at 2010-02-01T03:40:44Z)
Line 
1diff -rN -u old-tahoe/src/allmydata/immutable/download.py new-tahoe/src/allmydata/immutable/download.py
2--- old-tahoe/src/allmydata/immutable/download.py       2010-02-01 03:34:57.260000000 +0000
3+++ new-tahoe/src/allmydata/immutable/download.py       2010-02-01 03:35:01.431000000 +0000
4@@ -809,6 +809,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@@ -1109,6 +1113,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-tahoe/src/allmydata/test/no_network.py new-tahoe/src/allmydata/test/no_network.py
25--- old-tahoe/src/allmydata/test/no_network.py  2010-02-01 03:34:57.794000000 +0000
26+++ new-tahoe/src/allmydata/test/no_network.py  2010-02-01 03:35:01.974000000 +0000
27@@ -252,12 +252,26 @@
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 break_server_xyz(self, serverid):
36+        # mark the given server as broken, so it will throw exceptions when
37+        # asked to hold a share or serve a share
38         self.servers_by_id[serverid].broken = True
39 
40     def hang_server(self, serverid, until=defer.Deferred()):
41-        # hang the given server until 'until' fires
42-        self.servers_by_id[serverid].hung_until = until
43+        # hang the given server
44+        ss = self.servers_by_id[serverid]
45+        assert ss.hung_until is None
46+        ss.hung_until = until
47+
48+    def unhang_server(self, serverid):
49+        # unhang the given server
50+        ss = self.servers_by_id[serverid]
51+        assert ss.hung_until is not None
52+        ss.hung_until.callback(None)
53+        ss.hung_until = None
54 
55 
56 class GridTestMixin:
57diff -rN -u old-tahoe/src/allmydata/test/test_hung_server.py new-tahoe/src/allmydata/test/test_hung_server.py
58--- old-tahoe/src/allmydata/test/test_hung_server.py    2010-02-01 03:34:58.030000000 +0000
59+++ new-tahoe/src/allmydata/test/test_hung_server.py    2010-02-01 03:35:02.141000000 +0000
60@@ -8,7 +8,7 @@
61 from allmydata.mutable.common import UnrecoverableFileError
62 from allmydata.storage.common import storage_index_to_dir
63 from allmydata.test.no_network import GridTestMixin
64-from allmydata.test.common import ShouldFailMixin
65+from allmydata.test.common import ShouldFailMixin, _corrupt_share_data
66 from allmydata.interfaces import NotEnoughSharesError
67 
68 immutable_plaintext = "data" * 10000
69@@ -22,15 +22,27 @@
70             self.g.break_server(id)
71 
72     def _hang(self, servers, **kwargs):
73+        print "YYY 0 %s" % (servers,)
74         for (id, ss) in servers:
75             self.g.hang_server(id, **kwargs)
76 
77+    def _unhang(self, servers, **kwargs):
78+        print "YYY 1 %s" % (servers,)
79+        for (id, ss) in servers:
80+            self.g.unhang_server(id, **kwargs)
81+
82     def _delete_all_shares_from(self, servers):
83         serverids = [id for (id, ss) in servers]
84         for (i_shnum, i_serverid, i_sharefile) in self.shares:
85             if i_serverid in serverids:
86                 os.unlink(i_sharefile)
87 
88+    def _corrupt_all_shares_in(self, servers, corruptor_func):
89+        serverids = [id for (id, ss) in servers]
90+        for (i_shnum, i_serverid, i_sharefile) in self.shares:
91+            if i_serverid in serverids:
92+                self._corrupt_share((i_shnum, i_sharefile), corruptor_func)
93+
94     def _copy_all_shares_from(self, from_servers, to_server):
95         serverids = [id for (id, ss) in from_servers]
96         for (i_shnum, i_serverid, i_sharefile) in self.shares:
97@@ -38,19 +50,28 @@
98                 self._copy_share((i_shnum, i_sharefile), to_server)
99 
100     def _copy_share(self, share, to_server):
101-         (sharenum, sharefile) = share
102-         (id, ss) = to_server
103-         shares_dir = os.path.join(ss.original.storedir, "shares")
104-         si = uri.from_string(self.uri).get_storage_index()
105-         si_dir = os.path.join(shares_dir, storage_index_to_dir(si))
106-         if not os.path.exists(si_dir):
107-             os.makedirs(si_dir)
108-         new_sharefile = os.path.join(si_dir, str(sharenum))
109-         shutil.copy(sharefile, new_sharefile)
110-         self.shares = self.find_shares(self.uri)
111-         # Make sure that the storage server has the share.
112-         self.failUnless((sharenum, ss.original.my_nodeid, new_sharefile)
113-                         in self.shares)
114+        (sharenum, sharefile) = share
115+        (id, ss) = to_server
116+        shares_dir = os.path.join(ss.original.storedir, "shares")
117+        si = uri.from_string(self.uri).get_storage_index()
118+        si_dir = os.path.join(shares_dir, storage_index_to_dir(si))
119+        if not os.path.exists(si_dir):
120+            os.makedirs(si_dir)
121+        new_sharefile = os.path.join(si_dir, str(sharenum))
122+        shutil.copy(sharefile, new_sharefile)
123+        self.shares = self.find_shares(self.uri)
124+        # Make sure that the storage server has the share.
125+        self.failUnless((sharenum, ss.original.my_nodeid, new_sharefile)
126+                        in self.shares)
127+
128+    def _corrupt_share(self, share, corruptor_func):
129+        (sharenum, sharefile) = share
130+        data = open(sharefile, "rb").read()
131+        newdata = corruptor_func(data)
132+        os.unlink(sharefile)
133+        wf = open(sharefile, "wb")
134+        wf.write(newdata)
135+        wf.close()
136 
137     def _set_up(self, mutable, testdir, num_clients=1, num_servers=10):
138         self.mutable = mutable
139@@ -80,35 +101,51 @@
140             d.addCallback(_uploaded_immutable)
141         return d
142 
143-    def _check_download(self):
144+    def _start_download(self):
145         n = self.c0.create_node_from_uri(self.uri)
146         if self.mutable:
147             d = n.download_best_version()
148-            expected_plaintext = mutable_plaintext
149+            self._stage_4_d = None
150+        else:
151+            d = download_to_data(n)
152+            ciphertextdownloader = n._downloader._all_downloads.keys()[0]  # too ugly! FIXME
153+            self._stage_4_d = ciphertextdownloader._stage_4_d
154+        return d
155+
156+    def _wait_for_data(self, n):
157+        if self.mutable:
158+            d = n.download_best_version()
159         else:
160             d = download_to_data(n)
161-            expected_plaintext = immutable_plaintext
162-        def _got_data(data):
163-            self.failUnlessEqual(data, expected_plaintext)
164-        d.addCallback(_got_data)
165+        return d
166+
167+    def _check(self, resultingdata):
168+        if self.mutable:
169+            self.failUnlessEqual(resultingdata, mutable_plaintext)
170+        else:
171+            self.failUnlessEqual(resultingdata, immutable_plaintext)
172+
173+    def _download_and_check(self):
174+        d = self._start_download()
175+        d.addCallback(self._check)
176         return d
177 
178     def _should_fail_download(self):
179         if self.mutable:
180             return self.shouldFail(UnrecoverableFileError, self.basedir,
181                                    "no recoverable versions",
182-                                   self._check_download)
183+                                   self._download_and_check)
184         else:
185             return self.shouldFail(NotEnoughSharesError, self.basedir,
186                                    "Failed to get enough shareholders",
187-                                   self._check_download)
188+                                   self._download_and_check)
189 
190 
191     def test_10_good_sanity_check(self):
192         d = defer.succeed(None)
193         for mutable in [False, True]:
194             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_sanity_check"))
195-            d.addCallback(lambda ign: self._check_download())
196+            d.addCallback(lambda ign: self._download_and_check())
197         return d
198 
199     def test_10_good_copied_share(self):
200@@ -116,7 +153,7 @@
201         for mutable in [False, True]:
202             d.addCallback(lambda ign: self._set_up(mutable, "test_10_good_copied_share"))
203             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
204-            d.addCallback(lambda ign: self._check_download())
205+            d.addCallback(lambda ign: self._download_and_check())
206             return d
207 
208     def test_3_good_7_noshares(self):
209@@ -124,7 +161,7 @@
210         for mutable in [False, True]:
211             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_noshares"))
212             d.addCallback(lambda ign: self._delete_all_shares_from(self.servers[3:]))
213-            d.addCallback(lambda ign: self._check_download())
214+            d.addCallback(lambda ign: self._download_and_check())
215         return d
216 
217     def test_2_good_8_broken_fail(self):
218@@ -149,7 +186,7 @@
219             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_broken_copied_share"))
220             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[2:3], self.servers[0]))
221             d.addCallback(lambda ign: self._break(self.servers[2:]))
222-            d.addCallback(lambda ign: self._check_download())
223+            d.addCallback(lambda ign: self._download_and_check())
224         return d
225 
226     def test_2_good_8_broken_duplicate_share_fail(self):
227@@ -168,7 +205,7 @@
228         for mutable in [False]:
229             d.addCallback(lambda ign: self._set_up(mutable, "test_3_good_7_hung"))
230             d.addCallback(lambda ign: self._hang(self.servers[3:]))
231-            d.addCallback(lambda ign: self._check_download())
232+            d.addCallback(lambda ign: self._download_and_check())
233         return d
234 
235     def test_2_good_8_hung_then_1_recovers(self):
236@@ -179,17 +216,44 @@
237             d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
238             d.addCallback(lambda ign: self._hang(self.servers[3:]))
239             d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
240-            d.addCallback(lambda ign: self._check_download())
241+            d.addCallback(lambda ign: self._download_and_check())
242         return d
243 
244     def test_2_good_8_hung_then_1_recovers_with_2_shares(self):
245         d = defer.succeed(None)
246         for mutable in [False]:
247-            recovered = defer.Deferred()
248+            #recovered = defer.Deferred()
249             d.addCallback(lambda ign: self._set_up(mutable, "test_2_good_8_hung_then_1_recovers_with_2_shares"))
250             d.addCallback(lambda ign: self._copy_all_shares_from(self.servers[0:1], self.servers[2]))
251-            d.addCallback(lambda ign: self._hang(self.servers[2:3], until=recovered))
252+            #d.addCallback(lambda ign: self._hang(self.servers[2:3]))
253             d.addCallback(lambda ign: self._hang(self.servers[3:]))
254-            d.addCallback(lambda ign: reactor.callLater(5, recovered.callback, None))
255-            d.addCallback(lambda ign: self._check_download())
256+            #d.addCallback(lambda ign: self._unhang(self.servers[2:3]))
257+            d.addCallback(lambda ign: self._download_and_check())
258         return d
259+
260+    def test_failover_during_stage_4(self):
261+        # See #287
262+        d = defer.succeed(None)
263+        for mutable in [False]:
264+            #recovered = defer.Deferred()
265+            d.addCallback(lambda ign: self._set_up(mutable, "test_failover_during_stage_4"))
266+            d.addCallback(lambda ign: self._corrupt_all_shares_in(self.servers[2:3], _corrupt_share_data))
267+            d.addCallback(lambda ign: self._hang(self.servers[3:]))
268+            d.addCallback(lambda ign: self._start_download())
269+            def _after_starting_download(ign):
270+                d2 = defer.succeed(None)
271+                started4d = self._stage_4_d
272+                print "xxx 0 %s" % (started4d,)
273+                def p(x):
274+                    print "xxx1 %s" % (x,)
275+                    return x
276+                started4d.addCallback(p)
277+                started4d.addCallback(lambda ign: self._unhang(self.servers[3:4]))
278+                def p2(x):
279+                    print "xxx2 %s" % (x,)
280+                    return x
281+                d2.addCallback(p2)
282+                d2.addCallback(self._check)
283+                return d2
284+            d.addCallback(_after_starting_download)
285+        return d