| durin42 |
| hg-reviewers |
We previously only tracked which requests are receiving. A
misbehaving client could accidentally have multiple requests with
the same ID in flight.
We now explicitly track which request IDs are currently active.
We make it illegal to receive a frame associated with a request
ID that has already been dispatched.
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| Path | Packages | |||
|---|---|---|---|---|
| M | mercurial/wireprotoframing.py (32 lines) | |||
| M | tests/test-wireproto-serverreactor.py (54 lines) |
| send those frames. This is useful for half-duplex transports where the | send those frames. This is useful for half-duplex transports where the | ||||
| sender cannot receive until all data has been transmitted. | sender cannot receive until all data has been transmitted. | ||||
| """ | """ | ||||
| self._deferoutput = deferoutput | self._deferoutput = deferoutput | ||||
| self._state = 'idle' | self._state = 'idle' | ||||
| self._bufferedframegens = [] | self._bufferedframegens = [] | ||||
| # request id -> dict of commands that are actively being received. | # request id -> dict of commands that are actively being received. | ||||
| self._receivingcommands = {} | self._receivingcommands = {} | ||||
| # Request IDs that have been received and are actively being processed. | |||||
| # Once all output for a request has been sent, it is removed from this | |||||
| # set. | |||||
| self._activecommands = set() | |||||
| def onframerecv(self, frame): | def onframerecv(self, frame): | ||||
| """Process a frame that has been received off the wire. | """Process a frame that has been received off the wire. | ||||
| Returns a dict with an ``action`` key that details what action, | Returns a dict with an ``action`` key that details what action, | ||||
| if any, the consumer should take next. | if any, the consumer should take next. | ||||
| """ | """ | ||||
| handlers = { | handlers = { | ||||
| 'idle': self._onframeidle, | 'idle': self._onframeidle, | ||||
| 'command-receiving': self._onframecommandreceiving, | 'command-receiving': self._onframecommandreceiving, | ||||
| 'errored': self._onframeerrored, | 'errored': self._onframeerrored, | ||||
| } | } | ||||
| meth = handlers.get(self._state) | meth = handlers.get(self._state) | ||||
| if not meth: | if not meth: | ||||
| raise error.ProgrammingError('unhandled state: %s' % self._state) | raise error.ProgrammingError('unhandled state: %s' % self._state) | ||||
| return meth(frame) | return meth(frame) | ||||
| def onbytesresponseready(self, requestid, data): | def onbytesresponseready(self, requestid, data): | ||||
| """Signal that a bytes response is ready to be sent to the client. | """Signal that a bytes response is ready to be sent to the client. | ||||
| The raw bytes response is passed as an argument. | The raw bytes response is passed as an argument. | ||||
| """ | """ | ||||
| framegen = createbytesresponseframesfrombytes(requestid, data) | def sendframes(): | ||||
| for frame in createbytesresponseframesfrombytes(requestid, data): | |||||
| yield frame | |||||
| self._activecommands.remove(requestid) | |||||
| result = sendframes() | |||||
| if self._deferoutput: | if self._deferoutput: | ||||
| self._bufferedframegens.append(framegen) | self._bufferedframegens.append(result) | ||||
| return 'noop', {} | return 'noop', {} | ||||
| else: | else: | ||||
| return 'sendframes', { | return 'sendframes', { | ||||
| 'framegen': framegen, | 'framegen': result, | ||||
| } | } | ||||
| def oninputeof(self): | def oninputeof(self): | ||||
| """Signals that end of input has been received. | """Signals that end of input has been received. | ||||
| No more frames will be received. All pending activity should be | No more frames will be received. All pending activity should be | ||||
| completed. | completed. | ||||
| """ | """ | ||||
| entry = self._receivingcommands[requestid] | entry = self._receivingcommands[requestid] | ||||
| del self._receivingcommands[requestid] | del self._receivingcommands[requestid] | ||||
| if self._receivingcommands: | if self._receivingcommands: | ||||
| self._state = 'command-receiving' | self._state = 'command-receiving' | ||||
| else: | else: | ||||
| self._state = 'idle' | self._state = 'idle' | ||||
| assert requestid not in self._activecommands | |||||
| self._activecommands.add(requestid) | |||||
| return 'runcommand', { | return 'runcommand', { | ||||
| 'requestid': requestid, | 'requestid': requestid, | ||||
| 'command': entry['command'], | 'command': entry['command'], | ||||
| 'args': entry['args'], | 'args': entry['args'], | ||||
| 'data': entry['data'].getvalue() if entry['data'] else None, | 'data': entry['data'].getvalue() if entry['data'] else None, | ||||
| } | } | ||||
| def _makewantframeresult(self): | def _makewantframeresult(self): | ||||
| return self._makeerrorresult( | return self._makeerrorresult( | ||||
| _('expected command frame; got %d') % frame.typeid) | _('expected command frame; got %d') % frame.typeid) | ||||
| if frame.requestid in self._receivingcommands: | if frame.requestid in self._receivingcommands: | ||||
| self._state = 'errored' | self._state = 'errored' | ||||
| return self._makeerrorresult( | return self._makeerrorresult( | ||||
| _('request with ID %d already received') % frame.requestid) | _('request with ID %d already received') % frame.requestid) | ||||
| if frame.requestid in self._activecommands: | |||||
| self._state = 'errored' | |||||
| return self._makeerrorresult(( | |||||
| _('request with ID %d is already active') % frame.requestid)) | |||||
| expectingargs = bool(frame.flags & FLAG_COMMAND_NAME_HAVE_ARGS) | expectingargs = bool(frame.flags & FLAG_COMMAND_NAME_HAVE_ARGS) | ||||
| expectingdata = bool(frame.flags & FLAG_COMMAND_NAME_HAVE_DATA) | expectingdata = bool(frame.flags & FLAG_COMMAND_NAME_HAVE_DATA) | ||||
| self._receivingcommands[frame.requestid] = { | self._receivingcommands[frame.requestid] = { | ||||
| 'command': frame.payload, | 'command': frame.payload, | ||||
| 'args': {}, | 'args': {}, | ||||
| 'data': None, | 'data': None, | ||||
| 'expectingargs': expectingargs, | 'expectingargs': expectingargs, | ||||
| 'command frame')) | 'command frame')) | ||||
| def _onframecommandreceiving(self, frame): | def _onframecommandreceiving(self, frame): | ||||
| # It could be a new command request. Process it as such. | # It could be a new command request. Process it as such. | ||||
| if frame.typeid == FRAME_TYPE_COMMAND_NAME: | if frame.typeid == FRAME_TYPE_COMMAND_NAME: | ||||
| return self._onframeidle(frame) | return self._onframeidle(frame) | ||||
| # All other frames should be related to a command that is currently | # All other frames should be related to a command that is currently | ||||
| # receiving. | # receiving but is not active. | ||||
| if frame.requestid in self._activecommands: | |||||
| self._state = 'errored' | |||||
| return self._makeerrorresult( | |||||
| _('received frame for request that is still active: %d') % | |||||
| frame.requestid) | |||||
| if frame.requestid not in self._receivingcommands: | if frame.requestid not in self._receivingcommands: | ||||
| self._state = 'errored' | self._state = 'errored' | ||||
| return self._makeerrorresult( | return self._makeerrorresult( | ||||
| _('received frame for request that is not receiving: %d') % | _('received frame for request that is not receiving: %d') % | ||||
| frame.requestid) | frame.requestid) | ||||
| entry = self._receivingcommands[frame.requestid] | entry = self._receivingcommands[frame.requestid] | ||||
| 'message': b'command data frame without flags', | 'message': b'command data frame without flags', | ||||
| }) | }) | ||||
| def testframefornonreceivingrequest(self): | def testframefornonreceivingrequest(self): | ||||
| """Receiving a frame for a command that is not receiving is illegal.""" | """Receiving a frame for a command that is not receiving is illegal.""" | ||||
| results = list(sendframes(makereactor(), [ | results = list(sendframes(makereactor(), [ | ||||
| ffs(b'1 command-name eos command1'), | ffs(b'1 command-name eos command1'), | ||||
| ffs(b'3 command-name have-data command3'), | ffs(b'3 command-name have-data command3'), | ||||
| ffs(b'1 command-argument eoa ignored'), | ffs(b'5 command-argument eoa ignored'), | ||||
| ])) | ])) | ||||
| self.assertaction(results[2], 'error') | self.assertaction(results[2], 'error') | ||||
| self.assertEqual(results[2][1], { | self.assertEqual(results[2][1], { | ||||
| 'message': b'received frame for request that is not receiving: 1', | 'message': b'received frame for request that is not receiving: 5', | ||||
| }) | }) | ||||
| def testsimpleresponse(self): | def testsimpleresponse(self): | ||||
| """Bytes response to command sends result frames.""" | """Bytes response to command sends result frames.""" | ||||
| reactor = makereactor() | reactor = makereactor() | ||||
| list(sendcommandframes(reactor, 1, b'mycommand', {})) | list(sendcommandframes(reactor, 1, b'mycommand', {})) | ||||
| result = reactor.onbytesresponseready(1, b'response') | result = reactor.onbytesresponseready(1, b'response') | ||||
| result = reactor.oninputeof() | result = reactor.oninputeof() | ||||
| self.assertaction(result, 'sendframes') | self.assertaction(result, 'sendframes') | ||||
| self.assertframesequal(result[1]['framegen'], [ | self.assertframesequal(result[1]['framegen'], [ | ||||
| b'3 bytes-response eos response3', | b'3 bytes-response eos response3', | ||||
| b'1 bytes-response eos response1', | b'1 bytes-response eos response1', | ||||
| b'5 bytes-response eos response5', | b'5 bytes-response eos response5', | ||||
| ]) | ]) | ||||
| def testduplicaterequestonactivecommand(self): | |||||
| """Receiving a request ID that matches a request that isn't finished.""" | |||||
| reactor = makereactor() | |||||
| list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| results = list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| self.assertaction(results[0], 'error') | |||||
| self.assertEqual(results[0][1], { | |||||
| 'message': b'request with ID 1 is already active', | |||||
| }) | |||||
| def testduplicaterequestonactivecommandnosend(self): | |||||
| """Same as above but we've registered a response but haven't sent it.""" | |||||
| reactor = makereactor() | |||||
| list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| reactor.onbytesresponseready(1, b'response') | |||||
| # We've registered the response but haven't sent it. From the | |||||
| # perspective of the reactor, the command is still active. | |||||
| results = list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| self.assertaction(results[0], 'error') | |||||
| self.assertEqual(results[0][1], { | |||||
| 'message': b'request with ID 1 is already active', | |||||
| }) | |||||
| def testduplicaterequestargumentframe(self): | |||||
| """Variant on above except we sent an argument frame instead of name.""" | |||||
| reactor = makereactor() | |||||
| list(sendcommandframes(reactor, 1, b'command', {})) | |||||
| results = list(sendframes(reactor, [ | |||||
| ffs(b'3 command-name have-args command'), | |||||
| ffs(b'1 command-argument 0 ignored'), | |||||
| ])) | |||||
| self.assertaction(results[0], 'wantframe') | |||||
| self.assertaction(results[1], 'error') | |||||
| self.assertEqual(results[1][1], { | |||||
| 'message': 'received frame for request that is still active: 1', | |||||
| }) | |||||
| def testduplicaterequestaftersend(self): | |||||
| """We can use a duplicate request ID after we've sent the response.""" | |||||
| reactor = makereactor() | |||||
| list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| res = reactor.onbytesresponseready(1, b'response') | |||||
| list(res[1]['framegen']) | |||||
| results = list(sendcommandframes(reactor, 1, b'command1', {})) | |||||
| self.assertaction(results[0], 'runcommand') | |||||
| if __name__ == '__main__': | if __name__ == '__main__': | ||||
| import silenttestrunner | import silenttestrunner | ||||
| silenttestrunner.main(__name__) | silenttestrunner.main(__name__) | ||||