python/scannerpy/database.py
def run(self, bulk_job,
            force=False,
            work_packet_size=250,
            io_packet_size=-1,
            cpu_pool=None,
            gpu_pool=None,
            pipeline_instances_per_node=None,
            show_progress=False,
            profiling=False,
            load_sparsity_threshold=8,
            tasks_in_queue_per_pu=4,
            load_to_disk=False):
        """
        Runs a computation over a set of inputs.

        Args:
            tasks: The set of inputs to run the computation on. If tasks is a
                   Collection, then the computation is run on all frames of all
                   tables in the collection. Otherwise, tasks should be generated
                   by the Sampler.
            outputs: TODO(wcrichto)

        Kwargs:
            output_collection: If this is not None, then a new collection with
                               this name will be created for all the output
                               tables.
            force: TODO(wcrichto)
            work_item_size: TODO(wcrichto)
            io_item_size: TODO(wcrichto)
            cpu_pool: TODO(wcrichto)
            gpu_pool: TODO(wcrichto)
            pipeline_instances_per_node: TODO(wcrichto)
            show_progress: TODO(wcrichto)

        Returns:
            Either the output Collection if output_collection is specified
            or a list of Table objects.
        """
        assert isinstance(bulk_job, BulkJob)
        assert isinstance(bulk_job.output(), Op)

        # Collect compression annotations to add to job
        compression_options = []
        output_op = bulk_job.output()
        for out_col in output_op.inputs():
            opts = self.protobufs.OutputColumnCompression()
            opts.codec = 'default'
            if out_col._type == self.protobufs.Video:
                for k, v in out_col._encode_options.iteritems():
                    if k == 'codec':
                        opts.codec = v
                    else:
                        opts.options[k] = str(v)
            compression_options.append(opts)

        sorted_ops, input_ops, sampling_slicing_ops, output_ops = (
            self._toposort(bulk_job.output()))

        job_params = self.protobufs.BulkJobParameters()
        job_name = ''.join(choice(ascii_uppercase) for _ in range(12))
        job_params.job_name = job_name
        job_params.ops.extend(sorted_ops)
        job_output_table_names = []
        for job in bulk_job.jobs():
            j = job_params.jobs.add()
            output_table_name = None
            for op_col, args in job.op_args().iteritems():
                if isinstance(op_col, Op):
                    op = op_col
                else:
                    op = op_col._op
                if op in input_ops:
                    op_idx = input_ops[op]
                    col_input = j.inputs.add()
                    col_input.op_index = op_idx
                    if not args._table.committed():
                        raise ScannerException(
                            'Attempted to bind table {name} to Input Op but '
                            'table {name} is not committed.'
                            .format(name=args._table.name()))
                    col_input.table_name = args._table.name()
                    col_input.column_name = args.name()
                elif op in sampling_slicing_ops:
                    op_idx = sampling_slicing_ops[op]
                    saa = j.sampling_args_assignment.add()
                    saa.op_index = op_idx
                    if not isinstance(args, list):
                        args = [args]
                    for arg in args:
                        sa = saa.sampling_args.add()
                        sa.CopyFrom(arg)
                elif op in output_ops:
                    op_idx = output_ops[op]
                    assert isinstance(args, basestring)
                    output_table_name = args
                    job_output_table_names.append(args)
                else:
                    raise ScannerException(
                        'Attempted to bind arguments to Op {} which is not '
                        'an input, sampling, spacing, slicing, or output Op.'
                        .format(op.name()))  # FIXME(apoms): op.name() is unbound
            if output_table_name is None:
                raise ScannerException(
                    'Did not specify the output table name by binding a '
                    'string to the output Op.')
            j.output_table_name = output_table_name

        # Delete tables if they exist and force was specified
        to_delete = []
        for name in job_output_table_names:
            if self.has_table(name):
                if force:
                    to_delete.append(name)
                else:
                    raise ScannerException(
                        'Job would overwrite existing table {}'
                        .format(name))
        self.delete_tables(to_delete)

        job_params.compression.extend(compression_options)
        job_params.pipeline_instances_per_node = (
            pipeline_instances_per_node or -1)
        job_params.work_packet_size = work_packet_size
        job_params.io_packet_size = io_packet_size
        job_params.show_progress = show_progress
        job_params.profiling = profiling
        job_params.tasks_in_queue_per_pu = tasks_in_queue_per_pu
        job_params.load_sparsity_threshold = load_sparsity_threshold
        job_params.boundary_condition = (
            self.protobufs.BulkJobParameters.REPEAT_EDGE)

        job_params.memory_pool_config.pinned_cpu = False
        if cpu_pool is not None:
            job_params.memory_pool_config.cpu.use_pool = True
            if cpu_pool[0] == 'p':
                job_params.memory_pool_config.pinned_cpu = True
                cpu_pool = cpu_pool[1:]
            size = self._parse_size_string(cpu_pool)
            job_params.memory_pool_config.cpu.free_space = size

        if gpu_pool is not None:
            job_params.memory_pool_config.gpu.use_pool = True
            size = self._parse_size_string(gpu_pool)
            job_params.memory_pool_config.gpu.free_space = size

        job_params.load_to_disk = load_to_disk
        # Run the job
        self._try_rpc(lambda: self._master.NewJob(job_params))

        while True:
            try:
                result = self._master.IsJobDone(self.protobufs.Empty())
            except grpc.RpcError as e:
                raise ScannerException(e)
            if result.finished:
                break
            else:
                # Qian: no sleep!!
                time.sleep(1.0)
                pass

        if not result.result.success:
            raise ScannerException(result.result.msg)

        # Invalidate db metadata because of job run
        self._cached_db_metadata = None

        db_meta = self._load_db_metadata()
        job_id = None
        for job in db_meta.bulk_jobs:
            if job.name == job_name:
                job_id = job.id
        if job_id is None:
            raise ScannerException('Internal error: job id not found after run')

        return [self.table(t) for t in job_output_table_names]

        Runs a computation over a set of inputs.

        Args:
            tasks: The set of inputs to run the computation on. If tasks is a
                   Collection, then the computation is run on all frames of all
                   tables in the collection. Otherwise, tasks should be generated
                   by the Sampler.
            outputs: TODO(wcrichto)

        Kwargs:
            output_collection: If this is not None, then a new collection with
                               this name will be created for all the output
                               tables.
            force: TODO(wcrichto)
            work_item_size: TODO(wcrichto)
            io_item_size: TODO(wcrichto)
            cpu_pool: TODO(wcrichto)
            gpu_pool: TODO(wcrichto)
            pipeline_instances_per_node: TODO(wcrichto)
            show_progress: TODO(wcrichto)

        Returns:
            Either the output Collection if output_collection is specified
            or a list of Table objects.
        
Runs a computation over a set of inputs.
Runs a computation over a set of inputs .
def run(self, bulk_job,
            force=False,
            work_packet_size=250,
            io_packet_size=-1,
            cpu_pool=None,
            gpu_pool=None,
            pipeline_instances_per_node=None,
            show_progress=False,
            profiling=False,
            load_sparsity_threshold=8,
            tasks_in_queue_per_pu=4,
            load_to_disk=False):
        assert isinstance(bulk_job, BulkJob)
        assert isinstance(bulk_job.output(), Op)
        compression_options = []
        output_op = bulk_job.output()
        for out_col in output_op.inputs():
            opts = self.protobufs.OutputColumnCompression()
            opts.codec = 'default'
            if out_col._type == self.protobufs.Video:
                for k, v in out_col._encode_options.iteritems():
                    if k == 'codec':
                        opts.codec = v
                    else:
                        opts.options[k] = str(v)
            compression_options.append(opts)
        sorted_ops, input_ops, sampling_slicing_ops, output_ops = (
            self._toposort(bulk_job.output()))
        job_params = self.protobufs.BulkJobParameters()
        job_name = ''.join(choice(ascii_uppercase) for _ in range(12))
        job_params.job_name = job_name
        job_params.ops.extend(sorted_ops)
        job_output_table_names = []
        for job in bulk_job.jobs():
            j = job_params.jobs.add()
            output_table_name = None
            for op_col, args in job.op_args().iteritems():
                if isinstance(op_col, Op):
                    op = op_col
                else:
                    op = op_col._op
                if op in input_ops:
                    op_idx = input_ops[op]
                    col_input = j.inputs.add()
                    col_input.op_index = op_idx
                    if not args._table.committed():
                        raise ScannerException(
                            'Attempted to bind table {name} to Input Op but '
                            'table {name} is not committed.'
                            .format(name=args._table.name()))
                    col_input.table_name = args._table.name()
                    col_input.column_name = args.name()
                elif op in sampling_slicing_ops:
                    op_idx = sampling_slicing_ops[op]
                    saa = j.sampling_args_assignment.add()
                    saa.op_index = op_idx
                    if not isinstance(args, list):
                        args = [args]
                    for arg in args:
                        sa = saa.sampling_args.add()
                        sa.CopyFrom(arg)
                elif op in output_ops:
                    op_idx = output_ops[op]
                    assert isinstance(args, basestring)
                    output_table_name = args
                    job_output_table_names.append(args)
                else:
                    raise ScannerException(
                        'Attempted to bind arguments to Op {} which is not '
                        'an input, sampling, spacing, slicing, or output Op.'
                        .format(op.name()))  
            if output_table_name is None:
                raise ScannerException(
                    'Did not specify the output table name by binding a '
                    'string to the output Op.')
            j.output_table_name = output_table_name
        to_delete = []
        for name in job_output_table_names:
            if self.has_table(name):
                if force:
                    to_delete.append(name)
                else:
                    raise ScannerException(
                        'Job would overwrite existing table {}'
                        .format(name))
        self.delete_tables(to_delete)
        job_params.compression.extend(compression_options)
        job_params.pipeline_instances_per_node = (
            pipeline_instances_per_node or -1)
        job_params.work_packet_size = work_packet_size
        job_params.io_packet_size = io_packet_size
        job_params.show_progress = show_progress
        job_params.profiling = profiling
        job_params.tasks_in_queue_per_pu = tasks_in_queue_per_pu
        job_params.load_sparsity_threshold = load_sparsity_threshold
        job_params.boundary_condition = (
            self.protobufs.BulkJobParameters.REPEAT_EDGE)
        job_params.memory_pool_config.pinned_cpu = False
        if cpu_pool is not None:
            job_params.memory_pool_config.cpu.use_pool = True
            if cpu_pool[0] == 'p':
                job_params.memory_pool_config.pinned_cpu = True
                cpu_pool = cpu_pool[1:]
            size = self._parse_size_string(cpu_pool)
            job_params.memory_pool_config.cpu.free_space = size
        if gpu_pool is not None:
            job_params.memory_pool_config.gpu.use_pool = True
            size = self._parse_size_string(gpu_pool)
            job_params.memory_pool_config.gpu.free_space = size
        job_params.load_to_disk = load_to_disk
        self._try_rpc(lambda: self._master.NewJob(job_params))
        while True:
            try:
                result = self._master.IsJobDone(self.protobufs.Empty())
            except grpc.RpcError as e:
                raise ScannerException(e)
            if result.finished:
                break
            else:
                time.sleep(1.0)
                pass
        if not result.result.success:
            raise ScannerException(result.result.msg)
        self._cached_db_metadata = None
        db_meta = self._load_db_metadata()
        job_id = None
        for job in db_meta.bulk_jobs:
            if job.name == job_name:
                job_id = job.id
        if job_id is None:
            raise ScannerException('Internal error: job id not found after run')
        return [self.table(t) for t in job_output_table_names]
def run ( self , bulk_job , force = False , work_packet_size = 250 , io_packet_size = - 1 , cpu_pool = None , gpu_pool = None , pipeline_instances_per_node = None , show_progress = False , profiling = False , load_sparsity_threshold = 8 , tasks_in_queue_per_pu = 4 , load_to_disk = False ) : assert isinstance ( bulk_job , BulkJob ) assert isinstance ( bulk_job . output ( ) , Op ) compression_options = [ ] output_op = bulk_job . output ( ) for out_col in output_op . inputs ( ) : opts = self . protobufs . OutputColumnCompression ( ) opts . codec = 'default' if out_col . _type == self . protobufs . Video : for k , v in out_col . _encode_options . iteritems ( ) : if k == 'codec' : opts . codec = v else : opts . options [ k ] = str ( v ) compression_options . append ( opts ) sorted_ops , input_ops , sampling_slicing_ops , output_ops = ( self . _toposort ( bulk_job . output ( ) ) ) job_params = self . protobufs . BulkJobParameters ( ) job_name = '' . join ( choice ( ascii_uppercase ) for _ in range ( 12 ) ) job_params . job_name = job_name job_params . ops . extend ( sorted_ops ) job_output_table_names = [ ] for job in bulk_job . jobs ( ) : j = job_params . jobs . add ( ) output_table_name = None for op_col , args in job . op_args ( ) . iteritems ( ) : if isinstance ( op_col , Op ) : op = op_col else : op = op_col . _op if op in input_ops : op_idx = input_ops [ op ] col_input = j . inputs . add ( ) col_input . op_index = op_idx if not args . _table . committed ( ) : raise ScannerException ( 'Attempted to bind table {name} to Input Op but ' 'table {name} is not committed.' . format ( name = args . _table . name ( ) ) ) col_input . table_name = args . _table . name ( ) col_input . column_name = args . name ( ) elif op in sampling_slicing_ops : op_idx = sampling_slicing_ops [ op ] saa = j . sampling_args_assignment . add ( ) saa . op_index = op_idx if not isinstance ( args , list ) : args = [ args ] for arg in args : sa = saa . sampling_args . add ( ) sa . CopyFrom ( arg ) elif op in output_ops : op_idx = output_ops [ op ] assert isinstance ( args , basestring ) output_table_name = args job_output_table_names . append ( args ) else : raise ScannerException ( 'Attempted to bind arguments to Op {} which is not ' 'an input, sampling, spacing, slicing, or output Op.' . format ( op . name ( ) ) ) if output_table_name is None : raise ScannerException ( 'Did not specify the output table name by binding a ' 'string to the output Op.' ) j . output_table_name = output_table_name to_delete = [ ] for name in job_output_table_names : if self . has_table ( name ) : if force : to_delete . append ( name ) else : raise ScannerException ( 'Job would overwrite existing table {}' . format ( name ) ) self . delete_tables ( to_delete ) job_params . compression . extend ( compression_options ) job_params . pipeline_instances_per_node = ( pipeline_instances_per_node or - 1 ) job_params . work_packet_size = work_packet_size job_params . io_packet_size = io_packet_size job_params . show_progress = show_progress job_params . profiling = profiling job_params . tasks_in_queue_per_pu = tasks_in_queue_per_pu job_params . load_sparsity_threshold = load_sparsity_threshold job_params . boundary_condition = ( self . protobufs . BulkJobParameters . REPEAT_EDGE ) job_params . memory_pool_config . pinned_cpu = False if cpu_pool is not None : job_params . memory_pool_config . cpu . use_pool = True if cpu_pool [ 0 ] == 'p' : job_params . memory_pool_config . pinned_cpu = True cpu_pool = cpu_pool [ 1 : ] size = self . _parse_size_string ( cpu_pool ) job_params . memory_pool_config . cpu . free_space = size if gpu_pool is not None : job_params . memory_pool_config . gpu . use_pool = True size = self . _parse_size_string ( gpu_pool ) job_params . memory_pool_config . gpu . free_space = size job_params . load_to_disk = load_to_disk self . _try_rpc ( lambda : self . _master . NewJob ( job_params ) ) while True : try : result = self . _master . IsJobDone ( self . protobufs . Empty ( ) ) except grpc . RpcError as e : raise ScannerException ( e ) if result . finished : break else : time . sleep ( 1.0 ) pass if not result . result . success : raise ScannerException ( result . result . msg ) self . _cached_db_metadata = None db_meta = self . _load_db_metadata ( ) job_id = None for job in db_meta . bulk_jobs : if job . name == job_name : job_id = job . id if job_id is None : raise ScannerException ( 'Internal error: job id not found after run' ) return [ self . table ( t ) for t in job_output_table_names ]
Runs a computation over a set of inputs.
Runs a computation over a set of inputs .
"""
        Runs a computation over a set of inputs.

        Args:
            tasks: The set of inputs to run the computation on. If tasks is a
                   Collection, then the computation is run on all frames of all
                   tables in the collection. Otherwise, tasks should be generated
                   by the Sampler.
            outputs: TODO(wcrichto)

        Kwargs:
            output_collection: If this is not None, then a new collection with
                               this name will be created for all the output
                               tables.
            force: TODO(wcrichto)
            work_item_size: TODO(wcrichto)
            io_item_size: TODO(wcrichto)
            cpu_pool: TODO(wcrichto)
            gpu_pool: TODO(wcrichto)
            pipeline_instances_per_node: TODO(wcrichto)
            show_progress: TODO(wcrichto)

        Returns:
            Either the output Collection if output_collection is specified
            or a list of Table objects.
        """ # Collect compression annotations to add to job # FIXME(apoms): op.name() is unbound # Delete tables if they exist and force was specified # Run the job # Qian: no sleep!! # Invalidate db metadata because of job run
Either the output Collection if output_collection is specified
or a list of Table objects.
Either the output Collection if output_collection is specified or a list of Table objects .
self
bulk_job
force
work_packet_size
io_packet_size
cpu_pool
gpu_pool
pipeline_instances_per_node
show_progress
profiling
load_sparsity_threshold
tasks_in_queue_per_pu
load_to_disk
tasks
The set of inputs to run the computation on. If tasks is a
Collection, then the computation is run on all frames of all
tables in the collection. Otherwise, tasks should be generated
by the Sampler.
The set of inputs to run the computation on . If tasks is a Collection then the computation is run on all frames of all tables in the collection . Otherwise tasks should be generated by the Sampler .
outputs
None