A new Worker customizes itself to its WorkUnit at instantiation.
 Return output to the central server, marking the WorkUnit done.
490697
def complete_work_unit(result)
      keep_trying_to "complete work unit" do
        data = base_params.merge({:status => 'succeeded', :output => result})
        @node.central["/work/#{data[:id]}"].put(data)
        log "finished #{display_work_unit} in #{data[:time]} seconds"
      end
    end
747359
def result_worker(ctx, result_cls, get_object=False):
    
    g = ctx.obj
    ResultWorker = load_cls(None, None, result_cls)

    result_worker = ResultWorker(resultdb=g.resultdb, inqueue=g.processor2result)

    g.instances.append(result_worker)
    if g.get('testing_mode') or get_object:
        return result_worker

    result_worker.run()
1018960
public WorkUnit createWorkUnit(SubTask execUnit) {
        WorkUnit wu = new WorkUnit(this, execUnit);
        workUnits.add(wu);
        return wu;
    }
1199370
def worker(self):
        
        if self._worker is None:
            self._worker = XOrbLookupWorker(self.isThreadEnabled())
            
            # create worker connections
            self.loadRequested.connect(self._worker.loadRecords)
            self.loadBatchRequested.connect(self._worker.loadBatch)
            self.loadColumnsRequested.connect(self._worker.loadColumns)
            
            self._worker.loadingStarted.connect(self.markLoadingStarted)
            self._worker.loadingFinished.connect(self.markLoadingFinished)
            self._worker.loadedRecords[object].connect(self._loadRecords)
            self._worker.loadedRecords[object, object].connect(self._loadRecords)
            self._worker.loadedGroup.connect(self.createGroupItem)
            self._worker.columnLoaded.connect(self._loadColumns)
            self._worker.connectionLost.connect(self._connectionLost)
            
        return self._worker
609962
def init_workers():
    """"""
    setproctitle('oq-worker')
    # unregister raiseMasterKilled in oq-workers to avoid deadlock
    # since processes are terminated via pool.terminate()
    signal.signal(signal.SIGTERM, signal.SIG_DFL)
    # prctl is still useful (on Linux) to terminate all spawned processes
    # when master is killed via SIGKILL
    try:
        import prctl
    except ImportError:
        pass
    else:
        # if the parent dies, the children die
        prctl.set_pdeathsig(signal.SIGKILL)
946098
def initialize_worker(self, process_num=None):
        
        for p in self.producers:
            p.initialize_worker(process_num)
        # self.initial_state.process = process_num
        self.random.seed(hash(self.seed) + hash(process_num))
740682
function (opts) {
  if (!(this instanceof Worker)) {
    return new Worker(opts);
  }

  this.opts = opts;
  this.port = this.opts.port || 8888;
  this.apps = opts.apps;

  if (typeof (this.apps) !== 'object') {
    this.apps = JSON.parse(this.apps);
  }

  this.server = http.createServer(this._handleHttp.bind(this));
  return this;
}
634959
def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None):
    
    from ambry.library import new_library
    from ambry.run import get_runconfig
    import traceback

    assert maxtasks is None or (type(maxtasks) == int and maxtasks > 0)

    put = outqueue.put
    get = inqueue.get

    if hasattr(inqueue, '_writer'):
        inqueue._writer.close()
        outqueue._reader.close()

    if initializer is not None:
        initializer(*initargs)

    try:
        task = get()
    except (EOFError, IOError):
        debug('worker got EOFError or IOError -- exiting')
        return

    if task is None:
        debug('worker got sentinel -- exiting')
        return

    job, i, func, args, kwds = task

    # func = mapstar = map(*args)

    # Since there is only one source build per process, we know the structure
    # of the args beforehand.
    mp_func = args[0][0]
    mp_args = list(args[0][1][0])

    library = new_library(get_runconfig())
    library.database.close()  # Maybe it is still open after the fork.
    library.init_debug()

    bundle_vid = mp_args[0]

    try:

        b = library.bundle(bundle_vid)
        library.logger = b.logger # So library logs to the same file as the bundle.

        b = b.cast_to_subclass()
        b.multi = True  # In parent it is a number, in child, just needs to be true to get the right logger template
        b.is_subprocess = True
        b.limited_run = bool(int(os.getenv('AMBRY_LIMITED_RUN', 0)))

        assert b._progress == None  # Don't want to share connections across processes

        mp_args[0] = b
        result = (True, [mp_func(*mp_args)])

    except Exception as e:
        import traceback
        tb = traceback.format_exc()
        b.error('Subprocess {} raised an exception: {}'.format(os.getpid(), e.message), False)
        b.error(tb, False)
        result = (False, e)

    assert result

    b.progress.close()
    library.close()

    try:
        put((job, i, result))
    except Exception as e:
        wrapped = MaybeEncodingError(e, result[1])
        debug("Possible encoding error while sending result: %s" % (wrapped))
        put((job, i, (False, wrapped)))
944230
def worker(compute, task_queue, result_queue, log_queue, complete,
               *context):
        """"""
        try:
            MapReduce._forked = True
            log.debug('Worker process starting...')

            configure_worker_logging(log_queue)

            for obj in iter(task_queue.get, POISON_PILL):
                if complete.is_set():
                    log.debug('Worker received signal - exiting early')
                    break

                log.debug('Worker got %s', obj)
                result_queue.put(compute(obj, *context))
                log.debug('Worker finished %s', obj)

            result_queue.put(POISON_PILL)
            log.debug('Worker process exiting')

        except Exception as e:  # pylint: disable=broad-except
            result_queue.put(ExceptionWrapper(e))
214777
def _get_and_start_work(self):
        ""
        worker_id = nice_identifier()
        work_unit = self.task_master.get_work(worker_id, available_gb=self.available_gb())
        if work_unit is None:
            return None, None
        async_result = self.pool.apply_async(
            run_worker,
            (HeadlessWorker, self.task_master.registry.config,
             worker_id,
             work_unit.work_spec_name,
             work_unit.key),
            callback=self._finish_callback)
        return async_result, work_unit
516522
def new_workunit(self, name, labels=None, cmd='', log_config=None):
    
    with self.run_tracker.new_workunit(name=name, labels=labels, cmd=cmd, log_config=log_config) as workunit:
      yield workunit
486309
def post_worker_init(worker):
    
    quit_command = 'CTRL-BREAK' if sys.platform == 'win32' else 'CONTROL-C'
    sys.stdout.write(
        "Django version {djangover}, Gunicorn version {gunicornver}, "
        "using settings {settings!r}\n"
        "Starting development server at {urls}\n"
        "Quit the server with {quit_command}.\n".format(
            djangover=django.get_version(),
            gunicornver=gunicorn.__version__,
            settings=os.environ.get('DJANGO_SETTINGS_MODULE'),
            urls=', '.join('http://{0}/'.format(b) for b in worker.cfg.bind),
            quit_command=quit_command,
        ),
    )
355108
def init_worker(inpaths, g_args):
    
    global global_args
    global srcs
    global_args = g_args
    srcs = [rasterio.open(i) for i in inpaths]
940441
def createWorkers(self, num_workers):
        
        for i in range(num_workers):
            self.workers.append(WorkerThread(self._requests_queue))
895103
def setupWorkerTransition():
    """"""

    default_warn_method = getWarningMethod()

    def custom_warn_method(message, category, stacklevel):
        if stacklevel is not None:
            stacklevel += 1
        if _WORKER_WARNING_MARK in message:
            # Message contains our mark - it's Worker API Renaming warning,
            # issue it appropriately.
            message = message.replace(_WORKER_WARNING_MARK, "")
            warnings.warn(
                DeprecatedWorkerNameWarning(message), message, stacklevel)
        else:
            # Other's warning message
            default_warn_method(message, category, stacklevel)

    setWarningMethod(custom_warn_method)
243374
def _start_new_worker_process(self, server_socket):
        """"""
        from multiprocessing import Process
        p = Process(target=forked_child, args=self._get_child_args(server_socket))
        p.start()
        return p
396897
async def create_server(self, worker, protocol_factory, address=None,
                            sockets=None, idx=0):
        '''
        '''
        cfg = self.cfg
        max_requests = cfg.max_requests
        if max_requests:
            max_requests = int(lognormvariate(log(max_requests), 0.2))
        server = self.server_factory(
            protocol_factory,
            loop=worker._loop,
            max_requests=max_requests,
            keep_alive=cfg.keep_alive,
            name=self.name,
            logger=self.logger,
            server_software=cfg.server_software,
            cfg=cfg,
            idx=idx
        )
        for event in ('connection_made', 'pre_request', 'post_request',
                      'connection_lost'):
            callback = getattr(cfg, event)
            if callback != pass_through:
                server.event(event).bind(callback)
        await server.start_serving(
            sockets=sockets,
            address=address,
            backlog=cfg.backlog,
            sslcontext=self.sslcontext()
        )
        return server