refactor: multiprocessing helper testing codecov#225
Closed
NguyenHoangSon96 wants to merge 4 commits into
Closed
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #225 +/- ##
==========================================
- Coverage 87.21% 86.65% -0.56%
==========================================
Files 28 28
Lines 2072 2068 -4
==========================================
- Hits 1807 1792 -15
- Misses 265 276 +11 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
NguyenHoangSon96
changed the title
There was a problem hiding this comment.
Pull request overview
Refactors the MultiprocessingWriter helper to manage an internal multiprocessing process/context directly (rather than subclassing multiprocessing.Process), and updates integration tests and coverage configuration to support/validate multiprocessing start-method behavior.
Changes:
- Refactor
MultiprocessingWriterto create/manage its own process viamultiprocessing.get_context(start_method)andctx.Process(target=...). - Expand integration tests to validate multiprocessing start methods (
spawn,forkserver,fork) and adjust the existing multiprocessing helper test. - Add Coverage.py multiprocessing concurrency configuration in
pyproject.toml, and document the refactor inCHANGELOG.md.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 8 comments.
| File | Description |
|---|---|
tests/test_influxdb_client_3_integration.py |
Updates and expands integration tests for multiprocessing helper and start methods. |
pyproject.toml |
Adds Coverage.py multiprocessing concurrency/parallel configuration. |
influxdb_client_3/write_client/client/util/multiprocessing_helper.py |
Refactors multiprocessing writer to manage its own context/process and queue. |
CHANGELOG.md |
Adds release note entry describing the multiprocessing helper refactor. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+126
to
129
| 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``. | ||
| """ |
Comment on lines
+140
to
143
| 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 |
Comment on lines
176
to
179
| 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 |
Comment on lines
208
to
214
| if self.__started__: | ||
| self.queue_.put(_PoisonPill()) | ||
| self.queue_.join() | ||
| self.join() | ||
| self.process.join() | ||
| self.queue_ = None | ||
| self.__started__ = False | ||
| self.__disposed__ = True |
Comment on lines
+361
to
372
| self.assertEqual(mp.get_start_processing_method(), 'spawn') | ||
| mp.start() | ||
|
|
||
| measurement = f'test{random_hex(6)}'.lower() | ||
| for x in range(1, 10): | ||
| time.sleep(0.2) | ||
| writer.write( | ||
| 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()}" | ||
| ) | ||
| writer.__del__() | ||
| mp.__del__() | ||
|
|
| self.assertEqual(9, len(df)) | ||
| self.assertEqual(4, len(df)) | ||
|
|
||
| def test_multiprocessing_start_method_forkserver(self): |
| ) as mp: | ||
| self.assertEqual(mp.get_start_processing_method(), 'forkserver') | ||
|
|
||
| def test_multiprocessing_start_method_fork(self): |
Comment on lines
+28
to
+30
| 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`. |
NguyenHoangSon96
force-pushed
the
refactor/multiprocessing-helper-test
branch
from
July 23, 2026 08:39
a8154b8 to
c2175a3
Compare
NguyenHoangSon96
force-pushed
the
refactor/multiprocessing-helper-test
branch
from
July 23, 2026 09:25
806ffa8 to
d01d38b
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #
Proposed Changes
Briefly describe your proposed changes:
Checklist