[2934] use blocking I/O for sending XFR responses.

src/bin/xfrout/tests/xfrout_test.py.in

@@ -18,6 +18,8 @@
 
 import unittest
 import os
+import socket
+import fcntl
 from isc.testutils.tsigctx_mock import MockTSIGContext
 from isc.testutils.ccsession_mock import MockModuleCCSession
 from isc.cc.session import *
@@ -190,6 +192,24 @@ class Dbserver:
     def decrease_transfers_counter(self):
         self.transfer_counter -= 1
 
+class TestUtility(unittest.TestCase):
+    """Test some utility functions."""
+
+    def test_make_blockign(self):
+        def is_blocking(fd):
+            return (fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_NONBLOCK) == 0
+
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM,
+                           socket.IPPROTO_TCP) as sock:
+            # By default socket is made blocking
+            self.assertTrue(is_blocking(sock.fileno()))
+            # make_blocking(False) makes it non blocking
+            xfrout.make_blocking(sock.fileno(), False)
+            self.assertFalse(is_blocking(sock.fileno()))
+            # make_blocking(True) makes it blocking again
+            xfrout.make_blocking(sock.fileno(), True)
+            self.assertTrue(is_blocking(sock.fileno()))
+
 class TestXfroutSessionBase(unittest.TestCase):
     '''Base class for tests related to xfrout sessions
 
@@ -269,6 +289,12 @@ class TestXfroutSessionBase(unittest.TestCase):
             self.xfrsess._request_typestr = 'IXFR'
 
     def setUp(self):
+        # xfrout.make_blocking won't work with faked socket, and we'd like to
+        # examine how it's called, so we replace it without our mock.
+        self.__orig_make_blocking = xfrout.make_blocking
+        self._make_blocking_history = []
+        xfrout.make_blocking = lambda x, y: self.__make_blocking(x, y)
+
         self.sock = MySocket(socket.AF_INET,socket.SOCK_STREAM)
         self.xfrsess = MyXfroutSession(self.sock, None, Dbserver(),
                                        TSIGKeyRing(),
@@ -285,7 +311,11 @@ class TestXfroutSessionBase(unittest.TestCase):
         # original is used elsewhere.
         self.orig_get_rrset_len = xfrout.get_rrset_len
 
+    def __make_blocking(self, fd, on):
+        self._make_blocking_history.append((fd, on))
+
     def tearDown(self):
+        xfrout.make_blocking = self.__orig_make_blocking
         xfrout.get_rrset_len = self.orig_get_rrset_len
         # transfer_counter must be always be reset no matter happens within
         # the XfroutSession object.  We check the condition here.
@@ -306,7 +336,12 @@ class TestXfroutSession(TestXfroutSessionBase):
     def test_quota_ok(self):
         '''The default case in terms of the xfrout quota.
 
+        It also confirms that the socket is made blocking.
+
         '''
+        # make_blocking shouldn't be called yet.
+        self.assertEqual([], self._make_blocking_history)
+
         # set up a bogus request, which should result in FORMERR. (it only
         # has to be something that is different from the previous case)
         self.xfrsess._request_data = \
@@ -316,6 +351,26 @@ class TestXfroutSession(TestXfroutSessionBase):
         XfroutSession._handle(self.xfrsess)
         self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.FORMERR)
 
+        # make_blocking should have been called in _handle().  Note that
+        # in the test fixture we handle fileno as a faked socket object, not
+        # as integer.
+        self.assertEqual([(self.sock, True)], self._make_blocking_history)
+
+    def test_make_blocking_fail(self):
+        """Check what if make_blocking() raises an exception."""
+
+        # make_blocking() can fail due to OSError.  It shouldn't cause
+        # disruption, and xfrout_start shouldn't be called.
+
+        def raise_exception():
+            raise OSError('faked error')
+        xfrout.make_blocking = lambda x, y: raise_exception()
+        self.start_arg = []
+        self.xfrsess.dns_xfrout_start = \
+            lambda s, x, y: self.start_arg.append((x, y))
+        XfroutSession._handle(self.xfrsess)
+        self.assertEqual([], self.start_arg)
+
     def test_exception_from_session(self):
         '''Test the case where the main processing raises an exception.
 

src/bin/xfrout/xfrout.py.in

@@ -30,6 +30,7 @@ from isc.cc import SessionError, SessionTimeout
 from isc.statistics import Counters
 from isc.notify import notify_out
 import isc.util.process
+import fcntl
 import socket
 import select
 import errno
@@ -152,6 +153,24 @@ def get_soa_serial(soa_rdata):
     '''
     return Serial(int(soa_rdata.to_text().split()[2]))
 
+def make_blocking(filenum, on):
+    """A helper function to change blocking mode of the given file.
+
+    It sets the mode of blocking I/O for the file associated with filenum,
+    according to parameter 'on': if it's True the file will be made blocking;
+    otherwise it will be made non-blocking.
+
+    filenum(int): file number (descriptor) of the file to update.
+    on(bool): whether enable (True) or disable (False) blocking I/O.
+
+    """
+    flags = fcntl.fcntl(filenum, fcntl.F_GETFL)
+    if on:                      # make it blocking
+        flags &= ~os.O_NONBLOCK
+    else:                       # make it non blocking
+        flags |= os.O_NONBLOCK
+    fcntl.fcntl(filenum, fcntl.F_SETFL, flags)
+
 class XfroutSession():
     def __init__(self, sock_fd, request_data, server, tsig_key_ring, remote,
                  default_acl, zone_config, client_class=DataSourceClient):
@@ -190,6 +209,10 @@ class XfroutSession():
         quota_ok = self._server.increase_transfers_counter()
         ex = None
         try:
+            # Before start, make sure the socket uses blocking I/O because
+            # responses will be sent in the blocking mode; otherwise it could
+            # result in EWOULDBLOCK and disrupt the session.
+            make_blocking(self._sock_fd, True)
             self.dns_xfrout_start(self._sock_fd, self._request_data, quota_ok)
         except Exception as e:
             # To avoid resource leak we need catch all possible exceptions