Le mécanisme d'annulation de ForkJoinTask repose sur un drapeau coopératif plutôt que sur une interruption forcée des threads. Cela signifie que cancel() se contente de définir un statut interne volatile que les tâches doivent interroger explicitement pour observer les demandes de terminaison. En conséquence, cette conception ne parvient pas à débloquer les threads en attente d'opérations I/O monolithiques, telles que les lectures de FileChannel ou les opérations InputStream sur les sockets. Ces appels bloquants ne vérifient pas le drapeau d'annulation et ne peuvent pas être interrompus par les mécanismes standards d'interruption des threads.
Pour prévenir la famine du pool lorsque les travailleurs sont bloqués, l'API ForkJoinPool.managedBlock permet aux développeurs d'enregistrer une instance de ForkJoinPool.ManagedBlocker. Ce bloqueur signale au pool de lancer un thread de travail compensatoire, maintenant ainsi le niveau de parallélisme cible malgré le travail bloquant. La méthode isReleasable du bloqueur fournit une accroche pour vérifier l'état d'annulation ou interrompre l'opération bloquée par programmation. Cela permet au pool de se dégrader gracieusement plutôt que d'épuiser son budget de threads sur des I/O non réactifs.
Nous avons rencontré cette limitation lors de la construction d'un processeur de logs parallèle qui utilisait Files.lines() dans une RecursiveTask personnalisée. La tâche analysait des fichiers de log à l'échelle du téraoctet depuis un dispositif de stockage monté sur le réseau. Lorsque les utilisateurs demandaient l'annulation d'analyses longues, les threads de ForkJoinPool restaient bloqués dans les appels système read() pendant des minutes. Ils ignoraient complètement le drapeau d'annulation, empêchant de nouvelles tâches de démarrer et provoquant une grave famine de threads.
Nous avons considéré trois approches distinctes pour résoudre le blocage. La première approche consistait à abandonner complètement ForkJoinPool et à passer à un ThreadPoolExecutor mis en cache. Cela offrait des sémantiques d'interruption plus simples et un remplacement immédiat des threads, maisSacrifiant l'efficacité du vol de travail cruciale pour nos étapes d'analyse intensives en CPU.
La deuxième approche proposait d'envelopper chaque appel I/O dans une logique Thread.interrupt() et de passer à des canaux interruptibles comme SocketChannel. Bien que cela prenne en charge l'annulation immédiate, cela s'est révélé invasif et incompatible avec le code de bibliothèque hérité qui reposait sur des flux bloquants standards et des parseurs tiers.
La troisième approche a tiré parti de ForkJoinPool.managedBlock en implémentant un ManagedBlocker personnalisé qui enveloppait la boucle de lecture de fichiers. Ce bloqueur vérifiait périodiquement isCancelled() tout en permettant au pool de lancer des threads compensatoires via le protocole de bloqueur. Nous avons choisi la troisième solution car elle préservait l'architecture de flux parallèle existante tout en informant explicitement le pool des opérations bloquantes. Cela garantissait que la réactivité à l'annulation et le débit restaient équilibrés sans réécrire toute la couche I/O.
Le résultat était un système où les demandes d'annulation se propageaient en quelques secondes plutôt qu'en quelques minutes. Le pool a évolué dynamiquement jusqu'à cinquante threads pendant les pics d'I/O sans configuration manuelle. La saturation du CPU est restée élevée tout au long de la charge de travail, et la terminaison des tâches est devenue fiable même en cas de forte congestion réseau.
Comment le ForkJoinPool détecte-t-il le blocage des threads sans appels explicites à managedBlock, et quel est le seuil pour le lancement de threads compensatoires ?
Le pool suit en interne les états des threads de travail via un champ ctl de 64 bits représentant les comptes actifs par rapport aux comptes garés. Il compte les travailleurs comme "actifs" lorsqu'ils exécutent des tâches, mais ne peut pas distinguer entre le travail intensif en CPU et l'I/O bloquant sans indications du programmeur. Lorsqu'un travailleur est bloqué sur un moniteur de synchronisation ou une I/O sans utiliser managedBlock, le pool observe uniquement une réduction du travail volable et des travailleurs disponibles. Il peut finir par stagner si le niveau de parallélisme est atteint et aucun signal de progression n'arrive. Les threads compensatoires ne se lancent de manière fiable que lorsque managedBlock est invoqué, ou lorsque le blocage interne de la JVM est détecté via les compteurs Unsafe.park, mais le seuil par défaut est opaque et peu fiable pour le code de blocage personnalisé.
Pourquoi ForkJoinTask.join() ne retourne-t-il pas immédiatement lorsque la tâche est annulée, et en quoi cela diffère-t-il de Future.get() avec un timeout ?
join() appelle en interne doJoin(), qui implémente un mécanisme de "aide" où le thread appelant exécute ou vole d'autres travaux jusqu'à ce que la tâche cible soit terminée. Cela se produit indépendamment de l'état d'annulation, car l'annulation ne prévient que le fork de nouveaux sous-tâches et définit un drapeau de complétion. La méthode ne vérifie pas le drapeau d'annulation avant d'attendre, ni ne lance CancellationException à l'entrée. En revanche, Future.get() sur une ForkJoinTask (qui implémente Future) vérifie l'état d'annulation immédiatement et peut lancer CancellationException sans attendre. Cette distinction est vitale car join() est conçu pour la coopération intra-pool, tandis que get() est destiné aux clients externes s'attendant à des sémantiques standard de Future.
Quelle est l'interaction entre le niveau de parallélisme de ForkJoinPool et Runtime.availableProcessors(), et pourquoi le fait de définir un parallélisme supérieur au nombre de processeurs disponibles pourrait-il améliorer le débit pour les opérations bloquantes ?
Le pool commun par défaut s'initialise avec availableProcessors() - 1 pour réserver un cœur pour le thread applicatif ou la collecte des ordures. Le parallélisme définit le nombre cible de threads actifs, pas un maximum strict ; le pool peut créer plus de threads si managedBlock indique un travail bloquant, mais vise à garder uniquement les threads de parallélisme réellement actifs. Pour les opérations bloquantes, définir le parallélisme supérieur au nombre de cœurs (par exemple, 2x ou 3x cœurs) permet au planificateur de maintenir les CPU occupés pendant que d'autres threads attendent sur l'I/O. Cela modélise la limitation "thread-par-cœur" en s'assurant que des tâches exécutables existent pour chaque cœur malgré le blocage. Cependant, cela nécessite un réglage minutieux pour prévenir une surcharge de commutation de contexte excessive lorsque le ratio de blocage est mal estimé.