Una estructura arbol se recorre por medio de una función recursiva empezando desde el nodo raiz por medio de dos operaciones básicas, un método ProcessNode con el que procesamos la información de ese nodo, y otro GetNodeChilds con el que obtenemos los nodos hijos de un nodo, sería algo así:
static void BrowseTree(Node root)
{
ProcessNode(root);
foreach (Node n in GetNodeChilds(root))
{
BrowseTree(n);
}
}
Pero cuando esa estructura en arbol es realmente grande, ó alguna de las dos operaciones conlleva una latencia significativa, como puede ser por ejemplo procesar cada nodo localmente u obtener los hijos de un nodo a través de la red, puede ser muy útil recorrer dicha estructura con múltiples hilos si se dan las condiciones adecuadas, que pueden ser:
- Latencia elevada que provoca tiempos de Idle significativos al obtener los hijos.
- Múltiples CPUs en la máquina que navega el arbol.
- Múltiples CPUs en la máquina que contiene el arbol, capacidad para atender peticiones concurentes y preferiblemente concurrencia optimista.
El algorrítmo se complica un poco, pero gracias al modelo de programación asíncrono podemos simplificar bastante. A priori, puede parece que es tan simple como lanzar un hilo por cada nodo hijo del nodo raiz, pero si el nodo raiz tiene muchos hijos en su primer nivel podemos dejar vacio el ThreadPool pudiendo causar un deadlock ó en caso de usar Threads convencionales llegar a un número contraproducente de ellos, por lo que debemos controlar cuantos hilos vamos a usar para recorrer el arbol. Pero no esta todo solucionado, si el primer nodo hijo tiene 4 hijos, y el segundo nodo hijo tiene 400... el hilo que se encargue del primer nodo quedará desaprovechado mientras que el que se encargue del segundo tendrá mucho trabajo por delante, por lo que debemos poder reutilizar los hilos de forma dinámica.
Estos dos problemas quedan resueltos con el siguiente modelo:
Nota: Estos ejemplos de código son orientativos y simplificados.
static BrowseTreeDlgt_ BrowseTreeDlgt = new BrowseTreeDlgt_(BrowseTree);
static void BrowseTree(Node root)
{
ProcessNode(root);
foreach (Node n in GetNodeChilds(root))
{
if (runningThreads<maxThreads)
{
runningThreads++;
BrowseTreeDlgt.BeginInvoke(n,
delegate(IAsyncResult ia)
{
BrowseTreeDlgt.EndInvoke(ia);
runningThreads--;
},
null);
}
else
{
BrowseTreeDlgt.Invoke(n);
}
}
}
maxThreads define el número máximo de hilos a usar y runningThreads los que tenemos ya en funcionamiento, si hay hilos disponibles se lanza el nodo en otro hilo (.BeginInvoke) y si no, pues se continua con el hilo actual (.Invoke) como si de una función recursiva se tratase. Cada vez que creamos un hilo, incrementamos runningThreads y lo decrementamos cuando acaba, pero ... ¿en que orden exactamente? Pues el incremento lo antes posible y el decremento lo más tarde posible, de forma que se reduzca la posiblidad de crear hilos de más. De esta forma controlamos la cantidad de hilos.
Este modelo tiene una pega particular, y es el hecho de que cuando lanzamos nodos con todos sus hijos en otro hilo, la ejecución del hilo principal llegará al final y no sabremos si el resto de hilos secundarios han acabado ya de procesar todos sus nodos. El mejor planteamiento es bloquear el hilo que llama a la función hasta que se acabe de recorrer el arbol, para ello bloqueamos dicho hilo y cuando el número de hilos secundarios sea 0, lo liberamos. Sería algo así:
static void BrowseTree(Node root)
{
ProcessNode(root);
foreach (Node n in GetNodeChilds(root))
{
if (runningThreads<maxThreads)
{
runningThreads++;
BrowseTreeDlgt.BeginInvoke(n,
delegate(IAsyncResult ia)
{
BrowseTreeDlgt.EndInvoke(ia);
runningThreads--;
if (runningThreads == 0)
lock (endLock)
Monitor.Pulse(endLock);
},
null);
}
else
{
BrowseTreeDlgt.Invoke(n);
}
}
}
Y al llamar a la función:
lock (endLock)
{
BrowseTree(node);
Monitor.Wait(endLock);
}
Esto hará que se llame a BrowseTree y cuando termine el hilo principal quede bloquedado hasta que runningThreads sea 0. Pero siguen habiendo problemas... :D
Una de las cosas que debemos tener en cuenta en entornos multithreading es que no hay un orden concreto y que la memoria puede ser alterada en cualquier momento, incluso entre instrucción e instrucción. Si por cualquier motivo, el hilo principal acaba después de los hilos secundarios... (por que le ha tocado procesar más nodos..) ... el Pulse llegaría antes que el Wait, con lo que la aplicación quedaría en un deadlock. Para evitar esta situación, lo ideal es llamar a BrowseTree asíncronamente de forma que el hilo que llama a la función quede bloqueado en el Wait inmediatamente y aunque añadamos un hilo adicional... la carga es la misma:
lock (endLock)
{
BrowseTreeDlgt.BeginInvoke(node,
delegate(IAsyncResult ia)
{BrowseTreeDlgt.EndInvoke(ia);},
null);
Monitor.Wait(endLock);
}
Ahora que esta resuelto el tema de "bloquear hasta terminar", hay que dar un repaso a la condición que lo desbloquea, la de que los hilos secundarios en ejecución sea 0... Como decía la memoria puede ser alterada y/ó evaludada en cualquier momento, entonces se puede dar la situación (de hecho se da) de que se haga un Pulse sobre el bloqueo mientras que se esté a punto de iniciar otro hilo secundario porque hay más nodos que procesar en el bucle... Por este motivo, hay que tener en cuenta si hay nodos por procesar antes de hacer el Pulse, es decir, la condición para que se libere el bloqueo y por tanto se de por terminado el recorrido por el arbol es, que no haya hilos en ejecución y que los nodos por procesar sean 0, con lo que se asegura que cuando se den estas dos condiciones es el último hilo en ejecución. Ahora debemos comprobar tanto al final de las ejecuciones síncronas como asíncronas:
static void BrowseTree(Node root)
{
ProcessNode(root);
Node[] childs =GetNodeChilds(root);
nodesRemaining += childs.Length;
foreach (Node n in childs)
{
if (runningThreads < maxThreads)
{
runningThreads++;
BrowseTreeDlgt.BeginInvoke(n,
delegate(IAsyncResult ia)
{
BrowseTreeDlgt.EndInvoke(ia);
runningThreads--;
nodesRemaining--;
if ((runningThreads == 0)&&
(nodesRemaining==0))
lock (endLock)
Monitor.Pulse(endLock);
},
null);
}
else
{
BrowseTreeDlgt.Invoke(n);
nodesRemaining--;
if ((runningThreads == 0) &&
(nodesRemaining == 0))
lock (endLock)
Monitor.Pulse(endLock);
}
}
}
Como handicap el uso de multiples hilos trae los problemas que implica la sincronización, cosa en la que habrá que poner sumo cuidado y ... sobre todo ... imaginación, la principal herramienta del programador. La forma ideal de ejecutar esta función sería algo más compleja:
static BrowseTreeDlgt_ BrowseTreeDlgt = new BrowseTreeDlgt_(BrowseTree);
static void BrowseTree(Node root)
{
ProcessNode(root);
Node[] childs =GetNodeChilds(root);
lock(endLock)
nodesRemaining += childs.Length;
foreach (Node n in childs)
{
if (runningThreads < maxThreads)
{
Interlocked.Increment(ref runningThreads);
BrowseTreeDlgt.BeginInvoke(n,
delegate(IAsyncResult ia)
{
try
{
Monitor.Enter(endLock);
BrowseTreeDlgt.EndInvoke(ia);
runningThreads--;
nodesRemaining--;
}
finally
{
if ((runningThreads == 0) &&
(nodesRemaining == 0))
Monitor.Pulse(endLock);
Monitor.Exit(endLock);
}
},
null);
}
else
{
BrowseTreeDlgt.Invoke(n);
lock (endLock)
{
nodesRemaining--;
if ((runningThreads == 0) &&
(nodesRemaining == 0))
Monitor.Pulse(endLock);
}
}
}
}
Como se puede observar, el principal problema es no saber cuantos nodos tiene el arbol hasta que se han recorrido todos, pero con este planteamiento queda resuelto, al menos es lo que dicen las pruebas concienzudas con distintas CPUs y sistemas operativos que he hecho.
En el próximo artículo un ejemplo de una clase que realiza esta tarea mismo, con ejemplos incluidos ... ;)
Subscribir