<s> """<STR_LIT>""" <EOL> from time import time <EOL> from threading import Timer <EOL> from collections import defaultdict <EOL> from pprint import pprint <EOL> from ctypes import c_uint32 , byref , c_float <EOL> from . structures import SparseMap <EOL> from . lru import LRU <EOL> from . errors import ( GaugedAppendOnlyError , GaugedKeyOverflowError , <EOL> GaugedNaNError , GaugedUseAfterFreeError ) <EOL> from . bridge import Gauged <EOL> from . results import Statistics <EOL> from . utilities import to_bytes <EOL> class Writer ( object ) : <EOL> """<STR_LIT>""" <EOL> ERROR = <NUM_LIT:0> <EOL> IGNORE = <NUM_LIT:1> <EOL> REWRITE = <NUM_LIT:2> <EOL> KEY_OVERFLOW = - <NUM_LIT:1> <EOL> ALLOCATIONS = <NUM_LIT:0> <EOL> def __init__ ( self , driver , config ) : <EOL> self . driver = driver <EOL> self . config = config <EOL> self . key_cache = LRU ( config . key_cache_size ) <EOL> self . current_array = <NUM_LIT:0> <EOL> self . current_block = <NUM_LIT:0> <EOL> self . whitelist = None <EOL> self . flush_now = False <EOL> self . statistics = defaultdict ( Statistics ) <EOL> if config . flush_seconds : <EOL> self . start_flush_timer ( ) <EOL> self . flush_daemon = None <EOL> self . writer = None <EOL> self . allocate_writer ( ) <EOL> def add ( self , data , value = None , timestamp = None , namespace = None , <EOL> debug = False ) : <EOL> """<STR_LIT>""" <EOL> if value is not None : <EOL> return self . add ( ( ( data , value ) , ) , timestamp = timestamp , <EOL> namespace = namespace , debug = debug ) <EOL> writer = self . writer <EOL> if writer is None : <EOL> raise GaugedUseAfterFreeError <EOL> if timestamp is None : <EOL> timestamp = long ( time ( ) * <NUM_LIT:1000> ) <EOL> config = self . config <EOL> block_size = config . block_size <EOL> this_block = timestamp // block_size <EOL> this_array = ( timestamp % block_size ) // config . resolution <EOL> if namespace is None : <EOL> namespace = config . namespace <EOL> if this_block < self . current_block or ( this_block == self . current_block and <EOL> this_array < self . current_array ) : <EOL> if config . append_only_violation == Writer . ERROR : <EOL> msg = '<STR_LIT>' <EOL> raise GaugedAppendOnlyError ( msg ) <EOL> elif config . append_only_violation == Writer . REWRITE : <EOL> this_block = self . current_block <EOL> this_array = self . current_array <EOL> else : <EOL> return <EOL> if isinstance ( data , unicode ) : <EOL> data = data . encode ( '<STR_LIT:utf8>' ) <EOL> if debug : <EOL> return self . debug ( timestamp , namespace , data ) <EOL> if this_block > self . current_block : <EOL> self . flush_blocks ( ) <EOL> self . current_block = this_block <EOL> self . current_array = this_array <EOL> elif this_array > self . current_array : <EOL> if not Gauged . writer_flush_arrays ( writer , self . current_array ) : <EOL> raise MemoryError <EOL> self . current_array = this_array <EOL> data_points = <NUM_LIT:0> <EOL> namespace_statistics = self . statistics [ namespace ] <EOL> whitelist = config . key_whitelist <EOL> skip_long_keys = config . key_overflow == Writer . IGNORE <EOL> skip_gauge_nan = config . gauge_nan == Writer . IGNORE <EOL> if isinstance ( data , str ) and skip_gauge_nan and skip_long_keys and whitelist is None : <EOL> data_points = c_uint32 ( <NUM_LIT:0> ) <EOL> if not Gauged . writer_emit_pairs ( writer , namespace , data , <EOL> byref ( data_points ) ) : <EOL> raise MemoryError <EOL> data_points = data_points . value <EOL> else : <EOL> if isinstance ( data , dict ) : <EOL> data = data . iteritems ( ) <EOL> elif isinstance ( data , str ) : <EOL> data = self . parse_query ( data ) <EOL> emit = Gauged . writer_emit <EOL> for key , value in data : <EOL> key = to_bytes ( key ) <EOL> if whitelist is not None and key not in whitelist : <EOL> continue <EOL> try : <EOL> value = float ( value ) <EOL> except ValueError : <EOL> value = float ( '<STR_LIT>' ) <EOL> if value != value : <EOL> if skip_gauge_nan : <EOL> continue <EOL> raise GaugedNaNError <EOL> success = emit ( writer , namespace , key , c_float ( value ) ) <EOL> if success != <NUM_LIT:1> : <EOL> if not success : <EOL> raise MemoryError <EOL> elif success == Writer . KEY_OVERFLOW and not skip_long_keys : <EOL> msg = '<STR_LIT>' <EOL> msg += '<STR_LIT>' % key <EOL> raise GaugedKeyOverflowError ( msg ) <EOL> data_points += <NUM_LIT:1> <EOL> namespace_statistics . data_points += data_points <EOL> if self . flush_now : <EOL> self . flush ( ) <EOL> def flush ( self ) : <EOL> """<STR_LIT>""" <EOL> writer = self . writer <EOL> if writer is None : <EOL> raise GaugedUseAfterFreeError <EOL> self . flush_writer_position ( ) <EOL> keys = self . translate_keys ( ) <EOL> blocks = [ ] <EOL> current_block = self . current_block <EOL> statistics = self . statistics <EOL> driver = self . driver <EOL> flags = <NUM_LIT:0> <EOL> for namespace , key , block in self . pending_blocks ( ) : <EOL> length = block . byte_length ( ) <EOL> if not length : <EOL> continue <EOL> key_id = keys [ ( namespace , key ) ] <EOL> statistics [ namespace ] . byte_count += length <EOL> blocks . append ( ( namespace , current_block , key_id , block . buffer ( ) , <EOL> flags ) ) <EOL> if self . config . overwrite_blocks : <EOL> driver . replace_blocks ( blocks ) <EOL> else : <EOL> driver . insert_or_append_blocks ( blocks ) <EOL> if not Gauged . writer_flush_maps ( writer , True ) : <EOL> raise MemoryError <EOL> update_namespace = driver . add_namespace_statistics <EOL> for namespace , stats in statistics . iteritems ( ) : <EOL> update_namespace ( namespace , self . current_block , <EOL> stats . data_points , stats . byte_count ) <EOL> statistics . clear ( ) <EOL> driver . commit ( ) <EOL> self . flush_now = False <EOL> def resume_from ( self ) : <EOL> """<STR_LIT>""" <EOL> position = self . driver . get_writer_position ( self . config . writer_name ) <EOL> return position + self . config . resolution if position else <NUM_LIT:0> <EOL> def clear_from ( self , timestamp ) : <EOL> """<STR_LIT>""" <EOL> block_size = self . config . block_size <EOL> offset , remainder = timestamp // block_size , timestamp % block_size <EOL> if remainder : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . driver . clear_from ( offset , timestamp ) <EOL> def clear_key_before ( self , key , namespace = None , timestamp = None ) : <EOL> """<STR_LIT>""" <EOL> block_size = self . config . block_size <EOL> if namespace is None : <EOL> namespace = self . config . namespace <EOL> if timestamp is not None : <EOL> offset , remainder = divmod ( timestamp , block_size ) <EOL> if remainder : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if offset == <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> offset -= <NUM_LIT:1> <EOL> self . driver . clear_key_before ( key , namespace , offset , timestamp ) <EOL> else : <EOL> self . driver . clear_key_before ( key , namespace ) <EOL> def clear_key_after ( self , key , namespace = None , timestamp = None ) : <EOL> """<STR_LIT>""" <EOL> block_size = self . config . block_size <EOL> if namespace is None : <EOL> namespace = self . config . namespace <EOL> if timestamp is not None : <EOL> offset , remainder = divmod ( timestamp , block_size ) <EOL> if remainder : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . driver . clear_key_after ( key , namespace , offset , timestamp ) <EOL> else : <EOL> self . driver . clear_key_after ( key , namespace ) <EOL> def parse_query ( self , query ) : <EOL> """<STR_LIT>""" <EOL> writer = self . writer <EOL> if writer is None : <EOL> raise GaugedUseAfterFreeError <EOL> Gauged . writer_parse_query ( writer , query ) <EOL> position = <NUM_LIT:0> <EOL> writer_contents = writer . contents <EOL> size = writer_contents . buffer_size <EOL> pointers = writer_contents . buffer <EOL> while position < size : <EOL> yield pointers [ position ] , pointers [ position + <NUM_LIT:1> ] <EOL> position += <NUM_LIT:2> <EOL> def debug ( self , timestamp , namespace , data ) : <EOL> print '<STR_LIT>' % ( timestamp , namespace ) <EOL> if isinstance ( data , str ) : <EOL> data = self . parse_query ( data ) <EOL> elif isinstance ( data , dict ) : <EOL> data = data . iteritems ( ) <EOL> whitelist = self . config . key_whitelist <EOL> if whitelist is not None : <EOL> data_dict = { key : value for key , value in data if key in whitelist } <EOL> else : <EOL> data_dict = dict ( data ) <EOL> pprint ( data_dict ) <EOL> print '<STR_LIT>' <EOL> def flush_writer_position ( self ) : <EOL> config = self . config <EOL> timestamp = long ( self . current_block ) * config . block_size + long ( self . current_array ) * config . resolution <EOL> if timestamp : <EOL> self . driver . set_writer_position ( config . writer_name , timestamp ) <EOL> def translate_keys ( self ) : <EOL> keys = list ( self . pending_keys ( ) ) <EOL> if not len ( keys ) : <EOL> return { } <EOL> key_cache = self . key_cache <EOL> to_translate = [ key for key in keys if key not in key_cache ] <EOL> self . driver . insert_keys ( to_translate ) <EOL> ids = self . driver . lookup_ids ( to_translate ) <EOL> for key in keys : <EOL> if key in key_cache : <EOL> ids [ key ] = key_cache [ key ] <EOL> for key , id_ in ids . iteritems ( ) : <EOL> key_cache [ key ] = id_ <EOL> return ids <EOL> def flush_blocks ( self ) : <EOL> writer = self . writer <EOL> if not Gauged . writer_flush_arrays ( writer , self . current_array ) : <EOL> raise MemoryError <EOL> self . flush ( ) <EOL> if not Gauged . writer_flush_maps ( writer , False ) : <EOL> raise MemoryError <EOL> def start_flush_timer ( self ) : <EOL> period = self . config . flush_seconds <EOL> self . flush_daemon = Timer ( period , self . flush_timer_tick ) <EOL> self . flush_daemon . setDaemon ( True ) <EOL> self . flush_daemon . start ( ) <EOL> def flush_timer_tick ( self ) : <EOL> self . flush_now = True <EOL> self . start_flush_timer ( ) <EOL> def pending_keys ( self ) : <EOL> pending = self . writer . contents . pending . contents <EOL> head = pending . head <EOL> while True : <EOL> if not head : <EOL> break <EOL> node = head . contents <EOL> yield node . namespace , node . key <EOL> head = node . next <EOL> def pending_blocks ( self ) : <EOL> block = SparseMap ( ) <EOL> initial_ptr = block . _ptr <EOL> pending = self . writer . contents . pending . contents <EOL> head = pending . head <EOL> while True : <EOL> if not head : <EOL> break <EOL> node = head . contents <EOL> block . _ptr = node . map <EOL> yield node . namespace , node . key , block <EOL> head = node . next <EOL> block . _ptr = initial_ptr <EOL> block . free ( ) <EOL> def allocate_writer ( self ) : <EOL> self . current_array = <NUM_LIT:0> <EOL> self . current_block = <NUM_LIT:0> <EOL> if self . writer is None : <EOL> self . writer = Gauged . writer_new ( self . driver . MAX_KEY or <NUM_LIT:0> ) <EOL> Writer . ALLOCATIONS += <NUM_LIT:1> <EOL> if not self . writer : <EOL> raise MemoryError <EOL> def cleanup ( self ) : <EOL> if self . flush_daemon is not None : <EOL> self . flush_daemon . cancel ( ) <EOL> self . flush_daemon = None <EOL> self . flush_blocks ( ) <EOL> Gauged . writer_free ( self . writer ) <EOL> Writer . ALLOCATIONS -= <NUM_LIT:1> <EOL> self . writer = None <EOL> self . key_cache . clear ( ) <EOL> def __enter__ ( self ) : <EOL> self . allocate_writer ( ) <EOL> return self <EOL> def __exit__ ( self , type_ , value , traceback ) : <EOL> self . cleanup ( ) </s>
