Elastic Compute Service
Elastic Compute Service
Elastic Compute Service
Document
Elastic Compute Service
What's New
Announcement
OpenAPI related
Pricing and quotas are related
About products and plugins
Feature discontinuation
Overview
Billing
Billable items
Billing methods
Billing methods change
Getting Started
User Guide
Instances
Instance types
X86 Computing
Elastic Bare Metal Server
Heterogeneous Computing
Select instance billing method
Spot instance
Elastic Scheduled Instance
Offsetting Resources
Reserved Instances
Buying ECS instances
Logging in to an instance
Logging in to a Linux instance
Managing instance status
Modifying instance types
Using instance user data and metadata
Instance metadata
Images
Public images
veLinux images
Custom images
Importing custom images
Creating custom images
Image detection
End of life (EOL) of image
Elastic Block Storage
Basic volume operations
Partitioning and formatting a volume
Extending a volume
EBS Agent
Snapshots
Basic snapshot operations
Auto snapshot policies
DataInsights
Network
Network interface and IP address
IP Address
IPv4 address
Security group
Security
Passwords and key pairs
Deployment and scaling
Monitoring
Monitoring metrics
Event monitoring
Local volume event O&M
Method 1: Solve the local volume failure by replacing the faulty volume
O&M
Cloud assistant
Managing agents
Process choreography
Common operation and maintenance tasks
Batch operations on instances
Batch operations on EBS Volumes
Batch operations on images
Temporary upgrade of bandwidth
Managing application updates
orchestration plan
Patch management
Managing patch plans
Managing patch baselines
IAM
Tags and projects
Best Practices
AI applications
OpenClaw (formerly Moltbot)
Quickly deploying OpenClaw via application templates
Quickly deploying OpenClaw via marketplace images
Manually deploying OpenClaw
Self-service website creation
Building environments
Building a Web environment (Windows)
Building applications
Building FTP site
Deploying databases
Building game servers
Building websites
Building a WordPress blog
Building multiple websites
Manage software sources for Linux instances
AIGC practices
Performance test
Network performance test
Instance configuration
Setting up AD Domain Services on Windows OS
Windows Multi - User Login
Changing screen resolution
Instance data
Uploading or downloading files
Data restoration
ECS instance local time synchronization
Volume
Monitoring
Security
Images
Kernel fine-tuning
O&M
Troubleshooting
Remote log in issue
Linux operating system issue
Windows operating system issue
SSH log in issue
Operating system issue
veLinux operating system issue
Linux operating system issue
Windows operating system issue
Instance status issue
Instance unable to start issue
High instance load issue
GPU instances issue
Network connection issue
Network configuration issue
Storage classes issue
EBS Volume issue
Local Disk issue
Logs and monitoring issue
Third - party software issue
API Reference
Regions and availability zones
Instances
Images
HPC clusters
Key pairs
Tags
Volumes
Reserved storage capacity
Snapshot
Auto snapshot policy
Monitoring
Deployment sets
Launch templates
Cloud assistant
Patch management
SDK reference
CLI Reference
FAQ
Dedicated Host DDH
What's New
Billing
User Guide
Manage Dedicated Host
Manage Dedicated Host Cluster
Manage ECS instances
Monitoring and O&M
API reference
proprietary host
- Documentation
Elastic Compute ServiceElastic Compute ServiceBest PracticesSelf-service website creationAIGC practicesDeploying optimized WDL inference models on g4i instances
Copy page
Download PDFAIGC practices
Deploying optimized WDL inference models on g4i instances
Copy pageDownload PDFDeploying optimized WDL inference models on g4i instances
This topic describes how to deploy and optimize Wide & Deep Learning (WDL) models on ECS instances of the of the ecs.g4i type. By leveraging hardware acceleration, this approach enhances inference efficiency and speeds up the inference process.
Feature highlights
This practice uses the 4th-generation ECS instances of Volcano Engine, specifically the General-purpose g4i type. This instance type adopts the latest self-developed DPU architecture of Volcano Engine and is equipped with the 6th-generation Intel® Xeon® Scalable processor (Granite Rapids). This instance type also leverages next-generation hardware AI acceleration instructions: AMX-BF16 and AMX-FP16 (Intel Advanced Matrix Extensions). These features deliver system-level optimization and improve the inference efficiency of WDL models.
Background
A WDL model is a hybrid architecture that combines the Wide linear model and the Deep neural network. Primarily used for click-through rate (CTR) prediction, it is widely applied in recommendation systems, search ranking, online advertising, and numerous other fields. In a WDL model, the Wide component primarily processes highly sparse features, which requires minimal computation and is not computationally intensive. Notably, CPUs excel at handling both sparse features and irregular memory access patterns. In small and medium-scale inference scenarios, deploying an optimized WDL inference model on CPU instances delivers better cost-effectiveness than using GPU instances.
- Wide model: Excels at memorizing high-frequency feature combinations but lacks the ability to generalize to unseen feature interactions.
- Deep model: Generalizes low-frequency features via embedding vectors but exhibits low learning efficiency for high-frequency feature combinations.
Procedure
- Purchase an ecs.g4i instance. In the example provided in this topic, the following instance configurations are used.
- Instance type: ecs.g4i.16xlarge
- Image: Ubuntu 22.04 64 bit
- Network: an assigned EIP
- Execute the following command to install dependencies.
- apt updateapt install docker.io
- Execute the following command to obtain the WDL model image optimized for the ecs.g4i instance type.
- China (Beijing)China (Shanghai)China (Guangzhou)Asia Pacific (Johor)wget https://iaas-debug-beijing.tos-cn-beijing.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-shanghai.tos-cn-shanghai.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-guangzhou.tos-cn-guangzhou.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-southeast-1.tos-ap-southeast-1.ivolces.com/tool/cpu-serving/saved_model.tar
- (Optional) Modify the batch processing parameter for model inference based on your business requirements. In the example provided in this topic, the default value is used.
- Execute the following command to open the batching_config.txt file.
- Press the i key to enter edit mode, and modify the batch processing parameter based on your business requirements.
- max_batch_size { value: 4096 } batch_timeout_micros { value: 5000 } num_batch_threads { value: 64 } max_enqueued_batches { value: 204800 }
- Parameter descriptions:
- max_batch_size: The maximum number of samples per batch. The default value is 4096.
- batch_timeout_micros: The batch processing timeout period in microseconds. The default value is 5000.
- num_batch_threads: The number of batch processing threads. The default value is 64.
- max_enqueued_batches: The maximum number of enqueued batches. The default value is 204800.
- Press the Esc key to exit the edit mode, enter :wq, and then press Enter to save the changes and exit the file.
- Execute the following command to run the image.
- docker run -t --name server \--rm --network host \-v /data00/saved_model:/models/wdl \-v /data:/data \cr-for-diting-nodelete-cn-beijing.cr.volces.com/practice-images/tensorflow-serving:tf2.15-cpu-fp16 \--model_name=wdl \--model_base_path=/models/wdl \--enable_batching=true \--batching_parameters_file=/models/wdl/batching_config.txt \--grpc_channel_arguments=grpc.max_concurrent_streams=5000000 \--enable_profiler=true \--grpc_max_threads=1000 \--rest_api_num_threads=1000 \--mixed_precision=bfloat16
tip
- Change /data00/saved_model to the actual storage path of the WDL model.
- --rest_api_num_threads: Used to specify the maximum number of threads for the REST API service. In the example provided in this topic, this parameter is set to 1000.
- --mixed_precision: The mixed precision computation type. When FP16 (float16) or BF16 (bfloat16) is specified as data precision, you can enable the Advanced Matrix Extensions (AMX) instruction set of the CPU to improve inference performance. In the example provided in this topic, this parameter is set to bfloat16.
- Perform a stress test.
- Execute the following command to create a stress test script.
- vim grpc_client.py
- Press the i key to enter edit mode, and then copy the following content into the script file.
- *Script content *tip *You can modify the following parameters based on your business requirements: *request-rate: The required QPS. *max-concurrency: The maximum concurrency. *batch-size: The number of batches for each query. *num-requests: The total number of requests. * *Script contenttipYou can modify the following parameters based on your business requirements:
- request-rate: The required QPS.
- max-concurrency: The maximum concurrency.
- batch-size: The number of batches for each query.
- num-requests: The total number of requests.
import argparseimport asyncioimport timeimport numpy as npimport tensorflow as tffrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutorimport grpcfrom tensorflow_serving.apis import predict_pb2from tensorflow_serving.apis import prediction_service_pb2_grpcfrom statistics import meanimport multiprocessing# The log function, which determines whether to print logs based on the value of the debug parameter.def debug_print(message, debug=False):if debug:print(message)# The function for generating a single request (for parallel processing).def generate_single_request(batch_size=1):request = predict_pb2.PredictRequest()request.model_spec.name = 'wdl'request.model_spec.signature_name = "serving_default"# Generate numerical features I1-I13 (batch dimension).for i in range(1, 14):request.inputs[f"I{i}"].CopyFrom(tf.make_tensor_proto(np.random.uniform(0,1, [batch_size,1]), dtype=tf.float32))# Generate category features C1–C26 (batch dimension).for i in range(1, 27):categories = [f"cat_{i}_{j}" for j in range(5)]batch_values = [np.random.choice(categories) for _ in range(batch_size)]request.inputs[f"C{i}"].CopyFrom(tf.make_tensor_proto(batch_values, dtype=tf.string))return request# Pre-generate all request data. The batch_size parameter is supported.def generate_all_requests(num_requests, batch_size=1, num_workers=None, debug=False):# By default, the number of CPU cores is used as the number of worker processes.if num_workers is None:num_workers = min(multiprocessing.cpu_count(), num_requests)debug_print(f"{num_workers} processes are used to generate request data in parallel...", debug)# The number of requests generated by each process.requests_per_worker = [num_requests // num_workers] * num_workersfor i in range(num_requests % num_workers):requests_per_worker[i] +=1# Use a process pool to generate requests in parallel.with ProcessPoolExecutor(max_workers=num_workers) as executor:futures = []for count in requests_per_worker:# Specify a different random seed for each process to ensure data independence.seed = np.random.randint(0, 1000000)future = executor.submit(_worker_generate_requests, count, batch_size, seed)futures.append(future)# Collect all results.requests = []for future in futures:requests.extend(future.result())return requests# Worker process functiondef _worker_generate_requests(num_requests, batch_size, seed):# Specify random seeds to ensure that each process generates different data.np.random.seed(seed)return [generate_single_request(batch_size) for _ in range(num_requests)]# An asynchronous request generator (using a pre-generated request list)async def get_request(requests, request_rate, burstiness, max_batches, batch_counter, counter_lock, start_time, duration, debug=False):assert burstiness >0, "Burstiness factor must be positive."theta = 1.0 / (request_rate * burstiness)request_idx =0total_requests = len(requests)while True:# Check time limit.current_time = time.time()elapsed_time = current_time - start_timeif elapsed_time > duration:debug_print(f"[get_request] Time limit triggered: {elapsed_time:.4 f} s > {duration} s", debug)break# Use locks to protect counter operations.async with counter_lock:if request_idx >= total_requests or batch_counter[0] >= max_batches:reason = "Request index exhausted" if request_idx >= total_requests else "Batch count upper limit reached"debug_print(f"[get_request] Counter limit triggered: {reason}, Batch count: {batch_counter[0]}/{max_batches}", debug)break# Obtain the current request and add counters.current_request = requests[request_idx]request_idx +=1batch_counter[0] +=1yield current_requestif request_rate != float("inf"):interval = np.random.gamma(shape=burstiness, scale=theta)await asyncio.sleep(interval)# Call the gRPC interface.def call_grpc_stub(stub, request):start_time = time.time()result = stub.Predict(request, 10.0) # 10-second timeoutend_time = time.time()return end_time - start_time # Return the latency.# Concurrency monitorclass ConcurrencyMonitor:def __init__(self, max_concurrency):self.max_concurrency = max_concurrencyself.current =0self.lock = asyncio.Lock()self.highest =0async def increment(self):async with self.lock:self.current +=1if self.current > self.highest:self.highest = self.currentasync def decrement(self):async with self.lock:self.current -=1async def get_current(self):async with self.lock:return self.currentasync def get_highest(self):async with self.lock:return self.highest# Monitoring concurrent usage.async def concurrency_monitor(monitor, interval=3, debug=False):"""Periodically print concurrent usage information."""if not debug:return # If debug is not enabled, information is returned immediately.try:while True:current = await monitor.get_current()highest = await monitor.get_highest()print(f"[monitoring] Current concurrency: {current}/{monitor.max_concurrency}, historical maximum concurrency: {highest}")await asyncio.sleep(interval)except asyncio.CancelledError:# Normal cancellationdebug_print("[Monitoring] The monitoring task has stopped", debug)passexcept Exception as e:debug_print(f"[Monitoring] Monitoring task exception: {e}", debug)# Coroutine pool managerclass WorkerPool:def __init__(self, max_workers, stub, request_generator, semaphore, latencies, monitor, loop, start_time, duration, debug=False):self.max_workers = max_workersself.stub = stubself.request_generator = request_generatorself.semaphore = semaphoreself.latencies = latenciesself.monitor = monitorself.loop = loopself.start_time = start_timeself.duration = durationself.debug = debugself.workers = []self.queue = asyncio.Queue()self.running = True# Create initial worker coroutines.for i in range(min(max_workers,5)): # Initially start five or fewer worker coroutines.self._create_worker()def _create_worker(self):worker_id = len(self.workers) +1task = asyncio.create_task(self._worker_loop(worker_id))self.workers.append(task)debug_print(f"[WorkerPool] Create worker coroutines. {worker_id}, total:{len(self.workers)}", self.debug)async def _worker_loop(self, worker_id):debug_print(f"[worker-{worker_id}] start", self.debug)while self.running:try:# Wait for requests.request = await asyncio.wait_for(self.queue.get(), timeout=1.0)if request is None: # Exit signalbreak# Check whether the maximum duration has been reached.current_time = time.time()if current_time - self.start_time > self.duration:debug_print(f"[worker-{worker_id}] Exit: Time limit", self.debug)break# Increase the concurrency count.await self.monitor.increment()async with self.semaphore:try:# Specify request timeout period.remaining_time = self.duration - (time.time() - self.start_time)if remaining_time <=0:debug_print(f"[worker-{worker_id}] Insufficient remaining time for the request: {remaining_time:.4 f} s", self.debug)breaklatency = await asyncio.wait_for(self.loop.run_in_executor(ThreadPoolExecutor(), call_grpc_stub, self.stub, request),timeout=remaining_time)self.latencies.append(latency)debug_print(f"[worker-{worker_id}] Request completed, latency: {latency:.4 f} s", self.debug)except asyncio.TimeoutError:debug_print(f"[worker-{worker_id}] Request timeout: remaining time {remaining_time:.4 f} s", self.debug)pass # Timeout requests are not logged.except Exception as e:debug_print(f"[worker-{worker_id}] Request exception: {e}", self.debug)passfinally:# Reduce the concurrency count.await self.monitor.decrement()self.queue.task_done()except asyncio.TimeoutError:# Check if exit is needed.if not self.running:breakasync def submit(self, request):# If the queue length is greater than the current number of worker coroutines and the maximum number of coroutines has not been reached, create a new worker coroutine.if self.queue.qsize() >= len(self.workers) and len(self.workers) < self.max_workers:self._create_worker()# Add the request to the queue.await self.queue.put(request)async def close(self):# Stop receiving new requests.self.running = False# Put an exit signal into the queue.for _ in range(len(self.workers)):await self.queue.put(None)# Wait for all worker coroutines to complete.await asyncio.gather(*self.workers, return_exceptions=True)debug_print(f"[WorkerPool] All worker coroutines have stopped, total:{len(self.workers)}", self.debug)# Execute stress tests asynchronously.async def async_call_grpc(stub, requests, request_rate, burstiness, max_concurrency, duration, max_batches, debug=False):start_time = time.time()latencies = []semaphore = asyncio.Semaphore(max_concurrency)loop = asyncio.get_running_loop()batch_counter = [0] # Use a list as a mutable object for atomic batch counting.counter_lock = asyncio.Lock() # Lock protecting the counterstop_event = asyncio.Event() # Used to notify all coroutines to stop# Create a request generator.request_generator = get_request(requests, request_rate, burstiness, max_batches, batch_counter, counter_lock, start_time, duration, debug)# Create a concurrency monitor.concurrency_monitor_obj = ConcurrencyMonitor(max_concurrency)# Start concurrency monitoring (if debug is enabled).monitor_task = asyncio.create_task(concurrency_monitor(concurrency_monitor_obj,3, debug))# Create a coroutine pool.worker_pool = WorkerPool(max_workers=max_concurrency,stub=stub,request_generator=request_generator,semaphore=semaphore,latencies=latencies,monitor=concurrency_monitor_obj,loop=loop,start_time=start_time,duration=duration,debug=debug)# Obtain requests from the request generator and submit them to the coroutine pool.async def feeder():try:async for request in request_generator:if time.time() - start_time > duration or stop_event.is_set():breakawait worker_pool.submit(request)except Exception as e:debug_print(f"[feeder] exception: {e}", debug)finally:# Disable the coroutine pool.await worker_pool.close()stop_event.set()# Start the request-supplying coroutine.feeder_task = asyncio.create_task(feeder())# Wait for all tasks to complete.try:await asyncio.gather(feeder_task, return_exceptions=True)finally:# Stop monitoring tasks.monitor_task.cancel()await monitor_task# Obtain the maximum concurrency.highest_concurrency = await concurrency_monitor_obj.get_highest()debug_print(f"[async_call_grpc] Highest concurrency: {highest_concurrency}/{max_concurrency}", debug)# Filter requests to keep only valid ones (to avoid invalid data caused by timeout).valid_latencies = [t for t in latencies if t <= duration]return batch_counter[0], valid_latencies, highest_concurrency# Main functionasync def main():parser = argparse.ArgumentParser(description='TensorFlow Serving stress test tool')parser.add_argument('--request-rate', type=float, default=10.0, help='Target QPS')parser.add_argument('--burstiness', type=float, default=1.0, help='Precipitating factor')parser.add_argument('--duration', type=float, default=60.0, help='Stress test duration (seconds)')parser.add_argument('--server', type=str, default='localhost:8500', help='Service address')parser.add_argument('--num-requests', type=int, default=1000, help='Total request batch count')parser.add_argument('--max-concurrency', type=int, default=50, help='Maximum concurrency')parser.add_argument('--batch-size', type=int, default=1, help='Sample batch size for each request')parser.add_argument('--processes', type=int, default=None, help='Number of request-generating processes')parser.add_argument('--debug', action='store_true', help='Enable debugging information and monitoring')args = parser.parse_args()print(f"Stress test configuration:")print(f" Request rate: {args.request_rate} QPS")print(f" Stress test duration: {args.duration}second")print(f" Maximum number of requests: {args.num_requests} batches")print(f" Maximum concurrency: {args.max_concurrency}")print(f" Batch size: {args.batch_size}")print(f" Target server: {args.server}")print(f" Debugging mode:{'Enable' if args.debug else 'Disable'}")# Ensure sufficient requests are pre-generated.generate_count = max(args.num_requests, args.max_concurrency *2)print(f"Generating {generate_count} request data items, batch size: {args.batch_size}...")start_gen = time.time()requests = generate_all_requests(generate_count, args.batch_size, args.processes, args.debug)print(f"Request data generation completed, duration: {time.time()-start_gen:.2 f} seconds")# Create a gRPC connection.channel = grpc.insecure_channel(args.server)stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)# Perform a stress test.print(f"Start the stress test. Target QPS={args.request_rate}, concurrency={args.max_concurrency}, batch size={args.batch_size}")start_time = time.time()completed_batches, latencies, highest_concurrency = await async_call_grpc(stub, requests, args.request_rate, args.burstiness,args.max_concurrency, args.duration, args.num_requests,args.debug)# Calculate performance metrics.total_time = max(time.time() - start_time, 0.001)throughput = completed_batches / total_timep99_latency = np.percentile(latencies, 99) if latencies else 0avg_latency = np.mean(latencies) if latencies else 0total_samples = completed_batches * args.batch_sizesamples_per_second = total_samples / total_timeprint("\n============ Stress test results ============")print(f"Completed requests (batches): {completed_batches} / {args.num_requests}")print(f"Number of completed samples: {total_samples}")print(f"Total duration: {total_time:.2 f} seconds")print(f"Request throughput (QPS): {throughput:.2 f}")print(f"Sample throughput (samples/second): {samples_per_second:.2 f}")print(f"Average latency: {avg_latency:.4 f} seconds")print(f"P99 latency: {p99_latency:.4 f} seconds")print(f"Maximum concurrency: {args.max_concurrency}")print(f"Actual maximum concurrency: {highest_concurrency}")print(f"Batch size: {args.batch_size}")if __name__ == "__main__":# Windows System requires this condition.if hasattr(multiprocessing, 'freeze_support'):multiprocessing.freeze_support()asyncio.run(main())
- Press the Esc key to exit edit mode, enter :wq, and then press Enter to save and exit the file.
- Execute the following command to perform a stress test.
- python3 grpc_client.py \--request-rate 80 \--duration 30 \--num-requests 6000 \--max-concurrency 90 \--batch-size 2048 \--server 192.168.0.2:8500
Test conclusion
This topic demonstrates how to deploy and run a WDL model on an ecs.g4i.16xlarge instance, and provides the maximum throughput of the model before and after AMX acceleration is enabled.
Number of batch processing threads | Data precision | |
Number of batch processing threads | Data precision | |
| BF16 (AMX acceleration enabled) | FP32 (AMX acceleration not enabled) |
2,048 | 161.23 | 16.18 |
1,024 | 390.91 | 86.43 |
512 | 824.03 | 226.61 |
In scenarios where cost-effectiveness and abundant instance resources are prioritized, you can use instance types equipped with the 6th-generation Intel® Xeon® Scalable processor (Granite Rapids), such as the General-purpose g4i instances, as an alternative to GPU instances. This way, you can obtain the most cost-effective WDL model inference service by leveraging system-level acceleration solutions.
Last updated: 2026.06.18 17:48:27
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
