<s> from tornado . web import * <EOL> import json <EOL> from invocation import * <EOL> from exceptions import * <EOL> from tornado . options import define , options <EOL> from events import EventManager <EOL> from tornado . websocket import WebSocketHandler <EOL> import logging <EOL> from collections import deque <EOL> class ClientSideWorkerManager ( object ) : <EOL> def __init__ ( self ) : <EOL> self . __workers = { } <EOL> self . __worker_queue = deque ( ) <EOL> self . __operation_queue = deque ( ) <EOL> self . __operation_callbacks = { } <EOL> def add_worker ( self , worker ) : <EOL> self . __workers [ id ( worker ) ] = worker <EOL> self . __worker_queue . append ( id ( worker ) ) <EOL> self . run_operation ( ) <EOL> def remove_worker ( self , worker ) : <EOL> del self . __workers [ id ( worker ) ] <EOL> def run_operation ( self ) : <EOL> while self . __operation_queue and self . __worker_queue : <EOL> operation = self . __operation_queue . popleft ( ) <EOL> worker_id = self . __worker_queue . popleft ( ) <EOL> self . __workers [ worker_id ] . write_message ( { '<STR_LIT>' : operation [ <NUM_LIT:0> ] , '<STR_LIT>' : operation [ <NUM_LIT:1> ] } ) <EOL> if ( operation [ <NUM_LIT:2> ] ) : <EOL> self . __operation_queue . append ( operation ) <EOL> def add_operation ( self , operation_id , operation_script , callback_method = None , max_nodes = <NUM_LIT:1> , continuous = False ) : <EOL> nodes = min ( max_nodes , len ( self . __worker_queue ) ) or <NUM_LIT:1> <EOL> if callback_method : <EOL> self . __operation_callbacks [ operation_id ] = callback_method <EOL> for i in xrange ( nodes ) : <EOL> self . __operation_queue . append ( ( operation_id , operation_script , continuous ) ) <EOL> self . run_operation ( ) <EOL> def finish_operation ( self , worker , operation_id , result ) : <EOL> self . __worker_queue . append ( id ( worker ) ) <EOL> if operation_id in self . __operation_callbacks : <EOL> self . __operation_callbacks [ operation_id ] ( worker , result ) <EOL> self . run_operation ( ) <EOL> @ staticmethod <EOL> def instance ( ) : <EOL> if not hasattr ( ClientSideWorkerManager , '<STR_LIT>' ) : <EOL> ClientSideWorkerManager . _instance = ClientSideWorkerManager ( ) <EOL> return ClientSideWorkerManager . _instance <EOL> def worker_connected ( worker ) : <EOL> ClientSideWorkerManager . instance ( ) . add_worker ( worker ) <EOL> def worker_disconnected ( worker ) : <EOL> ClientSideWorkerManager . instance ( ) . remove_worker ( worker ) <EOL> def complete ( worker , data ) : <EOL> ClientSideWorkerManager . instance ( ) . finish_operation ( worker , data [ '<STR_LIT>' ] , data [ '<STR_LIT:result>' ] ) <EOL> class ClientSideWorkerSocketHandler ( WebSocketHandler ) : <EOL> @ classmethod <EOL> def configure ( cls ) : <EOL> if options . debug : <EOL> import traceback <EOL> def log_error ( self , e ) : <EOL> logging . error ( '<STR_LIT>' % ( traceback . format_exc ( ) , repr ( self . request . headers ) ) ) <EOL> cls . log_error = log_error <EOL> def log_error ( self , e ) : <EOL> if isinstance ( exception , TotoException ) : <EOL> logging . error ( "<STR_LIT>" % ( e . code , e . value ) ) <EOL> else : <EOL> logging . error ( "<STR_LIT>" % ( ERROR_SERVER , repr ( e ) ) ) <EOL> def open ( self ) : <EOL> ClientSideWorkerManager . instance ( ) . add_worker ( self ) <EOL> def on_message ( self , message_data ) : <EOL> message = json . loads ( message_data ) <EOL> ClientSideWorkerManager . instance ( ) . finish_operation ( self , message [ '<STR_LIT>' ] , message [ '<STR_LIT:result>' ] ) <EOL> def on_close ( self ) : <EOL> ClientSideWorkerManager . instance ( ) . remove_worker ( self ) </s>
