def test_above_fd_setsize(self):
# A scalable implementation should have no problem with more than
# FD_SETSIZE file descriptors. Since we don't know the value, we just
# try to set the soft RLIMIT_NOFILE to the hard RLIMIT_NOFILE ceiling.
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
if hard == resource.RLIM_INFINITY:
self.skipTest("RLIMIT_NOFILE is infinite")
try: # If we're on a *BSD system, the limit tag is different.
_, bsd_hard = resource.getrlimit(resource.RLIMIT_OFILE)
if bsd_hard == resource.RLIM_INFINITY:
self.skipTest("RLIMIT_OFILE is infinite")
if bsd_hard < hard:
hard = bsd_hard
# NOTE: AttributeError resource.RLIMIT_OFILE is not defined on Mac OS.
except (OSError, resource.error, AttributeError):
pass
try:
resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard))
self.addCleanup(resource.setrlimit, resource.RLIMIT_NOFILE,
(soft, hard))
limit_nofile = min(hard, 2 ** 16)
except (OSError, ValueError):
limit_nofile = soft
# Guard against already allocated FDs
limit_nofile -= 256
limit_nofile = max(0, limit_nofile)
s = self.make_selector()
for i in range(limit_nofile // 2):
rd, wr = self.make_socketpair()
s.register(rd, selectors2.EVENT_READ)
s.register(wr, selectors2.EVENT_WRITE)
self.assertEqual(limit_nofile // 2, len(s.select()))
评论列表
文章目录