Select the streams from `read_streams` that are ready for reading, and
    streams from `write_streams` ready for writing.

    Uses `select.select()` internally but only returns two lists of ready streams.
969962
def select(read_streams, write_streams, timeout=0):
    

    exception_streams = []

    try:
        return builtin_select.select(
            read_streams,
            write_streams,
            exception_streams,
            timeout,
        )[0:2]
    except builtin_select.error as e:
        # POSIX signals interrupt select()
        no = e.errno if six.PY3 else e[0]
        if no == errno.EINTR:
            return ([], [])
        else:
            raise e
291794
def ReadList(self, *branches, **kwargs):
        
        branches = map(lambda x: E.Read(x) if isinstance(x, str) else x, branches)

        return self.List(*branches, **kwargs)
101818
def items(self, *args, **kwargs):
        
        return self.get_stream()(self.get_object(*args, **kwargs))
1349127
def read_stream_stats(self):
        
        stream_stats = OrderedDict()
        for stream in self.streams.values():
            stream_stats[stream] = stream.read_stats()
        return stream_stats
21984
def readstream(self, stream):
        

        self._reset()

        try:
            # tokenize input stream
            self._lexer = SettingLexer()
            self._lexer.readstream(stream)

            # parse tokens into AST
            self._parse()
            return True

        except IOError:
            self._reset()
            return False
442727
def select(sockets, remain=None):
        
        if remain is not None:
            max_timeout = remain / len(sockets)
            for s in sockets:
                if s.timeout > max_timeout:
                    s.timeout = max_timeout

        # python-can sockets aren't selectable, so we return all of them
        # sockets, None (means use the socket's recv() )
        return sockets, None
1295142
def select(from: nil, to: nil, &block)
      retval = []

      each_with_depth(from: from, to: to) do |line, depth, index|
        retval << line if (block_given? == false || block.call(line, depth, index))
      end

      retval
    end
411715
public static <T> Collection<T> selectInstancesOf(Iterable<?> iterable, Class<T> clazz)
    {
        if (iterable instanceof MutableCollection)
        {
            return ((MutableCollection<?>) iterable).selectInstancesOf(clazz);
        }
        if (iterable instanceof ArrayList)
        {
            return ArrayListIterate.selectInstancesOf((ArrayList<?>) iterable, clazz);
        }
        if (iterable instanceof List)
        {
            return ListIterate.selectInstancesOf((List<?>) iterable, clazz);
        }
        if (iterable instanceof Collection)
        {
            return IterableIterate.selectInstancesOf(
                    iterable,
                    clazz,
                    DefaultSpeciesNewStrategy.INSTANCE.<T>speciesNew((Collection<?>) iterable));
        }
        if (iterable != null)
        {
            return IterableIterate.selectInstancesOf(iterable, clazz);
        }
        throw new IllegalArgumentException("Cannot perform a selectInstancesOf on null");
    }
727385
def do_select(opts, &block)

      skip = opts[:offset] || opts[:skip]
      limit = opts[:limit]
      count = opts[:count]

      hwis = fetch_all({})
      hwis = hwis.select(&block)

      hwis = hwis[skip..-1] if skip
      hwis = hwis[0, limit] if limit

      return hwis.size if count

      hwis.collect { |hwi| Ruote::Workitem.new(hwi) }
    end
636357
def trigger_streamer(*inputs, **kwargs):
    

    streamer_marker = kwargs['mark_streamer']

    try:
        reading = inputs[1].pop()
    except StreamEmptyError:
        return []
    finally:
        for input_x in inputs:
            input_x.skip_all()

    try:
        streamer_marker(reading.value)
    except ArgumentError:
        return []

    return [IOTileReading(0, 0, 0)]
131771
def streams(self):
        """"""
        if self._streams_api is None:
            self._streams_api = self.get_streams_api()
        return self._streams_api
935545
def _read_select_poll(poll):
        
        npipes = len(NonBlockingStreamReader._streams)
        for stream_descr in NonBlockingStreamReader._streams:
            if not stream_descr.has_error:
                poll.register(stream_descr.stream,
                              select.POLLIN | select.POLLERR | select.POLLHUP | select.POLLNVAL)

        while NonBlockingStreamReader._run_flag:
            for (file_descr, event) in poll.poll(500):
                if event == select.POLLIN:
                    NonBlockingStreamReader._read_fd(file_descr)
                else:
                    # Because event != select.POLLIN, the pipe is closed
                    # but we still want to read all bytes
                    while NonBlockingStreamReader._read_fd(file_descr) != 0:
                        pass
                    # Dut died, signal the processing thread so it notices that no lines coming in
                    stream_descr = NonBlockingStreamReader._get_sd(file_descr)
                    if stream_descr is None:
                        return # PIPE closed but DUT already disappeared
                    stream_descr.has_error = True
                    if stream_descr.callback is not None:
                        stream_descr.callback()
                        return # Force poll object to reregister only alive descriptors

            # Check if new pipes added, don't need mutext just for reading the size
            # If we will not get it right now, we will at next time
            if npipes != len(NonBlockingStreamReader._streams):
                return
296197
private function streamSelect(array &$read, array &$write, $timeout)
    {
        if ($read || $write)
        {
            $except = null;

            return @stream_select($read, $write, $except, $timeout === null ? null : 0, $timeout);
        }

        usleep($timeout);

        return 0;
    }
223273
def list_streams(region=None, key=None, keyid=None, profile=None):
    '''
    
    '''
    conn = _get_conn(region=region, key=key, keyid=keyid, profile=profile)
    streams = []
    exclusive_start_stream_name = ''
    while exclusive_start_stream_name is not None:
        args = {'ExclusiveStartStreamName': exclusive_start_stream_name} if exclusive_start_stream_name else {}
        ret = _execute_with_retries(conn, 'list_streams', **args)
        if 'error' in ret:
            return ret
        ret = ret['result'] if ret and ret.get('result') else {}
        streams += ret.get('StreamNames', [])
        exclusive_start_stream_name = streams[-1] if ret.get('HasMoreStreams', False) in (True, 'true') else None
    return {'result': streams}
713820
def select(query, data=None):
    
    connection = _State.connection()
    _State.new_transaction()
    if data is None:
        data = []

    result = connection.execute('select ' + query, data)

    rows = []
    for row in result:
        rows.append(dict(list(row.items())))

    return rows
95485
def _run_select(self):
        
        return self._connection.select(
            self.to_sql(), self.get_bindings(), not self._use_write_connection
        )
74004
def select(self, Class, set=None, recursive=True,  ignore=True):
        """"""
        if self.mode == Mode.MEMORY:
            for t in self.data:
                if Class.__name__ == 'Text':
                    yield t
                else:
                    for e in t.select(Class,set,recursive,ignore):
                        yield e