Tutoriel Lua: Les coroutines - Le multithreading non-préemptif.

Les coroutines: Le multithreading non-préemptif.

Comme vu précédemment, les coroutines sont une sorte de multithreading collaboratif.

Chaque coroutine est équivalent à un thread.

Une paire d'interrupteurs yield-resume contrôle les deux threads de l'un à l'autre.
Cependant, contrairement au "vrai" multithreading, les coroutines sont non-préemptives.

Lorsqu'une coroutine est en fonctionnement, elle ne pourra pas être arrêtée de l'extérieur. Elle ne suspendra son exécution que si une demande explicite est faite (via un appel à yield ). Pour plusieurs applications, ce n'est pas un problème, bien au contraire. La programmation est beaucoup plus facile en l'absence de préemption. Vous n'avez pas besoin d'être paranoïaque à propos de la synchronisation des bugs, car toute synchronisation entre les threads est explicite dans le programme. Vous n'avez qu'à vous assurer qu'une coroutine.yield() intervienne seulement quand elle est en dehors d'une région critique.

Cependant, avec le multithreading non-préemptif, chaque fois qu'un thread appelle une opération de blocage, c'est l'ensemble du programme qui s'arrête jusqu'à la fin de l'opération. Or, pour la plupart des applications, c'est un comportement inacceptable, ce qui entraîne de nombreux programmeurs à faire abstraction des coroutines comme d'une véritable alternative au traditionnel multithreading. Mais comme vous allez le voir ci-dessous, ce problème a une intéressante et évidente solution.

Supposez une situation typique de multithreading:
Vous voulez télécharger plusieurs fichiers à distance, via le protocole HTTP.

Bien sûr, pour télécharger plusieurs fichiers à distance, vous devez savoir comment télécharger un fichier distant.
Dans cet exemple, vous allez utiliser la bibliothèque luasocket, développée par Diego Nehab.

Pour télécharger un fichier, vous devez ouvrir une connexion à un site, envoyer une demande au fichier, recevoir le fichier (en blocs), et fermer la connexion.
Dans Lua, vous pouvez écrire cette tâche comme suit.

Premièrement, chargez la bibliothèque luasocket.

	  require "luasocket"

Ensuite, vous définissez l'hôte et le fichier que vous voulez télécharger. Dans cet exemple, vous allez télécharger la spécification HTML 3.2 à partir du Web Consortium Site Wide.

	   host = "www.w3.org" 
	   file = "/TR/REC-html32.html"

Ensuite, vous ouvrez une connexion TCP vers le port 80 (qui est le port standard pour les connexions HTTP) de ce site.

	   c = assert(socket.connect(host, 80))

L'opération retourne un objet de connexion que vous utiliserez pour envoyer la demande de fichier:

	   c:send("GET " .. file .. " HTTP/1.0\r\n\r\n")

La méthode de réception retourne toujours une chaîne de caractères, plus une autre chaîne avec l'état de l'opération.
Et finalement, vous fermez la connexion:

	   c:close()

Maintenant que vous savez comment télécharger un fichier, revenons au problème de téléchargement de plusieurs fichiers.

L'approche triviale consisterait à télécharger un fichier à la fois. Mais, cette approche séquentielle, où l'on ne commence à lire un nouveau fichier qu'après avoir terminé la lecture du précédent, est trop lente.

Lors de la lecture d'un fichier distant, un programme passe le plus clair de son temps à attendre que les données arrivent. Plus précisément, il passe le plus clair de son temps bloqué dans l'attente de recevoir.

Aussi, pourrait-il être beaucoup plus performant s'il téléchargeait tous les fichiers simultanément. Lorsqu'une connexion n'aurait plus de données disponibles, le programme pourrait lire une autre connexion.

De toute évidence, les coroutines offrent un moyen pratique de structurer ces téléchargements simultanés. Vous créez un nouveau thread pour chaque tâche de téléchargement. Quand un thread n'a pas de données disponibles, il cède le contrôle à un répartiteur simple, qui invoque un autre thread.

Pour réécrire le programme avec les coroutines, vous devez d'abord réécrire le code précédent, comme une fonction de téléchargement.

		function download(host, file)
		  local c = assert(socket.connect(host, 80))
		  local count = 0    -- compte le nb d'octets lus
		  c:send("GET " .. file .. " HTTP/1.0\r\n\r\n")
		  while true do
			 local s, status = receive(c)
			 count = count + string.len(s)
			 if status == "closed" then break end
		  end
		  c:close()
		  print(file, count)
		end

Puisque vous n'êtes pas intéressé par le contenu du fichier, cette fonction ne comptera que le nombre d'octets du fichier. Dans ce nouveau code, vous utiliserez une fonction auxiliaire ( receive(c) ) afin de recevoir les données de la connexion. Dans l'approche séquentielle, son code serait comme ceci.

		function receive(connexion)
		  return connexion:receive(2^10)
		end

Pour la mise en oeuvre simultanée, cette fonction doit recevoir les données sans blocage. Et, s'il n'y a pas assez de données disponibles, alors, coroutine.yield() rentrera en fonction.
Le nouveau code ressemble à ceci.

		function receive(connection)
		  connection:timeout(0)  
		  local s, status = connection:receive(2^10)
		  if status == "timeout" then
			 coroutine.yield(connection)
		  end
		return s, status
		end

L'appel à timeout(0) rend sur la connexion, toute opération "non-bloquante".

Lorsque l'état de l'opération est timeout, cela signifie que l'opération retournée n'est pas encore terminée et demande à la coroutine.yield(connection) de faire son office.

L'argument non-false passé à yield signale au répartiteur que le thread doit toujours remplir sa mission. (Plus tard vous verrez une autre version où le répartiteur a besoin d'une connexion interrompue.)

La fonction suivante garantit que chaque téléchargement sera exécuté dans un thread individuel.

		threads = {} -- liste de tous les threads.
		function get(host, file)
		  -- création de la coroutine.
		  local co = coroutine.create(function()
			 download(host, file)
		  end)
		-- insère dans la liste.
		table.insert(threads, co)
		end

Le tableau des threads conserve une liste de toutes les threads en direct, pour le répartiteur.

Le répartiteur est principalement une boucle qui passe par tous les threads, en les appelant un par un.

Il faut aussi enlever de la liste les threads qui terminent leurs tâches.

La boucle s'arrête lorsqu'il n'y a plus de thread.

		function Dispatcher()
		  while true do
			 local n = table.getn(threads)
			 if n == 0 then break end   -- il n'y a plus de thread
			 for i=1,n do
				local status, res = coroutine.resume(threads[i])
			  -- le thread a-t-il fini sa tâche?
				if not res then	  
				  table.remove(threads, i)
				  break
				end
			 end
		  end
		end

Enfin, le programme principal créer les threads dont il a besoin et appelle le répartiteur.

Par exemple, pour télécharger quatre documents sur le site du W3C, le programme principal pourrait ressembler à ceci.

		host = "www.w3.org"
			 
		get(host, "/TR/html401/html40.txt")
		get(host,"/TR/2002/REC-xhtml1-20020801/xhtml1.pdf")
		get(host,"/TR/REC-html32.html")
		get(host,"/TR/2000/REC-DOM-Level-2-Core-20001113
					 /DOM2-Core.txt")

		-- appel à la boucle principale.		
		 Dispatcher()

Si une machine met six secondes à télécharger ces fichiers en utilisant ces quatre coroutines. Avec la mise en oeuvre séquentielle, il lui faudra deux fois plus de temps (15 secondes).

Mais cette dernière application est loin d'être optimale.

Tout va pour le mieux tant qu'au moins un thread a quelque chose à lire.

Mais, dans le cas contraire, le répartiteur ne reste pas inactif et va de thread en thread à la recherche de quelque chose à se mettre sous la dent...

Ce qui signifie que cette application utilise le CPU presque 30 fois plus que la solution séquentielle!

Pour éviter un tel comportement, vous pouvez utiliser la sélection de fonction à partir de luasocket.

Ce qui permet au programme d'attendre sagement le changement d'état à l'intérieur du groupe de sockets. Les changements mis en oeuvre sont de petite taille. Vous avez seulement changé le répartiteur.

		function Dispatcher()
		  while true do
			 local n = table.getn(threads)
			 if n == 0 then break end  -- il n'y a plus de thread.
				local connexions = {}
				for i = 1,n do
				local status, res = coroutine.resume(threads[i])
				if not res then  -- le thread a-t-il fini sa tâche?
				  table.remove(threads, i)
				  break
				else  -- timeout.
				  table.insert(connexions, res)
				end
			 end
			 if table.getn(connexions) == n then
				socket.select(connexions)
			 end
		  end
		end

Ce nouveau Dispatcher() recueille les connexions "time-out" dans le tableau connexions.

Rappelez-vous que Receive() passe les connexions à yield et que resume les retourne.

Lorsque toutes les connexions à "time-out" sont écoulées, le dispatcher() appelle select().

Au final, cette mise en oeuvre tourne aussi vite que celle faite à partir des coroutines.

Mais, compte tenu du fait que lorsque le CPU n'est pas occupé... c'est qu'il attend, on en déduit que son utilisation est juste un peu plus importante que lors de la mise en oeuvre séquentielle.