diff --git a/.gitea/workflows/build-and-push.yml b/.gitea/workflows/build-and-push.yml new file mode 100644 index 0000000..09e0e40 --- /dev/null +++ b/.gitea/workflows/build-and-push.yml @@ -0,0 +1,54 @@ +name: Build and Deploy Docker Image + +# Controls when the workflow will run. Here, it runs on every push to the 'main' branch. +on: + push: + branches: ["main"] + +# Environment variables used across the workflow +env: + # The URL of your Gitea instance (without http:// or https://) + GITEA_INSTANCE_URL: git.deowl.ru + # The full name of your image (e.g., 'myusername/myproject') + IMAGE_NAME: vkrb/client + +jobs: + build-and-push: + # Runs the job on a runner with the 'ubuntu-latest' label. + runs-on: ubuntu-latest + # Optional but recommended: specifies the container image to use for the job. + # This ensures a consistent environment with Docker tools pre-installed. + container: + image: catthehacker/ubuntu:act-latest + + steps: + # 1. Check out your repository code so the workflow can access it. + - name: Checkout code + uses: actions/checkout@v4 + + # 2. Set up Docker Buildx, which is needed for building images. + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + # 3. Log in to your Gitea instance's Container Registry. + # It uses secrets you must define in your repository settings. + - name: Log in to Gitea Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.GITEA_INSTANCE_URL }} + username: ${{ gitea.repository_owner }} + # Use a secret for the password/token. See Step 3 for setup. + password: ${{ secrets.REGISTRY_TOKEN }} + + # 4. Build the Docker image from your 'dockerfile_build' directory + # and push it to the Gitea registry. + - name: Build and push Docker image + uses: docker/build-push-action@v6 + with: + # The path to the directory containing your Dockerfile + context: ./Dockerfile_build + push: true + # Tag the image with the Gitea instance, image name, and the git commit SHA. + tags: | + ${{ env.GITEA_INSTANCE_URL }}/${{ env.IMAGE_NAME }}:latest + ${{ env.GITEA_INSTANCE_URL }}/${{ env.IMAGE_NAME }}:${{ gitea.sha }} diff --git a/Dockerfile_build/source/connections/quantum_backend.py b/Dockerfile_build/source/connections/quantum_backend.py index 4dc10d5..bae2dae 100644 --- a/Dockerfile_build/source/connections/quantum_backend.py +++ b/Dockerfile_build/source/connections/quantum_backend.py @@ -5,7 +5,7 @@ import requests # Global var from env QUANTUM_BACKEND_URL = os.getenv( - "QUANTUM_BACKEND_URL", os.environ["QUNATUM_BACKEND_URL"] + "QUANTUM_BACKEND_URL", os.environ["QUANTUM_BACKEND_URL"] ) @@ -25,19 +25,29 @@ def get_or_create_device_by_name( Raises: requests.RequestException: If the request fails """ - url = f"{QUANTUM_BACKEND_URL}/machine/" + + url = f"{QUANTUM_BACKEND_URL}/machine" payload = {"system_name": system_name, "max_qubits": max_qubits} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}", + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Referer": url, + "Host": "quantum-backend.deowl.ru", } try: response = requests.post(url, json=payload, headers=headers) + print(response.content) + print(response.headers) + print(headers) response.raise_for_status() return response.json() - except requests.RequestException as e: + except Exception as e: + print(f"Failed to get or create device: {e}") raise Exception(f"Failed to get or create device: {e}") from e @@ -60,6 +70,11 @@ def get_device_by_id(system_id: int, access_token) -> Dict[str, Any]: headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}", + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Referer": url, + "Host": "quantum-backend.deowl.ru", } try: @@ -103,6 +118,11 @@ def update_device_data( headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}", + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Referer": url, + "Host": "quantum-backend.deowl.ru", } try: diff --git a/Dockerfile_build/source/connections/rabbitmq.py b/Dockerfile_build/source/connections/rabbitmq.py index 5a1133d..793292b 100644 --- a/Dockerfile_build/source/connections/rabbitmq.py +++ b/Dockerfile_build/source/connections/rabbitmq.py @@ -56,6 +56,7 @@ class RabbitMQManager: await self._consumer_channel.set_qos(prefetch_count=1, global_=True) print("Consumer channel created") return self._consumer_channel + return self._consumer_channel return None async def get_publisher_channel(self): diff --git a/Dockerfile_build/source/main.py b/Dockerfile_build/source/main.py index 2b7eb5c..49dfca5 100644 --- a/Dockerfile_build/source/main.py +++ b/Dockerfile_build/source/main.py @@ -14,6 +14,7 @@ async def start_heartbeat(): while True: try: await publish_heartbeat() + except Exception as e: print(f"Heartbeat failed: {e}") await asyncio.sleep(5) @@ -24,6 +25,7 @@ async def start_consumer(): while True: try: await consume_messages_topic() + await asyncio.Future() except Exception as e: print(f"Consumer failed: {e}") await asyncio.sleep(5) diff --git a/Dockerfile_build/source/modules/fasthtml.py b/Dockerfile_build/source/modules/fasthtml.py index cef4d5d..f9807f8 100644 --- a/Dockerfile_build/source/modules/fasthtml.py +++ b/Dockerfile_build/source/modules/fasthtml.py @@ -1,3 +1,4 @@ +import asyncio import logging import os from datetime import datetime, timedelta @@ -40,6 +41,7 @@ from fasthtml.common import ( from fasthtml.pico import Card, Container, picolink from fasthtml.xtend import Style from keycloak.pkce_utils import generate_code_challenge, generate_code_verifier +from modules.rabbitmq import consume_messages_topic logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -363,7 +365,7 @@ async def dashboard(): Div( id="client-status", hx_get="/client-status", - hx_trigger="load, every 30s", + hx_trigger="load, every 15s", ), ), Div(id="memory-stats", hx_get="/memory-stats", hx_trigger="load, every 5s"), @@ -382,10 +384,10 @@ async def dashboard(): @app.route("/client-status", methods="get") async def status(): """Получить статус клиента из центрального микросервиса""" - token_data = load_token() + access_token = get_valid_access_token() client_info = load_client_info() - if not token_data or not client_info: + if not access_token or not client_info: return P("Клиент не зарегистрирован", style="color: orange;") if rabbitmq_manager._connection: @@ -395,6 +397,8 @@ async def status(): ) # get_client_status(client_name, access_token) if status: + await consume_messages_topic() + return Div( P( Strong("Статус клиента: "), @@ -413,6 +417,17 @@ async def status(): ), style="padding: 10px; border-radius: 5px; margin: 10px 0;", ) + else: + return Div( + P( + Strong("Статус клиента: "), + Span("Не подключен", style="color: orange;"), + ), + P( + "Не удалось связаться с центральным сервисом или данному клиенту отказано в подключении" + ), + style="padding: 10px; border-radius: 5px; margin: 10px 0;", + ) @app.route("/memory-stats", methods="get") diff --git a/Dockerfile_build/source/modules/rabbitmq.py b/Dockerfile_build/source/modules/rabbitmq.py index 1f702f2..5e3ae44 100644 --- a/Dockerfile_build/source/modules/rabbitmq.py +++ b/Dockerfile_build/source/modules/rabbitmq.py @@ -16,7 +16,7 @@ from connections.quantum_backend import ( ) from connections.rabbitmq import rabbitmq_manager from fastcore.xtras import datetime -from modules.vqe import prepare_data, run_vqe +from modules.vqe import prepare_data, run_computation from pennylane.devices import Device from uvicorn.main import logger @@ -155,7 +155,7 @@ def run_vqe_process(conn: Connection, dev: Device, data: dict): """Run VQE in a separate process - process exits when this function returns.""" try: # Run VQE - run_vqe(conn=conn, dev1=dev, data=data) + run_computation(conn=conn, dev1=dev, data=data) # Send sentinel to indicate completion conn.send(None) # Signal completion @@ -179,6 +179,10 @@ def run_vqe_process(conn: Connection, dev: Device, data: dict): conn.close() +# Global variable to track current subscriptions +_current_subscriptions = {} # Format: {queue_name: {"team_id": team_id, "qubits": qubits, "consumer": consumer, "queue": queue}} + + async def consume_messages_topic(): """ Subscribe to team-specific qubit queues. @@ -186,6 +190,8 @@ async def consume_messages_topic(): where N is the number of qubits this system can handle (1 to max_qubits). This enables round-robin task distribution among systems with sufficient qubits. + + Dynamically manages subscriptions: adds new queues and removes obsolete ones. """ channel = await rabbitmq_manager.get_consumer_channel() if not channel: @@ -199,36 +205,114 @@ async def consume_messages_topic(): access_token = get_valid_access_token() device_data = get_device_by_id(system_id, access_token) teams = device_data["teams"] + if not teams: print(f"System {system_id} is not part of any team. No queues to subscribe.") + # Unsubscribe from all existing queues if no teams + await _unsubscribe_all() + return - # Subscribe to qubit-specific queues for each team - subscription_count = 0 + # Build expected queues from current teams + expected_queues = set() for team in teams: for qubits in range(1, team["num_qubits"] + 1): - # Queue name format: team_{team_id}.qubits_{qubits} queue_name = f"team_{team['team']['team_id']}.qubits_{qubits}" + expected_queues.add(queue_name) - # Declare the queue (durable, shared among multiple consumers) - queue = await channel.declare_queue( - queue_name, - durable=True, - arguments={ - "x-max-priority": 100, # Allow priorities 0-10 - }, - ) + # Get current subscriptions + current_queues = set(_current_subscriptions.keys()) - # Start consuming from this queue - await queue.consume( - process_message, - arguments={"x-priority": qubits, "x-priority-max": 10}, - ) - subscription_count += 1 + # Queues to add (in expected but not current) + queues_to_add = expected_queues - current_queues - print(f"System {system_id} subscribed to queue: {queue_name}") + # Queues to remove (in current but not expected) + queues_to_remove = current_queues - expected_queues - print(f"System {system_id} subscribed to {subscription_count} queues") - await asyncio.Future() # Keep running + # Unsubscribe from queues that no longer exist + for queue_name in queues_to_remove: + subscription_info = _current_subscriptions[queue_name] + queue = subscription_info["queue"] + cons = subscription_info["consumer"] + try: + # Cancel the consumer directly on the consumer object + await queue.cancel(consumer_tag=cons) + print(f"System {system_id} unsubscribed from queue: {queue_name}") + del _current_subscriptions[queue_name] + except Exception as e: + print(f"Error unsubscribing from {queue_name}: {e}") + + # Subscribe to new queues + for queue_name in queues_to_add: + # Extract qubits from queue name + qubits = int(queue_name.split(".qubits_")[1]) + team_id = queue_name.split(".qubits_")[0].replace("team_", "") + + # Declare the queue (durable, shared among multiple consumers) + queue = await channel.declare_queue( + queue_name, + durable=True, + arguments={ + "x-max-priority": 100, # Allow priorities 0-10 + }, + ) + + # Start consuming from this queue + # In aio-pika, queue.consume returns a Consumer object + consumer = await queue.consume( + process_message, + arguments={"x-priority": qubits, "x-priority-max": 10}, + ) + + # Store subscription info with consumer object + _current_subscriptions[queue_name] = { + "team_id": team_id, + "qubits": qubits, + "consumer": consumer, + "queue": queue, + } + + print(f"System {system_id} subscribed to queue: {queue_name}") + + # Print summary + if queues_to_add or queues_to_remove: + print( + f"System {system_id} subscription update - Active: {len(_current_subscriptions)} queues, " + f"Added: {len(queues_to_add)}, Removed: {len(queues_to_remove)}" + ) + else: + print( + f"System {system_id} subscriptions unchanged - Active: {len(_current_subscriptions)} queues" + ) + + +async def _unsubscribe_all(): + """ + Helper function to unsubscribe from all queues. + """ + for queue_name, subscription_info in list(_current_subscriptions.items()): + consumer = subscription_info["consumer"] + try: + # Cancel the consumer directly on the consumer object + await consumer.cancel() + print(f"Unsubscribed from queue: {queue_name}") + del _current_subscriptions[queue_name] + except Exception as e: + print(f"Error unsubscribing from {queue_name}: {e}") + + +def get_current_subscriptions(): + """ + Helper function to get current subscriptions for debugging. + """ + # Return a serializable version without the consumer and queue objects + return { + queue_name: { + "team_id": info["team_id"], + "qubits": info["qubits"], + "consumer_tag": info["consumer"].consumer_tag if info["consumer"] else None, + } + for queue_name, info in _current_subscriptions.items() + } # --- Heartbeat Publisher Logic --- diff --git a/Dockerfile_build/source/modules/vqe.py b/Dockerfile_build/source/modules/vqe.py index 8f06f92..bb0b8ed 100644 --- a/Dockerfile_build/source/modules/vqe.py +++ b/Dockerfile_build/source/modules/vqe.py @@ -1,5 +1,6 @@ import os from multiprocessing.connection import Connection +from typing import List import jax import pennylane as qml @@ -88,10 +89,11 @@ def prepare_data(data): } -def run_vqe(conn: Connection, dev1: Device, data: dict): +def run_computation(conn: Connection, dev1: Device, data: dict): coordinates = jnp.array(data.get("coordinates")) charge = int(data.get("charge")) multiplicity = int(data.get("multiplicity")) + molecule = qchem.Molecule( data.get("symbols"), coordinates, @@ -115,7 +117,10 @@ def run_vqe(conn: Connection, dev1: Device, data: dict): singles, doubles = qml.qchem.excitations(active_electrons, qubits) - params = np.array(np.zeros(len(singles) + len(doubles)), requires_grad=True) + if False: + params = np.array(last_state, requires_grad=True) + else: + params = np.array(np.zeros(len(singles) + len(doubles)), requires_grad=True) conn.send( { diff --git a/docker-compose.yml b/docker-compose.yml index 873b899..8366806 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,9 +4,7 @@ services: image: git.deowl.ru/vkrb/client:0.1.0 container_name: quantum-client restart: unless-stopped - extra_hosts: - - "auth.localhost:host-gateway" - - "host.docker.internal:host-gateway" + ports: - ${PORT}:${PORT} volumes: