<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> import sys <EOL> if sys . version < '<STR_LIT:3>' : <EOL> import Queue as queue <EOL> else : <EOL> import queue <EOL> import multiprocessing <EOL> import os <EOL> import random <EOL> import time <EOL> import numpy <EOL> import tables <EOL> def make_file ( file_path , n ) : <EOL> with tables . open_file ( file_path , '<STR_LIT:w>' ) as fobj : <EOL> array = fobj . create_carray ( '<STR_LIT:/>' , '<STR_LIT>' , tables . Int64Atom ( ) , ( n , n ) ) <EOL> for i in range ( n ) : <EOL> array [ i , : ] = i <EOL> class FileAccess ( multiprocessing . Process ) : <EOL> def __init__ ( self , h5_path , read_queue , result_queues , write_queue , <EOL> shutdown ) : <EOL> self . h5_path = h5_path <EOL> self . read_queue = read_queue <EOL> self . result_queues = result_queues <EOL> self . write_queue = write_queue <EOL> self . shutdown = shutdown <EOL> self . block_period = <NUM_LIT> <EOL> super ( FileAccess , self ) . __init__ ( ) <EOL> def run ( self ) : <EOL> self . h5_file = tables . open_file ( self . h5_path , '<STR_LIT>' ) <EOL> self . array = self . h5_file . get_node ( '<STR_LIT>' ) <EOL> another_loop = True <EOL> while another_loop : <EOL> if self . shutdown . poll ( ) : <EOL> another_loop = False <EOL> try : <EOL> row_num , proc_num = self . read_queue . get ( <EOL> True , self . block_period ) <EOL> result_queue = self . result_queues [ proc_num ] <EOL> print ( '<STR_LIT>' . format ( proc_num , <EOL> row_num ) ) <EOL> result_queue . put ( self . read_data ( row_num ) ) <EOL> another_loop = True <EOL> except queue . Empty : <EOL> pass <EOL> try : <EOL> row_num , data = self . write_queue . get ( True , self . block_period ) <EOL> print ( '<STR_LIT>' , row_num ) <EOL> self . write_data ( row_num , data ) <EOL> another_loop = True <EOL> except queue . Empty : <EOL> pass <EOL> self . h5_file . close ( ) <EOL> def read_data ( self , row_num ) : <EOL> return self . array [ row_num , : ] <EOL> def write_data ( self , row_num , data ) : <EOL> self . array [ row_num , : ] = data <EOL> class DataProcessor ( multiprocessing . Process ) : <EOL> def __init__ ( self , read_queue , result_queue , write_queue , proc_num , <EOL> array_size , output_file ) : <EOL> self . read_queue = read_queue <EOL> self . result_queue = result_queue <EOL> self . write_queue = write_queue <EOL> self . proc_num = proc_num <EOL> self . array_size = array_size <EOL> self . output_file = output_file <EOL> super ( DataProcessor , self ) . __init__ ( ) <EOL> def run ( self ) : <EOL> self . output_file = open ( self . output_file , '<STR_LIT:w>' ) <EOL> row_num = random . randint ( <NUM_LIT:0> , self . array_size - <NUM_LIT:1> ) <EOL> self . read_queue . put ( ( row_num , self . proc_num ) ) <EOL> self . output_file . write ( str ( row_num ) + '<STR_LIT:\n>' ) <EOL> self . output_file . write ( str ( self . result_queue . get ( ) ) + '<STR_LIT:\n>' ) <EOL> row_num = random . randint ( <NUM_LIT:0> , self . array_size - <NUM_LIT:1> ) <EOL> new_data = ( numpy . zeros ( ( <NUM_LIT:1> , self . array_size ) , '<STR_LIT>' ) + <EOL> <NUM_LIT:11> * ( self . proc_num + <NUM_LIT:1> ) ) <EOL> self . write_queue . put ( ( row_num , new_data ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . read_queue . put ( ( row_num , self . proc_num ) ) <EOL> self . output_file . write ( str ( row_num ) + '<STR_LIT:\n>' ) <EOL> self . output_file . write ( str ( self . result_queue . get ( ) ) + '<STR_LIT:\n>' ) <EOL> self . output_file . close ( ) <EOL> def make_queues ( num_processors ) : <EOL> read_queue = multiprocessing . Queue ( ) <EOL> write_queue = multiprocessing . Queue ( ) <EOL> shutdown_recv , shutdown_send = multiprocessing . Pipe ( False ) <EOL> result_queues = [ multiprocessing . Queue ( ) for i in range ( num_processors ) ] <EOL> file_access = FileAccess ( file_path , read_queue , result_queues , write_queue , <EOL> shutdown_recv ) <EOL> file_access . start ( ) <EOL> return read_queue , result_queues , write_queue , shutdown_send <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> file_path = '<STR_LIT>' <EOL> n = <NUM_LIT:10> <EOL> make_file ( file_path , n ) <EOL> num_processors = <NUM_LIT:3> <EOL> ( read_queue , result_queues , <EOL> write_queue , shutdown_send ) = make_queues ( num_processors ) <EOL> processors = [ ] <EOL> output_files = [ ] <EOL> for i in range ( num_processors ) : <EOL> result_queue = result_queues [ i ] <EOL> output_file = str ( i ) <EOL> processor = DataProcessor ( read_queue , result_queue , write_queue , i , n , <EOL> output_file ) <EOL> processors . append ( processor ) <EOL> output_files . append ( output_file ) <EOL> for processor in processors : <EOL> processor . start ( ) <EOL> for processor in processors : <EOL> processor . join ( ) <EOL> shutdown_send . send ( <NUM_LIT:0> ) <EOL> print ( ) <EOL> for output_file in output_files : <EOL> print ( ) <EOL> print ( '<STR_LIT>' . format ( output_file ) ) <EOL> print ( open ( output_file , '<STR_LIT:r>' ) . read ( ) ) <EOL> os . remove ( output_file ) <EOL> os . remove ( '<STR_LIT>' ) </s>
