We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
LogEmitterProvider.force_flush hangs and ultimately timeouts randomly
**Describe your environment** 
Python 3.6.8
Opentelemetry Python 1.7.1
Opentelemetry Collector 0.38.0

**Steps to reproduce**
Note that this is a random behavior.
Regularly call LogEmitterProvider.force_flush while emitting logs.
We're using the BatchLogProcessor with schedule_delay_millis=200

**What is the expected behavior?**
No hanging

**What is the actual behavior?**
Hanging randomly

**Additional context**



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import abc
16 import collections
17 import enum
18 import logging
19 import os
20 import sys
21 import threading
22 from os import linesep
23 from typing import IO, Callable, Deque, List, Optional, Sequence
24 
25 from opentelemetry.context import attach, detach, set_value
26 from opentelemetry.sdk._logs import LogData, LogProcessor, LogRecord
27 from opentelemetry.util._time import _time_ns
28 
29 _logger = logging.getLogger(__name__)
30 
31 
32 class LogExportResult(enum.Enum):
33     SUCCESS = 0
34     FAILURE = 1
35 
36 
37 class LogExporter(abc.ABC):
38     """Interface for exporting logs.
39 
40     Interface to be implemented by services that want to export logs received
41     in their own format.
42 
43     To export data this MUST be registered to the :class`opentelemetry.sdk._logs.LogEmitter` using a
44     log processor.
45     """
46 
47     @abc.abstractmethod
48     def export(self, batch: Sequence[LogData]):
49         """Exports a batch of logs.
50 
51         Args:
52             batch: The list of `LogData` objects to be exported
53 
54         Returns:
55             The result of the export
56         """
57 
58     @abc.abstractmethod
59     def shutdown(self):
60         """Shuts down the exporter.
61 
62         Called when the SDK is shut down.
63         """
64 
65 
66 class ConsoleLogExporter(LogExporter):
67     """Implementation of :class:`LogExporter` that prints log records to the
68     console.
69 
70     This class can be used for diagnostic purposes. It prints the exported
71     log records to the console STDOUT.
72     """
73 
74     def __init__(
75         self,
76         out: IO = sys.stdout,
77         formatter: Callable[[LogRecord], str] = lambda record: record.to_json()
78         + linesep,
79     ):
80         self.out = out
81         self.formatter = formatter
82 
83     def export(self, batch: Sequence[LogData]):
84         for data in batch:
85             self.out.write(self.formatter(data.log_record))
86         self.out.flush()
87         return LogExportResult.SUCCESS
88 
89     def shutdown(self):
90         pass
91 
92 
93 class SimpleLogProcessor(LogProcessor):
94     """This is an implementation of LogProcessor which passes
95     received logs in the export-friendly LogData representation to the
96     configured LogExporter, as soon as they are emitted.
97     """
98 
99     def __init__(self, exporter: LogExporter):
100         self._exporter = exporter
101         self._shutdown = False
102 
103     def emit(self, log_data: LogData):
104         if self._shutdown:
105             _logger.warning("Processor is already shutdown, ignoring call")
106             return
107         token = attach(set_value("suppress_instrumentation", True))
108         try:
109             self._exporter.export((log_data,))
110         except Exception:  # pylint: disable=broad-except
111             _logger.exception("Exception while exporting logs.")
112         detach(token)
113 
114     def shutdown(self):
115         self._shutdown = True
116         self._exporter.shutdown()
117 
118     def force_flush(
119         self, timeout_millis: int = 30000
120     ) -> bool:  # pylint: disable=no-self-use
121         return True
122 
123 
124 class _FlushRequest:
125     __slots__ = ["event", "num_log_records"]
126 
127     def __init__(self):
128         self.event = threading.Event()
129         self.num_log_records = 0
130 
131 
132 class BatchLogProcessor(LogProcessor):
133     """This is an implementation of LogProcessor which creates batches of
134     received logs in the export-friendly LogData representation and
135     send to the configured LogExporter, as soon as they are emitted.
136     """
137 
138     def __init__(
139         self,
140         exporter: LogExporter,
141         schedule_delay_millis: int = 5000,
142         max_export_batch_size: int = 512,
143         export_timeout_millis: int = 30000,
144     ):
145         self._exporter = exporter
146         self._schedule_delay_millis = schedule_delay_millis
147         self._max_export_batch_size = max_export_batch_size
148         self._export_timeout_millis = export_timeout_millis
149         self._queue = collections.deque()  # type: Deque[LogData]
150         self._worker_thread = threading.Thread(target=self.worker, daemon=True)
151         self._condition = threading.Condition(threading.Lock())
152         self._shutdown = False
153         self._flush_request = None  # type: Optional[_FlushRequest]
154         self._log_records = [
155             None
156         ] * self._max_export_batch_size  # type: List[Optional[LogData]]
157         self._worker_thread.start()
158         # Only available in *nix since py37.
159         if hasattr(os, "register_at_fork"):
160             os.register_at_fork(
161                 after_in_child=self._at_fork_reinit
162             )  # pylint: disable=protected-access
163 
164     def _at_fork_reinit(self):
165         self._condition = threading.Condition(threading.Lock())
166         self._queue.clear()
167         self._worker_thread = threading.Thread(target=self.worker, daemon=True)
168         self._worker_thread.start()
169 
170     def worker(self):
171         timeout = self._schedule_delay_millis / 1e3
172         flush_request = None  # type: Optional[_FlushRequest]
173         while not self._shutdown:
174             with self._condition:
175                 if self._shutdown:
176                     # shutdown may have been called, avoid further processing
177                     break
178                 flush_request = self._get_and_unset_flush_request()
179                 if (
180                     len(self._queue) < self._max_export_batch_size
181                     and self._flush_request is None
182                 ):
183                     self._condition.wait(timeout)
184 
185                     flush_request = self._get_and_unset_flush_request()
186                     if not self._queue:
187                         timeout = self._schedule_delay_millis / 1e3
188                         self._notify_flush_request_finished(flush_request)
189                         flush_request = None
190                         continue
191                     if self._shutdown:
192                         break
193 
194             start_ns = _time_ns()
195             self._export(flush_request)
196             end_ns = _time_ns()
197             # subtract the duration of this export call to the next timeout
198             timeout = self._schedule_delay_millis / 1e3 - (
199                 (end_ns - start_ns) / 1e9
200             )
201 
202             self._notify_flush_request_finished(flush_request)
203             flush_request = None
204 
205         # there might have been a new flush request while export was running
206         # and before the done flag switched to true
207         with self._condition:
208             shutdown_flush_request = self._get_and_unset_flush_request()
209 
210         # flush the remaining logs
211         self._drain_queue()
212         self._notify_flush_request_finished(flush_request)
213         self._notify_flush_request_finished(shutdown_flush_request)
214 
215     def _export(self, flush_request: Optional[_FlushRequest] = None):
216         """Exports logs considering the given flush_request.
217 
218         If flush_request is not None then logs are exported in batches
219         until the number of exported logs reached or exceeded the num of logs in
220         flush_request, otherwise exports at max max_export_batch_size logs.
221         """
222         if flush_request is None:
223             self._export_batch()
224             return
225 
226         num_log_records = flush_request.num_log_records
227         while self._queue:
228             exported = self._export_batch()
229             num_log_records -= exported
230 
231             if num_log_records <= 0:
232                 break
233 
234     def _export_batch(self) -> int:
235         """Exports at most max_export_batch_size logs and returns the number of
236         exported logs.
237         """
238         idx = 0
239         while idx < self._max_export_batch_size and self._queue:
240             record = self._queue.pop()
241             self._log_records[idx] = record
242             idx += 1
243         token = attach(set_value("suppress_instrumentation", True))
244         try:
245             self._exporter.export(self._log_records[:idx])  # type: ignore
246         except Exception:  # pylint: disable=broad-except
247             _logger.exception("Exception while exporting logs.")
248         detach(token)
249 
250         for index in range(idx):
251             self._log_records[index] = None
252         return idx
253 
254     def _drain_queue(self):
255         """Export all elements until queue is empty.
256 
257         Can only be called from the worker thread context because it invokes
258         `export` that is not thread safe.
259         """
260         while self._queue:
261             self._export_batch()
262 
263     def _get_and_unset_flush_request(self) -> Optional[_FlushRequest]:
264         flush_request = self._flush_request
265         self._flush_request = None
266         if flush_request is not None:
267             flush_request.num_log_records = len(self._queue)
268         return flush_request
269 
270     @staticmethod
271     def _notify_flush_request_finished(
272         flush_request: Optional[_FlushRequest] = None,
273     ):
274         if flush_request is not None:
275             flush_request.event.set()
276 
277     def _get_or_create_flush_request(self) -> _FlushRequest:
278         if self._flush_request is None:
279             self._flush_request = _FlushRequest()
280         return self._flush_request
281 
282     def emit(self, log_data: LogData) -> None:
283         """Adds the `LogData` to queue and notifies the waiting threads
284         when size of queue reaches max_export_batch_size.
285         """
286         if self._shutdown:
287             return
288         self._queue.appendleft(log_data)
289         if len(self._queue) >= self._max_export_batch_size:
290             with self._condition:
291                 self._condition.notify()
292 
293     def shutdown(self):
294         self._shutdown = True
295         with self._condition:
296             self._condition.notify_all()
297         self._worker_thread.join()
298         self._exporter.shutdown()
299 
300     def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
301         if timeout_millis is None:
302             timeout_millis = self._export_timeout_millis
303         if self._shutdown:
304             return True
305 
306         with self._condition:
307             flush_request = self._get_or_create_flush_request()
308             self._condition.notify_all()
309 
310         ret = flush_request.event.wait(timeout_millis / 1e3)
311         if not ret:
312             _logger.warning("Timeout was exceeded in force_flush().")
313         return ret
314
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py
--- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py
@@ -178,7 +178,7 @@
                 flush_request = self._get_and_unset_flush_request()
                 if (
                     len(self._queue) < self._max_export_batch_size
-                    and self._flush_request is None
+                    and flush_request is None
                 ):
                     self._condition.wait(timeout)

{"golden_diff": "diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py\n--- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py\n+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py\n@@ -178,7 +178,7 @@\n                 flush_request = self._get_and_unset_flush_request()\n                 if (\n                     len(self._queue) < self._max_export_batch_size\n-                    and self._flush_request is None\n+                    and flush_request is None\n                 ):\n                     self._condition.wait(timeout)\n", "issue": "LogEmitterProvider.force_flush hangs and ultimately timeouts randomly\n**Describe your environment** \r\nPython 3.6.8\r\nOpentelemetry Python 1.7.1\r\nOpentelemetry Collector 0.38.0\r\n\r\n**Steps to reproduce**\r\nNote that this is a random behavior.\r\nRegularly call LogEmitterProvider.force_flush while emitting logs.\r\nWe're using the BatchLogProcessor with schedule_delay_millis=200\r\n\r\n**What is the expected behavior?**\r\nNo hanging\r\n\r\n**What is the actual behavior?**\r\nHanging randomly\r\n\r\n**Additional context**\r\n\r\n\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport abc\nimport collections\nimport enum\nimport logging\nimport os\nimport sys\nimport threading\nfrom os import linesep\nfrom typing import IO, Callable, Deque, List, Optional, Sequence\n\nfrom opentelemetry.context import attach, detach, set_value\nfrom opentelemetry.sdk._logs import LogData, LogProcessor, LogRecord\nfrom opentelemetry.util._time import _time_ns\n\n_logger = logging.getLogger(__name__)\n\n\nclass LogExportResult(enum.Enum):\n    SUCCESS = 0\n    FAILURE = 1\n\n\nclass LogExporter(abc.ABC):\n    \"\"\"Interface for exporting logs.\n\n    Interface to be implemented by services that want to export logs received\n    in their own format.\n\n    To export data this MUST be registered to the :class`opentelemetry.sdk._logs.LogEmitter` using a\n    log processor.\n    \"\"\"\n\n    @abc.abstractmethod\n    def export(self, batch: Sequence[LogData]):\n        \"\"\"Exports a batch of logs.\n\n        Args:\n            batch: The list of `LogData` objects to be exported\n\n        Returns:\n            The result of the export\n        \"\"\"\n\n    @abc.abstractmethod\n    def shutdown(self):\n        \"\"\"Shuts down the exporter.\n\n        Called when the SDK is shut down.\n        \"\"\"\n\n\nclass ConsoleLogExporter(LogExporter):\n    \"\"\"Implementation of :class:`LogExporter` that prints log records to the\n    console.\n\n    This class can be used for diagnostic purposes. It prints the exported\n    log records to the console STDOUT.\n    \"\"\"\n\n    def __init__(\n        self,\n        out: IO = sys.stdout,\n        formatter: Callable[[LogRecord], str] = lambda record: record.to_json()\n        + linesep,\n    ):\n        self.out = out\n        self.formatter = formatter\n\n    def export(self, batch: Sequence[LogData]):\n        for data in batch:\n            self.out.write(self.formatter(data.log_record))\n        self.out.flush()\n        return LogExportResult.SUCCESS\n\n    def shutdown(self):\n        pass\n\n\nclass SimpleLogProcessor(LogProcessor):\n    \"\"\"This is an implementation of LogProcessor which passes\n    received logs in the export-friendly LogData representation to the\n    configured LogExporter, as soon as they are emitted.\n    \"\"\"\n\n    def __init__(self, exporter: LogExporter):\n        self._exporter = exporter\n        self._shutdown = False\n\n    def emit(self, log_data: LogData):\n        if self._shutdown:\n            _logger.warning(\"Processor is already shutdown, ignoring call\")\n            return\n        token = attach(set_value(\"suppress_instrumentation\", True))\n        try:\n            self._exporter.export((log_data,))\n        except Exception:  # pylint: disable=broad-except\n            _logger.exception(\"Exception while exporting logs.\")\n        detach(token)\n\n    def shutdown(self):\n        self._shutdown = True\n        self._exporter.shutdown()\n\n    def force_flush(\n        self, timeout_millis: int = 30000\n    ) -> bool:  # pylint: disable=no-self-use\n        return True\n\n\nclass _FlushRequest:\n    __slots__ = [\"event\", \"num_log_records\"]\n\n    def __init__(self):\n        self.event = threading.Event()\n        self.num_log_records = 0\n\n\nclass BatchLogProcessor(LogProcessor):\n    \"\"\"This is an implementation of LogProcessor which creates batches of\n    received logs in the export-friendly LogData representation and\n    send to the configured LogExporter, as soon as they are emitted.\n    \"\"\"\n\n    def __init__(\n        self,\n        exporter: LogExporter,\n        schedule_delay_millis: int = 5000,\n        max_export_batch_size: int = 512,\n        export_timeout_millis: int = 30000,\n    ):\n        self._exporter = exporter\n        self._schedule_delay_millis = schedule_delay_millis\n        self._max_export_batch_size = max_export_batch_size\n        self._export_timeout_millis = export_timeout_millis\n        self._queue = collections.deque()  # type: Deque[LogData]\n        self._worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self._condition = threading.Condition(threading.Lock())\n        self._shutdown = False\n        self._flush_request = None  # type: Optional[_FlushRequest]\n        self._log_records = [\n            None\n        ] * self._max_export_batch_size  # type: List[Optional[LogData]]\n        self._worker_thread.start()\n        # Only available in *nix since py37.\n        if hasattr(os, \"register_at_fork\"):\n            os.register_at_fork(\n                after_in_child=self._at_fork_reinit\n            )  # pylint: disable=protected-access\n\n    def _at_fork_reinit(self):\n        self._condition = threading.Condition(threading.Lock())\n        self._queue.clear()\n        self._worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self._worker_thread.start()\n\n    def worker(self):\n        timeout = self._schedule_delay_millis / 1e3\n        flush_request = None  # type: Optional[_FlushRequest]\n        while not self._shutdown:\n            with self._condition:\n                if self._shutdown:\n                    # shutdown may have been called, avoid further processing\n                    break\n                flush_request = self._get_and_unset_flush_request()\n                if (\n                    len(self._queue) < self._max_export_batch_size\n                    and self._flush_request is None\n                ):\n                    self._condition.wait(timeout)\n\n                    flush_request = self._get_and_unset_flush_request()\n                    if not self._queue:\n                        timeout = self._schedule_delay_millis / 1e3\n                        self._notify_flush_request_finished(flush_request)\n                        flush_request = None\n                        continue\n                    if self._shutdown:\n                        break\n\n            start_ns = _time_ns()\n            self._export(flush_request)\n            end_ns = _time_ns()\n            # subtract the duration of this export call to the next timeout\n            timeout = self._schedule_delay_millis / 1e3 - (\n                (end_ns - start_ns) / 1e9\n            )\n\n            self._notify_flush_request_finished(flush_request)\n            flush_request = None\n\n        # there might have been a new flush request while export was running\n        # and before the done flag switched to true\n        with self._condition:\n            shutdown_flush_request = self._get_and_unset_flush_request()\n\n        # flush the remaining logs\n        self._drain_queue()\n        self._notify_flush_request_finished(flush_request)\n        self._notify_flush_request_finished(shutdown_flush_request)\n\n    def _export(self, flush_request: Optional[_FlushRequest] = None):\n        \"\"\"Exports logs considering the given flush_request.\n\n        If flush_request is not None then logs are exported in batches\n        until the number of exported logs reached or exceeded the num of logs in\n        flush_request, otherwise exports at max max_export_batch_size logs.\n        \"\"\"\n        if flush_request is None:\n            self._export_batch()\n            return\n\n        num_log_records = flush_request.num_log_records\n        while self._queue:\n            exported = self._export_batch()\n            num_log_records -= exported\n\n            if num_log_records <= 0:\n                break\n\n    def _export_batch(self) -> int:\n        \"\"\"Exports at most max_export_batch_size logs and returns the number of\n        exported logs.\n        \"\"\"\n        idx = 0\n        while idx < self._max_export_batch_size and self._queue:\n            record = self._queue.pop()\n            self._log_records[idx] = record\n            idx += 1\n        token = attach(set_value(\"suppress_instrumentation\", True))\n        try:\n            self._exporter.export(self._log_records[:idx])  # type: ignore\n        except Exception:  # pylint: disable=broad-except\n            _logger.exception(\"Exception while exporting logs.\")\n        detach(token)\n\n        for index in range(idx):\n            self._log_records[index] = None\n        return idx\n\n    def _drain_queue(self):\n        \"\"\"Export all elements until queue is empty.\n\n        Can only be called from the worker thread context because it invokes\n        `export` that is not thread safe.\n        \"\"\"\n        while self._queue:\n            self._export_batch()\n\n    def _get_and_unset_flush_request(self) -> Optional[_FlushRequest]:\n        flush_request = self._flush_request\n        self._flush_request = None\n        if flush_request is not None:\n            flush_request.num_log_records = len(self._queue)\n        return flush_request\n\n    @staticmethod\n    def _notify_flush_request_finished(\n        flush_request: Optional[_FlushRequest] = None,\n    ):\n        if flush_request is not None:\n            flush_request.event.set()\n\n    def _get_or_create_flush_request(self) -> _FlushRequest:\n        if self._flush_request is None:\n            self._flush_request = _FlushRequest()\n        return self._flush_request\n\n    def emit(self, log_data: LogData) -> None:\n        \"\"\"Adds the `LogData` to queue and notifies the waiting threads\n        when size of queue reaches max_export_batch_size.\n        \"\"\"\n        if self._shutdown:\n            return\n        self._queue.appendleft(log_data)\n        if len(self._queue) >= self._max_export_batch_size:\n            with self._condition:\n                self._condition.notify()\n\n    def shutdown(self):\n        self._shutdown = True\n        with self._condition:\n            self._condition.notify_all()\n        self._worker_thread.join()\n        self._exporter.shutdown()\n\n    def force_flush(self, timeout_millis: Optional[int] = None) -> bool:\n        if timeout_millis is None:\n            timeout_millis = self._export_timeout_millis\n        if self._shutdown:\n            return True\n\n        with self._condition:\n            flush_request = self._get_or_create_flush_request()\n            self._condition.notify_all()\n\n        ret = flush_request.event.wait(timeout_millis / 1e3)\n        if not ret:\n            _logger.warning(\"Timeout was exceeded in force_flush().\")\n        return ret\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport abc\nimport collections\nimport enum\nimport logging\nimport os\nimport sys\nimport threading\nfrom os import linesep\nfrom typing import IO, Callable, Deque, List, Optional, Sequence\n\nfrom opentelemetry.context import attach, detach, set_value\nfrom opentelemetry.sdk._logs import LogData, LogProcessor, LogRecord\nfrom opentelemetry.util._time import _time_ns\n\n_logger = logging.getLogger(__name__)\n\n\nclass LogExportResult(enum.Enum):\n    SUCCESS = 0\n    FAILURE = 1\n\n\nclass LogExporter(abc.ABC):\n    \"\"\"Interface for exporting logs.\n\n    Interface to be implemented by services that want to export logs received\n    in their own format.\n\n    To export data this MUST be registered to the :class`opentelemetry.sdk._logs.LogEmitter` using a\n    log processor.\n    \"\"\"\n\n    @abc.abstractmethod\n    def export(self, batch: Sequence[LogData]):\n        \"\"\"Exports a batch of logs.\n\n        Args:\n            batch: The list of `LogData` objects to be exported\n\n        Returns:\n            The result of the export\n        \"\"\"\n\n    @abc.abstractmethod\n    def shutdown(self):\n        \"\"\"Shuts down the exporter.\n\n        Called when the SDK is shut down.\n        \"\"\"\n\n\nclass ConsoleLogExporter(LogExporter):\n    \"\"\"Implementation of :class:`LogExporter` that prints log records to the\n    console.\n\n    This class can be used for diagnostic purposes. It prints the exported\n    log records to the console STDOUT.\n    \"\"\"\n\n    def __init__(\n        self,\n        out: IO = sys.stdout,\n        formatter: Callable[[LogRecord], str] = lambda record: record.to_json()\n        + linesep,\n    ):\n        self.out = out\n        self.formatter = formatter\n\n    def export(self, batch: Sequence[LogData]):\n        for data in batch:\n            self.out.write(self.formatter(data.log_record))\n        self.out.flush()\n        return LogExportResult.SUCCESS\n\n    def shutdown(self):\n        pass\n\n\nclass SimpleLogProcessor(LogProcessor):\n    \"\"\"This is an implementation of LogProcessor which passes\n    received logs in the export-friendly LogData representation to the\n    configured LogExporter, as soon as they are emitted.\n    \"\"\"\n\n    def __init__(self, exporter: LogExporter):\n        self._exporter = exporter\n        self._shutdown = False\n\n    def emit(self, log_data: LogData):\n        if self._shutdown:\n            _logger.warning(\"Processor is already shutdown, ignoring call\")\n            return\n        token = attach(set_value(\"suppress_instrumentation\", True))\n        try:\n            self._exporter.export((log_data,))\n        except Exception:  # pylint: disable=broad-except\n            _logger.exception(\"Exception while exporting logs.\")\n        detach(token)\n\n    def shutdown(self):\n        self._shutdown = True\n        self._exporter.shutdown()\n\n    def force_flush(\n        self, timeout_millis: int = 30000\n    ) -> bool:  # pylint: disable=no-self-use\n        return True\n\n\nclass _FlushRequest:\n    __slots__ = [\"event\", \"num_log_records\"]\n\n    def __init__(self):\n        self.event = threading.Event()\n        self.num_log_records = 0\n\n\nclass BatchLogProcessor(LogProcessor):\n    \"\"\"This is an implementation of LogProcessor which creates batches of\n    received logs in the export-friendly LogData representation and\n    send to the configured LogExporter, as soon as they are emitted.\n    \"\"\"\n\n    def __init__(\n        self,\n        exporter: LogExporter,\n        schedule_delay_millis: int = 5000,\n        max_export_batch_size: int = 512,\n        export_timeout_millis: int = 30000,\n    ):\n        self._exporter = exporter\n        self._schedule_delay_millis = schedule_delay_millis\n        self._max_export_batch_size = max_export_batch_size\n        self._export_timeout_millis = export_timeout_millis\n        self._queue = collections.deque()  # type: Deque[LogData]\n        self._worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self._condition = threading.Condition(threading.Lock())\n        self._shutdown = False\n        self._flush_request = None  # type: Optional[_FlushRequest]\n        self._log_records = [\n            None\n        ] * self._max_export_batch_size  # type: List[Optional[LogData]]\n        self._worker_thread.start()\n        # Only available in *nix since py37.\n        if hasattr(os, \"register_at_fork\"):\n            os.register_at_fork(\n                after_in_child=self._at_fork_reinit\n            )  # pylint: disable=protected-access\n\n    def _at_fork_reinit(self):\n        self._condition = threading.Condition(threading.Lock())\n        self._queue.clear()\n        self._worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self._worker_thread.start()\n\n    def worker(self):\n        timeout = self._schedule_delay_millis / 1e3\n        flush_request = None  # type: Optional[_FlushRequest]\n        while not self._shutdown:\n            with self._condition:\n                if self._shutdown:\n                    # shutdown may have been called, avoid further processing\n                    break\n                flush_request = self._get_and_unset_flush_request()\n                if (\n                    len(self._queue) < self._max_export_batch_size\n                    and flush_request is None\n                ):\n                    self._condition.wait(timeout)\n\n                    flush_request = self._get_and_unset_flush_request()\n                    if not self._queue:\n                        timeout = self._schedule_delay_millis / 1e3\n                        self._notify_flush_request_finished(flush_request)\n                        flush_request = None\n                        continue\n                    if self._shutdown:\n                        break\n\n            start_ns = _time_ns()\n            self._export(flush_request)\n            end_ns = _time_ns()\n            # subtract the duration of this export call to the next timeout\n            timeout = self._schedule_delay_millis / 1e3 - (\n                (end_ns - start_ns) / 1e9\n            )\n\n            self._notify_flush_request_finished(flush_request)\n            flush_request = None\n\n        # there might have been a new flush request while export was running\n        # and before the done flag switched to true\n        with self._condition:\n            shutdown_flush_request = self._get_and_unset_flush_request()\n\n        # flush the remaining logs\n        self._drain_queue()\n        self._notify_flush_request_finished(flush_request)\n        self._notify_flush_request_finished(shutdown_flush_request)\n\n    def _export(self, flush_request: Optional[_FlushRequest] = None):\n        \"\"\"Exports logs considering the given flush_request.\n\n        If flush_request is not None then logs are exported in batches\n        until the number of exported logs reached or exceeded the num of logs in\n        flush_request, otherwise exports at max max_export_batch_size logs.\n        \"\"\"\n        if flush_request is None:\n            self._export_batch()\n            return\n\n        num_log_records = flush_request.num_log_records\n        while self._queue:\n            exported = self._export_batch()\n            num_log_records -= exported\n\n            if num_log_records <= 0:\n                break\n\n    def _export_batch(self) -> int:\n        \"\"\"Exports at most max_export_batch_size logs and returns the number of\n        exported logs.\n        \"\"\"\n        idx = 0\n        while idx < self._max_export_batch_size and self._queue:\n            record = self._queue.pop()\n            self._log_records[idx] = record\n            idx += 1\n        token = attach(set_value(\"suppress_instrumentation\", True))\n        try:\n            self._exporter.export(self._log_records[:idx])  # type: ignore\n        except Exception:  # pylint: disable=broad-except\n            _logger.exception(\"Exception while exporting logs.\")\n        detach(token)\n\n        for index in range(idx):\n            self._log_records[index] = None\n        return idx\n\n    def _drain_queue(self):\n        \"\"\"Export all elements until queue is empty.\n\n        Can only be called from the worker thread context because it invokes\n        `export` that is not thread safe.\n        \"\"\"\n        while self._queue:\n            self._export_batch()\n\n    def _get_and_unset_flush_request(self) -> Optional[_FlushRequest]:\n        flush_request = self._flush_request\n        self._flush_request = None\n        if flush_request is not None:\n            flush_request.num_log_records = len(self._queue)\n        return flush_request\n\n    @staticmethod\n    def _notify_flush_request_finished(\n        flush_request: Optional[_FlushRequest] = None,\n    ):\n        if flush_request is not None:\n            flush_request.event.set()\n\n    def _get_or_create_flush_request(self) -> _FlushRequest:\n        if self._flush_request is None:\n            self._flush_request = _FlushRequest()\n        return self._flush_request\n\n    def emit(self, log_data: LogData) -> None:\n        \"\"\"Adds the `LogData` to queue and notifies the waiting threads\n        when size of queue reaches max_export_batch_size.\n        \"\"\"\n        if self._shutdown:\n            return\n        self._queue.appendleft(log_data)\n        if len(self._queue) >= self._max_export_batch_size:\n            with self._condition:\n                self._condition.notify()\n\n    def shutdown(self):\n        self._shutdown = True\n        with self._condition:\n            self._condition.notify_all()\n        self._worker_thread.join()\n        self._exporter.shutdown()\n\n    def force_flush(self, timeout_millis: Optional[int] = None) -> bool:\n        if timeout_millis is None:\n            timeout_millis = self._export_timeout_millis\n        if self._shutdown:\n            return True\n\n        with self._condition:\n            flush_request = self._get_or_create_flush_request()\n            self._condition.notify_all()\n\n        ret = flush_request.event.wait(timeout_millis / 1e3)\n        if not ret:\n            _logger.warning(\"Timeout was exceeded in force_flush().\")\n        return ret\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_logs/export/__init__.py"}]}