Python: Implementing concurrent DNS requests (pipe

Question

I'm trying to write a python script that sends multiple DNS requests, using a different name server for each request.

Implementing a sequential solution is easy using dnspython but too slow for me. Adding concurrency using thread pool is not possible as, in my particular case, all requests use the same source port (REUSE_ADDRESS won't help here either).

For the above reasons I'm thinking about using the following solution (ditching the use of dnspython's resolver module but taking advantage of its message building and parsing modules):

• Allowing up to X requests to be in progress
• Send X requests simultaneously (just sending dns request packets using udp. Probably with added delay between sends to avoid bursts)
• A different thread wait for responses
• When a response arrives match it with the request (by address) and allow a new request to run
• If a response to request does not arrive within TIMEOUT seconds mark it as completed and allow a new request to run

My main issues here are:

• How to implement the task timeout easily
• Is it possible to implement it without the use of thread synchronization (e.g. using event loop?)
• Is there any existing library that can help implementing it (it really feels like I'm trying to reinvent the wheel here, I looked into the asycnio module but couldn't figure out a way to take advantage of it for my problem). Note that I don't want to use existing dns or network libraries as I need the flexibility of changing core functionality (e.g. using raw sockets, changing DNS header fields, etc.).

Answer 1

Did you try aiodns package? https://pypi.python.org/pypi/aiodns/

For timeouts asyncio has standard wait_for coroutine (https://docs.python.org/3/library/asyncio-task.html#asyncio.wait_for).

Answer 2

Using a simple select loop works well here. Here is a code snippet for completion:

def run(self, resolvers_iter):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
    sock.setblocking(False)

    try:
        pending = []

        # For rate limiting
        limit = float(self.timeout)/self.max_pending  # delay between sends
        last_sent = clock() - limit

        # Work as long as there are more resolvers to query
        completed_sending = False
        while not completed_sending or pending:

            # Can I send more requests
            want_to_write = False
            if not completed_sending and len(pending) < self.max_pending:
                want_to_write = True

            # Calculate nearest timeout time to make sure select returns on time
            timeout = None
            if pending:
                timeout = self.timeout - clock() + pending[0][0] + 0.001
                timeout = max(timeout, 0)

            # Rate limit
            time_passed_since_send = clock() - last_sent
            if want_to_write and time_passed_since_send + 0.001 < limit:
                timeout = min(timeout, limit-time_passed_since_send)
                timeout = max(timeout, 0)
                want_to_write = False

            # Poll socket - uses internally the select module
            readable, writable = self._select(readable=True, writable=want_to_write, timeout=timeout)

            # Can read
            if readable:
                # Read as many as possible
                while True:
                    try:
                        # Get response
                        response, from_address = DnsFacilities.read_response(sock)

                        # Check if not duplicate or already timed out
                        sent_time = None
                        for i, (t, ip) in enumerate(pending):
                            if ip == from_address[0]:
                                sent_time = t
                                del pending[i]
                                break

                        if sent_time is not None:
                            self.response_received((response, from_address, clock()-sent_time))

                    except socket.error, e:
                        if e[0] in (socket.errno.EWOULDBLOCK, socket.errno.EAGAIN):
                            break
                        elif e[0] in (socket.errno.WSAECONNRESET, socket.errno.WSAENETRESET):
                            pass
                        else:
                            raise

            # Can write
            if writable:
                try:
                    last_sent = clock()
                    resolver_address = resolvers_iter.next()
                    DnsFacilities.send_query(resolver_address)
                    pending.append((clock(), resolver_address)
                except StopIteration:
                    completed_sending = True

            # Check for timed out tasks
            now = clock()
            while pending and now - pending[0][0] > self.timeout:
                self.response_timeout(pending[0][1])
                del pending[0]

    finally:
        sock.close()