googleanalytics
/
lib
/
analytics-admin
/
vendor
/
google
/
grpc-gcp
/
src
/
GcpExtensionChannel.php
generated
3 years ago
ChannelRef.php
3 years ago
Config.php
3 years ago
CreatedByDeserializeCheck.php
3 years ago
GCPBidiStreamingCall.php
3 years ago
GCPCallInvoker.php
3 years ago
GCPClientStreamCall.php
3 years ago
GCPServerStreamCall.php
3 years ago
GCPUnaryCall.php
3 years ago
GcpBaseCall.php
3 years ago
GcpExtensionChannel.php
3 years ago
grpc_gcp.proto
3 years ago
GcpExtensionChannel.php
282 lines
| 1 | <?php |
| 2 | /* |
| 3 | * |
| 4 | * Copyright 2018 gRPC authors. |
| 5 | * |
| 6 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 7 | * you may not use this file except in compliance with the License. |
| 8 | * You may obtain a copy of the License at |
| 9 | * |
| 10 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | * |
| 12 | * Unless required by applicable law or agreed to in writing, software |
| 13 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 14 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 15 | * See the License for the specific language governing permissions and |
| 16 | * limitations under the License. |
| 17 | * |
| 18 | */ |
| 19 | namespace Grpc\Gcp; |
| 20 | |
| 21 | /** |
| 22 | * GcpExtensionChannel maintains an array of channels for certain API. |
| 23 | */ |
| 24 | class GcpExtensionChannel |
| 25 | { |
| 26 | public $max_size; |
| 27 | public $max_concurrent_streams_low_watermark; |
| 28 | public $target; |
| 29 | public $options; |
| 30 | public $affinity_by_method; |
| 31 | public $affinity_key_to_channel_ref; |
| 32 | public $channel_refs; |
| 33 | public $credentials; |
| 34 | public $affinity_conf; |
| 35 | |
| 36 | private $is_closed; |
| 37 | |
| 38 | /** |
| 39 | * @return array An array of ChannelRefs created for certain API. |
| 40 | */ |
| 41 | public function getChannelRefs() |
| 42 | { |
| 43 | return $this->channel_refs; |
| 44 | } |
| 45 | |
| 46 | /** |
| 47 | * @param string $hostname |
| 48 | * @param array $opts Options to create a \Grpc\Channel and affinity config |
| 49 | */ |
| 50 | public function __construct($hostname = null, $opts = array()) |
| 51 | { |
| 52 | if ($hostname == null || !is_array($opts)) { |
| 53 | throw new \InvalidArgumentException("Expected hostname is empty"); |
| 54 | } |
| 55 | $this->max_size = 10; |
| 56 | $this->max_concurrent_streams_low_watermark = 100; |
| 57 | if (isset($opts['affinity_conf'])) { |
| 58 | if (isset($opts['affinity_conf']['channelPool'])) { |
| 59 | if (isset($opts['affinity_conf']['channelPool']['maxSize'])) { |
| 60 | $this->max_size = $opts['affinity_conf']['channelPool']['maxSize']; |
| 61 | } |
| 62 | if (isset($opts['affinity_conf']['channelPool']['maxConcurrentStreamsLowWatermark'])) { |
| 63 | $this->max_concurrent_streams_low_watermark = |
| 64 | $opts['affinity_conf']['channelPool']['maxConcurrentStreamsLowWatermark']; |
| 65 | } |
| 66 | } |
| 67 | $this->affinity_by_method = $opts['affinity_conf']['affinity_by_method']; |
| 68 | $this->affinity_conf = $opts['affinity_conf']; |
| 69 | } |
| 70 | $this->target = $hostname; |
| 71 | $this->affinity_key_to_channel_ref = array(); |
| 72 | $this->channel_refs = array(); |
| 73 | $this->updateOpts($opts); |
| 74 | // Initiate a Grpc\Channel at the beginning in order to keep the same |
| 75 | // behavior as the Grpc. |
| 76 | $channel_ref = $this->getChannelRef(); |
| 77 | $channel_ref->getRealChannel($this->credentials); |
| 78 | } |
| 79 | |
| 80 | /** |
| 81 | * @param array $opts Options to create a \Grpc\Channel |
| 82 | */ |
| 83 | public function updateOpts($opts) |
| 84 | { |
| 85 | if (isset($opts['credentials'])) { |
| 86 | $this->credentials = $opts['credentials']; |
| 87 | } |
| 88 | unset($opts['affinity_conf']); |
| 89 | unset($opts['credentials']); |
| 90 | $this->options = $opts; |
| 91 | $this->is_closed = false; |
| 92 | } |
| 93 | |
| 94 | /** |
| 95 | * Bind the ChannelRef with the affinity key. This is a private method. |
| 96 | * |
| 97 | * @param ChannelRef $channel_ref |
| 98 | * @param string $affinity_key |
| 99 | * |
| 100 | * @return ChannelRef |
| 101 | */ |
| 102 | public function bind($channel_ref, $affinity_key) |
| 103 | { |
| 104 | if (!array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) { |
| 105 | $this->affinity_key_to_channel_ref[$affinity_key] = $channel_ref; |
| 106 | } |
| 107 | $channel_ref->affinityRefIncr(); |
| 108 | return $channel_ref; |
| 109 | } |
| 110 | |
| 111 | /** |
| 112 | * Unbind the affinity key. This is a private method. |
| 113 | * |
| 114 | * @param string $affinity_key |
| 115 | * |
| 116 | * @return ChannelRef |
| 117 | */ |
| 118 | public function unbind($affinity_key) |
| 119 | { |
| 120 | $channel_ref = null; |
| 121 | if (array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) { |
| 122 | $channel_ref = $this->affinity_key_to_channel_ref[$affinity_key]; |
| 123 | $channel_ref->affinityRefDecr(); |
| 124 | } |
| 125 | unset($this->affinity_key_to_channel_ref[$affinity_key]); |
| 126 | return $channel_ref; |
| 127 | } |
| 128 | |
| 129 | |
| 130 | public function cmp_by_active_stream_ref($a, $b) |
| 131 | { |
| 132 | return $a->getActiveStreamRef() - $b->getActiveStreamRef(); |
| 133 | } |
| 134 | |
| 135 | /** |
| 136 | * Pick or create a ChannelRef from the pool by affinity key. |
| 137 | * |
| 138 | * @param string $affinity_key |
| 139 | * |
| 140 | * @return ChannelRef |
| 141 | */ |
| 142 | public function getChannelRef($affinity_key = null) |
| 143 | { |
| 144 | if ($affinity_key) { |
| 145 | if (array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) { |
| 146 | return $this->affinity_key_to_channel_ref[$affinity_key]; |
| 147 | } |
| 148 | return $this->getChannelRef(); |
| 149 | } |
| 150 | usort($this->channel_refs, array($this, 'cmp_by_active_stream_ref')); |
| 151 | |
| 152 | if (count($this->channel_refs) > 0 && $this->channel_refs[0]->getActiveStreamRef() < |
| 153 | $this->max_concurrent_streams_low_watermark) { |
| 154 | return $this->channel_refs[0]; |
| 155 | } |
| 156 | $num_channel_refs = count($this->channel_refs); |
| 157 | if ($num_channel_refs < $this->max_size) { |
| 158 | // grpc_target_persist_bound stands for how many channels can be persisted for |
| 159 | // the same target in the C extension. It is possible that the user use the pure |
| 160 | // gRPC and this GCP extension at the same time, which share the same target. In this case |
| 161 | // pure gRPC channel may occupy positions in C extension, which deletes some channels created |
| 162 | // by this GCP extension. |
| 163 | // If that happens, it won't cause the script failure because we saves all arguments for creating |
| 164 | // a channel instead of a channel itself. If we watch to fetch a GCP channel already deleted, |
| 165 | // it will create a new channel. The only cons is the latency of the first RPC will high because |
| 166 | // it will establish the connection again. |
| 167 | if (!isset($this->options['grpc_target_persist_bound']) || |
| 168 | $this->options['grpc_target_persist_bound'] < $this->max_size) { |
| 169 | $this->options['grpc_target_persist_bound'] = $this->max_size; |
| 170 | } |
| 171 | $cur_opts = array_merge($this->options, |
| 172 | ['grpc_gcp_channel_id' => $num_channel_refs]); |
| 173 | $channel_ref = new ChannelRef($this->target, $num_channel_refs, $cur_opts); |
| 174 | array_unshift($this->channel_refs, $channel_ref); |
| 175 | } |
| 176 | return $this->channel_refs[0]; |
| 177 | } |
| 178 | |
| 179 | /** |
| 180 | * Get the connectivity state of the channel |
| 181 | * |
| 182 | * @param bool $try_to_connect try to connect on the channel |
| 183 | * |
| 184 | * @return int The grpc connectivity state |
| 185 | * @throws \InvalidArgumentException |
| 186 | */ |
| 187 | public function getConnectivityState($try_to_connect = false) |
| 188 | { |
| 189 | // Since getRealChannel is creating a PHP Channel object. However in gRPC, when a Channel |
| 190 | // object is closed, we only mark this Object to be invalid. Thus, we need a global variable |
| 191 | // to mark whether this GCPExtensionChannel is close or not. |
| 192 | if ($this->is_closed) { |
| 193 | throw new \RuntimeException("Channel has already been closed"); |
| 194 | } |
| 195 | $ready = 0; |
| 196 | $idle = 0; |
| 197 | $connecting = 0; |
| 198 | $transient_failure = 0; |
| 199 | $shutdown = 0; |
| 200 | foreach ($this->channel_refs as $channel_ref) { |
| 201 | $state = $channel_ref->getRealChannel($this->credentials)->getConnectivityState($try_to_connect); |
| 202 | switch ($state) { |
| 203 | case \Grpc\CHANNEL_READY: |
| 204 | $ready += 1; |
| 205 | break 2; |
| 206 | case \Grpc\CHANNEL_FATAL_FAILURE: |
| 207 | $shutdown += 1; |
| 208 | break; |
| 209 | case \Grpc\CHANNEL_CONNECTING: |
| 210 | $connecting += 1; |
| 211 | break; |
| 212 | case \Grpc\CHANNEL_TRANSIENT_FAILURE: |
| 213 | $transient_failure += 1; |
| 214 | break; |
| 215 | case \Grpc\CHANNEL_IDLE: |
| 216 | $idle += 1; |
| 217 | break; |
| 218 | } |
| 219 | } |
| 220 | if ($ready > 0) { |
| 221 | return \Grpc\CHANNEL_READY; |
| 222 | } elseif ($idle > 0) { |
| 223 | return \Grpc\CHANNEL_IDLE; |
| 224 | } elseif ($connecting > 0) { |
| 225 | return \Grpc\CHANNEL_CONNECTING; |
| 226 | } elseif ($transient_failure > 0) { |
| 227 | return \Grpc\CHANNEL_TRANSIENT_FAILURE; |
| 228 | } elseif ($shutdown > 0) { |
| 229 | return \Grpc\CHANNEL_SHUTDOWN; |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | /** |
| 234 | * Watch the connectivity state of the channel until it changed |
| 235 | * |
| 236 | * @param int $last_state The previous connectivity state of the channel |
| 237 | * @param Timeval $deadline_obj The deadline this function should wait until |
| 238 | * |
| 239 | * @return bool If the connectivity state changes from last_state |
| 240 | * before deadline |
| 241 | * @throws \InvalidArgumentException |
| 242 | */ |
| 243 | public function watchConnectivityState($last_state, $deadline_obj = null) |
| 244 | { |
| 245 | if ($deadline_obj == null || !is_a($deadline_obj, '\Grpc\Timeval')) { |
| 246 | throw new \InvalidArgumentException(""); |
| 247 | } |
| 248 | // Since getRealChannel is creating a PHP Channel object. However in gRPC, when a Channel |
| 249 | // object is closed, we only mark this Object to be invalid. Thus, we need a global variable |
| 250 | // to mark whether this GCPExtensionChannel is close or not. |
| 251 | if ($this->is_closed) { |
| 252 | throw new \RuntimeException("Channel has already been closed"); |
| 253 | } |
| 254 | $state = 0; |
| 255 | foreach ($this->channel_refs as $channel_ref) { |
| 256 | $state = $channel_ref->getRealChannel($this->credentials)->watchConnectivityState($last_state, $deadline_obj); |
| 257 | } |
| 258 | return $state; |
| 259 | } |
| 260 | |
| 261 | /** |
| 262 | * Get the endpoint this call/stream is connected to |
| 263 | * |
| 264 | * @return string The URI of the endpoint |
| 265 | */ |
| 266 | public function getTarget() |
| 267 | { |
| 268 | return $this->target; |
| 269 | } |
| 270 | |
| 271 | /** |
| 272 | * Close the channel |
| 273 | */ |
| 274 | public function close() |
| 275 | { |
| 276 | foreach ($this->channel_refs as $channel_ref) { |
| 277 | $channel_ref->getRealChannel($this->credentials)->close(); |
| 278 | } |
| 279 | $this->is_closed = true; |
| 280 | } |
| 281 | } |
| 282 |