Helper method that subscribes a single tensor to a list of side_effects. This method will check if the given tensor has already been subscribed or if it's a tensor returned by a previous call to `subscribe()` and, if so, will reuse the existing identity op, appending the given side effects to
(tensor, side_effects, control_cache)
| 215 | |
| 216 | |
| 217 | def _subscribe(tensor, side_effects, control_cache): |
| 218 | """Helper method that subscribes a single tensor to a list of side_effects. |
| 219 | |
| 220 | This method will check if the given tensor has already been subscribed or if |
| 221 | it's a tensor returned by a previous call to `subscribe()` and, if so, will |
| 222 | reuse the existing identity op, appending the given side effects to the list |
| 223 | of existing ones. |
| 224 | |
| 225 | Args: |
| 226 | tensor: The `tf.Tensor` to be subscribed. |
| 227 | side_effects: List of side_effect functions, see subscribe for details. |
| 228 | control_cache: `_ControlOutputCache` helper to get control_outputs faster. |
| 229 | |
| 230 | Returns: |
| 231 | The modified replacement to the passed in tensor which triggers the side |
| 232 | effects or the given tensor, if it was already been subscribed. |
| 233 | """ |
| 234 | # Check if the given tensor has a numpy compatible type (see dtypes.py). |
| 235 | # If not, we cannot subscribe it, so we just return the original tensor. |
| 236 | if not tensor.dtype.is_numpy_compatible: |
| 237 | logging.debug(('Tensor {} has an un-supported {} type and cannot be ' |
| 238 | 'subscribed.').format(tensor.name, tensor.dtype)) |
| 239 | return tensor |
| 240 | |
| 241 | if _is_subscribed_identity(tensor): |
| 242 | return _subscribe_extend(tensor, side_effects) |
| 243 | |
| 244 | # Check if the given tensor has already been subscribed by inspecting its |
| 245 | # outputs. |
| 246 | name_scope = tensor.op.name + '/subscription/Identity' |
| 247 | consumers = tensor.consumers() |
| 248 | matching_ops = [op for op in consumers if op.name.startswith(name_scope)] |
| 249 | assert len(matching_ops) <= 1, ('Op {} must only have one subscription ' |
| 250 | 'op connected to it').format(tensor.op.name) |
| 251 | if len(matching_ops) == 1: |
| 252 | candidate_tensor = matching_ops[0].outputs[0] |
| 253 | if _is_subscribed_identity(candidate_tensor): |
| 254 | return _subscribe_extend(candidate_tensor, side_effects) |
| 255 | |
| 256 | return _subscribe_new(tensor, side_effects, control_cache) |
| 257 | |
| 258 | |
| 259 | @contextlib.contextmanager |
no test coverage detected