@@ -0,0 +1,2578 @@
+# -*- coding: utf8 -*-%0A# Copyright (C) PyZMQ Developers%0A# Distributed under the terms of the Modified BSD License.%0A%0Aimport signal%0Aimport time%0Afrom threading import Thread%0A%0Aimport zmq%0Afrom zmq.tests import (%0A    BaseZMQTestCase, SkipTest, skip_pypy, skip_if%0A)%0Afrom zmq.utils.strtypes import b%0A%0A%0A# Partially based on EINTRBaseTest from CPython 3.5 eintr_tester%0A%0Aclass TestEINTRSysCall(BaseZMQTestCase):%0A    %22%22%22 Base class for EINTR tests. %22%22%22%0A%0A    # delay for initial signal delivery%0A    signal_delay = 0.1%0A    # timeout for tests. Must be %3E signal_delay%0A    timeout = .25%0A    timeout_ms = int(timeout * 1e3)%0A%0A    @skip_if(not hasattr(signal, 'setitimer'), 'EINTR tests require setitimer')%0A    @skip_pypy%0A    def alarm(self, t=None):%0A        %22%22%22start a timer to fire only once%0A        %0A        like signal.alarm, but with better resolution than integer seconds.%0A        %22%22%22%0A        if t is None:%0A            t = self.signal_delay%0A        self.timer_fired = False%0A        self.orig_handler = signal.signal(signal.SIGALRM, self.stop_timer)%0A        # signal_period ignored, since only one timer event is allowed to fire%0A        signal.setitimer(signal.ITIMER_REAL, t, 1000)%0A    %0A    def stop_timer(self, *args):%0A        self.timer_fired = True%0A        signal.setitimer(signal.ITIMER_REAL, 0, 0)%0A        signal.signal(signal.SIGALRM, self.orig_handler)%0A    %0A    @skip_if(not hasattr(zmq, 'RCVTIMEO'))%0A    def test_retry_recv(self):%0A        pull = self.socket(zmq.PULL)%0A        pull.rcvtimeo = self.timeout_ms%0A        self.alarm()%0A        with self.assertRaises(zmq.Again):%0A            pull.recv()%0A        assert self.timer_fired%0A%0A    @skip_if(not hasattr(zmq, 'SNDTIMEO'))%0A    def test_retry_send(self):%0A        push = self.socket(zmq.PUSH)%0A        push.sndtimeo = self.timeout_ms%0A        self.alarm()%0A        with self.assertRaises(zmq.Again):%0A            push.send(b('buf'))%0A        assert self.timer_fired%0A    %0A    def test_retry_poll(self):%0A        x, y = self.create_bound_pair()%0A        poller = zmq.Poller()%0A        poller.register(x, zmq.POLLIN)%0A        self.alarm()%0A        def send():%0A            time.sleep(2 * self.signal_delay)%0A            y.send(b('ping'))%0A        t = Thread(target=send)%0A        t.start()%0A        evts = dict(poller.poll(2 * self.timeout_ms))%0A        t.join()%0A        assert x in evts%0A        assert self.timer_fired%0A        x.recv()%0A    %0A    def test_retry_getsockopt(self):%0A        raise SkipTest(%22TODO: find a way to interrupt getsockopt%22)%0A    %0A    def test_retry_setsockopt(self):%0A        raise SkipTest(%22TODO: find a way to interrupt setsockopt%22)%0A

test retried system calls