Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _updateRegistration(self, fd):
try:
fileno = fd.fileno()
self._poller.unregister(fileno)
except (SocketError, IOError), e:
if e[0] == EBADF:
keys = [k for k, v in self._map.items() if v == fd]
for key in keys:
del self._map[key]
mask = 0
if fd in self._read:
mask = mask | EPOLLIN
if fd in self._write:
mask = mask | EPOLLOUT
if mask:
self._poller.register(fd, mask)
self._map[fileno] = fd
else:
super(EPoll, self).discard(fd)