Plots for the CIFS SPECsfs results.
[bluesky.git] / cloudbench / rangetest.py
1 #!/usr/bin/python
2 #
3 # Run a series of simple test requests against S3 for gathering some basic
4 # performance numbers.
5
6 import boto, time
7 from boto.s3.connection import SubdomainCallingFormat
8 from boto.s3.key import Key
9 import sys, threading, time, Queue
10 import azure
11
12 BUCKET_NAME = 'mvrable-benchmark'
13 SIZES = [(1 << s) for s in range(12, 23)]
14
15 class S3TestConnection:
16     def __init__(self):
17         self.conn = boto.connect_s3(is_secure=False,
18                                     calling_format=SubdomainCallingFormat())
19         self.bucket = self.conn.get_bucket(BUCKET_NAME)
20
21     def put_object(self, name, size):
22         buf = 'A' * size
23         k = Key(self.bucket, name)
24         start_time = time.time()
25         k.set_contents_from_string(buf)
26         #print "%s: %f" % (name, time.time() - start_time)
27
28     def get_object(self, name, byterange=None):
29         k = Key(self.bucket, name)
30         headers = {}
31         if byterange is not None:
32             headers['Range'] = 'bytes=%s-%s' % byterange
33         start_time = time.time()
34         buf = k.get_contents_as_string(headers=headers)
35         duration = time.time() - start_time
36         return (duration, len(buf))
37
38 def parallel_get(name, connections, delay1=0.0):
39     #print "Get: %s x %d" % (name, len(connections))
40     threads = []
41     q = Queue.Queue()
42     def launcher(c, name, result_queue):
43         result_queue.put(c.get_object(name))
44     for i in range(len(connections)):
45         c = connections[i]
46         threads.append(threading.Thread(target=launcher, args=(c, name, q)))
47     for i in range(len(threads)):
48         threads[i].start()
49     for t in threads: t.join()
50     res = []
51     while not q.empty():
52         res.append(q.get())
53
54     if len(res) == len(connections):
55         return res
56
57 def parallel_multiget(names, connections, repeat=1):
58     requests = Queue.Queue()
59     results = [[threading.Lock(), None] for n in names]
60     for i in range(len(names)):
61         for _ in range(repeat):
62             requests.put((names[i], results[i]))
63
64     threads = []
65     def launcher(c, requests):
66         while True:
67             try:
68                 (n, r) = requests.get(block=False)
69                 # Possible data race here but it should be harmless
70                 if r[1] is None:
71                     res = c.get_object(n)
72                     r[0].acquire()
73                     if r[1] is None: r[1] = time.time()
74                     r[0].release()
75                 requests.task_done()
76             except Queue.Empty:
77                 return
78     for i in range(len(connections)):
79         c = connections[i]
80         threads.append(threading.Thread(target=launcher, args=(c, requests)))
81     start_time = time.time()
82     for i in range(len(threads)):
83         threads[i].start()
84     requests.join()
85
86     return max(x[1] for x in results) - start_time
87
88 def run_test(size, threads, num, logfile=sys.stdout, delay=1.0):
89     connections = [S3TestConnection() for _ in range(threads)]
90     for i in range(num):
91         print "    ...test", i
92         res = parallel_get('file-%d-%d' % (size, i), connections)
93         if res is not None:
94             logfile.write(str(min(res)) + "\n")
95         if delay > 0:
96             time.sleep(delay)
97
98 # Ranges are specified as a start and a length.  Fractional values are
99 # multiplied by the total size of the object.  Negative start values measure
100 # from the end of the object.
101 TESTRANGES = [[None, None],                 # Cold read, then hot read
102               [None, (0, 256)],             # Cold read, hot partial read
103               [(0, 256), None],             # Cold partial, hot full
104               [(-256, 256), None],          # Cold partial end, hot full
105               [(0, 256), (0, 256)],         # Repeated range
106               [(0, 256), (256, 256)],       # Consecutive ranges
107               [(0, 256), (-256, 256)],      # Discontiguous ranges
108               [(-256, 256), (0, 256)]]      # Discontiguous ranges
109
110 connection = S3TestConnection()
111 logfile = open('multifetch-simulation.data', 'a')
112 RANGE = 100
113 for s in SIZES:
114     for i in range(RANGE):
115         name = 'file-%d-%d' % (s, i)
116         for r in TESTRANGES[i * len(TESTRANGES) // RANGE]:
117             if r is not None:
118                 (x, y) = r
119                 if abs(x) < 1: x = int(x * s)
120                 if abs(y) < 1: x = int(y * s)
121                 if x < 0: x += s
122                 r = (x, x + y - 1)
123             t = connection.get_object(name, r)
124             print "%s %s: %s" % (name, r or "", t)
125
126 sys.exit(0)