Process `limit` number of replica messages
542317
def service_replicas_outbox(self, limit: int = None) -> int:
        
        # TODO: rewrite this using Router

        num_processed = 0
        for message in self.replicas.get_output(limit):
            num_processed += 1
            if isinstance(message, (PrePrepare, Prepare, Commit, Checkpoint)):
                self.send(message)
            elif isinstance(message, Ordered):
                self.try_processing_ordered(message)
            elif isinstance(message, tuple) and isinstance(message[1], Reject):
                with self.metrics.measure_time(MetricsName.NODE_SEND_REJECT_TIME):
                    digest, reject = message
                    result_reject = Reject(
                        reject.identifier,
                        reject.reqId,
                        self.reasonForClientFromException(
                            reject.reason))
                    # TODO: What the case when reqKey will be not in requestSender dict
                    if digest in self.requestSender:
                        self.transmitToClient(result_reject, self.requestSender[digest])
                        self.doneProcessingReq(digest)
            elif isinstance(message, Exception):
                self.processEscalatedException(message)
            else:
                # TODO: should not this raise exception?
                logger.error("Received msg {} and don't "
                             "know how to handle it".format(message))
        return num_processed
1307675
function _reseed(callback) {
    if(ctx.pools[0].messageLength >= 32) {
      _seed();
      return callback();
    }
    // not enough seed data...
    var needed = (32 - ctx.pools[0].messageLength) << 5;
    ctx.seedFile(needed, function(err, bytes) {
      if(err) {
        return callback(err);
      }
      ctx.collect(bytes);
      _seed();
      callback();
    });
  }
258330
def run_bulk(t, first, last, incr = 1)
                retry_until { DRMAA.run_bulk_jobs(t.ptr, first, last, incr) }
            end
96871
async def serviceNodeMsgs(self, limit: int) -> int:
        
        with self.metrics.measure_time(MetricsName.SERVICE_NODE_STACK_TIME):
            n = await self.nodestack.service(limit, self.quota_control.node_quota)

        self.metrics.add_event(MetricsName.NODE_STACK_MESSAGES_PROCESSED, n)

        await self.processNodeInBox()
        return n
777412
def results(self, limit=100):
        
        limited = True if self.high_mark is not None else False
        rmax = self.high_mark - self.low_mark if limited else None
        rnum = 0

        params = self.get_params()
        params["offset"] = self.low_mark
        params["limit"] = limit

        while not limited and rmax is None or rnum < rmax:
            if limited or rmax is not None:
                rleft = rmax - rnum
                params["limit"] = rleft if rleft < limit else limit

            r = self.resource._meta.api.http_resource("GET", self.resource._meta.resource_name, params=params)
            data = self.resource._meta.api.resource_deserialize(r.text)

            if not limited:
                rmax = data["meta"]["total_count"]

            if data["meta"]["total_count"] < rmax:
                rmax = data["meta"]["total_count"]

            params["offset"] = data["meta"]["offset"] + data["meta"]["limit"]

            for item in data["objects"]:
                rnum += 1
                yield item
1060456
def bulk_write(self, metrics):
        
        try:
            for metric in metrics:
                self.producer.send(self.topic, metric)
            self.producer.flush()
        except (KafkaTimeoutError, NoBrokersAvailable) as exc:
            logger.warning('bulk_write metrics %r failure %r', metrics, exc)
524679
def reserve(limit: 1000)
      record_ids = Set.new
      while record_ids.size < limit && (record_id = Searchkick.with_redis { |r| r.rpop(redis_key) })
        record_ids << record_id
      end
      record_ids.to_a
    end
389184
def limit_mem(limit=(4 * 1024**3)):
    ""
    rsrc = resource.RLIMIT_DATA
    soft, hard = resource.getrlimit(rsrc)
    resource.setrlimit(rsrc, (limit, hard))  # 4GB
    softnew, _ = resource.getrlimit(rsrc)
    assert softnew == limit

    _log = logging.getLogger(__name__)
    _log.debug(': %s => %s', soft, softnew)
627911
def recv_loop(stream):
    """"""
    message = recv(stream)
    while message:
        yield message
        message = recv(stream)
383064
function collectMessage(message, processingQueue, cb) {
        const threshold = message.getRetryThreshold();
        const messageRetryThreshold = typeof threshold === 'number' ? threshold : dispatcher.getMessageRetryThreshold();

        const delay = message.getRetryDelay();
        const messageRetryDelay = typeof delay === 'number' ? delay : dispatcher.getMessageRetryDelay();

        let destQueueName = null;
        let delayed = false;
        let requeued = false;

        const multi = client.multi();

        /**
         * Only exceptions from non periodic messages are handled.
         * Periodic messages are ignored once they are delivered to a consumer.
         */
        if (!dispatcher.isPeriodic(message)) {
            const uuid = message.getId();

            /**
             * Attempts
             */
            const attempts = increaseAttempts(message);

            /**
             *
             */
            if (attempts < messageRetryThreshold) {
                debug(`Trying to consume message ID [${uuid}] again (attempts: [${attempts}]) ...`);
                if (messageRetryDelay) {
                    debug(`Scheduling message ID [${uuid}]  (delay: [${messageRetryDelay}])...`);
                    message.setScheduledDelay(messageRetryDelay);
                    dispatcher.schedule(message, multi);
                    delayed = true;
                } else {
                    debug(`Message ID [${uuid}] is going to be enqueued immediately...`);
                    destQueueName = keyQueueName;
                    requeued = true;
                }
            } else {
                debug(`Message ID [${uuid}] has exceeded max retry threshold...`);
                destQueueName = keyQueueNameDead;
            }

            /**
             *
             */
            if (destQueueName) {
                debug(`Moving message [${uuid}] to queue [${destQueueName}]...`);
                multi.lpush(destQueueName, message.toString());
            }
            multi.rpop(processingQueue);
            multi.exec((err) => {
                if (err) cb(err);
                else {
                    if (dispatcher.isTest()) {
                        if (requeued) dispatcher.emit(events.MESSAGE_REQUEUED, message);
                        else if (delayed) dispatcher.emit(events.MESSAGE_DELAYED, message);
                        else dispatcher.emit(events.MESSAGE_DEAD_LETTER, message);
                    }
                    cb();
                }
            });
        } else {
            client.rpop(processingQueue, cb);
        }
    }
1226467
def process_next_message(self, timeout):
        """"""
        message = self.worker_manager.receive(timeout)

        if isinstance(message, Acknowledgement):
            self.task_manager.task_start(message.task, message.worker)
        elif isinstance(message, Result):
            self.task_manager.task_done(message.task, message.result)
1369137
def update(self):
        
        ring = self._fetch()
        n_replicas = len(ring)
        replica_set = set([r[1] for r in self.replicas])
        self.ranges = []
        for n, (start, replica) in enumerate(ring):
            if replica in replica_set:
                end = ring[(n+1) % n_replicas][0] % RING_SIZE
                if start < end:
                    self.ranges.append((start, end))
                elif end < start:
                    self.ranges.append((start, RING_SIZE))
                    self.ranges.append((0, end))
                else:
                    self.ranges.append((0, RING_SIZE))
24059
def container_fs_limit_bytes(self, metric, scraper_config):
        
        pct_m_name = scraper_config['namespace'] + '.filesystem.usage_pct'
        if metric.type not in METRIC_TYPES:
            self.log.error("Metric type %s unsupported for metric %s" % (metric.type, metric.name))
            return
        self._process_limit_metric('', metric, self.fs_usage_bytes, scraper_config, pct_m_name)
463329
def limit(count)
      raise InvalidPageSize unless count && count > 0
      Specification.new(reader, result.dup { |r| r.count = count })
    end
635141
def queue(ctx, message):
    """"""
    if not valid_tweet(message):
        click.echo("Message is too long for twitter.")
        click.echo("Message: " + message)
        ctx.exit(2)

    if ctx.obj['DRYRUN']:
        click.echo("Message not queue due to dry-run mode.")
        ctx.exit(0)

    ctx.obj['TWEETLIST'].append(message)
92350
def set_recursion_limit(limit):
    """"""
    if limit < minimum_recursion_limit:
        raise CoconutException("--recursion-limit must be at least " + str(minimum_recursion_limit))
    sys.setrecursionlimit(limit)
1051072
def _request_bulk(self, urls: List[str]) -> List:
        """"""
        if not urls:
            raise Exception("No results were found")
        session: FuturesSession = FuturesSession(max_workers=len(urls))
        self.log.info("Bulk requesting: %d" % len(urls))
        futures = [session.get(u, headers=gen_headers(), timeout=3) for u in urls]
        done, incomplete = wait(futures)
        results: List = list()
        for response in done:
            try:
                results.append(response.result())
            except Exception as err:
                self.log.warn("Failed result: %s" % err)
        return results