From af37b809aa931856764e766ebe3e9baaaaaa0897 Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Tue, 29 Nov 2016 12:28:51 -0800 Subject: [PATCH 1/8] experimental --- test/http2_test/http2_base_server.py | 157 +++++++ test/http2_test/http2_test_server.py | 164 +++++++ test/http2_test/messages_pb2.py | 661 +++++++++++++++++++++++++++ 3 files changed, 982 insertions(+) create mode 100644 test/http2_test/http2_base_server.py create mode 100644 test/http2_test/http2_test_server.py create mode 100644 test/http2_test/messages_pb2.py diff --git a/test/http2_test/http2_base_server.py b/test/http2_test/http2_base_server.py new file mode 100644 index 00000000000..07bd37cae9a --- /dev/null +++ b/test/http2_test/http2_base_server.py @@ -0,0 +1,157 @@ +import struct +import messages_pb2 +import functools +import argparse +import logging +import time + +from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.internet.protocol import Protocol, Factory +from twisted.internet import endpoints, reactor, error, defer +from h2.connection import H2Connection +from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged, PingAcknowledged +from threading import Lock + +READ_CHUNK_SIZE = 16384 +GRPC_HEADER_SIZE = 5 + +class H2ProtocolBaseServer(Protocol): + def __init__(self): + self._conn = H2Connection(client_side=False) + self._recv_buffer = '' + self._handlers = {} + self._handlers['DataReceived'] = self.on_data_received_default + self._handlers['WindowUpdated'] = self.on_window_update_default + self._handlers['RequestReceived'] = self.on_request_received_default + self._handlers['SendDone'] = self.on_send_done_default + self._handlers['ConnectionLost'] = self.on_connection_lost + self._handlers['PingAcknowledged'] = self.on_ping_acknowledged_default + self._stream_status = {} + self._outstanding_pings = 0 + + def set_handlers(self, handlers): + self._handlers = handlers + + def connectionMade(self): + logging.info('Connection Made') + self._conn.initiate_connection() + self.transport.setTcpNoDelay(True) + self.transport.write(self._conn.data_to_send()) + + def connectionLost(self, reason): + self._handlers['ConnectionLost'](reason) + + def on_connection_lost(self, reason): + logging.info('Disconnected %s'%reason) + reactor.callFromThread(reactor.stop) + + def dataReceived(self, data): + events = self._conn.receive_data(data) + if self._conn.data_to_send: + self.transport.write(self._conn.data_to_send()) + for event in events: + if isinstance(event, RequestReceived) and self._handlers.has_key('RequestReceived'): + logging.info('RequestReceived Event') + self._handlers['RequestReceived'](event) + elif isinstance(event, DataReceived) and self._handlers.has_key('DataReceived'): + logging.info('DataReceived Event') + self._handlers['DataReceived'](event) + elif isinstance(event, WindowUpdated) and self._handlers.has_key('WindowUpdated'): + logging.info('WindowUpdated Event') + self._handlers['WindowUpdated'](event) + elif isinstance(event, PingAcknowledged) and self._handlers.has_key('PingAcknowledged'): + logging.info('PingAcknowledged Event') + self._handlers['PingAcknowledged'](event) + self.transport.write(self._conn.data_to_send()) + + def on_ping_acknowledged_default(self, event): + self._outstanding_pings -= 1 + + def on_data_received_default(self, event): + self._conn.acknowledge_received_data(len(event.data), event.stream_id) + self._recv_buffer += event.data + + def on_request_received_default(self, event): + self._recv_buffer = '' + self._stream_id = event.stream_id + self._stream_status[event.stream_id] = True + self._conn.send_headers( + stream_id=event.stream_id, + headers=[ + (':status', '200'), + ('content-type', 'application/grpc'), + ('grpc-encoding', 'identity'), + ('grpc-accept-encoding', 'identity,deflate,gzip'), + ], + ) + self.transport.write(self._conn.data_to_send()) + + def on_window_update_default(self, event): + pass + + def send_reset_stream(self): + self._conn.reset_stream(self._stream_id) + self.transport.write(self._conn.data_to_send()) + + def setup_send(self, data_to_send): + self._send_remaining = len(data_to_send) + self._send_offset = 0 + self._data_to_send = data_to_send + self.default_send() + + def default_send(self): + while self._send_remaining > 0: + lfcw = self._conn.local_flow_control_window(self._stream_id) + if lfcw == 0: + break + chunk_size = min(lfcw, READ_CHUNK_SIZE) + bytes_to_send = min(chunk_size, self._send_remaining) + logging.info('flow_control_window = %d. sending [%d:%d] stream_id %d'% + (lfcw, self._send_offset, self._send_offset + bytes_to_send, + self._stream_id)) + data = self._data_to_send[self._send_offset : self._send_offset + bytes_to_send] + self._conn.send_data(self._stream_id, data, False) + self._send_remaining -= bytes_to_send + self._send_offset += bytes_to_send + if self._send_remaining == 0: + self._handlers['SendDone']() + + def default_ping(self): + self._outstanding_pings += 1 + self._conn.ping(b'\x00'*8) + self.transport.write(self._conn.data_to_send()) + + def on_send_done_default(self): + if self._stream_status[self._stream_id]: + self._stream_status[self._stream_id] = False + self.default_send_trailer() + + def default_send_trailer(self): + logging.info('Sending trailer for stream id %d'%self._stream_id) + self._conn.send_headers(self._stream_id, + headers=[ ('grpc-status', '0') ], + end_stream=True + ) + self.transport.write(self._conn.data_to_send()) + + @staticmethod + def default_response_data(response_size): + sresp = messages_pb2.SimpleResponse() + sresp.payload.body = b'\x00'*response_size + serialized_resp_proto = sresp.SerializeToString() + response_data = b'\x00' + struct.pack('i', len(serialized_resp_proto))[::-1] + serialized_resp_proto + return response_data + + @staticmethod + def parse_received_data(recv_buffer): + """ returns a grpc framed string of bytes containing response proto of the size + asked in request """ + grpc_msg_size = struct.unpack('i',recv_buffer[1:5][::-1])[0] + if len(recv_buffer) != GRPC_HEADER_SIZE + grpc_msg_size: + logging.error('not enough data to decode req proto. size = %d, needed %s'%(len(recv_buffer), 5+grpc_msg_size)) + return None + req_proto_str = recv_buffer[5:5+grpc_msg_size] + sr = messages_pb2.SimpleRequest() + sr.ParseFromString(req_proto_str) + logging.info('Parsed request: response_size=%s'%sr.response_size) + return sr diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py new file mode 100644 index 00000000000..be5f1593ebc --- /dev/null +++ b/test/http2_test/http2_test_server.py @@ -0,0 +1,164 @@ +""" + HTTP2 Test Server. Highly experimental work in progress. +""" +import struct +import messages_pb2 +import argparse +import logging +import time + +from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.internet.protocol import Protocol, Factory +from twisted.internet import endpoints, reactor, error, defer +from h2.connection import H2Connection +from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged +from threading import Lock +import http2_base_server + +READ_CHUNK_SIZE = 16384 +GRPC_HEADER_SIZE = 5 + +class TestcaseRstStreamAfterHeader(object): + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + + def get_base_server(self): + return self._base_server + + def on_request_received(self, event): + # send initial headers + self._base_server.on_request_received_default(event) + # send reset stream + self._base_server.send_reset_stream() + +class TestcaseRstStreamAfterData(object): + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['DataReceived'] = self.on_data_received + + def get_base_server(self): + return self._base_server + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + assert(sr is not None) + assert(sr.response_size <= 2048) # so it can fit into one flow control window + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + self._base_server.setup_send(response_data) + # send reset stream + self._base_server.send_reset_stream() + +class TestcaseGoaway(object): + """ + Process incoming request normally. After sending trailer response, + send GOAWAY with stream id = 1. + assert that the next request is made on a different connection. + """ + def __init__(self, iteration): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['WindowUpdated'] = self.on_window_update_default + self._base_server._handlers['SendDone'] = self.on_send_done + self._base_server._handlers['ConnectionLost'] = self.on_connection_lost + self._ready_to_send = False + self._iteration = iteration + + def get_base_server(self): + return self._base_server + + def on_connection_lost(self, reason): + logging.info('Disconnect received. Count %d'%self._iteration) + # _iteration == 2 => Two different connections have been used. + if self._iteration == 2: + self._base_server.on_connection_lost(reason) + + def on_send_done(self): + self._base_server.on_send_done_default() + if self._base_server._stream_id == 1: + logging.info('Sending GOAWAY for stream 1') + self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=1) + + def on_request_received(self, event): + self._ready_to_send = False + self._base_server.on_request_received_default(event) + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + if sr: + time.sleep(1) + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + self._base_server.setup_send(response_data) + + def on_window_update_default(self, event): + if self._ready_to_send: + self._base_server.default_send() + +class TestcasePing(object): + """ + """ + def __init__(self, iteration): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['ConnectionLost'] = self.on_connection_lost + + def get_base_server(self): + return self._base_server + + def on_request_received(self, event): + self._base_server.default_ping() + self._base_server.on_request_received_default(event) + self._base_server.default_ping() + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._base_server.default_ping() + self._base_server.setup_send(response_data) + self._base_server.default_ping() + + def on_connection_lost(self, reason): + logging.info('Disconnect received. Ping Count %d'%self._base_server._outstanding_pings) + assert(self._base_server._outstanding_pings == 0) + self._base_server.on_connection_lost(reason) + +class H2Factory(Factory): + def __init__(self, testcase): + logging.info('In H2Factory') + self._num_streams = 0 + self._testcase = testcase + + def buildProtocol(self, addr): + self._num_streams += 1 + if self._testcase == 'rst_stream_after_header': + t = TestcaseRstStreamAfterHeader(self._num_streams) + elif self._testcase == 'rst_stream_after_data': + t = TestcaseRstStreamAfterData(self._num_streams) + elif self._testcase == 'goaway': + t = TestcaseGoaway(self._num_streams) + elif self._testcase == 'ping': + t = TestcasePing(self._num_streams) + else: + assert(0) + return t.get_base_server() + +if __name__ == "__main__": + logging.basicConfig(format = "%(levelname) -10s %(asctime)s %(module)s:%(lineno)s | %(message)s", level=logging.INFO) + parser = argparse.ArgumentParser() + parser.add_argument("test") + parser.add_argument("port") + args = parser.parse_args() + if args.test not in ['rst_stream_after_header', 'rst_stream_after_data', 'goaway', 'ping']: + print 'unknown test: ', args.test + endpoint = endpoints.TCP4ServerEndpoint(reactor, int(args.port), backlog=128) + endpoint.listen(H2Factory(args.test)) + reactor.run() diff --git a/test/http2_test/messages_pb2.py b/test/http2_test/messages_pb2.py new file mode 100644 index 00000000000..86cf5a8970f --- /dev/null +++ b/test/http2_test/messages_pb2.py @@ -0,0 +1,661 @@ +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: messages.proto + +import sys +_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +from google.protobuf.internal import enum_type_wrapper +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +from google.protobuf import descriptor_pb2 +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor.FileDescriptor( + name='messages.proto', + package='grpc.testing', + syntax='proto3', + serialized_pb=_b('\n\x0emessages.proto\x12\x0cgrpc.testing\"\x1a\n\tBoolValue\x12\r\n\x05value\x18\x01 \x01(\x08\"@\n\x07Payload\x12\'\n\x04type\x18\x01 \x01(\x0e\x32\x19.grpc.testing.PayloadType\x12\x0c\n\x04\x62ody\x18\x02 \x01(\x0c\"+\n\nEchoStatus\x12\x0c\n\x04\x63ode\x18\x01 \x01(\x05\x12\x0f\n\x07message\x18\x02 \x01(\t\"\xce\x02\n\rSimpleRequest\x12\x30\n\rresponse_type\x18\x01 \x01(\x0e\x32\x19.grpc.testing.PayloadType\x12\x15\n\rresponse_size\x18\x02 \x01(\x05\x12&\n\x07payload\x18\x03 \x01(\x0b\x32\x15.grpc.testing.Payload\x12\x15\n\rfill_username\x18\x04 \x01(\x08\x12\x18\n\x10\x66ill_oauth_scope\x18\x05 \x01(\x08\x12\x34\n\x13response_compressed\x18\x06 \x01(\x0b\x32\x17.grpc.testing.BoolValue\x12\x31\n\x0fresponse_status\x18\x07 \x01(\x0b\x32\x18.grpc.testing.EchoStatus\x12\x32\n\x11\x65xpect_compressed\x18\x08 \x01(\x0b\x32\x17.grpc.testing.BoolValue\"_\n\x0eSimpleResponse\x12&\n\x07payload\x18\x01 \x01(\x0b\x32\x15.grpc.testing.Payload\x12\x10\n\x08username\x18\x02 \x01(\t\x12\x13\n\x0boauth_scope\x18\x03 \x01(\t\"w\n\x19StreamingInputCallRequest\x12&\n\x07payload\x18\x01 \x01(\x0b\x32\x15.grpc.testing.Payload\x12\x32\n\x11\x65xpect_compressed\x18\x02 \x01(\x0b\x32\x17.grpc.testing.BoolValue\"=\n\x1aStreamingInputCallResponse\x12\x1f\n\x17\x61ggregated_payload_size\x18\x01 \x01(\x05\"d\n\x12ResponseParameters\x12\x0c\n\x04size\x18\x01 \x01(\x05\x12\x13\n\x0binterval_us\x18\x02 \x01(\x05\x12+\n\ncompressed\x18\x03 \x01(\x0b\x32\x17.grpc.testing.BoolValue\"\xe8\x01\n\x1aStreamingOutputCallRequest\x12\x30\n\rresponse_type\x18\x01 \x01(\x0e\x32\x19.grpc.testing.PayloadType\x12=\n\x13response_parameters\x18\x02 \x03(\x0b\x32 .grpc.testing.ResponseParameters\x12&\n\x07payload\x18\x03 \x01(\x0b\x32\x15.grpc.testing.Payload\x12\x31\n\x0fresponse_status\x18\x07 \x01(\x0b\x32\x18.grpc.testing.EchoStatus\"E\n\x1bStreamingOutputCallResponse\x12&\n\x07payload\x18\x01 \x01(\x0b\x32\x15.grpc.testing.Payload\"3\n\x0fReconnectParams\x12 \n\x18max_reconnect_backoff_ms\x18\x01 \x01(\x05\"3\n\rReconnectInfo\x12\x0e\n\x06passed\x18\x01 \x01(\x08\x12\x12\n\nbackoff_ms\x18\x02 \x03(\x05*\x1f\n\x0bPayloadType\x12\x10\n\x0c\x43OMPRESSABLE\x10\x00\x62\x06proto3') +) +_sym_db.RegisterFileDescriptor(DESCRIPTOR) + +_PAYLOADTYPE = _descriptor.EnumDescriptor( + name='PayloadType', + full_name='grpc.testing.PayloadType', + filename=None, + file=DESCRIPTOR, + values=[ + _descriptor.EnumValueDescriptor( + name='COMPRESSABLE', index=0, number=0, + options=None, + type=None), + ], + containing_type=None, + options=None, + serialized_start=1303, + serialized_end=1334, +) +_sym_db.RegisterEnumDescriptor(_PAYLOADTYPE) + +PayloadType = enum_type_wrapper.EnumTypeWrapper(_PAYLOADTYPE) +COMPRESSABLE = 0 + + + +_BOOLVALUE = _descriptor.Descriptor( + name='BoolValue', + full_name='grpc.testing.BoolValue', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='value', full_name='grpc.testing.BoolValue.value', index=0, + number=1, type=8, cpp_type=7, label=1, + has_default_value=False, default_value=False, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=32, + serialized_end=58, +) + + +_PAYLOAD = _descriptor.Descriptor( + name='Payload', + full_name='grpc.testing.Payload', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='type', full_name='grpc.testing.Payload.type', index=0, + number=1, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='body', full_name='grpc.testing.Payload.body', index=1, + number=2, type=12, cpp_type=9, label=1, + has_default_value=False, default_value=_b(""), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=60, + serialized_end=124, +) + + +_ECHOSTATUS = _descriptor.Descriptor( + name='EchoStatus', + full_name='grpc.testing.EchoStatus', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='code', full_name='grpc.testing.EchoStatus.code', index=0, + number=1, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='message', full_name='grpc.testing.EchoStatus.message', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=126, + serialized_end=169, +) + + +_SIMPLEREQUEST = _descriptor.Descriptor( + name='SimpleRequest', + full_name='grpc.testing.SimpleRequest', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='response_type', full_name='grpc.testing.SimpleRequest.response_type', index=0, + number=1, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='response_size', full_name='grpc.testing.SimpleRequest.response_size', index=1, + number=2, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='payload', full_name='grpc.testing.SimpleRequest.payload', index=2, + number=3, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='fill_username', full_name='grpc.testing.SimpleRequest.fill_username', index=3, + number=4, type=8, cpp_type=7, label=1, + has_default_value=False, default_value=False, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='fill_oauth_scope', full_name='grpc.testing.SimpleRequest.fill_oauth_scope', index=4, + number=5, type=8, cpp_type=7, label=1, + has_default_value=False, default_value=False, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='response_compressed', full_name='grpc.testing.SimpleRequest.response_compressed', index=5, + number=6, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='response_status', full_name='grpc.testing.SimpleRequest.response_status', index=6, + number=7, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='expect_compressed', full_name='grpc.testing.SimpleRequest.expect_compressed', index=7, + number=8, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=172, + serialized_end=506, +) + + +_SIMPLERESPONSE = _descriptor.Descriptor( + name='SimpleResponse', + full_name='grpc.testing.SimpleResponse', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='payload', full_name='grpc.testing.SimpleResponse.payload', index=0, + number=1, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='username', full_name='grpc.testing.SimpleResponse.username', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='oauth_scope', full_name='grpc.testing.SimpleResponse.oauth_scope', index=2, + number=3, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=508, + serialized_end=603, +) + + +_STREAMINGINPUTCALLREQUEST = _descriptor.Descriptor( + name='StreamingInputCallRequest', + full_name='grpc.testing.StreamingInputCallRequest', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='payload', full_name='grpc.testing.StreamingInputCallRequest.payload', index=0, + number=1, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='expect_compressed', full_name='grpc.testing.StreamingInputCallRequest.expect_compressed', index=1, + number=2, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=605, + serialized_end=724, +) + + +_STREAMINGINPUTCALLRESPONSE = _descriptor.Descriptor( + name='StreamingInputCallResponse', + full_name='grpc.testing.StreamingInputCallResponse', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='aggregated_payload_size', full_name='grpc.testing.StreamingInputCallResponse.aggregated_payload_size', index=0, + number=1, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=726, + serialized_end=787, +) + + +_RESPONSEPARAMETERS = _descriptor.Descriptor( + name='ResponseParameters', + full_name='grpc.testing.ResponseParameters', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='size', full_name='grpc.testing.ResponseParameters.size', index=0, + number=1, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='interval_us', full_name='grpc.testing.ResponseParameters.interval_us', index=1, + number=2, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='compressed', full_name='grpc.testing.ResponseParameters.compressed', index=2, + number=3, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=789, + serialized_end=889, +) + + +_STREAMINGOUTPUTCALLREQUEST = _descriptor.Descriptor( + name='StreamingOutputCallRequest', + full_name='grpc.testing.StreamingOutputCallRequest', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='response_type', full_name='grpc.testing.StreamingOutputCallRequest.response_type', index=0, + number=1, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='response_parameters', full_name='grpc.testing.StreamingOutputCallRequest.response_parameters', index=1, + number=2, type=11, cpp_type=10, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='payload', full_name='grpc.testing.StreamingOutputCallRequest.payload', index=2, + number=3, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='response_status', full_name='grpc.testing.StreamingOutputCallRequest.response_status', index=3, + number=7, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=892, + serialized_end=1124, +) + + +_STREAMINGOUTPUTCALLRESPONSE = _descriptor.Descriptor( + name='StreamingOutputCallResponse', + full_name='grpc.testing.StreamingOutputCallResponse', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='payload', full_name='grpc.testing.StreamingOutputCallResponse.payload', index=0, + number=1, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=1126, + serialized_end=1195, +) + + +_RECONNECTPARAMS = _descriptor.Descriptor( + name='ReconnectParams', + full_name='grpc.testing.ReconnectParams', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='max_reconnect_backoff_ms', full_name='grpc.testing.ReconnectParams.max_reconnect_backoff_ms', index=0, + number=1, type=5, cpp_type=1, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=1197, + serialized_end=1248, +) + + +_RECONNECTINFO = _descriptor.Descriptor( + name='ReconnectInfo', + full_name='grpc.testing.ReconnectInfo', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='passed', full_name='grpc.testing.ReconnectInfo.passed', index=0, + number=1, type=8, cpp_type=7, label=1, + has_default_value=False, default_value=False, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + _descriptor.FieldDescriptor( + name='backoff_ms', full_name='grpc.testing.ReconnectInfo.backoff_ms', index=1, + number=2, type=5, cpp_type=1, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + options=None), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=1250, + serialized_end=1301, +) + +_PAYLOAD.fields_by_name['type'].enum_type = _PAYLOADTYPE +_SIMPLEREQUEST.fields_by_name['response_type'].enum_type = _PAYLOADTYPE +_SIMPLEREQUEST.fields_by_name['payload'].message_type = _PAYLOAD +_SIMPLEREQUEST.fields_by_name['response_compressed'].message_type = _BOOLVALUE +_SIMPLEREQUEST.fields_by_name['response_status'].message_type = _ECHOSTATUS +_SIMPLEREQUEST.fields_by_name['expect_compressed'].message_type = _BOOLVALUE +_SIMPLERESPONSE.fields_by_name['payload'].message_type = _PAYLOAD +_STREAMINGINPUTCALLREQUEST.fields_by_name['payload'].message_type = _PAYLOAD +_STREAMINGINPUTCALLREQUEST.fields_by_name['expect_compressed'].message_type = _BOOLVALUE +_RESPONSEPARAMETERS.fields_by_name['compressed'].message_type = _BOOLVALUE +_STREAMINGOUTPUTCALLREQUEST.fields_by_name['response_type'].enum_type = _PAYLOADTYPE +_STREAMINGOUTPUTCALLREQUEST.fields_by_name['response_parameters'].message_type = _RESPONSEPARAMETERS +_STREAMINGOUTPUTCALLREQUEST.fields_by_name['payload'].message_type = _PAYLOAD +_STREAMINGOUTPUTCALLREQUEST.fields_by_name['response_status'].message_type = _ECHOSTATUS +_STREAMINGOUTPUTCALLRESPONSE.fields_by_name['payload'].message_type = _PAYLOAD +DESCRIPTOR.message_types_by_name['BoolValue'] = _BOOLVALUE +DESCRIPTOR.message_types_by_name['Payload'] = _PAYLOAD +DESCRIPTOR.message_types_by_name['EchoStatus'] = _ECHOSTATUS +DESCRIPTOR.message_types_by_name['SimpleRequest'] = _SIMPLEREQUEST +DESCRIPTOR.message_types_by_name['SimpleResponse'] = _SIMPLERESPONSE +DESCRIPTOR.message_types_by_name['StreamingInputCallRequest'] = _STREAMINGINPUTCALLREQUEST +DESCRIPTOR.message_types_by_name['StreamingInputCallResponse'] = _STREAMINGINPUTCALLRESPONSE +DESCRIPTOR.message_types_by_name['ResponseParameters'] = _RESPONSEPARAMETERS +DESCRIPTOR.message_types_by_name['StreamingOutputCallRequest'] = _STREAMINGOUTPUTCALLREQUEST +DESCRIPTOR.message_types_by_name['StreamingOutputCallResponse'] = _STREAMINGOUTPUTCALLRESPONSE +DESCRIPTOR.message_types_by_name['ReconnectParams'] = _RECONNECTPARAMS +DESCRIPTOR.message_types_by_name['ReconnectInfo'] = _RECONNECTINFO +DESCRIPTOR.enum_types_by_name['PayloadType'] = _PAYLOADTYPE + +BoolValue = _reflection.GeneratedProtocolMessageType('BoolValue', (_message.Message,), dict( + DESCRIPTOR = _BOOLVALUE, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.BoolValue) + )) +_sym_db.RegisterMessage(BoolValue) + +Payload = _reflection.GeneratedProtocolMessageType('Payload', (_message.Message,), dict( + DESCRIPTOR = _PAYLOAD, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.Payload) + )) +_sym_db.RegisterMessage(Payload) + +EchoStatus = _reflection.GeneratedProtocolMessageType('EchoStatus', (_message.Message,), dict( + DESCRIPTOR = _ECHOSTATUS, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.EchoStatus) + )) +_sym_db.RegisterMessage(EchoStatus) + +SimpleRequest = _reflection.GeneratedProtocolMessageType('SimpleRequest', (_message.Message,), dict( + DESCRIPTOR = _SIMPLEREQUEST, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.SimpleRequest) + )) +_sym_db.RegisterMessage(SimpleRequest) + +SimpleResponse = _reflection.GeneratedProtocolMessageType('SimpleResponse', (_message.Message,), dict( + DESCRIPTOR = _SIMPLERESPONSE, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.SimpleResponse) + )) +_sym_db.RegisterMessage(SimpleResponse) + +StreamingInputCallRequest = _reflection.GeneratedProtocolMessageType('StreamingInputCallRequest', (_message.Message,), dict( + DESCRIPTOR = _STREAMINGINPUTCALLREQUEST, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.StreamingInputCallRequest) + )) +_sym_db.RegisterMessage(StreamingInputCallRequest) + +StreamingInputCallResponse = _reflection.GeneratedProtocolMessageType('StreamingInputCallResponse', (_message.Message,), dict( + DESCRIPTOR = _STREAMINGINPUTCALLRESPONSE, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.StreamingInputCallResponse) + )) +_sym_db.RegisterMessage(StreamingInputCallResponse) + +ResponseParameters = _reflection.GeneratedProtocolMessageType('ResponseParameters', (_message.Message,), dict( + DESCRIPTOR = _RESPONSEPARAMETERS, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.ResponseParameters) + )) +_sym_db.RegisterMessage(ResponseParameters) + +StreamingOutputCallRequest = _reflection.GeneratedProtocolMessageType('StreamingOutputCallRequest', (_message.Message,), dict( + DESCRIPTOR = _STREAMINGOUTPUTCALLREQUEST, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.StreamingOutputCallRequest) + )) +_sym_db.RegisterMessage(StreamingOutputCallRequest) + +StreamingOutputCallResponse = _reflection.GeneratedProtocolMessageType('StreamingOutputCallResponse', (_message.Message,), dict( + DESCRIPTOR = _STREAMINGOUTPUTCALLRESPONSE, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.StreamingOutputCallResponse) + )) +_sym_db.RegisterMessage(StreamingOutputCallResponse) + +ReconnectParams = _reflection.GeneratedProtocolMessageType('ReconnectParams', (_message.Message,), dict( + DESCRIPTOR = _RECONNECTPARAMS, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.ReconnectParams) + )) +_sym_db.RegisterMessage(ReconnectParams) + +ReconnectInfo = _reflection.GeneratedProtocolMessageType('ReconnectInfo', (_message.Message,), dict( + DESCRIPTOR = _RECONNECTINFO, + __module__ = 'messages_pb2' + # @@protoc_insertion_point(class_scope:grpc.testing.ReconnectInfo) + )) +_sym_db.RegisterMessage(ReconnectInfo) + + +# @@protoc_insertion_point(module_scope) From 323bfa733d8b23c19d48601dd4d9d5f71ab3bc5f Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Thu, 1 Dec 2016 10:51:10 -0800 Subject: [PATCH 2/8] Broke up code in multiple modules.. --- test/http2_test/test_goaway.py | 51 ++++++++++++++++++++++++ test/http2_test/test_max_streams.py | 30 ++++++++++++++ test/http2_test/test_ping.py | 36 +++++++++++++++++ test/http2_test/test_rst_after_data.py | 23 +++++++++++ test/http2_test/test_rst_after_header.py | 19 +++++++++ 5 files changed, 159 insertions(+) create mode 100644 test/http2_test/test_goaway.py create mode 100644 test/http2_test/test_max_streams.py create mode 100644 test/http2_test/test_ping.py create mode 100644 test/http2_test/test_rst_after_data.py create mode 100644 test/http2_test/test_rst_after_header.py diff --git a/test/http2_test/test_goaway.py b/test/http2_test/test_goaway.py new file mode 100644 index 00000000000..419bd7b3f8c --- /dev/null +++ b/test/http2_test/test_goaway.py @@ -0,0 +1,51 @@ +import logging +import http2_base_server + +class TestcaseGoaway(object): + """ + This test does the following: + Process incoming request normally, i.e. send headers, data and trailers. + Then send a GOAWAY frame with the stream id of the processed request. + It assert that the next request is made on a different TCP connection. + """ + def __init__(self, iteration): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['WindowUpdated'] = self.on_window_update_default + self._base_server._handlers['SendDone'] = self.on_send_done + self._base_server._handlers['ConnectionLost'] = self.on_connection_lost + self._ready_to_send = False + self._iteration = iteration + + def get_base_server(self): + return self._base_server + + def on_connection_lost(self, reason): + logging.info('Disconnect received. Count %d'%self._iteration) + # _iteration == 2 => Two different connections have been used. + if self._iteration == 2: + self._base_server.on_connection_lost(reason) + + def on_send_done(self): + self._base_server.on_send_done_default() + if self._base_server._stream_id == 1: + logging.info('Sending GOAWAY for stream 1') + self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=1) + + def on_request_received(self, event): + self._ready_to_send = False + self._base_server.on_request_received_default(event) + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + if sr: + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + self._base_server.setup_send(response_data) + + def on_window_update_default(self, event): + if self._ready_to_send: + self._base_server.default_send() diff --git a/test/http2_test/test_max_streams.py b/test/http2_test/test_max_streams.py new file mode 100644 index 00000000000..a85dde48b5b --- /dev/null +++ b/test/http2_test/test_max_streams.py @@ -0,0 +1,30 @@ +import logging +import http2_base_server +from hyperframe.frame import SettingsFrame + +class TestcaseSettingsMaxStreams(object): + """ + This test sets MAX_CONCURRENT_STREAMS to 1 and asserts that at any point + only 1 stream is active. + """ + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['ConnectionMade'] = self.on_connection_made + + def get_base_server(self): + return self._base_server + + def on_connection_made(self): + logging.info('Connection Made') + self._base_server._conn.initiate_connection() + self._base_server._conn.update_settings({SettingsFrame.MAX_CONCURRENT_STREAMS: 1}) + self._base_server.transport.setTcpNoDelay(True) + self._base_server.transport.write(self._base_server._conn.data_to_send()) + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._base_server.setup_send(response_data) diff --git a/test/http2_test/test_ping.py b/test/http2_test/test_ping.py new file mode 100644 index 00000000000..bade9df9b1e --- /dev/null +++ b/test/http2_test/test_ping.py @@ -0,0 +1,36 @@ +import logging +import http2_base_server + +class TestcasePing(object): + """ + This test injects PING frames before and after header and data. Keeps count + of outstanding ping response and asserts when the count is non-zero at the + end of the test. + """ + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['ConnectionLost'] = self.on_connection_lost + + def get_base_server(self): + return self._base_server + + def on_request_received(self, event): + self._base_server.default_ping() + self._base_server.on_request_received_default(event) + self._base_server.default_ping() + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._base_server.default_ping() + self._base_server.setup_send(response_data) + self._base_server.default_ping() + + def on_connection_lost(self, reason): + logging.info('Disconnect received. Ping Count %d'%self._base_server._outstanding_pings) + assert(self._base_server._outstanding_pings == 0) + self._base_server.on_connection_lost(reason) diff --git a/test/http2_test/test_rst_after_data.py b/test/http2_test/test_rst_after_data.py new file mode 100644 index 00000000000..ef8d4084d96 --- /dev/null +++ b/test/http2_test/test_rst_after_data.py @@ -0,0 +1,23 @@ +import http2_base_server + +class TestcaseRstStreamAfterData(object): + """ + In response to an incoming request, this test sends headers, followed by + data, followed by a reset stream frame. Client asserts that the RPC failed. + """ + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['DataReceived'] = self.on_data_received + + def get_base_server(self): + return self._base_server + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + assert(sr is not None) + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + self._base_server.setup_send(response_data) + # send reset stream + self._base_server.send_reset_stream() diff --git a/test/http2_test/test_rst_after_header.py b/test/http2_test/test_rst_after_header.py new file mode 100644 index 00000000000..e9a6b1129cf --- /dev/null +++ b/test/http2_test/test_rst_after_header.py @@ -0,0 +1,19 @@ +import http2_base_server + +class TestcaseRstStreamAfterHeader(object): + """ + In response to an incoming request, this test sends headers, followed by + a reset stream frame. Client asserts that the RPC failed. + """ + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['RequestReceived'] = self.on_request_received + + def get_base_server(self): + return self._base_server + + def on_request_received(self, event): + # send initial headers + self._base_server.on_request_received_default(event) + # send reset stream + self._base_server.send_reset_stream() From 28d198008a6da19d28f0c34f7b3a4441f9c9d9ff Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Thu, 1 Dec 2016 10:56:52 -0800 Subject: [PATCH 3/8] minor cleanup.. --- test/http2_test/http2_base_server.py | 19 ++-- test/http2_test/http2_test_server.py | 149 +++------------------------ 2 files changed, 26 insertions(+), 142 deletions(-) diff --git a/test/http2_test/http2_base_server.py b/test/http2_test/http2_base_server.py index 07bd37cae9a..91caa74fcc2 100644 --- a/test/http2_test/http2_base_server.py +++ b/test/http2_test/http2_base_server.py @@ -1,16 +1,11 @@ import struct import messages_pb2 -import functools -import argparse import logging -import time -from twisted.internet.defer import Deferred, inlineCallbacks -from twisted.internet.protocol import Protocol, Factory -from twisted.internet import endpoints, reactor, error, defer +from twisted.internet.protocol import Protocol +from twisted.internet import reactor from h2.connection import H2Connection from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged, PingAcknowledged -from threading import Lock READ_CHUNK_SIZE = 16384 GRPC_HEADER_SIZE = 5 @@ -20,6 +15,7 @@ class H2ProtocolBaseServer(Protocol): self._conn = H2Connection(client_side=False) self._recv_buffer = '' self._handlers = {} + self._handlers['ConnectionMade'] = self.on_connection_made_default self._handlers['DataReceived'] = self.on_data_received_default self._handlers['WindowUpdated'] = self.on_window_update_default self._handlers['RequestReceived'] = self.on_request_received_default @@ -33,14 +29,17 @@ class H2ProtocolBaseServer(Protocol): self._handlers = handlers def connectionMade(self): + self._handlers['ConnectionMade']() + + def connectionLost(self, reason): + self._handlers['ConnectionLost'](reason) + + def on_connection_made_default(self): logging.info('Connection Made') self._conn.initiate_connection() self.transport.setTcpNoDelay(True) self.transport.write(self._conn.data_to_send()) - def connectionLost(self, reason): - self._handlers['ConnectionLost'](reason) - def on_connection_lost(self, reason): logging.info('Disconnected %s'%reason) reactor.callFromThread(reactor.stop) diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index be5f1593ebc..7ec781d2aa5 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -1,135 +1,17 @@ """ HTTP2 Test Server. Highly experimental work in progress. """ -import struct -import messages_pb2 import argparse import logging -import time -from twisted.internet.defer import Deferred, inlineCallbacks -from twisted.internet.protocol import Protocol, Factory -from twisted.internet import endpoints, reactor, error, defer -from h2.connection import H2Connection -from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged -from threading import Lock +from twisted.internet.protocol import Factory +from twisted.internet import endpoints, reactor import http2_base_server - -READ_CHUNK_SIZE = 16384 -GRPC_HEADER_SIZE = 5 - -class TestcaseRstStreamAfterHeader(object): - def __init__(self): - self._base_server = http2_base_server.H2ProtocolBaseServer() - self._base_server._handlers['RequestReceived'] = self.on_request_received - - def get_base_server(self): - return self._base_server - - def on_request_received(self, event): - # send initial headers - self._base_server.on_request_received_default(event) - # send reset stream - self._base_server.send_reset_stream() - -class TestcaseRstStreamAfterData(object): - def __init__(self): - self._base_server = http2_base_server.H2ProtocolBaseServer() - self._base_server._handlers['DataReceived'] = self.on_data_received - - def get_base_server(self): - return self._base_server - - def on_data_received(self, event): - self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - assert(sr is not None) - assert(sr.response_size <= 2048) # so it can fit into one flow control window - response_data = self._base_server.default_response_data(sr.response_size) - self._ready_to_send = True - self._base_server.setup_send(response_data) - # send reset stream - self._base_server.send_reset_stream() - -class TestcaseGoaway(object): - """ - Process incoming request normally. After sending trailer response, - send GOAWAY with stream id = 1. - assert that the next request is made on a different connection. - """ - def __init__(self, iteration): - self._base_server = http2_base_server.H2ProtocolBaseServer() - self._base_server._handlers['RequestReceived'] = self.on_request_received - self._base_server._handlers['DataReceived'] = self.on_data_received - self._base_server._handlers['WindowUpdated'] = self.on_window_update_default - self._base_server._handlers['SendDone'] = self.on_send_done - self._base_server._handlers['ConnectionLost'] = self.on_connection_lost - self._ready_to_send = False - self._iteration = iteration - - def get_base_server(self): - return self._base_server - - def on_connection_lost(self, reason): - logging.info('Disconnect received. Count %d'%self._iteration) - # _iteration == 2 => Two different connections have been used. - if self._iteration == 2: - self._base_server.on_connection_lost(reason) - - def on_send_done(self): - self._base_server.on_send_done_default() - if self._base_server._stream_id == 1: - logging.info('Sending GOAWAY for stream 1') - self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=1) - - def on_request_received(self, event): - self._ready_to_send = False - self._base_server.on_request_received_default(event) - - def on_data_received(self, event): - self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - if sr: - time.sleep(1) - logging.info('Creating response size = %s'%sr.response_size) - response_data = self._base_server.default_response_data(sr.response_size) - self._ready_to_send = True - self._base_server.setup_send(response_data) - - def on_window_update_default(self, event): - if self._ready_to_send: - self._base_server.default_send() - -class TestcasePing(object): - """ - """ - def __init__(self, iteration): - self._base_server = http2_base_server.H2ProtocolBaseServer() - self._base_server._handlers['RequestReceived'] = self.on_request_received - self._base_server._handlers['DataReceived'] = self.on_data_received - self._base_server._handlers['ConnectionLost'] = self.on_connection_lost - - def get_base_server(self): - return self._base_server - - def on_request_received(self, event): - self._base_server.default_ping() - self._base_server.on_request_received_default(event) - self._base_server.default_ping() - - def on_data_received(self, event): - self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - logging.info('Creating response size = %s'%sr.response_size) - response_data = self._base_server.default_response_data(sr.response_size) - self._base_server.default_ping() - self._base_server.setup_send(response_data) - self._base_server.default_ping() - - def on_connection_lost(self, reason): - logging.info('Disconnect received. Ping Count %d'%self._base_server._outstanding_pings) - assert(self._base_server._outstanding_pings == 0) - self._base_server.on_connection_lost(reason) +import test_rst_after_header +import test_rst_after_data +import test_goaway +import test_ping +import test_max_streams class H2Factory(Factory): def __init__(self, testcase): @@ -139,15 +21,18 @@ class H2Factory(Factory): def buildProtocol(self, addr): self._num_streams += 1 - if self._testcase == 'rst_stream_after_header': - t = TestcaseRstStreamAfterHeader(self._num_streams) - elif self._testcase == 'rst_stream_after_data': - t = TestcaseRstStreamAfterData(self._num_streams) + if self._testcase == 'rst_after_header': + t = test_rst_after_header.TestcaseRstStreamAfterHeader() + elif self._testcase == 'rst_after_data': + t = test_rst_after_data.TestcaseRstStreamAfterData() elif self._testcase == 'goaway': - t = TestcaseGoaway(self._num_streams) + t = test_goaway.TestcaseGoaway(self._num_streams) elif self._testcase == 'ping': - t = TestcasePing(self._num_streams) + t = test_ping.TestcasePing() + elif self._testcase == 'max_streams': + t = TestcaseSettingsMaxStreams(self._num_streams) else: + logging.error('Unknown test case: %s'%self._testcase) assert(0) return t.get_base_server() @@ -157,7 +42,7 @@ if __name__ == "__main__": parser.add_argument("test") parser.add_argument("port") args = parser.parse_args() - if args.test not in ['rst_stream_after_header', 'rst_stream_after_data', 'goaway', 'ping']: + if args.test not in ['rst_after_header', 'rst_after_data', 'goaway', 'ping', 'max_streams']: print 'unknown test: ', args.test endpoint = endpoints.TCP4ServerEndpoint(reactor, int(args.port), backlog=128) endpoint.listen(H2Factory(args.test)) From 4350e748e470be92abab9bb1f72ad8eb1ede5486 Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Thu, 1 Dec 2016 14:24:22 -0800 Subject: [PATCH 4/8] ability to deal with multiple streams in flight. --- test/http2_test/http2_base_server.py | 76 ++++++++++++++++---------- test/http2_test/test_goaway.py | 19 +++---- test/http2_test/test_max_streams.py | 9 +-- test/http2_test/test_ping.py | 13 +++-- test/http2_test/test_rst_after_data.py | 14 ++--- 5 files changed, 72 insertions(+), 59 deletions(-) diff --git a/test/http2_test/http2_base_server.py b/test/http2_test/http2_base_server.py index 91caa74fcc2..44fb575c0f3 100644 --- a/test/http2_test/http2_base_server.py +++ b/test/http2_test/http2_base_server.py @@ -6,6 +6,7 @@ from twisted.internet.protocol import Protocol from twisted.internet import reactor from h2.connection import H2Connection from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged, PingAcknowledged +from h2.exceptions import ProtocolError READ_CHUNK_SIZE = 16384 GRPC_HEADER_SIZE = 5 @@ -13,7 +14,7 @@ GRPC_HEADER_SIZE = 5 class H2ProtocolBaseServer(Protocol): def __init__(self): self._conn = H2Connection(client_side=False) - self._recv_buffer = '' + self._recv_buffer = {} self._handlers = {} self._handlers['ConnectionMade'] = self.on_connection_made_default self._handlers['DataReceived'] = self.on_data_received_default @@ -23,6 +24,7 @@ class H2ProtocolBaseServer(Protocol): self._handlers['ConnectionLost'] = self.on_connection_lost self._handlers['PingAcknowledged'] = self.on_ping_acknowledged_default self._stream_status = {} + self._send_remaining = {} self._outstanding_pings = 0 def set_handlers(self, handlers): @@ -45,18 +47,23 @@ class H2ProtocolBaseServer(Protocol): reactor.callFromThread(reactor.stop) def dataReceived(self, data): - events = self._conn.receive_data(data) + try: + events = self._conn.receive_data(data) + except ProtocolError: + # this try/except block catches exceptions due to race between sending + # GOAWAY and processing a response in flight. + return if self._conn.data_to_send: self.transport.write(self._conn.data_to_send()) for event in events: if isinstance(event, RequestReceived) and self._handlers.has_key('RequestReceived'): - logging.info('RequestReceived Event') + logging.info('RequestReceived Event for stream: %d'%event.stream_id) self._handlers['RequestReceived'](event) elif isinstance(event, DataReceived) and self._handlers.has_key('DataReceived'): - logging.info('DataReceived Event') + logging.info('DataReceived Event for stream: %d'%event.stream_id) self._handlers['DataReceived'](event) elif isinstance(event, WindowUpdated) and self._handlers.has_key('WindowUpdated'): - logging.info('WindowUpdated Event') + logging.info('WindowUpdated Event for stream: %d'%event.stream_id) self._handlers['WindowUpdated'](event) elif isinstance(event, PingAcknowledged) and self._handlers.has_key('PingAcknowledged'): logging.info('PingAcknowledged Event') @@ -68,10 +75,10 @@ class H2ProtocolBaseServer(Protocol): def on_data_received_default(self, event): self._conn.acknowledge_received_data(len(event.data), event.stream_id) - self._recv_buffer += event.data + self._recv_buffer[event.stream_id] += event.data def on_request_received_default(self, event): - self._recv_buffer = '' + self._recv_buffer[event.stream_id] = '' self._stream_id = event.stream_id self._stream_status[event.stream_id] = True self._conn.send_headers( @@ -86,48 +93,57 @@ class H2ProtocolBaseServer(Protocol): self.transport.write(self._conn.data_to_send()) def on_window_update_default(self, event): - pass + # send pending data, if any + self.default_send(event.stream_id) def send_reset_stream(self): self._conn.reset_stream(self._stream_id) self.transport.write(self._conn.data_to_send()) - def setup_send(self, data_to_send): - self._send_remaining = len(data_to_send) + def setup_send(self, data_to_send, stream_id): + logging.info('Setting up data to send for stream_id: %d'%stream_id) + self._send_remaining[stream_id] = len(data_to_send) self._send_offset = 0 self._data_to_send = data_to_send - self.default_send() + self.default_send(stream_id) - def default_send(self): - while self._send_remaining > 0: - lfcw = self._conn.local_flow_control_window(self._stream_id) + def default_send(self, stream_id): + if not self._send_remaining.has_key(stream_id): + # not setup to send data yet + return + + while self._send_remaining[stream_id] > 0: + if self._stream_status[stream_id] is False: + logging.info('Stream %d is closed.'%stream_id) + break + lfcw = self._conn.local_flow_control_window(stream_id) if lfcw == 0: break chunk_size = min(lfcw, READ_CHUNK_SIZE) - bytes_to_send = min(chunk_size, self._send_remaining) + bytes_to_send = min(chunk_size, self._send_remaining[stream_id]) logging.info('flow_control_window = %d. sending [%d:%d] stream_id %d'% (lfcw, self._send_offset, self._send_offset + bytes_to_send, - self._stream_id)) + stream_id)) data = self._data_to_send[self._send_offset : self._send_offset + bytes_to_send] - self._conn.send_data(self._stream_id, data, False) - self._send_remaining -= bytes_to_send + self._conn.send_data(stream_id, data, False) + self._send_remaining[stream_id] -= bytes_to_send self._send_offset += bytes_to_send - if self._send_remaining == 0: - self._handlers['SendDone']() + if self._send_remaining[stream_id] == 0: + self._handlers['SendDone'](stream_id) def default_ping(self): self._outstanding_pings += 1 self._conn.ping(b'\x00'*8) self.transport.write(self._conn.data_to_send()) - def on_send_done_default(self): - if self._stream_status[self._stream_id]: - self._stream_status[self._stream_id] = False - self.default_send_trailer() + def on_send_done_default(self, stream_id): + if self._stream_status[stream_id]: + self._stream_status[stream_id] = False + self.default_send_trailer(stream_id) - def default_send_trailer(self): - logging.info('Sending trailer for stream id %d'%self._stream_id) - self._conn.send_headers(self._stream_id, + def default_send_trailer(self, stream_id): + logging.info('Sending trailer for stream id %d'%stream_id) + self._conn.send_headers(stream_id, headers=[ ('grpc-status', '0') ], end_stream=True ) @@ -141,8 +157,8 @@ class H2ProtocolBaseServer(Protocol): response_data = b'\x00' + struct.pack('i', len(serialized_resp_proto))[::-1] + serialized_resp_proto return response_data - @staticmethod - def parse_received_data(recv_buffer): + def parse_received_data(self, stream_id): + recv_buffer = self._recv_buffer[stream_id] """ returns a grpc framed string of bytes containing response proto of the size asked in request """ grpc_msg_size = struct.unpack('i',recv_buffer[1:5][::-1])[0] @@ -152,5 +168,5 @@ class H2ProtocolBaseServer(Protocol): req_proto_str = recv_buffer[5:5+grpc_msg_size] sr = messages_pb2.SimpleRequest() sr.ParseFromString(req_proto_str) - logging.info('Parsed request: response_size=%s'%sr.response_size) + logging.info('Parsed request for stream %d: response_size=%s'%(stream_id, sr.response_size)) return sr diff --git a/test/http2_test/test_goaway.py b/test/http2_test/test_goaway.py index 419bd7b3f8c..7dd7cb7948f 100644 --- a/test/http2_test/test_goaway.py +++ b/test/http2_test/test_goaway.py @@ -12,7 +12,6 @@ class TestcaseGoaway(object): self._base_server = http2_base_server.H2ProtocolBaseServer() self._base_server._handlers['RequestReceived'] = self.on_request_received self._base_server._handlers['DataReceived'] = self.on_data_received - self._base_server._handlers['WindowUpdated'] = self.on_window_update_default self._base_server._handlers['SendDone'] = self.on_send_done self._base_server._handlers['ConnectionLost'] = self.on_connection_lost self._ready_to_send = False @@ -27,11 +26,11 @@ class TestcaseGoaway(object): if self._iteration == 2: self._base_server.on_connection_lost(reason) - def on_send_done(self): - self._base_server.on_send_done_default() - if self._base_server._stream_id == 1: - logging.info('Sending GOAWAY for stream 1') - self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=1) + def on_send_done(self, stream_id): + self._base_server.on_send_done_default(stream_id) + logging.info('Sending GOAWAY for stream %d:'%stream_id) + self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=stream_id) + self._base_server._stream_status[stream_id] = False def on_request_received(self, event): self._ready_to_send = False @@ -39,13 +38,9 @@ class TestcaseGoaway(object): def on_data_received(self, event): self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) + sr = self._base_server.parse_received_data(event.stream_id) if sr: logging.info('Creating response size = %s'%sr.response_size) response_data = self._base_server.default_response_data(sr.response_size) self._ready_to_send = True - self._base_server.setup_send(response_data) - - def on_window_update_default(self, event): - if self._ready_to_send: - self._base_server.default_send() + self._base_server.setup_send(response_data, event.stream_id) diff --git a/test/http2_test/test_max_streams.py b/test/http2_test/test_max_streams.py index a85dde48b5b..deb26770c3a 100644 --- a/test/http2_test/test_max_streams.py +++ b/test/http2_test/test_max_streams.py @@ -24,7 +24,8 @@ class TestcaseSettingsMaxStreams(object): def on_data_received(self, event): self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - logging.info('Creating response size = %s'%sr.response_size) - response_data = self._base_server.default_response_data(sr.response_size) - self._base_server.setup_send(response_data) + sr = self._base_server.parse_received_data(event.stream_id) + if sr: + logging.info('Creating response of size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._base_server.setup_send(response_data, event.stream_id) diff --git a/test/http2_test/test_ping.py b/test/http2_test/test_ping.py index bade9df9b1e..2e6dadbc079 100644 --- a/test/http2_test/test_ping.py +++ b/test/http2_test/test_ping.py @@ -23,12 +23,13 @@ class TestcasePing(object): def on_data_received(self, event): self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - logging.info('Creating response size = %s'%sr.response_size) - response_data = self._base_server.default_response_data(sr.response_size) - self._base_server.default_ping() - self._base_server.setup_send(response_data) - self._base_server.default_ping() + sr = self._base_server.parse_received_data(event.stream_id) + if sr: + logging.info('Creating response size = %s'%sr.response_size) + response_data = self._base_server.default_response_data(sr.response_size) + self._base_server.default_ping() + self._base_server.setup_send(response_data, event.stream_id) + self._base_server.default_ping() def on_connection_lost(self, reason): logging.info('Disconnect received. Ping Count %d'%self._base_server._outstanding_pings) diff --git a/test/http2_test/test_rst_after_data.py b/test/http2_test/test_rst_after_data.py index ef8d4084d96..c4ff56c889c 100644 --- a/test/http2_test/test_rst_after_data.py +++ b/test/http2_test/test_rst_after_data.py @@ -14,10 +14,10 @@ class TestcaseRstStreamAfterData(object): def on_data_received(self, event): self._base_server.on_data_received_default(event) - sr = self._base_server.parse_received_data(self._base_server._recv_buffer) - assert(sr is not None) - response_data = self._base_server.default_response_data(sr.response_size) - self._ready_to_send = True - self._base_server.setup_send(response_data) - # send reset stream - self._base_server.send_reset_stream() + sr = self._base_server.parse_received_data(event.stream_id) + if sr: + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + self._base_server.setup_send(response_data, event.stream_id) + # send reset stream + self._base_server.send_reset_stream() From ecc32885f097ece2c46187166c496e3dd2d70229 Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Thu, 1 Dec 2016 15:22:50 -0800 Subject: [PATCH 5/8] WIP --- test/http2_test/http2_base_server.py | 11 ++++++----- test/http2_test/http2_test_server.py | 1 + test/http2_test/test_goaway.py | 4 +++- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/test/http2_test/http2_base_server.py b/test/http2_test/http2_base_server.py index 44fb575c0f3..cbc26b11b1f 100644 --- a/test/http2_test/http2_base_server.py +++ b/test/http2_test/http2_base_server.py @@ -113,9 +113,6 @@ class H2ProtocolBaseServer(Protocol): return while self._send_remaining[stream_id] > 0: - if self._stream_status[stream_id] is False: - logging.info('Stream %d is closed.'%stream_id) - break lfcw = self._conn.local_flow_control_window(stream_id) if lfcw == 0: break @@ -125,7 +122,11 @@ class H2ProtocolBaseServer(Protocol): (lfcw, self._send_offset, self._send_offset + bytes_to_send, stream_id)) data = self._data_to_send[self._send_offset : self._send_offset + bytes_to_send] - self._conn.send_data(stream_id, data, False) + try: + self._conn.send_data(stream_id, data, False) + except ProtocolError: + logging.info('Stream %d is closed'%stream_id) + break self._send_remaining[stream_id] -= bytes_to_send self._send_offset += bytes_to_send if self._send_remaining[stream_id] == 0: @@ -163,7 +164,7 @@ class H2ProtocolBaseServer(Protocol): asked in request """ grpc_msg_size = struct.unpack('i',recv_buffer[1:5][::-1])[0] if len(recv_buffer) != GRPC_HEADER_SIZE + grpc_msg_size: - logging.error('not enough data to decode req proto. size = %d, needed %s'%(len(recv_buffer), 5+grpc_msg_size)) + #logging.error('not enough data to decode req proto. size = %d, needed %s'%(len(recv_buffer), 5+grpc_msg_size)) return None req_proto_str = recv_buffer[5:5+grpc_msg_size] sr = messages_pb2.SimpleRequest() diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index 7ec781d2aa5..35308c47874 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -21,6 +21,7 @@ class H2Factory(Factory): def buildProtocol(self, addr): self._num_streams += 1 + logging.info('New Connection: %d'%self._num_streams) if self._testcase == 'rst_after_header': t = test_rst_after_header.TestcaseRstStreamAfterHeader() elif self._testcase == 'rst_after_data': diff --git a/test/http2_test/test_goaway.py b/test/http2_test/test_goaway.py index 7dd7cb7948f..4c46cab8f9c 100644 --- a/test/http2_test/test_goaway.py +++ b/test/http2_test/test_goaway.py @@ -1,4 +1,5 @@ import logging +import time import http2_base_server class TestcaseGoaway(object): @@ -23,11 +24,12 @@ class TestcaseGoaway(object): def on_connection_lost(self, reason): logging.info('Disconnect received. Count %d'%self._iteration) # _iteration == 2 => Two different connections have been used. - if self._iteration == 2: + if self._iteration == 200: self._base_server.on_connection_lost(reason) def on_send_done(self, stream_id): self._base_server.on_send_done_default(stream_id) + time.sleep(1) logging.info('Sending GOAWAY for stream %d:'%stream_id) self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=stream_id) self._base_server._stream_status[stream_id] = False From 5b7070a15b7cd13115a15a619d447792d99d78a6 Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Thu, 1 Dec 2016 16:56:07 -0800 Subject: [PATCH 6/8] bugfix in rst_after_data --- test/http2_test/http2_test_server.py | 2 +- test/http2_test/test_rst_after_data.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index 35308c47874..c74fc4b1fb9 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -31,7 +31,7 @@ class H2Factory(Factory): elif self._testcase == 'ping': t = test_ping.TestcasePing() elif self._testcase == 'max_streams': - t = TestcaseSettingsMaxStreams(self._num_streams) + t = test_max_streams.TestcaseSettingsMaxStreams() else: logging.error('Unknown test case: %s'%self._testcase) assert(0) diff --git a/test/http2_test/test_rst_after_data.py b/test/http2_test/test_rst_after_data.py index c4ff56c889c..30163c2cebb 100644 --- a/test/http2_test/test_rst_after_data.py +++ b/test/http2_test/test_rst_after_data.py @@ -8,6 +8,7 @@ class TestcaseRstStreamAfterData(object): def __init__(self): self._base_server = http2_base_server.H2ProtocolBaseServer() self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['SendDone'] = self.on_send_done def get_base_server(self): return self._base_server @@ -20,4 +21,7 @@ class TestcaseRstStreamAfterData(object): self._ready_to_send = True self._base_server.setup_send(response_data, event.stream_id) # send reset stream - self._base_server.send_reset_stream() + + def on_send_done(self, stream_id): + self._base_server.send_reset_stream() + self._base_server._stream_status[stream_id] = False From a16ea7f9b129d34edbdb867e720d4a2ea71e9a7f Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Fri, 2 Dec 2016 10:17:03 -0800 Subject: [PATCH 7/8] added new test (rst_during_data) --- test/http2_test/http2_test_server.py | 41 +++++++++++++++---------- test/http2_test/test_goaway.py | 2 +- test/http2_test/test_rst_after_data.py | 1 + test/http2_test/test_rst_during_data.py | 30 ++++++++++++++++++ 4 files changed, 56 insertions(+), 18 deletions(-) create mode 100644 test/http2_test/test_rst_during_data.py diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index c74fc4b1fb9..5270ee4255e 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -9,10 +9,20 @@ from twisted.internet import endpoints, reactor import http2_base_server import test_rst_after_header import test_rst_after_data +import test_rst_during_data import test_goaway import test_ping import test_max_streams +test_case_mappings = { + 'rst_after_header': test_rst_after_header.TestcaseRstStreamAfterHeader, + 'rst_after_data': test_rst_after_data.TestcaseRstStreamAfterData, + 'rst_during_data': test_rst_during_data.TestcaseRstStreamDuringData, + 'goaway': test_goaway.TestcaseGoaway, + 'ping': test_ping.TestcasePing, + 'max_streams': test_max_streams.TestcaseSettingsMaxStreams, +} + class H2Factory(Factory): def __init__(self, testcase): logging.info('In H2Factory') @@ -22,20 +32,16 @@ class H2Factory(Factory): def buildProtocol(self, addr): self._num_streams += 1 logging.info('New Connection: %d'%self._num_streams) - if self._testcase == 'rst_after_header': - t = test_rst_after_header.TestcaseRstStreamAfterHeader() - elif self._testcase == 'rst_after_data': - t = test_rst_after_data.TestcaseRstStreamAfterData() - elif self._testcase == 'goaway': - t = test_goaway.TestcaseGoaway(self._num_streams) - elif self._testcase == 'ping': - t = test_ping.TestcasePing() - elif self._testcase == 'max_streams': - t = test_max_streams.TestcaseSettingsMaxStreams() - else: + if not test_case_mappings.has_key(self._testcase): logging.error('Unknown test case: %s'%self._testcase) assert(0) - return t.get_base_server() + else: + t = test_case_mappings[self._testcase] + + if self._testcase == 'goaway': + return t(self._num_streams).get_base_server() + else: + return t().get_base_server() if __name__ == "__main__": logging.basicConfig(format = "%(levelname) -10s %(asctime)s %(module)s:%(lineno)s | %(message)s", level=logging.INFO) @@ -43,8 +49,9 @@ if __name__ == "__main__": parser.add_argument("test") parser.add_argument("port") args = parser.parse_args() - if args.test not in ['rst_after_header', 'rst_after_data', 'goaway', 'ping', 'max_streams']: - print 'unknown test: ', args.test - endpoint = endpoints.TCP4ServerEndpoint(reactor, int(args.port), backlog=128) - endpoint.listen(H2Factory(args.test)) - reactor.run() + if args.test not in test_case_mappings.keys(): + logging.error('unknown test: %s'%args.test) + else: + endpoint = endpoints.TCP4ServerEndpoint(reactor, int(args.port), backlog=128) + endpoint.listen(H2Factory(args.test)) + reactor.run() diff --git a/test/http2_test/test_goaway.py b/test/http2_test/test_goaway.py index 4c46cab8f9c..7a915402b2b 100644 --- a/test/http2_test/test_goaway.py +++ b/test/http2_test/test_goaway.py @@ -24,7 +24,7 @@ class TestcaseGoaway(object): def on_connection_lost(self, reason): logging.info('Disconnect received. Count %d'%self._iteration) # _iteration == 2 => Two different connections have been used. - if self._iteration == 200: + if self._iteration == 2: self._base_server.on_connection_lost(reason) def on_send_done(self, stream_id): diff --git a/test/http2_test/test_rst_after_data.py b/test/http2_test/test_rst_after_data.py index 30163c2cebb..f7927cbcd30 100644 --- a/test/http2_test/test_rst_after_data.py +++ b/test/http2_test/test_rst_after_data.py @@ -4,6 +4,7 @@ class TestcaseRstStreamAfterData(object): """ In response to an incoming request, this test sends headers, followed by data, followed by a reset stream frame. Client asserts that the RPC failed. + Client needs to deliver the complete message to the application layer. """ def __init__(self): self._base_server = http2_base_server.H2ProtocolBaseServer() diff --git a/test/http2_test/test_rst_during_data.py b/test/http2_test/test_rst_during_data.py new file mode 100644 index 00000000000..105c9403bb7 --- /dev/null +++ b/test/http2_test/test_rst_during_data.py @@ -0,0 +1,30 @@ +import http2_base_server + +class TestcaseRstStreamDuringData(object): + """ + In response to an incoming request, this test sends headers, followed by + some data, followed by a reset stream frame. Client asserts that the RPC + failed and does not deliver the message to the application. + """ + def __init__(self): + self._base_server = http2_base_server.H2ProtocolBaseServer() + self._base_server._handlers['DataReceived'] = self.on_data_received + self._base_server._handlers['SendDone'] = self.on_send_done + + def get_base_server(self): + return self._base_server + + def on_data_received(self, event): + self._base_server.on_data_received_default(event) + sr = self._base_server.parse_received_data(event.stream_id) + if sr: + response_data = self._base_server.default_response_data(sr.response_size) + self._ready_to_send = True + response_len = len(response_data) + truncated_response_data = response_data[0:response_len/2] + self._base_server.setup_send(truncated_response_data, event.stream_id) + # send reset stream + + def on_send_done(self, stream_id): + self._base_server.send_reset_stream() + self._base_server._stream_status[stream_id] = False From 8c57917f56bc7af695f2c92f8e5678160da4276e Mon Sep 17 00:00:00 2001 From: Makarand Dharmapurikar Date: Mon, 12 Dec 2016 13:58:15 -0800 Subject: [PATCH 8/8] fixed feedback from review --- test/http2_test/http2_base_server.py | 65 +++++++++++++------------ test/http2_test/http2_test_server.py | 36 +++++++------- test/http2_test/test_goaway.py | 10 ++-- test/http2_test/test_max_streams.py | 9 ++-- test/http2_test/test_ping.py | 5 +- test/http2_test/test_rst_during_data.py | 1 - 6 files changed, 67 insertions(+), 59 deletions(-) diff --git a/test/http2_test/http2_base_server.py b/test/http2_test/http2_base_server.py index cbc26b11b1f..4d356d4385b 100644 --- a/test/http2_test/http2_base_server.py +++ b/test/http2_test/http2_base_server.py @@ -1,19 +1,19 @@ -import struct -import messages_pb2 import logging +import messages_pb2 +import struct -from twisted.internet.protocol import Protocol -from twisted.internet import reactor -from h2.connection import H2Connection -from h2.events import RequestReceived, DataReceived, WindowUpdated, RemoteSettingsChanged, PingAcknowledged -from h2.exceptions import ProtocolError +import h2 +import h2.connection +import twisted +import twisted.internet +import twisted.internet.protocol -READ_CHUNK_SIZE = 16384 -GRPC_HEADER_SIZE = 5 +_READ_CHUNK_SIZE = 16384 +_GRPC_HEADER_SIZE = 5 -class H2ProtocolBaseServer(Protocol): +class H2ProtocolBaseServer(twisted.internet.protocol.Protocol): def __init__(self): - self._conn = H2Connection(client_side=False) + self._conn = h2.connection.H2Connection(client_side=False) self._recv_buffer = {} self._handlers = {} self._handlers['ConnectionMade'] = self.on_connection_made_default @@ -43,34 +43,35 @@ class H2ProtocolBaseServer(Protocol): self.transport.write(self._conn.data_to_send()) def on_connection_lost(self, reason): - logging.info('Disconnected %s'%reason) - reactor.callFromThread(reactor.stop) + logging.info('Disconnected %s' % reason) + twisted.internet.reactor.callFromThread(twisted.internet.reactor.stop) def dataReceived(self, data): try: events = self._conn.receive_data(data) - except ProtocolError: + except h2.exceptions.ProtocolError: # this try/except block catches exceptions due to race between sending # GOAWAY and processing a response in flight. return if self._conn.data_to_send: self.transport.write(self._conn.data_to_send()) for event in events: - if isinstance(event, RequestReceived) and self._handlers.has_key('RequestReceived'): - logging.info('RequestReceived Event for stream: %d'%event.stream_id) + if isinstance(event, h2.events.RequestReceived) and self._handlers.has_key('RequestReceived'): + logging.info('RequestReceived Event for stream: %d' % event.stream_id) self._handlers['RequestReceived'](event) - elif isinstance(event, DataReceived) and self._handlers.has_key('DataReceived'): - logging.info('DataReceived Event for stream: %d'%event.stream_id) + elif isinstance(event, h2.events.DataReceived) and self._handlers.has_key('DataReceived'): + logging.info('DataReceived Event for stream: %d' % event.stream_id) self._handlers['DataReceived'](event) - elif isinstance(event, WindowUpdated) and self._handlers.has_key('WindowUpdated'): - logging.info('WindowUpdated Event for stream: %d'%event.stream_id) + elif isinstance(event, h2.events.WindowUpdated) and self._handlers.has_key('WindowUpdated'): + logging.info('WindowUpdated Event for stream: %d' % event.stream_id) self._handlers['WindowUpdated'](event) - elif isinstance(event, PingAcknowledged) and self._handlers.has_key('PingAcknowledged'): + elif isinstance(event, h2.events.PingAcknowledged) and self._handlers.has_key('PingAcknowledged'): logging.info('PingAcknowledged Event') self._handlers['PingAcknowledged'](event) self.transport.write(self._conn.data_to_send()) def on_ping_acknowledged_default(self, event): + logging.info('ping acknowledged') self._outstanding_pings -= 1 def on_data_received_default(self, event): @@ -101,7 +102,7 @@ class H2ProtocolBaseServer(Protocol): self.transport.write(self._conn.data_to_send()) def setup_send(self, data_to_send, stream_id): - logging.info('Setting up data to send for stream_id: %d'%stream_id) + logging.info('Setting up data to send for stream_id: %d' % stream_id) self._send_remaining[stream_id] = len(data_to_send) self._send_offset = 0 self._data_to_send = data_to_send @@ -116,16 +117,16 @@ class H2ProtocolBaseServer(Protocol): lfcw = self._conn.local_flow_control_window(stream_id) if lfcw == 0: break - chunk_size = min(lfcw, READ_CHUNK_SIZE) + chunk_size = min(lfcw, _READ_CHUNK_SIZE) bytes_to_send = min(chunk_size, self._send_remaining[stream_id]) - logging.info('flow_control_window = %d. sending [%d:%d] stream_id %d'% + logging.info('flow_control_window = %d. sending [%d:%d] stream_id %d' % (lfcw, self._send_offset, self._send_offset + bytes_to_send, stream_id)) data = self._data_to_send[self._send_offset : self._send_offset + bytes_to_send] try: self._conn.send_data(stream_id, data, False) - except ProtocolError: - logging.info('Stream %d is closed'%stream_id) + except h2.exceptions.ProtocolError: + logging.info('Stream %d is closed' % stream_id) break self._send_remaining[stream_id] -= bytes_to_send self._send_offset += bytes_to_send @@ -133,6 +134,7 @@ class H2ProtocolBaseServer(Protocol): self._handlers['SendDone'](stream_id) def default_ping(self): + logging.info('sending ping') self._outstanding_pings += 1 self._conn.ping(b'\x00'*8) self.transport.write(self._conn.data_to_send()) @@ -141,9 +143,11 @@ class H2ProtocolBaseServer(Protocol): if self._stream_status[stream_id]: self._stream_status[stream_id] = False self.default_send_trailer(stream_id) + else: + logging.error('Stream %d is already closed' % stream_id) def default_send_trailer(self, stream_id): - logging.info('Sending trailer for stream id %d'%stream_id) + logging.info('Sending trailer for stream id %d' % stream_id) self._conn.send_headers(stream_id, headers=[ ('grpc-status', '0') ], end_stream=True @@ -159,15 +163,14 @@ class H2ProtocolBaseServer(Protocol): return response_data def parse_received_data(self, stream_id): - recv_buffer = self._recv_buffer[stream_id] """ returns a grpc framed string of bytes containing response proto of the size asked in request """ + recv_buffer = self._recv_buffer[stream_id] grpc_msg_size = struct.unpack('i',recv_buffer[1:5][::-1])[0] - if len(recv_buffer) != GRPC_HEADER_SIZE + grpc_msg_size: - #logging.error('not enough data to decode req proto. size = %d, needed %s'%(len(recv_buffer), 5+grpc_msg_size)) + if len(recv_buffer) != _GRPC_HEADER_SIZE + grpc_msg_size: return None req_proto_str = recv_buffer[5:5+grpc_msg_size] sr = messages_pb2.SimpleRequest() sr.ParseFromString(req_proto_str) - logging.info('Parsed request for stream %d: response_size=%s'%(stream_id, sr.response_size)) + logging.info('Parsed request for stream %d: response_size=%s' % (stream_id, sr.response_size)) return sr diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index 5270ee4255e..1549d6da61c 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -3,18 +3,20 @@ """ import argparse import logging +import twisted +import twisted.internet +import twisted.internet.endpoints +import twisted.internet.reactor -from twisted.internet.protocol import Factory -from twisted.internet import endpoints, reactor import http2_base_server -import test_rst_after_header -import test_rst_after_data -import test_rst_during_data import test_goaway -import test_ping import test_max_streams +import test_ping +import test_rst_after_data +import test_rst_after_header +import test_rst_during_data -test_case_mappings = { +_TEST_CASE_MAPPING = { 'rst_after_header': test_rst_after_header.TestcaseRstStreamAfterHeader, 'rst_after_data': test_rst_after_data.TestcaseRstStreamAfterData, 'rst_during_data': test_rst_during_data.TestcaseRstStreamDuringData, @@ -23,20 +25,20 @@ test_case_mappings = { 'max_streams': test_max_streams.TestcaseSettingsMaxStreams, } -class H2Factory(Factory): +class H2Factory(twisted.internet.protocol.Factory): def __init__(self, testcase): - logging.info('In H2Factory') + logging.info('Creating H2Factory for new connection.') self._num_streams = 0 self._testcase = testcase def buildProtocol(self, addr): self._num_streams += 1 - logging.info('New Connection: %d'%self._num_streams) - if not test_case_mappings.has_key(self._testcase): - logging.error('Unknown test case: %s'%self._testcase) + logging.info('New Connection: %d' % self._num_streams) + if not _TEST_CASE_MAPPING.has_key(self._testcase): + logging.error('Unknown test case: %s' % self._testcase) assert(0) else: - t = test_case_mappings[self._testcase] + t = _TEST_CASE_MAPPING[self._testcase] if self._testcase == 'goaway': return t(self._num_streams).get_base_server() @@ -49,9 +51,9 @@ if __name__ == "__main__": parser.add_argument("test") parser.add_argument("port") args = parser.parse_args() - if args.test not in test_case_mappings.keys(): - logging.error('unknown test: %s'%args.test) + if args.test not in _TEST_CASE_MAPPING.keys(): + logging.error('unknown test: %s' % args.test) else: - endpoint = endpoints.TCP4ServerEndpoint(reactor, int(args.port), backlog=128) + endpoint = twisted.internet.endpoints.TCP4ServerEndpoint(twisted.internet.reactor, int(args.port), backlog=128) endpoint.listen(H2Factory(args.test)) - reactor.run() + twisted.internet.reactor.run() diff --git a/test/http2_test/test_goaway.py b/test/http2_test/test_goaway.py index 7a915402b2b..6deb883d4e0 100644 --- a/test/http2_test/test_goaway.py +++ b/test/http2_test/test_goaway.py @@ -1,5 +1,6 @@ import logging import time + import http2_base_server class TestcaseGoaway(object): @@ -7,7 +8,7 @@ class TestcaseGoaway(object): This test does the following: Process incoming request normally, i.e. send headers, data and trailers. Then send a GOAWAY frame with the stream id of the processed request. - It assert that the next request is made on a different TCP connection. + It checks that the next request is made on a different TCP connection. """ def __init__(self, iteration): self._base_server = http2_base_server.H2ProtocolBaseServer() @@ -22,15 +23,14 @@ class TestcaseGoaway(object): return self._base_server def on_connection_lost(self, reason): - logging.info('Disconnect received. Count %d'%self._iteration) + logging.info('Disconnect received. Count %d' % self._iteration) # _iteration == 2 => Two different connections have been used. if self._iteration == 2: self._base_server.on_connection_lost(reason) def on_send_done(self, stream_id): self._base_server.on_send_done_default(stream_id) - time.sleep(1) - logging.info('Sending GOAWAY for stream %d:'%stream_id) + logging.info('Sending GOAWAY for stream %d:' % stream_id) self._base_server._conn.close_connection(error_code=0, additional_data=None, last_stream_id=stream_id) self._base_server._stream_status[stream_id] = False @@ -42,7 +42,7 @@ class TestcaseGoaway(object): self._base_server.on_data_received_default(event) sr = self._base_server.parse_received_data(event.stream_id) if sr: - logging.info('Creating response size = %s'%sr.response_size) + logging.info('Creating response size = %s' % sr.response_size) response_data = self._base_server.default_response_data(sr.response_size) self._ready_to_send = True self._base_server.setup_send(response_data, event.stream_id) diff --git a/test/http2_test/test_max_streams.py b/test/http2_test/test_max_streams.py index deb26770c3a..10a9da3deb5 100644 --- a/test/http2_test/test_max_streams.py +++ b/test/http2_test/test_max_streams.py @@ -1,6 +1,7 @@ +import hyperframe.frame import logging + import http2_base_server -from hyperframe.frame import SettingsFrame class TestcaseSettingsMaxStreams(object): """ @@ -18,7 +19,8 @@ class TestcaseSettingsMaxStreams(object): def on_connection_made(self): logging.info('Connection Made') self._base_server._conn.initiate_connection() - self._base_server._conn.update_settings({SettingsFrame.MAX_CONCURRENT_STREAMS: 1}) + self._base_server._conn.update_settings( + {hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 1}) self._base_server.transport.setTcpNoDelay(True) self._base_server.transport.write(self._base_server._conn.data_to_send()) @@ -26,6 +28,7 @@ class TestcaseSettingsMaxStreams(object): self._base_server.on_data_received_default(event) sr = self._base_server.parse_received_data(event.stream_id) if sr: - logging.info('Creating response of size = %s'%sr.response_size) + logging.info('Creating response of size = %s' % sr.response_size) response_data = self._base_server.default_response_data(sr.response_size) self._base_server.setup_send(response_data, event.stream_id) + # TODO (makdharma): Add assertion to check number of live streams diff --git a/test/http2_test/test_ping.py b/test/http2_test/test_ping.py index 2e6dadbc079..873eb4f39ed 100644 --- a/test/http2_test/test_ping.py +++ b/test/http2_test/test_ping.py @@ -1,4 +1,5 @@ import logging + import http2_base_server class TestcasePing(object): @@ -25,13 +26,13 @@ class TestcasePing(object): self._base_server.on_data_received_default(event) sr = self._base_server.parse_received_data(event.stream_id) if sr: - logging.info('Creating response size = %s'%sr.response_size) + logging.info('Creating response size = %s' % sr.response_size) response_data = self._base_server.default_response_data(sr.response_size) self._base_server.default_ping() self._base_server.setup_send(response_data, event.stream_id) self._base_server.default_ping() def on_connection_lost(self, reason): - logging.info('Disconnect received. Ping Count %d'%self._base_server._outstanding_pings) + logging.info('Disconnect received. Ping Count %d' % self._base_server._outstanding_pings) assert(self._base_server._outstanding_pings == 0) self._base_server.on_connection_lost(reason) diff --git a/test/http2_test/test_rst_during_data.py b/test/http2_test/test_rst_during_data.py index 105c9403bb7..6767ccaf66b 100644 --- a/test/http2_test/test_rst_during_data.py +++ b/test/http2_test/test_rst_during_data.py @@ -23,7 +23,6 @@ class TestcaseRstStreamDuringData(object): response_len = len(response_data) truncated_response_data = response_data[0:response_len/2] self._base_server.setup_send(truncated_response_data, event.stream_id) - # send reset stream def on_send_done(self, stream_id): self._base_server.send_reset_stream()