source file: /home/buildslave/tahoe/edgy/build/src/allmydata/control.py
file stats: 169 lines, 150 executed: 88.8% covered
coverage versus previous test: 0 lines added, 0 lines removed
1.
2. import os, time
3. from zope.interface import implements
4. from twisted.application import service
5. from twisted.internet import defer
6. from foolscap.api import Referenceable
7. from allmydata.interfaces import RIControlClient
8. from allmydata.util import fileutil, mathutil
9. from allmydata.immutable import upload, download
10. from twisted.python import log
11.
12. def get_memory_usage():
13. # this is obviously linux-specific
14. stat_names = ("VmPeak",
15. "VmSize",
16. #"VmHWM",
17. "VmData")
18. stats = {}
19. try:
20. for line in open("/proc/self/status", "r").readlines():
21. name, right = line.split(":",2)
22. if name in stat_names:
23. assert right.endswith(" kB\n")
24. right = right[:-4]
25. stats[name] = int(right) * 1024
26. except:
27. # Probably not on (a compatible version of) Linux
28. stats['VmSize'] = 0
29. stats['VmPeak'] = 0
30. return stats
31.
32. def log_memory_usage(where=""):
33. stats = get_memory_usage()
34. log.msg("VmSize: %9d VmPeak: %9d %s" % (stats["VmSize"],
35. stats["VmPeak"],
36. where))
37.
38.
39. class ControlServer(Referenceable, service.Service):
40. implements(RIControlClient)
41.
42. def remote_wait_for_client_connections(self, num_clients):
43. return self.parent.debug_wait_for_client_connections(num_clients)
44.
45. def remote_upload_from_file_to_uri(self, filename, convergence):
46. uploader = self.parent.getServiceNamed("uploader")
47. u = upload.FileName(filename, convergence=convergence)
48. d = uploader.upload(u)
49. d.addCallback(lambda results: results.uri)
50. return d
51.
52. def remote_download_from_uri_to_file(self, uri, filename):
53. d = self.parent.downloader.download_to_filename(uri, filename)
54. d.addCallback(lambda res: filename)
55. return d
56.
57. def remote_speed_test(self, count, size, mutable):
58. assert size > 8
59. log.msg("speed_test: count=%d, size=%d, mutable=%s" % (count, size,
60. mutable))
61. st = SpeedTest(self.parent, count, size, mutable)
62. return st.run()
63.
64. def remote_get_memory_usage(self):
65. return get_memory_usage()
66.
67. def remote_measure_peer_response_time(self):
68. # I'd like to average together several pings, but I don't want this
69. # phase to take more than 10 seconds. Expect worst-case latency to be
70. # 300ms.
71. results = {}
72. sb = self.parent.get_storage_broker()
73. everyone = sb.get_all_servers()
74. num_pings = int(mathutil.div_ceil(10, (len(everyone) * 0.3)))
75. everyone = list(everyone) * num_pings
76. d = self._do_one_ping(None, everyone, results)
77. return d
78. def _do_one_ping(self, res, everyone_left, results):
79. if not everyone_left:
80. return results
81. peerid, connection = everyone_left.pop(0)
82. start = time.time()
83. d = connection.callRemote("get_buckets", "\x00"*16)
84. def _done(ignored):
85. stop = time.time()
86. elapsed = stop - start
87. if peerid in results:
88. results[peerid].append(elapsed)
89. else:
90. results[peerid] = [elapsed]
91. d.addCallback(_done)
92. d.addCallback(self._do_one_ping, everyone_left, results)
93. def _average(res):
94. averaged = {}
95. for peerid,times in results.iteritems():
96. averaged[peerid] = sum(times) / len(times)
97. return averaged
98. d.addCallback(_average)
99. return d
100.
101. class SpeedTest:
102. def __init__(self, parent, count, size, mutable):
103. self.parent = parent
104. self.count = count
105. self.size = size
106. self.mutable_mode = mutable
107. self.uris = {}
108. self.basedir = os.path.join(self.parent.basedir, "_speed_test_data")
109.
110. def run(self):
111. self.create_data()
112. d = self.do_upload()
113. d.addCallback(lambda res: self.do_download())
114. d.addBoth(self.do_cleanup)
115. d.addCallback(lambda res: (self.upload_time, self.download_time))
116. return d
117.
118. def create_data(self):
119. fileutil.make_dirs(self.basedir)
120. for i in range(self.count):
121. s = self.size
122. fn = os.path.join(self.basedir, str(i))
123. if os.path.exists(fn):
124. os.unlink(fn)
125. f = open(fn, "w")
126. f.write(os.urandom(8))
127. s -= 8
128. while s > 0:
129. chunk = min(s, 4096)
130. f.write("\x00" * chunk)
131. s -= chunk
132. f.close()
133.
134. def do_upload(self):
135. d = defer.succeed(None)
136. def _create_slot(res):
137. d1 = self.parent.create_mutable_file("")
138. def _created(n):
139. self._n = n
140. d1.addCallback(_created)
141. return d1
142. if self.mutable_mode == "upload":
143. d.addCallback(_create_slot)
144. def _start(res):
145. self._start = time.time()
146. d.addCallback(_start)
147.
148. def _record_uri(uri, i):
149. self.uris[i] = uri
150. def _upload_one_file(ignored, i):
151. if i >= self.count:
152. return
153. fn = os.path.join(self.basedir, str(i))
154. if self.mutable_mode == "create":
155. data = open(fn,"rb").read()
156. d1 = self.parent.create_mutable_file(data)
157. d1.addCallback(lambda n: n.get_uri())
158. elif self.mutable_mode == "upload":
159. data = open(fn,"rb").read()
160. d1 = self._n.overwrite(data)
161. d1.addCallback(lambda res: self._n.get_uri())
162. else:
163. up = upload.FileName(fn, convergence=None)
164. d1 = self.parent.upload(up)
165. d1.addCallback(lambda results: results.uri)
166. d1.addCallback(_record_uri, i)
167. d1.addCallback(_upload_one_file, i+1)
168. return d1
169. d.addCallback(_upload_one_file, 0)
170. def _upload_done(ignored):
171. stop = time.time()
172. self.upload_time = stop - self._start
173. d.addCallback(_upload_done)
174. return d
175.
176. def do_download(self):
177. start = time.time()
178. d = defer.succeed(None)
179. def _download_one_file(ignored, i):
180. if i >= self.count:
181. return
182. n = self.parent.create_node_from_uri(self.uris[i])
183. d1 = n.download(download.FileHandle(Discard()))
184. d1.addCallback(_download_one_file, i+1)
185. return d1
186. d.addCallback(_download_one_file, 0)
187. def _download_done(ignored):
188. stop = time.time()
189. self.download_time = stop - start
190. d.addCallback(_download_done)
191. return d
192.
193. def do_cleanup(self, res):
194. for i in range(self.count):
195. fn = os.path.join(self.basedir, str(i))
196. os.unlink(fn)
197. return res
198.
199. class Discard:
200. def write(self, data):
201. pass
202. # download_to_filehandle explicitly does not close the filehandle it was
203. # given: that is reserved for the provider of the filehandle. Therefore
204. # the lack of a close() method on this otherwise filehandle-like object
205. # is a part of the test.