Skip to content

fix: add missing sls core tests #274

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
justinmerrell merged 22 commits into from
Jan 30, 2024
Merged

fix: add missing sls core tests #274

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
ffdb382
fix: add missing sls core tests
justinmerrell Jan 18, 2024
e21c536
Update core.py
justinmerrell Jan 18, 2024
9ad6752
Update core.py
justinmerrell Jan 18, 2024
f7dd850
Update core.py
justinmerrell Jan 18, 2024
db35ebb
Update core.py
justinmerrell Jan 19, 2024
aa5b39a
Update core.py
justinmerrell Jan 19, 2024
b1827f5
Merge pull request #279 from runpod/main
justinmerrell Jan 21, 2024
89c1a65
Merge pull request #288 from runpod/main
justinmerrell Jan 29, 2024
2804818
feat: added logging
justinmerrell Jan 29, 2024
fad0d98
Update core.py
justinmerrell Jan 29, 2024
ccbfbfe
Update core.py
justinmerrell Jan 29, 2024
ee2b3b7
Update core.py
justinmerrell Jan 29, 2024
8e82450
Update core.py
justinmerrell Jan 29, 2024
1545377
Update core.py
justinmerrell Jan 29, 2024
4cd8d3d
Update core.py
justinmerrell Jan 29, 2024
3362456
Update sls_core.so
justinmerrell Jan 29, 2024
08851eb
Update sls_core.so
justinmerrell Jan 30, 2024
e90e72e
Update sls_core.so
justinmerrell Jan 30, 2024
c28ab8d
Update sls_core.so
justinmerrell Jan 30, 2024
5d9b5b8
Update core.py
justinmerrell Jan 30, 2024
561bb6b
Update sls_core.so
justinmerrell Jan 30, 2024
8cf43e9
Update CHANGELOG.md
justinmerrell Jan 30, 2024
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
88 changes: 78 additions & 10 deletions .github/tests.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,8 @@
{
"hardwareConfig": {
"endpointConfig": {
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80",
"name": "runpod-python E2E Test - Basic"
"name": "runpod-python E2E Test - Basic",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
}
},
"input": {
Expand All @@ -13,8 +13,8 @@
{
"hardwareConfig": {
"endpointConfig": {
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80",
"name": "runpod-python E2E Test - Long Job"
"name": "runpod-python E2E Test - Long Job",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
}
},
"input": {
Expand All @@ -25,8 +25,8 @@
{
"hardwareConfig": {
"endpointConfig": {
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80",
"name": "runpod-python E2E Test - Generator Handler"
"name": "runpod-python E2E Test - Generator Handler",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"dockerArgs": "python3 -u /handler.py --generator --return_aggregate_stream"
Expand All @@ -43,8 +43,8 @@
{
"hardwareConfig": {
"endpointConfig": {
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80",
"name": "runpod-python E2E Test - Async Generator Handler"
"name": "runpod-python E2E Test - Async Generator Handler",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"dockerArgs": "python3 -u /handler.py --async_generator --return_aggregate_stream"
Expand All @@ -61,8 +61,8 @@
{
"hardwareConfig": {
"endpointConfig": {
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80",
"name": "runpod-python E2E Test - Serverless Core"
"name": "runpod-python E2E Test - Serverless Core - Basic",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"env": [
Expand All @@ -76,5 +76,73 @@
"input": {
"mock_return": "this worked!"
}
},
{
"hardwareConfig": {
"endpointConfig": {
"name": "runpod-python E2E Test - Serverless Core - Long Job",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"env": [
{
"key": "RUNPOD_USE_CORE",
"value": "true"
}
]
}
},
"input": {
"mock_return": "Delay test successful returned after waiting 5 minutes.",
"mock_delay": 300
}
},
{
"hardwareConfig": {
"endpointConfig": {
"name": "runpod-python E2E Test - Serverless Core - Generator Handler",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"dockerArgs": "python3 -u /handler.py --generator --return_aggregate_stream",
"env": [
{
"key": "RUNPOD_USE_CORE",
"value": "true"
}
]
}
},
"input": {
"mock_return": [
"value1",
"value2",
"value3"
]
}
},
{
"hardwareConfig": {
"endpointConfig": {
"name": "runpod-python E2E Test - Serverless Core - Async Generator Handler",
"gpuIds": "ADA_24,AMPERE_16,AMPERE_24,AMPERE_48,AMPERE_80"
},
"templateConfig": {
"dockerArgs": "python3 -u /handler.py --async_generator --return_aggregate_stream",
"env": [
{
"key": "RUNPOD_USE_CORE",
"value": "true"
}
]
}
},
"input": {
"mock_return": [
"value1",
"value2",
"value3"
]
}
}
]
8 changes: 8 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,13 @@
# Change Log

## Release 1.6.0 (1/29/24)

### Fixed

- Rust Serverless Core Passing all tests.
- GitHub Action and Python package updates
- Changelog date typo

## Release 1.5.3 (1/25/24)

### Added
Expand Down
65 changes: 45 additions & 20 deletions runpod/serverless/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,9 @@
from ctypes import CDLL, byref, c_char_p, c_int
from typing import Any, Callable, List, Dict, Optional

from runpod.version import __version__ as runpod_version
from runpod.serverless.modules.rp_logger import RunPodLogger

from runpod.serverless.modules import rp_job

log = RunPodLogger()

Expand Down Expand Up @@ -44,6 +45,7 @@ class Hook: # pylint: disable=too-many-instance-attributes

def __new__(cls):
if Hook._instance is None:
log.debug("SLS Core | Initializing Hook.")
Hook._instance = object.__new__(cls)
Hook._initialized = False
return Hook._instance
Expand Down Expand Up @@ -136,7 +138,7 @@ def progress_update(self, job_id: str, json_data: bytes) -> bool:
c_char_p(json_data), c_int(len(json_data))
))

def stream_output(self, job_id: str, job_output: bytes) -> bool:
async def stream_output(self, job_id: str, job_output: bytes) -> bool:
"""
send part of a streaming result to AI-API.
"""
Expand Down Expand Up @@ -170,48 +172,70 @@ def finish_stream(self, job_id: str) -> bool:


# -------------------------------- Process Job ------------------------------- #
async def _process_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]:
async def _process_job(config: Dict[str, Any], job: Dict[str, Any], hook) -> Dict[str, Any]:
""" Process a single job. """
hook = Hook()
handler = config['handler']

result = {}
try:
result = handler(job)
except Exception as err:
raise RuntimeError(
f"run {job['id']}: user code raised an {type(err).__name__}") from err
if inspect.isgeneratorfunction(handler) or inspect.isasyncgenfunction(handler):
log.debug("SLS Core | Running job as a generator.")
generator_output = rp_job.run_job_generator(handler, job)
aggregated_output = {'output': []}

async for part in generator_output:
log.debug(f"SLS Core | Streaming output: {part}", job['id'])

if 'error' in part:
aggregated_output = part
break
if config.get('return_aggregate_stream', False):
aggregated_output['output'].append(part['output'])

await hook.stream_output(job['id'], part)

if inspect.isgeneratorfunction(handler):
for part in result:
hook.stream_output(job['id'], part)
log.debug("SLS Core | Finished streaming output.", job['id'])
hook.finish_stream(job['id'])
result = aggregated_output

hook.finish_stream(job['id'])
else:
log.debug("SLS Core | Running job as a standard function.")
result = await rp_job.run_job(handler, job)
result = result.get('output', result)

except Exception as err: # pylint: disable=broad-except
log.error(f"SLS Core | Error running job: {err}", job['id'])
result = {'error': str(err)}

else:
finally:
log.debug(f"SLS Core | Posting output: {result}", job['id'])
hook.post_output(job['id'], result)


# -------------------------------- Run Worker -------------------------------- #
# ---------------------------------------------------------------------------- #
# Run Worker #
# ---------------------------------------------------------------------------- #
async def run(config: Dict[str, Any]) -> None:
""" Run the worker.

Args:
config: A dictionary containing the following keys:
handler: A function that takes a job and returns a result.
"""
handler = config['handler']
max_concurrency = config.get('max_concurrency', 4)
max_jobs = config.get('max_jobs', 4)
max_concurrency = config.get('max_concurrency', 1)
max_jobs = config.get('max_jobs', 1)

hook = Hook()
serverless_hook = Hook()

while True:
jobs = hook.get_jobs(max_concurrency, max_jobs)
jobs = serverless_hook.get_jobs(max_concurrency, max_jobs)

if len(jobs) == 0 or jobs is None:
await asyncio.sleep(0)
continue

for job in jobs:
asyncio.create_task(_process_job(handler, job), name=job['id'])
asyncio.create_task(_process_job(config, job, serverless_hook), name=job['id'])
await asyncio.sleep(0)

await asyncio.sleep(0)
Expand All @@ -220,6 +244,7 @@ async def run(config: Dict[str, Any]) -> None:
def main(config: Dict[str, Any]) -> None:
"""Run the worker in an asyncio event loop."""
if config.get('handler') is None:
log.error("SLS Core | config must contain a handler function")
raise ValueError("config must contain a handler function")

try:
Expand Down
Binary file modified runpod/serverless/sls_core.so
View file
Open in desktop
Binary file not shown.