forked from luopingyi/xc_agent_platform_demo
23 lines
795 B
Python
23 lines
795 B
Python
"""Read queue hints without contacting the platform; unknown is conservative."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
def queue_depth():
|
|
value = os.environ.get('ASYNC_QUEUE_DEPTH')
|
|
if value is None and os.environ.get('QUEUE_SNAPSHOT_JSON'):
|
|
try:
|
|
data = json.loads(Path(os.environ['QUEUE_SNAPSHOT_JSON']).read_text())
|
|
if isinstance(data, dict):
|
|
value = next((data[k] for k in ('depth', 'queue_depth', 'async_queue_depth', 'ASYNC_QUEUE_DEPTH') if k in data), None)
|
|
except (OSError, ValueError):
|
|
return None
|
|
try:
|
|
if isinstance(value, bool):
|
|
return None
|
|
depth = int(str(value))
|
|
return depth if depth >= 0 else None
|
|
except (TypeError, ValueError):
|
|
return None
|