1. 模块列表
  2. 函数列表
  3. os.fork1()

Python os 模块，fork1() 实例源码

我们从Python开源项目中，提取了以下8个代码示例，用于说明如何使用os.fork1()。

项目：zippy    作者：securesystemslab    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            _thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = sorted(self.alive.keys())
        self.assertEqual(a, list(range(NUM_THREADS)))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            self.wait_impl(cpid)
            # Tell threads to die
            self.stop = 1
            time.sleep(2*SHORTSLEEP) # Wait for threads to die
项目：oil    作者：oilshell    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = self.alive.keys()
        a.sort()
        self.assertEqual(a, range(NUM_THREADS))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            self.wait_impl(cpid)
            # Tell threads to die
            self.stop = 1
            time.sleep(2*SHORTSLEEP) # Wait for threads to die
项目：python2-tracer    作者：extremecoders-re    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = self.alive.keys()
        a.sort()
        self.assertEqual(a, range(NUM_THREADS))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            self.wait_impl(cpid)
            # Tell threads to die
            self.stop = 1
            time.sleep(2*SHORTSLEEP) # Wait for threads to die
项目：web_ctp    作者：molebot    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            _thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = sorted(self.alive.keys())
        self.assertEqual(a, list(range(NUM_THREADS)))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            try:
                self.wait_impl(cpid)
            finally:
                # Tell threads to die
                self.stop = 1
项目：pefile.pypy    作者：cloudtracer    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = self.alive.keys()
        a.sort()
        self.assertEqual(a, range(NUM_THREADS))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            self.wait_impl(cpid)
            # Tell threads to die
            self.stop = 1
            time.sleep(2*SHORTSLEEP) # Wait for threads to die
项目：ouroboros    作者：pybee    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            _thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = sorted(self.alive.keys())
        self.assertEqual(a, list(range(NUM_THREADS)))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            try:
                self.wait_impl(cpid)
            finally:
                # Tell threads to die
                self.stop = 1
项目：ndk-python    作者：gittor    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = self.alive.keys()
        a.sort()
        self.assertEqual(a, range(NUM_THREADS))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            self.wait_impl(cpid)
            # Tell threads to die
            self.stop = 1
            time.sleep(2*SHORTSLEEP) # Wait for threads to die
项目：kbe_server    作者：xiaohaoppy    | 项目源码 | 文件源码 
def test_wait(self):
        for i in range(NUM_THREADS):
            _thread.start_new(self.f, (i,))

        time.sleep(LONGSLEEP)

        a = sorted(self.alive.keys())
        self.assertEqual(a, list(range(NUM_THREADS)))

        prefork_lives = self.alive.copy()

        if sys.platform in ['unixware7']:
            cpid = os.fork1()
        else:
            cpid = os.fork()

        if cpid == 0:
            # Child
            time.sleep(LONGSLEEP)
            n = 0
            for key in self.alive:
                if self.alive[key] != prefork_lives[key]:
                    n += 1
            os._exit(n)
        else:
            # Parent
            try:
                self.wait_impl(cpid)
            finally:
                # Tell threads to die
                self.stop = 1