Khi production system của bạn phụ thuộc vào các model AI, việc kiểm soát phiên bản API trở nên quan trọng hơn bao giờ hết. Bài viết này chia sẻ kinh nghiệm thực chiến của đội ngũ khi triển khai CI/CD pipeline với rollback tự động và canary deployment cho các API call đến model AI.
Tại sao cần Model Version Control?
Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem một tình huống thực tế: Đội ngũ của tôi từng gặp incident nghiêm trọng khi một model provider thay đổi behavior của API mà không báo trước. Response format thay đổi, field name khác, và 3 tiếng đồng hồ debugging mới tìm ra nguyên nhân. Kể từ đó, chúng tôi xây dựng một hệ thống version control hoàn chỉnh.
Kiến trúc tổng quan
Hệ thống Model Version Control bao gồm các thành phần chính:
- Version Registry: Lưu trữ mapping giữa model version và endpoint
- Traffic Router: Điều phối request đến các phiên bản khác nhau
- Health Monitor: Theo dõi error rate, latency, và quality metrics
- Rollback Controller: Tự động hoặc thủ công quay về version cũ
- Canary Deployer: Triển khai phiên bản mới cho % nhỏ traffic
Triển khai với HolySheep AI
Chúng tôi chọn HolySheep AI vì tỷ giá chỉ ¥1=$1, tiết kiệm 85%+ chi phí so với các provider khác, hỗ trợ WeChat/Alipay thanh toán, độ trễ trung bình dưới 50ms, và quan trọng nhất là API endpoint ổn định với versioning rõ ràng.
Code Implementation
1. Version Registry với HolySheep
// model_version_registry.py
import json
import hashlib
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime
@dataclass
class ModelVersion:
version: str
model_id: str
base_url: str
is_active: bool
traffic_percentage: int
deployed_at: datetime
health_score: float
class HolySheepVersionRegistry:
"""
Registry quản lý các phiên bản model với HolySheep AI
HolySheep cung cấp endpoint ổn định: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.versions: Dict[str, ModelVersion] = {}
self._initialize_default_versions()
def _initialize_default_versions(self):
"""Khởi tạo các phiên bản mặc định"""
# DeepSeek V3.2 - Phiên bản ổn định (production)
self.versions["deepseek-v3.2-stable"] = ModelVersion(
version="deepseek-v3.2-stable",
model_id="deepseek-v3.2",
base_url=self.BASE_URL,
is_active=True,
traffic_percentage=80,
deployed_at=datetime.now(),
health_score=0.98
)
# DeepSeek V3.2 - Phiên bản beta (testing)
self.versions["deepseek-v3.2-beta"] = ModelVersion(
version="deepseek-v3.2-beta",
model_id="deepseek-v3.2",
base_url=self.BASE_URL,
is_active=False,
traffic_percentage=20,
deployed_at=datetime.now(),
health_score=0.95
)
def get_active_version(self, version_name: str) -> Optional[ModelVersion]:
"""Lấy thông tin phiên bản đang active"""
return self.versions.get(version_name)
def switch_version(self, from_version: str, to_version: str) -> bool:
"""Chuyển đổi giữa các phiên bản"""
if from_version not in self.versions or to_version not in self.versions:
return False
self.versions[from_version].is_active = False
self.versions[to_version].is_active = True
# Log sự thay đổi để audit
self._log_version_switch(from_version, to_version)
return True
def _log_version_switch(self, from_v: str, to_v: str):
"""Ghi log khi chuyển đổi phiên bản"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"from_version": from_v,
"to_version": to_v,
"api_key_hash": hashlib.sha256(self.api_key.encode()).hexdigest()[:8]
}
print(f"[VERSION SWITCH] {json.dumps(log_entry)}")
def get_health_report(self) -> Dict:
"""Lấy báo cáo sức khỏe của tất cả phiên bản"""
return {
"versions": {
name: {
"is_active": v.is_active,
"traffic_percentage": v.traffic_percentage,
"health_score": v.health_score,
"deployed_at": v.deployed_at.isoformat()
}
for name, v in self.versions.items()
},
"total_versions": len(self.versions)
}2. Canary Deployment Controller
// canary_controller.py
import random
import asyncio
from typing import Callable, Dict, Any, Optional
from datetime import datetime, timedelta
from collections import defaultdict
class CanaryDeploymentController:
"""
Canary Deployment Controller cho HolySheep AI API
Triển khai phiên bản mới cho % nhỏ traffic trước khi rollout hoàn toàn
"""
def __init__(self):
self.traffic_splits: Dict[str, float] = {
"stable": 0.0,
"canary": 0.0
}
self.metrics = defaultdict(list)
self.rollback_threshold = {
"error_rate": 0.05, # 5% error rate threshold
"latency_p99": 2000, # 2000ms latency threshold
"min_requests": 100 # Minimum requests trước khi đánh giá
}
async def route_request(
self,
request_id: str,
payload: Dict[str, Any],
api_key: str
) -> Dict[str, Any]:
"""
Điều phối request đến phiên bản phù hợp
"""
# Xác định endpoint dựa trên traffic split
target_version = self._determine_version(request_id)
start_time = datetime.now()
try:
# Gọi HolySheep API
result = await self._call_holysheep(
target_version=target_version,
payload=payload,
api_key=api_key
)
# Ghi metrics
latency = (datetime.now() - start_time).total_seconds() * 1000
self._record_metric(target_version, "success", latency)
return {
"success": True,
"result": result,
"version": target_version,
"latency_ms": round(latency, 2)
}
except Exception as e:
latency = (datetime.now() - start_time).total_seconds() * 1000
self._record_metric(target_version, "error", latency)
return {
"success": False,
"error": str(e),
"version": target_version,
"latency_ms": round(latency, 2)
}
def _determine_version(self, request_id: str) -> str:
"""Xác định phiên bản nào nhận request"""
# Hash request_id để đảm bảo consistency
hash_value = hash(request_id) % 100
if hash_value < self.traffic_splits["canary"] * 100:
return "canary"
return "stable"
async def _call_holysheep(
self,
target_version: str,
payload: Dict[str, Any],
api_key: str
) -> Dict[str, Any]:
"""Gọi HolySheep AI API"""
import aiohttp
url = f"https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
raise Exception(f"API Error: {response.status}")
return await response.json()
def _record_metric(self, version: str, status: str, latency: float):
"""Ghi nhận metrics cho phiên bản"""
self.metrics[version].append({
"timestamp": datetime.now(),
"status": status,
"latency": latency
})
# Cleanup metrics cũ (giữ 1 giờ gần nhất)
cutoff = datetime.now() - timedelta(hours=1)
self.metrics[version] = [
m for m in self.metrics[version] if m["timestamp"] > cutoff
]
def set_traffic_split(self, stable_pct: float, canary_pct: float):
"""Thiết lập tỷ lệ traffic giữa các phiên bản"""
if stable_pct + canary_pct != 1.0:
raise ValueError("Tổng percentage phải bằng 1.0")
self.traffic_splits["stable"] = stable_pct
self.traffic_splits["canary"] = canary_pct
def should_rollback(self, version: str = "canary") -> tuple[bool, str]:
"""
Kiểm tra xem có nên rollback canary version không
Returns: (should_rollback, reason)
"""
version_metrics = self.metrics.get(version, [])
if len(version_metrics) < self.rollback_threshold["min_requests"]:
return False, "Chưa đủ request để đánh giá"
# Tính error rate
errors = sum(1 for m in version_metrics if m["status"] == "error")
error_rate = errors / len(version_metrics)
# Tính latency P99
latencies = sorted([m["latency"] for m in version_metrics])
p99_index = int(len(latencies) * 0.99)
latency_p99 = latencies[p99_index] if latencies else 0
# Kiểm tra các threshold
if error_rate > self.rollback_threshold["error_rate"]:
return True, f"Error rate cao: {error_rate:.2%} (threshold: {self.rollback_threshold['error_rate']:.2%})"
if latency_p99 > self.rollback_threshold["latency_p99"]:
return True, f"Latency P99 cao: {latency_p99:.0f}ms (threshold: {self.rollback_threshold['latency_p99']}ms)"
return False, "Tất cả metrics trong ngưỡng cho phép"
def get_metrics_summary(self) -> Dict[str, Any]:
"""Lấy tóm tắt metrics của các phiên bản"""
summary = {}
for version, metrics_list in self.metrics.items():
if not metrics_list:
continue
errors = sum(1 for m in metrics_list if m["status"] == "error")
latencies = [m["latency"] for m in metrics_list]
summary[version] = {
"total_requests": len(metrics_list),
"error_rate": errors / len(metrics_list),
"latency_avg": sum(latencies) / len(latencies),
"latency_p50": sorted(latencies)[len(latencies) // 2],
"latency_p99": sorted(latencies)[int(len(latencies) * 0.99)]
}
return summary3. Automatic Rollback System
// auto_rollback.py
import asyncio
import logging
from datetime import datetime, timedelta
from typing import Callable, Optional, List
from enum import Enum
class RollbackTrigger(Enum):
ERROR_RATE = "error_rate"
LATENCY = "latency"
MANUAL = "manual"
SCHEDULED = "scheduled"
class AutoRollbackSystem:
"""
Hệ thống Rollback tự động cho HolySheep AI API
Giám sát health metrics và tự động rollback khi phát hiện vấn đề
"""
def __init__(
self,
canary_controller,
notification_callback: Optional[Callable] = None
):
self.canary = canary_controller
self.notification = notification_callback
self.is_active = True
self.rollback_history: List[Dict] = []
# Cấu hình monitoring
self.check_interval = 30 # seconds
self.consecutive_failures_trigger = 3
# Metrics theo dõi
self.consecutive_failures = 0
self.last_healthy_check = datetime.now()
self.logger = logging.getLogger(__name__)
async def start_monitoring(self):
"""Bắt đầu giám sát và rollback tự động"""
self.logger.info("Bắt đầu Auto Rollback System")
while self.is_active:
try:
await self._check_health()
await asyncio.sleep(self.check_interval)
except Exception as e:
self.logger.error(f"Lỗi trong monitoring loop: {e}")
await asyncio.sleep(self.check_interval)
def stop_monitoring(self):
"""Dừng giám sát"""
self.is_active = False
self.logger.info("Dừng Auto Rollback System")
async def _check_health(self):
"""Kiểm tra sức khỏe của canary version"""
should_rollback, reason = self.canary.should_rollback("canary")
if should_rollback:
self.consecutive_failures += 1
self.logger.warning(
f"Phát hiện vấn đề với canary: {reason} "
f"(lần {self.consecutive_failures}/{self.consecutive_failures_trigger})"
)
if self.consecutive_failures >= self.consecutive_failures_trigger:
await self._execute_rollback(reason)
else:
self.consecutive_failures = 0
self.last_healthy_check = datetime.now()
self.logger.debug("Canary version khỏe mạnh")
async def _execute_rollback(self, reason: str):
"""Thực hiện rollback về phiên bản stable"""
self.logger.info(f"Bắt đầu rollback: {reason}")
# Lưu trạng thái trước rollback
pre_rollback_state = {
"traffic_splits": self.canary.traffic_splits.copy(),
"timestamp": datetime.now().isoformat(),
"reason": reason
}
try:
# Chuyển 100% traffic về stable
self.canary.set_traffic_split(stable_pct=1.0, canary_pct=0.0)
# Ghi nhận rollback thành công
rollback_record = {
**pre_rollback_state,
"status": "success",
"completed_at": datetime.now().isoformat()
}
self.rollback_history.append(rollback_record)
# Gửi thông báo
if self.notification:
await self.notification(
title="🚨 Auto Rollback Executed",
message=f"Canary đã rollback về stable.\nLý do: {reason}",
severity="critical"
)
self.logger.warning(f"Rollback thành công: {reason}")
except Exception as e:
self.logger.error(f"Rollback thất bại: {e}")
if self.notification:
await self.notification(
title="🚨 Rollback Failed",
message=f"Không thể rollback: {str(e)}",
severity="critical"
)
async def manual_rollback(self, reason: str = "Manual trigger"):
"""Trigger rollback thủ công"""
self.logger.info(f"Manual rollback triggered: {reason}")
await self._execute_rollback(reason)
def promote_canary_to_stable(self):
"""Đưa canary lên thành stable (promote sau khi canary ổn định)"""
self.logger.info("Promote canary lên stable")
# Clone canary config sang stable
# (Implementation tùy thuộc vào cách bạn lưu trữ config)
self.canary.set_traffic_split(stable_pct=1.0, canary_pct=0.0)
promote_record = {
"timestamp": datetime.now().isoformat(),
"action": "promote",
"from": "canary",
"to": "stable",
"status": "success"
}
self.rollback_history.append(promote_record)
def get_rollback_history(self, limit: int = 10) -> List[Dict]:
"""Lấy lịch sử rollback"""
return self.rollback_history[-limit:]
=== Example Usage với HolySheep AI ===
async def slack_notification(title: str, message: str, severity: str):
"""Example notification callback"""
print(f"[{severity.upper()}] {title}: {message}")
# Gửi Slack/Discord/PagerDuty notification ở đây
async def example_usage():
"""Ví dụ sử dụng Auto Rollback System"""
# Khởi tạo Canary Controller
canary = CanaryDeploymentController()
canary.set_traffic_split(stable_pct=0.9, canary_pct=0.1)
# Khởi tạo Auto Rollback System
rollback_system = AutoRollbackSystem(
canary_controller=canary,
notification_callback=slack_notification
)
# Bắt đầu monitoring trong background
monitor_task = asyncio.create_task(rollback_system.start_monitoring())
# Sau một thời gian, promote canary nếu ổn định
await asyncio.sleep(3600) # 1 giờ
rollback_system.promote_canary_to_stable()
# Hoặc manual rollback nếu cần
# await rollback_system.manual_rollback("Phát hiện output format không đúng")
# Dừng khi done
rollback_system.stop_monitoring()
await monitor_task4. Client SDK với Built-in Rollback
// holysheep_client.py
import time
import random
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime
@dataclass
class APIResponse:
success: bool
data: Optional[Dict]
error: Optional[str]
latency_ms: float
version: str
provider: str = "holysheep"
class HolySheepClient:
"""
HolySheep AI Client với built-in retry và automatic rollback
base_url: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.model_version = "deepseek-v3.2"
# Fallback versions
self.fallback_versions = [
"deepseek-v3.2",
"deepseek-v3.1",
"gpt-4o-mini"
]
# Retry configuration
self.max_retries = 3
self.retry_delay = 1.0
self.backoff_factor = 2.0
# Circuit breaker
self.circuit_breaker_threshold = 5
self.circuit_breaker_timeout = 60
self.failure_count = 0
self.circuit_open_since: Optional[datetime] = None
# Version health tracking
self.version_health: Dict[str, Dict] = {}
def chat_completions(
self,
messages: List[Dict[str, str]],
model: Optional[str] = None,
**kwargs
) -> APIResponse:
"""
Gọi Chat Completions API với automatic retry và fallback
"""
start_time = time.time()
version_to_use = model or self.model_version
for attempt in range(self.max_retries):
try:
# Kiểm tra circuit breaker
if self._is_circuit_open():
# Thử version khác
version_to_use = self._get_next_healthy_version()
if not version_to_use:
return APIResponse(
success=False,
data=None,
error="Circuit breaker open - all versions unavailable",
latency_ms=(time.time() - start_time) * 1000,
version="none"
)
# Thực hiện request
response = self._make_request(
version=version_to_use,
messages=messages,
**kwargs
)
# Success - ghi nhận
self._record_success(version_to_use)
return APIResponse(
success=True,
data=response,
error=None,
latency_ms=(time.time() - start_time) * 1000,
version=version_to_use
)
except Exception as e:
self._record_failure(version_to_use)
error_msg = str(e)
# Retry với exponential backoff
if attempt < self.max_retries - 1:
delay = self.retry_delay * (self.backoff_factor ** attempt)
time.sleep(delay)
# Thử version khác nếu version hiện tại có vấn đề
next_version = self._get_next_healthy_version()
if next_version and next_version != version_to_use:
version_to_use = next_version
# Tất cả retries thất bại
return APIResponse(
success=False,
data=None,
error=f"All retries failed. Last error: {error_msg}",
latency_ms=(time.time() - start_time) * 1000,
version=version_to_use
)
def _make_request(
self,
version: str,
messages: List[Dict[str, str]],
**kwargs
) -> Dict[str, Any]:
"""Thực hiện HTTP request đến HolySheep API"""
import requests
url = f"{self.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": version,
"messages": messages,
**kwargs
}
response = requests.post(url, json=payload, headers=headers, timeout=30)
if response.status_code == 429:
raise Exception("Rate limit exceeded")
elif response.status_code >= 500:
raise Exception(f"Server error: {response.status_code}")
elif response.status_code != 200:
raise Exception(f"API error: {response.status_code} - {response.text}")
return response.json()
def _is_circuit_open(self) -> bool:
"""Kiểm tra circuit breaker có open không"""
if self.failure_count < self.circuit_breaker_threshold:
return False
if self.circuit_open_since is None:
self.circuit_open_since = datetime.now()
return True
# Tự động thử lại sau timeout
elapsed = (datetime.now() - self.circuit_open_since).total_seconds()
if elapsed >= self.circuit_breaker_timeout:
self.circuit_open_since = None
return False
return True
def _record_success(self, version: str):
"""Ghi nhận thành công"""
self.failure_count = 0
self.circuit_open_since = None
if version not in self.version_health:
self.version_health[version] = {"success": 0, "failure": 0}
self.version_health[version]["success"] += 1
def _record_failure(self, version: str):
"""Ghi nhận thất bại"""
self.failure_count += 1
if version not in self.version_health:
self.version_health[version] = {"success": 0, "failure": 0}
self.version_health[version]["failure"] += 1
def _get_next_healthy_version(self) -> Optional[str]:
"""Lấy version khả dụng tiếp theo"""
for version in self.fallback_versions:
health = self.version_health.get(version, {"success": 0, "failure": 0})
total = health["success"] + health["failure"]
if total == 0:
continue
error_rate = health["failure"] / total
if error_rate < 0.1: # Chỉ dùng version có error rate < 10%
return version
return None
def get_health_status(self) -> Dict[str, Any]:
"""Lấy trạng thái sức khỏe của các versions"""
return {
"circuit_breaker": {
"is_open": self._is_circuit_open(),
"failure_count": self.failure_count,
"threshold": self.circuit_breaker_threshold
},
"versions": {
version: {
"success": health["success"],
"failure": health["failure"],
"error_rate": health["failure"] / max(1, health["success"] + health["failure"])
}
for version, health in self.version_health.items()
}
}
=== Usage Example ===
def main():
# Khởi tạo client với API key từ HolySheep
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Gọi API - automatic retry và fallback
response = client.chat_completions(
messages=[
{"role": "system", "content": "Bạn là trợ lý AI"},
{"role": "user", "content": "Giải thích về model versioning"}
],
temperature=0.7,
max_tokens=500
)
if response.success:
print(f"✅ Response từ {response.version} (latency: {response.latency_ms}ms)")
print(response.data)
else:
print(f"❌ Lỗi: {response.error}")
# Kiểm tra health status
print("\n📊 Health Status:")
print(client.get_health_status())
if __name__ == "__main__":
main()Bảng giá và ROI khi sử dụng HolySheep
So sánh chi phí giữa các provider cho 1 triệu tokens:
- GPT-4.1: $8.00 (provider khác) → Tương đương HolySheep với tỷ giá ¥1=$1
- Claude Sonnet 4.5: $15.00 (provider khác) → Tiết kiệm đáng kể với HolySheep
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42 — Model có chi phí thấp nhất, phù hợp cho batch processing
Với độ trễ trung bình dưới 50ms và 99.9% uptime, HolySheep là lựa chọn tối ưu cho production workloads.
Lỗi thường gặp và cách khắc phục
1. Lỗi "Circuit Breaker Open" - Tất cả versions unavailable
# Triệu chứng:
APIResponse(success=False, error="Circuit breaker open - all versions unavailable")
Nguyên nhân:
- Failure count vượt threshold do API provider gặp sự cố
- Tất cả fallback versions đều có error rate > 10%
Cách khắc phục:
1. Kiểm tra health status
health = client.get_health_status()
print(f"Circuit breaker: {health['circuit_breaker']}")
print(f"Versions: {health['versions']}")
2. Reset circuit breaker nếu cần (thủ công)
client.failure_count = 0
client.circuit_open_since = None
3. Chờ timeout tự động (60 giây mặc định)
4. Thêm version mới vào fallback list
client.fallback_versions.append("gpt-4o-mini")
5. Implement exponential backoff với jitter
import random
def retry_with_jitter(attempt, base_delay=1.0, max_delay=30.0):
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.1)
return delay + jitter
Sử dụng trong retry loop
for attempt in range(max_retries):
try:
response = client._make_request(...)
break
except Exception as e:
if attempt < max_retries - 1:
delay = retry_with_jitter(attempt)
time.sleep(delay)2. Lỗi "Latency cao" trong Canary Deployment
# Triệu chứng:
should_rollback() trả về True với "Latency P99 cao: >2000ms"
Nguyên nhân:
- Canary version có độ trễ cao hơn stable version
- Network congestion hoặc provider-side issues
- Request payload quá lớn
Cách khắc phục:
1. Tăng latency threshold tạm thời
canary.rollback_threshold["latency_p99"] = 5000 # 5 giây
2. Hoặc loại bỏ latency check tạm thời
canary.rollback_threshold["latency_p99"] = float('inf')
3. Optimize request payload
def optimize_payload(messages, max_history=10):
"""Giới hạn số lượng messages để giảm latency"""
if len(messages) <= max_history:
return messages
# Giữ system prompt và messages gần nhất
system_msg = [m for m in messages if m["role"] == "system"]
other_msgs = [m for m in messages if m["role"] != "system"]
return system_msg + other_msgs[-max_history:]
4. Sử dụng streaming cho response lớn
def chat_completions_streaming(client, messages):
"""Sử dụng streaming thay vì non-streaming"""
import requests
url = f"{client.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {client.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": client.model_version,
"messages": messages,
"stream": True
}
response = requests.post(url, json=payload, headers=headers, stream=True)
return response.iter_lines()
5. Implement adaptive timeout
def adaptive_timeout(base_timeout, attempt):
"""Tăng timeout cho từng attempt"""
timeouts = [30, 60, 120, 180] # seconds
return timeouts[min(attempt, len(timeouts) - 1)]3. Lỗi "Version Switch Failed" - Rollback không hoạt động
# Triệu chứng:
registry.switch_version() trả về False
Hoặc traffic vẫn đi vào version cũ sau khi switch
Nguyên nhân:
- Version name không tồn tại trong registry
- Race condition khi nhiều process cùng switch
- Cache không được clear sau khi switch
Cách khắc phục:
1. Kiểm tra available versions
registry = HolySheepVersionRegistry("YOUR_API_KEY")
print(f"Available versions: {list(registry.versions.keys())}")
2. Verify version tồn tại trước khi switch
def safe_switch(registry, from_v, to_v):
if from_v not in registry.versions:
raise ValueError(f"Version {from_v} không tồn tại")
if to_v not in registry.versions:
raise ValueError(f"Version {to_v} không tồn tại")
success = registry.switch_version(from_v, to_v)
if success:
# Force clear any local caches
clear_version_cache()
return success
3. Sử dụng distributed lock cho multi-process
import redis
import threading
class DistributedVersionSwitch:
def __init__(self, redis_url="redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.lock_key = "version_switch_lock