Async Signals

Overview

Async signals allow processing time-consuming operations in background threads without blocking the main thread, maintaining smooth game experience. GDSignalBus provides a complete async signal processing mechanism.

Basic Usage

Sending Async Signals

# Get SignalBus singleton
var bus = SignalBus.get_singleton()

# Basic async signal
bus.emit_async("save_game", [player_data])

# Async signal with callbacks
bus.subscribe_async_with_callbacks("data_processing",
    func(args):
        var data = args[0]
        print("[Async Thread] Starting data processing: ", data)
        OS.delay_msec(2000)  # Simulate time-consuming operation
        return "Processing result: " + str(data * 2),
    func(result):
        print("[Main Thread] Processing complete, result: ", result),
    func(error):
        print("[Main Thread] Processing failed: ", error)
)

# Emit async signal
bus.emit_async("data_processing", [42])

Async Callbacks

func _on_save_complete(success, result):
    if success:
        print("Save successful: ", result)
    else:
        print("Save failed: ", result)

func _on_load_complete(success, data):
    if success:
        apply_loaded_data(data)
    else:
        show_error_message("Load failed")

Async Task Types

File Operations

# Async file reading
bus.subscribe_async_with_callbacks("file_read",
    func(args):
        var path = args[0]
        var file = FileAccess.open(path, FileAccess.READ)
        if file:
            var content = file.get_as_text()
            file.close()
            return content
        return null,
    _on_file_read_complete,
    _on_file_read_error
)

# Async file writing
bus.subscribe_async_with_callbacks("file_write",
    func(args):
        var path = args[0]
        var content = args[1]
        var file = FileAccess.open(path, FileAccess.WRITE)
        if file:
            file.store_string(content)
            file.close()
            return true
        return false,
    _on_file_write_complete,
    _on_file_write_error
)

Network Requests

# HTTP requests
bus.subscribe_async_with_callbacks("http_request",
    func(args):
        var url = args[0]
        var method = args[1]
        var body = args[2] if args.size() > 2 else ""
        
        var http = HTTPRequest.new()
        # Need to implement actual HTTP request logic here
        # Simplified example, returns simulated result
        OS.delay_msec(1000)  # Simulate network delay
        return "Response from " + url,
    _on_response_complete,
    _on_response_error
)

Data Processing

# Complex calculations
bus.subscribe_async_with_callbacks("calculate_path",
    func(args):
        var start = args[0]
        var end = args[1]
        
        print("[Async Thread] Starting path calculation...")
        
        # Simulate A* pathfinding algorithm
        var path = []
        var current = start
        
        while current.distance_to(end) > 1.0:
            var next = current + (end - current).normalized() * 10.0
            path.append(next)
            current = next
            OS.delay_msec(50)  # Simulate calculation time
        
        path.append(end)
        print("[Async Thread] Path calculation complete")
        return path,
    _on_path_calculated,
    _on_path_error
)

# Large data processing
bus.subscribe_async_with_callbacks("process_inventory",
    func(args):
        var items = args[0]
        print("[Async Thread] Starting inventory data processing...")
        
        var processed_items = []
        for item in items:
            # Simulate complex calculation
            OS.delay_msec(100)
            
            var processed_item = {
                "id": item.id,
                "name": item.name,
                "value": item.value * 1.5
            }
            processed_items.append(processed_item)
        
        print("[Async Thread] Inventory processing complete")
        return processed_items,
    _on_inventory_processed,
    _on_inventory_error
)

Advanced Features

Task Management

# Get active async task count
var active_tasks = bus.get_active_async_task_count()
print("Active async tasks: ", active_tasks)

# Process completed async tasks (must be called in _process)
func _process(_delta):
    bus.process_async_tasks()

# Wait for all async tasks to complete
bus.wait_all_async_tasks()

# Cancel specific async task
var task_id = bus.emit_async("long_operation", [data])
bus.cancel_async_task(task_id)

Task Status Monitoring

func _ready():
    # Set up timer to monitor async task status
    var timer = Timer.new()
    timer.wait_time = 1.0
    timer.timeout.connect(_monitor_async_tasks)
    timer.autostart = true
    add_child(timer)

func _monitor_async_tasks():
    var active_count = bus.get_active_async_task_count()
    if active_count > 5:
        print("Warning: Too many active async tasks: ", active_count)

Error Handling

Timeout Handling

# Use async subscription with error callbacks
bus.subscribe_async_with_callbacks("network_request",
    func(args):
        var url = args[0]
        # Simulate operation that might fail
        OS.delay_msec(2000)
        if randf() < 0.3:  # 30% failure rate
            return null  # Return null to indicate failure
        return "Success response",
    func(result):
        print("Request successful: ", result),
    func(error):
        print("Request failed: ", error)
)

Retry Mechanism

# Implement retry logic
func retry_async_operation(signal_name, args, max_retries=3):
    var retry_count = 0
    
    func attempt_operation():
        bus.subscribe_async_with_callbacks(signal_name + "_retry",
            func(op_args):
                # Execute actual operation
                return perform_operation(op_args[0]),
            func(result):
                print("Operation successful")
                cleanup_retry_subscription(signal_name + "_retry"),
            func(error):
                retry_count += 1
                if retry_count < max_retries:
                    print("Retry ", retry_count, "/", max_retries)
                    attempt_operation()
                else:
                    print("Retries exhausted, operation failed")
                    cleanup_retry_subscription(signal_name + "_retry")
        )
        
        bus.emit(signal_name + "_retry", [args])
    
    attempt_operation()

Performance Optimization

Batch Processing

# Combine multiple small tasks into one batch task
var items_to_process = []

func add_item_to_queue(item):
    items_to_process.append(item)
    
    # Process when batch size reached or on timer
    if items_to_process.size() >= 10:
        process_batch()

func process_batch():
    if items_to_process.size() > 0:
        bus.emit_async("process_inventory_batch", [items_to_process.duplicate()])
        items_to_process.clear()

Resource Management

func _exit_tree():
    # Wait for all async tasks to complete
    bus.wait_all_async_tasks()
    
    # Or cancel specific tasks
    # bus.cancel_async_task(task_id)

Best Practices

1. Reasonable Use of Async

# Suitable for async operations
- File I/O
- Network requests
- Complex calculations
- Large data processing

# Not suitable for async operations
- Simple state updates
- UI related operations
- Operations requiring immediate response

2. Error Handling

func _on_async_complete(success, result):
    if success:
        handle_success(result)
    else:
        handle_error(result)
        # Consider retry or fallback strategy

3. Resource Management

func _exit_tree():
    # Cancel all async tasks for this object
    bus.unsubscribe_all(self)

Debugging and Monitoring

Async Task Monitoring

func _ready():
    # Enable debug mode
    bus.set_debug_enabled(true)

func _process(_delta):
    # Process async tasks
    bus.process_async_tasks()
    
    # Periodically check task status
    if Engine.get_frames_drawn() % 60 == 0:  # Check once per second
        var active_tasks = bus.get_active_async_task_count()
        if active_tasks > 10:
            print("Warning: Too many active async tasks: ", active_tasks)

Example: Complete Async Save System

# SaveManager.gd
extends Node

var bus

func _ready():
    bus = SignalBus.get_singleton()
    
    # Subscribe to async save signal
    bus.subscribe_async_with_callbacks("save_game",
        func(args):
            var save_data = args[0]
            var path = args[1]
            
            print("[Async Thread] Starting game save...")
            OS.delay_msec(2000)  # Simulate save time
            
            var file = FileAccess.open(path, FileAccess.WRITE)
            if file:
                file.store_var(save_data)
                file.close()
                return true
            return false,
        _on_save_complete,
        _on_save_error
    )

func save_game_async():
    var save_data = prepare_save_data()
    var save_path = "user://savegame.dat"
    
    # Show save indicator
    show_saving_indicator()
    
    # Async save
    bus.emit_async("save_game", [save_data, save_path])

func _on_save_complete(success):
    hide_saving_indicator()
    
    if success:
        show_message("Game saved successfully!")
        update_save_timestamp()
    else:
        show_error("Save failed")

func _on_save_error(error):
    hide_saving_indicator()
    show_error("Save error: " + str(error))

func _process(_delta):
    # Must process async tasks in main thread
    bus.process_async_tasks()

func _exit_tree():
    # Cancel all unfinished save tasks
    bus.unsubscribe_all(self)

Through the async signal system, you can build responsive game applications with good user experience while maintaining code simplicity and maintainability.