From 518e3fe700015759cedea57f2e4febf300d57ba7 Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Tue, 21 Jul 2015 15:49:42 -0700 Subject: Fixes for streaming compression. --- src/core/channel/compress_filter.c | 2 ++ src/core/transport/chttp2/frame_data.c | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) (limited to 'src/core') diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c index 14cb3da62d..9b9e82caba 100644 --- a/src/core/channel/compress_filter.c +++ b/src/core/channel/compress_filter.c @@ -174,6 +174,8 @@ static void process_send_ops(grpc_call_element *elem, size_t i; int did_compress = 0; + /* In streaming calls, we need to reset the previously accumulated slices */ + gpr_slice_buffer_reset_and_unref(&calld->slices); for (i = 0; i < send_ops->nops; ++i) { grpc_stream_op *sop = &send_ops->ops[i]; switch (sop->type) { diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c index 7a4c355f23..40bf2ebd79 100644 --- a/src/core/transport/chttp2/frame_data.c +++ b/src/core/transport/chttp2/frame_data.c @@ -92,7 +92,7 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( p->frame_type = *cur; switch (p->frame_type) { case 0: - /* noop */ + p->is_frame_compressed = 0; /* GPR_FALSE */ break; case 1: p->is_frame_compressed = 1; /* GPR_TRUE */ -- cgit v1.2.3 From 8b25f2aaebc261004a08fb05151795f0c332b066 Mon Sep 17 00:00:00 2001 From: yang-g Date: Tue, 21 Jul 2015 23:54:36 -0700 Subject: move fake_transport_security_credentials to private API --- Makefile | 2 - build.json | 2 - include/grpc/grpc_security.h | 8 --- src/core/security/credentials.h | 8 +++ src/node/ext/credentials.cc | 7 --- src/php/ext/grpc/credentials.c | 10 ---- .../grpc/_adapter/_c/types/client_credentials.c | 17 ------ .../src/grpc/_cython/_cygrpc/credentials.pyx | 11 ---- src/python/src/grpc/_cython/_cygrpc/grpc.pxd | 2 - test/cpp/end2end/end2end_test.cc | 60 +++++++++++++--------- test/cpp/util/fake_credentials.cc | 58 --------------------- test/cpp/util/fake_credentials.h | 51 ------------------ tools/run_tests/sources_and_headers.json | 3 -- 13 files changed, 44 insertions(+), 195 deletions(-) delete mode 100644 test/cpp/util/fake_credentials.cc delete mode 100644 test/cpp/util/fake_credentials.h (limited to 'src/core') diff --git a/Makefile b/Makefile index 2ae0cd052d..b5fc934806 100644 --- a/Makefile +++ b/Makefile @@ -4117,7 +4117,6 @@ LIBGRPC++_TEST_UTIL_SRC = \ $(GENDIR)/test/cpp/util/echo_duplicate.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.grpc.pb.cc \ test/cpp/util/cli_call.cc \ test/cpp/util/create_test_channel.cc \ - test/cpp/util/fake_credentials.cc \ test/cpp/util/subprocess.cc \ @@ -4164,7 +4163,6 @@ endif endif $(OBJDIR)/$(CONFIG)/test/cpp/util/cli_call.o: $(GENDIR)/test/cpp/util/messages.pb.cc $(GENDIR)/test/cpp/util/messages.grpc.pb.cc $(GENDIR)/test/cpp/util/echo.pb.cc $(GENDIR)/test/cpp/util/echo.grpc.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/create_test_channel.o: $(GENDIR)/test/cpp/util/messages.pb.cc $(GENDIR)/test/cpp/util/messages.grpc.pb.cc $(GENDIR)/test/cpp/util/echo.pb.cc $(GENDIR)/test/cpp/util/echo.grpc.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.grpc.pb.cc -$(OBJDIR)/$(CONFIG)/test/cpp/util/fake_credentials.o: $(GENDIR)/test/cpp/util/messages.pb.cc $(GENDIR)/test/cpp/util/messages.grpc.pb.cc $(GENDIR)/test/cpp/util/echo.pb.cc $(GENDIR)/test/cpp/util/echo.grpc.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/subprocess.o: $(GENDIR)/test/cpp/util/messages.pb.cc $(GENDIR)/test/cpp/util/messages.grpc.pb.cc $(GENDIR)/test/cpp/util/echo.pb.cc $(GENDIR)/test/cpp/util/echo.grpc.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.pb.cc $(GENDIR)/test/cpp/util/echo_duplicate.grpc.pb.cc diff --git a/build.json b/build.json index 2755703e1c..e61dc67a68 100644 --- a/build.json +++ b/build.json @@ -609,7 +609,6 @@ "headers": [ "test/cpp/util/cli_call.h", "test/cpp/util/create_test_channel.h", - "test/cpp/util/fake_credentials.h", "test/cpp/util/subprocess.h" ], "src": [ @@ -618,7 +617,6 @@ "test/cpp/util/echo_duplicate.proto", "test/cpp/util/cli_call.cc", "test/cpp/util/create_test_channel.cc", - "test/cpp/util/fake_credentials.cc", "test/cpp/util/subprocess.cc" ], "deps": [ diff --git a/include/grpc/grpc_security.h b/include/grpc/grpc_security.h index 37d66c04ae..18d9d2f899 100644 --- a/include/grpc/grpc_security.h +++ b/include/grpc/grpc_security.h @@ -140,9 +140,6 @@ grpc_credentials *grpc_access_token_credentials_create( grpc_credentials *grpc_iam_credentials_create(const char *authorization_token, const char *authority_selector); -/* Creates a fake transport security credentials object for testing. */ -grpc_credentials *grpc_fake_transport_security_credentials_create(void); - /* --- Secure channel creation. --- */ /* The caller of the secure_channel_create functions may override the target @@ -182,10 +179,6 @@ grpc_server_credentials *grpc_ssl_server_credentials_create( const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pairs, size_t num_key_cert_pairs); -/* Creates a fake server transport security credentials object for testing. */ -grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( - void); - /* --- Server-side secure ports. --- */ /* Add a HTTP2 over an encrypted link over tcp listener. @@ -206,7 +199,6 @@ grpc_call_error grpc_call_set_credentials(grpc_call *call, /* TODO(jboeuf): Define some well-known property names. */ #define GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME "transport_security_type" -#define GRPC_FAKE_TRANSPORT_SECURITY_TYPE "fake" #define GRPC_SSL_TRANSPORT_SECURITY_TYPE "ssl" #define GRPC_X509_CN_PROPERTY_NAME "x509_common_name" diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index d988901cf7..f5635122af 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -52,6 +52,8 @@ typedef enum { GRPC_CREDENTIALS_ERROR } grpc_credentials_status; +#define GRPC_FAKE_TRANSPORT_SECURITY_TYPE "fake" + #define GRPC_CREDENTIALS_TYPE_SSL "Ssl" #define GRPC_CREDENTIALS_TYPE_OAUTH2 "Oauth2" #define GRPC_CREDENTIALS_TYPE_JWT "Jwt" @@ -112,6 +114,12 @@ void grpc_credentials_md_store_unref(grpc_credentials_md_store *store); /* --- grpc_credentials. --- */ +/* Creates a fake transport security credentials object for testing. */ +grpc_credentials *grpc_fake_transport_security_credentials_create(void); +/* Creates a fake server transport security credentials object for testing. */ +grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( + void); + /* It is the caller's responsibility to gpr_free the result if not NULL. */ char *grpc_get_well_known_google_credentials_file_path(void); diff --git a/src/node/ext/credentials.cc b/src/node/ext/credentials.cc index 34872017ea..d6cff0631d 100644 --- a/src/node/ext/credentials.cc +++ b/src/node/ext/credentials.cc @@ -79,8 +79,6 @@ void Credentials::Init(Handle exports) { NanNew(CreateComposite)->GetFunction()); ctr->Set(NanNew("createGce"), NanNew(CreateGce)->GetFunction()); - ctr->Set(NanNew("createFake"), - NanNew(CreateFake)->GetFunction()); ctr->Set(NanNew("createIam"), NanNew(CreateIam)->GetFunction()); constructor = new NanCallback(ctr); @@ -180,11 +178,6 @@ NAN_METHOD(Credentials::CreateGce) { NanReturnValue(WrapStruct(grpc_compute_engine_credentials_create())); } -NAN_METHOD(Credentials::CreateFake) { - NanScope(); - NanReturnValue(WrapStruct(grpc_fake_transport_security_credentials_create())); -} - NAN_METHOD(Credentials::CreateIam) { NanScope(); if (!args[0]->IsString()) { diff --git a/src/php/ext/grpc/credentials.c b/src/php/ext/grpc/credentials.c index a262b9981f..3e781d4747 100644 --- a/src/php/ext/grpc/credentials.c +++ b/src/php/ext/grpc/credentials.c @@ -175,15 +175,6 @@ PHP_METHOD(Credentials, createGce) { RETURN_DESTROY_ZVAL(creds_object); } -/** - * Create fake credentials. Only to be used for testing. - * @return Credentials The new fake credentials object - */ -PHP_METHOD(Credentials, createFake) { - grpc_credentials *creds = grpc_fake_transport_security_credentials_create(); - zval *creds_object = grpc_php_wrap_credentials(creds); - RETURN_DESTROY_ZVAL(creds_object); -} static zend_function_entry credentials_methods[] = { PHP_ME(Credentials, createDefault, NULL, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC) @@ -191,7 +182,6 @@ static zend_function_entry credentials_methods[] = { PHP_ME(Credentials, createComposite, NULL, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC) PHP_ME(Credentials, createGce, NULL, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC) - PHP_ME(Credentials, createFake, NULL, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC) PHP_FE_END}; void grpc_init_credentials(TSRMLS_D) { diff --git a/src/python/src/grpc/_adapter/_c/types/client_credentials.c b/src/python/src/grpc/_adapter/_c/types/client_credentials.c index 6a4561c060..6e1a3e6f30 100644 --- a/src/python/src/grpc/_adapter/_c/types/client_credentials.c +++ b/src/python/src/grpc/_adapter/_c/types/client_credentials.c @@ -54,9 +54,6 @@ PyMethodDef pygrpc_ClientCredentials_methods[] = { METH_CLASS|METH_KEYWORDS, ""}, {"refresh_token", (PyCFunction)pygrpc_ClientCredentials_refresh_token, METH_CLASS|METH_KEYWORDS, ""}, - {"fake_transport_security", - (PyCFunction)pygrpc_ClientCredentials_fake_transport_security, - METH_CLASS|METH_NOARGS, ""}, {"iam", (PyCFunction)pygrpc_ClientCredentials_iam, METH_CLASS|METH_KEYWORDS, ""}, {NULL} @@ -249,20 +246,6 @@ ClientCredentials *pygrpc_ClientCredentials_refresh_token( return self; } -ClientCredentials *pygrpc_ClientCredentials_fake_transport_security( - PyTypeObject *type, PyObject *ignored) { - ClientCredentials *self = (ClientCredentials *)type->tp_alloc(type, 0); - self->c_creds = grpc_fake_transport_security_credentials_create(); - if (!self->c_creds) { - Py_DECREF(self); - PyErr_SetString(PyExc_RuntimeError, - "couldn't create fake credentials; " - "something is horribly wrong with the universe"); - return NULL; - } - return self; -} - ClientCredentials *pygrpc_ClientCredentials_iam( PyTypeObject *type, PyObject *args, PyObject *kwargs) { ClientCredentials *self; diff --git a/src/python/src/grpc/_cython/_cygrpc/credentials.pyx b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx index c14d8844dd..14e7ce84c0 100644 --- a/src/python/src/grpc/_cython/_cygrpc/credentials.pyx +++ b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx @@ -152,12 +152,6 @@ def client_credentials_refresh_token(json_refresh_token): credentials.references.append(json_refresh_token) return credentials -def client_credentials_fake_transport_security(): - cdef ClientCredentials credentials = ClientCredentials() - credentials.c_credentials = ( - grpc.grpc_fake_transport_security_credentials_create()) - return credentials - def client_credentials_iam(authorization_token, authority_selector): if isinstance(authorization_token, bytes): pass @@ -210,8 +204,3 @@ def server_credentials_ssl(pem_root_certs, pem_key_cert_pairs): ) return credentials -def server_credentials_fake_transport_security(): - cdef ServerCredentials credentials = ServerCredentials() - credentials.c_credentials = ( - grpc.grpc_fake_transport_security_server_credentials_create()) - return credentials diff --git a/src/python/src/grpc/_cython/_cygrpc/grpc.pxd b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd index 7db8fbe31c..7f36136250 100644 --- a/src/python/src/grpc/_cython/_cygrpc/grpc.pxd +++ b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd @@ -317,7 +317,6 @@ cdef extern from "grpc/grpc_security.h": gpr_timespec token_lifetime) grpc_credentials *grpc_refresh_token_credentials_create( const char *json_refresh_token) - grpc_credentials *grpc_fake_transport_security_credentials_create() grpc_credentials *grpc_iam_credentials_create(const char *authorization_token, const char *authority_selector) void grpc_credentials_release(grpc_credentials *creds) @@ -334,7 +333,6 @@ cdef extern from "grpc/grpc_security.h": const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pairs, size_t num_key_cert_pairs); - grpc_server_credentials *grpc_fake_transport_security_server_credentials_create() void grpc_server_credentials_release(grpc_server_credentials *creds) int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 8b4424c735..fc6b54305c 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -35,11 +35,11 @@ #include #include "src/core/security/credentials.h" +#include "test/core/end2end/data/ssl_test_data.h" #include "test/core/util/port.h" #include "test/core/util/test_config.h" #include "test/cpp/util/echo_duplicate.grpc.pb.h" #include "test/cpp/util/echo.grpc.pb.h" -#include "test/cpp/util/fake_credentials.h" #include #include #include @@ -83,13 +83,12 @@ void MaybeEchoDeadline(ServerContext* context, const EchoRequest* request, } } -template -void CheckAuthContext(T* context) { +void CheckServerAuthContext(const ServerContext* context) { std::shared_ptr auth_ctx = context->auth_context(); - std::vector fake = + std::vector ssl = auth_ctx->FindPropertyValues("transport_security_type"); - EXPECT_EQ(1u, fake.size()); - EXPECT_EQ("fake", fake[0]); + EXPECT_EQ(1u, ssl.size()); + EXPECT_EQ("ssl", ssl[0]); EXPECT_TRUE(auth_ctx->GetPeerIdentityPropertyName().empty()); EXPECT_TRUE(auth_ctx->GetPeerIdentity().empty()); } @@ -142,7 +141,7 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { } } if (request->has_param() && request->param().check_auth_context()) { - CheckAuthContext(context); + CheckServerAuthContext(context); } return Status::OK; } @@ -235,10 +234,15 @@ class End2endTest : public ::testing::Test { server_address_ << "localhost:" << port; // Setup server ServerBuilder builder; + SslServerCredentialsOptions::PemKeyCertPair pkcp = {test_server1_key, + test_server1_cert}; + SslServerCredentialsOptions ssl_opts; + ssl_opts.pem_root_certs = ""; + ssl_opts.pem_key_cert_pairs.push_back(pkcp); builder.AddListeningPort(server_address_.str(), - FakeTransportSecurityServerCredentials()); + SslServerCredentials(ssl_opts)); builder.RegisterService(&service_); - builder.RegisterService("special", &special_service_); + builder.RegisterService("foo.test.youtube.com", &special_service_); builder.SetMaxMessageSize( kMaxMessageSize_); // For testing max message size. builder.RegisterService(&dup_pkg_service_); @@ -249,12 +253,15 @@ class End2endTest : public ::testing::Test { void TearDown() GRPC_OVERRIDE { server_->Shutdown(); } void ResetStub() { - std::shared_ptr channel = - CreateChannel(server_address_.str(), FakeTransportSecurityCredentials(), - ChannelArguments()); - stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel)); + SslCredentialsOptions ssl_opts = {test_root_cert, "", ""}; + ChannelArguments args; + args.SetSslTargetNameOverride("foo.test.google.fr"); + channel_ = CreateChannel(server_address_.str(), SslCredentials(ssl_opts), + args); + stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel_)); } + std::shared_ptr channel_; std::unique_ptr stub_; std::unique_ptr server_; std::ostringstream server_address_; @@ -288,11 +295,11 @@ TEST_F(End2endTest, SimpleRpcWithHost) { request.set_message("Hello"); ClientContext context; - context.set_authority("special"); + context.set_authority("foo.test.youtube.com"); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); EXPECT_TRUE(response.has_param()); - EXPECT_EQ(response.param().host(), "special"); + EXPECT_EQ("special", response.param().host()); EXPECT_TRUE(s.ok()); } @@ -481,24 +488,19 @@ TEST_F(End2endTest, BidiStream) { // Talk to the two services with the same name but different package names. // The two stubs are created on the same channel. TEST_F(End2endTest, DiffPackageServices) { - std::shared_ptr channel = - CreateChannel(server_address_.str(), FakeTransportSecurityCredentials(), - ChannelArguments()); - + ResetStub(); EchoRequest request; EchoResponse response; request.set_message("Hello"); - std::unique_ptr stub( - grpc::cpp::test::util::TestService::NewStub(channel)); ClientContext context; - Status s = stub->Echo(&context, request, &response); + Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); EXPECT_TRUE(s.ok()); std::unique_ptr dup_pkg_stub( - grpc::cpp::test::util::duplicate::TestService::NewStub(channel)); + grpc::cpp::test::util::duplicate::TestService::NewStub(channel_)); ClientContext context2; s = dup_pkg_stub->Echo(&context2, request, &response); EXPECT_EQ("no package", response.message()); @@ -782,7 +784,17 @@ TEST_F(End2endTest, ClientAuthContext) { EXPECT_EQ(response.message(), request.message()); EXPECT_TRUE(s.ok()); - CheckAuthContext(&context); + std::shared_ptr auth_ctx = context.auth_context(); + std::vector ssl = + auth_ctx->FindPropertyValues("transport_security_type"); + EXPECT_EQ(1u, ssl.size()); + EXPECT_EQ("ssl", ssl[0]); + EXPECT_EQ("x509_subject_alternative_name", + auth_ctx->GetPeerIdentityPropertyName()); + EXPECT_EQ(3u, auth_ctx->GetPeerIdentity().size()); + EXPECT_EQ("*.test.google.fr", auth_ctx->GetPeerIdentity()[0]); + EXPECT_EQ("waterzooi.test.google.be", auth_ctx->GetPeerIdentity()[1]); + EXPECT_EQ("*.test.youtube.com", auth_ctx->GetPeerIdentity()[2]); } } // namespace testing diff --git a/test/cpp/util/fake_credentials.cc b/test/cpp/util/fake_credentials.cc deleted file mode 100644 index f5b83b8159..0000000000 --- a/test/cpp/util/fake_credentials.cc +++ /dev/null @@ -1,58 +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 -#include -#include -#include -#include "src/cpp/client/channel.h" -#include "src/cpp/client/secure_credentials.h" -#include "src/cpp/server/secure_server_credentials.h" - -namespace grpc { -namespace testing { - -std::shared_ptr FakeTransportSecurityCredentials() { - grpc_credentials* c_creds = grpc_fake_transport_security_credentials_create(); - return std::shared_ptr(new SecureCredentials(c_creds)); -} - -std::shared_ptr FakeTransportSecurityServerCredentials() { - grpc_server_credentials* c_creds = - grpc_fake_transport_security_server_credentials_create(); - return std::shared_ptr( - new SecureServerCredentials(c_creds)); -} - -} // namespace testing -} // namespace grpc diff --git a/test/cpp/util/fake_credentials.h b/test/cpp/util/fake_credentials.h deleted file mode 100644 index e1ba7bb9e4..0000000000 --- a/test/cpp/util/fake_credentials.h +++ /dev/null @@ -1,51 +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. - * - */ - -#ifndef GRPC_TEST_CPP_UTIL_FAKE_CREDENTIALS_H -#define GRPC_TEST_CPP_UTIL_FAKE_CREDENTIALS_H - -#include - -namespace grpc { -class Credentials; -class ServerCredentials; - -namespace testing { - -std::shared_ptr FakeTransportSecurityCredentials(); -std::shared_ptr FakeTransportSecurityServerCredentials(); - -} // namespace testing -} // namespace grpc - -#endif // GRPC_TEST_CPP_UTIL_FAKE_CREDENTIALS_H diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index abddaab699..15f6639662 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -10783,7 +10783,6 @@ "test/cpp/util/echo.pb.h", "test/cpp/util/echo_duplicate.grpc.pb.h", "test/cpp/util/echo_duplicate.pb.h", - "test/cpp/util/fake_credentials.h", "test/cpp/util/messages.grpc.pb.h", "test/cpp/util/messages.pb.h", "test/cpp/util/subprocess.h" @@ -10795,8 +10794,6 @@ "test/cpp/util/cli_call.h", "test/cpp/util/create_test_channel.cc", "test/cpp/util/create_test_channel.h", - "test/cpp/util/fake_credentials.cc", - "test/cpp/util/fake_credentials.h", "test/cpp/util/subprocess.cc", "test/cpp/util/subprocess.h" ] -- cgit v1.2.3 From fe4c3f4f1436e2fc2297d5544859ce14e03659bf Mon Sep 17 00:00:00 2001 From: Julien Boeuf Date: Wed, 22 Jul 2015 16:20:13 -0700 Subject: Renaming jwt_credentials to service_account_jwt_access_credentials. --- include/grpc++/credentials.h | 6 +++--- include/grpc/grpc_security.h | 4 ++-- src/core/security/credentials.c | 23 ++++++++++++---------- src/core/security/credentials.h | 5 +++-- src/core/security/google_default_credentials.c | 5 +++-- src/cpp/client/secure_credentials.cc | 8 ++++---- .../grpc/_adapter/_c/types/client_credentials.c | 3 ++- .../src/grpc/_cython/_cygrpc/credentials.pyx | 3 ++- src/python/src/grpc/_cython/_cygrpc/grpc.pxd | 2 +- test/core/security/credentials_test.c | 14 +++++++------ test/cpp/interop/client_helper.cc | 3 ++- 11 files changed, 43 insertions(+), 33 deletions(-) (limited to 'src/core') diff --git a/include/grpc++/credentials.h b/include/grpc++/credentials.h index 0eaaefcbca..a4f1e73118 100644 --- a/include/grpc++/credentials.h +++ b/include/grpc++/credentials.h @@ -106,13 +106,13 @@ std::shared_ptr ServiceAccountCredentials( const grpc::string& json_key, const grpc::string& scope, long token_lifetime_seconds); -// Builds JWT credentials. +// Builds Service Account JWT Access credentials. // json_key is the JSON key string containing the client's private key. // token_lifetime_seconds is the lifetime in seconds of each Json Web Token // (JWT) created with this credentials. It should not exceed // grpc_max_auth_token_lifetime or will be cropped to this value. -std::shared_ptr JWTCredentials(const grpc::string& json_key, - long token_lifetime_seconds); +std::shared_ptr ServiceAccountJWTAccessCredentials( + const grpc::string& json_key, long token_lifetime_seconds); // Builds refresh token credentials. // json_refresh_token is the JSON string containing the refresh token along diff --git a/include/grpc/grpc_security.h b/include/grpc/grpc_security.h index 37d66c04ae..4dd058063d 100644 --- a/include/grpc/grpc_security.h +++ b/include/grpc/grpc_security.h @@ -119,8 +119,8 @@ grpc_credentials *grpc_service_account_credentials_create( - token_lifetime is the lifetime of each Json Web Token (JWT) created with this credentials. It should not exceed grpc_max_auth_token_lifetime or will be cropped to this value. */ -grpc_credentials *grpc_jwt_credentials_create(const char *json_key, - gpr_timespec token_lifetime); +grpc_credentials *grpc_service_account_jwt_access_credentials_create( + const char *json_key, gpr_timespec token_lifetime); /* Creates an Oauth2 Refresh Token credentials object. May return NULL if the input is invalid. diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index fb59fa4b0e..38612cf308 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -315,7 +315,7 @@ grpc_server_credentials *grpc_ssl_server_credentials_create( /* -- Jwt credentials -- */ -static void jwt_reset_cache(grpc_jwt_credentials *c) { +static void jwt_reset_cache(grpc_service_account_jwt_access_credentials *c) { if (c->cached.jwt_md != NULL) { grpc_credentials_md_store_unref(c->cached.jwt_md); c->cached.jwt_md = NULL; @@ -328,7 +328,8 @@ static void jwt_reset_cache(grpc_jwt_credentials *c) { } static void jwt_destroy(grpc_credentials *creds) { - grpc_jwt_credentials *c = (grpc_jwt_credentials *)creds; + grpc_service_account_jwt_access_credentials *c = + (grpc_service_account_jwt_access_credentials *)creds; grpc_auth_json_key_destruct(&c->key); jwt_reset_cache(c); gpr_mu_destroy(&c->cache_mu); @@ -346,7 +347,8 @@ static void jwt_get_request_metadata(grpc_credentials *creds, const char *service_url, grpc_credentials_metadata_cb cb, void *user_data) { - grpc_jwt_credentials *c = (grpc_jwt_credentials *)creds; + grpc_service_account_jwt_access_credentials *c = + (grpc_service_account_jwt_access_credentials *)creds; gpr_timespec refresh_threshold = gpr_time_from_seconds( GRPC_SECURE_TOKEN_REFRESH_THRESHOLD_SECS, GPR_TIMESPAN); @@ -399,15 +401,16 @@ static grpc_credentials_vtable jwt_vtable = { jwt_destroy, jwt_has_request_metadata, jwt_has_request_metadata_only, jwt_get_request_metadata, NULL}; -grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials * +grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key key, gpr_timespec token_lifetime) { - grpc_jwt_credentials *c; + grpc_service_account_jwt_access_credentials *c; if (!grpc_auth_json_key_is_valid(&key)) { gpr_log(GPR_ERROR, "Invalid input for jwt credentials creation"); return NULL; } - c = gpr_malloc(sizeof(grpc_jwt_credentials)); - memset(c, 0, sizeof(grpc_jwt_credentials)); + c = gpr_malloc(sizeof(grpc_service_account_jwt_access_credentials)); + memset(c, 0, sizeof(grpc_service_account_jwt_access_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_JWT; gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &jwt_vtable; @@ -418,9 +421,9 @@ grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( return &c->base; } -grpc_credentials *grpc_jwt_credentials_create(const char *json_key, - gpr_timespec token_lifetime) { - return grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials *grpc_service_account_jwt_access_credentials_create( + const char *json_key, gpr_timespec token_lifetime) { + return grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key_create_from_string(json_key), token_lifetime); } diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index d988901cf7..7f4141967d 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -188,7 +188,8 @@ grpc_credentials *grpc_fake_oauth2_credentials_create( /* Private constructor for jwt credentials from an already parsed json key. Takes ownership of the key. */ -grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials * +grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key key, gpr_timespec token_lifetime); /* Private constructor for refresh token credentials from an already parsed @@ -240,7 +241,7 @@ typedef struct { grpc_auth_json_key key; gpr_timespec jwt_lifetime; -} grpc_jwt_credentials; +} grpc_service_account_jwt_access_credentials; /* -- Oauth2TokenFetcher credentials -- diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 833484310f..de1929fe76 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -140,8 +140,9 @@ static grpc_credentials *create_default_creds_from_path(char *creds_path) { /* First, try an auth json key. */ key = grpc_auth_json_key_create_from_json(json); if (grpc_auth_json_key_is_valid(&key)) { - result = grpc_jwt_credentials_create_from_auth_json_key( - key, grpc_max_auth_token_lifetime); + result = + grpc_service_account_jwt_access_credentials_create_from_auth_json_key( + key, grpc_max_auth_token_lifetime); goto end; } diff --git a/src/cpp/client/secure_credentials.cc b/src/cpp/client/secure_credentials.cc index 01c7f14f1a..abf0cb387e 100644 --- a/src/cpp/client/secure_credentials.cc +++ b/src/cpp/client/secure_credentials.cc @@ -99,8 +99,8 @@ std::shared_ptr ServiceAccountCredentials( } // Builds JWT credentials. -std::shared_ptr JWTCredentials(const grpc::string& json_key, - long token_lifetime_seconds) { +std::shared_ptr ServiceAccountJWTAccessCredentials( + const grpc::string& json_key, long token_lifetime_seconds) { if (token_lifetime_seconds <= 0) { gpr_log(GPR_ERROR, "Trying to create JWTCredentials with non-positive lifetime"); @@ -108,8 +108,8 @@ std::shared_ptr JWTCredentials(const grpc::string& json_key, } gpr_timespec lifetime = gpr_time_from_seconds(token_lifetime_seconds, GPR_TIMESPAN); - return WrapCredentials( - grpc_jwt_credentials_create(json_key.c_str(), lifetime)); + return WrapCredentials(grpc_service_account_jwt_access_credentials_create( + json_key.c_str(), lifetime)); } // Builds refresh token credentials. diff --git a/src/python/src/grpc/_adapter/_c/types/client_credentials.c b/src/python/src/grpc/_adapter/_c/types/client_credentials.c index 6a4561c060..9ea2b39cad 100644 --- a/src/python/src/grpc/_adapter/_c/types/client_credentials.c +++ b/src/python/src/grpc/_adapter/_c/types/client_credentials.c @@ -208,6 +208,7 @@ ClientCredentials *pygrpc_ClientCredentials_service_account( return self; } +/* TODO: Rename this credentials to something like service_account_jwt_access */ ClientCredentials *pygrpc_ClientCredentials_jwt( PyTypeObject *type, PyObject *args, PyObject *kwargs) { ClientCredentials *self; @@ -219,7 +220,7 @@ ClientCredentials *pygrpc_ClientCredentials_jwt( return NULL; } self = (ClientCredentials *)type->tp_alloc(type, 0); - self->c_creds = grpc_jwt_credentials_create( + self->c_creds = grpc_service_account_jwt_access_credentials_create( json_key, pygrpc_cast_double_to_gpr_timespec(lifetime)); if (!self->c_creds) { Py_DECREF(self); diff --git a/src/python/src/grpc/_cython/_cygrpc/credentials.pyx b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx index c14d8844dd..7bb3f798b2 100644 --- a/src/python/src/grpc/_cython/_cygrpc/credentials.pyx +++ b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx @@ -126,6 +126,7 @@ def client_credentials_service_account( credentials.references.extend([json_key, scope]) return credentials +#TODO rename to something like client_credentials_service_account_jwt_access. def client_credentials_jwt(json_key, records.Timespec token_lifetime not None): if isinstance(json_key, bytes): pass @@ -134,7 +135,7 @@ def client_credentials_jwt(json_key, records.Timespec token_lifetime not None): else: raise TypeError("expected json_key to be str or bytes") cdef ClientCredentials credentials = ClientCredentials() - credentials.c_credentials = grpc.grpc_jwt_credentials_create( + credentials.c_credentials = grpc.grpc_service_account_jwt_access_credentials_create( json_key, token_lifetime.c_time) credentials.references.append(json_key) return credentials diff --git a/src/python/src/grpc/_cython/_cygrpc/grpc.pxd b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd index 7db8fbe31c..a76ddfc9e1 100644 --- a/src/python/src/grpc/_cython/_cygrpc/grpc.pxd +++ b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd @@ -313,7 +313,7 @@ cdef extern from "grpc/grpc_security.h": grpc_credentials *grpc_compute_engine_credentials_create() grpc_credentials *grpc_service_account_credentials_create( const char *json_key, const char *scope, gpr_timespec token_lifetime) - grpc_credentials *grpc_jwt_credentials_create(const char *json_key, + grpc_credentials *grpc_service_account_jwt_access_credentials_create(const char *json_key, gpr_timespec token_lifetime) grpc_credentials *grpc_refresh_token_credentials_create( const char *json_refresh_token) diff --git a/test/core/security/credentials_test.c b/test/core/security/credentials_test.c index d3fea9680a..dd6e0d7bb3 100644 --- a/test/core/security/credentials_test.c +++ b/test/core/security/credentials_test.c @@ -826,8 +826,9 @@ static void on_jwt_creds_get_metadata_failure(void *user_data, static void test_jwt_creds_success(void) { char *json_key_string = test_json_key_str(); - grpc_credentials *jwt_creds = grpc_jwt_credentials_create( - json_key_string, grpc_max_auth_token_lifetime); + grpc_credentials *jwt_creds = + grpc_service_account_jwt_access_credentials_create( + json_key_string, grpc_max_auth_token_lifetime); GPR_ASSERT(grpc_credentials_has_request_metadata(jwt_creds)); GPR_ASSERT(grpc_credentials_has_request_metadata_only(jwt_creds)); @@ -858,8 +859,9 @@ static void test_jwt_creds_success(void) { static void test_jwt_creds_signing_failure(void) { char *json_key_string = test_json_key_str(); - grpc_credentials *jwt_creds = grpc_jwt_credentials_create( - json_key_string, grpc_max_auth_token_lifetime); + grpc_credentials *jwt_creds = + grpc_service_account_jwt_access_credentials_create( + json_key_string, grpc_max_auth_token_lifetime); GPR_ASSERT(grpc_credentials_has_request_metadata(jwt_creds)); GPR_ASSERT(grpc_credentials_has_request_metadata_only(jwt_creds)); @@ -900,7 +902,7 @@ static grpc_credentials *composite_inner_creds(grpc_credentials *creds, } static void test_google_default_creds_auth_key(void) { - grpc_jwt_credentials *jwt; + grpc_service_account_jwt_access_credentials *jwt; grpc_credentials *creds; char *json_key = test_json_key_str(); grpc_flush_cached_google_default_credentials(); @@ -909,7 +911,7 @@ static void test_google_default_creds_auth_key(void) { gpr_free(json_key); creds = grpc_google_default_credentials_create(); GPR_ASSERT(creds != NULL); - jwt = (grpc_jwt_credentials *)composite_inner_creds( + jwt = (grpc_service_account_jwt_access_credentials *)composite_inner_creds( creds, GRPC_CREDENTIALS_TYPE_JWT); GPR_ASSERT( strcmp(jwt->key.client_id, diff --git a/test/cpp/interop/client_helper.cc b/test/cpp/interop/client_helper.cc index 48b1b2e864..73d82f7b88 100644 --- a/test/cpp/interop/client_helper.cc +++ b/test/cpp/interop/client_helper.cc @@ -123,7 +123,8 @@ std::shared_ptr CreateChannelForTestCase( GPR_ASSERT(FLAGS_enable_ssl); grpc::string json_key = GetServiceAccountJsonKey(); std::chrono::seconds token_lifetime = std::chrono::hours(1); - creds = JWTCredentials(json_key, token_lifetime.count()); + creds = + ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count()); return CreateTestChannel(host_port, FLAGS_server_host_override, FLAGS_enable_ssl, FLAGS_use_prod_roots, creds); } else if (test_case == "oauth2_auth_token") { -- cgit v1.2.3 From 9835a136280e8debc8d61008b2c78bea45b2ff5b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 22 Jul 2015 17:37:39 -0700 Subject: Handle a race where a new call comes in post-shutdown a little better --- src/core/surface/server.c | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'src/core') diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 439452aea2..ce8b2af9c4 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -400,6 +400,15 @@ static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, call_data *calld = elem->call_data; int request_id; + if (gpr_atm_acq_load(&server->shutdown_flag)) { + gpr_mu_lock(&calld->mu_state); + calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); + grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); + grpc_iomgr_add_callback(&calld->kill_zombie_closure); + return; + } + request_id = gpr_stack_lockfree_pop(request_matcher->requests); if (request_id == -1) { gpr_mu_lock(&server->mu_call); -- cgit v1.2.3 From 6f30decf79da066bb199163a820a31fb057c0157 Mon Sep 17 00:00:00 2001 From: yang-g Date: Wed, 22 Jul 2015 23:11:56 -0700 Subject: Flow control fix --- src/core/transport/chttp2/writing.c | 3 ++- test/cpp/end2end/end2end_test.cc | 20 ++++++++++++++++++++ test/cpp/util/messages.proto | 1 + 3 files changed, 23 insertions(+), 1 deletion(-) (limited to 'src/core') diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index d8ec117aa5..041c4efd1f 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -66,7 +66,8 @@ int grpc_chttp2_unlocking_check_writes( /* for each grpc_chttp2_stream that's become writable, frame it's data (according to available window sizes) and add to the output buffer */ - while (grpc_chttp2_list_pop_writable_stream(transport_global, + while (transport_global->outgoing_window > 0 && + grpc_chttp2_list_pop_writable_stream(transport_global, transport_writing, &stream_global, &stream_writing)) { stream_writing->id = stream_global->id; diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 20e4c4ed55..c433b78948 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -144,6 +144,11 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { if (request->has_param() && request->param().check_auth_context()) { CheckAuthContext(context); } + if (request->has_param() && + request->param().response_message_length() > 0) { + response->set_message( + grpc::string(request->param().response_message_length(), '\0')); + } return Status::OK; } @@ -786,6 +791,21 @@ TEST_F(End2endTest, ClientAuthContext) { CheckAuthContext(&context); } +// Make the response larger than the flow control window. +TEST_F(End2endTest, HugeResponse) { + ResetStub(); + EchoRequest request; + EchoResponse response; + request.set_message("huge response"); + const int kResponseSize = 1024 * (1024 + 10); + request.mutable_param()->set_response_message_length(kResponseSize); + + ClientContext context; + Status s = stub_->Echo(&context, request, &response); + EXPECT_EQ(kResponseSize, response.message().size()); + EXPECT_TRUE(s.ok()); +} + } // namespace testing } // namespace grpc diff --git a/test/cpp/util/messages.proto b/test/cpp/util/messages.proto index 3708972b90..2fad8b42a2 100644 --- a/test/cpp/util/messages.proto +++ b/test/cpp/util/messages.proto @@ -38,6 +38,7 @@ message RequestParams { optional int32 server_cancel_after_us = 3; optional bool echo_metadata = 4; optional bool check_auth_context = 5; + optional int32 response_message_length = 6; } message EchoRequest { -- cgit v1.2.3 From 94329d09656f3eeb8eee40b72b96ec9cd3578559 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Jul 2015 09:52:11 -0700 Subject: Make the server report monotonic times for deadlines For very high performance systems, we're going to want to be able to simply push the value reported from the server down onto clients. If we report realtime now, then all wrapped languages are going to assume it, meaning that such a change will be impossible later. --- src/core/surface/call.c | 3 ++- src/core/surface/server.c | 6 ++++-- src/core/transport/chttp2/parsing.c | 2 +- src/cpp/util/time.cc | 3 ++- src/node/ext/timeval.cc | 1 + src/python/src/grpc/_adapter/_c/utility.c | 1 + src/ruby/ext/grpc/rb_grpc.c | 6 ++++-- src/ruby/ext/grpc/rb_server.c | 13 ++++++------- 8 files changed, 21 insertions(+), 14 deletions(-) (limited to 'src/core') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index e08273e451..1f73f9cf71 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -1368,7 +1368,8 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) { l->md = 0; } } - if (gpr_time_cmp(md->deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) != + 0) { set_deadline_alarm(call, md->deadline); } if (!is_trailing) { diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 439452aea2..a0d4ab3b0a 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -530,6 +530,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { static void server_on_recv(void *ptr, int success) { grpc_call_element *elem = ptr; call_data *calld = elem->call_data; + gpr_timespec op_deadline; if (success && !calld->got_initial_metadata) { size_t i; @@ -539,8 +540,9 @@ static void server_on_recv(void *ptr, int success) { grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA) continue; grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem); - if (0 != gpr_time_cmp(op->data.metadata.deadline, - gpr_inf_future(GPR_CLOCK_REALTIME))) { + op_deadline = op->data.metadata.deadline; + if (0 != + gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) { calld->deadline = op->data.metadata.deadline; } calld->got_initial_metadata = 1; diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 904b9afce7..50a2f752f6 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -607,7 +607,7 @@ static void on_header(void *tp, grpc_mdelem *md) { } grpc_chttp2_incoming_metadata_buffer_set_deadline( &stream_parsing->incoming_metadata, - gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), *cached_timeout)); + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), *cached_timeout)); GRPC_MDELEM_UNREF(md); } else { grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->incoming_metadata, diff --git a/src/cpp/util/time.cc b/src/cpp/util/time.cc index a814cad452..799c597e0b 100644 --- a/src/cpp/util/time.cc +++ b/src/cpp/util/time.cc @@ -79,9 +79,10 @@ void TimepointHR2Timespec(const high_resolution_clock::time_point& from, } system_clock::time_point Timespec2Timepoint(gpr_timespec t) { - if (gpr_time_cmp(t, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(t, gpr_inf_future(t.clock_type)) == 0) { return system_clock::time_point::max(); } + t = gpr_convert_clock_type(t, GPR_CLOCK_REALTIME); system_clock::time_point tp; tp += duration_cast(seconds(t.tv_sec)); tp += diff --git a/src/node/ext/timeval.cc b/src/node/ext/timeval.cc index 60de4d816d..bf68513c48 100644 --- a/src/node/ext/timeval.cc +++ b/src/node/ext/timeval.cc @@ -52,6 +52,7 @@ gpr_timespec MillisecondsToTimespec(double millis) { } double TimespecToMilliseconds(gpr_timespec timespec) { + timespec = gpr_convert_clock_type(timespec, GPR_CLOCK_REALTIME); if (gpr_time_cmp(timespec, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { return std::numeric_limits::infinity(); } else if (gpr_time_cmp(timespec, gpr_inf_past(GPR_CLOCK_REALTIME)) == 0) { diff --git a/src/python/src/grpc/_adapter/_c/utility.c b/src/python/src/grpc/_adapter/_c/utility.c index d9f911a41a..51f3c9be01 100644 --- a/src/python/src/grpc/_adapter/_c/utility.c +++ b/src/python/src/grpc/_adapter/_c/utility.c @@ -374,6 +374,7 @@ PyObject *pygrpc_consume_ops(grpc_op *op, size_t nops) { } double pygrpc_cast_gpr_timespec_to_double(gpr_timespec timespec) { + timespec = gpr_convert_clock_type(timespec, GPR_CLOCK_REALTIME); return timespec.tv_sec + 1e-9*timespec.tv_nsec; } diff --git a/src/ruby/ext/grpc/rb_grpc.c b/src/ruby/ext/grpc/rb_grpc.c index 829f825597..65d9c9a237 100644 --- a/src/ruby/ext/grpc/rb_grpc.c +++ b/src/ruby/ext/grpc/rb_grpc.c @@ -209,10 +209,12 @@ static ID id_to_s; /* Converts a wrapped time constant to a standard time. */ static VALUE grpc_rb_time_val_to_time(VALUE self) { gpr_timespec *time_const = NULL; + gpr_timespec real_time; TypedData_Get_Struct(self, gpr_timespec, &grpc_rb_timespec_data_type, time_const); - return rb_funcall(rb_cTime, id_at, 2, INT2NUM(time_const->tv_sec), - INT2NUM(time_const->tv_nsec)); + real_time = gpr_convert_clock_type(*time_const, GPR_CLOCK_REALTIME); + return rb_funcall(rb_cTime, id_at, 2, INT2NUM(real_time.tv_sec), + INT2NUM(real_time.tv_nsec)); } /* Invokes inspect on the ctime version of the time val. */ diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c index e3a0a5ad80..375a651d24 100644 --- a/src/ruby/ext/grpc/rb_server.c +++ b/src/ruby/ext/grpc/rb_server.c @@ -213,6 +213,7 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue, grpc_call_error err; request_call_stack st; VALUE result; + gpr_timespec deadline; TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s); if (s->wrapped == NULL) { rb_raise(rb_eRuntimeError, "destroyed!"); @@ -245,15 +246,13 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue, } /* build the NewServerRpc struct result */ + deadline = gpr_convert_clock_type(st.details.deadline, GPR_CLOCK_REALTIME); result = rb_struct_new( - grpc_rb_sNewServerRpc, - rb_str_new2(st.details.method), + grpc_rb_sNewServerRpc, rb_str_new2(st.details.method), rb_str_new2(st.details.host), - rb_funcall(rb_cTime, id_at, 2, INT2NUM(st.details.deadline.tv_sec), - INT2NUM(st.details.deadline.tv_nsec)), - grpc_rb_md_ary_to_h(&st.md_ary), - grpc_rb_wrap_call(call), - NULL); + rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec), + INT2NUM(deadline.tv_nsec)), + grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call), NULL); grpc_request_call_stack_cleanup(&st); return result; } -- cgit v1.2.3 From 994c2620e3540de869aee52b2ad0678006ef8a6b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Jul 2015 14:00:58 -0700 Subject: Fix flow control - sending of window updates is now integrated with the primary write path, making this far more robust - iomgr starts up after shutdown correctly again --- src/core/channel/client_channel.c | 2 +- src/core/iomgr/iomgr.c | 1 + src/core/iomgr/tcp_posix.c | 4 +- src/core/transport/chttp2/internal.h | 21 +++----- src/core/transport/chttp2/parsing.c | 3 +- src/core/transport/chttp2/stream_lists.c | 68 ++++++++++++----------- src/core/transport/chttp2/writing.c | 93 ++++++++++++++++++-------------- src/core/transport/chttp2_transport.c | 10 ++-- 8 files changed, 108 insertions(+), 94 deletions(-) (limited to 'src/core') diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 108a6dfdf1..ec6ca42889 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -460,7 +460,7 @@ static void cc_on_config_changed(void *arg, int iomgr_success) { while (wakeup_closures) { grpc_iomgr_closure *next = wakeup_closures->next; - grpc_iomgr_add_callback(wakeup_closures); + wakeup_closures->cb(wakeup_closures->cb_arg, 1); wakeup_closures = next; } diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index a18c176b30..aa4bc6e20d 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -88,6 +88,7 @@ void grpc_kick_poller(void) { void grpc_iomgr_init(void) { gpr_thd_id id; + g_shutdown = 0; gpr_mu_init(&g_mu); gpr_cv_init(&g_rcv); grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC)); diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index 1e8432d463..63a8a2720e 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -319,7 +319,7 @@ static void call_read_cb(grpc_tcp *tcp, gpr_slice *slices, size_t nslices, gpr_log(GPR_DEBUG, "read: status=%d", status); for (i = 0; i < nslices; i++) { char *dump = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); - gpr_log(GPR_DEBUG, "READ: %s", dump); + gpr_log(GPR_DEBUG, "READ %p: %s", tcp, dump); gpr_free(dump); } } @@ -448,7 +448,7 @@ static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); } else { tcp->handle_read_closure.cb_arg = tcp; - grpc_iomgr_add_callback(&tcp->handle_read_closure); + grpc_iomgr_add_delayed_callback(&tcp->handle_read_closure, 1); } } diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index e7901da510..f0eeb6de50 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -60,7 +60,6 @@ typedef enum { GRPC_CHTTP2_LIST_WRITABLE, GRPC_CHTTP2_LIST_WRITING, GRPC_CHTTP2_LIST_WRITTEN, - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE, GRPC_CHTTP2_LIST_PARSING_SEEN, GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING, GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING, @@ -383,6 +382,8 @@ typedef struct { gpr_uint8 published_cancelled; /** is this stream in the stream map? (boolean) */ gpr_uint8 in_stream_map; + /** is this stream actively being written? */ + gpr_uint8 writing_now; /** stream state already published to the upper layer */ grpc_stream_state published_state; @@ -475,11 +476,17 @@ void grpc_chttp2_publish_reads(grpc_chttp2_transport_global *global, void grpc_chttp2_list_add_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); +void grpc_chttp2_list_add_first_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global); int grpc_chttp2_list_pop_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_global **stream_global, grpc_chttp2_stream_writing **stream_writing); +void grpc_chttp2_list_remove_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global); void grpc_chttp2_list_add_incoming_window_updated( grpc_chttp2_transport_global *transport_global, @@ -511,18 +518,6 @@ int grpc_chttp2_list_pop_written_stream( grpc_chttp2_stream_global **stream_global, grpc_chttp2_stream_writing **stream_writing); -void grpc_chttp2_list_add_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_writing *transport_writing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_writing **stream_writing); -void grpc_chttp2_list_remove_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); - void grpc_chttp2_list_add_parsing_seen_stream( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing); diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 904b9afce7..9105746567 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -182,8 +182,7 @@ void grpc_chttp2_publish_reads( stream_global->max_recv_bytes -= stream_parsing->incoming_window_delta; stream_parsing->incoming_window_delta = 0; - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } /* update outgoing flow control window */ diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c index 590f6abfbc..9e68c1e146 100644 --- a/src/core/transport/chttp2/stream_lists.c +++ b/src/core/transport/chttp2/stream_lists.c @@ -108,6 +108,23 @@ static void stream_list_maybe_remove(grpc_chttp2_transport *t, } } +static void stream_list_add_head(grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + grpc_chttp2_stream_list_id id) { + grpc_chttp2_stream *old_head; + GPR_ASSERT(!s->included[id]); + old_head = t->lists[id].head; + s->links[id].next = old_head; + s->links[id].prev = NULL; + if (old_head) { + old_head->links[id].prev = s; + } else { + t->lists[id].tail = s; + } + t->lists[id].head = s; + s->included[id] = 1; +} + static void stream_list_add_tail(grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_chttp2_stream_list_id id) { @@ -119,7 +136,6 @@ static void stream_list_add_tail(grpc_chttp2_transport *t, if (old_tail) { old_tail->links[id].next = s; } else { - s->links[id].prev = NULL; t->lists[id].head = s; } t->lists[id].tail = s; @@ -144,6 +160,18 @@ void grpc_chttp2_list_add_writable_stream( STREAM_FROM_GLOBAL(stream_global), GRPC_CHTTP2_LIST_WRITABLE); } +void grpc_chttp2_list_add_first_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + GPR_ASSERT(stream_global->id != 0); + gpr_log(GPR_DEBUG, "add:%d:%d:%d:%d", stream_global->id, + stream_global->write_state, stream_global->in_stream_map, + stream_global->read_closed); + stream_list_add_head(TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_WRITABLE); +} + int grpc_chttp2_list_pop_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing, @@ -157,6 +185,14 @@ int grpc_chttp2_list_pop_writable_stream( return r; } +void grpc_chttp2_list_remove_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_WRITABLE); +} + void grpc_chttp2_list_add_writing_stream( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing) { @@ -202,36 +238,6 @@ int grpc_chttp2_list_pop_written_stream( return r; } -void grpc_chttp2_list_add_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - GPR_ASSERT(stream_global->id != 0); - stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); -} - -int grpc_chttp2_list_pop_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_writing *transport_writing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_writing **stream_writing) { - grpc_chttp2_stream *stream; - int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); - *stream_global = &stream->global; - *stream_writing = &stream->writing; - return r; -} - -void grpc_chttp2_list_remove_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); -} - void grpc_chttp2_list_add_parsing_seen_stream( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing) { diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index 041c4efd1f..54d38f2841 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -44,6 +44,7 @@ int grpc_chttp2_unlocking_check_writes( grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream_global *stream_global; grpc_chttp2_stream_writing *stream_writing; + grpc_chttp2_stream_global *first_reinserted_stream = NULL; gpr_uint32 window_delta; /* simple writes are queued to qbuf, and flushed here */ @@ -64,51 +65,53 @@ int grpc_chttp2_unlocking_check_writes( } /* for each grpc_chttp2_stream that's become writable, frame it's data - (according to - available window sizes) and add to the output buffer */ - while (transport_global->outgoing_window > 0 && - grpc_chttp2_list_pop_writable_stream(transport_global, - transport_writing, &stream_global, - &stream_writing)) { + (according to available window sizes) and add to the output buffer */ + while (grpc_chttp2_list_pop_writable_stream( + transport_global, transport_writing, &stream_global, &stream_writing)) { + if (stream_global == first_reinserted_stream) { + /* prevent infinite loop */ + grpc_chttp2_list_add_first_writable_stream(transport_global, + stream_global); + break; + } + stream_writing->id = stream_global->id; - window_delta = grpc_chttp2_preencode( - stream_global->outgoing_sopb->ops, &stream_global->outgoing_sopb->nops, - GPR_MIN(transport_global->outgoing_window, - stream_global->outgoing_window), - &stream_writing->sopb); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "write", transport_global, outgoing_window, -(gpr_int64)window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, - outgoing_window, -(gpr_int64)window_delta); - transport_global->outgoing_window -= window_delta; - stream_global->outgoing_window -= window_delta; - - if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE && - stream_global->outgoing_sopb->nops == 0) { - if (!transport_global->is_client && !stream_global->read_closed) { - stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM; - } else { - stream_writing->send_closed = GRPC_SEND_CLOSED; + stream_writing->send_closed = GRPC_DONT_SEND_CLOSED; + + if (stream_global->outgoing_sopb) { + window_delta = + grpc_chttp2_preencode(stream_global->outgoing_sopb->ops, + &stream_global->outgoing_sopb->nops, + GPR_MIN(transport_global->outgoing_window, + stream_global->outgoing_window), + &stream_writing->sopb); + GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( + "write", transport_global, outgoing_window, -(gpr_int64)window_delta); + GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, + outgoing_window, + -(gpr_int64)window_delta); + transport_global->outgoing_window -= window_delta; + stream_global->outgoing_window -= window_delta; + + if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE && + stream_global->outgoing_sopb->nops == 0) { + if (!transport_global->is_client && !stream_global->read_closed) { + stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM; + } else { + stream_writing->send_closed = GRPC_SEND_CLOSED; + } } - } - if (stream_writing->sopb.nops > 0 || - stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { - grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); - } - if (stream_global->outgoing_window > 0 && - stream_global->outgoing_sopb->nops != 0) { - grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + if (stream_global->outgoing_window > 0 && + stream_global->outgoing_sopb->nops != 0) { + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + if (first_reinserted_stream == NULL && + transport_global->outgoing_window == 0) { + first_reinserted_stream = stream_global; + } + } } - } - /* for each grpc_chttp2_stream that wants to update its window, add that - * window here */ - while (grpc_chttp2_list_pop_writable_window_update_stream(transport_global, - transport_writing, - &stream_global, - &stream_writing)) { - stream_writing->id = stream_global->id; if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) { stream_writing->announce_window = stream_global->unannounced_incoming_window; GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, @@ -119,6 +122,11 @@ int grpc_chttp2_unlocking_check_writes( stream_global->unannounced_incoming_window = 0; grpc_chttp2_list_add_incoming_window_updated(transport_global, stream_global); + stream_global->writing_now = 1; + grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); + } else if (stream_writing->sopb.nops > 0 || + stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { + stream_global->writing_now = 1; grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); } } @@ -206,6 +214,8 @@ void grpc_chttp2_cleanup_writing( while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { + GPR_ASSERT(stream_global->writing_now); + stream_global->writing_now = 0; if (stream_global->outgoing_sopb != NULL && stream_global->outgoing_sopb->nops == 0) { stream_global->outgoing_sopb = NULL; @@ -219,6 +229,9 @@ void grpc_chttp2_cleanup_writing( } grpc_chttp2_list_add_read_write_state_changed(transport_global, stream_global); + } else if (stream_global->read_closed) { + grpc_chttp2_list_add_read_write_state_changed(transport_global, + stream_global); } } transport_writing->outbuf.count = 0; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index eb435a2ee8..0540252546 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -395,7 +395,6 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { } grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global); - grpc_chttp2_list_remove_writable_window_update_stream(&t->global, &s->global); gpr_mu_unlock(&t->mu); @@ -576,8 +575,6 @@ static void maybe_start_some_streams( grpc_chttp2_list_add_incoming_window_updated(transport_global, stream_global); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); } /* cancel out streams that will never be started */ @@ -643,8 +640,7 @@ static void perform_stream_op_locked( if (stream_global->id != 0) { grpc_chttp2_list_add_read_write_state_changed(transport_global, stream_global); - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } @@ -752,6 +748,7 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) { if (!s) { s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id); } + grpc_chttp2_list_remove_writable_stream(&t->global, &s->global); GPR_ASSERT(s); s->global.in_stream_map = 0; if (t->parsing.incoming_stream == &s->parsing) { @@ -833,6 +830,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) { if (!stream_global->publish_sopb) { continue; } + if (stream_global->writing_now) { + continue; + } /* FIXME(ctiller): we include in_stream_map in our computation of whether the stream is write-closed. This is completely bogus, but has the effect of delaying stream-closed until the stream -- cgit v1.2.3 From 05cc0c4ba0efe3b838eacf364ea4a6413a8b5afe Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Jul 2015 16:01:27 -0700 Subject: Integration fix --- src/core/transport/chttp2/writing.c | 8 +++----- src/core/transport/chttp2_transport.c | 6 +++++- src/core/transport/transport_op_string.c | 10 +++++----- 3 files changed, 13 insertions(+), 11 deletions(-) (limited to 'src/core') diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index 54d38f2841..d39b0c42f7 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -77,6 +77,7 @@ int grpc_chttp2_unlocking_check_writes( stream_writing->id = stream_global->id; stream_writing->send_closed = GRPC_DONT_SEND_CLOSED; + GPR_ASSERT(!stream_global->writing_now); if (stream_global->outgoing_sopb) { window_delta = @@ -227,12 +228,9 @@ void grpc_chttp2_cleanup_writing( if (!transport_global->is_client) { stream_global->read_closed = 1; } - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); - } else if (stream_global->read_closed) { - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); } + grpc_chttp2_list_add_read_write_state_changed(transport_global, + stream_global); } transport_writing->outbuf.count = 0; transport_writing->outbuf.length = 0; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 0540252546..bca6090b6b 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -399,7 +399,11 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { gpr_mu_unlock(&t->mu); for (i = 0; i < STREAM_LIST_COUNT; i++) { - GPR_ASSERT(!s->included[i]); + if (s->included[i]) { + gpr_log(GPR_ERROR, "%s stream %d still included in list %d", + t->global.is_client ? "client" : "server", s->global.id, i); + abort(); + } } GPR_ASSERT(s->global.outgoing_sopb == NULL); diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 10d796fc15..f62c340e97 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -116,10 +116,9 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { if (op->send_ops) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_strvec_add(&b, gpr_strdup("SEND")); - if (op->is_last_send) { - gpr_strvec_add(&b, gpr_strdup("_LAST")); - } + gpr_asprintf(&tmp, "SEND%s:%p", op->is_last_send ? "_LAST" : "", + op->on_done_send); + gpr_strvec_add(&b, tmp); gpr_strvec_add(&b, gpr_strdup("[")); gpr_strvec_add(&b, grpc_sopb_string(op->send_ops)); gpr_strvec_add(&b, gpr_strdup("]")); @@ -128,7 +127,8 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { if (op->recv_ops) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "RECV:max_recv_bytes=%d", op->max_recv_bytes); + gpr_asprintf(&tmp, "RECV:%p:max_recv_bytes=%d", op->on_done_recv, + op->max_recv_bytes); gpr_strvec_add(&b, tmp); } -- cgit v1.2.3 From e3086f812e9a55ab665f4fbf49ac27861d2b6aea Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Jul 2015 16:25:00 -0700 Subject: Integration fix --- src/core/transport/chttp2_transport.c | 1 + 1 file changed, 1 insertion(+) (limited to 'src/core') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index bca6090b6b..5f49b2ddd6 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -395,6 +395,7 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { } grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global); + grpc_chttp2_list_remove_writable_stream(&t->global, &s->global); gpr_mu_unlock(&t->mu); -- cgit v1.2.3 From 3f7809d89cc677893a51da26cbf0b371167a46c4 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 24 Jul 2015 13:20:40 -0700 Subject: fix getpeername code on windows --- src/core/iomgr/tcp_server_windows.c | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) (limited to 'src/core') diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 8f634fcd7a..a175398825 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -250,6 +250,7 @@ static void on_accept(void *arg, int from_iocp) { DWORD transfered_bytes; DWORD flags; BOOL wsa_success; + int err; /* The general mechanism for shutting down is to queue abortion calls. While this is necessary in the read/write case, it's useless for the accept @@ -281,8 +282,21 @@ static void on_accept(void *arg, int from_iocp) { } } else { if (!sp->shutting_down) { - getpeername(sock, (struct sockaddr*)&peer_name, &peer_name_len); - peer_name_string = grpc_sockaddr_to_uri((struct sockaddr*)&peer_name); + peer_name_string = NULL; + err = setsockopt(sock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char *)&sp->socket->socket, sizeof(sp->socket->socket)); + if (err) { + char *utf8_message = gpr_format_message(WSAGetLastError()); + gpr_log(GPR_ERROR, "setsockopt error: %s", utf8_message); + gpr_free(utf8_message); + } + err = getpeername(sock, (struct sockaddr*)&peer_name, &peer_name_len); + if (!err) { + peer_name_string = grpc_sockaddr_to_uri((struct sockaddr*)&peer_name); + } else { + char *utf8_message = gpr_format_message(WSAGetLastError()); + gpr_log(GPR_ERROR, "getpeername error: %s", utf8_message); + gpr_free(utf8_message); + } gpr_asprintf(&fd_name, "tcp_server:%s", peer_name_string); ep = grpc_tcp_create(grpc_winsocket_create(sock, fd_name), peer_name_string); gpr_free(fd_name); -- cgit v1.2.3 From d298a9528ace23650b9b51cadb85312ec68e0ec2 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 24 Jul 2015 13:45:56 -0700 Subject: fix whitespaces and line length --- src/core/iomgr/tcp_server_windows.c | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) (limited to 'src/core') diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index a175398825..cc680507ff 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -283,10 +283,12 @@ static void on_accept(void *arg, int from_iocp) { } else { if (!sp->shutting_down) { peer_name_string = NULL; - err = setsockopt(sock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char *)&sp->socket->socket, sizeof(sp->socket->socket)); + err = setsockopt(sock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, + (char *)&sp->socket->socket, + sizeof(sp->socket->socket)); if (err) { char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "setsockopt error: %s", utf8_message); + gpr_log(GPR_ERROR, "setsockopt error: %s", utf8_message); gpr_free(utf8_message); } err = getpeername(sock, (struct sockaddr*)&peer_name, &peer_name_len); @@ -298,7 +300,8 @@ static void on_accept(void *arg, int from_iocp) { gpr_free(utf8_message); } gpr_asprintf(&fd_name, "tcp_server:%s", peer_name_string); - ep = grpc_tcp_create(grpc_winsocket_create(sock, fd_name), peer_name_string); + ep = grpc_tcp_create(grpc_winsocket_create(sock, fd_name), + peer_name_string); gpr_free(fd_name); gpr_free(peer_name_string); } -- cgit v1.2.3