Merge github.com:grpc/grpc into test_affine

pull/6645/head
Craig Tiller 9 years ago
commit 9766f4f424
  1. 7
      include/grpc++/impl/codegen/call.h
  2. 11
      include/grpc++/impl/codegen/proto_utils.h
  3. 12
      src/core/lib/channel/compress_filter.c
  4. 2
      src/core/lib/channel/compress_filter.h
  5. 17
      src/core/lib/iomgr/exec_ctx.c
  6. 29
      src/core/lib/iomgr/exec_ctx.h
  7. 2
      src/core/lib/support/string_util_win32.c
  8. 19
      src/core/lib/surface/byte_buffer_reader.c
  9. 32
      src/core/lib/surface/call.c
  10. 2
      src/core/lib/surface/init.c
  11. 10
      src/php/ext/grpc/call.c
  12. 4
      src/php/ext/grpc/call.h
  13. 12
      src/php/ext/grpc/call_credentials.c
  14. 4
      src/php/ext/grpc/channel.c
  15. 2
      src/php/ext/grpc/channel.h
  16. 8
      src/php/ext/grpc/channel_credentials.c
  17. 8
      src/php/ext/grpc/server.c
  18. 4
      src/php/ext/grpc/server_credentials.c
  19. 14
      src/php/ext/grpc/timeval.c
  20. 2
      src/php/ext/grpc/timeval.h
  21. 20
      src/python/grpcio/grpc/_adapter/_low.py
  22. 4
      src/python/grpcio/grpc/_adapter/_types.py
  23. 38
      src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
  24. 4
      src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi
  25. 74
      src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
  26. 9
      src/python/grpcio/grpc/_cython/imports.generated.h
  27. 2
      src/python/grpcio/grpc/beta/interfaces.py
  28. 28
      src/python/grpcio/tests/health_check/__init__.py
  29. 75
      src/python/grpcio/tests/health_check/_health_servicer_test.py
  30. 5
      src/python/grpcio/tests/tests.json
  31. 2
      src/python/grpcio/tests/unit/_cython/_channel_test.py
  32. 58
      src/python/grpcio/tests/unit/_cython/cygrpc_test.py
  33. 9
      src/python/grpcio/tests/unit/framework/common/test_constants.py
  34. 10
      src/python/grpcio/tests/unit/framework/interfaces/face/_blocking_invocation_inline_service.py
  35. 8
      src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
  36. 5
      src/python/grpcio_health_checking/.gitignore
  37. 5
      src/python/grpcio_health_checking/MANIFEST.in
  38. 49
      src/python/grpcio_health_checking/grpc/health/v1/health.proto
  39. 129
      src/python/grpcio_health_checking/grpc/health/v1/health.py
  40. 0
      src/python/grpcio_health_checking/grpc_health/__init__.py
  41. 0
      src/python/grpcio_health_checking/grpc_health/health/__init__.py
  42. 0
      src/python/grpcio_health_checking/grpc_health/health/v1/__init__.py
  43. 66
      src/python/grpcio_health_checking/grpc_health/health/v1/health.py
  44. 30
      src/python/grpcio_health_checking/health_commands.py
  45. 15
      src/python/grpcio_health_checking/setup.py
  46. 9
      templates/src/python/grpcio/grpc/_cython/imports.generated.h.template
  47. 5
      test/core/end2end/dualstack_socket_test.c
  48. 74
      test/core/end2end/fuzzers/api_fuzzer.c
  49. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/001ea98069c10f808c281da9bbdd84cc05c3bad1
  50. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/01f52e31dfffdab89d83acd39925c3dd81baa76f
  51. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/02c3cf8d52fbc43f89b5f516a17cea23b68fc8d5
  52. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/032744b59cafd3320cc932ad39926a9bc92f589e
  53. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0385c7b41263419e25a4342fbfc44fbd65eb2ed5
  54. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/04d93c9df413717f71abd091592b5238afb799e8
  55. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/064d50aee4416ccf32f4e4fe7b770b7802265ffe
  56. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/066e7fcb68e83b432c414f63f6de73e5f5099e49
  57. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/06c714e289673cf982ce2ac0670707a15f2ac5ea
  58. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/071247b8fddda8aa520d9142c89039fbf8bf6cee
  59. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/07cb3b9baca1bbcce2e199e551073ba2fdd4e05c
  60. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/07fa2b6ed650d436f423adcccfcbe63ce6253de0
  61. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/081e3248dfca2b32837c4738daee3a4698caaf15
  62. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/088bf259e854abd9508d91b23983737f8e9e242c
  63. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0976de1461fb037c6987d77d088416440b524dde
  64. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0adaf5f559e1fb9cd8cd5b29911e13bca315c606
  65. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0af5adf68560b3a7036ad23af62e4f9749eca690
  66. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0becc6ede499ddc452fd4e6c3c0413a1107a8373
  67. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0c531e03e56a5cf48bdd531a8c11a19e4a3b0aeb
  68. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0c65733bc09e8527347e20f5c876c5b64570d423
  69. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0c7b763d22885462527123656fa17af7520fc55d
  70. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0d604693a9d3e76f54d28a26142abd729b0a9acd
  71. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0d993b34021ec088f1aa3e5acdd98089b4104b07
  72. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0def53b5575cc6ab2fbbd17e2bc6a24de9656f84
  73. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0e2ddbe92c08eb9ad3cbee1d0db2264baaca12df
  74. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0ea509d249ae28faba8980aacb972c7ea28d3fd5
  75. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0f16eeeecdebcb59022bda5a0972d1b3429648fd
  76. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/0f81830560dbb9c6d3889b5d581b918c6cade65f
  77. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/105d9784648fe2d6c22fbefa69c9a26fff1c6481
  78. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/11153bfeee3cdede86a52151dbb939c3ffee48ed
  79. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/113c1d1bac15d550124f1ffb9012c32755adf27f
  80. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/11759723c597e6806f8873e5062d31516cdb97ea
  81. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/12abf5dcf2aba770f7b94ce5d96d7a8565a9aa19
  82. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/12b904b97ed234fa45073b4e346ebe3211558528
  83. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/12c00ed8945bdae03f03142cb964a47ea0c5786e
  84. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/12f977ee18a7499d18a503a47e71b4f241052640
  85. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/145acf7c03a0bc6c4a40d710ba5813b9f28efe2a
  86. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/14ccbe1d9d7302d642e51ede3d4d846e85310fc2
  87. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1586adc8c21b5796ba52203379faeb5f251f5c1d
  88. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1608a688768bdecdb205a455401ce5d9a1424a22
  89. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/162b4ec7cf39df091898e01057b2fa39605b34bb
  90. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/16858f1f9db0e248a15ce09d9848612de1f4bba6
  91. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/16e681f1867a1ac5612e1a88fddaed0bcb4521e7
  92. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/16ebac3f7cea2b46f660ec6a5ef3401c3e17a2e9
  93. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/17ec0503991dc248d2b188edfa3d28573a1c2154
  94. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/17fb35db0b73c331a66120dbc491300b2d1665e0
  95. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/190c4ca0cf29c99bc987d2792c7f62e1007c0245
  96. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1949f4a75f7d501d5279a01f58a444640379bd78
  97. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1972f535ae202777efdd15a09138efc37e07ac01
  98. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/198e691a9dabd23ed5c156f3a6e2c06a4379c15b
  99. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1a16a4b32cb0cb3a759ec20edf332cdfc5d1717e
  100. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/1af0744fe0ccad11d6df023803ab699e1464c8da
  101. Some files were not shown because too many files have changed in this diff Show More

@ -329,8 +329,11 @@ class CallOpGenericRecvMessage {
template <class R>
void RecvMessage(R* message) {
deserialize_.reset(
new CallOpGenericRecvMessageHelper::DeserializeFuncType<R>(message));
// Use an explicit base class pointer to avoid resolution error in the
// following unique_ptr::reset for some old implementations.
CallOpGenericRecvMessageHelper::DeserializeFunc* func =
new CallOpGenericRecvMessageHelper::DeserializeFuncType<R>(message);
deserialize_.reset(func);
}
bool got_message;

@ -49,7 +49,7 @@ namespace grpc {
extern CoreCodegenInterface* g_core_codegen_interface;
namespace {
namespace internal {
const int kGrpcBufferWriterMaxBufferLength = 8192;
@ -166,7 +166,7 @@ class GrpcBufferReader GRPC_FINAL
grpc_byte_buffer_reader reader_;
gpr_slice slice_;
};
} // namespace
} // namespace internal
template <class T>
class SerializationTraits<T, typename std::enable_if<std::is_base_of<
@ -176,7 +176,7 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
grpc_byte_buffer** bp, bool* own_buffer) {
*own_buffer = true;
int byte_size = msg.ByteSize();
if (byte_size <= kGrpcBufferWriterMaxBufferLength) {
if (byte_size <= internal::kGrpcBufferWriterMaxBufferLength) {
gpr_slice slice = g_core_codegen_interface->gpr_slice_malloc(byte_size);
GPR_CODEGEN_ASSERT(
GPR_SLICE_END_PTR(slice) ==
@ -185,7 +185,8 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
g_core_codegen_interface->gpr_slice_unref(slice);
return g_core_codegen_interface->ok();
} else {
GrpcBufferWriter writer(bp, kGrpcBufferWriterMaxBufferLength);
internal::GrpcBufferWriter writer(
bp, internal::kGrpcBufferWriterMaxBufferLength);
return msg.SerializeToZeroCopyStream(&writer)
? g_core_codegen_interface->ok()
: Status(StatusCode::INTERNAL, "Failed to serialize message");
@ -200,7 +201,7 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
}
Status result = g_core_codegen_interface->ok();
{
GrpcBufferReader reader(buffer);
internal::GrpcBufferReader reader(buffer);
::grpc::protobuf::io::CodedInputStream decoder(&reader);
if (max_message_size > 0) {
decoder.SetTotalBytesLimit(max_message_size, max_message_size);

@ -47,7 +47,7 @@
#include "src/core/lib/support/string.h"
#include "src/core/lib/transport/static_metadata.h"
int grpc_compress_filter_trace = 0;
int grpc_compression_trace = 0;
typedef struct call_data {
gpr_slice_buffer slices; /**< Buffers up input slices to be compressed */
@ -171,7 +171,7 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx,
did_compress =
grpc_msg_compress(calld->compression_algorithm, &calld->slices, &tmp);
if (did_compress) {
if (grpc_compress_filter_trace) {
if (grpc_compression_trace) {
char *algo_name;
const size_t before_size = calld->slices.length;
const size_t after_size = tmp.length;
@ -185,12 +185,14 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx,
gpr_slice_buffer_swap(&calld->slices, &tmp);
calld->send_flags |= GRPC_WRITE_INTERNAL_COMPRESS;
} else {
if (grpc_compress_filter_trace) {
if (grpc_compression_trace) {
char *algo_name;
GPR_ASSERT(grpc_compression_algorithm_name(calld->compression_algorithm,
&algo_name));
gpr_log(GPR_DEBUG, "Algorithm '%s' enabled but decided not to compress.",
algo_name);
gpr_log(
GPR_DEBUG,
"Algorithm '%s' enabled but decided not to compress. Input size: %d",
algo_name, calld->slices.length);
}
}

@ -38,7 +38,7 @@
#define GRPC_COMPRESS_REQUEST_ALGORITHM_KEY "grpc-internal-encoding-request"
extern int grpc_compress_filter_trace;
extern int grpc_compression_trace;
/** Compression filter for outgoing data.
*

@ -39,6 +39,22 @@
#include "src/core/lib/profiling/timers.h"
bool grpc_exec_ctx_ready_to_finish(grpc_exec_ctx *exec_ctx) {
if (!exec_ctx->cached_ready_to_finish) {
exec_ctx->cached_ready_to_finish = exec_ctx->check_ready_to_finish(
exec_ctx, exec_ctx->check_ready_to_finish_arg);
}
return exec_ctx->cached_ready_to_finish;
}
bool grpc_never_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored) {
return false;
}
bool grpc_always_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored) {
return true;
}
#ifndef GRPC_EXECUTION_CONTEXT_SANITIZER
bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
bool did_something = 0;
@ -61,6 +77,7 @@ bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
}
void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx) {
exec_ctx->cached_ready_to_finish = true;
grpc_exec_ctx_flush(exec_ctx);
}

@ -53,6 +53,9 @@ typedef struct grpc_workqueue grpc_workqueue;
* - track a list of work that needs to be delayed until the top of the
* call stack (this provides a convenient mechanism to run callbacks
* without worrying about locking issues)
* - provide a decision maker (via grpc_exec_ctx_ready_to_finish) that provides
* signal as to whether a borrowed thread should continue to do work or
* should actively try to finish up and get this thread back to its owner
*
* CONVENTIONS:
* Instance of this must ALWAYS be constructed on the stack, never
@ -63,18 +66,26 @@ typedef struct grpc_workqueue grpc_workqueue;
*/
struct grpc_exec_ctx {
grpc_closure_list closure_list;
bool cached_ready_to_finish;
void *check_ready_to_finish_arg;
bool (*check_ready_to_finish)(grpc_exec_ctx *exec_ctx, void *arg);
};
#define GRPC_EXEC_CTX_INIT \
{ GRPC_CLOSURE_LIST_INIT }
#define GRPC_EXEC_CTX_INIT_WITH_FINISH_CHECK(finish_check, finish_check_arg) \
{ GRPC_CLOSURE_LIST_INIT, false, finish_check_arg, finish_check }
#else
struct grpc_exec_ctx {
int unused;
bool cached_ready_to_finish;
void *check_ready_to_finish_arg;
bool (*check_ready_to_finish)(grpc_exec_ctx *exec_ctx, void *arg);
};
#define GRPC_EXEC_CTX_INIT \
{ 0 }
#define GRPC_EXEC_CTX_INIT_WITH_FINISH_CHECK(finish_check, finish_check_arg) \
{ false, finish_check_arg, finish_check }
#endif
#define GRPC_EXEC_CTX_INIT \
GRPC_EXEC_CTX_INIT_WITH_FINISH_CHECK(grpc_never_ready_to_finish, NULL)
/** Flush any work that has been enqueued onto this grpc_exec_ctx.
* Caller must guarantee that no interfering locks are held.
* Returns true if work was performed, false otherwise. */
@ -86,6 +97,14 @@ void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx);
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
bool success,
grpc_workqueue *offload_target_or_null);
/** Returns true if we'd like to leave this execution context as soon as
possible: useful for deciding whether to do something more or not depending
on outside context */
bool grpc_exec_ctx_ready_to_finish(grpc_exec_ctx *exec_ctx);
/** A finish check that is never ready to finish */
bool grpc_never_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored);
/** A finish check that is always ready to finish */
bool grpc_always_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored);
/** Add a list of closures to be executed at the next flush/finish point.
* Leaves \a list empty. */
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,

@ -83,7 +83,7 @@ char *gpr_format_message(int messageid) {
DWORD status = FormatMessage(
FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
FORMAT_MESSAGE_IGNORE_INSERTS,
NULL, (DWORD)messageid, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
NULL, (DWORD)messageid, MAKELANGID(LANG_ENGLISH, SUBLANG_DEFAULT),
(LPTSTR)(&tmessage), 0, NULL);
if (status == 0) return gpr_strdup("Unable to retrieve error string");
message = gpr_tchar_to_char(tmessage);

@ -62,12 +62,19 @@ void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
case GRPC_BB_RAW:
gpr_slice_buffer_init(&decompressed_slices_buffer);
if (is_compressed(reader->buffer_in)) {
grpc_msg_decompress(reader->buffer_in->data.raw.compression,
&reader->buffer_in->data.raw.slice_buffer,
&decompressed_slices_buffer);
reader->buffer_out =
grpc_raw_byte_buffer_create(decompressed_slices_buffer.slices,
decompressed_slices_buffer.count);
if (grpc_msg_decompress(reader->buffer_in->data.raw.compression,
&reader->buffer_in->data.raw.slice_buffer,
&decompressed_slices_buffer) == 0) {
gpr_log(GPR_ERROR,
"Unexpected error decompressing data for algorithm with enum "
"value '%d'. Reading data as if it were uncompressed.",
reader->buffer_in->data.raw.compression);
reader->buffer_out = reader->buffer_in;
} else { /* all fine */
reader->buffer_out =
grpc_raw_byte_buffer_create(decompressed_slices_buffer.slices,
decompressed_slices_buffer.count);
}
gpr_slice_buffer_destroy(&decompressed_slices_buffer);
} else { /* not compressed, use the input buffer as output */
reader->buffer_out = reader->buffer_in;

@ -261,6 +261,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call,
call->channel = channel;
call->cq = cq;
call->parent = parent_call;
/* Always support no compression */
GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_COMPRESS_NONE);
call->is_client = server_transport_data == NULL;
if (call->is_client) {
GPR_ASSERT(add_initial_metadata_count < MAX_SEND_EXTRA_METADATA_COUNT);
@ -408,6 +410,7 @@ static void set_status_code(grpc_call *call, status_source source,
static void set_compression_algorithm(grpc_call *call,
grpc_compression_algorithm algo) {
GPR_ASSERT(algo < GRPC_COMPRESS_ALGORITHMS_COUNT);
call->compression_algorithm = algo;
}
@ -828,12 +831,16 @@ static uint32_t decode_status(grpc_mdelem *md) {
return status;
}
static uint32_t decode_compression(grpc_mdelem *md) {
static grpc_compression_algorithm decode_compression(grpc_mdelem *md) {
grpc_compression_algorithm algorithm =
grpc_compression_algorithm_from_mdstr(md->value);
if (algorithm == GRPC_COMPRESS_ALGORITHMS_COUNT) {
const char *md_c_str = grpc_mdstr_as_c_string(md->value);
gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'", md_c_str);
gpr_log(GPR_ERROR,
"Invalid incoming compression algorithm: '%s'. Interpreting "
"incoming data as uncompressed.",
md_c_str);
return GRPC_COMPRESS_NONE;
}
return algorithm;
}
@ -1087,6 +1094,24 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
&call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
grpc_metadata_batch_filter(md, recv_initial_filter, call);
/* make sure the received grpc-encoding is amongst the ones listed in
* grpc-accept-encoding */
GPR_ASSERT(call->encodings_accepted_by_peer != 0);
if (!GPR_BITGET(call->encodings_accepted_by_peer,
call->compression_algorithm)) {
extern int grpc_compression_trace;
if (grpc_compression_trace) {
char *algo_name;
grpc_compression_algorithm_name(call->compression_algorithm,
&algo_name);
gpr_log(GPR_ERROR,
"Compression algorithm (grpc-encoding = '%s') not present in "
"the bitset of accepted encodings (grpc-accept-encodings: "
"'0x%x')",
algo_name, call->encodings_accepted_by_peer);
}
}
if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) !=
0 &&
!call->is_client) {
@ -1474,7 +1499,8 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
grpc_call_error err;
GRPC_API_TRACE(
"grpc_call_start_batch(call=%p, ops=%p, nops=%lu, tag=%p, reserved=%p)",
"grpc_call_start_batch(call=%p, ops=%p, nops=%lu, tag=%p, "
"reserved=%p)",
5, (call, ops, (unsigned long)nops, tag, reserved));
if (reserved != NULL) {

@ -164,7 +164,7 @@ void grpc_init(void) {
grpc_register_tracer("channel_stack_builder",
&grpc_trace_channel_stack_builder);
grpc_register_tracer("http1", &grpc_http1_trace);
grpc_register_tracer("compression", &grpc_compress_filter_trace);
grpc_register_tracer("compression", &grpc_compression_trace);
grpc_security_pre_init();
grpc_iomgr_init();
grpc_executor_init();

@ -89,7 +89,7 @@ zend_object_value create_wrapped_grpc_call(zend_class_entry *class_type
/* Wraps a grpc_call struct in a PHP object. Owned indicates whether the struct
should be destroyed at the end of the object's lifecycle */
zval *grpc_php_wrap_call(grpc_call *wrapped, bool owned) {
zval *grpc_php_wrap_call(grpc_call *wrapped, bool owned TSRMLS_DC) {
zval *call_object;
MAKE_STD_ZVAL(call_object);
object_init_ex(call_object, grpc_ce_call);
@ -102,7 +102,7 @@ zval *grpc_php_wrap_call(grpc_call *wrapped, bool owned) {
/* Creates and returns a PHP array object with the data in a
* grpc_metadata_array. Returns NULL on failure */
zval *grpc_parse_metadata_array(grpc_metadata_array *metadata_array) {
zval *grpc_parse_metadata_array(grpc_metadata_array *metadata_array TSRMLS_DC) {
int count = metadata_array->count;
grpc_metadata *elements = metadata_array->metadata;
int i;
@ -127,7 +127,7 @@ zval *grpc_parse_metadata_array(grpc_metadata_array *metadata_array) {
if (zend_hash_find(array_hash, str_key, key_len, (void **)data) ==
SUCCESS) {
if (Z_TYPE_P(*data) != IS_ARRAY) {
zend_throw_exception(zend_exception_get_default(),
zend_throw_exception(zend_exception_get_default(TSRMLS_C),
"Metadata hash somehow contains wrong types.",
1 TSRMLS_CC);
efree(str_key);
@ -454,7 +454,7 @@ PHP_METHOD(Call, startBatch) {
add_property_bool(result, "send_status", true);
break;
case GRPC_OP_RECV_INITIAL_METADATA:
array = grpc_parse_metadata_array(&recv_metadata);
array = grpc_parse_metadata_array(&recv_metadata TSRMLS_CC);
add_property_zval(result, "metadata", array);
Z_DELREF_P(array);
break;
@ -470,7 +470,7 @@ PHP_METHOD(Call, startBatch) {
case GRPC_OP_RECV_STATUS_ON_CLIENT:
MAKE_STD_ZVAL(recv_status);
object_init(recv_status);
array = grpc_parse_metadata_array(&recv_trailing_metadata);
array = grpc_parse_metadata_array(&recv_trailing_metadata TSRMLS_CC);
add_property_zval(recv_status, "metadata", array);
Z_DELREF_P(array);
add_property_long(recv_status, "code", status);

@ -60,11 +60,11 @@ typedef struct wrapped_grpc_call {
void grpc_init_call(TSRMLS_D);
/* Creates a Call object that wraps the given grpc_call struct */
zval *grpc_php_wrap_call(grpc_call *wrapped, bool owned);
zval *grpc_php_wrap_call(grpc_call *wrapped, bool owned TSRMLS_DC);
/* Creates and returns a PHP associative array of metadata from a C array of
* call metadata */
zval *grpc_parse_metadata_array(grpc_metadata_array *metadata_array);
zval *grpc_parse_metadata_array(grpc_metadata_array *metadata_array TSRMLS_DC);
/* Populates a grpc_metadata_array with the data in a PHP array object.
Returns true on success and false on failure */

@ -83,7 +83,7 @@ zend_object_value create_wrapped_grpc_call_credentials(
return retval;
}
zval *grpc_php_wrap_call_credentials(grpc_call_credentials *wrapped) {
zval *grpc_php_wrap_call_credentials(grpc_call_credentials *wrapped TSRMLS_DC) {
zval *credentials_object;
MAKE_STD_ZVAL(credentials_object);
object_init_ex(credentials_object, grpc_ce_call_credentials);
@ -122,7 +122,7 @@ PHP_METHOD(CallCredentials, createComposite) {
grpc_call_credentials *creds =
grpc_composite_call_credentials_create(cred1->wrapped, cred2->wrapped,
NULL);
zval *creds_object = grpc_php_wrap_call_credentials(creds);
zval *creds_object = grpc_php_wrap_call_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}
@ -141,7 +141,7 @@ PHP_METHOD(CallCredentials, createFromPlugin) {
memset(fci_cache, 0, sizeof(zend_fcall_info_cache));
/* "f" == 1 function */
if (zend_parse_parameters(ZEND_NUM_ARGS(), "f", fci,
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "f", fci,
fci_cache,
fci->params,
fci->param_count) == FAILURE) {
@ -167,7 +167,7 @@ PHP_METHOD(CallCredentials, createFromPlugin) {
grpc_call_credentials *creds = grpc_metadata_credentials_create_from_plugin(
plugin, NULL);
zval *creds_object = grpc_php_wrap_call_credentials(creds);
zval *creds_object = grpc_php_wrap_call_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}
@ -175,6 +175,8 @@ PHP_METHOD(CallCredentials, createFromPlugin) {
void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context,
grpc_credentials_plugin_metadata_cb cb,
void *user_data) {
TSRMLS_FETCH();
plugin_state *state = (plugin_state *)ptr;
/* prepare to call the user callback function with info from the
@ -192,7 +194,7 @@ void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context,
state->fci->retval_ptr_ptr = &retval;
/* call the user callback function */
zend_call_function(state->fci, state->fci_cache);
zend_call_function(state->fci, state->fci_cache TSRMLS_CC);
if (Z_TYPE_P(retval) != IS_ARRAY) {
zend_throw_exception(spl_ce_InvalidArgumentException,

@ -84,7 +84,7 @@ zend_object_value create_wrapped_grpc_channel(zend_class_entry *class_type
return retval;
}
void php_grpc_read_args_array(zval *args_array, grpc_channel_args *args) {
void php_grpc_read_args_array(zval *args_array, grpc_channel_args *args TSRMLS_DC) {
HashTable *array_hash;
HashPosition array_pointer;
int args_index;
@ -168,7 +168,7 @@ PHP_METHOD(Channel, __construct) {
zend_hash_del(array_hash, "credentials", 12);
}
}
php_grpc_read_args_array(args_array, &args);
php_grpc_read_args_array(args_array, &args TSRMLS_CC);
if (creds == NULL) {
channel->wrapped = grpc_insecure_channel_create(target, &args, NULL);
} else {

@ -59,6 +59,6 @@ typedef struct wrapped_grpc_channel {
void grpc_init_channel(TSRMLS_D);
/* Iterates through a PHP array and populates args with the contents */
void php_grpc_read_args_array(zval *args_array, grpc_channel_args *args);
void php_grpc_read_args_array(zval *args_array, grpc_channel_args *args TSRMLS_DC);
#endif /* NET_GRPC_PHP_GRPC_CHANNEL_H_ */

@ -82,7 +82,7 @@ zend_object_value create_wrapped_grpc_channel_credentials(
return retval;
}
zval *grpc_php_wrap_channel_credentials(grpc_channel_credentials *wrapped) {
zval *grpc_php_wrap_channel_credentials(grpc_channel_credentials *wrapped TSRMLS_DC) {
zval *credentials_object;
MAKE_STD_ZVAL(credentials_object);
object_init_ex(credentials_object, grpc_ce_channel_credentials);
@ -99,7 +99,7 @@ zval *grpc_php_wrap_channel_credentials(grpc_channel_credentials *wrapped) {
*/
PHP_METHOD(ChannelCredentials, createDefault) {
grpc_channel_credentials *creds = grpc_google_default_credentials_create();
zval *creds_object = grpc_php_wrap_channel_credentials(creds);
zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}
@ -134,7 +134,7 @@ PHP_METHOD(ChannelCredentials, createSsl) {
grpc_channel_credentials *creds = grpc_ssl_credentials_create(
pem_root_certs,
pem_key_cert_pair.private_key == NULL ? NULL : &pem_key_cert_pair, NULL);
zval *creds_object = grpc_php_wrap_channel_credentials(creds);
zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}
@ -165,7 +165,7 @@ PHP_METHOD(ChannelCredentials, createComposite) {
grpc_channel_credentials *creds =
grpc_composite_channel_credentials_create(cred1->wrapped, cred2->wrapped,
NULL);
zval *creds_object = grpc_php_wrap_channel_credentials(creds);
zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}

@ -111,7 +111,7 @@ PHP_METHOD(Server, __construct) {
if (args_array == NULL) {
server->wrapped = grpc_server_create(NULL, NULL);
} else {
php_grpc_read_args_array(args_array, &args);
php_grpc_read_args_array(args_array, &args TSRMLS_CC);
server->wrapped = grpc_server_create(&args, NULL);
efree(args.args);
}
@ -154,12 +154,12 @@ PHP_METHOD(Server, requestCall) {
1 TSRMLS_CC);
goto cleanup;
}
add_property_zval(result, "call", grpc_php_wrap_call(call, true));
add_property_zval(result, "call", grpc_php_wrap_call(call, true TSRMLS_CC));
add_property_string(result, "method", details.method, true);
add_property_string(result, "host", details.host, true);
add_property_zval(result, "absolute_deadline",
grpc_php_wrap_timeval(details.deadline));
add_property_zval(result, "metadata", grpc_parse_metadata_array(&metadata));
grpc_php_wrap_timeval(details.deadline TSRMLS_CC));
add_property_zval(result, "metadata", grpc_parse_metadata_array(&metadata TSRMLS_CC));
cleanup:
grpc_call_details_destroy(&details);
grpc_metadata_array_destroy(&metadata);

@ -81,7 +81,7 @@ zend_object_value create_wrapped_grpc_server_credentials(
return retval;
}
zval *grpc_php_wrap_server_credentials(grpc_server_credentials *wrapped) {
zval *grpc_php_wrap_server_credentials(grpc_server_credentials *wrapped TSRMLS_DC) {
zval *server_credentials_object;
MAKE_STD_ZVAL(server_credentials_object);
object_init_ex(server_credentials_object, grpc_ce_server_credentials);
@ -120,7 +120,7 @@ PHP_METHOD(ServerCredentials, createSsl) {
grpc_server_credentials *creds = grpc_ssl_server_credentials_create_ex(
pem_root_certs, &pem_key_cert_pair, 1,
GRPC_SSL_DONT_REQUEST_CLIENT_CERTIFICATE, NULL);
zval *creds_object = grpc_php_wrap_server_credentials(creds);
zval *creds_object = grpc_php_wrap_server_credentials(creds TSRMLS_CC);
RETURN_DESTROY_ZVAL(creds_object);
}

@ -72,7 +72,7 @@ zend_object_value create_wrapped_grpc_timeval(zend_class_entry *class_type
return retval;
}
zval *grpc_php_wrap_timeval(gpr_timespec wrapped) {
zval *grpc_php_wrap_timeval(gpr_timespec wrapped TSRMLS_DC) {
zval *timeval_object;
MAKE_STD_ZVAL(timeval_object);
object_init_ex(timeval_object, grpc_ce_timeval);
@ -122,7 +122,7 @@ PHP_METHOD(Timeval, add) {
wrapped_grpc_timeval *other =
(wrapped_grpc_timeval *)zend_object_store_get_object(other_obj TSRMLS_CC);
zval *sum =
grpc_php_wrap_timeval(gpr_time_add(self->wrapped, other->wrapped));
grpc_php_wrap_timeval(gpr_time_add(self->wrapped, other->wrapped) TSRMLS_CC);
RETURN_DESTROY_ZVAL(sum);
}
@ -146,7 +146,7 @@ PHP_METHOD(Timeval, subtract) {
wrapped_grpc_timeval *other =
(wrapped_grpc_timeval *)zend_object_store_get_object(other_obj TSRMLS_CC);
zval *diff =
grpc_php_wrap_timeval(gpr_time_sub(self->wrapped, other->wrapped));
grpc_php_wrap_timeval(gpr_time_sub(self->wrapped, other->wrapped) TSRMLS_CC);
RETURN_DESTROY_ZVAL(diff);
}
@ -208,7 +208,7 @@ PHP_METHOD(Timeval, similar) {
* @return Timeval The current time
*/
PHP_METHOD(Timeval, now) {
zval *now = grpc_php_wrap_timeval(gpr_now(GPR_CLOCK_REALTIME));
zval *now = grpc_php_wrap_timeval(gpr_now(GPR_CLOCK_REALTIME) TSRMLS_CC);
RETURN_DESTROY_ZVAL(now);
}
@ -218,7 +218,7 @@ PHP_METHOD(Timeval, now) {
*/
PHP_METHOD(Timeval, zero) {
zval *grpc_php_timeval_zero =
grpc_php_wrap_timeval(gpr_time_0(GPR_CLOCK_REALTIME));
grpc_php_wrap_timeval(gpr_time_0(GPR_CLOCK_REALTIME) TSRMLS_CC);
RETURN_ZVAL(grpc_php_timeval_zero,
false, /* Copy original before returning? */
true /* Destroy original before returning */);
@ -230,7 +230,7 @@ PHP_METHOD(Timeval, zero) {
*/
PHP_METHOD(Timeval, infFuture) {
zval *grpc_php_timeval_inf_future =
grpc_php_wrap_timeval(gpr_inf_future(GPR_CLOCK_REALTIME));
grpc_php_wrap_timeval(gpr_inf_future(GPR_CLOCK_REALTIME) TSRMLS_CC);
RETURN_DESTROY_ZVAL(grpc_php_timeval_inf_future);
}
@ -240,7 +240,7 @@ PHP_METHOD(Timeval, infFuture) {
*/
PHP_METHOD(Timeval, infPast) {
zval *grpc_php_timeval_inf_past =
grpc_php_wrap_timeval(gpr_inf_past(GPR_CLOCK_REALTIME));
grpc_php_wrap_timeval(gpr_inf_past(GPR_CLOCK_REALTIME) TSRMLS_CC);
RETURN_DESTROY_ZVAL(grpc_php_timeval_inf_past);
}

@ -63,6 +63,6 @@ void grpc_init_timeval(TSRMLS_D);
void grpc_shutdown_timeval(TSRMLS_D);
/* Creates a Timeval object that wraps the given timeval struct */
zval *grpc_php_wrap_timeval(gpr_timespec wrapped);
zval *grpc_php_wrap_timeval(gpr_timespec wrapped TSRMLS_DC);
#endif /* NET_GRPC_PHP_GRPC_TIMEVAL_H_ */

@ -195,26 +195,30 @@ class Call(_types.Call):
translated_op = cygrpc.operation_send_initial_metadata(
cygrpc.Metadata(
cygrpc.Metadatum(key, value)
for key, value in op.initial_metadata))
for key, value in op.initial_metadata),
op.flags)
elif op.type == _types.OpType.SEND_MESSAGE:
translated_op = cygrpc.operation_send_message(op.message)
translated_op = cygrpc.operation_send_message(op.message, op.flags)
elif op.type == _types.OpType.SEND_CLOSE_FROM_CLIENT:
translated_op = cygrpc.operation_send_close_from_client()
translated_op = cygrpc.operation_send_close_from_client(op.flags)
elif op.type == _types.OpType.SEND_STATUS_FROM_SERVER:
translated_op = cygrpc.operation_send_status_from_server(
cygrpc.Metadata(
cygrpc.Metadatum(key, value)
for key, value in op.trailing_metadata),
op.status.code,
op.status.details)
op.status.details,
op.flags)
elif op.type == _types.OpType.RECV_INITIAL_METADATA:
translated_op = cygrpc.operation_receive_initial_metadata()
translated_op = cygrpc.operation_receive_initial_metadata(
op.flags)
elif op.type == _types.OpType.RECV_MESSAGE:
translated_op = cygrpc.operation_receive_message()
translated_op = cygrpc.operation_receive_message(op.flags)
elif op.type == _types.OpType.RECV_STATUS_ON_CLIENT:
translated_op = cygrpc.operation_receive_status_on_client()
translated_op = cygrpc.operation_receive_status_on_client(
op.flags)
elif op.type == _types.OpType.RECV_CLOSE_ON_SERVER:
translated_op = cygrpc.operation_receive_close_on_server()
translated_op = cygrpc.operation_receive_close_on_server(op.flags)
else:
raise ValueError('unexpected operation type {}'.format(op.type))
translated_ops.append(translated_op)

@ -152,7 +152,7 @@ class OpArgs(collections.namedtuple(
'trailing_metadata',
'message',
'status',
'write_flags',
'flags',
])):
"""Arguments passed into a GRPC operation.
@ -165,7 +165,7 @@ class OpArgs(collections.namedtuple(
message (bytes): Only valid if type == OpType.SEND_MESSAGE, else is None.
status (Status): Only valid if type == OpType.SEND_STATUS_FROM_SERVER, else
is None.
write_flags (int): a bit OR'ing of 0 or more OpWriteFlags values.
flags (int): a bitwise OR'ing of 0 or more OpWriteFlags values.
"""
@staticmethod

@ -140,6 +140,9 @@ cdef extern from "grpc/_cython/loader.h":
const char *GRPC_ARG_PRIMARY_USER_AGENT_STRING
const char *GRPC_ARG_SECONDARY_USER_AGENT_STRING
const char *GRPC_SSL_TARGET_NAME_OVERRIDE_ARG
const char *GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM
const char *GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL
const char *GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET
const int GRPC_WRITE_BUFFER_HINT
const int GRPC_WRITE_NO_COMPRESS
@ -425,3 +428,38 @@ cdef extern from "grpc/_cython/loader.h":
grpc_call_credentials *grpc_metadata_credentials_create_from_plugin(
grpc_metadata_credentials_plugin plugin, void *reserved) nogil
ctypedef enum grpc_compression_algorithm:
GRPC_COMPRESS_NONE
GRPC_COMPRESS_DEFLATE
GRPC_COMPRESS_GZIP
GRPC_COMPRESS_ALGORITHMS_COUNT
ctypedef enum grpc_compression_level:
GRPC_COMPRESS_LEVEL_NONE
GRPC_COMPRESS_LEVEL_LOW
GRPC_COMPRESS_LEVEL_MED
GRPC_COMPRESS_LEVEL_HIGH
GRPC_COMPRESS_LEVEL_COUNT
ctypedef struct grpc_compression_options:
uint32_t enabled_algorithms_bitset
grpc_compression_algorithm default_compression_algorithm
int grpc_compression_algorithm_parse(
const char *name, size_t name_length,
grpc_compression_algorithm *algorithm) nogil
int grpc_compression_algorithm_name(grpc_compression_algorithm algorithm,
char **name) nogil
grpc_compression_algorithm grpc_compression_algorithm_for_level(
grpc_compression_level level, uint32_t accepted_encodings) nogil
void grpc_compression_options_init(grpc_compression_options *opts) nogil
void grpc_compression_options_enable_algorithm(
grpc_compression_options *opts,
grpc_compression_algorithm algorithm) nogil
void grpc_compression_options_disable_algorithm(
grpc_compression_options *opts,
grpc_compression_algorithm algorithm) nogil
int grpc_compression_options_is_algorithm_enabled(
const grpc_compression_options *opts,
grpc_compression_algorithm algorithm) nogil

@ -124,3 +124,7 @@ cdef class Operations:
cdef size_t c_nops
cdef list operations
cdef class CompressionOptions:
cdef grpc_compression_options c_options

@ -103,6 +103,19 @@ class OperationType:
receive_close_on_server = GRPC_OP_RECV_CLOSE_ON_SERVER
class CompressionAlgorithm:
none = GRPC_COMPRESS_NONE
deflate = GRPC_COMPRESS_DEFLATE
gzip = GRPC_COMPRESS_GZIP
class CompressionLevel:
none = GRPC_COMPRESS_LEVEL_NONE
low = GRPC_COMPRESS_LEVEL_LOW
medium = GRPC_COMPRESS_LEVEL_MED
high = GRPC_COMPRESS_LEVEL_HIGH
cdef class Timespec:
def __cinit__(self, time):
@ -472,6 +485,10 @@ cdef class Operation:
def type(self):
return self.c_op.type
@property
def flags(self):
return self.c_op.flags
@property
def has_status(self):
return self.c_op.type == GRPC_OP_RECV_STATUS_ON_CLIENT
@ -553,9 +570,10 @@ cdef class Operation:
with nogil:
gpr_free(self._received_status_details)
def operation_send_initial_metadata(Metadata metadata):
def operation_send_initial_metadata(Metadata metadata, int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_SEND_INITIAL_METADATA
op.c_op.flags = flags
op.c_op.data.send_initial_metadata.count = metadata.c_metadata_array.count
op.c_op.data.send_initial_metadata.metadata = (
metadata.c_metadata_array.metadata)
@ -563,23 +581,25 @@ def operation_send_initial_metadata(Metadata metadata):
op.is_valid = True
return op
def operation_send_message(data):
def operation_send_message(data, int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_SEND_MESSAGE
op.c_op.flags = flags
byte_buffer = ByteBuffer(data)
op.c_op.data.send_message = byte_buffer.c_byte_buffer
op.references.append(byte_buffer)
op.is_valid = True
return op
def operation_send_close_from_client():
def operation_send_close_from_client(int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_SEND_CLOSE_FROM_CLIENT
op.c_op.flags = flags
op.is_valid = True
return op
def operation_send_status_from_server(
Metadata metadata, grpc_status_code code, details):
Metadata metadata, grpc_status_code code, details, int flags):
if isinstance(details, bytes):
pass
elif isinstance(details, basestring):
@ -588,6 +608,7 @@ def operation_send_status_from_server(
raise TypeError("expected a str or bytes object for details")
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_SEND_STATUS_FROM_SERVER
op.c_op.flags = flags
op.c_op.data.send_status_from_server.trailing_metadata_count = (
metadata.c_metadata_array.count)
op.c_op.data.send_status_from_server.trailing_metadata = (
@ -599,18 +620,20 @@ def operation_send_status_from_server(
op.is_valid = True
return op
def operation_receive_initial_metadata():
def operation_receive_initial_metadata(int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_RECV_INITIAL_METADATA
op.c_op.flags = flags
op._received_metadata = Metadata([])
op.c_op.data.receive_initial_metadata = (
&op._received_metadata.c_metadata_array)
op.is_valid = True
return op
def operation_receive_message():
def operation_receive_message(int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_RECV_MESSAGE
op.c_op.flags = flags
op._received_message = ByteBuffer(None)
# n.b. the c_op.data.receive_message field needs to be deleted by us,
# anyway, so we just let that be handled by the ByteBuffer() we allocated
@ -619,9 +642,10 @@ def operation_receive_message():
op.is_valid = True
return op
def operation_receive_status_on_client():
def operation_receive_status_on_client(int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_RECV_STATUS_ON_CLIENT
op.c_op.flags = flags
op._received_metadata = Metadata([])
op.c_op.data.receive_status_on_client.trailing_metadata = (
&op._received_metadata.c_metadata_array)
@ -634,9 +658,10 @@ def operation_receive_status_on_client():
op.is_valid = True
return op
def operation_receive_close_on_server():
def operation_receive_close_on_server(int flags):
cdef Operation op = Operation()
op.c_op.type = GRPC_OP_RECV_CLOSE_ON_SERVER
op.c_op.flags = flags
op.c_op.data.receive_close_on_server.cancelled = &op._received_cancelled
op.is_valid = True
return op
@ -692,3 +717,36 @@ cdef class Operations:
def __iter__(self):
return _OperationsIterator(self)
cdef class CompressionOptions:
def __cinit__(self):
with nogil:
grpc_compression_options_init(&self.c_options)
def enable_algorithm(self, grpc_compression_algorithm algorithm):
with nogil:
grpc_compression_options_enable_algorithm(&self.c_options, algorithm)
def disable_algorithm(self, grpc_compression_algorithm algorithm):
with nogil:
grpc_compression_options_disable_algorithm(&self.c_options, algorithm)
def is_algorithm_enabled(self, grpc_compression_algorithm algorithm):
cdef int result
with nogil:
result = grpc_compression_options_is_algorithm_enabled(
&self.c_options, algorithm)
return result
def to_channel_arg(self):
return ChannelArg(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET,
self.c_options.enabled_algorithms_bitset)
def compression_algorithm_name(grpc_compression_algorithm algorithm):
cdef char* name
with nogil:
grpc_compression_algorithm_name(algorithm, &name)
# Let Cython do the right thing with string casting
return name

@ -873,14 +873,15 @@ void pygrpc_load_imports(HMODULE library);
#else /* !GPR_WIN32 */
#include <grpc/support/alloc.h>
#include <grpc/support/slice.h>
#include <grpc/support/time.h>
#include <grpc/status.h>
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>
#include <grpc/compression.h>
#include <grpc/grpc.h>
#include <grpc/grpc_security.h>
#include <grpc/support/alloc.h>
#include <grpc/support/slice.h>
#include <grpc/support/time.h>
#include <grpc/status.h>
#endif /* !GPR_WIN32 */

@ -235,7 +235,7 @@ class Server(six.with_metaclass(abc.ABCMeta)):
This method may be called at any time and is idempotent. Passing a smaller
grace value than has been passed in a previous call will have the effect of
stopping the Server sooner. Passing a larger grace value than has been
passed in a previous call will not have the effect of stopping the sooner
passed in a previous call will not have the effect of stopping the server
later.
Args:

@ -0,0 +1,28 @@
# 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.

@ -0,0 +1,75 @@
# 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.
"""Tests of grpc_health.health.v1.health."""
import unittest
from grpc_health.health.v1 import health
from grpc_health.health.v1 import health_pb2
class HealthServicerTest(unittest.TestCase):
def setUp(self):
self.servicer = health.HealthServicer()
self.servicer.set('', health_pb2.HealthCheckResponse.SERVING)
self.servicer.set('grpc.test.TestServiceServing',
health_pb2.HealthCheckResponse.SERVING)
self.servicer.set('grpc.test.TestServiceUnknown',
health_pb2.HealthCheckResponse.UNKNOWN)
self.servicer.set('grpc.test.TestServiceNotServing',
health_pb2.HealthCheckResponse.NOT_SERVING)
def test_empty_service(self):
request = health_pb2.HealthCheckRequest()
resp = self.servicer.Check(request, None)
self.assertEqual(resp.status, health_pb2.HealthCheckResponse.SERVING)
def test_serving_service(self):
request = health_pb2.HealthCheckRequest(
service='grpc.test.TestServiceServing')
resp = self.servicer.Check(request, None)
self.assertEqual(resp.status, health_pb2.HealthCheckResponse.SERVING)
def test_unknown_serivce(self):
request = health_pb2.HealthCheckRequest(
service='grpc.test.TestServiceUnknown')
resp = self.servicer.Check(request, None)
self.assertEqual(resp.status, health_pb2.HealthCheckResponse.UNKNOWN)
def test_not_serving_service(self):
request = health_pb2.HealthCheckRequest(
service='grpc.test.TestServiceNotServing')
resp = self.servicer.Check(request, None)
self.assertEqual(resp.status, health_pb2.HealthCheckResponse.NOT_SERVING)
if __name__ == '__main__':
unittest.main(verbosity=2)

@ -28,7 +28,8 @@
"_face_interface_test.GenericInvokerBlockingInvocationInlineServiceTest",
"_face_interface_test.GenericInvokerFutureInvocationAsynchronousEventServiceTest",
"_face_interface_test.MultiCallableInvokerBlockingInvocationInlineServiceTest",
"_face_interface_test.MultiCallableInvokerFutureInvocationAsynchronousEventServiceTest",
"_face_interface_test.MultiCallableInvokerFutureInvocationAsynchronousEventServiceTest",
"_health_servicer_test.HealthServicerTest",
"_implementations_test.ChannelCredentialsTest",
"_insecure_interop_test.InsecureInteropTest",
"_intermediary_low_test.CancellationTest",
@ -50,4 +51,4 @@
"cygrpc_test.InsecureServerInsecureClient",
"cygrpc_test.SecureServerSecureClient",
"cygrpc_test.TypeSmokeTest"
]
]

@ -60,7 +60,7 @@ def _create_loop_destroy():
def _in_parallel(behavior, arguments):
threads = tuple(
threading.Thread(target=behavior, args=arguments)
for _ in range(test_constants.PARALLELISM))
for _ in range(test_constants.THREAD_CONCURRENCY))
for thread in threads:
thread.start()
for thread in threads:

@ -40,6 +40,7 @@ from tests.unit import resources
_SSL_HOST_OVERRIDE = 'foo.test.google.fr'
_CALL_CREDENTIALS_METADATA_KEY = 'call-creds-key'
_CALL_CREDENTIALS_METADATA_VALUE = 'call-creds-value'
_EMPTY_FLAGS = 0
def _metadata_plugin_callback(context, callback):
callback(cygrpc.Metadata(
@ -76,7 +77,7 @@ class TypeSmokeTest(unittest.TestCase):
def testOperationsIteration(self):
operations = cygrpc.Operations([
cygrpc.operation_send_message('asdf')])
cygrpc.operation_send_message('asdf', _EMPTY_FLAGS)])
iterator = iter(operations)
operation = next(iterator)
self.assertIsInstance(operation, cygrpc.Operation)
@ -85,6 +86,11 @@ class TypeSmokeTest(unittest.TestCase):
with self.assertRaises(StopIteration):
next(iterator)
def testOperationFlags(self):
operation = cygrpc.operation_send_message('asdf',
cygrpc.WriteFlag.no_compress)
self.assertEqual(cygrpc.WriteFlag.no_compress, operation.flags)
def testTimespec(self):
now = time.time()
timespec = cygrpc.Timespec(now)
@ -188,12 +194,13 @@ class InsecureServerInsecureClient(unittest.TestCase):
CLIENT_METADATA_ASCII_VALUE),
cygrpc.Metadatum(CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)])
client_start_batch_result = client_call.start_batch(cygrpc.Operations([
cygrpc.operation_send_initial_metadata(client_initial_metadata),
cygrpc.operation_send_message(REQUEST),
cygrpc.operation_send_close_from_client(),
cygrpc.operation_receive_initial_metadata(),
cygrpc.operation_receive_message(),
cygrpc.operation_receive_status_on_client()
cygrpc.operation_send_initial_metadata(client_initial_metadata,
_EMPTY_FLAGS),
cygrpc.operation_send_message(REQUEST, _EMPTY_FLAGS),
cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
cygrpc.operation_receive_message(_EMPTY_FLAGS),
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS)
]), client_call_tag)
self.assertEqual(cygrpc.CallError.ok, client_start_batch_result)
client_event_future = test_utilities.CompletionQueuePollFuture(
@ -223,12 +230,14 @@ class InsecureServerInsecureClient(unittest.TestCase):
cygrpc.Metadatum(SERVER_TRAILING_METADATA_KEY,
SERVER_TRAILING_METADATA_VALUE)])
server_start_batch_result = server_call.start_batch([
cygrpc.operation_send_initial_metadata(server_initial_metadata),
cygrpc.operation_receive_message(),
cygrpc.operation_send_message(RESPONSE),
cygrpc.operation_receive_close_on_server(),
cygrpc.operation_send_initial_metadata(server_initial_metadata,
_EMPTY_FLAGS),
cygrpc.operation_receive_message(_EMPTY_FLAGS),
cygrpc.operation_send_message(RESPONSE, _EMPTY_FLAGS),
cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),
cygrpc.operation_send_status_from_server(
server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS)
server_trailing_metadata, SERVER_STATUS_CODE,
SERVER_STATUS_DETAILS, _EMPTY_FLAGS)
], server_call_tag)
self.assertEqual(cygrpc.CallError.ok, server_start_batch_result)
@ -349,12 +358,13 @@ class SecureServerSecureClient(unittest.TestCase):
CLIENT_METADATA_ASCII_VALUE),
cygrpc.Metadatum(CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)])
client_start_batch_result = client_call.start_batch(cygrpc.Operations([
cygrpc.operation_send_initial_metadata(client_initial_metadata),
cygrpc.operation_send_message(REQUEST),
cygrpc.operation_send_close_from_client(),
cygrpc.operation_receive_initial_metadata(),
cygrpc.operation_receive_message(),
cygrpc.operation_receive_status_on_client()
cygrpc.operation_send_initial_metadata(client_initial_metadata,
_EMPTY_FLAGS),
cygrpc.operation_send_message(REQUEST, _EMPTY_FLAGS),
cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
cygrpc.operation_receive_message(_EMPTY_FLAGS),
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS)
]), client_call_tag)
self.assertEqual(cygrpc.CallError.ok, client_start_batch_result)
client_event_future = test_utilities.CompletionQueuePollFuture(
@ -387,12 +397,14 @@ class SecureServerSecureClient(unittest.TestCase):
cygrpc.Metadatum(SERVER_TRAILING_METADATA_KEY,
SERVER_TRAILING_METADATA_VALUE)])
server_start_batch_result = server_call.start_batch([
cygrpc.operation_send_initial_metadata(server_initial_metadata),
cygrpc.operation_receive_message(),
cygrpc.operation_send_message(RESPONSE),
cygrpc.operation_receive_close_on_server(),
cygrpc.operation_send_initial_metadata(server_initial_metadata,
_EMPTY_FLAGS),
cygrpc.operation_receive_message(_EMPTY_FLAGS),
cygrpc.operation_send_message(RESPONSE, _EMPTY_FLAGS),
cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),
cygrpc.operation_send_status_from_server(
server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS)
server_trailing_metadata, SERVER_STATUS_CODE,
SERVER_STATUS_DETAILS, _EMPTY_FLAGS)
], server_call_tag)
self.assertEqual(cygrpc.CallError.ok, server_start_batch_result)

@ -49,8 +49,13 @@ STREAM_LENGTH = 200
# The size of payloads to transmit in tests.
PAYLOAD_SIZE = 256 * 1024 + 17
# The parallelism to use in tests of parallel RPCs.
PARALLELISM = 200
# The concurrency to use in tests of concurrent RPCs that will not create as
# many threads as RPCs.
RPC_CONCURRENCY = 200
# The concurrency to use in tests of concurrent RPCs that will create as many
# threads as RPCs.
THREAD_CONCURRENCY = 25
# The size of thread pools to use in tests.
POOL_SIZE = 10

@ -146,13 +146,13 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
test_messages.verify(second_request, second_response, self)
def testParallelInvocations(self):
pool = logging_pool.pool(test_constants.PARALLELISM)
pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
for (group, method), test_messages_sequence in (
six.iteritems(self._digest.unary_unary_messages_sequences)):
for test_messages in test_messages_sequence:
requests = []
response_futures = []
for _ in range(test_constants.PARALLELISM):
for _ in range(test_constants.THREAD_CONCURRENCY):
request = test_messages.request()
response_future = pool.submit(
self._invoker.blocking(group, method), request,
@ -168,13 +168,13 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
pool.shutdown(wait=True)
def testWaitingForSomeButNotAllParallelInvocations(self):
pool = logging_pool.pool(test_constants.PARALLELISM)
pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
for (group, method), test_messages_sequence in (
six.iteritems(self._digest.unary_unary_messages_sequences)):
for test_messages in test_messages_sequence:
requests = []
response_futures_to_indices = {}
for index in range(test_constants.PARALLELISM):
for index in range(test_constants.THREAD_CONCURRENCY):
request = test_messages.request()
response_future = pool.submit(
self._invoker.blocking(group, method), request,
@ -184,7 +184,7 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
some_completed_response_futures_iterator = itertools.islice(
futures.as_completed(response_futures_to_indices),
test_constants.PARALLELISM // 2)
test_constants.THREAD_CONCURRENCY // 2)
for response_future in some_completed_response_futures_iterator:
index = response_futures_to_indices[response_future]
test_messages.verify(requests[index], response_future.result(), self)

@ -249,7 +249,7 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
for test_messages in test_messages_sequence:
requests = []
response_futures = []
for _ in range(test_constants.PARALLELISM):
for _ in range(test_constants.THREAD_CONCURRENCY):
request = test_messages.request()
response_future = self._invoker.future(group, method)(
request, test_constants.LONG_TIMEOUT)
@ -263,13 +263,13 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
test_messages.verify(request, response, self)
def testWaitingForSomeButNotAllParallelInvocations(self):
pool = logging_pool.pool(test_constants.PARALLELISM)
pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
for (group, method), test_messages_sequence in (
six.iteritems(self._digest.unary_unary_messages_sequences)):
for test_messages in test_messages_sequence:
requests = []
response_futures_to_indices = {}
for index in range(test_constants.PARALLELISM):
for index in range(test_constants.THREAD_CONCURRENCY):
request = test_messages.request()
inner_response_future = self._invoker.future(group, method)(
request, test_constants.LONG_TIMEOUT)
@ -279,7 +279,7 @@ class TestCase(six.with_metaclass(abc.ABCMeta, test_coverage.Coverage, unittest.
some_completed_response_futures_iterator = itertools.islice(
futures.as_completed(response_futures_to_indices),
test_constants.PARALLELISM // 2)
test_constants.THREAD_CONCURRENCY // 2)
for response_future in some_completed_response_futures_iterator:
index = response_futures_to_indices[response_future]
test_messages.verify(requests[index], response_future.result(), self)

@ -0,0 +1,5 @@
*.proto
*_pb2.py
build/
grpcio_health_checking.egg-info/
dist/

@ -1,2 +1,3 @@
graft grpc
include commands.py
include health_commands.py
graft grpc_health
global-exclude *.pyc

@ -1,49 +0,0 @@
// Copyright 2015, 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.
syntax = "proto3";
package grpc.health.v1;
message HealthCheckRequest {
string service = 1;
}
message HealthCheckResponse {
enum ServingStatus {
UNKNOWN = 0;
SERVING = 1;
NOT_SERVING = 2;
}
ServingStatus status = 1;
}
service Health {
rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
}

@ -1,129 +0,0 @@
# Copyright 2015, 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.
"""Reference implementation for health checking in gRPC Python."""
import abc
import enum
import threading
from grpc.health.v1 import health_pb2
@enum.unique
class HealthStatus(enum.Enum):
"""Statuses for a service mirroring the reference health.proto's values."""
UNKNOWN = health_pb2.HealthCheckResponse.UNKNOWN
SERVING = health_pb2.HealthCheckResponse.SERVING
NOT_SERVING = health_pb2.HealthCheckResponse.NOT_SERVING
class _HealthServicer(health_pb2.EarlyAdopterHealthServicer):
"""Servicer handling RPCs for service statuses."""
def __init__(self):
self._server_status_lock = threading.Lock()
self._server_status = {}
def Check(self, request, context):
with self._server_status_lock:
if request.service not in self._server_status:
# TODO(atash): once the Python API has a way of setting the server
# status, bring us into conformance with the health check spec by
# returning the NOT_FOUND status here.
raise NotImplementedError()
else:
return health_pb2.HealthCheckResponse(
status=self._server_status[request.service].value)
def set(service, status):
if not isinstance(status, HealthStatus):
raise TypeError('expected grpc.health.v1.health.HealthStatus '
'for argument `status` but got {}'.format(status))
with self._server_status_lock:
self._server_status[service] = status
class HealthServer(health_pb2.EarlyAdopterHealthServer):
"""Interface for the reference gRPC Python health server."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def start(self):
raise NotImplementedError()
@abc.abstractmethod
def stop(self):
raise NotImplementedError()
@abc.abstractmethod
def set(self, service, status):
"""Set the status of the given service.
Args:
service (str): service name of the service to set the reported status of
status (HealthStatus): status to set for the specified service
"""
raise NotImplementedError()
class _HealthServerImplementation(HealthServer):
"""Implementation for the reference gRPC Python health server."""
def __init__(self, server, servicer):
self._server = server
self._servicer = servicer
def start(self):
self._server.start()
def stop(self):
self._server.stop()
def set(self, service, status):
self._servicer.set(service, status)
def create_Health_server(port, private_key=None, certificate_chain=None):
"""Get a HealthServer instance.
Args:
port (int): port number passed through to health_pb2 server creation
routine.
private_key (str): to-be-created server's desired private key
certificate_chain (str): to-be-created server's desired certificate chain
Returns:
An instance of HealthServer (conforming thus to
EarlyAdopterHealthServer and providing a method to set server status)."""
servicer = _HealthServicer()
server = health_pb2.early_adopter_create_Health_server(
servicer, port=port, private_key=private_key,
certificate_chain=certificate_chain)
return _HealthServerImplementation(server, servicer)

@ -0,0 +1,66 @@
# Copyright 2015, 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.
"""Reference implementation for health checking in gRPC Python."""
import threading
from grpc_health.health.v1 import health_pb2
class HealthServicer(health_pb2.BetaHealthServicer):
"""Servicer handling RPCs for service statuses."""
def __init__(self):
self._server_status_lock = threading.Lock()
self._server_status = {}
def Check(self, request, context):
with self._server_status_lock:
if request.service not in self._server_status:
# TODO(atash): once the Python API has a way of setting the server
# status, bring us into conformance with the health check spec by
# returning the NOT_FOUND status here.
raise NotImplementedError()
else:
return health_pb2.HealthCheckResponse(
status=self._server_status[request.service])
def set(self, service, status):
"""Sets the status of a service.
Args:
service: string, the name of the service.
NOTE, '' must be set.
status: HealthCheckResponse.status enum value indicating
the status of the service
"""
with self._server_status_lock:
self._server_status[service] = status

@ -33,11 +33,16 @@ import distutils
import glob
import os
import os.path
import shutil
import subprocess
import sys
import setuptools
from setuptools.command import build_py
from setuptools.command import sdist
ROOT_DIR = os.path.abspath(os.path.dirname(os.path.abspath(__file__)))
HEALTH_PROTO = os.path.join(ROOT_DIR, '../../proto/grpc/health/v1/health.proto')
class BuildProtoModules(setuptools.Command):
@ -76,9 +81,34 @@ class BuildProtoModules(setuptools.Command):
raise Exception('{}\nOutput:\n{}'.format(e.message, e.output))
class CopyProtoModules(setuptools.Command):
"""Command to copy proto modules from grpc/src/proto."""
def initialize_options(self):
pass
def finalize_options(self):
pass
def run(self):
if os.path.isfile(HEALTH_PROTO):
shutil.copyfile(
HEALTH_PROTO,
os.path.join(ROOT_DIR, 'grpc_health/health/v1/health.proto'))
class BuildPy(build_py.build_py):
"""Custom project build command."""
def run(self):
self.run_command('copy_proto_modules')
self.run_command('build_proto_modules')
build_py.build_py.run(self)
class SDist(sdist.sdist):
"""Custom project build command."""
def run(self):
self.run_command('copy_proto_modules')
sdist.sdist.run(self)

@ -40,7 +40,7 @@ import setuptools
os.chdir(os.path.dirname(os.path.abspath(__file__)))
# Break import-style to ensure we can actually find our commands module.
import commands
import health_commands
_PACKAGES = (
setuptools.find_packages('.')
@ -51,22 +51,21 @@ _PACKAGE_DIRECTORIES = {
}
_INSTALL_REQUIRES = (
'grpcio>=0.11.0b0',
'grpcio>=0.13.1',
)
_SETUP_REQUIRES = _INSTALL_REQUIRES
_COMMAND_CLASS = {
'build_proto_modules': commands.BuildProtoModules,
'build_py': commands.BuildPy,
'copy_proto_modules': health_commands.CopyProtoModules,
'build_proto_modules': health_commands.BuildProtoModules,
'build_py': health_commands.BuildPy,
'sdist': health_commands.SDist,
}
setuptools.setup(
name='grpcio_health_checking',
version='0.11.0b0',
version='0.14.0b0',
packages=list(_PACKAGES),
package_dir=_PACKAGE_DIRECTORIES,
install_requires=_INSTALL_REQUIRES,
setup_requires=_SETUP_REQUIRES,
cmdclass=_COMMAND_CLASS
)

@ -64,14 +64,15 @@
#else /* !GPR_WIN32 */
#include <grpc/support/alloc.h>
#include <grpc/support/slice.h>
#include <grpc/support/time.h>
#include <grpc/status.h>
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>
#include <grpc/compression.h>
#include <grpc/grpc.h>
#include <grpc/grpc_security.h>
#include <grpc/support/alloc.h>
#include <grpc/support/slice.h>
#include <grpc/support/time.h>
#include <grpc/status.h>
#endif /* !GPR_WIN32 */

@ -88,9 +88,11 @@ void test_connect(const char *server_host, const char *client_host, int port,
int was_cancelled = 2;
grpc_call_details call_details;
char *peer;
int picked_port = 0;
if (port == 0) {
port = grpc_pick_unused_port_or_die();
picked_port = 1;
}
gpr_join_host_port(&server_hostport, server_host, port);
@ -263,6 +265,9 @@ void test_connect(const char *server_host, const char *client_host, int port,
grpc_call_details_destroy(&call_details);
gpr_free(details);
if (picked_port) {
grpc_recycle_unused_port(port);
}
}
int external_dns_works(const char *host) {

@ -340,6 +340,8 @@ static void free_non_null(void *p) {
typedef enum { ROOT, CLIENT, SERVER, PENDING_SERVER } call_state_type;
#define DONE_FLAG_CALL_CLOSED ((uint64_t)(1 << 0))
typedef struct call_state {
call_state_type type;
grpc_call *call;
@ -352,6 +354,10 @@ typedef struct call_state {
int cancelled;
int pending_ops;
grpc_call_details call_details;
grpc_byte_buffer *send_message;
// starts at 0, individual flags from DONE_FLAG_xxx are set
// as different operations are completed
uint64_t done_flags;
// array of pointers to free later
size_t num_to_free;
@ -449,10 +455,41 @@ static void finished_request_call(void *csp, bool success) {
}
}
static void finished_batch(void *csp, bool success) {
call_state *cs = csp;
--cs->pending_ops;
maybe_delete_call_state(cs);
typedef struct {
call_state *cs;
uint8_t has_ops;
} batch_info;
static void finished_batch(void *p, bool success) {
batch_info *bi = p;
--bi->cs->pending_ops;
if ((bi->has_ops & (1u << GRPC_OP_RECV_MESSAGE)) &&
(bi->cs->done_flags & DONE_FLAG_CALL_CLOSED)) {
GPR_ASSERT(bi->cs->recv_message == NULL);
}
if ((bi->has_ops & (1u << GRPC_OP_RECV_MESSAGE) &&
bi->cs->recv_message != NULL)) {
grpc_byte_buffer_destroy(bi->cs->recv_message);
bi->cs->recv_message = NULL;
}
if ((bi->has_ops & (1u << GRPC_OP_SEND_MESSAGE))) {
grpc_byte_buffer_destroy(bi->cs->send_message);
bi->cs->send_message = NULL;
}
if ((bi->has_ops & (1u << GRPC_OP_RECV_STATUS_ON_CLIENT)) ||
(bi->has_ops & (1u << GRPC_OP_RECV_CLOSE_ON_SERVER))) {
bi->cs->done_flags |= DONE_FLAG_CALL_CLOSED;
}
maybe_delete_call_state(bi->cs);
gpr_free(bi);
}
static validator *make_finished_batch_validator(call_state *cs,
uint8_t has_ops) {
batch_info *bi = gpr_malloc(sizeof(*bi));
bi->cs = cs;
bi->has_ops = has_ops;
return create_validator(finished_batch, bi);
}
int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
@ -579,6 +616,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
} else {
end(&inp);
}
break;
}
// begin server shutdown
case 5: {
@ -700,6 +738,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
bool ok = true;
size_t i;
grpc_op *op;
uint8_t has_ops = 0;
for (i = 0; i < num_ops; i++) {
op = &ops[i];
switch (next_byte(&inp)) {
@ -710,19 +749,28 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
break;
case GRPC_OP_SEND_INITIAL_METADATA:
op->op = GRPC_OP_SEND_INITIAL_METADATA;
has_ops |= 1 << GRPC_OP_SEND_INITIAL_METADATA;
read_metadata(&inp, &op->data.send_initial_metadata.count,
&op->data.send_initial_metadata.metadata,
g_active_call);
break;
case GRPC_OP_SEND_MESSAGE:
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message = read_message(&inp);
if (g_active_call->send_message != NULL) {
ok = false;
} else {
has_ops |= 1 << GRPC_OP_SEND_MESSAGE;
g_active_call->send_message = op->data.send_message =
read_message(&inp);
}
break;
case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
has_ops |= 1 << GRPC_OP_SEND_CLOSE_FROM_CLIENT;
break;
case GRPC_OP_SEND_STATUS_FROM_SERVER:
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
has_ops |= 1 << GRPC_OP_SEND_STATUS_FROM_SERVER;
read_metadata(
&inp,
&op->data.send_status_from_server.trailing_metadata_count,
@ -734,11 +782,13 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
break;
case GRPC_OP_RECV_INITIAL_METADATA:
op->op = GRPC_OP_RECV_INITIAL_METADATA;
has_ops |= 1 << GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata =
&g_active_call->recv_initial_metadata;
break;
case GRPC_OP_RECV_MESSAGE:
op->op = GRPC_OP_RECV_MESSAGE;
has_ops |= 1 << GRPC_OP_RECV_MESSAGE;
op->data.recv_message = &g_active_call->recv_message;
break;
case GRPC_OP_RECV_STATUS_ON_CLIENT:
@ -753,6 +803,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
break;
case GRPC_OP_RECV_CLOSE_ON_SERVER:
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
has_ops |= 1 << GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled =
&g_active_call->cancelled;
break;
@ -761,7 +812,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
op->flags = read_uint32(&inp);
}
if (ok) {
validator *v = create_validator(finished_batch, g_active_call);
validator *v = make_finished_batch_validator(g_active_call, has_ops);
g_active_call->pending_ops++;
grpc_call_error error =
grpc_call_start_batch(g_active_call->call, ops, num_ops, v, NULL);
@ -772,17 +823,18 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
} else {
end(&inp);
}
if (!ok && (has_ops & (1 << GRPC_OP_SEND_MESSAGE))) {
grpc_byte_buffer_destroy(g_active_call->send_message);
g_active_call->send_message = NULL;
}
for (i = 0; i < num_ops; i++) {
op = &ops[i];
switch (op->op) {
case GRPC_OP_SEND_INITIAL_METADATA:
break;
case GRPC_OP_SEND_MESSAGE:
grpc_byte_buffer_destroy(op->data.send_message);
break;
case GRPC_OP_SEND_STATUS_FROM_SERVER:
gpr_free((void *)op->data.send_status_from_server.status_details);
break;
case GRPC_OP_SEND_MESSAGE:
case GRPC_OP_SEND_INITIAL_METADATA:
case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
case GRPC_OP_RECV_INITIAL_METADATA:
case GRPC_OP_RECV_MESSAGE:

Some files were not shown because too many files have changed in this diff Show More

Loading…
Cancel
Save