_pipe_queue, if set to Queue is a CPU hog

[ghost]

Code like in the example below (simplified) will consume 4-5% of CPU, which means that you can barely have 25 processes running at the same time...

from sh import sleep

for line in sleep('infinity', _iter='out', _err_to_out=True):
    pass

The 25 processes is actually an optimistic prognosis, you'd still need to account for Python's process creation tax and such. So, more like 20...

Here's an example profiler output:

         13108795 function calls in 600.691 seconds

   Ordered by: standard name

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
        1    0.000    0.000  600.691  600.691 <string>:1(<module>)
        1    0.000    0.000    0.000    0.000 __init__.py:1169(__init__)
        2    0.000    0.000    0.000    0.000 __init__.py:1175(append)
       10    0.000    0.000    0.000    0.000 __init__.py:1221(getLogger)
        5    0.000    0.000    0.000    0.000 __init__.py:1272(_fixupParents)
        5    0.000    0.000    0.000    0.000 __init__.py:1341(__init__)
        8    0.000    0.000    0.000    0.000 __init__.py:1361(debug)
        6    0.000    0.000    0.000    0.000 __init__.py:1373(info)
        2    0.000    0.000    0.000    0.000 __init__.py:1605(getEffectiveLevel)
       14    0.000    0.000    0.000    0.000 __init__.py:1619(isEnabledFor)
        5    0.000    0.000    0.000    0.000 __init__.py:187(_checkLevel)
       10    0.000    0.000    0.000    0.000 __init__.py:1930(getLogger)
       14    0.000    0.000    0.000    0.000 __init__.py:212(_acquireLock)
       14    0.000    0.000    0.000    0.000 __init__.py:221(_releaseLock)
        4    0.000    0.000    0.000    0.000 __init__.py:253(_at_fork_weak_calls)
        2    0.000    0.000    0.000    0.000 __init__.py:264(_before_at_fork_weak_calls)
        2    0.000    0.000    0.000    0.000 __init__.py:268(_after_at_fork_weak_calls)
        1    0.000    0.000    0.000    0.000 __init__.py:274(load)
        1    0.000    0.000    0.000    0.000 __init__.py:299(loads)
        5    0.000    0.000    0.000    0.000 __init__.py:727(__init__)
        4    0.000    0.000    0.000    0.000 __init__.py:849(acquire)
        4    0.000    0.000    0.000    0.000 __init__.py:856(release)
        2    0.000    0.000    0.000    0.000 _bootlocale.py:33(getpreferredencoding)
        4    0.000    0.000    0.000    0.000 _weakrefset.py:16(__init__)
        4    0.000    0.000    0.000    0.000 _weakrefset.py:20(__enter__)
        4    0.000    0.000    0.000    0.000 _weakrefset.py:26(__exit__)
        1    0.000    0.000    0.000    0.000 _weakrefset.py:38(_remove)
        4    0.000    0.000    0.000    0.000 _weakrefset.py:52(_commit_removals)
       12    0.000    0.000    0.000    0.000 _weakrefset.py:58(__iter__)
        6    0.000    0.000    0.000    0.000 _weakrefset.py:81(add)
        2    0.000    0.000    0.000    0.000 codecs.py:186(__init__)
        1    0.000    0.000    0.000    0.000 decoder.py:332(decode)
        1    0.000    0.000    0.000    0.000 decoder.py:343(raw_decode)
        1    0.000    0.000    0.000    0.000 fio.py:37(get_io_depth)
        1    0.000    0.000  600.691  600.691 fio.py:48(run_test)
        1    0.000    0.000    0.000    0.000 fio.py:56(<listcomp>)
        1    0.000    0.000    0.000    0.000 posixpath.py:144(basename)
        3    0.000    0.000    0.000    0.000 posixpath.py:41(_get_sep)
        2    0.000    0.000    0.000    0.000 posixpath.py:75(join)
        2    0.000    0.000    0.000    0.000 pty.py:21(openpty)
   546263    5.439    0.000  598.483    0.001 queue.py:153(get)
       10    0.000    0.000    0.000    0.000 queue.py:205(_init)
  1092203    0.801    0.000    1.152    0.000 queue.py:208(_qsize)
      325    0.000    0.000    0.000    0.000 queue.py:216(_get)
       10    0.000    0.000    0.000    0.000 queue.py:33(__init__)
       10    0.000    0.000    0.000    0.000 sh.py:1005(ob_is_pipe)
        2    0.000    0.000    0.000    0.000 sh.py:1014(tty_in_validator)
        2    0.000    0.000    0.000    0.000 sh.py:1026(bufsize_validator)
       10    0.000    0.000    0.000    0.000 sh.py:1213(__getattribute__)
        2    0.000    0.000    0.000    0.000 sh.py:1236(_extract_call_args)
        2    0.000    0.000    0.006    0.003 sh.py:1324(__call__)
        2    0.000    0.000    0.000    0.000 sh.py:1430(compile_args)
        2    0.000    0.000    0.000    0.000 sh.py:1471(aggregate_keywords)
        4    0.000    0.000    0.001    0.000 sh.py:1537(_start_daemon_thread)
        2    0.000    0.000    0.000    0.000 sh.py:1629(get_exc_exit_code_would_raise)
        1    0.000    0.000    0.000    0.000 sh.py:1646(handle_process_exit_code)
        1    0.000    0.000    0.000    0.000 sh.py:1660(no_interrupt)
        2    0.000    0.000    0.005    0.003 sh.py:1690(__init__)
        2    0.000    0.000    0.000    0.000 sh.py:2199(__repr__)
        1    0.000    0.000    0.000    0.000 sh.py:2329(_process_just_ended)
        2    0.000    0.000    0.001    0.001 sh.py:2346(wait)
        3    0.000    0.000    0.000    0.000 sh.py:2654(bufsize_type_to_bufsize)
        3    0.000    0.000    0.000    0.000 sh.py:268(encode_to_py3bytes_or_py2str)
        3    0.000    0.000    0.000    0.000 sh.py:2777(determine_how_to_feed_output)
        3    0.000    0.000    0.000    0.000 sh.py:2871(__init__)
        3    0.000    0.000    0.000    0.000 sh.py:2956(__init__)
       10    0.000    0.000    0.000    0.000 sh.py:606(__init__)
       14    0.000    0.000    0.000    0.000 sh.py:611(_format_msg)
       12    0.000    0.000    0.000    0.000 sh.py:616(set_context)
        4    0.000    0.000    0.000    0.000 sh.py:621(get_child)
        6    0.000    0.000    0.000    0.000 sh.py:627(info)
        8    0.000    0.000    0.000    0.000 sh.py:630(debug)
        4    0.000    0.000    0.000    0.000 sh.py:640(default_logger_str)
        2    0.000    0.000    0.006    0.003 sh.py:680(__init__)
        2    0.000    0.000    0.000    0.000 sh.py:695(<listcomp>)
        2    0.000    0.000    0.001    0.001 sh.py:777(wait)
        2    0.000    0.000    0.000    0.000 sh.py:805(handle_command_exit_code)
       24    0.000    0.000    0.000    0.000 sh.py:81(callable)
        1    0.000    0.000    0.000    0.000 sh.py:844(__iter__)
      325    2.201    0.007  600.684    1.848 sh.py:847(next)
        4    0.000    0.000    0.000    0.000 sh.py:947(output_redirect_is_filename)
        2    0.000    0.000    0.000    0.000 sh.py:951(get_prepend_stack)
        2    0.000    0.000    0.000    0.000 sh.py:958(special_kwarg_validator)
       20    0.000    0.000    0.000    0.000 sh.py:978(get_fileno)
       10    0.000    0.000    0.000    0.000 sh.py:997(ob_is_tty)
        4    0.000    0.000    0.000    0.000 threading.py:1000(join)
        4    0.000    0.000    0.000    0.000 threading.py:1038(_wait_for_tstate_lock)
        6    0.000    0.000    0.000    0.000 threading.py:1096(daemon)
        4    0.000    0.000    0.000    0.000 threading.py:1112(daemon)
        2    0.000    0.000    0.000    0.000 threading.py:1143(__init__)
        2    0.000    0.000    0.000    0.000 threading.py:1151(cancel)
       10    0.000    0.000    0.000    0.000 threading.py:1206(current_thread)
       42    0.000    0.000    0.000    0.000 threading.py:216(__init__)
   546273    0.294    0.000    0.478    0.000 threading.py:240(__enter__)
   546273    0.512    0.000    0.670    0.000 threading.py:243(__exit__)
   545946    0.264    0.000    0.358    0.000 threading.py:249(_release_save)
   545946    0.658    0.000    1.216    0.000 threading.py:252(_acquire_restore)
   546275    0.291    0.000    0.611    0.000 threading.py:255(_is_owned)
   545946    3.633    0.000  590.279    0.001 threading.py:264(wait)
      329    0.000    0.000    0.001    0.000 threading.py:335(notify)
        4    0.000    0.000    0.000    0.000 threading.py:358(notify_all)
       12    0.000    0.000    0.000    0.000 threading.py:499(__init__)
       14    0.000    0.000    0.000    0.000 threading.py:507(is_set)
        4    0.000    0.000    0.000    0.000 threading.py:513(set)
        6    0.000    0.000    0.001    0.000 threading.py:534(wait)
        2    0.000    0.000    0.000    0.000 threading.py:728(_newname)
        3    0.000    0.000    0.000    0.000 threading.py:75(RLock)
        6    0.000    0.000    0.000    0.000 threading.py:758(__init__)
        6    0.000    0.000    0.001    0.000 threading.py:829(start)
        4    0.000    0.000    0.000    0.000 threading.py:968(_stop)
        1    0.000    0.000    0.000    0.000 {built-in method _csv.writer}
        2    0.000    0.000    0.000    0.000 {built-in method _locale.nl_langinfo}
   545972    0.293    0.000    0.293    0.000 {built-in method _thread.allocate_lock}
       10    0.000    0.000    0.000    0.000 {built-in method _thread.get_ident}
        6    0.000    0.000    0.000    0.000 {built-in method _thread.start_new_thread}
        1    0.000    0.000  600.691  600.691 {built-in method builtins.exec}
       30    0.000    0.000    0.000    0.000 {built-in method builtins.getattr}
       29    0.000    0.000    0.000    0.000 {built-in method builtins.hasattr}
       73    0.000    0.000    0.000    0.000 {built-in method builtins.isinstance}
  1092208    0.351    0.000    0.351    0.000 {built-in method builtins.len}
        2    0.000    0.000    0.000    0.000 {built-in method builtins.repr}
        2    0.000    0.000    0.000    0.000 {built-in method gc.disable}
        2    0.000    0.000    0.000    0.000 {built-in method gc.enable}
        2    0.000    0.000    0.000    0.000 {built-in method gc.isenabled}
        2    0.000    0.000    0.000    0.000 {built-in method io.open}
        1    0.000    0.000    0.000    0.000 {built-in method posix.WEXITSTATUS}
        1    0.000    0.000    0.000    0.000 {built-in method posix.WIFEXITED}
        1    0.000    0.000    0.000    0.000 {built-in method posix.WIFSIGNALED}
       16    0.000    0.000    0.000    0.000 {built-in method posix.close}
        2    0.000    0.000    0.000    0.000 {built-in method posix.dup}
        2    0.001    0.000    0.001    0.001 {built-in method posix.fork}
        3    0.000    0.000    0.000    0.000 {built-in method posix.fspath}
        2    0.000    0.000    0.000    0.000 {built-in method posix.openpty}
        7    0.000    0.000    0.000    0.000 {built-in method posix.pipe}
        4    0.001    0.000    0.001    0.000 {built-in method posix.read}
        1    0.000    0.000    0.000    0.000 {built-in method posix.waitpid}
  1638141    0.466    0.000    0.466    0.000 {built-in method time.monotonic}
        2    0.000    0.000    0.000    0.000 {built-in method time.time}
   546273    0.184    0.000    0.184    0.000 {method '__enter__' of '_thread.lock' objects}
   546273    0.157    0.000    0.157    0.000 {method '__exit__' of '_thread.lock' objects}
       18    0.000    0.000    0.000    0.000 {method 'acquire' of '_thread.RLock' objects}
  2184117  584.668    0.000  584.668    0.000 {method 'acquire' of '_thread.lock' objects}
       10    0.000    0.000    0.000    0.000 {method 'add' of 'set' objects}
   545946    0.122    0.000    0.122    0.000 {method 'append' of 'collections.deque' objects}
        5    0.000    0.000    0.000    0.000 {method 'append' of 'list' objects}
        6    0.000    0.000    0.000    0.000 {method 'copy' of 'dict' objects}
      332    0.000    0.000    0.000    0.000 {method 'decode' of 'bytes' objects}
        1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}
        1    0.000    0.000    0.000    0.000 {method 'discard' of 'set' objects}
        2    0.000    0.000    0.000    0.000 {method 'end' of 're.Match' objects}
        2    0.000    0.000    0.000    0.000 {method 'endswith' of 'str' objects}
        6    0.000    0.000    0.000    0.000 {method 'extend' of 'list' objects}
        2    0.000    0.000    0.000    0.000 {method 'format' of 'str' objects}
       10    0.000    0.000    0.000    0.000 {method 'get' of 'dict' objects}
        2    0.000    0.000    0.000    0.000 {method 'insert' of 'list' objects}
       12    0.000    0.000    0.000    0.000 {method 'issuperset' of 'set' objects}
        5    0.000    0.000    0.000    0.000 {method 'items' of 'dict' objects}
        2    0.000    0.000    0.000    0.000 {method 'join' of 'str' objects}
        2    0.000    0.000    0.000    0.000 {method 'keys' of 'dict' objects}
        4    0.000    0.000    0.000    0.000 {method 'locked' of '_thread.lock' objects}
        2    0.000    0.000    0.000    0.000 {method 'match' of 're.Pattern' objects}
        2    0.000    0.000    0.000    0.000 {method 'pop' of 'list' objects}
      325    0.000    0.000    0.000    0.000 {method 'popleft' of 'collections.deque' objects}
        2    0.000    0.000    0.000    0.000 {method 'read' of '_io.StringIO' objects}
       18    0.000    0.000    0.000    0.000 {method 'release' of '_thread.RLock' objects}
   545954    0.094    0.000    0.094    0.000 {method 'release' of '_thread.lock' objects}
   545942    0.257    0.000    0.257    0.000 {method 'remove' of 'collections.deque' objects}
        4    0.000    0.000    0.000    0.000 {method 'remove' of 'set' objects}
        8    0.000    0.000    0.000    0.000 {method 'replace' of 'str' objects}
       11    0.000    0.000    0.000    0.000 {method 'rfind' of 'str' objects}
        2    0.000    0.000    0.000    0.000 {method 'seek' of '_io.StringIO' objects}
        3    0.000    0.000    0.000    0.000 {method 'split' of 'str' objects}
       13    0.000    0.000    0.000    0.000 {method 'startswith' of 'str' objects}
        4    0.000    0.000    0.000    0.000 {method 'update' of 'dict' objects}
      324    0.000    0.000    0.000    0.000 {method 'write' of '_io.StringIO' objects}
        1    0.000    0.000    0.000    0.000 {method 'write' of '_io.TextIOWrapper' objects}
        2    0.000    0.000    0.000    0.000 {method 'writerow' of '_csv.writer' objects}

[amoffat]

Thanks for reporting. I've raised the queue.get timeout time by two orders of magnitude and made it configurable. This should reduce the cpu spinning significantly.