`fetch` can be called recursively in the `yield`ed block, so must
 make sure not to try to lock the Mutex if it's already locked by the
 current thread, as this would result in an error.
357072
def fetch(included_filename)
      if source = @cache[included_filename]
        return source
      end

      if @m.locked? && @owns_lock == Thread.current
        @cache[included_filename] = yield
      else
        ret = nil
        @cache[included_filename] = @m.synchronize do
          begin
            @owns_lock = Thread.current
            ret = yield
          ensure
            @owns_lock = nil
          end
        end
        ret
      end
    end
1040983
def transaction(force_sync = false, &block)
      # Ruby 1.9.3 does not support @mutex.owned?
      if @mutex.respond_to?(:owned?)
        force_sync = false if @mutex.locked? && @mutex.owned?
      else
        # If we allowed this in Ruby 1.9.3, it might possibly cause recursive
        # locking within the same thread.
        force_sync = false
      end
      if !force_sync && (@in_transaction || options[:without_mutex])
        block.call self
      else
        @mutex.synchronize do
          @in_transaction = true
          result = block.call
          @in_transaction = false
          result
        end
      end
    end
1144687
def fetch(**kwargs):
    '''
    
    '''
    # fetch continues when no controlling terminal is present
    pre = ''
    post = ''
    run_args = {}
    if float(__grains__['osrelease']) >= 10.2:
        post += '--not-running-from-cron'
    else:
        pre += ' env PAGER=cat'
        run_args['python_shell'] = True
    return _wrapper('fetch', pre=pre, post=post, run_args=run_args, **kwargs)
775118
def _fetch(self, request: Request, is_file: bool):
        '''
        '''
        _logger.info(_('Fetching ‘{url}’.'), url=request.url)

        self._item_session.request = request
        response = None

        try:
            with self._processor.ftp_client.session() as session:
                if is_file:
                    response = yield from session.start(request)
                else:
                    response = yield from session.start_listing(request)

                self._item_session.response = response

                action = self._result_rule.handle_pre_response(
                    self._item_session
                )

                if action in (Actions.RETRY, Actions.FINISH):
                    raise HookPreResponseBreak()

                self._file_writer_session.process_response(response)

                if not response.body:
                    response.body = Body(
                        directory=self._item_session.app_session.root_path,
                        hint='resp_cb')

                duration_timeout = self._fetch_rule.duration_timeout

                if is_file:
                    yield from session.download(
                        response.body, duration_timeout=duration_timeout)
                else:
                    yield from session.download_listing(
                        response.body, duration_timeout=duration_timeout)

        except HookPreResponseBreak:
            if response:
                response.body.close()

        except REMOTE_ERRORS as error:
            self._log_error(request, error)

            self._result_rule.handle_error(self._item_session, error)

            wait_time = self._result_rule.get_wait_time(
                self._item_session, error=error
            )

            if response:
                response.body.close()

            return wait_time
        else:
            self._log_response(request, response)
            self._handle_response(request, response)

            wait_time = self._result_rule.get_wait_time(
                self._item_session
            )

            if is_file and \
                    self._processor.fetch_params.preserve_permissions and \
                    hasattr(response.body, 'name'):
                yield from self._apply_unix_permissions(request, response)

            response.body.close()

            return wait_time
492748
async def fetch(self, n, *, timeout=None):
        
        """
        self._check_ready()
        if n <= 0:
            raise exceptions.InterfaceError('n must be greater than zero')
        if self._exhausted:
            return []
        recs = await self._exec(n, timeout)
        if len(recs) < n:
            self._exhausted = True
        return recs
494731
async def iter_lines(self, chunk_size=1024):
        
        pending = b''
        async for chunk in self.iter_chunks(chunk_size):
            lines = (pending + chunk).splitlines(True)
            for line in lines[:-1]:
                await yield_(line.splitlines()[0])
            pending = lines[-1]
        if pending:
            await yield_(pending.splitlines()[0])
183589
private void fetchResultBatchWithRetry(RecordSetList<JsonElement> rs)
      throws AsyncApiException, DataRecordException, IOException {
    boolean success = false;
    int retryCount = 0;
    int recordCountBeforeFetch = this.bulkRecordCount;

    do {
      try {
        // reinitialize the reader to establish a new connection to handle transient network errors
        if (retryCount > 0) {
          reinitializeBufferedReader();
        }

        // on retries there may already be records in rs, so pass the number of records as the initial count
        fetchResultBatch(rs, this.bulkRecordCount - recordCountBeforeFetch);
        success = true;
      } catch (IOException e) {
        if (retryCount < this.fetchRetryLimit) {
          log.info("Exception while fetching data, retrying: " + e.getMessage(), e);
          retryCount++;
        } else {
          log.error("Exception while fetching data: " + e.getMessage(), e);
          throw e;
        }
      }
    } while (!success);
  }
689399
public ByteBuffer fetch(long position) throws InterruptedException, ExecutionException {
    long blockIndex = position / bufSize;
    boolean goingBack = false;
    for (WorkUnit w : full) {
      if (w.blockIndex == blockIndex) {
        ensureFetching(blockIndex + 1);
        nbHit++;
        return w.buf;
      } else if (w.blockIndex > blockIndex) {
        goingBack = true;
      }
    }
    if (goingBack) {
      // user is asking for a block with a lower index than we've already fetched -
      // in other words they are not following the expected pattern of increasing indexes.
      nbGoingBack++;
    }
    if (null == fetching) {
      ensureFetching(blockIndex);
    }
    WorkUnit candidate = fetching;
    // block until we have the buffer
    ByteBuffer buf = candidate.getBuf();
    full.add(candidate);
    fetching = null;
    if (candidate.blockIndex == blockIndex) {
      // this is who we were waiting for
      nbNearHit++;
      ensureFetching(blockIndex + 1);
      return buf;
    } else {
      // wrong block. Let's fetch the right one now.
      nbMiss++;
      ensureFetching(blockIndex);
      candidate = fetching;
      buf = candidate.getBuf();
      full.add(candidate);
      fetching = null;
      ensureFetching(blockIndex + 1);
      return buf;
    }
  }
1280355
async def item(source, index):
    
    # Prepare
    if index >= 0:
        source = skip.raw(source, index)
    else:
        source = takelast(source, abs(index))
    async with streamcontext(source) as streamer:
        # Get first item
        try:
            result = await anext(streamer)
        except StopAsyncIteration:
            raise IndexError("Index out of range")
        # Check length
        if index < 0:
            count = 1
            async for _ in streamer:
                count += 1
            if count != abs(index):
                raise IndexError("Index out of range")
        # Yield result
        yield result
627027
def can_fetch(self, request: Request, file=None) -> bool:
        '''
        '''
        try:
            return self.can_fetch_pool(request)
        except NotInPoolError:
            pass

        yield from self.fetch_robots_txt(request, file=file)

        return self.can_fetch_pool(request)
918920
def fetch_values(*keys, **options)
      return super if options.include?(:raw)
      new_expires = expires_at(options, nil)
      options = Utils.without(options, :expires)
      substituted = {}
      block = if block_given?
                lambda do |key|
                  substituted[key] = true
                  yield key
                end
              end

      with_updates(options) do |updates|
        keys.zip(@adapter.fetch_values(*keys, **options, &block)).map do |key, entry|
          next entry if substituted[key]
          entry = invalidate_entry(key, entry, new_expires) do |new_entry|
            updates[key] = new_entry
          end
          if entry.nil?
            value = if block_given?
                      yield key
                    end
          else
            value, _ = entry
          end
          value
        end
      end
    end
1361569
def get_results_generator(future_list, timeout=None, sort_by_completed=False):
    """"""
    try:
        # python2 not support yield from
        if sort_by_completed:
            for future in as_completed(future_list, timeout=timeout):
                yield future.x
        else:
            for future in future_list:
                yield future.x
    except TimeoutError:
        return
768511
private void enterExclusiveThread() {
		// we use simple get, check, set here, rather than CAS
		// we don't need lock-style correctness, this is only a sanity-check and we thus
		// favor speed at the cost of some false negatives in this check
		Thread previous = currentThread;
		Thread thisThread = Thread.currentThread();

		if (previous == null) {
			currentThread = thisThread;
		}
		else if (previous != thisThread) {
			throw new IllegalStateException(
				"Concurrent access to KryoSerializer. Thread 1: " + thisThread.getName() +
					" , Thread 2: " + previous.getName());
		}
	}
28577
function flow(fn) {
  if (!isGenerator(fn)) {
    throw new Error(NOT_GENERATOR);
  }
  var gen = fn();
  next();
  function next(e, v) {
    var r = e ? gen.throw(e) : gen.next(v);
    if (!r.done) {
      if (isThunk(r.value)) {
        r.value(function(e, v){next(e, v);});
      } else if (isThunkArray(r.value)) {
        parallel(r.value, next);
      } else {
        next(new Error(NOT_THUNK));
      }
    }
  }
}
821334
private synchronized void initiateRetry() {
    retryCount += 1;
    currentListener = new RetryingBlockFetchListener();

    logger.info("Retrying fetch ({}/{}) for {} outstanding blocks after {} ms",
      retryCount, maxRetries, outstandingBlocksIds.size(), retryWaitTime);

    executorService.submit(() -> {
      Uninterruptibles.sleepUninterruptibly(retryWaitTime, TimeUnit.MILLISECONDS);
      fetchAllOutstanding();
    });
  }
508630
function (fn) { var cache = {}
        return function (value, then) {
            if (value in cache) {                   //  there's a flaw: cache updates after fetch completes, so while fetch is running,
                then (cache[value]) }               //  any subsequent call (until cache is ready) will trigger fetch (as it doesnt know that result is already fetching)
            else {
                fn.call (this, value, function (result) {
                    then (cache[value] = result) }) } } }
1219332
def get(self, exclude_recurse=None):
        
        reqs = Queue()
        pending = 1
        path = self.path
        zk = self.zk

        def child_of(path):
            return zk.get_children_async(path)

        def dispatch(path):
            return Request(path, child_of(path))

        stat = zk.exists(path)
        if stat is None or stat.numChildren == 0:
            return

        reqs.put(dispatch(path))

        while pending:
            req = reqs.get()

            try:
                children = req.value
                for child in children:
                    cpath = os.path.join(req.path, child)
                    if exclude_recurse is None or exclude_recurse not in child:
                        pending += 1
                        reqs.put(dispatch(cpath))
                    yield cpath
            except (NoNodeError, NoAuthError): pass

            pending -= 1