IntenseWebs/openvino
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Telemetry sender and MO instrumentation (#3804)

Browse Source 
* Draft implementation of the telemetry sender utility

* Examples of sending telemetry from the MO

* More statistic about the model.

* Intentional broken file to fail Mask-RCNN ONNX model conversion

* Added joined list of ops used

* Added requests to the requrements file and update BOM to include necessary files related to telemetry

* Send telemetry alwasys

* Refactored usage of GUID usage in the telemetry

* Enabled sending telemetry always

* Simplified function "TelemetryBackend.send"

* Use other approach to send information about session to GA

* Added automatic registration of the telemetry backends and allow to choose it during the telemetry class instantiation

* Added "requests" as a requirement. Wrapped usage of requests module to not crash the app

* Added timeout for sending data to GA. Increased the queue size to 1000

* Finalize Telemetry class implementation

* Do not fail MO if non-critical component is not installed and updated Telemetry GA with the default property

* Added sending version to a separate event

* Use default TID to send the data

* Set lower bound for the requests module which does not contain vulnerabilities

Co-authored-by: Evgeny Lazarev <elazarev.nnov@gmail.com>
This commit is contained in:
Evgeny Lazarev
2021-02-10 10:51:31 +03:00
committed by GitHub
co-authored by Evgeny Lazarev
parent 73f846648c
commit 66f4c69b90
21 changed files with 588 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
model-optimizer/automation/package_BOM.txt
+9
View File
@@ -1031,3 +1031,12 @@ requirements_mxnet.txt
requirements_onnx.txt
requirements_tf.txt
requirements_tf2.txt
telemetry/__init__.py
telemetry/backend/__init__.py
telemetry/backend/backend.py
telemetry/backend/backend_ga.py
telemetry/telemetry.py
telemetry/utils/__init__.py
telemetry/utils/isip.py
telemetry/utils/message.py
telemetry/utils/sender.py
model-optimizer/mo/main.py
+20 -5
View File
@@ -24,10 +24,10 @@ from collections import OrderedDict
import numpy as np
import telemetry.telemetry as tm
from extensions.back.SpecialNodesFinalization import RemoveConstOps, CreateConstNodesReplacement, NormalizeTI
from mo.utils.get_ov_update_message import get_ov_update_message
from mo.graph.graph import Graph
from mo.middle.pattern_match import for_graph_and_each_sub_graph_recursively, for_each_sub_graph_recursively
from mo.middle.pattern_match import for_graph_and_each_sub_graph_recursively
from mo.pipeline.common import prepare_emit_ir, get_ir_version
from mo.pipeline.unified import unified_pipeline
from mo.utils import import_extensions
@@ -35,6 +35,7 @@ from mo.utils.cli_parser import get_placeholder_shapes, get_tuple_values, get_mo
get_common_cli_options, get_caffe_cli_options, get_tf_cli_options, get_mxnet_cli_options, get_kaldi_cli_options, \
get_onnx_cli_options, get_mean_scale_dictionary, parse_tuple_pairs, get_freeze_placeholder_values, get_meta_info
from mo.utils.error import Error, FrameworkError
from mo.utils.get_ov_update_message import get_ov_update_message
from mo.utils.guess_framework import deduce_framework_by_namespace
from mo.utils.logger import init_logger
from mo.utils.model_analysis import AnalysisResults
@@ -113,6 +114,8 @@ def prepare_ir(argv: argparse.Namespace):
log.debug(str(argv))
log.debug("Model Optimizer started")
t = tm.Telemetry()
t.start_session()
model_name = "<UNKNOWN_NAME>"
if argv.model_name:
@@ -151,12 +154,10 @@ def prepare_ir(argv: argparse.Namespace):
if is_tf and argv.tensorflow_use_custom_operations_config is not None:
argv.transformations_config = argv.tensorflow_use_custom_operations_config
mean_file_offsets = None
if is_caffe and argv.mean_file and argv.mean_values:
raise Error('Both --mean_file and mean_values are specified. Specify either mean file or mean values. ' +
refer_to_faq_msg(17))
elif is_caffe and argv.mean_file and argv.mean_file_offsets:
values = get_tuple_values(argv.mean_file_offsets, t=int, num_exp_values=2)
mean_file_offsets = np.array([int(x) for x in values[0].split(',')])
if not all([offset >= 0 for offset in mean_file_offsets]):
@@ -207,7 +208,6 @@ def prepare_ir(argv: argparse.Namespace):
log.debug("Placeholder shapes : {}".format(argv.placeholder_shapes))
ret_res = 1
if hasattr(argv, 'extensions') and argv.extensions and argv.extensions != '':
extensions = argv.extensions.split(',')
else:
@@ -216,18 +216,23 @@ def prepare_ir(argv: argparse.Namespace):
argv.freeze_placeholder_with_value, argv.input = get_freeze_placeholder_values(argv.input,
argv.freeze_placeholder_with_value)
if is_tf:
t.send_event('mo', 'framework', 'tf')
from mo.front.tf.register_custom_ops import get_front_classes
import_extensions.load_dirs(argv.framework, extensions, get_front_classes)
elif is_caffe:
t.send_event('mo', 'framework', 'caffe')
from mo.front.caffe.register_custom_ops import get_front_classes
import_extensions.load_dirs(argv.framework, extensions, get_front_classes)
elif is_mxnet:
t.send_event('mo', 'framework', 'mxnet')
from mo.front.mxnet.register_custom_ops import get_front_classes
import_extensions.load_dirs(argv.framework, extensions, get_front_classes)
elif is_kaldi:
t.send_event('mo', 'framework', 'kaldi')
from mo.front.kaldi.register_custom_ops import get_front_classes
import_extensions.load_dirs(argv.framework, extensions, get_front_classes)
elif is_onnx:
t.send_event('mo', 'framework', 'onnx')
from mo.front.onnx.register_custom_ops import get_front_classes
import_extensions.load_dirs(argv.framework, extensions, get_front_classes)
graph = unified_pipeline(argv)
@@ -282,6 +287,9 @@ def driver(argv: argparse.Namespace):
def main(cli_parser: argparse.ArgumentParser, framework: str):
telemetry = tm.Telemetry(app_name='Model Optimizer', app_version=get_version())
telemetry.start_session()
telemetry.send_event('mo', 'version', get_version())
try:
# Initialize logger with 'ERROR' as default level to be able to form nice messages
# before arg parser deliver log_level requested by user
@@ -297,6 +305,9 @@ def main(cli_parser: argparse.ArgumentParser, framework: str):
ret_code = driver(argv)
if ov_update_message:
print(ov_update_message)
telemetry.send_event('mo', 'conversion_result', 'success')
telemetry.end_session()
telemetry.force_shutdown(1.0)
return ret_code
except (FileNotFoundError, NotADirectoryError) as e:
log.error('File {} was not found'.format(str(e).split('No such file or directory:')[1]))
@@ -320,4 +331,8 @@ def main(cli_parser: argparse.ArgumentParser, framework: str):
log.error(traceback.format_exc())
log.error("---------------- END OF BUG REPORT --------------")
log.error("-------------------------------------------------")
telemetry.send_event('mo', 'conversion_result', 'fail')
telemetry.end_session()
telemetry.force_shutdown(1.0)
return 1
model-optimizer/mo/utils/versions_checker.py
+1 -3
View File
@@ -25,7 +25,7 @@ modules = {
"protobuf": "google.protobuf",
"test-generator": "generator",
}
critical_modules = ["networkx"]
critical_modules = ["networkx", "defusedxml", "numpy"]
message = "\nDetected not satisfied dependencies:\n" \
"{}\n" \
@@ -244,14 +244,12 @@ def check_requirements(framework=None):
not_satisfied_versions.append((name, 'not installed', 'required: {} {}'.format(key, required_version)))
else:
not_satisfied_versions.append((name, 'not installed', ''))
exit_code = 1
continue
except Exception as e:
log.error('Error happened while importing {} module. It may happen due to unsatisfied requirements of '
'that module. Please run requirements installation script once more.\n'
'Details on module importing failure: {}'.format(name, e))
not_satisfied_versions.append((name, 'package error', 'required: {} {}'.format(key, required_version)))
exit_code = 1
continue
if len(not_satisfied_versions) != 0:
model-optimizer/requirements.txt
+1
View File
@@ -8,3 +8,4 @@ onnx>=1.1.2
test-generator==0.1.1
defusedxml>=0.5.0
urllib3>=1.25.9
requests>=2.20.0
model-optimizer/requirements_caffe.txt
+1
View File
@@ -3,3 +3,4 @@ numpy>=1.14.0
protobuf>=3.6.1
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/requirements_dev.txt
+1
View File
@@ -3,3 +3,4 @@ pylint==2.5.0
pyenchant==1.6.11
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/requirements_kaldi.txt
+1
View File
@@ -2,3 +2,4 @@ networkx>=1.11
numpy>=1.14.0
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/requirements_mxnet.txt
+1
View File
@@ -4,3 +4,4 @@ numpy>=1.14.0
test-generator==0.1.1
defusedxml>=0.5.0
urllib3>=1.25.9
requests>=2.20.0
model-optimizer/requirements_onnx.txt
+1
View File
@@ -3,3 +3,4 @@ networkx>=1.11
numpy>=1.14.0
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/requirements_tf.txt
+1
View File
@@ -4,3 +4,4 @@ networkx>=1.11
numpy>=1.14.0,<1.19.0
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/requirements_tf2.txt
+1
View File
@@ -3,3 +3,4 @@ networkx>=1.11
numpy>=1.14.0
test-generator==0.1.1
defusedxml>=0.5.0
requests>=2.20.0
model-optimizer/telemetry/__init__.py
View File
model-optimizer/telemetry/backend/__init__.py
+1
View File
@@ -0,0 +1 @@
from .backend_ga import *
model-optimizer/telemetry/backend/backend.py
+94
View File
@@ -0,0 +1,94 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import abc
from telemetry.utils.message import Message
class BackendRegistry:
"""
The class that stores information about all registered telemetry backends
"""
r = {}
@classmethod
def register_backend(cls, id: str, backend):
cls.r[id] = backend
@classmethod
def get_backend(cls, id: str):
assert id in cls.r, 'The backend with id "{}" is not registered'.format(id)
return cls.r.get(id)
class TelemetryBackendMetaClass(abc.ABCMeta):
def __init__(cls, clsname, bases, methods):
super().__init__(clsname, bases, methods)
if cls.id is not None:
BackendRegistry.register_backend(cls.id, cls)
class TelemetryBackend(metaclass=TelemetryBackendMetaClass):
id = None
@abc.abstractmethod
def __init__(self, tid: str, app_name: str, app_version: str):
"""
Initializer of the class
:param tid: database id
:param app_name: name of the application
:param app_version: version of the application
"""
@abc.abstractmethod
def send(self, message: Message):
"""
Sends the message to the backend.
:param message: The Message object to send
:return: None
"""
@abc.abstractmethod
def build_event_message(self, event_category: str, event_action: str, event_label: str, event_value: int = 1,
**kwargs):
"""
Should return the Message object build from the event message.
"""
@abc.abstractmethod
def build_error_message(self, error_msg: str, **kwargs):
"""
Should return the Message object build from the error message.
"""
@abc.abstractmethod
def build_stack_trace_message(self, error_msg: str, **kwargs):
"""
Should return the Message object build from the stack trace message.
"""
@abc.abstractmethod
def build_session_start_message(self, **kwargs):
"""
Should return the Message object corresponding to the session start.
"""
@abc.abstractmethod
def build_session_end_message(self, **kwargs):
"""
Should return the Message object corresponding to the session end.
"""
model-optimizer/telemetry/backend/backend_ga.py
+99
View File
@@ -0,0 +1,99 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import uuid
from telemetry.backend.backend import TelemetryBackend
from telemetry.utils.message import Message, MessageType
from telemetry.utils.guid import get_or_generate_uid
class GABackend(TelemetryBackend):
backend_url = 'https://www.google-analytics.com/collect'
id = 'ga'
def __init__(self, tid: str = None, app_name: str = None, app_version: str = None):
super(GABackend, self).__init__(tid, app_name, app_version)
if tid is None:
tid = 'UA-17808594-29'
self.tid = tid
self.uid = get_or_generate_uid('openvino_ga_uid', lambda: str(uuid.uuid4()), is_valid_uuid4)
self.app_name = app_name
self.app_version = app_version
self.default_message_attrs = {
'v': '1', # API Version
'tid': self.tid,
'cid': self.uid,
'an': self.app_name,
'av': self.app_version,
'ua': 'Opera/9.80 (Windows NT 6.0) Presto/2.12.388 Version/12.14' # dummy identifier of the browser
}
def send(self, message: Message):
try:
import requests
requests.post(self.backend_url, message.attrs, timeout=1.0)
except Exception:
pass
def build_event_message(self, event_category: str, event_action: str, event_label: str, event_value: int = 1,
**kwargs):
data = self.default_message_attrs.copy()
data.update({
't': 'event',
'ec': event_category,
'ea': event_action,
'el': event_label,
'ev': event_value,
})
return Message(MessageType.EVENT, data)
def build_session_start_message(self, **kwargs):
data = self.default_message_attrs.copy()
data.update({
'sc': 'start',
't': 'event',
'ec': 'session',
'ea': 'control',
'el': 'start',
'ev': 1,
})
return Message(MessageType.SESSION_START, data)
def build_session_end_message(self, **kwargs):
data = self.default_message_attrs.copy()
data.update({
'sc': 'end',
't': 'event',
'ec': 'session',
'ea': 'control',
'el': 'end',
'ev': 1,
})
return Message(MessageType.SESSION_END, data)
def build_error_message(self, error_msg: str, **kwargs):
pass
def build_stack_trace_message(self, error_msg: str, **kwargs):
pass
def is_valid_uuid4(uid: str):
try:
uuid.UUID(uid, version=4)
except ValueError:
return False
return True
model-optimizer/telemetry/telemetry.py
+103
View File
@@ -0,0 +1,103 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import telemetry.utils.isip as isip
from telemetry.backend.backend import BackendRegistry
from telemetry.utils.sender import TelemetrySender
class SingletonMetaClass(type):
def __init__(self, cls_name, super_classes, dic):
self.__single_instance = None
super().__init__(cls_name, super_classes, dic)
def __call__(cls, *args, **kwargs):
if cls.__single_instance is None:
cls.__single_instance = super(SingletonMetaClass, cls).__call__(*args, **kwargs)
return cls.__single_instance
class Telemetry(metaclass=SingletonMetaClass):
"""
The main class to send telemetry data. It uses singleton pattern. The instance should be initialized with the
application name, version and tracking id just once. Later the instance can be created without parameters.
"""
def __init__(self, app_name: str = None, app_version: str = None, tid: [None, str] = None,
backend: [str, None] = 'ga'):
if not hasattr(self, 'tid'):
self.tid = None
if app_name is not None:
self.consent = isip.isip_consent() == isip.ISIPConsent.APPROVED
# override default tid
if tid is not None:
self.tid = tid
self.backend = BackendRegistry.get_backend(backend)(self.tid, app_name, app_version)
self.sender = TelemetrySender()
else: # use already configured instance
assert self.sender is not None, 'The first instantiation of the Telemetry should be done with the ' \
'application name and version'
def force_shutdown(self, timeout: float = 1.0):
"""
Stops currently running threads which may be hanging because of no Internet connection.
:param timeout: maximum timeout time
:return: None
"""
self.sender.force_shutdown(timeout)
def send_event(self, event_category: str, event_action: str, event_label: str, event_value: int = 1, **kwargs):
"""
Send single event.
:param event_category: category of the event
:param event_action: action of the event
:param event_label: the label associated with the action
:param event_value: the integer value corresponding to this label
:param kwargs: additional parameters
:return: None
"""
if self.consent:
self.sender.send(self.backend, self.backend.build_event_message(event_category, event_action, event_label,
event_value, **kwargs))
def start_session(self, **kwargs):
"""
Sends a message about starting of a new session.
:param kwargs: additional parameters
:return: None
"""
if self.consent:
self.sender.send(self.backend, self.backend.build_session_start_message(**kwargs))
def end_session(self, **kwargs):
"""
Sends a message about ending of the current session.
:param kwargs: additional parameters
:return: None
"""
if self.consent:
self.sender.send(self.backend, self.backend.build_session_end_message(**kwargs))
def send_error(self, error_msg: str, **kwargs):
if self.consent:
pass
def send_stack_trace(self, stack_trace: str, **kwargs):
if self.consent:
pass
model-optimizer/telemetry/utils/__init__.py
View File
model-optimizer/telemetry/utils/guid.py
+77
View File
@@ -0,0 +1,77 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import os
from platform import system
import telemetry.utils.isip as isip
def save_uid_to_file(file_name: str, uid: str):
"""
Save the uid to the specified file
"""
try:
# create directories recursively first
os.makedirs(os.path.dirname(file_name), exist_ok=True)
with open(file_name, 'w') as file:
file.write(uid)
except Exception as e:
print('Failed to generate the UID file: {}'.format(str(e)))
return False
return True
def get_or_generate_uid(file_name: str, generator: callable, validator: [callable, None]):
"""
Get existing UID or generate a new one.
:param file_name: name of the file with the UID
:param generator: the function to generate the UID
:param validator: the function to validate the UID
:return: existing or a new UID file
"""
full_path = os.path.join(get_uid_path(), file_name)
uid = None
if os.path.exists(full_path):
with open(full_path, 'r') as file:
uid = file.readline().strip()
if uid is not None and (validator is not None and not validator(uid)):
uid = None
if uid is None:
uid = generator()
save_uid_to_file(full_path, uid)
return uid
def get_uid_path():
"""
Returns a directory with the the OpenVINO randomly generated UUID file.
:return: the directory with the the UUID file
"""
platform = system()
subdir = None
if platform == 'Windows':
subdir = 'Intel Corporation'
elif platform in ['Linux', 'Darwin']:
subdir = '.intel'
if subdir is None:
raise Exception('Failed to determine the operation system type')
return os.path.join(isip.isip_consent_base_dir(), subdir)
model-optimizer/telemetry/utils/isip.py
+82
View File
@@ -0,0 +1,82 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import os
from enum import Enum
from platform import system
class ISIPConsent(Enum):
APPROVED = 0
DECLINED = 1
UNKNOWN = 2
def isip_consent_base_dir():
"""
Returns the base directory with the ISIP consent file. The full directory may not have write access on Linux/OSX
systems so that is why the base directory is used.
:return:
"""
platform = system()
dir_to_check = None
if platform == 'Windows':
dir_to_check = '$LOCALAPPDATA'
elif platform in ['Linux', 'Darwin']:
dir_to_check = '$HOME'
if dir_to_check is None:
raise Exception('Failed to find location of the ISIP consent')
return os.path.expandvars(dir_to_check)
def _isip_consent_sub_directory():
platform = system()
if platform == 'Windows':
return 'Intel Corporation'
elif platform in ['Linux', 'Darwin']:
return 'intel'
raise Exception('Failed to find location of the ISIP consent')
def _isip_consent_dir():
dir_to_check = os.path.join(isip_consent_base_dir(), _isip_consent_sub_directory())
return os.path.expandvars(dir_to_check)
def _isip_consent_file():
return os.path.join(_isip_consent_dir(), 'isip')
def isip_consent():
file_to_check = _isip_consent_file()
if not os.path.exists(file_to_check):
return ISIPConsent.UNKNOWN
try:
with open(file_to_check, 'r') as file:
content = file.readline().strip()
if content == '1':
return ISIPConsent.APPROVED
else:
return ISIPConsent.DECLINED
except Exception as e:
pass
# unknown value in the file is considered as a unknown consent
return ISIPConsent.UNKNOWN
model-optimizer/telemetry/utils/message.py
+31
View File
@@ -0,0 +1,31 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
from enum import Enum
class MessageType(Enum):
EVENT = 0
ERROR = 1
STACK_TRACE = 2
SESSION_START = 3
SESSION_END = 4
class Message:
def __init__(self, type: MessageType, attrs: dict):
self.type = type
self.attrs = attrs.copy()
model-optimizer/telemetry/utils/sender.py
+63
View File
@@ -0,0 +1,63 @@
"""
Copyright (C) 2017-2021 Intel Corporation
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.
"""
import threading
from concurrent import futures
from time import sleep
from telemetry.backend.backend import TelemetryBackend
from telemetry.utils.message import Message
MAX_QUEUE_SIZE = 1000
class TelemetrySender:
def __init__(self, max_workers=None):
self.executor = futures.ThreadPoolExecutor(max_workers=max_workers)
self.queue_size = 0
self._lock = threading.Lock()
def send(self, backend: TelemetryBackend, message: Message):
def _future_callback(future):
with self._lock:
self.queue_size -= 1
with self._lock:
if self.queue_size < MAX_QUEUE_SIZE:
fut = self.executor.submit(backend.send, message)
fut.add_done_callback(_future_callback)
self.queue_size += 1
else:
pass # dropping a message because the queue is full
def force_shutdown(self, timeout: float):
"""
Forces all threads to be stopped after timeout. The "shutdown" method of the ThreadPoolExecutor removes only not
yet scheduled threads and keep running the existing one. In order to stop the already running use some low-level
attribute. The operation with low-level attributes is wrapped with the try/except to avoid potential crash if
these attributes will removed or renamed.
:param timeout: timeout to wait before the shutdown
:return: None
"""
try:
with self._lock:
if self.queue_size > 0:
sleep(timeout)
self.executor.shutdown(wait=False)
self.executor._threads.clear()
futures.thread._threads_queues.clear()
except Exception:
pass