<s> """<STR_LIT>""" <EOL> __authors__ = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> __license__ = "<STR_LIT>" <EOL> __contact__ = "<STR_LIT>" <EOL> import copy <EOL> try : <EOL> import dill as cPickle <EOL> except ImportError : <EOL> import cPickle <EOL> import hashlib <EOL> import logging <EOL> import optparse <EOL> import os <EOL> import shutil <EOL> import signal <EOL> import socket <EOL> import subprocess <EOL> import sys <EOL> import time <EOL> import urlparse <EOL> import warnings <EOL> import numpy <EOL> import pymongo <EOL> import gridfs <EOL> from bson import SON <EOL> logger = logging . getLogger ( __name__ ) <EOL> from . base import JOB_STATES <EOL> from . base import ( JOB_STATE_NEW , JOB_STATE_RUNNING , JOB_STATE_DONE , <EOL> JOB_STATE_ERROR ) <EOL> from . base import Trials <EOL> from . base import trials_from_docs <EOL> from . base import InvalidTrial <EOL> from . base import Ctrl <EOL> from . base import SONify <EOL> from . base import spec_from_misc <EOL> from . utils import coarse_utcnow <EOL> from . utils import fast_isin <EOL> from . utils import get_most_recent_inds <EOL> from . utils import json_call <EOL> from . utils import working_dir , temp_dir <EOL> import plotting <EOL> class OperationFailure ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class Shutdown ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class WaitQuit ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class InvalidMongoTrial ( InvalidTrial ) : <EOL> pass <EOL> class DomainSwapError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class ReserveTimeout ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def read_pw ( ) : <EOL> username = '<STR_LIT>' <EOL> password = open ( os . path . join ( os . getenv ( '<STR_LIT>' ) , "<STR_LIT>" ) ) . read ( ) [ : - <NUM_LIT:1> ] <EOL> return dict ( <EOL> username = username , <EOL> password = password ) <EOL> def authenticate_for_db ( db ) : <EOL> d = read_pw ( ) <EOL> db . authenticate ( d [ '<STR_LIT:username>' ] , d [ '<STR_LIT:password>' ] ) <EOL> def parse_url ( url , pwfile = None ) : <EOL> """<STR_LIT>""" <EOL> protocol = url [ : url . find ( '<STR_LIT::>' ) ] <EOL> ftp_url = '<STR_LIT>' + url [ url . find ( '<STR_LIT::>' ) : ] <EOL> tmp = urlparse . urlparse ( ftp_url ) <EOL> logger . info ( '<STR_LIT>' % protocol ) <EOL> logger . info ( '<STR_LIT>' % tmp . username ) <EOL> logger . info ( '<STR_LIT>' % tmp . hostname ) <EOL> logger . info ( '<STR_LIT>' % tmp . port ) <EOL> logger . info ( '<STR_LIT>' % tmp . path ) <EOL> try : <EOL> _ , dbname , collection = tmp . path . split ( '<STR_LIT:/>' ) <EOL> except : <EOL> print >> sys . stderr , "<STR_LIT>" % ( str ( tmp . path ) ) <EOL> raise <EOL> logger . info ( '<STR_LIT>' % dbname ) <EOL> logger . info ( '<STR_LIT>' % collection ) <EOL> if tmp . password is None : <EOL> if ( tmp . username is not None ) and pwfile : <EOL> password = open ( pwfile ) . read ( ) [ : - <NUM_LIT:1> ] <EOL> else : <EOL> password = None <EOL> else : <EOL> password = tmp . password <EOL> logger . info ( '<STR_LIT>' % password ) <EOL> return ( protocol , tmp . username , password , tmp . hostname , tmp . port , dbname , <EOL> collection ) <EOL> def connection_with_tunnel ( host = '<STR_LIT:localhost>' , <EOL> auth_dbname = '<STR_LIT>' , port = <NUM_LIT> , <EOL> ssh = False , user = '<STR_LIT>' , pw = None ) : <EOL> if ssh : <EOL> local_port = numpy . random . randint ( low = <NUM_LIT> , high = <NUM_LIT> ) <EOL> ssh_tunnel = subprocess . Popen ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' % ( local_port , '<STR_LIT:127.0.0.1>' , port ) , <EOL> host ] , <EOL> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> connection = pymongo . Connection ( '<STR_LIT:127.0.0.1>' , local_port , <EOL> document_class = SON ) <EOL> else : <EOL> connection = pymongo . Connection ( host , port , document_class = SON ) <EOL> if user : <EOL> if user == '<STR_LIT>' : <EOL> authenticate_for_db ( connection [ auth_dbname ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> ssh_tunnel = None <EOL> connection . write_concern [ '<STR_LIT:w>' ] = <NUM_LIT:1> <EOL> connection . write_concern [ '<STR_LIT>' ] = True <EOL> return connection , ssh_tunnel <EOL> def connection_from_string ( s ) : <EOL> protocol , user , pw , host , port , db , collection = parse_url ( s ) <EOL> if protocol == '<STR_LIT>' : <EOL> ssh = False <EOL> elif protocol in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> ssh = True <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' , protocol ) <EOL> connection , tunnel = connection_with_tunnel ( <EOL> ssh = ssh , <EOL> user = user , <EOL> pw = pw , <EOL> host = host , <EOL> port = port , <EOL> ) <EOL> return connection , tunnel , connection [ db ] , connection [ db ] [ collection ] <EOL> class MongoJobs ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , jobs , gfs , conn , tunnel , config_name ) : <EOL> """<STR_LIT>""" <EOL> self . db = db <EOL> self . jobs = jobs <EOL> assert jobs . write_concern [ '<STR_LIT:w>' ] >= <NUM_LIT:1> <EOL> self . gfs = gfs <EOL> self . conn = conn <EOL> self . tunnel = tunnel <EOL> self . config_name = config_name <EOL> coll = property ( lambda s : s . jobs ) <EOL> @ classmethod <EOL> def alloc ( cls , dbname , host = '<STR_LIT:localhost>' , <EOL> auth_dbname = '<STR_LIT>' , port = <NUM_LIT> , <EOL> jobs_coll = '<STR_LIT>' , gfs_coll = '<STR_LIT>' , ssh = False , user = None , pw = None ) : <EOL> connection , tunnel = connection_with_tunnel ( <EOL> host , auth_dbname , port , ssh , user , pw ) <EOL> db = connection [ dbname ] <EOL> gfs = gridfs . GridFS ( db , collection = gfs_coll ) <EOL> return cls ( db , db [ jobs_coll ] , gfs , connection , tunnel ) <EOL> @ classmethod <EOL> def new_from_connection_str ( cls , conn_str , gfs_coll = '<STR_LIT>' , config_name = '<STR_LIT>' ) : <EOL> connection , tunnel , db , coll = connection_from_string ( conn_str ) <EOL> gfs = gridfs . GridFS ( db , collection = gfs_coll ) <EOL> return cls ( db , coll , gfs , connection , tunnel , config_name ) <EOL> def __iter__ ( self ) : <EOL> return self . jobs . find ( ) <EOL> def __len__ ( self ) : <EOL> try : <EOL> return self . jobs . count ( ) <EOL> except : <EOL> return <NUM_LIT:0> <EOL> def create_jobs_indexes ( self ) : <EOL> jobs = self . db . jobs <EOL> for k in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> jobs . create_index ( k ) <EOL> def create_drivers_indexes ( self ) : <EOL> drivers = self . db . drivers <EOL> drivers . create_index ( '<STR_LIT>' , unique = True ) <EOL> def create_indexes ( self ) : <EOL> self . create_jobs_indexes ( ) <EOL> self . create_drivers_indexes ( ) <EOL> def jobs_complete ( self , cursor = False ) : <EOL> c = self . jobs . find ( spec = dict ( state = JOB_STATE_DONE ) ) <EOL> return c if cursor else list ( c ) <EOL> def jobs_error ( self , cursor = False ) : <EOL> c = self . jobs . find ( spec = dict ( state = JOB_STATE_ERROR ) ) <EOL> return c if cursor else list ( c ) <EOL> def jobs_running ( self , cursor = False ) : <EOL> if cursor : <EOL> raise NotImplementedError ( ) <EOL> rval = list ( self . jobs . find ( spec = dict ( state = JOB_STATE_RUNNING ) ) ) <EOL> rval = [ r for r in rval if not r . get ( '<STR_LIT>' , False ) ] <EOL> return rval <EOL> def jobs_dead ( self , cursor = False ) : <EOL> if cursor : <EOL> raise NotImplementedError ( ) <EOL> rval = list ( self . jobs . find ( spec = dict ( state = JOB_STATE_RUNNING ) ) ) <EOL> rval = [ r for r in rval if r . get ( '<STR_LIT>' , False ) ] <EOL> return rval <EOL> def jobs_queued ( self , cursor = False ) : <EOL> c = self . jobs . find ( spec = dict ( state = JOB_STATE_NEW ) ) <EOL> return c if cursor else list ( c ) <EOL> def insert ( self , job ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> cpy = copy . deepcopy ( job ) <EOL> _id = self . jobs . insert ( cpy , check_keys = True ) <EOL> assert _id == cpy [ '<STR_LIT>' ] <EOL> return cpy <EOL> except pymongo . errors . OperationFailure , e : <EOL> raise OperationFailure ( e ) <EOL> def delete ( self , job ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . jobs . remove ( job ) <EOL> except pymongo . errors . OperationFailure , e : <EOL> raise OperationFailure ( e ) <EOL> def delete_all ( self , cond = None ) : <EOL> """<STR_LIT>""" <EOL> if cond is None : <EOL> cond = { } <EOL> try : <EOL> for d in self . jobs . find ( spec = cond , fields = [ '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> logger . info ( '<STR_LIT>' % d [ '<STR_LIT>' ] ) <EOL> for name , file_id in d . get ( '<STR_LIT>' , [ ] ) : <EOL> try : <EOL> self . gfs . delete ( file_id ) <EOL> except gridfs . errors . NoFile : <EOL> logger . error ( '<STR_LIT>' % ( <EOL> name , file_id ) ) <EOL> self . jobs . remove ( d ) <EOL> except pymongo . errors . OperationFailure , e : <EOL> raise OperationFailure ( e ) <EOL> def delete_all_error_jobs ( self ) : <EOL> return self . delete_all ( cond = { '<STR_LIT:state>' : JOB_STATE_ERROR } ) <EOL> def reserve ( self , host_id , cond = None , exp_key = None ) : <EOL> now = coarse_utcnow ( ) <EOL> if cond is None : <EOL> cond = { } <EOL> else : <EOL> cond = copy . copy ( cond ) <EOL> if exp_key is not None : <EOL> cond [ '<STR_LIT>' ] = exp_key <EOL> if cond . get ( '<STR_LIT>' ) is not None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> else : <EOL> cond [ '<STR_LIT>' ] = None <EOL> cond [ '<STR_LIT:state>' ] = JOB_STATE_NEW <EOL> try : <EOL> rval = self . jobs . find_and_modify ( <EOL> cond , <EOL> { '<STR_LIT>' : <EOL> { '<STR_LIT>' : host_id , <EOL> '<STR_LIT>' : now , <EOL> '<STR_LIT:state>' : JOB_STATE_RUNNING , <EOL> '<STR_LIT>' : now , <EOL> } <EOL> } , <EOL> new = True , <EOL> upsert = False ) <EOL> except pymongo . errors . OperationFailure , e : <EOL> logger . error ( '<STR_LIT>' % str ( e ) ) <EOL> rval = None <EOL> return rval <EOL> def refresh ( self , doc ) : <EOL> self . update ( doc , dict ( refresh_time = coarse_utcnow ( ) ) ) <EOL> def update ( self , doc , dct , collection = None , do_sanity_checks = True ) : <EOL> """<STR_LIT>""" <EOL> if collection is None : <EOL> collection = self . coll <EOL> dct = copy . deepcopy ( dct ) <EOL> if '<STR_LIT>' not in doc : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in dct : <EOL> if dct [ '<STR_LIT>' ] != doc [ '<STR_LIT>' ] : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> del dct [ '<STR_LIT>' ] <EOL> if '<STR_LIT:version>' in dct : <EOL> if dct [ '<STR_LIT:version>' ] != doc [ '<STR_LIT:version>' ] : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> if '<STR_LIT:version>' in doc : <EOL> doc_query = dict ( _id = doc [ '<STR_LIT>' ] , version = doc [ '<STR_LIT:version>' ] ) <EOL> dct [ '<STR_LIT:version>' ] = doc [ '<STR_LIT:version>' ] + <NUM_LIT:1> <EOL> else : <EOL> doc_query = dict ( _id = doc [ '<STR_LIT>' ] ) <EOL> dct [ '<STR_LIT:version>' ] = <NUM_LIT:1> <EOL> try : <EOL> collection . update ( <EOL> doc_query , <EOL> { '<STR_LIT>' : dct } , <EOL> upsert = False , <EOL> multi = False , ) <EOL> except pymongo . errors . OperationFailure , e : <EOL> raise OperationFailure ( e ) <EOL> doc . update ( dct ) <EOL> if do_sanity_checks : <EOL> server_doc = collection . find_one ( <EOL> dict ( _id = doc [ '<STR_LIT>' ] , version = doc [ '<STR_LIT:version>' ] ) ) <EOL> if server_doc is None : <EOL> raise OperationFailure ( '<STR_LIT>' <EOL> % str ( doc ) ) <EOL> elif server_doc != doc : <EOL> if <NUM_LIT:0> : <EOL> mismatching_keys = [ ] <EOL> for k , v in server_doc . items ( ) : <EOL> if k in doc : <EOL> if doc [ k ] != v : <EOL> mismatching_keys . append ( ( k , v , doc [ k ] ) ) <EOL> else : <EOL> mismatching_keys . append ( ( k , v , '<STR_LIT>' ) ) <EOL> for k , v in doc . items ( ) : <EOL> if k not in server_doc : <EOL> mismatching_keys . append ( ( k , '<STR_LIT>' , v ) ) <EOL> raise OperationFailure ( '<STR_LIT>' % <EOL> repr ( ( doc , server_doc , mismatching_keys ) ) ) <EOL> return doc <EOL> def attachment_names ( self , doc ) : <EOL> def as_str ( name_id ) : <EOL> assert isinstance ( name_id [ <NUM_LIT:0> ] , basestring ) , name_id <EOL> return str ( name_id [ <NUM_LIT:0> ] ) <EOL> return map ( as_str , doc . get ( '<STR_LIT>' , [ ] ) ) <EOL> def set_attachment ( self , doc , blob , name , collection = None ) : <EOL> """<STR_LIT>""" <EOL> attachments = doc . get ( '<STR_LIT>' , [ ] ) <EOL> name_matches = [ a for a in attachments if a [ <NUM_LIT:0> ] == name ] <EOL> new_file_id = self . gfs . put ( blob , filename = '<STR_LIT>' % ( doc [ '<STR_LIT>' ] , name ) ) <EOL> logger . info ( '<STR_LIT>' % ( <EOL> len ( blob ) , str ( new_file_id ) , doc [ '<STR_LIT>' ] , name ) ) <EOL> new_attachments = ( [ a for a in attachments if a [ <NUM_LIT:0> ] != name ] <EOL> + [ ( name , new_file_id ) ] ) <EOL> try : <EOL> ii = <NUM_LIT:0> <EOL> doc = self . update ( doc , { '<STR_LIT>' : new_attachments } , <EOL> collection = collection ) <EOL> while ii < len ( name_matches ) : <EOL> self . gfs . delete ( name_matches [ ii ] [ <NUM_LIT:1> ] ) <EOL> ii += <NUM_LIT:1> <EOL> except : <EOL> while ii < len ( name_matches ) : <EOL> logger . warning ( "<STR_LIT>" % ( <EOL> name_matches [ ii ] [ <NUM_LIT:1> ] ) ) <EOL> ii += <NUM_LIT:1> <EOL> raise <EOL> assert len ( [ n for n in self . attachment_names ( doc ) if n == name ] ) == <NUM_LIT:1> <EOL> def get_attachment ( self , doc , name ) : <EOL> """<STR_LIT>""" <EOL> attachments = doc . get ( '<STR_LIT>' , [ ] ) <EOL> file_ids = [ a [ <NUM_LIT:1> ] for a in attachments if a [ <NUM_LIT:0> ] == name ] <EOL> if not file_ids : <EOL> raise OperationFailure ( '<STR_LIT>' % name ) <EOL> if len ( file_ids ) > <NUM_LIT:1> : <EOL> raise OperationFailure ( '<STR_LIT>' , ( name , file_ids ) ) <EOL> return self . gfs . get ( file_ids [ <NUM_LIT:0> ] ) . read ( ) <EOL> def delete_attachment ( self , doc , name , collection = None ) : <EOL> attachments = doc . get ( '<STR_LIT>' , [ ] ) <EOL> file_id = None <EOL> for i , a in enumerate ( attachments ) : <EOL> if a [ <NUM_LIT:0> ] == name : <EOL> file_id = a [ <NUM_LIT:1> ] <EOL> break <EOL> if file_id is None : <EOL> raise OperationFailure ( '<STR_LIT>' % name ) <EOL> del attachments [ i ] <EOL> self . update ( doc , { '<STR_LIT>' : attachments } , collection = collection ) <EOL> self . gfs . delete ( file_id ) <EOL> class MongoTrials ( Trials ) : <EOL> """<STR_LIT>""" <EOL> async = True <EOL> def __init__ ( self , arg , exp_key = None , cmd = None , workdir = None , <EOL> refresh = True ) : <EOL> if isinstance ( arg , MongoJobs ) : <EOL> self . handle = arg <EOL> else : <EOL> connection_string = arg <EOL> self . handle = MongoJobs . new_from_connection_str ( connection_string ) <EOL> self . handle . create_indexes ( ) <EOL> self . _exp_key = exp_key <EOL> self . cmd = cmd <EOL> self . workdir = workdir <EOL> if refresh : <EOL> self . refresh ( ) <EOL> def view ( self , exp_key = None , cmd = None , workdir = None , refresh = True ) : <EOL> rval = self . __class__ ( self . handle , <EOL> exp_key = self . _exp_key if exp_key is None else exp_key , <EOL> cmd = self . cmd if cmd is None else cmd , <EOL> workdir = self . workdir if workdir is None else workdir , <EOL> refresh = refresh ) <EOL> return rval <EOL> def refresh_tids ( self , tids ) : <EOL> """<STR_LIT>""" <EOL> exp_key = self . _exp_key <EOL> if exp_key != None : <EOL> query = { '<STR_LIT>' : exp_key } <EOL> else : <EOL> query = { } <EOL> t0 = time . time ( ) <EOL> query [ '<STR_LIT:state>' ] = { '<STR_LIT>' : JOB_STATE_ERROR } <EOL> if tids is not None : <EOL> query [ '<STR_LIT>' ] = { '<STR_LIT>' : list ( tids ) } <EOL> orig_trials = getattr ( self , '<STR_LIT>' , [ ] ) <EOL> _trials = orig_trials [ : ] <EOL> if _trials : <EOL> db_data = list ( self . handle . jobs . find ( query , <EOL> fields = [ '<STR_LIT>' , '<STR_LIT:version>' ] ) ) <EOL> if db_data : <EOL> db_data = numpy . rec . array ( [ ( x [ '<STR_LIT>' ] , int ( x [ '<STR_LIT:version>' ] ) ) <EOL> for x in db_data ] , <EOL> names = [ '<STR_LIT>' , '<STR_LIT:version>' ] ) <EOL> db_data . sort ( order = [ '<STR_LIT>' , '<STR_LIT:version>' ] ) <EOL> db_data = db_data [ get_most_recent_inds ( db_data ) ] <EOL> existing_data = numpy . rec . array ( [ ( x [ '<STR_LIT>' ] , <EOL> int ( x [ '<STR_LIT:version>' ] ) ) for x in _trials ] , <EOL> names = [ '<STR_LIT>' , '<STR_LIT:version>' ] ) <EOL> existing_data . sort ( order = [ '<STR_LIT>' , '<STR_LIT:version>' ] ) <EOL> db_in_existing = fast_isin ( db_data [ '<STR_LIT>' ] , existing_data [ '<STR_LIT>' ] ) <EOL> existing_in_db = fast_isin ( existing_data [ '<STR_LIT>' ] , db_data [ '<STR_LIT>' ] ) <EOL> _trials = [ _trials [ _ind ] for _ind in existing_in_db . nonzero ( ) [ <NUM_LIT:0> ] ] <EOL> new_data = db_data [ numpy . invert ( db_in_existing ) ] <EOL> db_data = db_data [ db_in_existing ] <EOL> existing_data = existing_data [ existing_in_db ] <EOL> try : <EOL> assert len ( db_data ) == len ( existing_data ) <EOL> assert ( existing_data [ '<STR_LIT>' ] == db_data [ '<STR_LIT>' ] ) . all ( ) <EOL> assert ( existing_data [ '<STR_LIT:version>' ] <= db_data [ '<STR_LIT:version>' ] ) . all ( ) <EOL> except : <EOL> reportpath = os . path . join ( os . getcwd ( ) , <EOL> '<STR_LIT>' + str ( numpy . random . randint ( <NUM_LIT> ) ) + '<STR_LIT>' ) <EOL> logger . error ( '<STR_LIT>' % reportpath ) <EOL> _file = open ( reportpath , '<STR_LIT:w>' ) <EOL> cPickle . dump ( { '<STR_LIT>' : db_data , <EOL> '<STR_LIT>' : existing_data } , <EOL> _file ) <EOL> _file . close ( ) <EOL> raise <EOL> same_version = existing_data [ '<STR_LIT:version>' ] == db_data [ '<STR_LIT:version>' ] <EOL> _trials = [ _trials [ _ind ] for _ind in same_version . nonzero ( ) [ <NUM_LIT:0> ] ] <EOL> version_changes = existing_data [ numpy . invert ( same_version ) ] <EOL> update_ids = new_data [ '<STR_LIT>' ] . tolist ( ) + version_changes [ '<STR_LIT>' ] . tolist ( ) <EOL> num_new = len ( update_ids ) <EOL> update_query = copy . deepcopy ( query ) <EOL> update_query [ '<STR_LIT>' ] = { '<STR_LIT>' : update_ids } <EOL> updated_trials = list ( self . handle . jobs . find ( update_query ) ) <EOL> _trials . extend ( updated_trials ) <EOL> else : <EOL> num_new = <NUM_LIT:0> <EOL> _trials = [ ] <EOL> else : <EOL> _trials = list ( self . handle . jobs . find ( query ) ) <EOL> if _trials : <EOL> _trials = [ _trials [ _i ] for _i in get_most_recent_inds ( _trials ) ] <EOL> num_new = len ( _trials ) <EOL> logger . debug ( '<STR_LIT>' % <EOL> ( time . time ( ) - t0 , num_new ) ) <EOL> if tids is not None : <EOL> new_trials = _trials <EOL> tids_set = set ( tids ) <EOL> assert all ( t [ '<STR_LIT>' ] in tids_set for t in new_trials ) <EOL> old_trials = [ t for t in orig_trials if t [ '<STR_LIT>' ] not in tids_set ] <EOL> _trials = new_trials + old_trials <EOL> jarray = numpy . array ( [ j [ '<STR_LIT>' ] for j in _trials ] ) <EOL> jobsort = jarray . argsort ( ) <EOL> self . _trials = [ _trials [ _idx ] for _idx in jobsort ] <EOL> self . _specs = [ _trials [ _idx ] [ '<STR_LIT>' ] for _idx in jobsort ] <EOL> self . _results = [ _trials [ _idx ] [ '<STR_LIT:result>' ] for _idx in jobsort ] <EOL> self . _miscs = [ _trials [ _idx ] [ '<STR_LIT>' ] for _idx in jobsort ] <EOL> def refresh ( self ) : <EOL> self . refresh_tids ( None ) <EOL> def _insert_trial_docs ( self , docs ) : <EOL> rval = [ ] <EOL> for doc in docs : <EOL> rval . append ( self . handle . jobs . insert ( doc ) ) <EOL> return rval <EOL> def count_by_state_unsynced ( self , arg ) : <EOL> exp_key = self . _exp_key <EOL> if isinstance ( arg , int ) : <EOL> if arg not in JOB_STATES : <EOL> raise ValueError ( '<STR_LIT>' , arg ) <EOL> query = dict ( state = arg ) <EOL> else : <EOL> assert hasattr ( arg , '<STR_LIT>' ) <EOL> states = list ( arg ) <EOL> assert all ( [ x in JOB_STATES for x in states ] ) <EOL> query = dict ( state = { '<STR_LIT>' : states } ) <EOL> if exp_key != None : <EOL> query [ '<STR_LIT>' ] = exp_key <EOL> rval = self . handle . jobs . find ( query ) . count ( ) <EOL> return rval <EOL> def delete_all ( self , cond = None ) : <EOL> if cond is None : <EOL> cond = { } <EOL> else : <EOL> cond = dict ( cond ) <EOL> if self . _exp_key : <EOL> cond [ '<STR_LIT>' ] = self . _exp_key <EOL> self . handle . delete_all ( cond ) <EOL> gfs = self . handle . gfs <EOL> for filename in gfs . list ( ) : <EOL> try : <EOL> fdoc = gfs . get_last_version ( filename = filename , ** cond ) <EOL> except gridfs . errors . NoFile : <EOL> continue <EOL> gfs . delete ( fdoc . _id ) <EOL> self . refresh ( ) <EOL> def new_trial_ids ( self , N ) : <EOL> db = self . handle . db <EOL> query = { '<STR_LIT:a>' : <NUM_LIT:0> } <EOL> doc = None <EOL> while doc is None : <EOL> doc = db . job_ids . find_and_modify ( <EOL> query , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : N } } , <EOL> upsert = True ) <EOL> if doc is None : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:1.0> ) <EOL> lid = doc . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> return range ( lid , lid + N ) <EOL> def trial_attachments ( self , trial ) : <EOL> """<STR_LIT>""" <EOL> class Attachments ( object ) : <EOL> def __contains__ ( _self , name ) : <EOL> return name in self . handle . attachment_names ( doc = trial ) <EOL> def __len__ ( _self ) : <EOL> return len ( self . handle . attachment_names ( doc = trial ) ) <EOL> def __iter__ ( _self ) : <EOL> return iter ( self . handle . attachment_names ( doc = trial ) ) <EOL> def __getitem__ ( _self , name ) : <EOL> try : <EOL> return self . handle . get_attachment ( <EOL> doc = trial , <EOL> name = name ) <EOL> except OperationFailure : <EOL> raise KeyError ( name ) <EOL> def __setitem__ ( _self , name , value ) : <EOL> self . handle . set_attachment ( <EOL> doc = trial , <EOL> blob = value , <EOL> name = name , <EOL> collection = self . handle . db . jobs ) <EOL> def __delitem__ ( _self , name ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def keys ( self ) : <EOL> return [ k for k in self ] <EOL> def values ( self ) : <EOL> return [ self [ k ] for k in self ] <EOL> def items ( self ) : <EOL> return [ ( k , self [ k ] ) for k in self ] <EOL> return Attachments ( ) <EOL> @ property <EOL> def attachments ( self ) : <EOL> """<STR_LIT>""" <EOL> gfs = self . handle . gfs <EOL> query = { } <EOL> if self . _exp_key : <EOL> query [ '<STR_LIT>' ] = self . _exp_key <EOL> class Attachments ( object ) : <EOL> def __iter__ ( _self ) : <EOL> if query : <EOL> filenames = [ fname <EOL> for fname in gfs . list ( ) <EOL> if fname in _self ] <EOL> else : <EOL> filenames = gfs . list ( ) <EOL> return iter ( filenames ) <EOL> def __contains__ ( _self , name ) : <EOL> return gfs . exists ( filename = name , ** query ) <EOL> def __getitem__ ( _self , name ) : <EOL> try : <EOL> rval = gfs . get_version ( filename = name , ** query ) . read ( ) <EOL> return rval <EOL> except gridfs . NoFile : <EOL> raise KeyError ( name ) <EOL> def __setitem__ ( _self , name , value ) : <EOL> if gfs . exists ( filename = name , ** query ) : <EOL> gout = gfs . get_last_version ( filename = name , ** query ) <EOL> gfs . delete ( gout . _id ) <EOL> gfs . put ( value , filename = name , ** query ) <EOL> def __delitem__ ( _self , name ) : <EOL> gout = gfs . get_last_version ( filename = name , ** query ) <EOL> gfs . delete ( gout . _id ) <EOL> return Attachments ( ) <EOL> class MongoWorker ( object ) : <EOL> poll_interval = <NUM_LIT> <EOL> workdir = None <EOL> def __init__ ( self , mj , <EOL> poll_interval = poll_interval , <EOL> workdir = workdir , <EOL> exp_key = None , <EOL> logfilename = '<STR_LIT>' , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> self . mj = mj <EOL> self . poll_interval = poll_interval <EOL> self . workdir = workdir <EOL> self . exp_key = exp_key <EOL> self . logfilename = logfilename <EOL> def make_log_handler ( self ) : <EOL> self . log_handler = logging . FileHandler ( self . logfilename ) <EOL> self . log_handler . setFormatter ( <EOL> logging . Formatter ( <EOL> fmt = '<STR_LIT>' ) ) <EOL> self . log_handler . setLevel ( logging . INFO ) <EOL> def run_one ( self , <EOL> host_id = None , <EOL> reserve_timeout = None , <EOL> erase_created_workdir = False , <EOL> ) : <EOL> if host_id == None : <EOL> host_id = '<STR_LIT>' % ( socket . gethostname ( ) , os . getpid ( ) ) , <EOL> job = None <EOL> start_time = time . time ( ) <EOL> mj = self . mj <EOL> while job is None : <EOL> if ( reserve_timeout <EOL> and ( time . time ( ) - start_time ) > reserve_timeout ) : <EOL> raise ReserveTimeout ( ) <EOL> job = mj . reserve ( host_id , exp_key = self . exp_key ) <EOL> if not job : <EOL> interval = ( <NUM_LIT:1> + <EOL> numpy . random . rand ( ) <EOL> * ( float ( self . poll_interval ) - <NUM_LIT:1.0> ) ) <EOL> logger . info ( '<STR_LIT>' % interval ) <EOL> time . sleep ( interval ) <EOL> logger . debug ( '<STR_LIT>' % str ( job ) ) <EOL> spec = spec_from_misc ( job [ '<STR_LIT>' ] ) <EOL> ctrl = MongoCtrl ( <EOL> trials = MongoTrials ( mj , exp_key = job [ '<STR_LIT>' ] , refresh = False ) , <EOL> read_only = False , <EOL> current_trial = job ) <EOL> if self . workdir is None : <EOL> workdir = job [ '<STR_LIT>' ] . get ( '<STR_LIT>' , os . getcwd ( ) ) <EOL> if workdir is None : <EOL> workdir = '<STR_LIT>' <EOL> workdir = os . path . join ( workdir , str ( job [ '<STR_LIT>' ] ) ) <EOL> else : <EOL> workdir = self . workdir <EOL> workdir = os . path . abspath ( os . path . expanduser ( workdir ) ) <EOL> try : <EOL> root_logger = logging . getLogger ( ) <EOL> if self . logfilename : <EOL> self . make_log_handler ( ) <EOL> root_logger . addHandler ( self . log_handler ) <EOL> cmd = job [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> cmd_protocol = cmd [ <NUM_LIT:0> ] <EOL> try : <EOL> if cmd_protocol == '<STR_LIT>' : <EOL> worker_fn = cPickle . loads ( cmd [ <NUM_LIT:1> ] ) <EOL> elif cmd_protocol == '<STR_LIT>' : <EOL> bandit = cPickle . loads ( cmd [ <NUM_LIT:1> ] ) <EOL> worker_fn = bandit . evaluate <EOL> elif cmd_protocol == '<STR_LIT>' : <EOL> cmd_toks = cmd [ <NUM_LIT:1> ] . split ( '<STR_LIT:.>' ) <EOL> cmd_module = '<STR_LIT:.>' . join ( cmd_toks [ : - <NUM_LIT:1> ] ) <EOL> worker_fn = exec_import ( cmd_module , cmd [ <NUM_LIT:1> ] ) <EOL> elif cmd_protocol == '<STR_LIT>' : <EOL> bandit = json_call ( cmd [ <NUM_LIT:1> ] ) <EOL> worker_fn = bandit . evaluate <EOL> elif cmd_protocol == '<STR_LIT>' : <EOL> blob = ctrl . trials . attachments [ cmd [ <NUM_LIT:1> ] ] <EOL> bandit_name , bandit_args , bandit_kwargs = cPickle . loads ( blob ) <EOL> worker_fn = json_call ( bandit_name , <EOL> args = bandit_args , <EOL> kwargs = bandit_kwargs ) . evaluate <EOL> elif cmd_protocol == '<STR_LIT>' : <EOL> blob = ctrl . trials . attachments [ cmd [ <NUM_LIT:1> ] ] <EOL> try : <EOL> domain = cPickle . loads ( blob ) <EOL> except BaseException , e : <EOL> logger . info ( '<STR_LIT>' ) <EOL> raise <EOL> worker_fn = domain . evaluate <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' , cmd_protocol ) <EOL> with temp_dir ( workdir , erase_created_workdir ) , working_dir ( workdir ) : <EOL> result = worker_fn ( spec , ctrl ) <EOL> result = SONify ( result ) <EOL> except BaseException , e : <EOL> logger . info ( '<STR_LIT>' % str ( e ) ) <EOL> ctrl . checkpoint ( ) <EOL> mj . update ( job , <EOL> { '<STR_LIT:state>' : JOB_STATE_ERROR , <EOL> '<STR_LIT:error>' : ( str ( type ( e ) ) , str ( e ) ) } ) <EOL> raise <EOL> finally : <EOL> if self . logfilename : <EOL> root_logger . removeHandler ( self . log_handler ) <EOL> logger . info ( '<STR_LIT>' % str ( job [ '<STR_LIT>' ] ) ) <EOL> attachments = result . pop ( '<STR_LIT>' , { } ) <EOL> for aname , aval in attachments . items ( ) : <EOL> logger . info ( <EOL> '<STR_LIT>' % ( <EOL> aname , len ( aval ) ) ) <EOL> ctrl . attachments [ aname ] = aval <EOL> ctrl . checkpoint ( result ) <EOL> mj . update ( job , { '<STR_LIT:state>' : JOB_STATE_DONE } ) <EOL> class MongoCtrl ( Ctrl ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , trials , current_trial , read_only ) : <EOL> self . trials = trials <EOL> self . current_trial = current_trial <EOL> self . read_only = read_only <EOL> def debug ( self , * args , ** kwargs ) : <EOL> return logger . debug ( * args , ** kwargs ) <EOL> def info ( self , * args , ** kwargs ) : <EOL> return logger . info ( * args , ** kwargs ) <EOL> def warn ( self , * args , ** kwargs ) : <EOL> return logger . warn ( * args , ** kwargs ) <EOL> def error ( self , * args , ** kwargs ) : <EOL> return logger . error ( * args , ** kwargs ) <EOL> def checkpoint ( self , result = None ) : <EOL> if not self . read_only : <EOL> handle = self . trials . handle <EOL> handle . refresh ( self . current_trial ) <EOL> if result is not None : <EOL> return handle . update ( self . current_trial , dict ( result = result ) ) <EOL> @ property <EOL> def attachments ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . trials . trial_attachments ( trial = self . current_trial ) <EOL> @ property <EOL> def set_attachment ( self ) : <EOL> raise RuntimeError ( <EOL> '<STR_LIT>' ) <EOL> def exec_import ( cmd_module , cmd ) : <EOL> worker_fn = None <EOL> exec ( '<STR_LIT>' % ( cmd_module , cmd ) ) <EOL> return worker_fn <EOL> def as_mongo_str ( s ) : <EOL> if s . startswith ( '<STR_LIT>' ) : <EOL> return s <EOL> else : <EOL> return '<STR_LIT>' % s <EOL> def main_worker_helper ( options , args ) : <EOL> N = int ( options . max_jobs ) <EOL> if options . last_job_timeout is not None : <EOL> last_job_timeout = time . time ( ) + float ( options . last_job_timeout ) <EOL> else : <EOL> last_job_timeout = None <EOL> def sighandler_shutdown ( signum , frame ) : <EOL> logger . info ( '<STR_LIT>' % signum ) <EOL> raise Shutdown ( signum ) <EOL> def sighandler_wait_quit ( signum , frame ) : <EOL> logger . info ( '<STR_LIT>' % signum ) <EOL> raise WaitQuit ( signum ) <EOL> signal . signal ( signal . SIGINT , sighandler_shutdown ) <EOL> signal . signal ( signal . SIGHUP , sighandler_shutdown ) <EOL> signal . signal ( signal . SIGTERM , sighandler_shutdown ) <EOL> signal . signal ( signal . SIGUSR1 , sighandler_wait_quit ) <EOL> if N > <NUM_LIT:1> : <EOL> proc = None <EOL> cons_errs = <NUM_LIT:0> <EOL> if last_job_timeout and time . time ( ) > last_job_timeout : <EOL> logger . info ( "<STR_LIT>" ) <EOL> return <EOL> while N and cons_errs < int ( options . max_consecutive_failures ) : <EOL> try : <EOL> sub_argv = [ sys . argv [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' % options . poll_interval , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' % options . mongo , <EOL> '<STR_LIT>' % options . reserve_timeout ] <EOL> if options . workdir is not None : <EOL> sub_argv . append ( '<STR_LIT>' % options . workdir ) <EOL> if options . exp_key is not None : <EOL> sub_argv . append ( '<STR_LIT>' % options . exp_key ) <EOL> proc = subprocess . Popen ( sub_argv ) <EOL> retcode = proc . wait ( ) <EOL> proc = None <EOL> except Shutdown : <EOL> if proc : <EOL> os . kill ( proc . pid , signal . SIGTERM ) <EOL> return proc . wait ( ) <EOL> else : <EOL> return <NUM_LIT:0> <EOL> except WaitQuit : <EOL> if proc : <EOL> return proc . wait ( ) <EOL> else : <EOL> return <NUM_LIT:0> <EOL> if retcode != <NUM_LIT:0> : <EOL> cons_errs += <NUM_LIT:1> <EOL> else : <EOL> cons_errs = <NUM_LIT:0> <EOL> N -= <NUM_LIT:1> <EOL> logger . info ( "<STR_LIT>" % ( <EOL> N , cons_errs ) ) <EOL> elif N == <NUM_LIT:1> : <EOL> mj = MongoJobs . new_from_connection_str ( <EOL> as_mongo_str ( options . mongo ) + '<STR_LIT>' ) <EOL> mworker = MongoWorker ( mj , <EOL> float ( options . poll_interval ) , <EOL> workdir = options . workdir , <EOL> exp_key = options . exp_key ) <EOL> mworker . run_one ( reserve_timeout = float ( options . reserve_timeout ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def main_worker ( ) : <EOL> parser = optparse . OptionParser ( usage = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> default = None , <EOL> metavar = '<STR_LIT:str>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> metavar = '<STR_LIT:T>' , <EOL> default = None , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = "<STR_LIT>" , <EOL> metavar = '<STR_LIT:N>' , <EOL> default = <NUM_LIT:4> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> default = sys . maxint , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> metavar = '<STR_LIT:N>' , <EOL> default = <NUM_LIT:5> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = '<STR_LIT>' , <EOL> metavar = '<STR_LIT:T>' , <EOL> default = <NUM_LIT> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , <EOL> dest = "<STR_LIT>" , <EOL> default = None , <EOL> help = "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" ) <EOL> ( options , args ) = parser . parse_args ( ) <EOL> if args : <EOL> parser . print_help ( ) <EOL> return - <NUM_LIT:1> <EOL> return main_worker_helper ( options , args ) </s>
