Skip to content

feat: Use a daemon thread to monitor the go feature server exclusively #2391

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 23 commits into from
Mar 10, 2022
Merged

feat: Use a daemon thread to monitor the go feature server exclusively #2391

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
612c962
feat: Use a daemon thread to monitor the go feature server exclusively
achals Mar 9, 2022
f14ac9b
fix workflow
achals Mar 9, 2022
78bbd66
Fix setup.py build_python_protos
achals Mar 9, 2022
8014662
CR comments
achals Mar 9, 2022
2ed95ec
mkdir is needed
achals Mar 9, 2022
bff6e92
explicit commit
achals Mar 9, 2022
7f33768
flush file explicitly
achals Mar 9, 2022
99fb066
Flush registry changes
achals Mar 9, 2022
88db3ec
Signal handler can only be triggered from main thread
achals Mar 9, 2022
31dbbf0
join on background thread when cancelling and update test
achals Mar 9, 2022
0f3afa3
fix redis
achals Mar 9, 2022
f5710ff
add back go_build.py
achals Mar 9, 2022
88e2c49
wait on grpc connection check
achals Mar 10, 2022
2fa1ccf
Add a more robust wait
achals Mar 10, 2022
92ff707
Add a sleep
achals Mar 10, 2022
54f1a08
Remove explicitt cleanup
achals Mar 10, 2022
67c7c17
More defensive
achals Mar 10, 2022
a79ec1c
Even more defensive
achals Mar 10, 2022
f8f0402
Stop join
achals Mar 10, 2022
635bd2a
Clean up the process to ensure next one is fine
achals Mar 10, 2022
3ea3fac
join thread
achals Mar 10, 2022
b83d4b3
remove a sleep
achals Mar 10, 2022
39a66e3
Cleanup
achals Mar 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
wait on grpc connection check 
Signed-off-by: Achal Shah <achals@gmail.com>
  • Loading branch information
@achals
achals committed Mar 10, 2022
commit 88e2c49052c238032a9321b9cac86af57a6a2556
12 changes: 10 additions & 2 deletions sdk/python/feast/go_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,26 +250,34 @@ def run(self):
while not self._is_cancelled:

# If we fail to connect to grpc stub, terminate subprocess and repeat
_logger.info("Connecting to subprocess")
if not self._shared_connection.connect():
_logger.info("Failed to connect, killing and retrying")
self._shared_connection.kill_process()
continue
else:
_logger.debug("Go feature server started")
self._go_server_started.set()
_logger.info("Status: %s", self._is_cancelled)
while not self._is_cancelled:
try:
# Making a blocking wait by setting timeout to a very long time so we don't waste cpu cycle
self._shared_connection.wait_for_process(3600)
Comment thread
achals marked this conversation as resolved.
except subprocess.TimeoutExpired:
pass

_logger.info(
"No longer waiting for process: %s, %s, %s",
self._shared_connection._process.pid,
self._shared_connection._process.returncode,
self._shared_connection.is_process_alive(),
)
if not self._shared_connection.is_process_alive():
break
finally:
# Main thread exits
self._shared_connection.kill_process()

def stop(self):
_logger.debug("Stopping monitoring thread and terminating go feature server")
_logger.info("Stopping monitoring thread and terminating go feature server")
self._is_cancelled = True
self._shared_connection.kill_process()
20 changes: 12 additions & 8 deletions sdk/python/tests/integration/online_store/test_universal_online.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1020,14 +1020,17 @@ def test_go_server_life_cycle(go_cycle_environment, go_data_sources):
os.kill(go_fs_pid, signal.SIGTERM)
# At the same time checking that resources are clean up properly once child process is killed
# Check that background thread has terminated
monitor_thread = None
for thread in threading.enumerate():
if thread.name == "GoServerMonitorThread":
monitor_thread = thread
assert monitor_thread
assert monitor_thread.is_alive()
monitor_thread_alive = False
monitor_thread = fs._go_server._go_server_background_thread
assert monitor_thread.daemon

print(f"Monitor thread: {monitor_thread}, {monitor_thread.ident}")

for thread in threading.enumerate():
if thread.ident == monitor_thread.ident and thread.is_alive():
monitor_thread_alive = True
assert monitor_thread_alive

# Check if go server subprocess is still active even if background thread and process are killed
go_server_still_alive = False
for proc in psutil.process_iter():
Expand Down Expand Up @@ -1055,6 +1058,7 @@ def test_go_server_life_cycle(go_cycle_environment, go_data_sources):
)
new_go_fs_pid = fs._go_server._shared_connection._process.pid
assert new_go_fs_pid != go_fs_pid
fs._go_server._shared_connection._check_grpc_connection()

# Ensure process is still running.
assert fs._go_server._shared_connection._process.poll() is None
Expand All @@ -1068,8 +1072,8 @@ def test_go_server_life_cycle(go_cycle_environment, go_data_sources):
# Ensure process is dead.
assert fs._go_server._shared_connection._process.poll() is not None
Comment thread
achals marked this conversation as resolved.
Outdated
# Ensure monitoring thread is also dead.
live_threads = [t.name for t in threading.enumerate()]
assert "GoServerMonitorThread" not in live_threads
live_threads = [t.ident for t in threading.enumerate()]
assert monitor_thread.ident not in live_threads


def response_feature_name(feature: str, full_feature_names: bool) -> str:
Expand Down