qnxnto.py (1905B)
1 #!/usr/bin/env python 2 # encoding: utf-8 3 # Jérôme Carretero 2011 (zougloub) 4 # QNX neutrino compatibility functions 5 6 import sys, os 7 from waflib import Utils 8 9 class Popen(object): 10 """ 11 Popen cannot work on QNX from a threaded program: 12 Forking in threads is not implemented in neutrino. 13 14 Python's os.popen / spawn / fork won't work when running in threads (they will if in the main program thread) 15 16 In waf, this happens mostly in build. 17 And the use cases can be replaced by os.system() calls. 18 """ 19 __slots__ = ["prog", "kw", "popen", "verbose"] 20 verbose = 0 21 def __init__(self, prog, **kw): 22 try: 23 self.prog = prog 24 self.kw = kw 25 self.popen = None 26 if Popen.verbose: 27 sys.stdout.write("Popen created: %r, kw=%r..." % (prog, kw)) 28 29 do_delegate = kw.get('stdout') == -1 and kw.get('stderr') == -1 30 if do_delegate: 31 if Popen.verbose: 32 print("Delegating to real Popen") 33 self.popen = self.real_Popen(prog, **kw) 34 else: 35 if Popen.verbose: 36 print("Emulating") 37 except Exception as e: 38 if Popen.verbose: 39 print("Exception: %s" % e) 40 raise 41 42 def __getattr__(self, name): 43 if Popen.verbose: 44 sys.stdout.write("Getattr: %s..." % name) 45 if name in Popen.__slots__: 46 return object.__getattribute__(self, name) 47 else: 48 if self.popen is not None: 49 if Popen.verbose: 50 print("from Popen") 51 return getattr(self.popen, name) 52 else: 53 if name == "wait": 54 return self.emu_wait 55 else: 56 raise Exception("subprocess emulation: not implemented: %s" % name) 57 58 def emu_wait(self): 59 if Popen.verbose: 60 print("emulated wait (%r kw=%r)" % (self.prog, self.kw)) 61 if isinstance(self.prog, str): 62 cmd = self.prog 63 else: 64 cmd = " ".join(self.prog) 65 if 'cwd' in self.kw: 66 cmd = 'cd "%s" && %s' % (self.kw['cwd'], cmd) 67 return os.system(cmd) 68 69 if sys.platform == "qnx6": 70 Popen.real_Popen = Utils.subprocess.Popen 71 Utils.subprocess.Popen = Popen 72