1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
# 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.
"""Implementations of in-order work deference."""
import abc
import enum
import threading
from grpc.framework.foundation import activated
from grpc.framework.foundation import logging_pool
_NULL_BEHAVIOR = lambda unused_value: None
class Relay(object):
"""Performs work submitted to it in another thread.
Performs work in the order in which work was submitted to it; otherwise there
would be no reason to use an implementation of this interface instead of a
thread pool.
"""
@abc.abstractmethod
def add_value(self, value):
"""Adds a value to be passed to the behavior registered with this Relay.
Args:
value: A value that will be passed to a call made in another thread to the
behavior registered with this Relay.
"""
raise NotImplementedError()
@abc.abstractmethod
def set_behavior(self, behavior):
"""Sets the behavior that this Relay should call when passed values.
Args:
behavior: The behavior that this Relay should call in another thread when
passed a value, or None to have passed values ignored.
"""
raise NotImplementedError()
class _PoolRelay(activated.Activated, Relay):
@enum.unique
class _State(enum.Enum):
INACTIVE = 'inactive'
IDLE = 'idle'
SPINNING = 'spinning'
def __init__(self, pool, behavior):
self._condition = threading.Condition()
self._pool = pool
self._own_pool = pool is None
self._state = _PoolRelay._State.INACTIVE
self._activated = False
self._spinning = False
self._values = []
self._behavior = _NULL_BEHAVIOR if behavior is None else behavior
def _spin(self, behavior, value):
while True:
behavior(value)
with self._condition:
if self._values:
value = self._values.pop(0)
behavior = self._behavior
else:
self._state = _PoolRelay._State.IDLE
self._condition.notify_all()
break
def add_value(self, value):
with self._condition:
if self._state is _PoolRelay._State.INACTIVE:
raise ValueError('add_value not valid on inactive Relay!')
elif self._state is _PoolRelay._State.IDLE:
self._pool.submit(self._spin, self._behavior, value)
self._state = _PoolRelay._State.SPINNING
else:
self._values.append(value)
def set_behavior(self, behavior):
with self._condition:
self._behavior = _NULL_BEHAVIOR if behavior is None else behavior
def _start(self):
with self._condition:
self._state = _PoolRelay._State.IDLE
if self._own_pool:
self._pool = logging_pool.pool(1)
return self
def _stop(self):
with self._condition:
while self._state is _PoolRelay._State.SPINNING:
self._condition.wait()
if self._own_pool:
self._pool.shutdown(wait=True)
self._state = _PoolRelay._State.INACTIVE
def __enter__(self):
return self._start()
def __exit__(self, exc_type, exc_val, exc_tb):
self._stop()
return False
def start(self):
return self._start()
def stop(self):
self._stop()
def relay(behavior):
"""Creates a Relay.
Args:
behavior: The behavior to be called by the created Relay, or None to have
passed values dropped until a different behavior is given to the returned
Relay later.
Returns:
An object that is both an activated.Activated and a Relay. The object is
only valid for use as a Relay when activated.
"""
return _PoolRelay(None, behavior)
def pool_relay(pool, behavior):
"""Creates a Relay that uses a given thread pool.
This object will make use of at most one thread in the given pool.
Args:
pool: A futures.ThreadPoolExecutor for use by the created Relay.
behavior: The behavior to be called by the created Relay, or None to have
passed values dropped until a different behavior is given to the returned
Relay later.
Returns:
An object that is both an activated.Activated and a Relay. The object is
only valid for use as a Relay when activated.
"""
return _PoolRelay(pool, behavior)
|
