chore(deps): bump google.golang.org/api from 0.134.0 to 0.136.0 (#2681)
Bumps [google.golang.org/api](https://github.com/googleapis/google-api-go-client) from 0.134.0 to 0.136.0. - [Release notes](https://github.com/googleapis/google-api-go-client/releases) - [Changelog](https://github.com/googleapis/google-api-go-client/blob/main/CHANGES.md) - [Commits](https://github.com/googleapis/google-api-go-client/compare/v0.134.0...v0.136.0) --- updated-dependencies: - dependency-name: google.golang.org/api dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
This commit is contained in:
dependabot[bot]
GitHub
parent
232c77de5d
commit
dd8315ab78
16
go.mod
16
go.mod
|
|
@ -48,10 +48,10 @@ require (
|
|||
)
|
||||
|
||||
require (
|
||||
cloud.google.com/go v0.110.4 // indirect
|
||||
cloud.google.com/go/compute v1.20.1 // indirect
|
||||
cloud.google.com/go v0.110.6 // indirect
|
||||
cloud.google.com/go/compute v1.23.0 // indirect
|
||||
cloud.google.com/go/compute/metadata v0.2.3 // indirect
|
||||
cloud.google.com/go/iam v1.1.0 // indirect
|
||||
cloud.google.com/go/iam v1.1.1 // indirect
|
||||
github.com/Azure/azure-pipeline-go v0.2.3 // indirect
|
||||
github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
|
||||
github.com/Azure/go-autorest v14.2.0+incompatible // indirect
|
||||
|
|
@ -149,10 +149,10 @@ require (
|
|||
golang.org/x/text v0.12.0 // indirect
|
||||
golang.org/x/time v0.1.0 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
|
||||
google.golang.org/api v0.134.0
|
||||
google.golang.org/api v0.136.0
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 // indirect
|
||||
google.golang.org/grpc v1.56.2 // indirect
|
||||
google.golang.org/genproto v0.0.0-20230803162519-f966b187b2e5 // indirect
|
||||
google.golang.org/grpc v1.57.0 // indirect
|
||||
google.golang.org/protobuf v1.31.0 // indirect
|
||||
gopkg.in/warnings.v0 v0.1.2 // indirect
|
||||
)
|
||||
|
|
@ -179,8 +179,8 @@ require (
|
|||
github.com/spf13/viper v1.8.1 // indirect
|
||||
github.com/subosito/gotenv v1.2.0 // indirect
|
||||
github.com/toqueteos/webbrowser v1.2.0 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230720185612-659f7aaaa771 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577 // indirect
|
||||
gopkg.in/ini.v1 v1.62.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
)
|
||||
|
|
|
|||
32
go.sum
32
go.sum
|
|
@ -20,23 +20,23 @@ cloud.google.com/go v0.75.0/go.mod h1:VGuuCn7PG0dwsd5XPVm2Mm3wlh3EL55/79EKB6hlPT
|
|||
cloud.google.com/go v0.78.0/go.mod h1:QjdrLG0uq+YwhjoVOLsS1t7TW8fs36kLs4XO5R5ECHg=
|
||||
cloud.google.com/go v0.79.0/go.mod h1:3bzgcEeQlzbuEAYu4mrWhKqWjmpprinYgKJLgKHnbb8=
|
||||
cloud.google.com/go v0.81.0/go.mod h1:mk/AM35KwGk/Nm2YSeZbxXdrNK3KZOYHmLkOqC2V6E0=
|
||||
cloud.google.com/go v0.110.4 h1:1JYyxKMN9hd5dR2MYTPWkGUgcoxVVhg0LKNKEo0qvmk=
|
||||
cloud.google.com/go v0.110.4/go.mod h1:+EYjdK8e5RME/VY/qLCAtuyALQ9q67dvuum8i+H5xsI=
|
||||
cloud.google.com/go v0.110.6 h1:8uYAkj3YHTP/1iwReuHPxLSbdcyc+dSBbzFMrVwDR6Q=
|
||||
cloud.google.com/go v0.110.6/go.mod h1:+EYjdK8e5RME/VY/qLCAtuyALQ9q67dvuum8i+H5xsI=
|
||||
cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
|
||||
cloud.google.com/go/bigquery v1.3.0/go.mod h1:PjpwJnslEMmckchkHFfq+HTD2DmtT67aNFKH1/VBDHE=
|
||||
cloud.google.com/go/bigquery v1.4.0/go.mod h1:S8dzgnTigyfTmLBfrtrhyYhwRxG72rYxvftPBK2Dvzc=
|
||||
cloud.google.com/go/bigquery v1.5.0/go.mod h1:snEHRnqQbz117VIFhE8bmtwIDY80NLUZUMb4Nv6dBIg=
|
||||
cloud.google.com/go/bigquery v1.7.0/go.mod h1://okPTzCYNXSlb24MZs83e2Do+h+VXtc4gLoIoXIAPc=
|
||||
cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM77hZzJN/fQ=
|
||||
cloud.google.com/go/compute v1.20.1 h1:6aKEtlUiwEpJzM001l0yFkpXmUVXaN8W+fbkb2AZNbg=
|
||||
cloud.google.com/go/compute v1.20.1/go.mod h1:4tCnrn48xsqlwSAiLf1HXMQk8CONslYbdiEZc9FEIbM=
|
||||
cloud.google.com/go/compute v1.23.0 h1:tP41Zoavr8ptEqaW6j+LQOnyBBhO7OkOMAGrgLopTwY=
|
||||
cloud.google.com/go/compute v1.23.0/go.mod h1:4tCnrn48xsqlwSAiLf1HXMQk8CONslYbdiEZc9FEIbM=
|
||||
cloud.google.com/go/compute/metadata v0.2.3 h1:mg4jlk7mCAj6xXp9UJ4fjI9VUI5rubuGBW5aJ7UnBMY=
|
||||
cloud.google.com/go/compute/metadata v0.2.3/go.mod h1:VAV5nSsACxMJvgaAuX6Pk2AawlZn8kiOGuCv6gTkwuA=
|
||||
cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
|
||||
cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
|
||||
cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqClKRT5SZwBmk=
|
||||
cloud.google.com/go/iam v1.1.0 h1:67gSqaPukx7O8WLLHMa0PNs3EBGd2eE4d+psbO/CO94=
|
||||
cloud.google.com/go/iam v1.1.0/go.mod h1:nxdHjaKfCr7fNYx/HJMM8LgiMugmveWlkatear5gVyk=
|
||||
cloud.google.com/go/iam v1.1.1 h1:lW7fzj15aVIXYHREOqjRBV9PsH0Z6u8Y46a1YGvQP4Y=
|
||||
cloud.google.com/go/iam v1.1.1/go.mod h1:A5avdyVL2tCppe4unb0951eI9jreack+RJ0/d+KUZOU=
|
||||
cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
|
||||
cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
|
||||
cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
|
||||
|
|
@ -1023,8 +1023,8 @@ google.golang.org/api v0.40.0/go.mod h1:fYKFpnQN0DsDSKRVRcQSDQNtqWPfM9i+zNPxepjR
|
|||
google.golang.org/api v0.41.0/go.mod h1:RkxM5lITDfTzmyKFPt+wGrCJbVfniCr2ool8kTBzRTU=
|
||||
google.golang.org/api v0.43.0/go.mod h1:nQsDGjRXMo4lvh5hP0TKqF244gqhGcr/YSIykhUk/94=
|
||||
google.golang.org/api v0.44.0/go.mod h1:EBOGZqzyhtvMDoxwS97ctnh0zUmYY6CxqXsc1AvkYD8=
|
||||
google.golang.org/api v0.134.0 h1:ktL4Goua+UBgoP1eL1/60LwZJqa1sIzkLmvoR3hR6Gw=
|
||||
google.golang.org/api v0.134.0/go.mod h1:sjRL3UnjTx5UqNQS9EWr9N8p7xbHpy1k0XGRLCf3Spk=
|
||||
google.golang.org/api v0.136.0 h1:e/6enzUE1s4tGPa6Q3ZYShKTtvRc+1Jq0rrafhppmOs=
|
||||
google.golang.org/api v0.136.0/go.mod h1:XtJfF+V2zgUxelOn5Zs3kECtluMxneJG8ZxUTlLNTPA=
|
||||
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
|
||||
google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
|
||||
google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
|
||||
|
|
@ -1076,12 +1076,12 @@ google.golang.org/genproto v0.0.0-20210310155132-4ce2db91004e/go.mod h1:FWY/as6D
|
|||
google.golang.org/genproto v0.0.0-20210319143718-93e7006c17a6/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
|
||||
google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaEmUacj36I+k7YKbEc5CXzPIeORRgDAUOu28A=
|
||||
google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
|
||||
google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 h1:Au6te5hbKUV8pIYWHqOUZ1pva5qK/rwbIhoXEUB9Lu8=
|
||||
google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130/go.mod h1:O9kGHb51iE/nOGvQaDUuadVYqovW56s5emA88lQnj6Y=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130 h1:XVeBY8d/FaK4848myy41HBqnDwvxeV3zMZhwN1TvAMU=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130/go.mod h1:mPBs5jNgx2GuQGvFwUvVKqtn6HsUw9nP64BedgvqEsQ=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230720185612-659f7aaaa771 h1:Z8qdAF9GFsmcUuWQ5KVYIpP3PCKydn/YKORnghIalu4=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230720185612-659f7aaaa771/go.mod h1:TUfxEVdsvPg18p6AslUXFoLdpED4oBnGwyqk3dV1XzM=
|
||||
google.golang.org/genproto v0.0.0-20230803162519-f966b187b2e5 h1:L6iMMGrtzgHsWofoFcihmDEMYeDR9KN/ThbPWGrh++g=
|
||||
google.golang.org/genproto v0.0.0-20230803162519-f966b187b2e5/go.mod h1:oH/ZOT02u4kWEp7oYBGYFFkCdKS/uYR9Z7+0/xuuFp8=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5 h1:nIgk/EEq3/YlnmVVXVnm14rC2oxgs1o0ong4sD/rd44=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5/go.mod h1:5DZzOUPCLYL3mNkQ0ms0F3EuUNZ7py1Bqeq6sxzI7/Q=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577 h1:wukfNtZmZUurLN/atp2hiIeTKn7QJWIQdHzqmsOnAOk=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577/go.mod h1:+Bk1OCOj40wS2hwAMA+aCW9ypzm63QTBBHp6lQ3p+9M=
|
||||
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
|
||||
google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
|
||||
google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
|
||||
|
|
@ -1103,8 +1103,8 @@ google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAG
|
|||
google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
|
||||
google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
|
||||
google.golang.org/grpc v1.45.0/go.mod h1:lN7owxKUQEqMfSyQikvvk5tf/6zMPsrK+ONuO11+0rQ=
|
||||
google.golang.org/grpc v1.56.2 h1:fVRFRnXvU+x6C4IlHZewvJOVHoOv1TUuQyoRsYnB4bI=
|
||||
google.golang.org/grpc v1.56.2/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
|
||||
google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
|
||||
google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
|
||||
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
|
||||
google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
|
||||
google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
|
||||
|
|
|
|||
|
|
@ -15,4 +15,4 @@
|
|||
package internal
|
||||
|
||||
// Version is the current tagged release of the library.
|
||||
const Version = "1.20.1"
|
||||
const Version = "1.23.0"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,13 @@
|
|||
# Changes
|
||||
|
||||
|
||||
## [1.1.1](https://github.com/googleapis/google-cloud-go/compare/iam/v1.1.0...iam/v1.1.1) (2023-06-20)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* **iam:** REST query UpdateMask bug ([df52820](https://github.com/googleapis/google-cloud-go/commit/df52820b0e7721954809a8aa8700b93c5662dc9b))
|
||||
|
||||
## [1.1.0](https://github.com/googleapis/google-cloud-go/compare/iam/v1.0.1...iam/v1.1.0) (2023-05-30)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@
|
|||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.30.0
|
||||
// protoc v4.23.1
|
||||
// protoc v4.23.2
|
||||
// source: google/iam/v1/iam_policy.proto
|
||||
|
||||
package iampb
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@
|
|||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.30.0
|
||||
// protoc v4.23.1
|
||||
// protoc v4.23.2
|
||||
// source: google/iam/v1/options.proto
|
||||
|
||||
package iampb
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@
|
|||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.30.0
|
||||
// protoc v4.23.1
|
||||
// protoc v4.23.2
|
||||
// source: google/iam/v1/policy.proto
|
||||
|
||||
package iampb
|
||||
|
|
|
|||
|
|
@ -29,6 +29,16 @@
|
|||
"release_level": "preview",
|
||||
"library_type": "GAPIC_AUTO"
|
||||
},
|
||||
"cloud.google.com/go/ai/generativelanguage/apiv1beta2": {
|
||||
"api_shortname": "generativelanguage",
|
||||
"distribution_name": "cloud.google.com/go/ai/generativelanguage/apiv1beta2",
|
||||
"description": "Generative Language API",
|
||||
"language": "go",
|
||||
"client_library_type": "generated",
|
||||
"client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/ai/latest/generativelanguage/apiv1beta2",
|
||||
"release_level": "preview",
|
||||
"library_type": "GAPIC_AUTO"
|
||||
},
|
||||
"cloud.google.com/go/aiplatform/apiv1": {
|
||||
"api_shortname": "aiplatform",
|
||||
"distribution_name": "cloud.google.com/go/aiplatform/apiv1",
|
||||
|
|
@ -1059,26 +1069,6 @@
|
|||
"release_level": "preview",
|
||||
"library_type": "CORE"
|
||||
},
|
||||
"cloud.google.com/go/gaming/apiv1": {
|
||||
"api_shortname": "gameservices",
|
||||
"distribution_name": "cloud.google.com/go/gaming/apiv1",
|
||||
"description": "Game Services API",
|
||||
"language": "go",
|
||||
"client_library_type": "generated",
|
||||
"client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/gaming/latest/apiv1",
|
||||
"release_level": "stable",
|
||||
"library_type": "GAPIC_AUTO"
|
||||
},
|
||||
"cloud.google.com/go/gaming/apiv1beta": {
|
||||
"api_shortname": "gameservices",
|
||||
"distribution_name": "cloud.google.com/go/gaming/apiv1beta",
|
||||
"description": "Game Services API",
|
||||
"language": "go",
|
||||
"client_library_type": "generated",
|
||||
"client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/gaming/latest/apiv1beta",
|
||||
"release_level": "preview",
|
||||
"library_type": "GAPIC_AUTO"
|
||||
},
|
||||
"cloud.google.com/go/gkebackup/apiv1": {
|
||||
"api_shortname": "gkebackup",
|
||||
"distribution_name": "cloud.google.com/go/gkebackup/apiv1",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,136 @@
|
|||
// Copyright 2017 The Go Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style
|
||||
// license that can be found in the LICENSE file.
|
||||
|
||||
// Package semaphore provides a weighted semaphore implementation.
|
||||
package semaphore // import "golang.org/x/sync/semaphore"
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type waiter struct {
|
||||
n int64
|
||||
ready chan<- struct{} // Closed when semaphore acquired.
|
||||
}
|
||||
|
||||
// NewWeighted creates a new weighted semaphore with the given
|
||||
// maximum combined weight for concurrent access.
|
||||
func NewWeighted(n int64) *Weighted {
|
||||
w := &Weighted{size: n}
|
||||
return w
|
||||
}
|
||||
|
||||
// Weighted provides a way to bound concurrent access to a resource.
|
||||
// The callers can request access with a given weight.
|
||||
type Weighted struct {
|
||||
size int64
|
||||
cur int64
|
||||
mu sync.Mutex
|
||||
waiters list.List
|
||||
}
|
||||
|
||||
// Acquire acquires the semaphore with a weight of n, blocking until resources
|
||||
// are available or ctx is done. On success, returns nil. On failure, returns
|
||||
// ctx.Err() and leaves the semaphore unchanged.
|
||||
//
|
||||
// If ctx is already done, Acquire may still succeed without blocking.
|
||||
func (s *Weighted) Acquire(ctx context.Context, n int64) error {
|
||||
s.mu.Lock()
|
||||
if s.size-s.cur >= n && s.waiters.Len() == 0 {
|
||||
s.cur += n
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
if n > s.size {
|
||||
// Don't make other Acquire calls block on one that's doomed to fail.
|
||||
s.mu.Unlock()
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
ready := make(chan struct{})
|
||||
w := waiter{n: n, ready: ready}
|
||||
elem := s.waiters.PushBack(w)
|
||||
s.mu.Unlock()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
err := ctx.Err()
|
||||
s.mu.Lock()
|
||||
select {
|
||||
case <-ready:
|
||||
// Acquired the semaphore after we were canceled. Rather than trying to
|
||||
// fix up the queue, just pretend we didn't notice the cancelation.
|
||||
err = nil
|
||||
default:
|
||||
isFront := s.waiters.Front() == elem
|
||||
s.waiters.Remove(elem)
|
||||
// If we're at the front and there're extra tokens left, notify other waiters.
|
||||
if isFront && s.size > s.cur {
|
||||
s.notifyWaiters()
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
return err
|
||||
|
||||
case <-ready:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// TryAcquire acquires the semaphore with a weight of n without blocking.
|
||||
// On success, returns true. On failure, returns false and leaves the semaphore unchanged.
|
||||
func (s *Weighted) TryAcquire(n int64) bool {
|
||||
s.mu.Lock()
|
||||
success := s.size-s.cur >= n && s.waiters.Len() == 0
|
||||
if success {
|
||||
s.cur += n
|
||||
}
|
||||
s.mu.Unlock()
|
||||
return success
|
||||
}
|
||||
|
||||
// Release releases the semaphore with a weight of n.
|
||||
func (s *Weighted) Release(n int64) {
|
||||
s.mu.Lock()
|
||||
s.cur -= n
|
||||
if s.cur < 0 {
|
||||
s.mu.Unlock()
|
||||
panic("semaphore: released more than held")
|
||||
}
|
||||
s.notifyWaiters()
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Weighted) notifyWaiters() {
|
||||
for {
|
||||
next := s.waiters.Front()
|
||||
if next == nil {
|
||||
break // No more waiters blocked.
|
||||
}
|
||||
|
||||
w := next.Value.(waiter)
|
||||
if s.size-s.cur < w.n {
|
||||
// Not enough tokens for the next waiter. We could keep going (to try to
|
||||
// find a waiter with a smaller request), but under load that could cause
|
||||
// starvation for large requests; instead, we leave all remaining waiters
|
||||
// blocked.
|
||||
//
|
||||
// Consider a semaphore used as a read-write lock, with N tokens, N
|
||||
// readers, and one writer. Each reader can Acquire(1) to obtain a read
|
||||
// lock. The writer can Acquire(N) to obtain a write lock, excluding all
|
||||
// of the readers. If we allow the readers to jump ahead in the queue,
|
||||
// the writer will starve — there is always one token available for every
|
||||
// reader.
|
||||
break
|
||||
}
|
||||
|
||||
s.cur += w.n
|
||||
s.waiters.Remove(next)
|
||||
close(w.ready)
|
||||
}
|
||||
}
|
||||
|
|
@ -91,16 +91,10 @@ func getTransportConfig(settings *DialSettings) (*transportConfig, error) {
|
|||
s2aMTLSEndpoint: "",
|
||||
}
|
||||
|
||||
// Check the env to determine whether to use S2A.
|
||||
if !isGoogleS2AEnabled() {
|
||||
if !shouldUseS2A(clientCertSource, settings) {
|
||||
return &defaultTransportConfig, nil
|
||||
}
|
||||
|
||||
// If client cert is found, use that over S2A.
|
||||
// If MTLS is not enabled for the endpoint, skip S2A.
|
||||
if clientCertSource != nil || !mtlsEndpointEnabledForS2A() {
|
||||
return &defaultTransportConfig, nil
|
||||
}
|
||||
s2aMTLSEndpoint := settings.DefaultMTLSEndpoint
|
||||
// If there is endpoint override, honor it.
|
||||
if settings.Endpoint != "" {
|
||||
|
|
@ -118,10 +112,6 @@ func getTransportConfig(settings *DialSettings) (*transportConfig, error) {
|
|||
}, nil
|
||||
}
|
||||
|
||||
func isGoogleS2AEnabled() bool {
|
||||
return strings.ToLower(os.Getenv(googleAPIUseS2AEnv)) == "true"
|
||||
}
|
||||
|
||||
// getClientCertificateSource returns a default client certificate source, if
|
||||
// not provided by the user.
|
||||
//
|
||||
|
|
@ -275,8 +265,36 @@ func GetHTTPTransportConfigAndEndpoint(settings *DialSettings) (cert.Source, fun
|
|||
return nil, dialTLSContextFunc, config.s2aMTLSEndpoint, nil
|
||||
}
|
||||
|
||||
func shouldUseS2A(clientCertSource cert.Source, settings *DialSettings) bool {
|
||||
// If client cert is found, use that over S2A.
|
||||
if clientCertSource != nil {
|
||||
return false
|
||||
}
|
||||
// If EXPERIMENTAL_GOOGLE_API_USE_S2A is not set to true, skip S2A.
|
||||
if !isGoogleS2AEnabled() {
|
||||
return false
|
||||
}
|
||||
// If DefaultMTLSEndpoint is not set, skip S2A.
|
||||
if settings.DefaultMTLSEndpoint == "" {
|
||||
return false
|
||||
}
|
||||
// If MTLS is not enabled for this endpoint, skip S2A.
|
||||
if !mtlsEndpointEnabledForS2A() {
|
||||
return false
|
||||
}
|
||||
// If custom HTTP client is provided, skip S2A.
|
||||
if settings.HTTPClient != nil {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// mtlsEndpointEnabledForS2A checks if the endpoint is indeed MTLS-enabled, so that we can use S2A for MTLS connection.
|
||||
var mtlsEndpointEnabledForS2A = func() bool {
|
||||
// TODO(xmenxk): determine this via discovery config.
|
||||
return true
|
||||
}
|
||||
|
||||
func isGoogleS2AEnabled() bool {
|
||||
return strings.ToLower(os.Getenv(googleAPIUseS2AEnv)) == "true"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,4 +5,4 @@
|
|||
package internal
|
||||
|
||||
// Version is the current tagged release of the library.
|
||||
const Version = "0.134.0"
|
||||
const Version = "0.136.0"
|
||||
|
|
|
|||
|
|
@ -112,19 +112,31 @@ func (a *Attributes) String() string {
|
|||
sb.WriteString("{")
|
||||
first := true
|
||||
for k, v := range a.m {
|
||||
var key, val string
|
||||
if str, ok := k.(interface{ String() string }); ok {
|
||||
key = str.String()
|
||||
}
|
||||
if str, ok := v.(interface{ String() string }); ok {
|
||||
val = str.String()
|
||||
}
|
||||
if !first {
|
||||
sb.WriteString(", ")
|
||||
}
|
||||
sb.WriteString(fmt.Sprintf("%q: %q, ", key, val))
|
||||
sb.WriteString(fmt.Sprintf("%q: %q ", str(k), str(v)))
|
||||
first = false
|
||||
}
|
||||
sb.WriteString("}")
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func str(x interface{}) string {
|
||||
if v, ok := x.(fmt.Stringer); ok {
|
||||
return v.String()
|
||||
} else if v, ok := x.(string); ok {
|
||||
return v
|
||||
}
|
||||
return fmt.Sprintf("<%p>", x)
|
||||
}
|
||||
|
||||
// MarshalJSON helps implement the json.Marshaler interface, thereby rendering
|
||||
// the Attributes correctly when printing (via pretty.JSON) structs containing
|
||||
// Attributes as fields.
|
||||
//
|
||||
// Is it impossible to unmarshal attributes from a JSON representation and this
|
||||
// method is meant only for debugging purposes.
|
||||
func (a *Attributes) MarshalJSON() ([]byte, error) {
|
||||
return []byte(a.String()), nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ import (
|
|||
"google.golang.org/grpc/internal/backoff"
|
||||
"google.golang.org/grpc/internal/channelz"
|
||||
"google.golang.org/grpc/internal/grpcsync"
|
||||
"google.golang.org/grpc/internal/pretty"
|
||||
iresolver "google.golang.org/grpc/internal/resolver"
|
||||
"google.golang.org/grpc/internal/transport"
|
||||
"google.golang.org/grpc/keepalive"
|
||||
|
|
@ -867,6 +868,20 @@ func (cc *ClientConn) handleSubConnStateChange(sc balancer.SubConn, s connectivi
|
|||
cc.balancerWrapper.updateSubConnState(sc, s, err)
|
||||
}
|
||||
|
||||
// Makes a copy of the input addresses slice and clears out the balancer
|
||||
// attributes field. Addresses are passed during subconn creation and address
|
||||
// update operations. In both cases, we will clear the balancer attributes by
|
||||
// calling this function, and therefore we will be able to use the Equal method
|
||||
// provided by the resolver.Address type for comparison.
|
||||
func copyAddressesWithoutBalancerAttributes(in []resolver.Address) []resolver.Address {
|
||||
out := make([]resolver.Address, len(in))
|
||||
for i := range in {
|
||||
out[i] = in[i]
|
||||
out[i].BalancerAttributes = nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// newAddrConn creates an addrConn for addrs and adds it to cc.conns.
|
||||
//
|
||||
// Caller needs to make sure len(addrs) > 0.
|
||||
|
|
@ -874,7 +889,7 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
|
|||
ac := &addrConn{
|
||||
state: connectivity.Idle,
|
||||
cc: cc,
|
||||
addrs: addrs,
|
||||
addrs: copyAddressesWithoutBalancerAttributes(addrs),
|
||||
scopts: opts,
|
||||
dopts: cc.dopts,
|
||||
czData: new(channelzData),
|
||||
|
|
@ -995,8 +1010,9 @@ func equalAddresses(a, b []resolver.Address) bool {
|
|||
// connections or connection attempts.
|
||||
func (ac *addrConn) updateAddrs(addrs []resolver.Address) {
|
||||
ac.mu.Lock()
|
||||
channelz.Infof(logger, ac.channelzID, "addrConn: updateAddrs curAddr: %v, addrs: %v", ac.curAddr, addrs)
|
||||
channelz.Infof(logger, ac.channelzID, "addrConn: updateAddrs curAddr: %v, addrs: %v", pretty.ToJSON(ac.curAddr), pretty.ToJSON(addrs))
|
||||
|
||||
addrs = copyAddressesWithoutBalancerAttributes(addrs)
|
||||
if equalAddresses(ac.addrs, addrs) {
|
||||
ac.mu.Unlock()
|
||||
return
|
||||
|
|
@ -1807,19 +1823,70 @@ func (cc *ClientConn) parseTargetAndFindResolver() error {
|
|||
}
|
||||
|
||||
// parseTarget uses RFC 3986 semantics to parse the given target into a
|
||||
// resolver.Target struct containing scheme, authority and url. Query
|
||||
// params are stripped from the endpoint.
|
||||
// resolver.Target struct containing url. Query params are stripped from the
|
||||
// endpoint.
|
||||
func parseTarget(target string) (resolver.Target, error) {
|
||||
u, err := url.Parse(target)
|
||||
if err != nil {
|
||||
return resolver.Target{}, err
|
||||
}
|
||||
|
||||
return resolver.Target{
|
||||
Scheme: u.Scheme,
|
||||
Authority: u.Host,
|
||||
URL: *u,
|
||||
}, nil
|
||||
return resolver.Target{URL: *u}, nil
|
||||
}
|
||||
|
||||
func encodeAuthority(authority string) string {
|
||||
const upperhex = "0123456789ABCDEF"
|
||||
|
||||
// Return for characters that must be escaped as per
|
||||
// Valid chars are mentioned here:
|
||||
// https://datatracker.ietf.org/doc/html/rfc3986#section-3.2
|
||||
shouldEscape := func(c byte) bool {
|
||||
// Alphanum are always allowed.
|
||||
if 'a' <= c && c <= 'z' || 'A' <= c && c <= 'Z' || '0' <= c && c <= '9' {
|
||||
return false
|
||||
}
|
||||
switch c {
|
||||
case '-', '_', '.', '~': // Unreserved characters
|
||||
return false
|
||||
case '!', '$', '&', '\'', '(', ')', '*', '+', ',', ';', '=': // Subdelim characters
|
||||
return false
|
||||
case ':', '[', ']', '@': // Authority related delimeters
|
||||
return false
|
||||
}
|
||||
// Everything else must be escaped.
|
||||
return true
|
||||
}
|
||||
|
||||
hexCount := 0
|
||||
for i := 0; i < len(authority); i++ {
|
||||
c := authority[i]
|
||||
if shouldEscape(c) {
|
||||
hexCount++
|
||||
}
|
||||
}
|
||||
|
||||
if hexCount == 0 {
|
||||
return authority
|
||||
}
|
||||
|
||||
required := len(authority) + 2*hexCount
|
||||
t := make([]byte, required)
|
||||
|
||||
j := 0
|
||||
// This logic is a barebones version of escape in the go net/url library.
|
||||
for i := 0; i < len(authority); i++ {
|
||||
switch c := authority[i]; {
|
||||
case shouldEscape(c):
|
||||
t[j] = '%'
|
||||
t[j+1] = upperhex[c>>4]
|
||||
t[j+2] = upperhex[c&15]
|
||||
j += 3
|
||||
default:
|
||||
t[j] = authority[i]
|
||||
j++
|
||||
}
|
||||
}
|
||||
return string(t)
|
||||
}
|
||||
|
||||
// Determine channel authority. The order of precedence is as follows:
|
||||
|
|
@ -1872,7 +1939,11 @@ func (cc *ClientConn) determineAuthority() error {
|
|||
// the channel authority given the user's dial target. For resolvers
|
||||
// which don't implement this interface, we will use the endpoint from
|
||||
// "scheme://authority/endpoint" as the default authority.
|
||||
cc.authority = endpoint
|
||||
// Escape the endpoint to handle use cases where the endpoint
|
||||
// might not be a valid authority by default.
|
||||
// For example an endpoint which has multiple paths like
|
||||
// 'a/b/c', which is not a valid authority by default.
|
||||
cc.authority = encodeAuthority(endpoint)
|
||||
}
|
||||
channelz.Infof(logger, cc.channelzID, "Channel authority set to %q", cc.authority)
|
||||
return nil
|
||||
|
|
|
|||
57
vendor/google.golang.org/grpc/credentials/alts/internal/handshaker/handshaker.go
generated
vendored
57
vendor/google.golang.org/grpc/credentials/alts/internal/handshaker/handshaker.go
generated
vendored
|
|
@ -25,8 +25,8 @@ import (
|
|||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
|
||||
"golang.org/x/sync/semaphore"
|
||||
grpc "google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials"
|
||||
|
|
@ -35,15 +35,13 @@ import (
|
|||
"google.golang.org/grpc/credentials/alts/internal/conn"
|
||||
altsgrpc "google.golang.org/grpc/credentials/alts/internal/proto/grpc_gcp"
|
||||
altspb "google.golang.org/grpc/credentials/alts/internal/proto/grpc_gcp"
|
||||
"google.golang.org/grpc/internal/envconfig"
|
||||
)
|
||||
|
||||
const (
|
||||
// The maximum byte size of receive frames.
|
||||
frameLimit = 64 * 1024 // 64 KB
|
||||
rekeyRecordProtocolName = "ALTSRP_GCM_AES128_REKEY"
|
||||
// maxPendingHandshakes represents the maximum number of concurrent
|
||||
// handshakes.
|
||||
maxPendingHandshakes = 100
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
@ -59,9 +57,9 @@ var (
|
|||
return conn.NewAES128GCMRekey(s, keyData)
|
||||
},
|
||||
}
|
||||
// control number of concurrent created (but not closed) handshakers.
|
||||
mu sync.Mutex
|
||||
concurrentHandshakes = int64(0)
|
||||
// control number of concurrent created (but not closed) handshakes.
|
||||
clientHandshakes = semaphore.NewWeighted(int64(envconfig.ALTSMaxConcurrentHandshakes))
|
||||
serverHandshakes = semaphore.NewWeighted(int64(envconfig.ALTSMaxConcurrentHandshakes))
|
||||
// errDropped occurs when maxPendingHandshakes is reached.
|
||||
errDropped = errors.New("maximum number of concurrent ALTS handshakes is reached")
|
||||
// errOutOfBound occurs when the handshake service returns a consumed
|
||||
|
|
@ -77,30 +75,6 @@ func init() {
|
|||
}
|
||||
}
|
||||
|
||||
func acquire() bool {
|
||||
mu.Lock()
|
||||
// If we need n to be configurable, we can pass it as an argument.
|
||||
n := int64(1)
|
||||
success := maxPendingHandshakes-concurrentHandshakes >= n
|
||||
if success {
|
||||
concurrentHandshakes += n
|
||||
}
|
||||
mu.Unlock()
|
||||
return success
|
||||
}
|
||||
|
||||
func release() {
|
||||
mu.Lock()
|
||||
// If we need n to be configurable, we can pass it as an argument.
|
||||
n := int64(1)
|
||||
concurrentHandshakes -= n
|
||||
if concurrentHandshakes < 0 {
|
||||
mu.Unlock()
|
||||
panic("bad release")
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
|
||||
// ClientHandshakerOptions contains the client handshaker options that can
|
||||
// provided by the caller.
|
||||
type ClientHandshakerOptions struct {
|
||||
|
|
@ -134,10 +108,6 @@ func DefaultServerHandshakerOptions() *ServerHandshakerOptions {
|
|||
return &ServerHandshakerOptions{}
|
||||
}
|
||||
|
||||
// TODO: add support for future local and remote endpoint in both client options
|
||||
// and server options (server options struct does not exist now. When
|
||||
// caller can provide endpoints, it should be created.
|
||||
|
||||
// altsHandshaker is used to complete an ALTS handshake between client and
|
||||
// server. This handshaker talks to the ALTS handshaker service in the metadata
|
||||
// server.
|
||||
|
|
@ -185,10 +155,10 @@ func NewServerHandshaker(ctx context.Context, conn *grpc.ClientConn, c net.Conn,
|
|||
// ClientHandshake starts and completes a client ALTS handshake for GCP. Once
|
||||
// done, ClientHandshake returns a secure connection.
|
||||
func (h *altsHandshaker) ClientHandshake(ctx context.Context) (net.Conn, credentials.AuthInfo, error) {
|
||||
if !acquire() {
|
||||
if !clientHandshakes.TryAcquire(1) {
|
||||
return nil, nil, errDropped
|
||||
}
|
||||
defer release()
|
||||
defer clientHandshakes.Release(1)
|
||||
|
||||
if h.side != core.ClientSide {
|
||||
return nil, nil, errors.New("only handshakers created using NewClientHandshaker can perform a client handshaker")
|
||||
|
|
@ -238,10 +208,10 @@ func (h *altsHandshaker) ClientHandshake(ctx context.Context) (net.Conn, credent
|
|||
// ServerHandshake starts and completes a server ALTS handshake for GCP. Once
|
||||
// done, ServerHandshake returns a secure connection.
|
||||
func (h *altsHandshaker) ServerHandshake(ctx context.Context) (net.Conn, credentials.AuthInfo, error) {
|
||||
if !acquire() {
|
||||
if !serverHandshakes.TryAcquire(1) {
|
||||
return nil, nil, errDropped
|
||||
}
|
||||
defer release()
|
||||
defer serverHandshakes.Release(1)
|
||||
|
||||
if h.side != core.ServerSide {
|
||||
return nil, nil, errors.New("only handshakers created using NewServerHandshaker can perform a server handshaker")
|
||||
|
|
@ -264,8 +234,6 @@ func (h *altsHandshaker) ServerHandshake(ctx context.Context) (net.Conn, credent
|
|||
}
|
||||
|
||||
// Prepare server parameters.
|
||||
// TODO: currently only ALTS parameters are provided. Might need to use
|
||||
// more options in the future.
|
||||
params := make(map[int32]*altspb.ServerHandshakeParameters)
|
||||
params[int32(altspb.HandshakeProtocol_ALTS)] = &altspb.ServerHandshakeParameters{
|
||||
RecordProtocols: recordProtocols,
|
||||
|
|
@ -391,3 +359,10 @@ func (h *altsHandshaker) Close() {
|
|||
h.stream.CloseSend()
|
||||
}
|
||||
}
|
||||
|
||||
// ResetConcurrentHandshakeSemaphoreForTesting resets the handshake semaphores
|
||||
// to allow numberOfAllowedHandshakes concurrent handshakes each.
|
||||
func ResetConcurrentHandshakeSemaphoreForTesting(numberOfAllowedHandshakes int64) {
|
||||
clientHandshakes = semaphore.NewWeighted(numberOfAllowedHandshakes)
|
||||
serverHandshakes = semaphore.NewWeighted(numberOfAllowedHandshakes)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -78,6 +78,7 @@ type dialOptions struct {
|
|||
defaultServiceConfigRawJSON *string
|
||||
resolvers []resolver.Builder
|
||||
idleTimeout time.Duration
|
||||
recvBufferPool SharedBufferPool
|
||||
}
|
||||
|
||||
// DialOption configures how we set up the connection.
|
||||
|
|
@ -628,6 +629,7 @@ func defaultDialOptions() dialOptions {
|
|||
ReadBufferSize: defaultReadBufSize,
|
||||
UseProxy: true,
|
||||
},
|
||||
recvBufferPool: nopBufferPool{},
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -676,3 +678,24 @@ func WithIdleTimeout(d time.Duration) DialOption {
|
|||
o.idleTimeout = d
|
||||
})
|
||||
}
|
||||
|
||||
// WithRecvBufferPool returns a DialOption that configures the ClientConn
|
||||
// to use the provided shared buffer pool for parsing incoming messages. Depending
|
||||
// on the application's workload, this could result in reduced memory allocation.
|
||||
//
|
||||
// If you are unsure about how to implement a memory pool but want to utilize one,
|
||||
// begin with grpc.NewSharedBufferPool.
|
||||
//
|
||||
// Note: The shared buffer pool feature will not be active if any of the following
|
||||
// options are used: WithStatsHandler, EnableTracing, or binary logging. In such
|
||||
// cases, the shared buffer pool will be ignored.
|
||||
//
|
||||
// # Experimental
|
||||
//
|
||||
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
|
||||
// later release.
|
||||
func WithRecvBufferPool(bufferPool SharedBufferPool) DialOption {
|
||||
return newFuncDialOption(func(o *dialOptions) {
|
||||
o.recvBufferPool = bufferPool
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,6 +40,9 @@ var (
|
|||
// pick_first LB policy, which can be enabled by setting the environment
|
||||
// variable "GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG" to "true".
|
||||
PickFirstLBConfig = boolFromEnv("GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG", false)
|
||||
// ALTSMaxConcurrentHandshakes is the maximum number of concurrent ALTS
|
||||
// handshakes that can be performed.
|
||||
ALTSMaxConcurrentHandshakes = uint64FromEnv("GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES", 100, 1, 100)
|
||||
)
|
||||
|
||||
func boolFromEnv(envVar string, def bool) bool {
|
||||
|
|
|
|||
|
|
@ -80,6 +80,13 @@ func Uint32() uint32 {
|
|||
return r.Uint32()
|
||||
}
|
||||
|
||||
// ExpFloat64 implements rand.ExpFloat64 on the grpcrand global source.
|
||||
func ExpFloat64() float64 {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return r.ExpFloat64()
|
||||
}
|
||||
|
||||
// Shuffle implements rand.Shuffle on the grpcrand global source.
|
||||
var Shuffle = func(n int, f func(int, int)) {
|
||||
mu.Lock()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,136 @@
|
|||
/*
|
||||
*
|
||||
* Copyright 2023 gRPC authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*
|
||||
*/
|
||||
|
||||
package grpcsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Subscriber represents an entity that is subscribed to messages published on
|
||||
// a PubSub. It wraps the callback to be invoked by the PubSub when a new
|
||||
// message is published.
|
||||
type Subscriber interface {
|
||||
// OnMessage is invoked when a new message is published. Implementations
|
||||
// must not block in this method.
|
||||
OnMessage(msg interface{})
|
||||
}
|
||||
|
||||
// PubSub is a simple one-to-many publish-subscribe system that supports
|
||||
// messages of arbitrary type. It guarantees that messages are delivered in
|
||||
// the same order in which they were published.
|
||||
//
|
||||
// Publisher invokes the Publish() method to publish new messages, while
|
||||
// subscribers interested in receiving these messages register a callback
|
||||
// via the Subscribe() method.
|
||||
//
|
||||
// Once a PubSub is stopped, no more messages can be published, and
|
||||
// it is guaranteed that no more subscriber callback will be invoked.
|
||||
type PubSub struct {
|
||||
cs *CallbackSerializer
|
||||
cancel context.CancelFunc
|
||||
|
||||
// Access to the below fields are guarded by this mutex.
|
||||
mu sync.Mutex
|
||||
msg interface{}
|
||||
subscribers map[Subscriber]bool
|
||||
stopped bool
|
||||
}
|
||||
|
||||
// NewPubSub returns a new PubSub instance.
|
||||
func NewPubSub() *PubSub {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &PubSub{
|
||||
cs: NewCallbackSerializer(ctx),
|
||||
cancel: cancel,
|
||||
subscribers: map[Subscriber]bool{},
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe registers the provided Subscriber to the PubSub.
|
||||
//
|
||||
// If the PubSub contains a previously published message, the Subscriber's
|
||||
// OnMessage() callback will be invoked asynchronously with the existing
|
||||
// message to begin with, and subsequently for every newly published message.
|
||||
//
|
||||
// The caller is responsible for invoking the returned cancel function to
|
||||
// unsubscribe itself from the PubSub.
|
||||
func (ps *PubSub) Subscribe(sub Subscriber) (cancel func()) {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
|
||||
if ps.stopped {
|
||||
return func() {}
|
||||
}
|
||||
|
||||
ps.subscribers[sub] = true
|
||||
|
||||
if ps.msg != nil {
|
||||
msg := ps.msg
|
||||
ps.cs.Schedule(func(context.Context) {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if !ps.subscribers[sub] {
|
||||
return
|
||||
}
|
||||
sub.OnMessage(msg)
|
||||
})
|
||||
}
|
||||
|
||||
return func() {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
delete(ps.subscribers, sub)
|
||||
}
|
||||
}
|
||||
|
||||
// Publish publishes the provided message to the PubSub, and invokes
|
||||
// callbacks registered by subscribers asynchronously.
|
||||
func (ps *PubSub) Publish(msg interface{}) {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
|
||||
if ps.stopped {
|
||||
return
|
||||
}
|
||||
|
||||
ps.msg = msg
|
||||
for sub := range ps.subscribers {
|
||||
s := sub
|
||||
ps.cs.Schedule(func(context.Context) {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if !ps.subscribers[s] {
|
||||
return
|
||||
}
|
||||
s.OnMessage(msg)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Stop shuts down the PubSub and releases any resources allocated by it.
|
||||
// It is guaranteed that no subscriber callbacks would be invoked once this
|
||||
// method returns.
|
||||
func (ps *PubSub) Stop() {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
ps.stopped = true
|
||||
|
||||
ps.cancel()
|
||||
}
|
||||
|
|
@ -62,7 +62,8 @@ const (
|
|||
defaultPort = "443"
|
||||
defaultDNSSvrPort = "53"
|
||||
golang = "GO"
|
||||
// txtPrefix is the prefix string to be prepended to the host name for txt record lookup.
|
||||
// txtPrefix is the prefix string to be prepended to the host name for txt
|
||||
// record lookup.
|
||||
txtPrefix = "_grpc_config."
|
||||
// In DNS, service config is encoded in a TXT record via the mechanism
|
||||
// described in RFC-1464 using the attribute name grpc_config.
|
||||
|
|
@ -86,14 +87,14 @@ var (
|
|||
minDNSResRate = 30 * time.Second
|
||||
)
|
||||
|
||||
var customAuthorityDialler = func(authority string) func(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
return func(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
var addressDialer = func(address string) func(context.Context, string, string) (net.Conn, error) {
|
||||
return func(ctx context.Context, network, _ string) (net.Conn, error) {
|
||||
var dialer net.Dialer
|
||||
return dialer.DialContext(ctx, network, authority)
|
||||
return dialer.DialContext(ctx, network, address)
|
||||
}
|
||||
}
|
||||
|
||||
var customAuthorityResolver = func(authority string) (netResolver, error) {
|
||||
var newNetResolver = func(authority string) (netResolver, error) {
|
||||
host, port, err := parseTarget(authority, defaultDNSSvrPort)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
@ -103,7 +104,7 @@ var customAuthorityResolver = func(authority string) (netResolver, error) {
|
|||
|
||||
return &net.Resolver{
|
||||
PreferGo: true,
|
||||
Dial: customAuthorityDialler(authorityWithPort),
|
||||
Dial: addressDialer(authorityWithPort),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
|
@ -114,7 +115,8 @@ func NewBuilder() resolver.Builder {
|
|||
|
||||
type dnsBuilder struct{}
|
||||
|
||||
// Build creates and starts a DNS resolver that watches the name resolution of the target.
|
||||
// Build creates and starts a DNS resolver that watches the name resolution of
|
||||
// the target.
|
||||
func (b *dnsBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
|
||||
host, port, err := parseTarget(target.Endpoint(), defaultPort)
|
||||
if err != nil {
|
||||
|
|
@ -143,7 +145,7 @@ func (b *dnsBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts
|
|||
if target.URL.Host == "" {
|
||||
d.resolver = defaultResolver
|
||||
} else {
|
||||
d.resolver, err = customAuthorityResolver(target.URL.Host)
|
||||
d.resolver, err = newNetResolver(target.URL.Host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -180,19 +182,22 @@ type dnsResolver struct {
|
|||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
cc resolver.ClientConn
|
||||
// rn channel is used by ResolveNow() to force an immediate resolution of the target.
|
||||
// rn channel is used by ResolveNow() to force an immediate resolution of the
|
||||
// target.
|
||||
rn chan struct{}
|
||||
// wg is used to enforce Close() to return after the watcher() goroutine has finished.
|
||||
// Otherwise, data race will be possible. [Race Example] in dns_resolver_test we
|
||||
// replace the real lookup functions with mocked ones to facilitate testing.
|
||||
// If Close() doesn't wait for watcher() goroutine finishes, race detector sometimes
|
||||
// will warns lookup (READ the lookup function pointers) inside watcher() goroutine
|
||||
// has data race with replaceNetFunc (WRITE the lookup function pointers).
|
||||
// wg is used to enforce Close() to return after the watcher() goroutine has
|
||||
// finished. Otherwise, data race will be possible. [Race Example] in
|
||||
// dns_resolver_test we replace the real lookup functions with mocked ones to
|
||||
// facilitate testing. If Close() doesn't wait for watcher() goroutine
|
||||
// finishes, race detector sometimes will warns lookup (READ the lookup
|
||||
// function pointers) inside watcher() goroutine has data race with
|
||||
// replaceNetFunc (WRITE the lookup function pointers).
|
||||
wg sync.WaitGroup
|
||||
disableServiceConfig bool
|
||||
}
|
||||
|
||||
// ResolveNow invoke an immediate resolution of the target that this dnsResolver watches.
|
||||
// ResolveNow invoke an immediate resolution of the target that this
|
||||
// dnsResolver watches.
|
||||
func (d *dnsResolver) ResolveNow(resolver.ResolveNowOptions) {
|
||||
select {
|
||||
case d.rn <- struct{}{}:
|
||||
|
|
@ -220,8 +225,8 @@ func (d *dnsResolver) watcher() {
|
|||
|
||||
var timer *time.Timer
|
||||
if err == nil {
|
||||
// Success resolving, wait for the next ResolveNow. However, also wait 30 seconds at the very least
|
||||
// to prevent constantly re-resolving.
|
||||
// Success resolving, wait for the next ResolveNow. However, also wait 30
|
||||
// seconds at the very least to prevent constantly re-resolving.
|
||||
backoffIndex = 1
|
||||
timer = newTimerDNSResRate(minDNSResRate)
|
||||
select {
|
||||
|
|
@ -231,7 +236,8 @@ func (d *dnsResolver) watcher() {
|
|||
case <-d.rn:
|
||||
}
|
||||
} else {
|
||||
// Poll on an error found in DNS Resolver or an error received from ClientConn.
|
||||
// Poll on an error found in DNS Resolver or an error received from
|
||||
// ClientConn.
|
||||
timer = newTimer(backoff.DefaultExponential.Backoff(backoffIndex))
|
||||
backoffIndex++
|
||||
}
|
||||
|
|
@ -278,7 +284,8 @@ func (d *dnsResolver) lookupSRV() ([]resolver.Address, error) {
|
|||
}
|
||||
|
||||
func handleDNSError(err error, lookupType string) error {
|
||||
if dnsErr, ok := err.(*net.DNSError); ok && !dnsErr.IsTimeout && !dnsErr.IsTemporary {
|
||||
dnsErr, ok := err.(*net.DNSError)
|
||||
if ok && !dnsErr.IsTimeout && !dnsErr.IsTemporary {
|
||||
// Timeouts and temporary errors should be communicated to gRPC to
|
||||
// attempt another DNS query (with backoff). Other errors should be
|
||||
// suppressed (they may represent the absence of a TXT record).
|
||||
|
|
@ -307,10 +314,12 @@ func (d *dnsResolver) lookupTXT() *serviceconfig.ParseResult {
|
|||
res += s
|
||||
}
|
||||
|
||||
// TXT record must have "grpc_config=" attribute in order to be used as service config.
|
||||
// TXT record must have "grpc_config=" attribute in order to be used as
|
||||
// service config.
|
||||
if !strings.HasPrefix(res, txtAttribute) {
|
||||
logger.Warningf("dns: TXT record %v missing %v attribute", res, txtAttribute)
|
||||
// This is not an error; it is the equivalent of not having a service config.
|
||||
// This is not an error; it is the equivalent of not having a service
|
||||
// config.
|
||||
return nil
|
||||
}
|
||||
sc := canaryingSC(strings.TrimPrefix(res, txtAttribute))
|
||||
|
|
@ -352,9 +361,10 @@ func (d *dnsResolver) lookup() (*resolver.State, error) {
|
|||
return &state, nil
|
||||
}
|
||||
|
||||
// formatIP returns ok = false if addr is not a valid textual representation of an IP address.
|
||||
// If addr is an IPv4 address, return the addr and ok = true.
|
||||
// If addr is an IPv6 address, return the addr enclosed in square brackets and ok = true.
|
||||
// formatIP returns ok = false if addr is not a valid textual representation of
|
||||
// an IP address. If addr is an IPv4 address, return the addr and ok = true.
|
||||
// If addr is an IPv6 address, return the addr enclosed in square brackets and
|
||||
// ok = true.
|
||||
func formatIP(addr string) (addrIP string, ok bool) {
|
||||
ip := net.ParseIP(addr)
|
||||
if ip == nil {
|
||||
|
|
@ -366,10 +376,10 @@ func formatIP(addr string) (addrIP string, ok bool) {
|
|||
return "[" + addr + "]", true
|
||||
}
|
||||
|
||||
// parseTarget takes the user input target string and default port, returns formatted host and port info.
|
||||
// If target doesn't specify a port, set the port to be the defaultPort.
|
||||
// If target is in IPv6 format and host-name is enclosed in square brackets, brackets
|
||||
// are stripped when setting the host.
|
||||
// parseTarget takes the user input target string and default port, returns
|
||||
// formatted host and port info. If target doesn't specify a port, set the port
|
||||
// to be the defaultPort. If target is in IPv6 format and host-name is enclosed
|
||||
// in square brackets, brackets are stripped when setting the host.
|
||||
// examples:
|
||||
// target: "www.google.com" defaultPort: "443" returns host: "www.google.com", port: "443"
|
||||
// target: "ipv4-host:80" defaultPort: "443" returns host: "ipv4-host", port: "80"
|
||||
|
|
@ -385,12 +395,14 @@ func parseTarget(target, defaultPort string) (host, port string, err error) {
|
|||
}
|
||||
if host, port, err = net.SplitHostPort(target); err == nil {
|
||||
if port == "" {
|
||||
// If the port field is empty (target ends with colon), e.g. "[::1]:", this is an error.
|
||||
// If the port field is empty (target ends with colon), e.g. "[::1]:",
|
||||
// this is an error.
|
||||
return "", "", errEndsWithColon
|
||||
}
|
||||
// target has port, i.e ipv4-host:port, [ipv6-host]:port, host-name:port
|
||||
if host == "" {
|
||||
// Keep consistent with net.Dial(): If the host is empty, as in ":80", the local system is assumed.
|
||||
// Keep consistent with net.Dial(): If the host is empty, as in ":80",
|
||||
// the local system is assumed.
|
||||
host = "localhost"
|
||||
}
|
||||
return host, port, nil
|
||||
|
|
|
|||
|
|
@ -238,7 +238,7 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
|
|||
kp.Timeout = defaultServerKeepaliveTimeout
|
||||
}
|
||||
if kp.Time != infinity {
|
||||
if err = syscall.SetTCPUserTimeout(conn, kp.Timeout); err != nil {
|
||||
if err = syscall.SetTCPUserTimeout(rawConn, kp.Timeout); err != nil {
|
||||
return nil, connectionErrorf(false, err, "transport: failed to set TCP_USER_TIMEOUT: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -142,6 +142,10 @@ type Address struct {
|
|||
|
||||
// Equal returns whether a and o are identical. Metadata is compared directly,
|
||||
// not with any recursive introspection.
|
||||
//
|
||||
// This method compares all fields of the address. When used to tell apart
|
||||
// addresses during subchannel creation or connection establishment, it might be
|
||||
// more appropriate for the caller to implement custom equality logic.
|
||||
func (a Address) Equal(o Address) bool {
|
||||
return a.Addr == o.Addr && a.ServerName == o.ServerName &&
|
||||
a.Attributes.Equal(o.Attributes) &&
|
||||
|
|
@ -264,10 +268,6 @@ type ClientConn interface {
|
|||
// - "unknown_scheme://authority/endpoint"
|
||||
// Target{Scheme: resolver.GetDefaultScheme(), Endpoint: "unknown_scheme://authority/endpoint"}
|
||||
type Target struct {
|
||||
// Deprecated: use URL.Scheme instead.
|
||||
Scheme string
|
||||
// Deprecated: use URL.Host instead.
|
||||
Authority string
|
||||
// URL contains the parsed dial target with an optional default scheme added
|
||||
// to it if the original dial target contained no scheme or contained an
|
||||
// unregistered scheme. Any query params specified in the original dial
|
||||
|
|
|
|||
|
|
@ -577,6 +577,9 @@ type parser struct {
|
|||
// The header of a gRPC message. Find more detail at
|
||||
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
|
||||
header [5]byte
|
||||
|
||||
// recvBufferPool is the pool of shared receive buffers.
|
||||
recvBufferPool SharedBufferPool
|
||||
}
|
||||
|
||||
// recvMsg reads a complete gRPC message from the stream.
|
||||
|
|
@ -610,9 +613,7 @@ func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byt
|
|||
if int(length) > maxReceiveMessageSize {
|
||||
return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
|
||||
}
|
||||
// TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
|
||||
// of making it for each message:
|
||||
msg = make([]byte, int(length))
|
||||
msg = p.recvBufferPool.Get(int(length))
|
||||
if _, err := p.r.Read(msg); err != nil {
|
||||
if err == io.EOF {
|
||||
err = io.ErrUnexpectedEOF
|
||||
|
|
@ -726,12 +727,12 @@ type payloadInfo struct {
|
|||
}
|
||||
|
||||
func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) {
|
||||
pf, d, err := p.recvMsg(maxReceiveMessageSize)
|
||||
pf, buf, err := p.recvMsg(maxReceiveMessageSize)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if payInfo != nil {
|
||||
payInfo.compressedLength = len(d)
|
||||
payInfo.compressedLength = len(buf)
|
||||
}
|
||||
|
||||
if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil {
|
||||
|
|
@ -743,10 +744,10 @@ func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxRecei
|
|||
// To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
|
||||
// use this decompressor as the default.
|
||||
if dc != nil {
|
||||
d, err = dc.Do(bytes.NewReader(d))
|
||||
size = len(d)
|
||||
buf, err = dc.Do(bytes.NewReader(buf))
|
||||
size = len(buf)
|
||||
} else {
|
||||
d, size, err = decompress(compressor, d, maxReceiveMessageSize)
|
||||
buf, size, err = decompress(compressor, buf, maxReceiveMessageSize)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message: %v", err)
|
||||
|
|
@ -757,7 +758,7 @@ func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxRecei
|
|||
return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message after decompression larger than max (%d vs. %d)", size, maxReceiveMessageSize)
|
||||
}
|
||||
}
|
||||
return d, nil
|
||||
return buf, nil
|
||||
}
|
||||
|
||||
// Using compressor, decompress d, returning data and size.
|
||||
|
|
@ -792,15 +793,17 @@ func decompress(compressor encoding.Compressor, d []byte, maxReceiveMessageSize
|
|||
// dc takes precedence over compressor.
|
||||
// TODO(dfawley): wrap the old compressor/decompressor using the new API?
|
||||
func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error {
|
||||
d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor)
|
||||
buf, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := c.Unmarshal(d, m); err != nil {
|
||||
if err := c.Unmarshal(buf, m); err != nil {
|
||||
return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message: %v", err)
|
||||
}
|
||||
if payInfo != nil {
|
||||
payInfo.uncompressedBytes = d
|
||||
payInfo.uncompressedBytes = buf
|
||||
} else {
|
||||
p.recvBufferPool.Put(&buf)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -174,6 +174,7 @@ type serverOptions struct {
|
|||
maxHeaderListSize *uint32
|
||||
headerTableSize *uint32
|
||||
numServerWorkers uint32
|
||||
recvBufferPool SharedBufferPool
|
||||
}
|
||||
|
||||
var defaultServerOptions = serverOptions{
|
||||
|
|
@ -182,6 +183,7 @@ var defaultServerOptions = serverOptions{
|
|||
connectionTimeout: 120 * time.Second,
|
||||
writeBufferSize: defaultWriteBufSize,
|
||||
readBufferSize: defaultReadBufSize,
|
||||
recvBufferPool: nopBufferPool{},
|
||||
}
|
||||
var globalServerOptions []ServerOption
|
||||
|
||||
|
|
@ -552,6 +554,27 @@ func NumStreamWorkers(numServerWorkers uint32) ServerOption {
|
|||
})
|
||||
}
|
||||
|
||||
// RecvBufferPool returns a ServerOption that configures the server
|
||||
// to use the provided shared buffer pool for parsing incoming messages. Depending
|
||||
// on the application's workload, this could result in reduced memory allocation.
|
||||
//
|
||||
// If you are unsure about how to implement a memory pool but want to utilize one,
|
||||
// begin with grpc.NewSharedBufferPool.
|
||||
//
|
||||
// Note: The shared buffer pool feature will not be active if any of the following
|
||||
// options are used: StatsHandler, EnableTracing, or binary logging. In such
|
||||
// cases, the shared buffer pool will be ignored.
|
||||
//
|
||||
// # Experimental
|
||||
//
|
||||
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
|
||||
// later release.
|
||||
func RecvBufferPool(bufferPool SharedBufferPool) ServerOption {
|
||||
return newFuncServerOption(func(o *serverOptions) {
|
||||
o.recvBufferPool = bufferPool
|
||||
})
|
||||
}
|
||||
|
||||
// serverWorkerResetThreshold defines how often the stack must be reset. Every
|
||||
// N requests, by spawning a new goroutine in its place, a worker can reset its
|
||||
// stack so that large stacks don't live in memory forever. 2^16 should allow
|
||||
|
|
@ -1296,7 +1319,7 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
|
|||
if len(shs) != 0 || len(binlogs) != 0 {
|
||||
payInfo = &payloadInfo{}
|
||||
}
|
||||
d, err := recvAndDecompress(&parser{r: stream}, stream, dc, s.opts.maxReceiveMessageSize, payInfo, decomp)
|
||||
d, err := recvAndDecompress(&parser{r: stream, recvBufferPool: s.opts.recvBufferPool}, stream, dc, s.opts.maxReceiveMessageSize, payInfo, decomp)
|
||||
if err != nil {
|
||||
if e := t.WriteStatus(stream, status.Convert(err)); e != nil {
|
||||
channelz.Warningf(logger, s.channelzID, "grpc: Server.processUnaryRPC failed to write status: %v", e)
|
||||
|
|
@ -1506,7 +1529,7 @@ func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transp
|
|||
ctx: ctx,
|
||||
t: t,
|
||||
s: stream,
|
||||
p: &parser{r: stream},
|
||||
p: &parser{r: stream, recvBufferPool: s.opts.recvBufferPool},
|
||||
codec: s.getCodec(stream.ContentSubtype()),
|
||||
maxReceiveMessageSize: s.opts.maxReceiveMessageSize,
|
||||
maxSendMessageSize: s.opts.maxSendMessageSize,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,154 @@
|
|||
/*
|
||||
*
|
||||
* Copyright 2023 gRPC authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*
|
||||
*/
|
||||
|
||||
package grpc
|
||||
|
||||
import "sync"
|
||||
|
||||
// SharedBufferPool is a pool of buffers that can be shared, resulting in
|
||||
// decreased memory allocation. Currently, in gRPC-go, it is only utilized
|
||||
// for parsing incoming messages.
|
||||
//
|
||||
// # Experimental
|
||||
//
|
||||
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
|
||||
// later release.
|
||||
type SharedBufferPool interface {
|
||||
// Get returns a buffer with specified length from the pool.
|
||||
//
|
||||
// The returned byte slice may be not zero initialized.
|
||||
Get(length int) []byte
|
||||
|
||||
// Put returns a buffer to the pool.
|
||||
Put(*[]byte)
|
||||
}
|
||||
|
||||
// NewSharedBufferPool creates a simple SharedBufferPool with buckets
|
||||
// of different sizes to optimize memory usage. This prevents the pool from
|
||||
// wasting large amounts of memory, even when handling messages of varying sizes.
|
||||
//
|
||||
// # Experimental
|
||||
//
|
||||
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
|
||||
// later release.
|
||||
func NewSharedBufferPool() SharedBufferPool {
|
||||
return &simpleSharedBufferPool{
|
||||
pools: [poolArraySize]simpleSharedBufferChildPool{
|
||||
newBytesPool(level0PoolMaxSize),
|
||||
newBytesPool(level1PoolMaxSize),
|
||||
newBytesPool(level2PoolMaxSize),
|
||||
newBytesPool(level3PoolMaxSize),
|
||||
newBytesPool(level4PoolMaxSize),
|
||||
newBytesPool(0),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// simpleSharedBufferPool is a simple implementation of SharedBufferPool.
|
||||
type simpleSharedBufferPool struct {
|
||||
pools [poolArraySize]simpleSharedBufferChildPool
|
||||
}
|
||||
|
||||
func (p *simpleSharedBufferPool) Get(size int) []byte {
|
||||
return p.pools[p.poolIdx(size)].Get(size)
|
||||
}
|
||||
|
||||
func (p *simpleSharedBufferPool) Put(bs *[]byte) {
|
||||
p.pools[p.poolIdx(cap(*bs))].Put(bs)
|
||||
}
|
||||
|
||||
func (p *simpleSharedBufferPool) poolIdx(size int) int {
|
||||
switch {
|
||||
case size <= level0PoolMaxSize:
|
||||
return level0PoolIdx
|
||||
case size <= level1PoolMaxSize:
|
||||
return level1PoolIdx
|
||||
case size <= level2PoolMaxSize:
|
||||
return level2PoolIdx
|
||||
case size <= level3PoolMaxSize:
|
||||
return level3PoolIdx
|
||||
case size <= level4PoolMaxSize:
|
||||
return level4PoolIdx
|
||||
default:
|
||||
return levelMaxPoolIdx
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
level0PoolMaxSize = 16 // 16 B
|
||||
level1PoolMaxSize = level0PoolMaxSize * 16 // 256 B
|
||||
level2PoolMaxSize = level1PoolMaxSize * 16 // 4 KB
|
||||
level3PoolMaxSize = level2PoolMaxSize * 16 // 64 KB
|
||||
level4PoolMaxSize = level3PoolMaxSize * 16 // 1 MB
|
||||
)
|
||||
|
||||
const (
|
||||
level0PoolIdx = iota
|
||||
level1PoolIdx
|
||||
level2PoolIdx
|
||||
level3PoolIdx
|
||||
level4PoolIdx
|
||||
levelMaxPoolIdx
|
||||
poolArraySize
|
||||
)
|
||||
|
||||
type simpleSharedBufferChildPool interface {
|
||||
Get(size int) []byte
|
||||
Put(interface{})
|
||||
}
|
||||
|
||||
type bufferPool struct {
|
||||
sync.Pool
|
||||
|
||||
defaultSize int
|
||||
}
|
||||
|
||||
func (p *bufferPool) Get(size int) []byte {
|
||||
bs := p.Pool.Get().(*[]byte)
|
||||
|
||||
if cap(*bs) < size {
|
||||
p.Pool.Put(bs)
|
||||
|
||||
return make([]byte, size)
|
||||
}
|
||||
|
||||
return (*bs)[:size]
|
||||
}
|
||||
|
||||
func newBytesPool(size int) simpleSharedBufferChildPool {
|
||||
return &bufferPool{
|
||||
Pool: sync.Pool{
|
||||
New: func() interface{} {
|
||||
bs := make([]byte, size)
|
||||
return &bs
|
||||
},
|
||||
},
|
||||
defaultSize: size,
|
||||
}
|
||||
}
|
||||
|
||||
// nopBufferPool is a buffer pool just makes new buffer without pooling.
|
||||
type nopBufferPool struct {
|
||||
}
|
||||
|
||||
func (nopBufferPool) Get(length int) []byte {
|
||||
return make([]byte, length)
|
||||
}
|
||||
|
||||
func (nopBufferPool) Put(*[]byte) {
|
||||
}
|
||||
|
|
@ -507,7 +507,7 @@ func (a *csAttempt) newStream() error {
|
|||
return toRPCErr(nse.Err)
|
||||
}
|
||||
a.s = s
|
||||
a.p = &parser{r: s}
|
||||
a.p = &parser{r: s, recvBufferPool: a.cs.cc.dopts.recvBufferPool}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -1270,7 +1270,7 @@ func newNonRetryClientStream(ctx context.Context, desc *StreamDesc, method strin
|
|||
return nil, err
|
||||
}
|
||||
as.s = s
|
||||
as.p = &parser{r: s}
|
||||
as.p = &parser{r: s, recvBufferPool: ac.dopts.recvBufferPool}
|
||||
ac.incrCallsStarted()
|
||||
if desc != unaryStreamDesc {
|
||||
// Listen on stream context to cleanup when the stream context is
|
||||
|
|
|
|||
|
|
@ -19,4 +19,4 @@
|
|||
package grpc
|
||||
|
||||
// Version is the current grpc version.
|
||||
const Version = "1.56.2"
|
||||
const Version = "1.57.0"
|
||||
|
|
|
|||
|
|
@ -1,16 +1,16 @@
|
|||
# cloud.google.com/go v0.110.4
|
||||
# cloud.google.com/go v0.110.6
|
||||
## explicit; go 1.19
|
||||
cloud.google.com/go/internal
|
||||
cloud.google.com/go/internal/optional
|
||||
cloud.google.com/go/internal/trace
|
||||
cloud.google.com/go/internal/version
|
||||
# cloud.google.com/go/compute v1.20.1
|
||||
# cloud.google.com/go/compute v1.23.0
|
||||
## explicit; go 1.19
|
||||
cloud.google.com/go/compute/internal
|
||||
# cloud.google.com/go/compute/metadata v0.2.3
|
||||
## explicit; go 1.19
|
||||
cloud.google.com/go/compute/metadata
|
||||
# cloud.google.com/go/iam v1.1.0
|
||||
# cloud.google.com/go/iam v1.1.1
|
||||
## explicit; go 1.19
|
||||
cloud.google.com/go/iam
|
||||
cloud.google.com/go/iam/apiv1/iampb
|
||||
|
|
@ -903,6 +903,7 @@ golang.org/x/oauth2/jwt
|
|||
# golang.org/x/sync v0.3.0
|
||||
## explicit; go 1.17
|
||||
golang.org/x/sync/errgroup
|
||||
golang.org/x/sync/semaphore
|
||||
golang.org/x/sync/syncmap
|
||||
# golang.org/x/sys v0.11.0
|
||||
## explicit; go 1.17
|
||||
|
|
@ -944,7 +945,7 @@ golang.org/x/tools/internal/typesinternal
|
|||
## explicit; go 1.17
|
||||
golang.org/x/xerrors
|
||||
golang.org/x/xerrors/internal
|
||||
# google.golang.org/api v0.134.0
|
||||
# google.golang.org/api v0.136.0
|
||||
## explicit; go 1.19
|
||||
google.golang.org/api/googleapi
|
||||
google.golang.org/api/googleapi/transport
|
||||
|
|
@ -976,21 +977,21 @@ google.golang.org/appengine/internal/socket
|
|||
google.golang.org/appengine/internal/urlfetch
|
||||
google.golang.org/appengine/socket
|
||||
google.golang.org/appengine/urlfetch
|
||||
# google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130
|
||||
# google.golang.org/genproto v0.0.0-20230803162519-f966b187b2e5
|
||||
## explicit; go 1.19
|
||||
google.golang.org/genproto/googleapis/type/date
|
||||
google.golang.org/genproto/googleapis/type/expr
|
||||
google.golang.org/genproto/internal
|
||||
# google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130
|
||||
# google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5
|
||||
## explicit; go 1.19
|
||||
google.golang.org/genproto/googleapis/api
|
||||
google.golang.org/genproto/googleapis/api/annotations
|
||||
# google.golang.org/genproto/googleapis/rpc v0.0.0-20230720185612-659f7aaaa771
|
||||
# google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577
|
||||
## explicit; go 1.19
|
||||
google.golang.org/genproto/googleapis/rpc/code
|
||||
google.golang.org/genproto/googleapis/rpc/errdetails
|
||||
google.golang.org/genproto/googleapis/rpc/status
|
||||
# google.golang.org/grpc v1.56.2
|
||||
# google.golang.org/grpc v1.57.0
|
||||
## explicit; go 1.17
|
||||
google.golang.org/grpc
|
||||
google.golang.org/grpc/attributes
|
||||
|
|
|
|||
Loading…
Reference in New Issue