ExecuteRequests processes requests concurrently with comprehensive error handling and resource management. This method is the main entry point for bulk permission checking. It: 1. Stops collecting new requests 2. Sorts all collected requests 3. Processes them concurrently with controlled parallelism
(size uint32)
| 259 | // Returns: |
| 260 | // - error: Any error that occurred during processing (context cancellation is not considered an error) |
| 261 | func (bc *BulkChecker) ExecuteRequests(size uint32) error { // Main execution entry point |
| 262 | if size == 0 { |
| 263 | return fmt.Errorf("size must be greater than 0") |
| 264 | } |
| 265 | // Stop collecting new requests and wait for collection to complete |
| 266 | bc.StopCollectingRequests() // Ensure no new requests are added |
| 267 | |
| 268 | // Get sorted requests for processing |
| 269 | requests := bc.getSortedRequests() |
| 270 | if len(requests) == 0 { |
| 271 | return nil // No requests to process |
| 272 | } |
| 273 | |
| 274 | // Initialize execution state for tracking progress |
| 275 | bc.executionState = &executionState{ |
| 276 | results: make([]base.CheckResult, len(requests)), |
| 277 | limit: int64(size), |
| 278 | } |
| 279 | |
| 280 | // Create execution context with cancellation for graceful shutdown |
| 281 | execCtx, execCancel := context.WithCancel(bc.ctx) |
| 282 | defer execCancel() |
| 283 | |
| 284 | // Create semaphore to control concurrency |
| 285 | sem := semaphore.NewWeighted(int64(bc.config.ConcurrencyLimit)) |
| 286 | |
| 287 | // Create error group for managing goroutines and error propagation |
| 288 | g, ctx := errgroup.WithContext(execCtx) |
| 289 | |
| 290 | // Process requests concurrently |
| 291 | for i, req := range requests { |
| 292 | // Check if we've reached the success limit |
| 293 | if atomic.LoadInt64(&bc.executionState.successCount) >= int64(size) { |
| 294 | break |
| 295 | } |
| 296 | |
| 297 | index := i |
| 298 | request := req |
| 299 | |
| 300 | // Launch goroutine for each request |
| 301 | g.Go(func() error { |
| 302 | return bc.processRequest(ctx, sem, index, request) |
| 303 | }) |
| 304 | } |
| 305 | |
| 306 | // Wait for all goroutines to complete and handle any errors |
| 307 | if err := g.Wait(); err != nil { |
| 308 | if isContextError(err) { |
| 309 | return nil // Context cancellation is not an error |
| 310 | } |
| 311 | return fmt.Errorf("bulk execution failed: %w", err) |
| 312 | } |
| 313 | |
| 314 | return nil |
| 315 | } |
| 316 | |
| 317 | // processRequest handles a single request with comprehensive error handling. |
| 318 | // This method is executed in a separate goroutine for each request and: |