Change in vdsm[master]: json-rpc: Protocol detection

nsoffer at redhat.com nsoffer at redhat.com
Thu May 22 10:43:43 UTC 2014


Nir Soffer has posted comments on this change.

Change subject: json-rpc: Protocol detection
......................................................................


Patch Set 17:

(15 comments)

This is very nice code now, and a joy to review!

http://gerrit.ovirt.org/#/c/26300/17/vdsm/protocolDetector.py
File vdsm/protocolDetector.py:

Line 51: 
Line 52:         self._jsonBinding = None
Line 53:         self._xmlBinding = None
Line 54: 
Line 55:     def _set_non_blocking_pipes(self, fd):
It works, but this is not related to pipes - _set_non_blokcing is more correct name.
Line 56:         flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
Line 57:         flags = flags | os.O_NONBLOCK
Line 58:         fcntl.fcntl(fd, fcntl.F_SETFL, flags)
Line 59: 


Line 94:     def init_handlers(self):
Line 95:         self.handlers = [_StompDetector(self._jsonBinding),
Line 96:                          _XmlDetector(self._xmlBinding)]
Line 97: 
Line 98:     def stop(self):
The best way to stop when stop is called from another thread, is to set a flag, and wakeup the server.

    def stop(self):
        self._isRunning = False
        self.wakeup()

and the cleanup code can go into serve_forever, or called from there:

    def serve_forever(self):
        try:
            while self.isRunning:
            process events...
        finally:        
            cleanup...
Line 99:         stopHandlers = self.handlers
Line 100:         self.handlers.append(self._jsonBinding)
Line 101:         for handler in stopHandlers:
Line 102:             if handler:


Line 95:         self.handlers = [_StompDetector(self._jsonBinding),
Line 96:                          _XmlDetector(self._xmlBinding)]
Line 97: 
Line 98:     def stop(self):
Line 99:         stopHandlers = self.handlers
What are you trying to do here?

If you like to create a list of handlers for stopping, you need to:

    stop = self.handlers[:] # Creates a copy of the original list
    stop.append(self._jsonBinding)
    for handler in stop:
        handler.stop()

But this is simpler and more clear:

    self._jsonBinding.stop()
    self._xmlBinding.stop()
   
    for handler in self.handlers:
        handler.stop()
Line 100:         self.handlers.append(self._jsonBinding)
Line 101:         for handler in stopHandlers:
Line 102:             if handler:
Line 103:                 handler.stop()


Line 98:     def stop(self):
Line 99:         stopHandlers = self.handlers
Line 100:         self.handlers.append(self._jsonBinding)
Line 101:         for handler in stopHandlers:
Line 102:             if handler:
You can get None?! Please fix the code responsible for this.
Line 103:                 handler.stop()
Line 104: 
Line 105:         self._isRunning = False
Line 106:         self.poller.unregister(self.socket)


Line 102:             if handler:
Line 103:                 handler.stop()
Line 104: 
Line 105:         self._isRunning = False
Line 106:         self.poller.unregister(self.socket)
Is this thread safe? probably not.
Line 107:         self.socket.close()
Line 108:         self.wakeup()
Line 109: 
Line 110:     def wakeup(self):


Line 107:         self.socket.close()
Line 108:         self.wakeup()
Line 109: 
Line 110:     def wakeup(self):
Line 111:         os.write(self._write_fd, '1')
You must check and ignore EAGAIN and  probably EINPROGRESS here. If we cannot write to the pipe (unlikely), it is full and the event loop is about to wake up anyway.
Line 112: 
Line 113:     def _cleanup_wakeup_pipe(self):
Line 114:         os.read(self._read_fd, 128)
Line 115: 


Line 110:     def wakeup(self):
Line 111:         os.write(self._write_fd, '1')
Line 112: 
Line 113:     def _cleanup_wakeup_pipe(self):
Line 114:         os.read(self._read_fd, 128)
You must check and ignore EAGAIN and  probably EINPROGRESS here. If you cannot read, there is nothing in the pipe (unlikely), and there is nothing to clean.
Line 115: 
Line 116:     def _accept_conn(self):
Line 117:         client_socket, _ = self.socket.accept()
Line 118:         client_socket.settimeout(self._timeout)


Line 114:         os.read(self._read_fd, 128)
Line 115: 
Line 116:     def _accept_conn(self):
Line 117:         client_socket, _ = self.socket.accept()
Line 118:         client_socket.settimeout(self._timeout)
You cannot use timeout and non-blocking mode in the same time. Remove this line.
Line 119:         client_socket.setblocking(0)
Line 120: 
Line 121:         self.fd_to_socket[client_socket.fileno()] = client_socket
Line 122:         self.poller.register(client_socket, self.READ_ONLY)


Line 117:         client_socket, _ = self.socket.accept()
Line 118:         client_socket.settimeout(self._timeout)
Line 119:         client_socket.setblocking(0)
Line 120: 
Line 121:         self.fd_to_socket[client_socket.fileno()] = client_socket
fd_to_socket is correct, but this is not the main theme here. This map is not keeping random sockets, but keeping pending connections, waiting for protocol detection.

So it would be nice if you pick a name that describe this, like self.pending_connections.
Line 122:         self.poller.register(client_socket, self.READ_ONLY)
Line 123: 
Line 124:     def _conn_handle(self, fd):
Line 125:         try:


Line 120: 
Line 121:         self.fd_to_socket[client_socket.fileno()] = client_socket
Line 122:         self.poller.register(client_socket, self.READ_ONLY)
Line 123: 
Line 124:     def _conn_handle(self, fd):
Lets be consistent with other method names (cleanup_x, acceps_y) and call it handle_read. If you like to be more specific, handle_connection_read.
Line 125:         try:
Line 126:             client_socket = self.fd_to_socket[fd]
Line 127:             data = client_socket.recv(4096, socket.MSG_PEEK)
Line 128:         except socket.error as e:


Line 123: 
Line 124:     def _conn_handle(self, fd):
Line 125:         try:
Line 126:             client_socket = self.fd_to_socket[fd]
Line 127:             data = client_socket.recv(4096, socket.MSG_PEEK)
You should ignore here EAGAIN and probbaly EINPROGRESS. There is not point to log an error if we cannot read from a socket when poll wakes up. This is unlikely but possible.
Line 128:         except socket.error as e:
Line 129:             self.log.error(e)
Line 130:             data = None
Line 131: 


Line 127:             data = client_socket.recv(4096, socket.MSG_PEEK)
Line 128:         except socket.error as e:
Line 129:             self.log.error(e)
Line 130:             data = None
Line 131: 
There is no point to continue if the recv failed. And of course that you don't want to fail this connection and close it in this case.
Line 132:         socket_address = client_socket.getsockname()
Line 133: 
Line 134:         handler = self.detect_protocol(data)
Line 135:         if handler:


Line 128:         except socket.error as e:
Line 129:             self.log.error(e)
Line 130:             data = None
Line 131: 
Line 132:         socket_address = client_socket.getsockname()
You need this only if you detected the protocol - why add a variable in this scope that nobody needes?
Line 133: 
Line 134:         handler = self.detect_protocol(data)
Line 135:         if handler:
Line 136:             self.poller.unregister(client_socket)


Line 133: 
Line 134:         handler = self.detect_protocol(data)
Line 135:         if handler:
Line 136:             self.poller.unregister(client_socket)
Line 137:             del self.fd_to_socket[fd]
Get the socket_address here, and return the socket to blocking mode, since you don't know (you know but the detector should not) if the handler expect the socket in non-blocking mode.
Line 138:             handler.handleSocket(client_socket,
Line 139:                                  socket_address)
Line 140:         else:
Line 141:             self.log.warn("Unrecognized protocol from %s",


Line 139:                                  socket_address)
Line 140:         else:
Line 141:             self.log.warn("Unrecognized protocol from %s",
Line 142:                           socket_address)
Line 143:             client_socket.close()
You should try multiple times to get the protocol, probably with a timeout. I thin that this can work:

adding a pending connection:

    self.pending_connections[fd] = (time.time(), connection)

checking connections on events

    accepted, connection = self.pending_connections[fd]
    ....
    if time.time() - accepted > 30.0:
        Kill it

And to make sure you check connections from time to time, change the event loop to wake up every 30 seconds, and kill pending connections.

    while running:
        wait for events up to 30 seconds:
        if timed out:
            cleanup pending connections
Line 144: 
Line 145:     def _create_socket(self, host, port):
Line 146:         addr = socket.getaddrinfo(host, port, socket.AF_INET,
Line 147:                                   socket.SOCK_STREAM)


-- 
To view, visit http://gerrit.ovirt.org/26300
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: comment
Gerrit-Change-Id: Id739a40e2b37dcc175137ec91cd5ec166ad24a75
Gerrit-PatchSet: 17
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski at gmail.com>
Gerrit-Reviewer: Barak Azulay <bazulay at redhat.com>
Gerrit-Reviewer: Nir Soffer <nsoffer at redhat.com>
Gerrit-Reviewer: Piotr Kliczewski <piotr.kliczewski at gmail.com>
Gerrit-Reviewer: Saggi Mizrahi <smizrahi at redhat.com>
Gerrit-Reviewer: Yaniv Bronhaim <ybronhei at redhat.com>
Gerrit-Reviewer: Yeela Kaplan <ykaplan at redhat.com>
Gerrit-Reviewer: automation at ovirt.org
Gerrit-Reviewer: oVirt Jenkins CI Server
Gerrit-HasComments: Yes


More information about the vdsm-patches mailing list