mirror of https://github.com/grpc/grpc.git
Merge remote-tracking branch 'upstream/master' into revert-9063-revert-8951-revert-8949-revert-8922-slice_cleanup
Mark D. Roth
8 years ago
commit
1311e3448c
52 changed files with 638 additions and 133 deletions
@ -0,0 +1,29 @@ |
||||
# C++ Performance Notes |
||||
|
||||
## Streaming write buffering |
||||
|
||||
Generally, each write operation (Write(), WritesDone()) implies a syscall. |
||||
gRPC will try to batch together separate write operations from different |
||||
threads, but currently cannot automatically infer batching in a single stream. |
||||
|
||||
If message k+1 in a stream does not rely on responses from message k, it's |
||||
possible to enable write batching by passing a WriteOptions argument to Write |
||||
with the buffer_hint set: |
||||
|
||||
~~~{.cpp} |
||||
stream_writer->Write(message, WriteOptions().set_buffer_hint()); |
||||
~~~ |
||||
|
||||
The write will be buffered until one of the following is true: |
||||
- the per-stream buffer is filled (controllable with the channel argument |
||||
GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE) - this prevents infinite buffering leading |
||||
to OOM |
||||
- a subsequent Write without buffer_hint set is posted |
||||
- the call is finished for writing (WritesDone() called on the client, |
||||
or Finish() called on an async server stream, or the service handler returns |
||||
for a sync server stream) |
||||
|
||||
## Completion Queues and Threading in the Async API |
||||
|
||||
Right now, the best performance trade-off is having numcpu's threads and one |
||||
completion queue per thread. |
||||
@ -0,0 +1,5 @@ |
||||
Optional plugins for gRPC Core: Modules in this directory extend gRPC Core in |
||||
useful ways. All optional code belongs here. |
||||
|
||||
NOTE: The movement of code between lib and ext is an ongoing effort, so this |
||||
directory currently contains too much of the core library. |
||||
@ -0,0 +1,4 @@ |
||||
# Resolver |
||||
|
||||
Implementations of various name resolution schemes. |
||||
See the [naming spec](/doc/naming.md). |
||||
@ -0,0 +1 @@ |
||||
Transports for gRPC |
||||
@ -0,0 +1 @@ |
||||
CHTTP2 - gRPC's implementation of a HTTP2 based transport |
||||
@ -0,0 +1,6 @@ |
||||
Required elements of gRPC Core: Each module in this directory is required to |
||||
build gRPC. If it's possible to envisage a configuration where code is not |
||||
required, then that code belongs in ext/ instead. |
||||
|
||||
NOTE: The movement of code between lib and ext is an ongoing effort, so this |
||||
directory currently contains too much of the core library. |
||||
@ -0,0 +1,4 @@ |
||||
# Channel |
||||
|
||||
Provides channel/call stack implementation, and implementation of common filters |
||||
for that implementation. |
||||
@ -0,0 +1,6 @@ |
||||
# iomgr |
||||
|
||||
Platform abstractions for I/O (mostly network). |
||||
|
||||
Provides abstractions over TCP/UDP I/O, file loading, polling, and concurrency |
||||
management for various operating systems. |
||||
@ -0,0 +1,4 @@ |
||||
# Surface |
||||
|
||||
Surface provides the bulk of the gRPC Core public API, and translates it into |
||||
calls against core components. |
||||
@ -0,0 +1,7 @@ |
||||
# Transport |
||||
|
||||
Common implementation details for gRPC Transports. |
||||
|
||||
Transports multiplex messages across some single connection. In ext/ there are |
||||
implementations atop [a custom http2 implementation](/src/core/ext/transport/chttp2/README.md) |
||||
and atop [cronet](/src/core/ext/transport/cronet/README.md). |
||||
@ -0,0 +1,2 @@ |
||||
# Transport Security Interface |
||||
An abstraction library over crypto and auth modules (typically OpenSSL) |
||||
@ -0,0 +1,247 @@ |
||||
# Copyright 2016, Google Inc. |
||||
# All rights reserved. |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are |
||||
# met: |
||||
# |
||||
# * Redistributions of source code must retain the above copyright |
||||
# notice, this list of conditions and the following disclaimer. |
||||
# * Redistributions in binary form must reproduce the above |
||||
# copyright notice, this list of conditions and the following disclaimer |
||||
# in the documentation and/or other materials provided with the |
||||
# distribution. |
||||
# * Neither the name of Google Inc. nor the names of its |
||||
# contributors may be used to endorse or promote products derived from |
||||
# this software without specific prior written permission. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
import itertools |
||||
import threading |
||||
import unittest |
||||
from concurrent import futures |
||||
|
||||
import grpc |
||||
from grpc.framework.foundation import logging_pool |
||||
|
||||
from tests.unit.framework.common import test_constants |
||||
from tests.unit.framework.common import test_control |
||||
|
||||
_SERIALIZE_REQUEST = lambda bytestring: bytestring * 2 |
||||
_DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:] |
||||
_SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3 |
||||
_DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3] |
||||
|
||||
_UNARY_UNARY = '/test/UnaryUnary' |
||||
_UNARY_STREAM = '/test/UnaryStream' |
||||
_STREAM_UNARY = '/test/StreamUnary' |
||||
_STREAM_STREAM = '/test/StreamStream' |
||||
|
||||
|
||||
class _Callback(object): |
||||
def __init__(self): |
||||
self._condition = threading.Condition() |
||||
self._value = None |
||||
self._called = False |
||||
|
||||
def __call__(self, value): |
||||
with self._condition: |
||||
self._value = value |
||||
self._called = True |
||||
self._condition.notify_all() |
||||
|
||||
def value(self): |
||||
with self._condition: |
||||
while not self._called: |
||||
self._condition.wait() |
||||
return self._value |
||||
|
||||
|
||||
class _Handler(object): |
||||
def __init__(self, control): |
||||
self._control = control |
||||
|
||||
def handle_unary_unary(self, request, servicer_context): |
||||
self._control.control() |
||||
if servicer_context is not None: |
||||
servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) |
||||
return request |
||||
|
||||
def handle_unary_stream(self, request, servicer_context): |
||||
for _ in range(test_constants.STREAM_LENGTH): |
||||
self._control.control() |
||||
yield request |
||||
self._control.control() |
||||
if servicer_context is not None: |
||||
servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) |
||||
|
||||
def handle_stream_unary(self, request_iterator, servicer_context): |
||||
if servicer_context is not None: |
||||
servicer_context.invocation_metadata() |
||||
self._control.control() |
||||
response_elements = [] |
||||
for request in request_iterator: |
||||
self._control.control() |
||||
response_elements.append(request) |
||||
self._control.control() |
||||
if servicer_context is not None: |
||||
servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) |
||||
return b''.join(response_elements) |
||||
|
||||
def handle_stream_stream(self, request_iterator, servicer_context): |
||||
self._control.control() |
||||
if servicer_context is not None: |
||||
servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) |
||||
for request in request_iterator: |
||||
self._control.control() |
||||
yield request |
||||
self._control.control() |
||||
|
||||
|
||||
class _MethodHandler(grpc.RpcMethodHandler): |
||||
def __init__( |
||||
self, request_streaming, response_streaming, request_deserializer, |
||||
response_serializer, unary_unary, unary_stream, stream_unary, |
||||
stream_stream): |
||||
self.request_streaming = request_streaming |
||||
self.response_streaming = response_streaming |
||||
self.request_deserializer = request_deserializer |
||||
self.response_serializer = response_serializer |
||||
self.unary_unary = unary_unary |
||||
self.unary_stream = unary_stream |
||||
self.stream_unary = stream_unary |
||||
self.stream_stream = stream_stream |
||||
|
||||
|
||||
class _GenericHandler(grpc.GenericRpcHandler): |
||||
def __init__(self, handler): |
||||
self._handler = handler |
||||
|
||||
def service(self, handler_call_details): |
||||
if handler_call_details.method == _UNARY_UNARY: |
||||
return _MethodHandler( |
||||
False, False, None, None, self._handler.handle_unary_unary, None, |
||||
None, None) |
||||
elif handler_call_details.method == _UNARY_STREAM: |
||||
return _MethodHandler( |
||||
False, True, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, |
||||
self._handler.handle_unary_stream, None, None) |
||||
elif handler_call_details.method == _STREAM_UNARY: |
||||
return _MethodHandler( |
||||
True, False, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, None, |
||||
self._handler.handle_stream_unary, None) |
||||
elif handler_call_details.method == _STREAM_STREAM: |
||||
return _MethodHandler( |
||||
True, True, None, None, None, None, None, |
||||
self._handler.handle_stream_stream) |
||||
else: |
||||
return None |
||||
|
||||
|
||||
class FailAfterFewIterationsCounter(object): |
||||
def __init__(self, high, bytestring): |
||||
self._current = 0 |
||||
self._high = high |
||||
self._bytestring = bytestring |
||||
|
||||
def __iter__(self): |
||||
return self |
||||
|
||||
def __next__(self): |
||||
if self._current >= self._high: |
||||
raise Exception("This is a deliberate failure in a unit test.") |
||||
else: |
||||
self._current += 1 |
||||
return self._bytestring |
||||
|
||||
|
||||
def _unary_unary_multi_callable(channel): |
||||
return channel.unary_unary(_UNARY_UNARY) |
||||
|
||||
|
||||
def _unary_stream_multi_callable(channel): |
||||
return channel.unary_stream( |
||||
_UNARY_STREAM, |
||||
request_serializer=_SERIALIZE_REQUEST, |
||||
response_deserializer=_DESERIALIZE_RESPONSE) |
||||
|
||||
|
||||
def _stream_unary_multi_callable(channel): |
||||
return channel.stream_unary( |
||||
_STREAM_UNARY, |
||||
request_serializer=_SERIALIZE_REQUEST, |
||||
response_deserializer=_DESERIALIZE_RESPONSE) |
||||
|
||||
|
||||
def _stream_stream_multi_callable(channel): |
||||
return channel.stream_stream(_STREAM_STREAM) |
||||
|
||||
|
||||
class InvocationDefectsTest(unittest.TestCase): |
||||
def setUp(self): |
||||
self._control = test_control.PauseFailControl() |
||||
self._handler = _Handler(self._control) |
||||
self._server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY) |
||||
|
||||
self._server = grpc.server(self._server_pool) |
||||
port = self._server.add_insecure_port('[::]:0') |
||||
self._server.add_generic_rpc_handlers((_GenericHandler(self._handler),)) |
||||
self._server.start() |
||||
|
||||
self._channel = grpc.insecure_channel('localhost:%d' % port) |
||||
|
||||
def tearDown(self): |
||||
self._server.stop(0) |
||||
|
||||
def testIterableStreamRequestBlockingUnaryResponse(self): |
||||
requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)] |
||||
multi_callable = _stream_unary_multi_callable(self._channel) |
||||
|
||||
with self.assertRaises(grpc.RpcError): |
||||
response = multi_callable( |
||||
requests, |
||||
metadata=(('test', 'IterableStreamRequestBlockingUnaryResponse'),)) |
||||
|
||||
def testIterableStreamRequestFutureUnaryResponse(self): |
||||
requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)] |
||||
multi_callable = _stream_unary_multi_callable(self._channel) |
||||
response_future = multi_callable.future( |
||||
requests, |
||||
metadata=( |
||||
('test', 'IterableStreamRequestFutureUnaryResponse'),)) |
||||
|
||||
with self.assertRaises(grpc.RpcError): |
||||
response = response_future.result() |
||||
|
||||
def testIterableStreamRequestStreamResponse(self): |
||||
requests = [b'\x77\x58' for _ in range(test_constants.STREAM_LENGTH)] |
||||
multi_callable = _stream_stream_multi_callable(self._channel) |
||||
response_iterator = multi_callable( |
||||
requests, |
||||
metadata=(('test', 'IterableStreamRequestStreamResponse'),)) |
||||
|
||||
with self.assertRaises(grpc.RpcError): |
||||
next(response_iterator) |
||||
|
||||
def testIteratorStreamRequestStreamResponse(self): |
||||
requests_iterator = FailAfterFewIterationsCounter( |
||||
test_constants.STREAM_LENGTH // 2, b'\x07\x08') |
||||
multi_callable = _stream_stream_multi_callable(self._channel) |
||||
response_iterator = multi_callable( |
||||
requests_iterator, |
||||
metadata=(('test', 'IteratorStreamRequestStreamResponse'),)) |
||||
|
||||
with self.assertRaises(grpc.RpcError): |
||||
for _ in range(test_constants.STREAM_LENGTH // 2 + 1): |
||||
next(response_iterator) |
||||
@ -1,39 +0,0 @@ |
||||
@rem Copyright 2016, Google Inc. |
||||
@rem All rights reserved. |
||||
@rem |
||||
@rem Redistribution and use in source and binary forms, with or without |
||||
@rem modification, are permitted provided that the following conditions are |
||||
@rem met: |
||||
@rem |
||||
@rem * Redistributions of source code must retain the above copyright |
||||
@rem notice, this list of conditions and the following disclaimer. |
||||
@rem * Redistributions in binary form must reproduce the above |
||||
@rem copyright notice, this list of conditions and the following disclaimer |
||||
@rem in the documentation and/or other materials provided with the |
||||
@rem distribution. |
||||
@rem * Neither the name of Google Inc. nor the names of its |
||||
@rem contributors may be used to endorse or promote products derived from |
||||
@rem this software without specific prior written permission. |
||||
@rem |
||||
@rem THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
@rem "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
@rem LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
@rem A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
@rem OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
@rem SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
@rem LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
@rem DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
@rem THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
@rem (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
@rem OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
@rem Convenience wrapper that runs specified gRPC target using msbuild |
||||
@rem Usage: build_vs2010.bat TARGET_NAME |
||||
|
||||
setlocal |
||||
@rem Set VS variables (uses Visual Studio 2010) |
||||
@call "%VS100COMNTOOLS%\..\..\vc\vcvarsall.bat" x86 |
||||
|
||||
msbuild %* |
||||
exit /b %ERRORLEVEL% |
||||
endlocal |
||||
Loading…
Reference in new issue