From 6f6cb895194d2230f872c7d21100bea6fcaa3cd8 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Fri, 12 Jun 2026 19:07:52 +0700 Subject: [PATCH 1/7] refactor: multiprocess helper class --- CHANGELOG.md | 4 + .../client/util/multiprocessing_helper.py | 63 +++++++------ pyproject.toml | 7 +- tests/test_influxdb_client_3_integration.py | 88 ++++++++++++++----- 4 files changed, 107 insertions(+), 55 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dcd2226..878a71e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,10 @@ - `debug` - boolean - toggle debug level logging. - Any clients explicitly using the `Configuration`, `ApiClient`, `WriteService` or _legacy_ `InfluxDBClient` classes, will need to migrate their settings to the `InfluxDBClient3` constructor. +1. [#222](https://github.com/InfluxCommunity/influxdb3-python/pull/222): Refactor `MultiprocessingWriter` class. + - New Process will be created by using `DefaultContext.Process(target)` to prevent erratic crashes, handle cross-platform code behavior safely, and coordinate complex resource sharing. + - Users can now choose one of the start methods `fork`, `spawn` or `forkserver` when creating new Process. The default will be `spawn`. + ## 0.20.0 [2026-06-11] ### Features diff --git a/influxdb_client_3/write_client/client/util/multiprocessing_helper.py b/influxdb_client_3/write_client/client/util/multiprocessing_helper.py index 96ba6469..1bc8245a 100644 --- a/influxdb_client_3/write_client/client/util/multiprocessing_helper.py +++ b/influxdb_client_3/write_client/client/util/multiprocessing_helper.py @@ -7,9 +7,9 @@ import logging import multiprocessing -from influxdb_client_3 import InfluxDBClient3, write_client_options -from influxdb_client_3.write_client import WriteOptions +from influxdb_client_3 import write_client_options from influxdb_client_3.exceptions import InfluxDBError +from influxdb_client_3.write_client import WriteOptions, WriteApi logger = logging.getLogger('influxdb_client.client.util.multiprocessing_helper') @@ -35,9 +35,9 @@ class _PoisonPill: pass -class MultiprocessingWriter(multiprocessing.Process): +class MultiprocessingWriter: """ - The Helper class to write data into InfluxDB in independent OS process. + The Helper class to write data into InfluxDB in an independent OS process. Example: .. code-block:: python @@ -119,27 +119,27 @@ def main(): __started__ = False __disposed__ = False - def __init__(self, **kwargs) -> None: + def __init__(self, start_method='spawn', **kwargs) -> None: """ Initialize defaults. - For more information how to initialize the writer see the examples above. + For more information on how to initialize the writer, see the examples above. - :param kwargs: arguments are passed into ``__init__`` function of ``InfluxDBClient`` and ``write_api``. + :param kwargs: arguments are passed into the `` _ _init__`` function of ``InfluxDBClient`` and ``write_api``. """ - multiprocessing.Process.__init__(self) + self.ctx = multiprocessing.get_context(start_method) + self.process = self.ctx.Process(target=self.run) self.kwargs = kwargs - self.client = None self.write_api = None - self.queue_ = multiprocessing.Manager().Queue() + self.queue_ = self.ctx.JoinableQueue() def write(self, **kwargs) -> None: """ - Append time-series data into underlying queue. + Append time-series data into the underlying queue. - For more information how to pass arguments see the examples above. + For more information on how to pass arguments, see the examples above. - :param kwargs: arguments are passed into ``write`` function of ``WriteApi`` + :param kwargs: arguments are passed into the `` write `` function of ``WriteApi`` :return: None """ assert self.__disposed__ is False, 'Cannot write data: the writer is closed.' @@ -155,16 +155,13 @@ def run(self): retry_callback=self.kwargs.get('retry_callback', _retry_callback) ) - # Still need to create the InfluxDBClient3 because the init logics of InfluxDBClient3 will create the WriteApi. - # it will make WriteApi class created properly. - self.client = InfluxDBClient3(write_client_options=wco, **self.kwargs) - - # Close and set _query_api to None because query_api is not needed in this process. - # We only need write_api. - self.client._query_api.close() - self.client._query_api = None - - self.write_api = self.client._write_api + self.write_api = WriteApi( + bucket=self.kwargs.get('database'), + org=self.kwargs.get('org'), + default_header=self.kwargs.get('default_header'), + rest_client=self.kwargs.get('rest_client'), + **wco + ) # Infinite loop - until poison pill while True: next_record = self.queue_.get() @@ -177,26 +174,26 @@ def run(self): self.queue_.task_done() def start(self) -> None: - """Start independent process for writing data into InfluxDB.""" - super().start() + """Start an independent process for writing data into InfluxDB.""" + self.process.start() self.__started__ = True def terminate(self) -> None: """ - Cleanup resources in independent process. + Cleanup resources in an independent process. This function **cannot be used** to terminate the ``MultiprocessingWriter``. - If you want to finish your writes please call: ``__del__``. + If you want to finish your writes, please call: ``__del__``. """ if self.write_api: logger.info("flushing data...") - self.write_api.__del__() + self.write_api.close() self.write_api = None - if self.client: - self.client.close() - self.client = None logger.info("closed") + def get_start_processing_method(self): + return self.ctx.get_start_method() + def __enter__(self): """Enter the runtime context related to this object.""" self.start() @@ -207,11 +204,11 @@ def __exit__(self, exc_type, exc_value, traceback): self.__del__() def __del__(self): - """Dispose the client and write_api.""" + """Dispose of the client and write_api.""" if self.__started__: self.queue_.put(_PoisonPill()) self.queue_.join() - self.join() + self.process.join() self.queue_ = None self.__started__ = False self.__disposed__ = True diff --git a/pyproject.toml b/pyproject.toml index 758d2a02..d69ac4aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,8 @@ [build-system] requires = ["setuptools>=82.0.1"] -build-backend = "setuptools.build_meta" \ No newline at end of file +build-backend = "setuptools.build_meta" + +[tool.coverage.run] +concurrency = ["multiprocessing"] +parallel = true +source = ["influxdb_client_3.write_client.client.util"] diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index b6225bb5..ee39362a 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -1,10 +1,10 @@ +import asyncio import json import logging import os import random import string import time -import asyncio import unittest import pandas as pd @@ -21,8 +21,6 @@ from influxdb_client_3.write_client.write_exceptions import ApiException from tests.util import asyncio_run, lp_to_py_object -running_on_posix = os.name == 'posix' - def random_hex(len=6): return ''.join(random.choice(string.hexdigits) for i in range(len)) @@ -343,28 +341,76 @@ def test_batch_write_closed(self): list_results = reader.to_pylist() self.assertEqual(data_size, len(list_results)) - @pytest.mark.skipif(running_on_posix, reason="Skipping this test in POSIX environments") def test_multiprocessing_helper(self): - org = 'my-org' - writer = MultiprocessingWriter( - host=self.host, - database=self.database, - token=self.token, - org=org, - write_options=WriteOptions(batch_size=1)) - writer.start() - measurement = f'test{random_hex(6)}'.lower() - for x in range(1, 10): - time.sleep(0.2) - writer.write( - bucket=self.database, - record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" - ) - writer.__del__() + default_header = { + 'Authorization': f'Token {self.token}' + } + rest = rest_client.RestClient( + base_url=self.host, + default_header=default_header, + ) + + with MultiprocessingWriter( + host=self.host, + database=self.database, + token=self.token, + org='my-org', + default_header=default_header, + rest_client=rest, + write_options=WriteOptions(batch_size=1)) as mp: + self.assertEqual(mp.get_start_processing_method(), 'spawn') + + measurement = f'test{random_hex(6)}'.lower() + for x in range(1, 5): + time.sleep(0.5) + mp.write( + bucket=self.database, + record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" + ) time.sleep(1) df = self.client.query(f'select * from {measurement}', mode="pandas") - self.assertEqual(9, len(df)) + self.assertEqual(4, len(df)) + + # def test_multiprocessing_start_method_forkserver(self): + # default_header = { + # 'Authorization': f'Token {self.token}' + # } + # + # with MultiprocessingWriter( + # host=self.host, + # database=self.database, + # token=self.token, + # org='my-org', + # default_header=default_header, + # rest_client=(rest_client.RestClient( + # base_url=self.host, + # default_header=default_header, + # )), + # write_options=WriteOptions(batch_size=1), + # start_method='forkserver' + # ) as mp: + # self.assertEqual(mp.get_start_processing_method(), 'forkserver') + # + # def test_multiprocessing_start_method_fork(self): + # default_header = { + # 'Authorization': f'Token {self.token}' + # } + # + # with MultiprocessingWriter( + # host=self.host, + # database=self.database, + # token=self.token, + # org='my-org', + # default_header=default_header, + # rest_client=(rest_client.RestClient( + # base_url=self.host, + # default_header=default_header, + # )), + # write_options=WriteOptions(batch_size=1), + # start_method='fork' + # ) as mp: + # self.assertEqual(mp.get_start_processing_method(), 'fork') test_cert = """-----BEGIN CERTIFICATE----- MIIDUzCCAjugAwIBAgIUZB55ULutbc9gy6xLp1BkTQU7siowDQYJKoZIhvcNAQEL From 6fc57914e2b593e304b2f1b95533f92202b8799d Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 14:34:46 +0700 Subject: [PATCH 2/7] wip --- tests/test_influxdb_client_3_integration.py | 24 +++++++++++---------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index ee39362a..374b9a43 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -350,23 +350,25 @@ def test_multiprocessing_helper(self): default_header=default_header, ) - with MultiprocessingWriter( + mp = MultiprocessingWriter( host=self.host, database=self.database, token=self.token, org='my-org', default_header=default_header, rest_client=rest, - write_options=WriteOptions(batch_size=1)) as mp: - self.assertEqual(mp.get_start_processing_method(), 'spawn') - - measurement = f'test{random_hex(6)}'.lower() - for x in range(1, 5): - time.sleep(0.5) - mp.write( - bucket=self.database, - record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" - ) + write_options=WriteOptions(batch_size=1)) + self.assertEqual(mp.get_start_processing_method(), 'spawn') + mp.start() + + measurement = f'test{random_hex(6)}'.lower() + for x in range(1, 5): + time.sleep(0.5) + mp.write( + bucket=self.database, + record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" + ) + mp.__del__() time.sleep(1) df = self.client.query(f'select * from {measurement}', mode="pandas") From 845cd7cb734087333d9daefbd67e739832f25291 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 14:41:16 +0700 Subject: [PATCH 3/7] wip --- tests/test_influxdb_client_3_integration.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index 374b9a43..9be73701 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -351,13 +351,13 @@ def test_multiprocessing_helper(self): ) mp = MultiprocessingWriter( - host=self.host, - database=self.database, - token=self.token, - org='my-org', - default_header=default_header, - rest_client=rest, - write_options=WriteOptions(batch_size=1)) + host=self.host, + database=self.database, + token=self.token, + org='my-org', + default_header=default_header, + rest_client=rest, + write_options=WriteOptions(batch_size=1)) self.assertEqual(mp.get_start_processing_method(), 'spawn') mp.start() From d5b0d41dde893255e7085c1e94de41817297f6e9 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 17:50:52 +0700 Subject: [PATCH 4/7] refactor: multiprocess helper class --- codecov.yml | 7 ++ tests/test_influxdb_client_3_integration.py | 78 ++++++++++----------- 2 files changed, 46 insertions(+), 39 deletions(-) create mode 100644 codecov.yml diff --git a/codecov.yml b/codecov.yml new file mode 100644 index 00000000..90364494 --- /dev/null +++ b/codecov.yml @@ -0,0 +1,7 @@ +coverage: + status: + project: + default: + target: auto + removed_code_behavior: fully_covered_patch + threshold: 1% \ No newline at end of file diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index 9be73701..a4f2692d 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -374,45 +374,45 @@ def test_multiprocessing_helper(self): df = self.client.query(f'select * from {measurement}', mode="pandas") self.assertEqual(4, len(df)) - # def test_multiprocessing_start_method_forkserver(self): - # default_header = { - # 'Authorization': f'Token {self.token}' - # } - # - # with MultiprocessingWriter( - # host=self.host, - # database=self.database, - # token=self.token, - # org='my-org', - # default_header=default_header, - # rest_client=(rest_client.RestClient( - # base_url=self.host, - # default_header=default_header, - # )), - # write_options=WriteOptions(batch_size=1), - # start_method='forkserver' - # ) as mp: - # self.assertEqual(mp.get_start_processing_method(), 'forkserver') - # - # def test_multiprocessing_start_method_fork(self): - # default_header = { - # 'Authorization': f'Token {self.token}' - # } - # - # with MultiprocessingWriter( - # host=self.host, - # database=self.database, - # token=self.token, - # org='my-org', - # default_header=default_header, - # rest_client=(rest_client.RestClient( - # base_url=self.host, - # default_header=default_header, - # )), - # write_options=WriteOptions(batch_size=1), - # start_method='fork' - # ) as mp: - # self.assertEqual(mp.get_start_processing_method(), 'fork') + def test_multiprocessing_start_method_forkserver(self): + default_header = { + 'Authorization': f'Token {self.token}' + } + + with MultiprocessingWriter( + host=self.host, + database=self.database, + token=self.token, + org='my-org', + default_header=default_header, + rest_client=(rest_client.RestClient( + base_url=self.host, + default_header=default_header, + )), + write_options=WriteOptions(batch_size=1), + start_method='forkserver' + ) as mp: + self.assertEqual(mp.get_start_processing_method(), 'forkserver') + + def test_multiprocessing_start_method_fork(self): + default_header = { + 'Authorization': f'Token {self.token}' + } + + with MultiprocessingWriter( + host=self.host, + database=self.database, + token=self.token, + org='my-org', + default_header=default_header, + rest_client=(rest_client.RestClient( + base_url=self.host, + default_header=default_header, + )), + write_options=WriteOptions(batch_size=1), + start_method='fork' + ) as mp: + self.assertEqual(mp.get_start_processing_method(), 'fork') test_cert = """-----BEGIN CERTIFICATE----- MIIDUzCCAjugAwIBAgIUZB55ULutbc9gy6xLp1BkTQU7siowDQYJKoZIhvcNAQEL From 7112514ada50b5542cf2152b338b27ad12bbaf72 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 17:58:55 +0700 Subject: [PATCH 5/7] refactor: multiprocess helper class --- codecov.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/codecov.yml b/codecov.yml index 90364494..b2539fd7 100644 --- a/codecov.yml +++ b/codecov.yml @@ -3,5 +3,5 @@ coverage: project: default: target: auto - removed_code_behavior: fully_covered_patch - threshold: 1% \ No newline at end of file + removed_code_behavior: adjust_base +# threshold: 1% \ No newline at end of file From f40fdbda5ec711bea0bc04827dd1d43c466af197 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 18:04:37 +0700 Subject: [PATCH 6/7] refactor: multiprocess helper class --- codecov.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codecov.yml b/codecov.yml index b2539fd7..e75919da 100644 --- a/codecov.yml +++ b/codecov.yml @@ -4,4 +4,4 @@ coverage: default: target: auto removed_code_behavior: adjust_base -# threshold: 1% \ No newline at end of file + threshold: 1% \ No newline at end of file From 28bff8fd2025f00d317b8eb6648e0efa9f68cd27 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 23 Jul 2026 18:15:56 +0700 Subject: [PATCH 7/7] refactor: multiprocess helper class --- tests/test_influxdb_client_3_integration.py | 36 ++++++++++----------- 1 file changed, 17 insertions(+), 19 deletions(-) diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index a4f2692d..9f22173a 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -350,25 +350,23 @@ def test_multiprocessing_helper(self): default_header=default_header, ) - mp = MultiprocessingWriter( - host=self.host, - database=self.database, - token=self.token, - org='my-org', - default_header=default_header, - rest_client=rest, - write_options=WriteOptions(batch_size=1)) - self.assertEqual(mp.get_start_processing_method(), 'spawn') - mp.start() - - measurement = f'test{random_hex(6)}'.lower() - for x in range(1, 5): - time.sleep(0.5) - mp.write( - bucket=self.database, - record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" - ) - mp.__del__() + with MultiprocessingWriter( + host=self.host, + database=self.database, + token=self.token, + org='my-org', + default_header=default_header, + rest_client=rest, + write_options=WriteOptions(batch_size=1)) as mp: + self.assertEqual(mp.get_start_processing_method(), 'spawn') + + measurement = f'test{random_hex(6)}'.lower() + for x in range(1, 5): + time.sleep(0.5) + mp.write( + bucket=self.database, + record=f"{measurement},tag=a value=\"number{x}\" {time.time_ns()}" + ) time.sleep(1) df = self.client.query(f'select * from {measurement}', mode="pandas")