-
Notifications
You must be signed in to change notification settings - Fork 198
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
put each radio type in its own module
in preparation for moving radio-specific radio code into these modules too
- Loading branch information
1 parent
2552add
commit 4d358d3
Showing
13 changed files
with
221 additions
and
201 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
This file was deleted.
Oops, something went wrong.
Empty file.
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,13 @@ | ||
import logging | ||
from abc import ABCMeta, abstractmethod | ||
from typing import Optional | ||
|
||
_db_manager_excepts: Optional[Exception] | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class MonitoringRadioSender(metaclass=ABCMeta): | ||
@abstractmethod | ||
def send(self, message: object) -> None: | ||
pass |
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,52 @@ | ||
import logging | ||
import os | ||
import pickle | ||
import uuid | ||
|
||
from parsl.monitoring.radios.base import MonitoringRadioSender | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class FilesystemRadioSender(MonitoringRadioSender): | ||
"""A MonitoringRadioSender that sends messages over a shared filesystem. | ||
The messsage directory structure is based on maildir, | ||
https://en.wikipedia.org/wiki/Maildir | ||
The writer creates a message in tmp/ and then when it is fully | ||
written, moves it atomically into new/ | ||
The reader ignores tmp/ and only reads and deletes messages from | ||
new/ | ||
This avoids a race condition of reading partially written messages. | ||
This radio is likely to give higher shared filesystem load compared to | ||
the UDP radio, but should be much more reliable. | ||
""" | ||
|
||
def __init__(self, *, monitoring_url: str, timeout: int = 10, run_dir: str): | ||
logger.info("filesystem based monitoring channel initializing") | ||
self.base_path = f"{run_dir}/monitor-fs-radio/" | ||
self.tmp_path = f"{self.base_path}/tmp" | ||
self.new_path = f"{self.base_path}/new" | ||
|
||
os.makedirs(self.tmp_path, exist_ok=True) | ||
os.makedirs(self.new_path, exist_ok=True) | ||
|
||
def send(self, message: object) -> None: | ||
logger.info("Sending a monitoring message via filesystem") | ||
|
||
unique_id = str(uuid.uuid4()) | ||
|
||
tmp_filename = f"{self.tmp_path}/{unique_id}" | ||
new_filename = f"{self.new_path}/{unique_id}" | ||
buffer = message | ||
|
||
# this will write the message out then atomically | ||
# move it into new/, so that a partially written | ||
# file will never be observed in new/ | ||
with open(tmp_filename, "wb") as f: | ||
pickle.dump(buffer, f) | ||
os.rename(tmp_filename, new_filename) |
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
import logging | ||
import pickle | ||
|
||
from parsl.monitoring.radios.base import MonitoringRadioSender | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class HTEXRadioSender(MonitoringRadioSender): | ||
|
||
def __init__(self, monitoring_url: str, timeout: int = 10): | ||
""" | ||
Parameters | ||
---------- | ||
monitoring_url : str | ||
URL of the form <scheme>://<IP>:<PORT> | ||
timeout : int | ||
timeout, default=10s | ||
""" | ||
logger.info("htex-based monitoring channel initialising") | ||
|
||
def send(self, message: object) -> None: | ||
""" Sends a message to the UDP receiver | ||
Parameter | ||
--------- | ||
message: object | ||
Arbitrary pickle-able object that is to be sent | ||
Returns: | ||
None | ||
""" | ||
|
||
import parsl.executors.high_throughput.monitoring_info | ||
|
||
result_queue = parsl.executors.high_throughput.monitoring_info.result_queue | ||
|
||
# this message needs to go in the result queue tagged so that it is treated | ||
# i) as a monitoring message by the interchange, and then further more treated | ||
# as a RESOURCE_INFO message when received by monitoring (rather than a NODE_INFO | ||
# which is the implicit default for messages from the interchange) | ||
|
||
# for the interchange, the outer wrapper, this needs to be a dict: | ||
|
||
interchange_msg = { | ||
'type': 'monitoring', | ||
'payload': message | ||
} | ||
|
||
if result_queue: | ||
result_queue.put(pickle.dumps(interchange_msg)) | ||
else: | ||
logger.error("result_queue is uninitialized - cannot put monitoring message") | ||
|
||
return |
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,17 @@ | ||
from multiprocessing.queues import Queue | ||
|
||
from parsl.monitoring.radios.base import MonitoringRadioSender | ||
|
||
|
||
class MultiprocessingQueueRadioSender(MonitoringRadioSender): | ||
"""A monitoring radio which connects over a multiprocessing Queue. | ||
This radio is intended to be used on the submit side, where components | ||
in the submit process, or processes launched by multiprocessing, will have | ||
access to a Queue shared with the monitoring database code (bypassing the | ||
monitoring router). | ||
""" | ||
def __init__(self, queue: Queue) -> None: | ||
self.queue = queue | ||
|
||
def send(self, message: object) -> None: | ||
self.queue.put(message) |
Oops, something went wrong.