<s> import json <EOL> import re <EOL> import sys <EOL> import uuid <EOL> import time <EOL> import random <EOL> import signal <EOL> import socket <EOL> import logging <EOL> from contextlib import contextmanager <EOL> from optparse import Values <EOL> import eventlet <EOL> import eventlet . pools <EOL> from eventlet . green . httplib import CannotSendRequest <EOL> import swiftclient as client <EOL> from swiftbench . utils import config_true_value , using_http_proxy <EOL> try : <EOL> from swift . common import direct_client <EOL> except ImportError : <EOL> direct_client = None <EOL> HTTP_CONFLICT = <NUM_LIT> <EOL> def _func_on_containers ( logger , conf , concurrency_key , func , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> bench = Bench ( logger , conf , [ ] ) <EOL> pool = eventlet . GreenPool ( int ( getattr ( conf , concurrency_key ) ) ) <EOL> for container in conf . containers : <EOL> pool . spawn_n ( func , bench . url , bench . token , container , ** kwargs ) <EOL> pool . waitall ( ) <EOL> def delete_containers ( logger , conf ) : <EOL> """<STR_LIT>""" <EOL> def _deleter ( url , token , container ) : <EOL> try : <EOL> client . delete_container ( url , token , container ) <EOL> except client . ClientException as e : <EOL> if e . http_status != HTTP_CONFLICT : <EOL> logger . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( container , e . http_status ) ) <EOL> _func_on_containers ( logger , conf , '<STR_LIT>' , _deleter ) <EOL> def create_containers ( logger , conf ) : <EOL> """<STR_LIT>""" <EOL> if conf . policy_name : <EOL> logger . info ( "<STR_LIT>" % <EOL> conf . policy_name ) <EOL> _func_on_containers ( logger , conf , '<STR_LIT>' , <EOL> client . put_container , <EOL> headers = { '<STR_LIT>' : conf . policy_name } ) <EOL> else : <EOL> _func_on_containers ( logger , conf , '<STR_LIT>' , <EOL> client . put_container ) <EOL> class SourceFile ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , size , chunk_size = <NUM_LIT> * <NUM_LIT:64> ) : <EOL> self . pos = <NUM_LIT:0> <EOL> self . size = size <EOL> self . chunk_size = chunk_size <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def __len__ ( self ) : <EOL> return self . size <EOL> def next ( self ) : <EOL> if self . pos >= self . size : <EOL> raise StopIteration <EOL> chunk_size = min ( self . size - self . pos , self . chunk_size ) <EOL> self . pos += chunk_size <EOL> return '<STR_LIT:0>' * chunk_size <EOL> def read ( self , desired_size ) : <EOL> chunk_size = min ( self . size - self . pos , desired_size ) <EOL> self . pos += chunk_size <EOL> return '<STR_LIT:0>' * chunk_size <EOL> class ConnectionPool ( eventlet . pools . Pool ) : <EOL> def __init__ ( self , url , size ) : <EOL> self . url = url <EOL> eventlet . pools . Pool . __init__ ( self , size , size ) <EOL> def create ( self ) : <EOL> return client . http_connection ( self . url ) <EOL> class BenchServer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , logger , bind_ip , bind_port ) : <EOL> self . logger = logger <EOL> self . bind_ip = bind_ip <EOL> self . bind_port = int ( bind_port ) <EOL> def run ( self ) : <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . logger . info ( '<STR_LIT>' , self . bind_ip , self . bind_port ) <EOL> s . bind ( ( self . bind_ip , self . bind_port ) ) <EOL> s . listen ( <NUM_LIT:20> ) <EOL> while True : <EOL> client , address = s . accept ( ) <EOL> self . logger . debug ( '<STR_LIT>' , * address ) <EOL> client_file = client . makefile ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> json_data = client_file . read ( ) <EOL> conf = Values ( json . loads ( json_data ) ) <EOL> self . logger . info ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , address [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] , <EOL> conf . put_concurrency , conf . get_concurrency , <EOL> conf . del_concurrency , conf . num_objects , conf . num_gets ) <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> level = logging . DEBUG if conf . log_level . lower ( ) == '<STR_LIT>' else logging . INFO <EOL> logger . setLevel ( level ) <EOL> loghandler = logging . StreamHandler ( client_file ) <EOL> logformat = logging . Formatter ( <EOL> '<STR_LIT>' ) <EOL> loghandler . setFormatter ( logformat ) <EOL> logger . addHandler ( loghandler ) <EOL> controller = BenchController ( logger , conf ) <EOL> try : <EOL> controller . run ( ) <EOL> except socket . error : <EOL> logger . warning ( '<STR_LIT>' , exc_info = <NUM_LIT:1> ) <EOL> logger . removeHandler ( loghandler ) <EOL> client_file . close ( ) <EOL> client . close ( ) <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> class Bench ( object ) : <EOL> def __init__ ( self , logger , conf , names ) : <EOL> self . logger = logger <EOL> self . aborted = False <EOL> self . user = conf . user <EOL> self . key = conf . key <EOL> self . auth_url = conf . auth <EOL> self . use_proxy = config_true_value ( conf . use_proxy ) <EOL> if not self . use_proxy and direct_client is None : <EOL> self . logger . critical ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> self . auth_version = conf . auth_version <EOL> self . logger . info ( "<STR_LIT>" % self . auth_version ) <EOL> if self . use_proxy : <EOL> if using_http_proxy ( self . auth_url ) : <EOL> logger . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> url , token = client . get_auth ( self . auth_url , self . user , self . key , <EOL> auth_version = self . auth_version ) <EOL> self . token = token <EOL> self . account = url . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] <EOL> if conf . url == '<STR_LIT>' : <EOL> self . url = url <EOL> else : <EOL> self . url = conf . url <EOL> else : <EOL> self . token = '<STR_LIT>' <EOL> self . account = conf . account <EOL> self . url = conf . url <EOL> self . ip , self . port = self . url . split ( '<STR_LIT:/>' ) [ <NUM_LIT:2> ] . split ( '<STR_LIT::>' ) <EOL> if using_http_proxy ( self . url ) : <EOL> logger . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . object_size = int ( conf . object_size ) <EOL> self . object_sources = conf . object_sources <EOL> self . lower_object_size = int ( conf . lower_object_size ) <EOL> self . upper_object_size = int ( conf . upper_object_size ) <EOL> self . files = [ ] <EOL> if self . object_sources : <EOL> self . object_sources = self . object_sources . split ( ) <EOL> self . files = [ file ( f , '<STR_LIT:rb>' ) . read ( ) for f in self . object_sources ] <EOL> self . put_concurrency = int ( conf . put_concurrency ) <EOL> self . get_concurrency = int ( conf . get_concurrency ) <EOL> self . del_concurrency = int ( conf . del_concurrency ) <EOL> self . total_objects = int ( conf . num_objects ) <EOL> self . total_gets = int ( conf . num_gets ) <EOL> self . timeout = int ( conf . timeout ) <EOL> self . devices = conf . devices . split ( ) <EOL> self . names = names <EOL> self . conn_pool = ConnectionPool ( self . url , <EOL> max ( self . put_concurrency , <EOL> self . get_concurrency , <EOL> self . del_concurrency ) ) <EOL> def _log_status ( self , title ) : <EOL> total = time . time ( ) - self . beginbeat <EOL> self . logger . info ( <EOL> '<STR_LIT>' , <EOL> { '<STR_LIT:title>' : title , '<STR_LIT>' : self . complete , <EOL> '<STR_LIT>' : self . failures , <EOL> '<STR_LIT>' : ( float ( self . complete ) / total ) } ) <EOL> @ contextmanager <EOL> def connection ( self ) : <EOL> try : <EOL> hc = self . conn_pool . get ( ) <EOL> try : <EOL> yield hc <EOL> except CannotSendRequest : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> try : <EOL> hc . close ( ) <EOL> except Exception : <EOL> pass <EOL> self . failures += <NUM_LIT:1> <EOL> hc = self . conn_pool . create ( ) <EOL> finally : <EOL> self . conn_pool . put ( hc ) <EOL> def run ( self ) : <EOL> pool = eventlet . GreenPool ( self . concurrency ) <EOL> self . beginbeat = self . heartbeat = time . time ( ) <EOL> self . heartbeat -= <NUM_LIT> <EOL> self . failures = <NUM_LIT:0> <EOL> self . complete = <NUM_LIT:0> <EOL> for i in xrange ( self . total ) : <EOL> if self . aborted : <EOL> break <EOL> pool . spawn_n ( self . _run , i ) <EOL> pool . waitall ( ) <EOL> self . _log_status ( self . msg + '<STR_LIT>' ) <EOL> def _run ( self , thread ) : <EOL> return <EOL> class DistributedBenchController ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , logger , conf ) : <EOL> self . logger = logger <EOL> self . final_re = re . compile ( <EOL> '<STR_LIT>' ) <EOL> self . clients = conf . bench_clients <EOL> del conf . bench_clients <EOL> for key , minval in [ ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) ] : <EOL> setattr ( conf , key , <EOL> max ( minval , int ( getattr ( conf , key ) ) / len ( self . clients ) ) ) <EOL> self . conf = conf <EOL> def run ( self ) : <EOL> eventlet . patcher . monkey_patch ( socket = True ) <EOL> pool = eventlet . GreenPool ( size = len ( self . clients ) ) <EOL> pile = eventlet . GreenPile ( pool ) <EOL> for c in self . clients : <EOL> pile . spawn ( self . do_run , c ) <EOL> results = { <EOL> '<STR_LIT>' : dict ( count = <NUM_LIT:0> , failures = <NUM_LIT:0> , rate = <NUM_LIT:0.0> ) , <EOL> '<STR_LIT>' : dict ( count = <NUM_LIT:0> , failures = <NUM_LIT:0> , rate = <NUM_LIT:0.0> ) , <EOL> '<STR_LIT>' : dict ( count = <NUM_LIT:0> , failures = <NUM_LIT:0> , rate = <NUM_LIT:0.0> ) , <EOL> } <EOL> for result in pile : <EOL> for k , v in result . iteritems ( ) : <EOL> target = results [ k ] <EOL> target [ '<STR_LIT:count>' ] += int ( v [ '<STR_LIT:count>' ] ) <EOL> target [ '<STR_LIT>' ] += int ( v [ '<STR_LIT>' ] ) <EOL> target [ '<STR_LIT>' ] += float ( v [ '<STR_LIT>' ] ) <EOL> for k in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> v = results [ k ] <EOL> self . logger . info ( '<STR_LIT>' % ( <EOL> v [ '<STR_LIT:count>' ] , k , v [ '<STR_LIT>' ] , v [ '<STR_LIT>' ] ) ) <EOL> def do_run ( self , client ) : <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> ip , port = client . split ( '<STR_LIT::>' ) <EOL> s . connect ( ( ip , int ( port ) ) ) <EOL> s . sendall ( json . dumps ( self . conf . __dict__ ) ) <EOL> s . shutdown ( socket . SHUT_WR ) <EOL> s_file = s . makefile ( '<STR_LIT:rb>' , <NUM_LIT:1> ) <EOL> result = { } <EOL> for line in s_file : <EOL> match = self . final_re . search ( line ) <EOL> if match : <EOL> g = match . groups ( ) <EOL> result [ g [ <NUM_LIT:1> ] ] = { <EOL> '<STR_LIT:count>' : g [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : g [ <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : g [ <NUM_LIT:3> ] , <EOL> } <EOL> else : <EOL> sys . stderr . write ( '<STR_LIT>' % ( client , line ) ) <EOL> return result <EOL> class BenchController ( object ) : <EOL> def __init__ ( self , logger , conf ) : <EOL> self . logger = logger <EOL> self . conf = conf <EOL> self . names = [ ] <EOL> self . delete = config_true_value ( conf . delete ) <EOL> self . gets = int ( conf . num_gets ) <EOL> self . aborted = False <EOL> self . delay = int ( self . conf . delay ) <EOL> def sigint1 ( self , signum , frame ) : <EOL> if self . delete : <EOL> print >> sys . stderr , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . aborted = True <EOL> if self . running and not isinstance ( self . running , BenchDELETE ) : <EOL> self . running . aborted = True <EOL> signal . signal ( signal . SIGINT , self . sigint2 ) <EOL> else : <EOL> self . sigint2 ( signum , frame ) <EOL> def sigint2 ( self , signum , frame ) : <EOL> sys . exit ( '<STR_LIT>' ) <EOL> def run ( self ) : <EOL> eventlet . patcher . monkey_patch ( socket = True ) <EOL> signal . signal ( signal . SIGINT , self . sigint1 ) <EOL> puts = BenchPUT ( self . logger , self . conf , self . names ) <EOL> self . running = puts <EOL> puts . run ( ) <EOL> if self . gets and not self . aborted : <EOL> gets = BenchGET ( self . logger , self . conf , self . names ) <EOL> self . running = gets <EOL> gets . run ( ) <EOL> if self . delete : <EOL> if self . delay != <NUM_LIT:0> : <EOL> self . logger . info ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % self . delay ) <EOL> time . sleep ( self . delay ) <EOL> dels = BenchDELETE ( self . logger , self . conf , self . names ) <EOL> self . running = dels <EOL> dels . run ( ) <EOL> class BenchDELETE ( Bench ) : <EOL> def __init__ ( self , logger , conf , names ) : <EOL> Bench . __init__ ( self , logger , conf , names ) <EOL> self . concurrency = self . del_concurrency <EOL> self . total = len ( names ) <EOL> self . msg = '<STR_LIT>' <EOL> def _run ( self , thread ) : <EOL> if time . time ( ) - self . heartbeat >= <NUM_LIT:15> : <EOL> self . heartbeat = time . time ( ) <EOL> self . _log_status ( '<STR_LIT>' ) <EOL> device , partition , name , container_name = self . names . pop ( ) <EOL> with self . connection ( ) as conn : <EOL> try : <EOL> if self . use_proxy : <EOL> client . delete_object ( self . url , self . token , <EOL> container_name , name , http_conn = conn ) <EOL> else : <EOL> node = { '<STR_LIT>' : self . ip , '<STR_LIT:port>' : self . port , '<STR_LIT>' : device } <EOL> direct_client . direct_delete_object ( node , partition , <EOL> self . account , <EOL> container_name , name ) <EOL> except client . ClientException as e : <EOL> self . logger . debug ( str ( e ) ) <EOL> self . failures += <NUM_LIT:1> <EOL> self . complete += <NUM_LIT:1> <EOL> class BenchGET ( Bench ) : <EOL> def __init__ ( self , logger , conf , names ) : <EOL> Bench . __init__ ( self , logger , conf , names ) <EOL> self . concurrency = self . get_concurrency <EOL> self . total = self . total_gets <EOL> self . msg = '<STR_LIT>' <EOL> def _run ( self , thread ) : <EOL> if time . time ( ) - self . heartbeat >= <NUM_LIT:15> : <EOL> self . heartbeat = time . time ( ) <EOL> self . _log_status ( '<STR_LIT>' ) <EOL> device , partition , name , container_name = random . choice ( self . names ) <EOL> with self . connection ( ) as conn : <EOL> try : <EOL> if self . use_proxy : <EOL> client . get_object ( self . url , self . token , <EOL> container_name , name , http_conn = conn ) <EOL> else : <EOL> node = { '<STR_LIT>' : self . ip , '<STR_LIT:port>' : self . port , '<STR_LIT>' : device } <EOL> direct_client . direct_get_object ( node , partition , <EOL> self . account , <EOL> container_name , name ) <EOL> except client . ClientException as e : <EOL> self . logger . debug ( str ( e ) ) <EOL> self . failures += <NUM_LIT:1> <EOL> self . complete += <NUM_LIT:1> <EOL> class BenchPUT ( Bench ) : <EOL> def __init__ ( self , logger , conf , names ) : <EOL> Bench . __init__ ( self , logger , conf , names ) <EOL> self . concurrency = self . put_concurrency <EOL> self . total = self . total_objects <EOL> self . msg = '<STR_LIT>' <EOL> self . containers = conf . containers <EOL> def _run ( self , thread ) : <EOL> if time . time ( ) - self . heartbeat >= <NUM_LIT:15> : <EOL> self . heartbeat = time . time ( ) <EOL> self . _log_status ( '<STR_LIT>' ) <EOL> name = uuid . uuid4 ( ) . hex <EOL> if self . object_sources : <EOL> source = random . choice ( self . files ) <EOL> elif self . upper_object_size > self . lower_object_size : <EOL> source = SourceFile ( random . randint ( self . lower_object_size , <EOL> self . upper_object_size ) ) <EOL> else : <EOL> source = SourceFile ( self . object_size ) <EOL> device = random . choice ( self . devices ) <EOL> partition = str ( random . randint ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> container_name = random . choice ( self . containers ) <EOL> with self . connection ( ) as conn : <EOL> try : <EOL> if self . use_proxy : <EOL> client . put_object ( self . url , self . token , <EOL> container_name , name , source , <EOL> content_length = len ( source ) , <EOL> http_conn = conn ) <EOL> else : <EOL> node = { '<STR_LIT>' : self . ip , '<STR_LIT:port>' : self . port , '<STR_LIT>' : device } <EOL> direct_client . direct_put_object ( node , partition , <EOL> self . account , <EOL> container_name , name , <EOL> source , <EOL> content_length = len ( source ) ) <EOL> except client . ClientException as e : <EOL> self . logger . debug ( str ( e ) ) <EOL> self . failures += <NUM_LIT:1> <EOL> else : <EOL> self . names . append ( ( device , partition , name , container_name ) ) <EOL> self . complete += <NUM_LIT:1> </s>
