Merge remote-tracking branch 'upstream/master' into mock

pull/1478/head
Yang Gao 10 years ago
commit 691ff71da5
  1. 1
      .travis.yml
  2. 448
      Makefile
  3. 2
      gRPC.podspec
  4. 5
      include/grpc++/config.h
  5. 10
      include/grpc++/impl/call.h
  6. 7
      include/grpc++/server.h
  7. 6
      include/grpc++/server_builder.h
  8. 3
      include/grpc/grpc.h
  9. 19
      src/compiler/cpp_generator.cc
  10. 141
      src/core/channel/call_op_string.c
  11. 44
      src/core/channel/context.h
  12. 2
      src/core/iomgr/endpoint_pair_windows.c
  13. 6
      src/core/iomgr/iocp_windows.c
  14. 4
      src/core/iomgr/iomgr_windows.c
  15. 3
      src/core/iomgr/pollset_kick_windows.h
  16. 5
      src/core/iomgr/pollset_windows.c
  17. 8
      src/core/iomgr/pollset_windows.h
  18. 53
      src/core/iomgr/socket_windows.c
  19. 55
      src/core/iomgr/socket_windows.h
  20. 48
      src/core/iomgr/tcp_client_windows.c
  21. 156
      src/core/iomgr/tcp_posix.c
  22. 93
      src/core/iomgr/tcp_server_windows.c
  23. 112
      src/core/iomgr/tcp_windows.c
  24. 98
      src/core/profiling/basic_timers.c
  25. 6
      src/core/profiling/stap_timers.c
  26. 36
      src/core/profiling/timers.h
  27. 45
      src/core/profiling/timers_preciseclock.h
  28. 4
      src/core/support/cpu_windows.c
  29. 61
      src/core/surface/call.c
  30. 8
      src/core/surface/call.h
  31. 1
      src/core/surface/init.c
  32. 86
      src/core/transport/chttp2_transport.c
  33. 1
      src/core/transport/chttp2_transport.h
  34. 3
      src/core/transport/transport.h
  35. 16
      src/cpp/common/call.cc
  36. 9
      src/cpp/proto/proto_utils.cc
  37. 3
      src/cpp/proto/proto_utils.h
  38. 37
      src/cpp/server/server.cc
  39. 5
      src/cpp/server/server_builder.cc
  40. 24
      src/php/tests/interop/interop_client.php
  41. 14
      src/python/src/grpc/_adapter/_tag.h
  42. 7
      templates/Makefile.template
  43. 4
      test/build/systemtap.c
  44. 2
      test/core/end2end/gen_build_json.py
  45. 210
      test/core/end2end/tests/max_message_length.c
  46. 223
      test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
  47. 24
      test/cpp/end2end/end2end_test.cc
  48. 133
      test/cpp/qps/client_async.cc
  49. 9
      tools/dockerfile/grpc_php/Dockerfile
  50. 2
      tools/gce_setup/cloud_prod_runner.sh
  51. 31
      tools/gce_setup/grpc_docker.sh
  52. 4
      tools/gce_setup/shared_startup_funcs.sh
  53. 11
      tools/run_tests/run_sanity.sh
  54. 198
      tools/run_tests/tests.json
  55. 66
      vsprojects/Grpc.mak
  56. 13
      vsprojects/openssl.props
  57. 13
      vsprojects/zlib.props

@ -25,7 +25,6 @@ env:
- CONFIG=opt TEST=python
- CONFIG=opt TEST=csharp
- USE_GCC=4.4 CONFIG=opt TEST=build
- USE_GCC=4.5 CONFIG=opt TEST=build
script:
- rvm use $RUBY_VERSION
- gem install bundler

File diff suppressed because one or more lines are too long

@ -4,7 +4,7 @@ Pod::Spec.new do |s|
s.summary = 'Generic gRPC client library for iOS'
s.homepage = 'https://www.grpc.io'
s.license = 'New BSD'
s.authors = { 'Jorge Canizales' => 'jcanizales@google.com'
s.authors = { 'Jorge Canizales' => 'jcanizales@google.com',
'Michael Lumish' => 'mlumish@google.com' }
# s.source = { :git => 'https://github.com/grpc/grpc.git', :tag => 'release-0_5_0' }

@ -93,13 +93,17 @@
#endif
#ifndef GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM
#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream.h>
#define GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM \
::google::protobuf::io::ZeroCopyOutputStream
#define GRPC_CUSTOM_ZEROCOPYINPUTSTREAM \
::google::protobuf::io::ZeroCopyInputStream
#define GRPC_CUSTOM_CODEDINPUTSTREAM \
::google::protobuf::io::CodedInputStream
#endif
#ifdef GRPC_CXX0X_NO_NULLPTR
#include <memory>
const class {
@ -126,6 +130,7 @@ typedef GRPC_CUSTOM_PROTOBUF_INT64 int64;
namespace io {
typedef GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM ZeroCopyOutputStream;
typedef GRPC_CUSTOM_ZEROCOPYINPUTSTREAM ZeroCopyInputStream;
typedef GRPC_CUSTOM_CODEDINPUTSTREAM CodedInputStream;
} // namespace io
} // namespace protobuf

@ -80,6 +80,10 @@ class CallOpBuffer : public CompletionQueueTag {
// Called by completion queue just prior to returning from Next() or Pluck()
bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
void set_max_message_size(int max_message_size) {
max_message_size_ = max_message_size;
}
bool got_message;
private:
@ -99,6 +103,7 @@ class CallOpBuffer : public CompletionQueueTag {
grpc::protobuf::Message* recv_message_;
ByteBuffer* recv_message_buffer_;
grpc_byte_buffer* recv_buf_;
int max_message_size_;
// Client send close
bool client_send_close_;
// Client recv status
@ -130,16 +135,21 @@ class Call GRPC_FINAL {
public:
/* call is owned by the caller */
Call(grpc_call* call, CallHook* call_hook_, CompletionQueue* cq);
Call(grpc_call* call, CallHook* call_hook_, CompletionQueue* cq,
int max_message_size);
void PerformOps(CallOpBuffer* buffer);
grpc_call* call() { return call_; }
CompletionQueue* cq() { return cq_; }
int max_message_size() { return max_message_size_; }
private:
CallHook* call_hook_;
CompletionQueue* cq_;
grpc_call* call_;
int max_message_size_;
};
} // namespace grpc

@ -79,7 +79,8 @@ class Server GRPC_FINAL : public GrpcLibrary,
class AsyncRequest;
// ServerBuilder use only
Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned);
Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
int max_message_size);
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance.
bool RegisterService(RpcService* service);
@ -106,6 +107,8 @@ class Server GRPC_FINAL : public GrpcLibrary,
ServerAsyncStreamingInterface* stream,
CompletionQueue* cq, void* tag);
const int max_message_size_;
// Completion queue.
CompletionQueue cq_;
@ -126,7 +129,7 @@ class Server GRPC_FINAL : public GrpcLibrary,
// Whether the thread pool is created and owned by the server.
bool thread_pool_owned_;
private:
Server() : server_(NULL) { abort(); }
Server() : max_message_size_(-1), server_(NULL) { abort(); }
};
} // namespace grpc

@ -68,6 +68,11 @@ class ServerBuilder {
// Register a generic service.
void RegisterAsyncGenericService(AsyncGenericService* service);
// Set max message size in bytes.
void SetMaxMessageSize(int max_message_size) {
max_message_size_ = max_message_size;
}
// Add a listening port. Can be called multiple times.
void AddListeningPort(const grpc::string& addr,
std::shared_ptr<ServerCredentials> creds,
@ -87,6 +92,7 @@ class ServerBuilder {
int* selected_port;
};
int max_message_size_;
std::vector<RpcService*> services_;
std::vector<AsynchronousService*> async_services_;
std::vector<Port> ports_;

@ -111,6 +111,9 @@ typedef struct {
#define GRPC_ARG_MAX_CONCURRENT_STREAMS "grpc.max_concurrent_streams"
/* Maximum message length that the channel can receive */
#define GRPC_ARG_MAX_MESSAGE_LENGTH "grpc.max_message_length"
/* Initial sequence number for http2 transports */
#define GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER \
"grpc.http2.initial_sequence_number"
/* Result of a grpc call. If the caller satisfies the prerequisites of a
particular operation, the grpc_call_error returned will be GRPC_CALL_OK.

@ -1065,9 +1065,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
" new ::grpc::RpcMethodHandler< $ns$$Service$::Service, "
"$Request$, "
"$Response$>(\n"
" std::function< ::grpc::Status($ns$$Service$::Service*, "
"::grpc::ServerContext*, const $Request$*, $Response$*)>("
"&$ns$$Service$::Service::$Method$), this),\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this),\n"
" new $Request$, new $Response$));\n");
} else if (ClientOnlyStreaming(method)) {
printer->Print(
@ -1077,10 +1075,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
" ::grpc::RpcMethod::CLIENT_STREAMING,\n"
" new ::grpc::ClientStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::function< ::grpc::Status($ns$$Service$::Service*, "
"::grpc::ServerContext*, "
"::grpc::ServerReader< $Request$>*, $Response$*)>("
"&$ns$$Service$::Service::$Method$), this),\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this),\n"
" new $Request$, new $Response$));\n");
} else if (ServerOnlyStreaming(method)) {
printer->Print(
@ -1090,10 +1085,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
" ::grpc::RpcMethod::SERVER_STREAMING,\n"
" new ::grpc::ServerStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::function< ::grpc::Status($ns$$Service$::Service*, "
"::grpc::ServerContext*, "
"const $Request$*, ::grpc::ServerWriter< $Response$>*)>("
"&$ns$$Service$::Service::$Method$), this),\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this),\n"
" new $Request$, new $Response$));\n");
} else if (BidiStreaming(method)) {
printer->Print(
@ -1103,10 +1095,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
" ::grpc::RpcMethod::BIDI_STREAMING,\n"
" new ::grpc::BidiStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::function< ::grpc::Status($ns$$Service$::Service*, "
"::grpc::ServerContext*, "
"::grpc::ServerReaderWriter< $Response$, $Request$>*)>("
"&$ns$$Service$::Service::$Method$), this),\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this),\n"
" new $Request$, new $Response$));\n");
}
}

@ -1,141 +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.
*
*/
#include "src/core/channel/channel_stack.h"
#include <stdarg.h>
#include <stdio.h>
#include <string.h>
#include "src/core/support/string.h"
#include <grpc/support/alloc.h>
#include <grpc/support/useful.h>
static void put_metadata(gpr_strvec *b, grpc_mdelem *md) {
gpr_strvec_add(b, gpr_strdup(" key="));
gpr_strvec_add(
b, gpr_hexdump((char *)GPR_SLICE_START_PTR(md->key->slice),
GPR_SLICE_LENGTH(md->key->slice), GPR_HEXDUMP_PLAINTEXT));
gpr_strvec_add(b, gpr_strdup(" value="));
gpr_strvec_add(b, gpr_hexdump((char *)GPR_SLICE_START_PTR(md->value->slice),
GPR_SLICE_LENGTH(md->value->slice),
GPR_HEXDUMP_PLAINTEXT));
}
static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) {
grpc_linked_mdelem *m;
for (m = md.list.head; m != NULL; m = m->next) {
put_metadata(b, m->md);
}
if (gpr_time_cmp(md.deadline, gpr_inf_future) != 0) {
char *tmp;
gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec,
md.deadline.tv_nsec);
gpr_strvec_add(b, tmp);
}
}
char *grpc_call_op_string(grpc_call_op *op) {
char *tmp;
char *out;
gpr_strvec b;
gpr_strvec_init(&b);
switch (op->dir) {
case GRPC_CALL_DOWN:
gpr_strvec_add(&b, gpr_strdup(">"));
break;
case GRPC_CALL_UP:
gpr_strvec_add(&b, gpr_strdup("<"));
break;
}
switch (op->type) {
case GRPC_SEND_METADATA:
gpr_strvec_add(&b, gpr_strdup("SEND_METADATA"));
put_metadata_list(&b, op->data.metadata);
break;
case GRPC_SEND_MESSAGE:
gpr_strvec_add(&b, gpr_strdup("SEND_MESSAGE"));
break;
case GRPC_SEND_PREFORMATTED_MESSAGE:
gpr_strvec_add(&b, gpr_strdup("SEND_PREFORMATTED_MESSAGE"));
break;
case GRPC_SEND_FINISH:
gpr_strvec_add(&b, gpr_strdup("SEND_FINISH"));
break;
case GRPC_REQUEST_DATA:
gpr_strvec_add(&b, gpr_strdup("REQUEST_DATA"));
break;
case GRPC_RECV_METADATA:
gpr_strvec_add(&b, gpr_strdup("RECV_METADATA"));
put_metadata_list(&b, op->data.metadata);
break;
case GRPC_RECV_MESSAGE:
gpr_strvec_add(&b, gpr_strdup("RECV_MESSAGE"));
break;
case GRPC_RECV_HALF_CLOSE:
gpr_strvec_add(&b, gpr_strdup("RECV_HALF_CLOSE"));
break;
case GRPC_RECV_FINISH:
gpr_strvec_add(&b, gpr_strdup("RECV_FINISH"));
break;
case GRPC_RECV_SYNTHETIC_STATUS:
gpr_asprintf(&tmp, "RECV_SYNTHETIC_STATUS status=%d message='%s'",
op->data.synthetic_status.status,
op->data.synthetic_status.message);
gpr_strvec_add(&b, tmp);
break;
case GRPC_CANCEL_OP:
gpr_strvec_add(&b, gpr_strdup("CANCEL_OP"));
break;
}
gpr_asprintf(&tmp, " flags=0x%08x", op->flags);
gpr_strvec_add(&b, tmp);
if (op->bind_pollset) {
gpr_strvec_add(&b, gpr_strdup("bind_pollset"));
}
out = gpr_strvec_flatten(&b, NULL);
gpr_strvec_destroy(&b);
return out;
}
void grpc_call_log_op(char *file, int line, gpr_log_severity severity,
grpc_call_element *elem, grpc_call_op *op) {
char *str = grpc_call_op_string(op);
gpr_log(file, line, severity, "OP[%s:%p]: %s", elem->filter->name, elem, str);
gpr_free(str);
}

@ -0,0 +1,44 @@
/*
*
* 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.
*
*/
#ifndef GRPC_INTERNAL_CORE_CHANNEL_CONTEXT_H
#define GRPC_INTERNAL_CORE_CHANNEL_CONTEXT_H
/* Call object context pointers */
typedef enum {
GRPC_CONTEXT_SECURITY = 0,
GRPC_CONTEXT_TRACING,
GRPC_CONTEXT_COUNT
} grpc_context_index;
#endif

@ -50,7 +50,7 @@ static void create_sockets(SOCKET sv[2]) {
SOCKET lst_sock = INVALID_SOCKET;
SOCKET cli_sock = INVALID_SOCKET;
SOCKADDR_IN addr;
int addr_len;
int addr_len = sizeof(addr);
lst_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
GPR_ASSERT(lst_sock != INVALID_SOCKET);

@ -172,9 +172,15 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) {
}
void grpc_iocp_socket_orphan(grpc_winsocket *socket) {
GPR_ASSERT(!socket->orphan);
gpr_atm_full_fetch_add(&g_orphans, 1);
socket->orphan = 1;
}
/* Calling notify_on_read or write means either of two things:
-) The IOCP already completed in the background, and we need to call
the callback now.
-) The IOCP hasn't completed yet, and we're queuing it for later. */
static void socket_notify_on_iocp(grpc_winsocket *socket,
void(*cb)(void *, int), void *opaque,
grpc_winsocket_callback_info *info) {

@ -43,6 +43,10 @@
#include "src/core/iomgr/iocp_windows.h"
#include "src/core/iomgr/iomgr.h"
/* Windows' io manager is going to be fully designed using IO completion
ports. All of what we're doing here is basically make sure that
Windows sockets are initialized in and out. */
static void winsock_init(void) {
WSADATA wsaData;
int status = WSAStartup(MAKEWORD(2, 0), &wsaData);

@ -36,6 +36,9 @@
#include <grpc/support/sync.h>
/* There isn't really any such thing as a pollset under Windows, due to the
nature of the IO completion ports. */
struct grpc_kick_fd_info;
typedef struct grpc_pollset_kick_state {

@ -41,6 +41,11 @@
#include "src/core/iomgr/iomgr_internal.h"
#include "src/core/iomgr/pollset_windows.h"
/* There isn't really any such thing as a pollset under Windows, due to the
nature of the IO completion ports. We're still going to provide a minimal
set of features for the sake of the rest of grpc. But grpc_pollset_work
won't actually do any polling, and return as quickly as possible. */
void grpc_pollset_init(grpc_pollset *pollset) {
gpr_mu_init(&pollset->mu);
gpr_cv_init(&pollset->cv);

@ -40,10 +40,10 @@
#include "src/core/iomgr/pollset_kick.h"
#include "src/core/iomgr/socket_windows.h"
/* forward declare only in this file to avoid leaking impl details via
pollset.h; real users of grpc_fd should always include 'fd_posix.h' and not
use the struct tag */
struct grpc_fd;
/* There isn't really any such thing as a pollset under Windows, due to the
nature of the IO completion ports. A Windows "pollset" is merely a mutex
and a condition variable, as this is the minimal set of features we need
implemented for the rest of grpc. But we won't use them directly. */
typedef struct grpc_pollset {
gpr_mu mu;

@ -32,17 +32,18 @@
*/
#include <grpc/support/port_platform.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#ifdef GPR_WINSOCK_SOCKET
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/iomgr/iocp_windows.h"
#include "src/core/iomgr/iomgr.h"
#include "src/core/iomgr/iomgr_internal.h"
#include "src/core/iomgr/socket_windows.h"
#include "src/core/iomgr/pollset.h"
#include "src/core/iomgr/pollset_windows.h"
#include "src/core/iomgr/socket_windows.h"
grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
grpc_winsocket *r = gpr_malloc(sizeof(grpc_winsocket));
@ -54,26 +55,44 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
return r;
}
static void shutdown_op(grpc_winsocket_callback_info *info) {
if (!info->cb) return;
grpc_iomgr_add_delayed_callback(info->cb, info->opaque, 0);
}
/* Schedule a shutdown of the socket operations. Will call the pending
operations to abort them. We need to do that this way because of the
various callsites of that function, which happens to be in various
mutex hold states, and that'd be unsafe to call them directly. */
void grpc_winsocket_shutdown(grpc_winsocket *socket) {
shutdown_op(&socket->read_info);
shutdown_op(&socket->write_info);
gpr_mu_lock(&socket->state_mu);
if (socket->read_info.cb) {
grpc_iomgr_add_delayed_callback(socket->read_info.cb,
socket->read_info.opaque, 0);
}
if (socket->write_info.cb) {
grpc_iomgr_add_delayed_callback(socket->write_info.cb,
socket->write_info.opaque, 0);
}
gpr_mu_unlock(&socket->state_mu);
}
void grpc_winsocket_orphan(grpc_winsocket *socket) {
grpc_iocp_socket_orphan(socket);
socket->orphan = 1;
/* Abandons a socket. Either we're going to queue it up for garbage collecting
from the IO Completion Port thread, or destroy it immediately. Note that this
mechanisms assumes that we're either always waiting for an operation, or we
explicitely know that we don't. If there is a future case where we can have
an "idle" socket which is neither trying to read or write, we'd start leaking
both memory and sockets. */
void grpc_winsocket_orphan(grpc_winsocket *winsocket) {
SOCKET socket = winsocket->socket;
if (!winsocket->closed_early) {
grpc_iocp_socket_orphan(winsocket);
}
if (winsocket->closed_early) {
grpc_winsocket_destroy(winsocket);
}
closesocket(socket);
grpc_iomgr_unref();
closesocket(socket->socket);
}
void grpc_winsocket_destroy(grpc_winsocket *socket) {
gpr_mu_destroy(&socket->state_mu);
gpr_free(socket);
void grpc_winsocket_destroy(grpc_winsocket *winsocket) {
gpr_mu_destroy(&winsocket->state_mu);
gpr_free(winsocket);
}
#endif /* GPR_WINSOCK_SOCKET */

@ -39,21 +39,43 @@
#include <grpc/support/sync.h>
#include <grpc/support/atm.h>
/* This holds the data for an outstanding read or write on a socket.
The mutex to protect the concurrent access to that data is the one
inside the winsocket wrapper. */
typedef struct grpc_winsocket_callback_info {
/* This is supposed to be a WSAOVERLAPPED, but in order to get that
* definition, we need to include ws2tcpip.h, which needs to be included
* from the top, otherwise it'll clash with a previous inclusion of
* windows.h that in turns includes winsock.h. If anyone knows a way
* to do it properly, feel free to send a patch.
*/
definition, we need to include ws2tcpip.h, which needs to be included
from the top, otherwise it'll clash with a previous inclusion of
windows.h that in turns includes winsock.h. If anyone knows a way
to do it properly, feel free to send a patch. */
OVERLAPPED overlapped;
/* The callback information for the pending operation. May be empty if the
caller hasn't registered a callback yet. */
void(*cb)(void *opaque, int success);
void *opaque;
/* A boolean to describe if the IO Completion Port got a notification for
that operation. This will happen if the operation completed before the
called had time to register a callback. We could avoid that behavior
altogether by forcing the caller to always register its callback before
proceeding queue an operation, but it is frequent for an IO Completion
Port to trigger quickly. This way we avoid a context switch for calling
the callback. We also simplify the read / write operations to avoid having
to hold a mutex for a long amount of time. */
int has_pending_iocp;
/* The results of the overlapped operation. */
DWORD bytes_transfered;
int wsa_error;
} grpc_winsocket_callback_info;
/* This is a wrapper to a Windows socket. A socket can have one outstanding
read, and one outstanding write. Doing an asynchronous accept means waiting
for a read operation. Doing an asynchronous connect means waiting for a
write operation. These are completely abitrary ties between the operation
and the kind of event, because we can have one overlapped per pending
operation, whichever its nature is. So we could have more dedicated pending
operation callbacks for connect and listen. But given the scope of listen
and accept, we don't need to go to that extent and waste memory. Also, this
is closer to what happens in posix world. */
typedef struct grpc_winsocket {
SOCKET socket;
@ -62,16 +84,35 @@ typedef struct grpc_winsocket {
gpr_mu state_mu;
/* You can't add the same socket twice to the same IO Completion Port.
This prevents that. */
int added_to_iocp;
/* A boolean to indicate that the caller has abandonned that socket, but
there is a pending operation that the IO Completion Port will have to
wait for. The socket will be collected at that time. */
int orphan;
/* A boolean to indicate that the socket was already closed somehow, and
that no operation is going to be pending. Trying to abandon a socket in
that state won't result in an orphan, but will instead be destroyed
without further delay. We could avoid that boolean by adding one into
grpc_winsocket_callback_info describing that the operation is pending,
but that 1) waste memory more and 2) obfuscate the intent a bit more. */
int closed_early;
} grpc_winsocket;
/* Create a wrapped windows handle.
This takes ownership of closing it. */
/* Create a wrapped windows handle. This takes ownership of it, meaning that
it will be responsible for closing it. */
grpc_winsocket *grpc_winsocket_create(SOCKET socket);
/* Initiate an asynchronous shutdown of the socket. Will call off any pending
operation to cancel them. */
void grpc_winsocket_shutdown(grpc_winsocket *socket);
/* Abandon a socket. */
void grpc_winsocket_orphan(grpc_winsocket *socket);
/* Destroy a socket. Should only be called by the IO Completion Port thread,
or by grpc_winsocket_orphan if there's no pending operation. */
void grpc_winsocket_destroy(grpc_winsocket *socket);
#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */

@ -59,6 +59,7 @@ typedef struct {
gpr_timespec deadline;
grpc_alarm alarm;
int refs;
int aborted;
} async_connect;
static void async_connect_cleanup(async_connect *ac) {
@ -70,26 +71,31 @@ static void async_connect_cleanup(async_connect *ac) {
}
}
static void on_alarm(void *acp, int success) {
static void on_alarm(void *acp, int occured) {
async_connect *ac = acp;
gpr_mu_lock(&ac->mu);
if (ac->socket != NULL && success) {
/* If the alarm didn't occor, it got cancelled. */
if (ac->socket != NULL && occured) {
grpc_winsocket_shutdown(ac->socket);
}
async_connect_cleanup(ac);
}
static void on_connect(void *acp, int success) {
static void on_connect(void *acp, int from_iocp) {
async_connect *ac = acp;
SOCKET sock = ac->socket->socket;
grpc_endpoint *ep = NULL;
grpc_winsocket_callback_info *info = &ac->socket->write_info;
void(*cb)(void *arg, grpc_endpoint *tcp) = ac->cb;
void *cb_arg = ac->cb_arg;
int aborted;
grpc_alarm_cancel(&ac->alarm);
if (success) {
gpr_mu_lock(&ac->mu);
aborted = ac->aborted;
if (from_iocp) {
DWORD transfered_bytes = 0;
DWORD flags;
BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped,
@ -107,20 +113,40 @@ static void on_connect(void *acp, int success) {
}
} else {
gpr_log(GPR_ERROR, "on_connect is shutting down");
goto finish;
/* If the connection timeouts, we will still get a notification from
the IOCP whatever happens. So we're just going to flag that connection
as being in the process of being aborted, and wait for the IOCP. We
can't just orphan the socket now, because the IOCP might already have
gotten a successful connection, which is our worst-case scenario.
We need to call our callback now to respect the deadline. */
ac->aborted = 1;
gpr_mu_unlock(&ac->mu);
cb(cb_arg, NULL);
return;
}
abort();
finish:
gpr_mu_lock(&ac->mu);
if (!ep) {
/* If we don't have an endpoint, it means the connection failed,
so it doesn't matter if it aborted or failed. We need to orphan
that socket. */
if (!ep || aborted) {
/* If the connection failed, it means we won't get an IOCP notification,
so let's flag it as already closed. But if the connection was aborted,
while we still got an endpoint, we have to wait for the IOCP to collect
that socket. So let's properly flag that. */
ac->socket->closed_early = !ep;
grpc_winsocket_orphan(ac->socket);
}
async_connect_cleanup(ac);
cb(cb_arg, ep);
/* If the connection was aborted, the callback was already called when
the deadline was met. */
if (!aborted) cb(cb_arg, ep);
}
/* Tries to issue one async connection, then schedules both an IOCP
notification request for the connection, and one timeout alert. */
void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
void *arg, const struct sockaddr *addr,
int addr_len, gpr_timespec deadline) {
@ -156,6 +182,8 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
goto failure;
}
/* Grab the function pointer for ConnectEx for that specific socket.
It may change depending on the interface. */
status = WSAIoctl(sock, SIO_GET_EXTENSION_FUNCTION_POINTER,
&guid, sizeof(guid), &ConnectEx, sizeof(ConnectEx),
&ioctl_num_bytes, NULL, NULL);
@ -178,6 +206,8 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
info = &socket->write_info;
success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped);
/* It wouldn't be unusual to get a success immediately. But we'll still get
an IOCP notification, so let's ignore it. */
if (!success) {
int error = WSAGetLastError();
if (error != ERROR_IO_PENDING) {
@ -192,6 +222,7 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
ac->socket = socket;
gpr_mu_init(&ac->mu);
ac->refs = 2;
ac->aborted = 0;
grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now());
grpc_socket_notify_on_write(socket, on_connect, ac);
@ -202,6 +233,7 @@ failure:
gpr_log(GPR_ERROR, message, utf8_message);
gpr_free(utf8_message);
if (socket) {
socket->closed_early = 1;
grpc_winsocket_orphan(socket);
} else if (sock != INVALID_SOCKET) {
closesocket(sock);

@ -258,6 +258,8 @@ typedef struct {
grpc_endpoint base;
grpc_fd *em_fd;
int fd;
int iov_size; /* Number of slices to allocate per read attempt */
int finished_edge;
size_t slice_size;
gpr_refcount refcount;
@ -315,9 +317,7 @@ static void call_read_cb(grpc_tcp *tcp, gpr_slice *slices, size_t nslices,
#define INLINE_SLICE_BUFFER_SIZE 8
#define MAX_READ_IOVEC 4
static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
int iov_size = 1;
static void grpc_tcp_continue_read(grpc_tcp *tcp) {
gpr_slice static_read_slices[INLINE_SLICE_BUFFER_SIZE];
struct msghdr msg;
struct iovec iov[MAX_READ_IOVEC];
@ -327,88 +327,103 @@ static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success) {
gpr_slice *final_slices;
size_t final_nslices;
GPR_ASSERT(!tcp->finished_edge);
GRPC_TIMER_BEGIN(GRPC_PTAG_HANDLE_READ, 0);
slice_state_init(&read_state, static_read_slices, INLINE_SLICE_BUFFER_SIZE,
0);
if (!success) {
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
grpc_tcp_unref(tcp);
return;
allocated_bytes = slice_state_append_blocks_into_iovec(
&read_state, iov, tcp->iov_size, tcp->slice_size);
msg.msg_name = NULL;
msg.msg_namelen = 0;
msg.msg_iov = iov;
msg.msg_iovlen = tcp->iov_size;
msg.msg_control = NULL;
msg.msg_controllen = 0;
msg.msg_flags = 0;
GRPC_TIMER_BEGIN(GRPC_PTAG_RECVMSG, 0);
do {
read_bytes = recvmsg(tcp->fd, &msg, 0);
} while (read_bytes < 0 && errno == EINTR);
GRPC_TIMER_END(GRPC_PTAG_RECVMSG, 0);
if (read_bytes < allocated_bytes) {
/* TODO(klempner): Consider a second read first, in hopes of getting a
* quick EAGAIN and saving a bunch of allocations. */
slice_state_remove_last(&read_state, read_bytes < 0
? allocated_bytes
: allocated_bytes - read_bytes);
}
/* TODO(klempner): Limit the amount we read at once. */
for (;;) {
allocated_bytes = slice_state_append_blocks_into_iovec(
&read_state, iov, iov_size, tcp->slice_size);
msg.msg_name = NULL;
msg.msg_namelen = 0;
msg.msg_iov = iov;
msg.msg_iovlen = iov_size;
msg.msg_control = NULL;
msg.msg_controllen = 0;
msg.msg_flags = 0;
GRPC_TIMER_BEGIN(GRPC_PTAG_RECVMSG, 0);
do {
read_bytes = recvmsg(tcp->fd, &msg, 0);
} while (read_bytes < 0 && errno == EINTR);
GRPC_TIMER_END(GRPC_PTAG_RECVMSG, 0);
if (read_bytes < allocated_bytes) {
/* TODO(klempner): Consider a second read first, in hopes of getting a
* quick EAGAIN and saving a bunch of allocations. */
slice_state_remove_last(&read_state, read_bytes < 0
? allocated_bytes
: allocated_bytes - read_bytes);
}
if (read_bytes < 0) {
/* NB: After calling the user_cb a parallel call of the read handler may
* be running. */
if (errno == EAGAIN) {
if (slice_state_has_available(&read_state)) {
/* TODO(klempner): We should probably do the call into the application
without all this junk on the stack */
/* FIXME(klempner): Refcount properly */
slice_state_transfer_ownership(&read_state, &final_slices,
&final_nslices);
call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_OK);
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
} else {
/* Spurious read event, consume it here */
slice_state_destroy(&read_state);
grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure);
}
} else {
/* TODO(klempner): Log interesting errors */
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_ERROR);
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
if (read_bytes < 0) {
/* NB: After calling the user_cb a parallel call of the read handler may
* be running. */
if (errno == EAGAIN) {
if (tcp->iov_size > 1) {
tcp->iov_size /= 2;
}
return;
} else if (read_bytes == 0) {
/* 0 read size ==> end of stream */
if (slice_state_has_available(&read_state)) {
/* there were bytes already read: pass them up to the application */
/* TODO(klempner): We should probably do the call into the application
without all this junk on the stack */
/* FIXME(klempner): Refcount properly */
slice_state_transfer_ownership(&read_state, &final_slices,
&final_nslices);
call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_EOF);
tcp->finished_edge = 1;
call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_OK);
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
} else {
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_EOF);
/* We've consumed the edge, request a new one */
slice_state_destroy(&read_state);
grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure);
}
} else {
/* TODO(klempner): Log interesting errors */
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_ERROR);
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
return;
} else if (iov_size < MAX_READ_IOVEC) {
++iov_size;
}
} else if (read_bytes == 0) {
/* 0 read size ==> end of stream */
if (slice_state_has_available(&read_state)) {
/* there were bytes already read: pass them up to the application */
slice_state_transfer_ownership(&read_state, &final_slices,
&final_nslices);
call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_EOF);
} else {
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_EOF);
}
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
} else {
if (tcp->iov_size < MAX_READ_IOVEC) {
++tcp->iov_size;
}
GPR_ASSERT(slice_state_has_available(&read_state));
slice_state_transfer_ownership(&read_state, &final_slices,
&final_nslices);
call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_OK);
slice_state_destroy(&read_state);
grpc_tcp_unref(tcp);
}
GRPC_TIMER_END(GRPC_PTAG_HANDLE_READ, 0);
}
static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
GPR_ASSERT(!tcp->finished_edge);
if (!success) {
call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
grpc_tcp_unref(tcp);
} else {
grpc_tcp_continue_read(tcp);
}
}
static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
void *user_data) {
grpc_tcp *tcp = (grpc_tcp *)ep;
@ -416,7 +431,12 @@ static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
tcp->read_cb = cb;
tcp->read_user_data = user_data;
gpr_ref(&tcp->refcount);
grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure);
if (tcp->finished_edge) {
tcp->finished_edge = 0;
grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure);
} else {
grpc_iomgr_add_callback(grpc_tcp_handle_read, tcp);
}
}
#define MAX_WRITE_IOVEC 16
@ -554,6 +574,8 @@ grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size) {
tcp->read_user_data = NULL;
tcp->write_user_data = NULL;
tcp->slice_size = slice_size;
tcp->iov_size = 1;
tcp->finished_edge = 1;
slice_state_init(&tcp->write_state, NULL, 0, 0);
/* paired with unref in grpc_tcp_destroy */
gpr_ref_init(&tcp->refcount, 1);

@ -55,11 +55,17 @@
/* one listening port */
typedef struct server_port {
gpr_uint8 addresses[sizeof(struct sockaddr_in6) * 2 + 32];
/* This seemingly magic number comes from AcceptEx's documentation. each
address buffer needs to have at least 16 more bytes at their end. */
gpr_uint8 addresses[(sizeof(struct sockaddr_in6) + 16) * 2];
/* This will hold the socket for the next accept. */
SOCKET new_socket;
/* The listener winsocked. */
grpc_winsocket *socket;
grpc_tcp_server *server;
/* The cached AcceptEx for that port. */
LPFN_ACCEPTEX AcceptEx;
int shutting_down;
} server_port;
/* the overall server */
@ -79,6 +85,8 @@ struct grpc_tcp_server {
size_t port_capacity;
};
/* Public function. Allocates the proper data structures to hold a
grpc_tcp_server. */
grpc_tcp_server *grpc_tcp_server_create(void) {
grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server));
gpr_mu_init(&s->mu);
@ -92,24 +100,30 @@ grpc_tcp_server *grpc_tcp_server_create(void) {
return s;
}
/* Public function. Stops and destroys a grpc_tcp_server. */
void grpc_tcp_server_destroy(grpc_tcp_server *s,
void (*shutdown_done)(void *shutdown_done_arg),
void *shutdown_done_arg) {
size_t i;
gpr_mu_lock(&s->mu);
/* shutdown all fd's */
/* First, shutdown all fd's. This will queue abortion calls for all
of the pending accepts. */
for (i = 0; i < s->nports; i++) {
grpc_winsocket_shutdown(s->ports[i].socket);
server_port *sp = &s->ports[i];
grpc_winsocket_shutdown(sp->socket);
}
/* wait while that happens */
/* This happens asynchronously. Wait while that happens. */
while (s->active_ports) {
gpr_cv_wait(&s->cv, &s->mu, gpr_inf_future);
}
gpr_mu_unlock(&s->mu);
/* delete ALL the things */
/* Now that the accepts have been aborted, we can destroy the sockets.
The IOCP won't get notified on these, so we can flag them as already
closed by the system. */
for (i = 0; i < s->nports; i++) {
server_port *sp = &s->ports[i];
sp->socket->closed_early = 1;
grpc_winsocket_orphan(sp->socket);
}
gpr_free(s->ports);
@ -120,7 +134,7 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s,
}
}
/* Prepare a recently-created socket for listening. */
/* Prepare (bind) a recently-created socket for listening. */
static int prepare_socket(SOCKET sock, const struct sockaddr *addr,
int addr_len) {
struct sockaddr_storage sockname_temp;
@ -168,8 +182,11 @@ error:
return -1;
}
static void on_accept(void *arg, int success);
/* start_accept will reference that for the IOCP notification request. */
static void on_accept(void *arg, int from_iocp);
/* In order to do an async accept, we need to create a socket first which
will be the one assigned to the new incoming connection. */
static void start_accept(server_port *port) {
SOCKET sock = INVALID_SOCKET;
char *message;
@ -191,12 +208,13 @@ static void start_accept(server_port *port) {
goto failure;
}
/* TODO(jtattermusch): probably a race here, we regularly get use-after-free on server shutdown */
GPR_ASSERT(port->socket != (grpc_winsocket*)0xfeeefeee);
/* Start the "accept" asynchronously. */
success = port->AcceptEx(port->socket->socket, sock, port->addresses, 0,
addrlen, addrlen, &bytes_received,
&port->socket->read_info.overlapped);
/* It is possible to get an accept immediately without delay. However, we
will still get an IOCP notification for it. So let's just ignore it. */
if (!success) {
int error = WSAGetLastError();
if (error != ERROR_IO_PENDING) {
@ -205,6 +223,8 @@ static void start_accept(server_port *port) {
}
}
/* We're ready to do the accept. Calling grpc_socket_notify_on_read may
immediately process an accept that happened in the meantime. */
port->new_socket = sock;
grpc_socket_notify_on_read(port->socket, on_accept, port);
return;
@ -216,14 +236,30 @@ failure:
if (sock != INVALID_SOCKET) closesocket(sock);
}
/* event manager callback when reads are ready */
static void on_accept(void *arg, int success) {
/* Event manager callback when reads are ready. */
static void on_accept(void *arg, int from_iocp) {
server_port *sp = arg;
SOCKET sock = sp->new_socket;
grpc_winsocket_callback_info *info = &sp->socket->read_info;
grpc_endpoint *ep = NULL;
if (success) {
/* The shutdown sequence is done in two parts. This is the second
part here, acknowledging the IOCP notification, and doing nothing
else, especially not queuing a new accept. */
if (sp->shutting_down) {
GPR_ASSERT(from_iocp);
sp->shutting_down = 0;
gpr_mu_lock(&sp->server->mu);
if (0 == --sp->server->active_ports) {
gpr_cv_broadcast(&sp->server->cv);
}
gpr_mu_unlock(&sp->server->mu);
return;
}
if (from_iocp) {
/* The IOCP notified us of a completed operation. Let's grab the results,
and act accordingly. */
DWORD transfered_bytes = 0;
DWORD flags;
BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped,
@ -237,18 +273,31 @@ static void on_accept(void *arg, int success) {
ep = grpc_tcp_create(grpc_winsocket_create(sock));
}
} else {
closesocket(sock);
gpr_mu_lock(&sp->server->mu);
if (0 == --sp->server->active_ports) {
gpr_cv_broadcast(&sp->server->cv);
/* If we're not notified from the IOCP, it means we are asked to shutdown.
This will initiate that shutdown. Calling closesocket will trigger an
IOCP notification, that will call this function a second time, from
the IOCP thread. Of course, this only works if the socket was, in fact,
listening. If that's not the case, we'd wait indefinitely. That's a bit
of a degenerate case, but it can happen if you create a server, but
don't start it. So let's support that by recursing once. */
sp->shutting_down = 1;
sp->new_socket = INVALID_SOCKET;
if (sock != INVALID_SOCKET) {
closesocket(sock);
} else {
on_accept(sp, 1);
}
gpr_mu_unlock(&sp->server->mu);
return;
}
/* The only time we should call our callback, is where we successfully
managed to accept a connection, and created an endpoint. */
if (ep) sp->server->cb(sp->server->cb_arg, ep);
if (success) {
start_accept(sp);
}
/* As we were notified from the IOCP of one and exactly one accept,
the former socked we created has now either been destroy or assigned
to the new connection. We need to create a new one for the next
connection. */
start_accept(sp);
}
static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
@ -262,6 +311,8 @@ static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
if (sock == INVALID_SOCKET) return -1;
/* We need to grab the AcceptEx pointer for that port, as it may be
interface-dependent. We'll cache it to avoid doing that again. */
status =
WSAIoctl(sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid),
&AcceptEx, sizeof(AcceptEx), &ioctl_num_bytes, NULL, NULL);
@ -286,7 +337,9 @@ static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
sp = &s->ports[s->nports++];
sp->server = s;
sp->socket = grpc_winsocket_create(sock);
sp->shutting_down = 0;
sp->AcceptEx = AcceptEx;
sp->new_socket = INVALID_SOCKET;
GPR_ASSERT(sp->socket);
gpr_mu_unlock(&s->mu);
}

@ -76,8 +76,11 @@ int grpc_tcp_prepare_socket(SOCKET sock) {
}
typedef struct grpc_tcp {
/* This is our C++ class derivation emulation. */
grpc_endpoint base;
/* The one socket this endpoint is using. */
grpc_winsocket *socket;
/* Refcounting how many operations are in progress. */
gpr_refcount refcount;
grpc_endpoint_read_cb read_cb;
@ -90,6 +93,10 @@ typedef struct grpc_tcp {
gpr_slice_buffer write_slices;
int outstanding_write;
/* The IO Completion Port runs from another thread. We need some mechanism
to protect ourselves when requesting a shutdown. */
gpr_mu mu;
int shutting_down;
} grpc_tcp;
static void tcp_ref(grpc_tcp *tcp) {
@ -100,11 +107,13 @@ static void tcp_unref(grpc_tcp *tcp) {
if (gpr_unref(&tcp->refcount)) {
gpr_slice_buffer_destroy(&tcp->write_slices);
grpc_winsocket_orphan(tcp->socket);
gpr_mu_destroy(&tcp->mu);
gpr_free(tcp);
}
}
static void on_read(void *tcpp, int success) {
/* Asynchronous callback from the IOCP, or the background thread. */
static void on_read(void *tcpp, int from_iocp) {
grpc_tcp *tcp = (grpc_tcp *) tcpp;
grpc_winsocket *socket = tcp->socket;
gpr_slice sub;
@ -114,22 +123,32 @@ static void on_read(void *tcpp, int success) {
grpc_endpoint_read_cb cb = tcp->read_cb;
grpc_winsocket_callback_info *info = &socket->read_info;
void *opaque = tcp->read_user_data;
int do_abort = 0;
gpr_mu_lock(&tcp->mu);
if (!from_iocp || tcp->shutting_down) {
/* If we are here with from_iocp set to true, it means we got raced to
shutting down the endpoint. No actual abort callback will happen
though, so we're going to do it from here. */
do_abort = 1;
}
gpr_mu_unlock(&tcp->mu);
GPR_ASSERT(tcp->outstanding_read);
if (!success) {
if (do_abort) {
if (from_iocp) gpr_slice_unref(tcp->read_slice);
tcp_unref(tcp);
cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
return;
}
tcp->outstanding_read = 0;
GPR_ASSERT(tcp->outstanding_read);
if (socket->read_info.wsa_error != 0) {
char *utf8_message = gpr_format_message(info->wsa_error);
gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
gpr_free(utf8_message);
status = GRPC_ENDPOINT_CB_ERROR;
socket->closed_early = 1;
} else {
if (info->bytes_transfered != 0) {
sub = gpr_slice_sub(tcp->read_slice, 0, info->bytes_transfered);
@ -141,6 +160,9 @@ static void on_read(void *tcpp, int success) {
status = GRPC_ENDPOINT_CB_EOF;
}
}
tcp->outstanding_read = 0;
tcp_unref(tcp);
cb(opaque, slice, nslices, status);
}
@ -157,6 +179,7 @@ static void win_notify_on_read(grpc_endpoint *ep,
WSABUF buffer;
GPR_ASSERT(!tcp->outstanding_read);
GPR_ASSERT(!tcp->shutting_down);
tcp_ref(tcp);
tcp->outstanding_read = 1;
tcp->read_cb = cb;
@ -167,10 +190,12 @@ static void win_notify_on_read(grpc_endpoint *ep,
buffer.len = GPR_SLICE_LENGTH(tcp->read_slice);
buffer.buf = (char *)GPR_SLICE_START_PTR(tcp->read_slice);
/* First let's try a synchronous, non-blocking read. */
status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
NULL, NULL);
info->wsa_error = status == 0 ? 0 : WSAGetLastError();
/* Did we get data immediately ? Yay. */
if (info->wsa_error != WSAEWOULDBLOCK) {
info->bytes_transfered = bytes_read;
/* This might heavily recurse. */
@ -178,6 +203,7 @@ static void win_notify_on_read(grpc_endpoint *ep,
return;
}
/* Otherwise, let's retry, by queuing a read. */
memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED));
status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
&info->overlapped, NULL);
@ -191,30 +217,53 @@ static void win_notify_on_read(grpc_endpoint *ep,
if (error != WSA_IO_PENDING) {
char *utf8_message = gpr_format_message(WSAGetLastError());
__debugbreak();
gpr_log(GPR_ERROR, "WSARecv error: %s", utf8_message);
gpr_log(GPR_ERROR, "WSARecv error: %s - this means we're going to leak.",
utf8_message);
gpr_free(utf8_message);
/* would the IO completion port be called anyway... ? Let's assume not. */
/* I'm pretty sure this is a very bad situation there. Hence the log.
What will happen now is that the socket will neither wait for read
or write, unless the caller retry, which is unlikely, but I am not
sure if that's guaranteed. And there might also be a write pending.
This means that the future orphanage of that socket will be in limbo,
and we're going to leak it. I have no idea what could cause this
specific case however, aside from a parameter error from our call.
Normal read errors would actually happen during the overlapped
operation, which is the supported way to go for that. */
tcp->outstanding_read = 0;
tcp_unref(tcp);
cb(arg, NULL, 0, GRPC_ENDPOINT_CB_ERROR);
/* Per the comment above, I'm going to treat that case as a hard failure
for now, and leave the option to catch that and debug. */
__debugbreak();
return;
}
grpc_socket_notify_on_read(tcp->socket, on_read, tcp);
}
static void on_write(void *tcpp, int success) {
/* Asynchronous callback from the IOCP, or the background thread. */
static void on_write(void *tcpp, int from_iocp) {
grpc_tcp *tcp = (grpc_tcp *) tcpp;
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->write_info;
grpc_endpoint_cb_status status = GRPC_ENDPOINT_CB_OK;
grpc_endpoint_write_cb cb = tcp->write_cb;
void *opaque = tcp->write_user_data;
int do_abort = 0;
gpr_mu_lock(&tcp->mu);
if (!from_iocp || tcp->shutting_down) {
/* If we are here with from_iocp set to true, it means we got raced to
shutting down the endpoint. No actual abort callback will happen
though, so we're going to do it from here. */
do_abort = 1;
}
gpr_mu_unlock(&tcp->mu);
GPR_ASSERT(tcp->outstanding_write);
if (!success) {
if (do_abort) {
if (from_iocp) gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp_unref(tcp);
cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
return;
@ -225,6 +274,7 @@ static void on_write(void *tcpp, int success) {
gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
gpr_free(utf8_message);
status = GRPC_ENDPOINT_CB_ERROR;
tcp->socket->closed_early = 1;
} else {
GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length);
}
@ -236,6 +286,7 @@ static void on_write(void *tcpp, int success) {
cb(opaque, status);
}
/* Initiates a write. */
static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
gpr_slice *slices, size_t nslices,
grpc_endpoint_write_cb cb,
@ -251,11 +302,13 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
WSABUF *buffers = local_buffers;
GPR_ASSERT(!tcp->outstanding_write);
GPR_ASSERT(!tcp->shutting_down);
tcp_ref(tcp);
tcp->outstanding_write = 1;
tcp->write_cb = cb;
tcp->write_user_data = arg;
gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);
if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) {
@ -268,10 +321,14 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]);
}
/* First, let's try a synchronous, non-blocking write. */
status = WSASend(socket->socket, buffers, tcp->write_slices.count,
&bytes_sent, 0, NULL, NULL);
info->wsa_error = status == 0 ? 0 : WSAGetLastError();
/* We would kind of expect to get a WSAEWOULDBLOCK here, especially on a busy
connection that has its send queue filled up. But if we don't, then we can
avoid doing an async write operation at all. */
if (info->wsa_error != WSAEWOULDBLOCK) {
grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR;
if (status == 0) {
@ -289,25 +346,42 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
return ret;
}
/* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same
operation, this time asynchronously. */
memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED));
status = WSASend(socket->socket, buffers, tcp->write_slices.count,
&bytes_sent, 0, &socket->write_info.overlapped, NULL);
if (allocated) gpr_free(allocated);
/* It is possible the operation completed then. But we'd still get an IOCP
notification. So let's ignore it and wait for the IOCP. */
if (status != 0) {
int error = WSAGetLastError();
if (error != WSA_IO_PENDING) {
char *utf8_message = gpr_format_message(WSAGetLastError());
__debugbreak();
gpr_log(GPR_ERROR, "WSASend error: %s", utf8_message);
gpr_log(GPR_ERROR, "WSASend error: %s - this means we're going to leak.",
utf8_message);
gpr_free(utf8_message);
/* would the IO completion port be called anyway ? Let's assume not. */
/* I'm pretty sure this is a very bad situation there. Hence the log.
What will happen now is that the socket will neither wait for read
or write, unless the caller retry, which is unlikely, but I am not
sure if that's guaranteed. And there might also be a read pending.
This means that the future orphanage of that socket will be in limbo,
and we're going to leak it. I have no idea what could cause this
specific case however, aside from a parameter error from our call.
Normal read errors would actually happen during the overlapped
operation, which is the supported way to go for that. */
tcp->outstanding_write = 0;
tcp_unref(tcp);
/* Per the comment above, I'm going to treat that case as a hard failure
for now, and leave the option to catch that and debug. */
__debugbreak();
return GRPC_ENDPOINT_WRITE_ERROR;
}
}
/* As all is now setup, we can now ask for the IOCP notification. It may
trigger the callback immediately however, but no matter. */
grpc_socket_notify_on_write(socket, on_write, tcp);
return GRPC_ENDPOINT_WRITE_PENDING;
}
@ -317,9 +391,20 @@ static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) {
grpc_iocp_add_socket(tcp->socket);
}
/* Initiates a shutdown of the TCP endpoint. This will queue abort callbacks
for the potential read and write operations. It is up to the caller to
guarantee this isn't called in parallel to a read or write request, so
we're not going to protect against these. However the IO Completion Port
callback will happen from another thread, so we need to protect against
concurrent access of the data structure in that regard. */
static void win_shutdown(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *) ep;
gpr_mu_lock(&tcp->mu);
/* At that point, what may happen is that we're already inside the IOCP
callback. See the comments in on_read and on_write. */
tcp->shutting_down = 1;
grpc_winsocket_shutdown(tcp->socket);
gpr_mu_unlock(&tcp->mu);
}
static void win_destroy(grpc_endpoint *ep) {
@ -336,6 +421,7 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket) {
memset(tcp, 0, sizeof(grpc_tcp));
tcp->base.vtable = &vtable;
tcp->socket = socket;
gpr_mu_init(&tcp->mu);
gpr_slice_buffer_init(&tcp->write_slices);
gpr_ref_init(&tcp->refcount, 1);
return &tcp->base;

@ -45,108 +45,78 @@
#include <grpc/support/thd.h>
#include <stdio.h>
typedef enum { BEGIN = '{', END = '}', MARK = '.' } marker_type;
typedef struct grpc_timer_entry {
grpc_precise_clock tm;
gpr_thd_id thd;
int tag;
marker_type type;
void* id;
const char* file;
int line;
} grpc_timer_entry;
struct grpc_timers_log {
gpr_mu mu;
grpc_timer_entry* log;
int num_entries;
int capacity;
int capacity_limit;
FILE* fp;
};
grpc_timers_log* grpc_timers_log_global = NULL;
static grpc_timers_log* grpc_timers_log_create(int capacity_limit, FILE* dump) {
grpc_timers_log* log = gpr_malloc(sizeof(*log));
/* TODO (vpai): Allow allocation below limit */
log->log = gpr_malloc(capacity_limit * sizeof(*log->log));
/* TODO (vpai): Improve concurrency, do per-thread logging? */
gpr_mu_init(&log->mu);
log->num_entries = 0;
log->capacity = log->capacity_limit = capacity_limit;
#define MAX_COUNT (1024 * 1024 / sizeof(grpc_timer_entry))
log->fp = dump;
static __thread grpc_timer_entry log[MAX_COUNT];
static __thread int count;
return log;
}
static void log_report_locked(grpc_timers_log* log) {
FILE* fp = log->fp;
static void log_report() {
int i;
for (i = 0; i < log->num_entries; i++) {
grpc_timer_entry* entry = &(log->log[i]);
fprintf(fp, "GRPC_LAT_PROF ");
grpc_precise_clock_print(&entry->tm, fp);
fprintf(fp, " %p %d %p %s %d\n", (void*)(gpr_intptr)entry->thd, entry->tag,
entry->id, entry->file, entry->line);
for (i = 0; i < count; i++) {
grpc_timer_entry* entry = &(log[i]);
printf("GRPC_LAT_PROF " GRPC_PRECISE_CLOCK_FORMAT " %p %c %d %p %s %d\n",
GRPC_PRECISE_CLOCK_PRINTF_ARGS(&entry->tm),
(void*)(gpr_intptr)gpr_thd_currentid(), entry->type, entry->tag,
entry->id, entry->file, entry->line);
}
/* Now clear out the log */
log->num_entries = 0;
}
static void grpc_timers_log_destroy(grpc_timers_log* log) {
gpr_mu_lock(&log->mu);
log_report_locked(log);
gpr_mu_unlock(&log->mu);
gpr_free(log->log);
gpr_mu_destroy(&log->mu);
gpr_free(log);
count = 0;
}
static void grpc_timers_log_add(grpc_timers_log* log, int tag, void* id,
static void grpc_timers_log_add(int tag, marker_type type, void* id,
const char* file, int line) {
grpc_timer_entry* entry;
/* TODO (vpai) : Improve concurrency */
gpr_mu_lock(&log->mu);
if (log->num_entries == log->capacity_limit) {
log_report_locked(log);
if (count == MAX_COUNT) {
log_report();
}
entry = &log->log[log->num_entries++];
entry = &log[count++];
grpc_precise_clock_now(&entry->tm);
entry->tag = tag;
entry->type = type;
entry->id = id;
entry->file = file;
entry->line = line;
entry->thd = gpr_thd_currentid();
gpr_mu_unlock(&log->mu);
}
/* Latency profiler API implementation. */
void grpc_timer_add_mark(int tag, void* id, const char* file, int line) {
grpc_timers_log_add(grpc_timers_log_global, tag, id, file, line);
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) {
grpc_timers_log_add(tag, MARK, id, file, line);
}
}
void grpc_timer_begin(int tag, void* id, const char *file, int line) {}
void grpc_timer_end(int tag, void* id, const char *file, int line) {}
/* Basic profiler specific API functions. */
void grpc_timers_global_init(void) {
grpc_timers_log_global = grpc_timers_log_create(100000, stdout);
void grpc_timer_begin(int tag, void* id, const char* file, int line) {
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) {
grpc_timers_log_add(tag, BEGIN, id, file, line);
}
}
void grpc_timers_global_destroy(void) {
grpc_timers_log_destroy(grpc_timers_log_global);
void grpc_timer_end(int tag, void* id, const char* file, int line) {
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) {
grpc_timers_log_add(tag, END, id, file, line);
}
}
/* Basic profiler specific API functions. */
void grpc_timers_global_init(void) {}
void grpc_timers_global_destroy(void) {}
#else /* !GRPC_BASIC_PROFILER */
void grpc_timers_global_init(void) {}

@ -42,15 +42,15 @@
#include "src/core/profiling/stap_probes.h"
/* Latency profiler API implementation. */
void grpc_timer_add_mark(int tag, void* id, const char *file, int line) {
void grpc_timer_add_mark(int tag, void* id, const char* file, int line) {
_STAP_ADD_MARK(tag);
}
void grpc_timer_begin(int tag, void* id, const char *file, int line) {
void grpc_timer_begin(int tag, void* id, const char* file, int line) {
_STAP_TIMING_NS_BEGIN(tag);
}
void grpc_timer_end(int tag, void* id, const char *file, int line) {
void grpc_timer_end(int tag, void* id, const char* file, int line) {
_STAP_TIMING_NS_END(tag);
}

@ -41,9 +41,9 @@ extern "C" {
void grpc_timers_global_init(void);
void grpc_timers_global_destroy(void);
void grpc_timer_add_mark(int tag, void* id, const char *file, int line);
void grpc_timer_begin(int tag, void* id, const char *file, int line);
void grpc_timer_end(int tag, void* id, const char *file, int line);
void grpc_timer_add_mark(int tag, void *id, const char *file, int line);
void grpc_timer_begin(int tag, void *id, const char *file, int line);
void grpc_timer_end(int tag, void *id, const char *file, int line);
enum grpc_profiling_tags {
/* Any GRPC_PTAG_* >= than the threshold won't generate any profiling mark. */
@ -60,11 +60,16 @@ enum grpc_profiling_tags {
GRPC_PTAG_POLL_FINISHED = 203 + GRPC_PTAG_IGNORE_THRESHOLD,
GRPC_PTAG_TCP_CB_WRITE = 204 + GRPC_PTAG_IGNORE_THRESHOLD,
GRPC_PTAG_TCP_WRITE = 205 + GRPC_PTAG_IGNORE_THRESHOLD,
GRPC_PTAG_CALL_ON_DONE_RECV = 206 + GRPC_PTAG_IGNORE_THRESHOLD,
/* C++ */
GRPC_PTAG_CPP_CALL_CREATED = 300 + GRPC_PTAG_IGNORE_THRESHOLD,
GRPC_PTAG_CPP_PERFORM_OPS = 301 + GRPC_PTAG_IGNORE_THRESHOLD,
/* Transports */
GRPC_PTAG_HTTP2_UNLOCK = 401 + GRPC_PTAG_IGNORE_THRESHOLD,
GRPC_PTAG_HTTP2_UNLOCK_CLEANUP = 402 + GRPC_PTAG_IGNORE_THRESHOLD,
/* > 1024 Unassigned reserved. For any miscellaneous use.
* Use addition to generate tags from this base or take advantage of the 10
* zero'd bits for OR-ing. */
@ -74,13 +79,16 @@ enum grpc_profiling_tags {
#if !(defined(GRPC_STAP_PROFILER) + defined(GRPC_BASIC_PROFILER))
/* No profiling. No-op all the things. */
#define GRPC_TIMER_MARK(tag, id) \
do {} while(0)
do { \
} while (0)
#define GRPC_TIMER_BEGIN(tag, id) \
do {} while(0)
do { \
} while (0)
#define GRPC_TIMER_END(tag, id) \
do {} while(0)
do { \
} while (0)
#else /* at least one profiler requested... */
/* ... hopefully only one. */
@ -94,14 +102,14 @@ enum grpc_profiling_tags {
grpc_timer_add_mark(tag, ((void *)(gpr_intptr)(id)), __FILE__, __LINE__); \
}
#define GRPC_TIMER_BEGIN(tag, id) \
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) { \
grpc_timer_begin(tag, ((void *)(gpr_intptr)(id)), __FILE__, __LINE__); \
#define GRPC_TIMER_BEGIN(tag, id) \
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) { \
grpc_timer_begin(tag, ((void *)(gpr_intptr)(id)), __FILE__, __LINE__); \
}
#define GRPC_TIMER_END(tag, id) \
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) { \
grpc_timer_end(tag, ((void *)(gpr_intptr)(id)), __FILE__, __LINE__); \
#define GRPC_TIMER_END(tag, id) \
if (tag < GRPC_PTAG_IGNORE_THRESHOLD) { \
grpc_timer_end(tag, ((void *)(gpr_intptr)(id)), __FILE__, __LINE__); \
}
#ifdef GRPC_STAP_PROFILER
@ -109,9 +117,7 @@ enum grpc_profiling_tags {
#endif /* GRPC_STAP_PROFILER */
#ifdef GRPC_BASIC_PROFILER
typedef struct grpc_timers_log grpc_timers_log;
extern grpc_timers_log *grpc_timers_log_global;
/* Empty placeholder for now. */
#endif /* GRPC_BASIC_PROFILER */
#endif /* at least one profiler requested. */

@ -34,20 +34,59 @@
#ifndef GRPC_CORE_PROFILING_TIMERS_PRECISECLOCK_H
#define GRPC_CORE_PROFILING_TIMERS_PRECISECLOCK_H
#include <grpc/support/sync.h>
#include <grpc/support/time.h>
#include <stdio.h>
typedef struct grpc_precise_clock grpc_precise_clock;
#ifdef GRPC_TIMERS_RDTSC
#error RDTSC timers not currently supported
typedef long long int grpc_precise_clock;
#if defined(__i386__)
static void grpc_precise_clock_now(grpc_precise_clock *clk) {
grpc_precise_clock ret;
__asm__ volatile("rdtsc" : "=A"(ret));
*clk = ret;
}
// ----------------------------------------------------------------
#elif defined(__x86_64__) || defined(__amd64__)
static void grpc_precise_clock_now(grpc_precise_clock *clk) {
unsigned long long low, high;
__asm__ volatile("rdtsc" : "=a"(low), "=d"(high));
*clk = (high << 32) | low;
}
#endif
static gpr_once precise_clock_init = GPR_ONCE_INIT;
static double cycles_per_second = 0.0;
static void grpc_precise_clock_init() {
time_t start = time(NULL);
grpc_precise_clock start_time;
grpc_precise_clock end_time;
while (time(NULL) == start)
;
grpc_precise_clock_now(&start_time);
while (time(NULL) == start + 1)
;
grpc_precise_clock_now(&end_time);
cycles_per_second = end_time - start_time;
}
static double grpc_precise_clock_scaling_factor() {
gpr_once_init(&precise_clock_init, grpc_precise_clock_init);
return 1e6 / cycles_per_second;
}
#define GRPC_PRECISE_CLOCK_FORMAT "%f"
#define GRPC_PRECISE_CLOCK_PRINTF_ARGS(clk) \
(*(clk)*grpc_precise_clock_scaling_factor())
#else
typedef struct grpc_precise_clock grpc_precise_clock;
struct grpc_precise_clock {
gpr_timespec clock;
};
static void grpc_precise_clock_now(grpc_precise_clock* clk) {
clk->clock = gpr_now();
}
#define GRPC_PRECISE_CLOCK_FORMAT "%ld.%09d"
#define GRPC_PRECISE_CLOCK_PRINTF_ARGS(clk) \
(clk)->clock.tv_sec, (clk)->clock.tv_nsec
static void grpc_precise_clock_print(const grpc_precise_clock* clk, FILE* fp) {
fprintf(fp, "%ld.%09d", clk->clock.tv_sec, clk->clock.tv_nsec);
}

@ -43,8 +43,6 @@ unsigned gpr_cpu_num_cores(void) {
return si.dwNumberOfProcessors;
}
unsigned gpr_cpu_current_cpu(void) {
return GetCurrentProcessorNumber();
}
unsigned gpr_cpu_current_cpu(void) { return GetCurrentProcessorNumber(); }
#endif /* GPR_WIN32 */

@ -34,6 +34,7 @@
#include "src/core/surface/call.h"
#include "src/core/channel/channel_stack.h"
#include "src/core/iomgr/alarm.h"
#include "src/core/profiling/timers.h"
#include "src/core/support/string.h"
#include "src/core/surface/byte_buffer_queue.h"
#include "src/core/surface/channel.h"
@ -204,6 +205,9 @@ struct grpc_call {
/* Received call statuses from various sources */
received_status status[STATUS_SOURCE_COUNT];
void *context[GRPC_CONTEXT_COUNT];
void (*destroy_context[GRPC_CONTEXT_COUNT])(void *);
/* Deadline alarm - if have_alarm is non-zero */
grpc_alarm alarm;
@ -246,6 +250,9 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op);
static void execute_op(grpc_call *call, grpc_transport_op *op);
static void recv_metadata(grpc_call *call, grpc_metadata_batch *metadata);
static void finish_read_ops(grpc_call *call);
static grpc_call_error cancel_with_status(
grpc_call *c, grpc_status_code status, const char *description,
gpr_uint8 locked);
grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
const void *server_transport_data,
@ -291,6 +298,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
initial_op.recv_state = &call->recv_state;
initial_op.on_done_recv = call_on_done_recv;
initial_op.recv_user_data = call;
initial_op.context = call->context;
call->receiving = 1;
GRPC_CALL_INTERNAL_REF(call, "receiving");
initial_op_ptr = &initial_op;
@ -343,6 +351,11 @@ static void destroy_call(void *call, int ignored_success) {
for (i = 0; i < c->send_initial_metadata_count; i++) {
grpc_mdelem_unref(c->send_initial_metadata[i].md);
}
for (i = 0; i < GRPC_CONTEXT_COUNT; i++) {
if (c->destroy_context[i]) {
c->destroy_context[i](c->context[i]);
}
}
grpc_sopb_destroy(&c->send_ops);
grpc_sopb_destroy(&c->recv_ops);
grpc_bbq_destroy(&c->incoming_queue);
@ -405,14 +418,14 @@ static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); }
static int need_more_data(grpc_call *call) {
return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) ||
is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) ||
(is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && grpc_bbq_empty(&call->incoming_queue)) ||
is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) ||
is_op_live(call, GRPC_IOREQ_RECV_STATUS) ||
is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) ||
(is_op_live(call, GRPC_IOREQ_RECV_CLOSE) &&
grpc_bbq_empty(&call->incoming_queue)) ||
(call->write_state == WRITE_STATE_INITIAL && !call->is_client &&
call->read_state != READ_STATE_STREAM_CLOSED);
call->read_state < READ_STATE_GOT_INITIAL_METADATA);
}
static void unlock(grpc_call *call) {
@ -627,7 +640,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
gpr_asprintf(
&message, "Message terminated early; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
}
@ -638,7 +651,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
&message,
"Maximum message length of %d exceeded by a message of length %d",
grpc_channel_get_max_message_length(call->channel), msg.length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
} else if (msg.length > 0) {
@ -658,9 +671,9 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
}
/* we have to be reading a message to know what to do here */
if (!call->reading_message) {
grpc_call_cancel_with_status(
cancel_with_status(
call, GRPC_STATUS_INVALID_ARGUMENT,
"Received payload data while not reading a message");
"Received payload data while not reading a message", 1);
return 0;
}
/* append the slice to the incoming buffer */
@ -671,7 +684,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
gpr_asprintf(
&message, "Receiving message overflow; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
} else if (call->incoming_message.length == call->incoming_message_length) {
@ -685,6 +698,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
static void call_on_done_recv(void *pc, int success) {
grpc_call *call = pc;
size_t i;
GRPC_TIMER_BEGIN(GRPC_PTAG_CALL_ON_DONE_RECV, 0);
lock(call);
call->receiving = 0;
if (success) {
@ -729,6 +743,7 @@ static void call_on_done_recv(void *pc, int success) {
unlock(call);
GRPC_CALL_INTERNAL_UNREF(call, "receiving", 0);
GRPC_TIMER_BEGIN(GRPC_PTAG_CALL_ON_DONE_RECV, 0);
}
static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count,
@ -996,6 +1011,12 @@ grpc_call_error grpc_call_cancel(grpc_call *call) {
grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
grpc_status_code status,
const char *description) {
return cancel_with_status(c, status, description, 0);
}
static grpc_call_error cancel_with_status(
grpc_call *c, grpc_status_code status, const char *description,
gpr_uint8 locked) {
grpc_transport_op op;
grpc_mdstr *details =
description ? grpc_mdstr_from_string(c->metadata_context, description)
@ -1003,10 +1024,14 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
memset(&op, 0, sizeof(op));
op.cancel_with_status = status;
lock(c);
if (locked == 0) {
lock(c);
}
set_status_code(c, STATUS_FROM_API_OVERRIDE, status);
set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
unlock(c);
if (locked == 0) {
unlock(c);
}
execute_op(c, &op);
@ -1016,6 +1041,7 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
static void execute_op(grpc_call *call, grpc_transport_op *op) {
grpc_call_element *elem;
elem = CALL_ELEM_FROM_CALL(call, 0);
op->context = call->context;
elem->filter->start_transport_op(elem, op);
}
@ -1027,8 +1053,8 @@ static void call_alarm(void *arg, int success) {
grpc_call *call = arg;
if (success) {
if (call->is_client) {
grpc_call_cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
"Deadline Exceeded");
cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
"Deadline Exceeded", 0);
} else {
grpc_call_cancel(call);
}
@ -1253,3 +1279,16 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_batch,
tag);
}
void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value,
void (*destroy)(void *value)) {
if (call->destroy_context[elem]) {
call->destroy_context[elem](value);
}
call->context[elem] = value;
call->destroy_context[elem] = destroy;
}
void *grpc_call_context_get(grpc_call *call, grpc_context_index elem) {
return call->context[elem];
}

@ -35,6 +35,7 @@
#define GRPC_INTERNAL_CORE_SURFACE_CALL_H
#include "src/core/channel/channel_stack.h"
#include "src/core/channel/context.h"
#include <grpc/grpc.h>
/* Primitive operation types - grpc_op's get rewritten into these */
@ -120,6 +121,13 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity,
grpc_call *call, const grpc_op *ops, size_t nops,
void *tag);
/* Set a context pointer.
No thread safety guarantees are made wrt this value. */
void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value,
void (*destroy)(void *value));
/* Get a context pointer. */
void *grpc_call_context_get(grpc_call *call, grpc_context_index elem);
#define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \
if (grpc_trace_batch) grpc_call_log_batch(sev, call, ops, nops, tag)

@ -59,6 +59,7 @@ void grpc_init(void) {
grpc_register_tracer("channel", &grpc_trace_channel);
grpc_register_tracer("surface", &grpc_surface_trace);
grpc_register_tracer("http", &grpc_http_trace);
grpc_register_tracer("flowctl", &grpc_flowctl_trace);
grpc_register_tracer("batch", &grpc_trace_batch);
grpc_security_pre_init();
grpc_iomgr_init();

@ -37,6 +37,7 @@
#include <stdio.h>
#include <string.h>
#include "src/core/profiling/timers.h"
#include "src/core/support/string.h"
#include "src/core/transport/chttp2/frame_data.h"
#include "src/core/transport/chttp2/frame_goaway.h"
@ -60,10 +61,13 @@
#define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024)
#define MAX_WINDOW 0x7fffffffu
#define MAX_CLIENT_STREAM_ID 0x7fffffffu
#define CLIENT_CONNECT_STRING "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
#define CLIENT_CONNECT_STRLEN 24
int grpc_http_trace = 0;
int grpc_flowctl_trace = 0;
typedef struct transport transport;
typedef struct stream stream;
@ -74,6 +78,12 @@ typedef struct stream stream;
else \
stmt
#define FLOWCTL_TRACE(t, obj, dir, id, delta) \
if (!grpc_flowctl_trace) \
; \
else \
flowctl_trace(t, #dir, obj->dir##_window, id, delta)
/* streams are kept in various linked lists depending on what things need to
happen to them... this enum labels each list */
typedef enum {
@ -382,6 +392,12 @@ static void add_to_pollset_locked(transport *t, grpc_pollset *pollset);
static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op);
static void add_metadata_batch(transport *t, stream *s);
static void flowctl_trace(transport *t, const char *flow, gpr_int32 window,
gpr_uint32 id, gpr_int32 delta) {
gpr_log(GPR_DEBUG, "HTTP:FLOW:%p:%d:%s: %d + %d = %d", t, id, flow, window,
delta, window + delta);
}
/*
* CONSTRUCTION/DESTRUCTION/REFCOUNTING
*/
@ -524,6 +540,19 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup,
push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS,
channel_args->args[i].value.integer);
}
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER)) {
if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
gpr_log(GPR_ERROR, "%s: must be an integer",
GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER);
} else if ((t->next_stream_id & 1) !=
(channel_args->args[i].value.integer & 1)) {
gpr_log(GPR_ERROR, "%s: low bit must be %d on %s",
GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER, t->next_stream_id & 1,
t->is_client ? "client" : "server");
} else {
t->next_stream_id = channel_args->args[i].value.integer;
}
}
}
}
@ -772,6 +801,8 @@ static void unlock(transport *t) {
grpc_stream_op_buffer nuke_now;
const grpc_transport_callbacks *cb = t->cb;
GRPC_TIMER_BEGIN(GRPC_PTAG_HTTP2_UNLOCK, 0);
grpc_sopb_init(&nuke_now);
if (t->nuke_later_sopb.nops) {
grpc_sopb_swap(&nuke_now, &t->nuke_later_sopb);
@ -820,6 +851,8 @@ static void unlock(transport *t) {
/* finally unlock */
gpr_mu_unlock(&t->mu);
GRPC_TIMER_MARK(GRPC_PTAG_HTTP2_UNLOCK_CLEANUP, 0);
/* perform some callbacks if necessary */
for (i = 0; i < num_goaways; i++) {
cb->goaway(t->cb_user_data, &t->base, goaways[i].status, goaways[i].debug);
@ -850,6 +883,8 @@ static void unlock(transport *t) {
grpc_sopb_destroy(&nuke_now);
gpr_free(goaways);
GRPC_TIMER_END(GRPC_PTAG_HTTP2_UNLOCK, 0);
}
/*
@ -896,6 +931,8 @@ static int prepare_write(transport *t) {
window_delta = grpc_chttp2_preencode(
s->outgoing_sopb->ops, &s->outgoing_sopb->nops,
GPR_MIN(t->outgoing_window, s->outgoing_window), &s->writing_sopb);
FLOWCTL_TRACE(t, t, outgoing, 0, -(gpr_int64)window_delta);
FLOWCTL_TRACE(t, s, outgoing, s->id, -(gpr_int64)window_delta);
t->outgoing_window -= window_delta;
s->outgoing_window -= window_delta;
@ -924,6 +961,7 @@ static int prepare_write(transport *t) {
if (!s->read_closed && window_delta) {
gpr_slice_buffer_add(
&t->outbuf, grpc_chttp2_window_update_create(s->id, window_delta));
FLOWCTL_TRACE(t, s, incoming, s->id, window_delta);
s->incoming_window += window_delta;
}
}
@ -933,6 +971,7 @@ static int prepare_write(transport *t) {
window_delta = t->connection_window_target - t->incoming_window;
gpr_slice_buffer_add(&t->outbuf,
grpc_chttp2_window_update_create(0, window_delta));
FLOWCTL_TRACE(t, t, incoming, 0, window_delta);
t->incoming_window += window_delta;
}
@ -1006,16 +1045,36 @@ static void perform_write(transport *t, grpc_endpoint *ep) {
}
}
static void add_goaway(transport *t, gpr_uint32 goaway_error, gpr_slice goaway_text) {
if (t->num_pending_goaways == t->cap_pending_goaways) {
t->cap_pending_goaways = GPR_MAX(1, t->cap_pending_goaways * 2);
t->pending_goaways =
gpr_realloc(t->pending_goaways,
sizeof(pending_goaway) * t->cap_pending_goaways);
}
t->pending_goaways[t->num_pending_goaways].status =
grpc_chttp2_http2_error_to_grpc_status(goaway_error);
t->pending_goaways[t->num_pending_goaways].debug = goaway_text;
t->num_pending_goaways++;
}
static void maybe_start_some_streams(transport *t) {
/* start streams where we have free stream ids and free concurrency */
while (
t->next_stream_id <= MAX_CLIENT_STREAM_ID &&
grpc_chttp2_stream_map_size(&t->stream_map) <
t->settings[PEER_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS]) {
stream *s = stream_list_remove_head(t, WAITING_FOR_CONCURRENCY);
if (!s) break;
if (!s) return;
IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Allocating new stream %p to id %d",
t->is_client ? "CLI" : "SVR", s, t->next_stream_id));
if (t->next_stream_id == MAX_CLIENT_STREAM_ID) {
add_goaway(t, GRPC_CHTTP2_NO_ERROR, gpr_slice_from_copied_string("Exceeded sequence number limit"));
}
GPR_ASSERT(s->id == 0);
s->id = t->next_stream_id;
t->next_stream_id += 2;
@ -1026,6 +1085,13 @@ static void maybe_start_some_streams(transport *t) {
grpc_chttp2_stream_map_add(&t->stream_map, s->id, s);
stream_list_join(t, s, WRITABLE);
}
/* cancel out streams that will never be started */
while (t->next_stream_id > MAX_CLIENT_STREAM_ID) {
stream *s = stream_list_remove_head(t, WAITING_FOR_CONCURRENCY);
if (!s) return;
cancel_stream(t, s, GRPC_STATUS_UNAVAILABLE, grpc_chttp2_grpc_status_to_http2_error(GRPC_STATUS_UNAVAILABLE), NULL, 0);
}
}
static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) {
@ -1259,6 +1325,8 @@ static grpc_chttp2_parse_error update_incoming_window(transport *t, stream *s) {
return GRPC_CHTTP2_CONNECTION_ERROR;
}
FLOWCTL_TRACE(t, t, incoming, 0, -(gpr_int64)t->incoming_frame_size);
FLOWCTL_TRACE(t, s, incoming, s->id, -(gpr_int64)t->incoming_frame_size);
t->incoming_window -= t->incoming_frame_size;
s->incoming_window -= t->incoming_frame_size;
@ -1581,16 +1649,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) {
grpc_chttp2_ping_create(1, t->simple_parsers.ping.opaque_8bytes));
}
if (st.goaway) {
if (t->num_pending_goaways == t->cap_pending_goaways) {
t->cap_pending_goaways = GPR_MAX(1, t->cap_pending_goaways * 2);
t->pending_goaways =
gpr_realloc(t->pending_goaways,
sizeof(pending_goaway) * t->cap_pending_goaways);
}
t->pending_goaways[t->num_pending_goaways].status =
grpc_chttp2_http2_error_to_grpc_status(st.goaway_error);
t->pending_goaways[t->num_pending_goaways].debug = st.goaway_text;
t->num_pending_goaways++;
add_goaway(t, st.goaway_error, st.goaway_text);
}
if (st.process_ping_reply) {
for (i = 0; i < t->ping_count; i++) {
@ -1608,6 +1667,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) {
for (i = 0; i < t->stream_map.count; i++) {
stream *s = (stream *)(t->stream_map.values[i]);
int was_window_empty = s->outgoing_window <= 0;
FLOWCTL_TRACE(t, s, outgoing, s->id, st.initial_window_update);
s->outgoing_window += st.initial_window_update;
if (was_window_empty && s->outgoing_window > 0 && s->outgoing_sopb &&
s->outgoing_sopb->nops > 0) {
@ -1626,6 +1686,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) {
GRPC_CHTTP2_FLOW_CONTROL_ERROR),
GRPC_CHTTP2_FLOW_CONTROL_ERROR, NULL, 1);
} else {
FLOWCTL_TRACE(t, s, outgoing, s->id, st.window_update);
s->outgoing_window += st.window_update;
/* if this window update makes outgoing ops writable again,
flag that */
@ -1640,6 +1701,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) {
if (!is_window_update_legal(st.window_update, t->outgoing_window)) {
drop_connection(t);
} else {
FLOWCTL_TRACE(t, t, outgoing, 0, st.window_update);
t->outgoing_window += st.window_update;
}
}
@ -1754,7 +1816,7 @@ static int process_read(transport *t, gpr_slice slice) {
/* fallthrough */
case DTS_FH_5:
GPR_ASSERT(cur < end);
t->incoming_stream_id = (((gpr_uint32)*cur) << 24) & 0x7f;
t->incoming_stream_id = (((gpr_uint32)*cur) & 0x7f) << 24;
if (++cur == end) {
t->deframe_state = DTS_FH_6;
return 1;

@ -38,6 +38,7 @@
#include "src/core/transport/transport.h"
extern int grpc_http_trace;
extern int grpc_flowctl_trace;
void grpc_create_chttp2_transport(grpc_transport_setup_callback setup,
void *arg,

@ -76,6 +76,9 @@ typedef struct grpc_transport_op {
grpc_status_code cancel_with_status;
grpc_mdstr *cancel_message;
/* Indexes correspond to grpc_context_index enum values */
void *const *context;
} grpc_transport_op;
/* Callbacks made from the transport to the upper layers of grpc. */

@ -55,6 +55,7 @@ CallOpBuffer::CallOpBuffer()
recv_message_(nullptr),
recv_message_buffer_(nullptr),
recv_buf_(nullptr),
max_message_size_(-1),
client_send_close_(false),
recv_trailing_metadata_(nullptr),
recv_status_(nullptr),
@ -311,7 +312,8 @@ bool CallOpBuffer::FinalizeResult(void** tag, bool* status) {
got_message = *status;
if (recv_message_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, 0);
*status = *status && DeserializeProto(recv_buf_, recv_message_);
*status = *status &&
DeserializeProto(recv_buf_, recv_message_, max_message_size_);
grpc_byte_buffer_destroy(recv_buf_);
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, 0);
} else {
@ -338,9 +340,19 @@ bool CallOpBuffer::FinalizeResult(void** tag, bool* status) {
}
Call::Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq)
: call_hook_(call_hook), cq_(cq), call_(call) {}
: call_hook_(call_hook), cq_(cq), call_(call), max_message_size_(-1) {}
Call::Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
int max_message_size)
: call_hook_(call_hook),
cq_(cq),
call_(call),
max_message_size_(max_message_size) {}
void Call::PerformOps(CallOpBuffer* buffer) {
if (max_message_size_ > 0) {
buffer->set_max_message_size(max_message_size_);
}
call_hook_->PerformOpsOnCall(buffer, this);
}

@ -158,10 +158,15 @@ bool SerializeProto(const grpc::protobuf::Message& msg, grpc_byte_buffer** bp) {
return msg.SerializeToZeroCopyStream(&writer);
}
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg) {
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg,
int max_message_size) {
if (!buffer) return false;
GrpcBufferReader reader(buffer);
return msg->ParseFromZeroCopyStream(&reader);
::grpc::protobuf::io::CodedInputStream decoder(&reader);
if (max_message_size > 0) {
decoder.SetTotalBytesLimit(max_message_size, max_message_size);
}
return msg->ParseFromCodedStream(&decoder) && decoder.ConsumedEntireMessage();
}
} // namespace grpc

@ -47,7 +47,8 @@ bool SerializeProto(const grpc::protobuf::Message& msg,
grpc_byte_buffer** buffer);
// The caller keeps ownership of buffer and msg.
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg);
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg,
int max_message_size);
} // namespace grpc

@ -100,7 +100,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
public:
explicit CallData(Server* server, SyncRequest* mrd)
: cq_(mrd->cq_),
call_(mrd->call_, server, &cq_),
call_(mrd->call_, server, &cq_, server->max_message_size_),
ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
mrd->request_metadata_.count),
has_request_payload_(mrd->has_request_payload_),
@ -126,8 +126,11 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
if (has_request_payload_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
req.reset(method_->AllocateRequestProto());
if (!DeserializeProto(request_payload_, req.get())) {
abort(); // for now
if (!DeserializeProto(request_payload_, req.get(),
call_.max_message_size())) {
// FIXME(yangg) deal with deserialization failure
cq_.Shutdown();
return;
}
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
}
@ -176,12 +179,27 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
grpc_completion_queue* cq_;
};
Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned)
: started_(false),
grpc_server* CreateServer(grpc_completion_queue* cq, int max_message_size) {
if (max_message_size > 0) {
grpc_arg arg;
arg.type = GRPC_ARG_INTEGER;
arg.key = const_cast<char*>(GRPC_ARG_MAX_MESSAGE_LENGTH);
arg.value.integer = max_message_size;
grpc_channel_args args = {1, &arg};
return grpc_server_create(cq, &args);
} else {
return grpc_server_create(cq, nullptr);
}
}
Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
int max_message_size)
: max_message_size_(max_message_size),
started_(false),
shutdown_(false),
num_running_cb_(0),
sync_methods_(new std::list<SyncRequest>),
server_(grpc_server_create(cq_.cq(), nullptr)),
server_(CreateServer(cq_.cq(), max_message_size)),
thread_pool_(thread_pool),
thread_pool_owned_(thread_pool_owned) {}
@ -220,7 +238,7 @@ bool Server::RegisterAsyncService(AsynchronousService* service) {
GPR_ASSERT(service->dispatch_impl_ == nullptr &&
"Can only register an asynchronous service against one server.");
service->dispatch_impl_ = this;
service->request_args_ = new void* [service->method_count_];
service->request_args_ = new void*[service->method_count_];
for (size_t i = 0; i < service->method_count_; ++i) {
void* tag =
grpc_server_register_method(server_, service->method_names_[i], nullptr,
@ -347,7 +365,8 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
if (*status && request_) {
if (payload_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, call_);
*status = DeserializeProto(payload_, request_);
*status =
DeserializeProto(payload_, request_, server_->max_message_size_);
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, call_);
} else {
*status = false;
@ -374,7 +393,7 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
}
ctx->call_ = call_;
ctx->cq_ = cq_;
Call call(call_, server_, cq_);
Call call(call_, server_, cq_, server_->max_message_size_);
if (orig_status && call_) {
ctx->BeginCompletionOp(&call);
}

@ -42,7 +42,7 @@
namespace grpc {
ServerBuilder::ServerBuilder()
: generic_service_(nullptr), thread_pool_(nullptr) {}
: max_message_size_(-1), generic_service_(nullptr), thread_pool_(nullptr) {}
void ServerBuilder::RegisterService(SynchronousService* service) {
services_.push_back(service->service());
@ -86,7 +86,8 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
thread_pool_ = new ThreadPool(cores);
thread_pool_owned = true;
}
std::unique_ptr<Server> server(new Server(thread_pool_, thread_pool_owned));
std::unique_ptr<Server> server(
new Server(thread_pool_, thread_pool_owned, max_message_size_));
for (auto service = services_.begin(); service != services_.end();
service++) {
if (!server->RegisterService(*service)) {

@ -125,6 +125,24 @@ function serviceAccountCreds($stub, $args) {
'invalid oauth scope returned');
}
/**
* Run the compute engine credentials auth test.
* Has not been run from gcloud as of 2015-05-05
* @param $stub Stub object that has service methods
* @param $args array command line args
*/
function computeEngineCreds($stub, $args) {
if (!array_key_exists('oauth_scope', $args)) {
throw new Exception('Missing oauth scope');
}
if (!array_key_exists('default_service_account', $args)) {
throw new Exception('Missing default_service_account');
}
$result = performLargeUnary($stub, $fillUsername=true, $fillOauthScope=true);
hardAssert($args['default_service_account'] == $result->getUsername(),
'invalid email returned');
}
/**
* Run the client_streaming test.
* Passes when run against the Node server as of 2015-04-30
@ -240,7 +258,8 @@ function cancelAfterFirstResponse($stub) {
}
$args = getopt('', array('server_host:', 'server_port:', 'test_case:',
'server_host_override:', 'oauth_scope:'));
'server_host_override:', 'oauth_scope:',
'default_service_account:'));
if (!array_key_exists('server_host', $args) ||
!array_key_exists('server_port', $args) ||
!array_key_exists('test_case', $args)) {
@ -301,6 +320,9 @@ switch ($args['test_case']) {
case 'service_account_creds':
serviceAccountCreds($stub, $args);
break;
case 'compute_engine_creds':
computeEngineCreds($stub, $args);
break;
default:
exit(1);
}

@ -44,14 +44,14 @@
replacement for its descriptive functionality until Python can move its whole
C and C adapter stack to more closely resemble the core batching API. */
typedef enum {
PYGRPC_SERVER_RPC_NEW = 0,
PYGRPC_INITIAL_METADATA = 1,
PYGRPC_READ = 2,
PYGRPC_WRITE_ACCEPTED = 3,
PYGRPC_FINISH_ACCEPTED = 4,
PYGRPC_SERVER_RPC_NEW = 0,
PYGRPC_INITIAL_METADATA = 1,
PYGRPC_READ = 2,
PYGRPC_WRITE_ACCEPTED = 3,
PYGRPC_FINISH_ACCEPTED = 4,
PYGRPC_CLIENT_METADATA_READ = 5,
PYGRPC_FINISHED_CLIENT = 6,
PYGRPC_FINISHED_SERVER = 7
PYGRPC_FINISHED_CLIENT = 6,
PYGRPC_FINISHED_SERVER = 7
} pygrpc_tag_type;
typedef struct {

@ -106,7 +106,7 @@ CC_basicprof = $(DEFAULT_CC)
CXX_basicprof = $(DEFAULT_CXX)
LD_basicprof = $(DEFAULT_CC)
LDXX_basicprof = $(DEFAULT_CXX)
CPPFLAGS_basicprof = -O2 -DGRPC_BASIC_PROFILER
CPPFLAGS_basicprof = -O2 -DGRPC_BASIC_PROFILER -DGRPC_TIMERS_RDTSC
LDFLAGS_basicprof =
DEFINES_basicprof = NDEBUG
@ -1191,11 +1191,14 @@ endif
lib_deps = ' $(ZLIB_DEP)'
mingw_libs = ''
mingw_lib_deps = ' $(ZLIB_DEP)'
if lib.language == 'c++':
lib_deps += ' $(PROTOBUF_DEP)'
mingw_lib_deps += ' $(PROTOBUF_DEP)'
for dep in lib.get('deps', []):
libs = libs + ' -l' + dep
lib_deps = lib_deps + ' $(LIBDIR)/$(CONFIG)/lib' + dep + '.$(SHARED_EXT)'
mingw_libs = mingw_libs + ' -l' + dep + '-imp'
mingw_lib_deps = mingw_lib_deps + '$(LIBDIR)/$(CONFIG)/' + dep + '.$(SHARED_EXT)'
mingw_lib_deps = mingw_lib_deps + ' $(LIBDIR)/$(CONFIG)/' + dep + '.$(SHARED_EXT)'
if lib.get('secure', 'check') == 'yes':
common = common + ' $(LDLIBS_SECURE) $(OPENSSL_MERGE_LIBS)'

@ -37,6 +37,4 @@
#error "_SYS_SDT_H not defined, despite <sys/sdt.h> being present."
#endif
int main() {
return 0;
}
int main() { return 0; }

@ -62,6 +62,7 @@ END2END_TESTS = {
'graceful_server_shutdown': True,
'invoke_large_request': False,
'max_concurrent_streams': True,
'max_message_length': True,
'no_op': True,
'ping_pong_streaming': True,
'request_response_with_binary_metadata_and_payload': True,
@ -71,6 +72,7 @@ END2END_TESTS = {
'request_with_payload': True,
'simple_delayed_request': True,
'simple_request': True,
'simple_request_with_high_initial_sequence_number': True,
'registered_call': True,
}

@ -0,0 +1,210 @@
/*
*
* 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.
*
*/
#include "test/core/end2end/end2end_tests.h"
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <grpc/byte_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "test/core/end2end/cq_verifier.h"
enum { TIMEOUT = 200000 };
static void *tag(gpr_intptr t) { return (void *)t; }
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
const char *test_name,
grpc_channel_args *client_args,
grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
f = config.create_fixture(client_args, server_args);
config.init_client(&f, client_args);
config.init_server(&f, server_args);
return f;
}
static gpr_timespec n_seconds_time(int n) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
}
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
grpc_event *ev;
grpc_completion_type type;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
GPR_ASSERT(ev);
type = ev->type;
grpc_event_finish(ev);
} while (type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown(f->server);
grpc_server_destroy(f->server);
f->server = NULL;
}
static void shutdown_client(grpc_end2end_test_fixture *f) {
if (!f->client) return;
grpc_channel_destroy(f->client);
f->client = NULL;
}
static void end_test(grpc_end2end_test_fixture *f) {
shutdown_server(f);
shutdown_client(f);
grpc_completion_queue_shutdown(f->server_cq);
drain_cq(f->server_cq);
grpc_completion_queue_destroy(f->server_cq);
grpc_completion_queue_shutdown(f->client_cq);
drain_cq(f->client_cq);
grpc_completion_queue_destroy(f->client_cq);
}
static void test_max_message_length(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f;
grpc_arg server_arg;
grpc_channel_args server_args;
grpc_call *c;
grpc_call *s;
cq_verifier *v_client;
cq_verifier *v_server;
grpc_op ops[6];
grpc_op *op;
gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
grpc_byte_buffer *request_payload =
grpc_byte_buffer_create(&request_payload_slice, 1);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_status_code status;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
server_arg.key = GRPC_ARG_MAX_MESSAGE_LENGTH;
server_arg.type = GRPC_ARG_INTEGER;
server_arg.value.integer = 5;
server_args.num_args = 1;
server_args.args = &server_arg;
f = begin_test(config, __FUNCTION__, NULL, &server_args);
v_client = cq_verifier_create(f.client_cq);
v_server = cq_verifier_create(f.server_cq);
c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
"foo.test.google.fr:1234", gpr_inf_future);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message = request_payload;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
cq_verify(v_server);
cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
GPR_ASSERT(0 == strcmp(details, "Cancelled"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234"));
GPR_ASSERT(was_cancelled == 1);
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(v_client);
cq_verifier_destroy(v_server);
end_test(&f);
config.tear_down_data(&f);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_max_message_length(config);
}

@ -0,0 +1,223 @@
/*
*
* 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.
*
*/
#include "test/core/end2end/end2end_tests.h"
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include "src/core/support/string.h"
#include <grpc/byte_buffer.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "test/core/end2end/cq_verifier.h"
enum { TIMEOUT = 200000 };
static void *tag(gpr_intptr t) { return (void *)t; }
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
const char *test_name,
grpc_channel_args *client_args,
grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
f = config.create_fixture(client_args, server_args);
config.init_client(&f, client_args);
config.init_server(&f, server_args);
return f;
}
static gpr_timespec n_seconds_time(int n) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
}
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
grpc_event *ev;
grpc_completion_type type;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
GPR_ASSERT(ev);
type = ev->type;
grpc_event_finish(ev);
} while (type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown(f->server);
grpc_server_destroy(f->server);
f->server = NULL;
}
static void shutdown_client(grpc_end2end_test_fixture *f) {
if (!f->client) return;
grpc_channel_destroy(f->client);
f->client = NULL;
}
static void end_test(grpc_end2end_test_fixture *f) {
shutdown_server(f);
shutdown_client(f);
grpc_completion_queue_shutdown(f->server_cq);
drain_cq(f->server_cq);
grpc_completion_queue_destroy(f->server_cq);
grpc_completion_queue_shutdown(f->client_cq);
drain_cq(f->client_cq);
grpc_completion_queue_destroy(f->client_cq);
}
static void simple_request_body(grpc_end2end_test_fixture f) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
cq_verifier *v_client = cq_verifier_create(f.client_cq);
cq_verifier *v_server = cq_verifier_create(f.server_cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_status_code status;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
op->data.send_status_from_server.status_details = "xyz";
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
cq_verify(v_server);
cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234"));
GPR_ASSERT(was_cancelled == 0);
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(v_client);
cq_verifier_destroy(v_server);
}
static void test_invoke_10_simple_requests(grpc_end2end_test_config config, int initial_sequence_number) {
int i;
grpc_end2end_test_fixture f;
grpc_arg client_arg;
grpc_channel_args client_args;
client_arg.type = GRPC_ARG_INTEGER;
client_arg.key = GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER;
client_arg.value.integer = initial_sequence_number;
client_args.num_args = 1;
client_args.args = &client_arg;
f = begin_test(config, __FUNCTION__, &client_args, NULL);
for (i = 0; i < 10; i++) {
simple_request_body(f);
gpr_log(GPR_INFO, "Passed simple request %d", i);
}
end_test(&f);
config.tear_down_data(&f);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_invoke_10_simple_requests(config, 16777213);
if (config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION) {
test_invoke_10_simple_requests(config, 2147483645);
}
}

@ -172,7 +172,7 @@ class TestServiceImplDupPkg
class End2endTest : public ::testing::Test {
protected:
End2endTest() : thread_pool_(2) {}
End2endTest() : kMaxMessageSize_(8192), thread_pool_(2) {}
void SetUp() GRPC_OVERRIDE {
int port = grpc_pick_unused_port_or_die();
@ -182,6 +182,8 @@ class End2endTest : public ::testing::Test {
builder.AddListeningPort(server_address_.str(),
InsecureServerCredentials());
builder.RegisterService(&service_);
builder.SetMaxMessageSize(
kMaxMessageSize_); // For testing max message size.
builder.RegisterService(&dup_pkg_service_);
builder.SetThreadPool(&thread_pool_);
server_ = builder.BuildAndStart();
@ -198,6 +200,7 @@ class End2endTest : public ::testing::Test {
std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_;
std::unique_ptr<Server> server_;
std::ostringstream server_address_;
const int kMaxMessageSize_;
TestServiceImpl service_;
TestServiceImplDupPkg dup_pkg_service_;
ThreadPool thread_pool_;
@ -426,8 +429,7 @@ TEST_F(End2endTest, DiffPackageServices) {
// rpc and stream should fail on bad credentials.
TEST_F(End2endTest, BadCredentials) {
std::unique_ptr<Credentials> bad_creds =
ServiceAccountCredentials("", "", 1);
std::unique_ptr<Credentials> bad_creds = ServiceAccountCredentials("", "", 1);
EXPECT_EQ(nullptr, bad_creds.get());
std::shared_ptr<ChannelInterface> channel =
CreateChannel(server_address_.str(), bad_creds, ChannelArguments());
@ -501,14 +503,13 @@ TEST_F(End2endTest, ClientCancelsRequestStream) {
auto stream = stub_->RequestStream(&context, &response);
EXPECT_TRUE(stream->Write(request));
EXPECT_TRUE(stream->Write(request));
context.TryCancel();
Status s = stream->Finish();
EXPECT_EQ(grpc::StatusCode::CANCELLED, s.code());
EXPECT_EQ(response.message(), "");
EXPECT_EQ(response.message(), "");
}
// Client cancels server stream after sending some messages
@ -588,6 +589,17 @@ TEST_F(End2endTest, ThreadStress) {
}
}
TEST_F(End2endTest, RpcMaxMessageSize) {
ResetStub();
EchoRequest request;
EchoResponse response;
request.set_message(string(kMaxMessageSize_ * 2, 'a'));
ClientContext context;
Status s = stub_->Echo(&context, request, &response);
EXPECT_FALSE(s.IsOk());
}
} // namespace testing
} // namespace grpc

@ -130,39 +130,26 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
response_reader_;
};
class AsyncUnaryClient GRPC_FINAL : public Client {
class AsyncClient : public Client {
public:
explicit AsyncUnaryClient(const ClientConfig& config) : Client(config) {
explicit AsyncClient(const ClientConfig& config,
std::function<void(CompletionQueue*, TestService::Stub*,
const SimpleRequest&)> setup_ctx) :
Client(config) {
for (int i = 0; i < config.async_client_threads(); i++) {
cli_cqs_.emplace_back(new CompletionQueue);
}
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
int t = 0;
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
for (auto channel = channels_.begin(); channel != channels_.end();
channel++) {
auto* cq = cli_cqs_[t].get();
t = (t + 1) % cli_cqs_.size();
auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx,
const SimpleRequest& request, void* tag) {
return stub->AsyncUnaryCall(ctx, request, cq, tag);
};
TestService::Stub* stub = channel->get_stub();
const SimpleRequest& request = request_;
new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
stub, request, start_req, check_done);
setup_ctx(cq, channel->get_stub(), request_);
}
}
StartThreads(config.async_client_threads());
}
~AsyncUnaryClient() GRPC_OVERRIDE {
EndThreads();
virtual ~AsyncClient() {
for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
(*cq)->Shutdown();
void* got_tag;
@ -173,10 +160,13 @@ class AsyncUnaryClient GRPC_FINAL : public Client {
}
}
bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
bool ThreadFunc(Histogram* histogram, size_t thread_idx)
GRPC_OVERRIDE GRPC_FINAL {
void* got_tag;
bool ok;
switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, std::chrono::system_clock::now() + std::chrono::seconds(1))) {
switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok,
std::chrono::system_clock::now() +
std::chrono::seconds(1))) {
case CompletionQueue::SHUTDOWN: return false;
case CompletionQueue::TIMEOUT: return true;
case CompletionQueue::GOT_EVENT: break;
@ -192,10 +182,30 @@ class AsyncUnaryClient GRPC_FINAL : public Client {
return true;
}
private:
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
};
class AsyncUnaryClient GRPC_FINAL : public AsyncClient {
public:
explicit AsyncUnaryClient(const ClientConfig& config) :
AsyncClient(config, SetupCtx) {
StartThreads(config.async_client_threads());
}
~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
private:
static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub,
const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx,
const SimpleRequest& request, void* tag) {
return stub->AsyncUnaryCall(ctx, request, cq, tag);
};
new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
stub, req, start_req, check_done);
}
};
template <class RequestType, class ResponseType>
class ClientRpcContextStreamingImpl : public ClientRpcContext {
public:
@ -241,7 +251,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
return(false);
}
next_state_ = &ClientRpcContextStreamingImpl::ReadDone;
stream_->Read(&response_, ClientRpcContext::tag(this));
stream_->Read(&response_, ClientRpcContext::tag(this));
return true;
}
bool ReadDone(bool ok, Histogram *hist) {
@ -263,71 +273,26 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
stream_;
};
class AsyncStreamingClient GRPC_FINAL : public Client {
class AsyncStreamingClient GRPC_FINAL : public AsyncClient {
public:
explicit AsyncStreamingClient(const ClientConfig &config) : Client(config) {
for (int i = 0; i < config.async_client_threads(); i++) {
cli_cqs_.emplace_back(new CompletionQueue);
}
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
int t = 0;
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
for (auto channel = channels_.begin(); channel != channels_.end();
channel++) {
auto* cq = cli_cqs_[t].get();
t = (t + 1) % cli_cqs_.size();
auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx,
void *tag) {
auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
return stream;
};
TestService::Stub *stub = channel->get_stub();
const SimpleRequest &request = request_;
new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
stub, request, start_req, check_done);
}
}
explicit AsyncStreamingClient(const ClientConfig &config) :
AsyncClient(config, SetupCtx) {
StartThreads(config.async_client_threads());
}
~AsyncStreamingClient() GRPC_OVERRIDE {
EndThreads();
for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
(*cq)->Shutdown();
void *got_tag;
bool ok;
while ((*cq)->Next(&got_tag, &ok)) {
delete ClientRpcContext::detag(got_tag);
}
}
}
bool ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE {
void *got_tag;
bool ok;
switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, std::chrono::system_clock::now() + std::chrono::seconds(1))) {
case CompletionQueue::SHUTDOWN: return false;
case CompletionQueue::TIMEOUT: return true;
case CompletionQueue::GOT_EVENT: break;
}
ClientRpcContext *ctx = ClientRpcContext::detag(got_tag);
if (ctx->RunNextState(ok, histogram) == false) {
// call the callback and then delete it
ctx->RunNextState(ok, histogram);
ctx->StartNewClone();
delete ctx;
}
return true;
~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
private:
static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub,
const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx,
void *tag) {
auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
return stream;
};
new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
stub, req, start_req, check_done);
}
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
};
std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig& args) {

@ -48,6 +48,13 @@ RUN cd /var/local/git/grpc/src/php/ext/grpc \
RUN cd /var/local/git/grpc/src/php && composer install
# Add a cacerts directory containing the Google root pem file, allowing the
# php client to access the production test instance
ADD cacerts cacerts
# Add a service_account directory containing the auth creds file
ADD service_account service_account
RUN cd /var/local/git/grpc/src/php && protoc-gen-php -i tests/interop/ -o tests/interop/ tests/interop/test.proto
RUN cd /var/local/git/grpc/src/php && ./bin/run_tests.sh
RUN cd /var/local/git/grpc/src/php && ./bin/run_tests.sh

@ -38,7 +38,7 @@ main() {
source grpc_docker.sh
test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming cancel_after_begin cancel_after_first_response)
auth_test_cases=(service_account_creds compute_engine_creds jwt_token_creds)
clients=(cxx java go ruby node csharp_mono python)
clients=(cxx java go ruby node csharp_mono python php)
for test_case in "${test_cases[@]}"
do
for client in "${clients[@]}"

@ -1250,6 +1250,37 @@ grpc_interop_gen_php_cmd() {
echo $the_cmd
}
# constructs the full dockerized php service_account auth interop test cmd.
#
# call-seq:
# flags= .... # generic flags to include the command
# cmd=$($grpc_gen_test_cmd $flags)
grpc_cloud_prod_auth_service_account_creds_gen_php_cmd() {
local env_flag="-e SSL_CERT_FILE=/cacerts/roots.pem "
env_flag+="-e GOOGLE_APPLICATION_CREDENTIALS=/service_account/stubbyCloudTestingTest-7dd63462c60c.json "
local cmd_prefix="sudo docker run $env_flag grpc/php";
local test_script="/var/local/git/grpc/src/php/bin/interop_client.sh";
local gfe_flags=$(_grpc_prod_gfe_flags);
local added_gfe_flags=$(_grpc_default_creds_test_flags)
local the_cmd="$cmd_prefix $test_script $gfe_flags $added_gfe_flags $@";
echo $the_cmd
}
# constructs the full dockerized php compute_engine auth interop test cmd.
#
# call-seq:
# flags= .... # generic flags to include the command
# cmd=$($grpc_gen_test_cmd $flags)
grpc_cloud_prod_auth_compute_engine_creds_gen_php_cmd() {
local env_flag="-e SSL_CERT_FILE=/cacerts/roots.pem "
local cmd_prefix="sudo docker run $env_flag grpc/php";
local test_script="/var/local/git/grpc/src/php/bin/interop_client.sh";
local gfe_flags=$(_grpc_prod_gfe_flags);
local added_gfe_flags=$(_grpc_gce_test_flags)
local the_cmd="$cmd_prefix $test_script $gfe_flags $added_gfe_flags $@";
echo $the_cmd
}
# constructs the full dockerized node interop test cmd.
#
# call-seq:

@ -421,6 +421,10 @@ grpc_dockerfile_install() {
grpc_docker_sync_roots_pem $dockerfile_dir/cacerts || return 1;
grpc_docker_sync_service_account $dockerfile_dir/service_account || return 1;
}
[[ $image_label == "grpc/php" ]] && {
grpc_docker_sync_roots_pem $dockerfile_dir/cacerts || return 1;
grpc_docker_sync_service_account $dockerfile_dir/service_account || return 1;
}
[[ $image_label == "grpc/cxx" ]] && {
grpc_docker_sync_roots_pem $dockerfile_dir/cacerts || return 1;
grpc_docker_sync_service_account $dockerfile_dir/service_account || return 1;

@ -37,3 +37,14 @@ export TEST=true
cd `dirname $0`/../..
./tools/buildgen/generate_projects.sh
submodules=`mktemp`
git submodule > $submodules
diff -u $submodules - << EOF
05b155ff59114735ec8cd089f669c4c3d8f59029 third_party/gflags (v2.1.0-45-g05b155f)
3df69d3aefde7671053d4e3c242b228e5d79c83f third_party/openssl (OpenSSL_1_0_2a)
644a6a1da71385e9d7a7a26b3476c93fdd71788c third_party/protobuf (v3.0.0-alpha-1-35-g644a6a1)
50893291621658f355bc5b4d450a8d06a563053d third_party/zlib (v1.2.8)
EOF

@ -819,6 +819,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fake_security_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -909,6 +918,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fake_security_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1035,6 +1053,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1125,6 +1152,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1251,6 +1287,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1341,6 +1386,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1467,6 +1521,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1557,6 +1620,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_fullstack_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1683,6 +1755,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_with_oauth2_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1773,6 +1854,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_with_oauth2_fullstack_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1899,6 +1989,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1989,6 +2088,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2115,6 +2223,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2205,6 +2322,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_simple_request_with_high_initial_sequence_number_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2331,6 +2457,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2421,6 +2556,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_simple_request_with_high_initial_sequence_number_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2547,6 +2691,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2637,6 +2790,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_simple_request_with_high_initial_sequence_number_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2763,6 +2925,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2853,6 +3024,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_simple_request_with_high_initial_sequence_number_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2979,6 +3159,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3068,6 +3257,15 @@
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_simple_request_with_high_initial_sequence_number_unsecure_test",
"platforms": [
"windows",
"posix"
]
}
]

File diff suppressed because one or more lines are too long

@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<ImportGroup Label="PropertySheets" />
<PropertyGroup Label="UserMacros" />
<PropertyGroup />
<ItemDefinitionGroup>
<Link>
<AdditionalDependencies>ssleay32.lib;libeay32.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalLibraryDirectories>$(MSBuildProjectDirectory)\..\packages\grpc.dependencies.openssl.1.0.2.2\build\native\lib\$(PlatformToolset)\$(Platform)\$(Configuration)\static;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories>
</Link>
</ItemDefinitionGroup>
<ItemGroup />
</Project>

@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<ImportGroup Label="PropertySheets" />
<PropertyGroup Label="UserMacros" />
<PropertyGroup />
<ItemDefinitionGroup>
<Link>
<AdditionalDependencies>zlib.lib;%(AdditionalDependencies)</AdditionalDependencies>
<AdditionalLibraryDirectories>$(MSBuildProjectDirectory)\..\packages\grpc.dependencies.zlib.1.2.8.9\build\native\lib\$(PlatformToolset)\$(Platform)\$(Configuration)\static\cdecl;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories>
</Link>
</ItemDefinitionGroup>
<ItemGroup />
</Project>
Loading…
Cancel
Save