standalone test case demonstrating that EPOLLHUP triggers the issue

test_read_pty.py (text/x-python), 1.23 KB, created by Zac Medico on 2014-12-26 07:34:02 UTC

#!/usr/bin/env python

import os
import pty
import select

BUFSIZE = 4096

def test_read(open_pipe, data):

	consumer_fd, producer_fd = open_pipe()

	pid = os.fork()
	if pid == 0:
		os.close(consumer_fd)
		while data:
			data = data[os.write(producer_fd, data):]
		os._exit(os.EX_OK)

	os.close(producer_fd)
	epoll_obj = select.epoll()

	try:
		epoll_obj.register(consumer_fd,
			select.EPOLLIN | select.EPOLLHUP | select.EPOLLERR)

		received = []
		eof = None

		while eof is None:
			for fd, event in epoll_obj.poll(-1):

				if event & select.EPOLLIN:
					try:
						received.append(os.read(fd, BUFSIZE))
					except OSError as e:
						eof = e
						break
					else:
						if not received[-1]:
							eof = "os.read returned empty string"
							break

				if event & select.EPOLLHUP:
					eof = "EPOLLHUP"
					break

				if event & select.EPOLLERR:
					eof = "EPOLLERR"
					break

		received = b"".join(received)

		os.waitpid(pid, 0)

		if len(data) != len(received):
			raise AssertionError("%s != %s eof: %s" %
				(len(data), len(received), eof))
	finally:
		epoll_obj.close()
		os.close(consumer_fd)

def main():
	for x in (1, 2, 5, 6, 7, 8, 2**5, 2**10, 2**12, 2**13, 2**14):
		test_read(pty.openpty, x * b"a")

if __name__ == '__main__':
	main()