cloudbench/s3pipeline.py

   1 #!/usr/bin/python2.6
   2 #
   3 # Copyright (C) 2010  The Regents of the University of California
   4 # Written by Michael Vrable <mvrable@cs.ucsd.edu>
   5 #
   6 # Redistribution and use in source and binary forms, with or without
   7 # modification, are permitted provided that the following conditions
   8 # are met:
   9 # 1. Redistributions of source code must retain the above copyright
  10 #    notice, this list of conditions and the following disclaimer.
  11 # 2. Redistributions in binary form must reproduce the above copyright
  12 #    notice, this list of conditions and the following disclaimer in the
  13 #    documentation and/or other materials provided with the distribution.
  14 # 3. Neither the name of the University nor the names of its contributors
  15 #    may be used to endorse or promote products derived from this software
  16 #    without specific prior written permission.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
  19 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  20 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  21 # ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  22 # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  23 # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  24 # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  25 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  27 # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  28 # SUCH DAMAGE.
  29
  30 import boto, httplib, socket, sys, time
  31 from boto.s3.key import Key
  32 from threading import Thread
  33
  34 bucket_name = 'mvrable-benchmark'
  35 conn = boto.connect_s3(is_secure=False)
  36
  37 class HttpResponseParser:
  38     def __init__(self):
  39         self.buf = ""
  40         self.header = True
  41         self.content_length = None
  42
  43     def process(self, data):
  44         self.buf = self.buf + data
  45         while self.parse():
  46             pass
  47
  48     def parse(self):
  49         if len(self.buf) == 0:
  50             return False
  51
  52         if not self.header and self.content_length is not None:
  53             consumed = min(self.content_length, len(self.buf))
  54             #print "Got", consumed, "bytes of data"
  55             self.content_length -= consumed
  56             self.buf = self.buf[consumed:]
  57
  58             if self.content_length == 0:
  59                 print "Completed reading body"
  60                 self.content_length = None
  61                 self.header = True
  62
  63             return True
  64
  65         crlf = self.buf.find("\r\n")
  66         if crlf < 0:
  67             return False
  68
  69         line = self.buf[0:crlf]
  70         self.buf = self.buf[crlf+2:]
  71         #print "Header line:", line
  72         if line.lower().startswith('content-length'):
  73             self.content_length = int(line[16:])
  74             print "Content length:", self.content_length
  75         if line == "":
  76             self.header = False
  77         return True
  78
  79 class PipelinedRequester:
  80     def __init__(self, bucket):
  81         self.bucket = bucket
  82         self.host = conn.calling_format.build_host(conn.server_name(), bucket)
  83         self.sock = socket.create_connection((self.host, 80))
  84
  85         t = Thread(target=self.reader_thread)
  86         t.setDaemon(True)
  87         t.start()
  88
  89     def reader_thread(self):
  90         hrp = HttpResponseParser()
  91         while True:
  92             buf = self.sock.recv(4096)
  93             if len(buf) == 0: break
  94             hrp.process(buf)
  95
  96     def send_request(self, key):
  97         method = 'GET'
  98         path = conn.calling_format.build_path_base(self.bucket, key)
  99         auth_path = conn.calling_format.build_auth_path(self.bucket, key)
 100
 101         headers = {'User-Agent': boto.UserAgent + " (pipelined)",
 102                    'Content-Length': str(0)}
 103         conn.add_aws_auth_header(headers, method, auth_path)
 104
 105         req = "%s %s HTTP/1.1\r\nHost: %s\r\n" % (method, path, self.host)
 106         req = req + ''.join("%s: %s\r\n" % h for h in headers.items()) + "\r\n"
 107         self.sock.sendall(req)
 108
 109 requester = PipelinedRequester(bucket_name)
 110 for i in range(12, 18):
 111     requester.send_request('file-%d-1' % (1 << i,))
 112     if i == 12:
 113         time.sleep(2)
 114 for i in range(32):
 115     requester.send_request('file-8192-%d' % (i,))
 116
 117 time.sleep(5)