Multihilos en xHarbour - Parte VI

En estas primeras entregas sobre los multihilos de xHarbour, he hablado de funciones para crear un nuevo hilo, para sincronizar ejecuciones y para sincronizar accesos a recursos.

Sin embargo, puede suceder que tengamos un hilo al que necesitamos finalizar su ejecución porque no se comporta como esperábamos.
Para eso existen 2 funciones. Stopthread y Killthread.

Además, junto con Stopthread, existe una función para sincronizar la finalización del thread con el thread que lo está finalizando.

Stopthread

Sirve para detener la ejecución de un thread de forma amigable, marcando al thread para que se cierre cuando llegue a uno de los puntos de control.

Los puntos de interrupción son:

• antes y después de un acceso a disco.
• antes y después de acceso a TCP/IP.
• antes y después de llamadas a bloqueos.
• cada vez que finaliza la ejecución de una función PRG, codebloc o macro.
• cada 5000 ejecuciones de PCODE dentro de un mismo PRG.
  

Un thread no puede ser detenido amigablemente cuando esta:

• esperando por un mutex.
• esperando por conexión o datos de TCP/IP.
• mientras ejecuta funciones de C o de PRG compilado en C nativo, que quedan en un loop infinito o muy largo.
  

Killthread

En general hay que evitar el uso de esta función, siempre hay que tratar de finalizar los threads amigablemente.

Windows en su documentación indica que se evite su uso porque puede dejar incluso registros del kernel en estado incorrecto si al matar el thread el proceso esta ejecutando una función en modo kernel.
También la documentación dice que si el thread tiene un bloqueo, este podría no ser desbloqueado o no ser avisados otros threads de que el mutex esta disponible.

Nos encontramos ante la disyuntiva de matar o no a un thread que no responde.
Si no lo matamos, no podremos finalizar la ejecución.
Si lo matamos, quizás tampoco podamos finalizarla.

Por este motivo, la función Killthread inicialmente marca al thread que se le indica para que se cierre amigablemente, si ya esta marcado entonces sí se lo mata.

En conclusión. Hay que tratar por todos los medios de detener los threads amigablemente, para esto es necesario diseñar el código pensando en incluir puntos de interrupción o de control, si fuera necesario, para ayudar a finalizar el/los threads por las buenas.

Jointhread

Esta función se complementa con Stopthread permitiendo sincronizar la finalización del thread.
Se usa para esperar a que el thread indicado termine su ejecución.

Ejemplo:


Procedure Main()
Local thThread

cls
@8,10 say "Presione una tecla para parar el hilo"
thThread := StartThread( @Work() )
StartThread( @Join(), thThread )

inkey(0)
StopThread( thThread )
WaitForThreads()
Return

Procedure Work()
Local aStat := "\|/-", nStat := 1
do while .t.
  DevOut( "Trabajando ... "+aStat[nStat],,10,10)
  if ++nStat > 4
    nStat := 1
  endif
enddo
return

Procedure Join( thThread )
DevOut("Esperando que finalize el thread de trabajo",,12,10)
JoinThread( thThread )
DevOut("El thread de trabajo ha finalizado correctamente",,12,10)
return


Sobre la Reunión del 30 de Agosto 2008

En primer lugar agradecer el aliento y los comentarios recibidos. Tambien la presencia de todos los que se pudieron acercarce, en especial a aquellos que llegaron de lejos o de países vecinos.

Tuvimos un ambiente de total camaradería y literaltemte nos encerramos en la reunión, a tal punto que ni siquiera fuimos a almorzar (de 9 a 19). Es que había mucho interes.

Respecto de los temas tratados, en orden de presentación fueron:

• Paradigmas: una charla de Gustavo Valentin acerca de los paradigmas y como afectan la forma en que programamos.
  
• Objetos: Walter explicó los conceptos básicos de objetos partiendo de un mini programa que no los tenía y llevandolo en una serie de modificaciones a uno que sí los tenía.
  
• Impresoras Fiscales: Walter y Gustavo contaron que es y como funciona una impresora fiscal, la forma de comunicación y los programas que ellos utilizan. Luego se generó un debate entre los que estan necesitados de implementar impresora fiscal en sus programas y, con la donación de código por parte de softmagic quedaron en realizar una clase tanto para epson como para hasar.
  
• Hash: si bien no estaba en la agenda de la reunión surgió espontaneamente. Walter nos explicó como inicializar un hash y luego se generó un debate sobre los diferente lugares donde utilizarlo. Fué la frutilla del postre.
  
• Se habló de hacer la reunión mas seguido.

Multihilos en xHarbour - Parte V

Funciones CRITICAL
Una funcion marcada como critical tiene un mutex interno que automáticamente se activa y desactiva al ingresar y salir de dicha función.
La ventaja es que no hay que crear un mutex para hacer esta tarea, lo cual reduce y simplifica el código.
La desventaja es que bloquea a una sola función.

Pueden existir muchas funciones CRITICAL ejecutandose en paralelo, pero un sólo thread por vez ejecutara cada una de ellas.

Forma de uso:

CRITICAL FUNCTION SoloUno()
CRITICAL PROCEDURE SoloUno()
CRITICAL STATIC FUNCTION SoloUno()
CRITICAL STATIC PROCEDURE SoloUno()



// No superar 32768 porque HB_Random() solo genera
// hasta 32768 numeros diferentes

#define MAXVALUE 30000

//#define PARALELO

Static nCount := 0, lWork := .t.
Static nTime, lEnd := .f.

Function Main()
Local aValues := Array(MAXVALUE)
cls
StartThread(@generate(),aValues)
StartThread(@generate(),aValues)
StartThread(@process())
DevOut("Presione una tecla para finalizar...",,5,0)
do while lEnd .and. inkey(0.1) == 0
  lWork := .f.
enddo
WaitForThreads()
If lEnd
  Devout("Tiempo de trabajo="+str(nTime),,8,0)
Endif
Return nil

Procedure Generate(aValues)
Local nValue, lRet
Do While lWork
  nValue := Int(HB_Random(0,MAXVALUE))+1
#ifdef PARALELO
  If aValues[nValue] == nil
#endif
    lRet := AddUnique(aValues,nValue)
    If lRet
      Exit
    Endif
#ifdef PARALELO
  ElseIf nCount == MAXVALUE
    Exit
  Endif
#endif
Enddo
Return

Critical Function AddUnique(aValues,nValue)
If nCount == 0
  nTime := Seconds()
Endif
If aValues[nValue] == nil
  ++nCount
  aValues[nValue] := nCount
Endif
If nCount == MAXVALUE
  If !lEnd
    nTime := Seconds()-nTime
  Endif
  lEnd := .t.
  Return .t.
Endif
Return .f.

Procedure Process(nThread)
Local cPaso := "\|/-", nPaso := 1
do while nCount < MAXVALUE .and. lWork
  DevOut(nCount,,7,0)
  DevOut(cPaso[nPaso],,7,15)
  If ++nPaso > 4
    nPaso := 1
  Endif
  ThreadSleep(100)
enddo
DevOut(nCount,,7,0)
Return


El código del ejemplo tiene 2 modos de funcionamiento.
El modo 1 es tal como está escrito.
El modo 2 es "des-comentando" la línea #define PARALELO

En el modo 1, realiza lo siguiente:
Unos hilos en paralelo, generan unos números aleatorios y luego llaman a una función que controla si dicho número está cargado en un array.
Si no está, lo carga. Si ya existe, lo descarta.
Cuando el array está completo, cambia el estado de la bandera lEnd para que finalice el programa.

En el modo 2, realiza lo siguiente:
El proceso es similar, solo que cada hilo controla por su lado, si el número generado ya existe. Si no existe, llama a la función para que lo cargue. Si ya existe, lo descarta y vuelve a calcular otro número.
Puede darse el caso que el hilo 1 calcula un número y detecta que el número no existe y llama a la función, pero el hilo 2 pudo haber calculado el mismo número y haberse anticipado en la llamada a la función AddUnique, pero esto no es problema ya que cuando el hilo 1 llame a la función, el número será descartado sin problemas como en el modo 1.

Muestra el uso del procesador en modo 1.

Muestra el uso del procesador en modo 2.

Si en un procesador multicore, se nota que el modo 2 se ejecuta más rápidamente, ya que los 2 hilos pueden estar en forma paralela averiguando si el número obtenido existe o no en el array, y solo en el caso de no existir llamará a la función AddUnique para que lo agregue.
Si bien la ejecución es bastante rápida, podemos concluir que en el modo 2, inicialmente el uso de los 2 cores (uno por cada hilo) es de un 60%, pero a medida que se va llenando el array, el uso tiende al 100%.
Cómo es esto?
El hilo 1, calcula un número, lo busca en el array y no existe, asi que llama a AddUnique. Por su parte, el hilo 2, calcula otro número, lo busca en el array y tampoco existe, asi que tambien llama a AddUnique.
El tema es que los hilos deben turnarse para ejecutar AddUnique, ya que es una función CRITICAL.
Pero a medida que crece la densidad de uso del array, los números aleatorios comienzan a duplicarse y la tasa de encontrados aumenta y por consiguiente, aumenta los reintentos (volver a calcular el número aleatorio). Cada hilo por su cuenta reintenta varias veces la búsqueda hasta encontrar un número que no existe.
Pero debido a que los hilos no se están turnando, hacen más trabajo en menos tiempo y permite terminar la tarea antes.

En el modo 1, los hilos siempre se están turnando, con lo cual el uso del procesador permanece constante en un 60% durante todo el proceso. Al hacer menos trabajo en el mismo tiempo, la tarea demora más en terminarse.

Con estas explicaciones me estoy adelantando a temas que voy a tocar en próximas entregas, pero bien viene la reseña.
Al trabajar con mutex es importante saber cuando y cuanto proteger para evitar ser "sobreprotectores" y tener un código que funcione lentamente.

LetoDB rdd cliente servidor para xHarbour

Desde la soleada Mallorca leemos una muy interesante nota de BielSys (Gabriel Maimó "Biel") acerca de un proyecto Open Source de Alexander Kresin.

http://bielsys.blogspot.com/2008/07/letodb-rdd-cliente-servidor-para.html

Gracias BielSys, muy interesante tu blog.

Multihilos en xHarbour - Parte IV

En las entregas anteriores vimos como ejecutar varios hilos simultáneamente, lanzandolos uno a uno.
En esta oportunidad veremos un tema un poco más avanzado pero muy útil, los avisos o mensajes entre hilos.

Supongamos que estamos lanzando hilos para ejecutar un juego de una carrera.
Creamos un código que irá mostrando el avance de cada participante y un mutex nos permitirá identificar el orden de llegada de cada uno de ellos.

Lo primero que se nos ocurre es lanzarlos tal como vimos anteriormente y nos resulta un código como el del ejemplo 1.

NOTA: Los ejemplos contienen código que permite mostrar una velocidad pareja de ejecución independiente del procesador donde se ejecute, con el fin de facilitar
la visualización de la explicación.
De todas formas, debido a que el tiempo de ejecución es considerablemente alto (del orden de algunos segundos) una cantidad de cores menor a la cantidad de hilos a ejecutar, puede hacer que el ejemplo no siempre muestre el resultado que busco mostrar a pesar de algunas compensaciones de tiempo para evitarlo.


Static mtxGano
Static nGanador := 0, nPuesto := 1

Function Main()
Local nSec := Seconds() + 0.5, n
Local nCount := 0

cls
mtxGano := HB_MutexCreate()
do while Seconds() < nSec
   nCount++
enddo

For n := 1 To 5
   ThreadSleep(200) // Simula hacer otras tareas
   StartThread(@listos_ya(),n,int(nCount/50))
Next
WaitForThreads()

Procedure listos_ya( nHilo, nCount )
Local nMax := nCount * 60, nPos := 1, n

For n := 0 to nMax
   if n % nCount == 0
      DevOut(Replicate("-",nPos-1),,nHilo+5,1)
      DevOut(chr(2),,nHilo+5,nPos)
      nPos++
   endif
Next
DevOut(Replicate("-",nPos-1),,nHilo+5,1)
DevOut(chr(2),,nHilo+5,nPos)

HB_MutexLock(mtxGano)
If nGanador == 0
   nGanador := nHilo
   DevOut("GANADOR",,nHilo+5,65)
   nPuesto ++
Else
   DevOut(str(nPuesto,1,0)+" lugar",,nHilo+5,65)
   nPuesto++
Endif
HB_MutexUnlock(mtxGano)
Return


Ejemplo 1

Este ejemplo nos muestra una clara tendencia a que el jugador 1 gane la carrera.
Lo cual no es muy conveniente si estamos diseñando un juego.

Con lo cual necesitamos sincronizar el arranque.
Quizás lo visto en la entrega de Mutex nos sirva para mejorar el código, así que modificamos el código anterior para que cada hilo tenga un mutex y queden todos a la espera de la señal de arranque y nos queda algo como el ejemplo 2.


Static mtxGano
Static nGanador := 0, nPuesto := 1

Function Main()
Local nSec := Seconds() + 0.5, n
Local nCount := 0, aMutex := {}

cls
mtxGano := HB_MutexCreate()
do while Seconds() < nSec
   nCount++
enddo

For n := 1 To 5
   aadd(aMutex,HB_MutexCreate())
   HB_MutexLock( aMutex[-1] )
   StartThread(@listos_ya(),n,int(nCount/50),aMutex[-1])
Next
For n := 1 To 5
   ThreadSleep(100) // Simula hacer otras tareas
   HB_MutexUnlock(aMutex[n])
Next
WaitForThreads()

Procedure listos_ya( nHilo, nCount, mtxSincro )
Local nMax := nCount * 60, nPos := 1, n
HB_MutexLock(mtxSincro)
HB_MutexUnlock(mtxSincro)
For n := 0 to nMax
   if n % nCount == 0
      DevOut(Replicate("-",nPos-1),,nHilo+5,1)
      DevOut(chr(2),,nHilo+5,nPos)
      nPos++
   endif
Next
DevOut(Replicate("-",nPos-1),,nHilo+5,1)
DevOut(chr(2),,nHilo+5,nPos)

HB_MutexLock(mtxGano)
If nGanador == 0
   nGanador := nHilo
   DevOut("GANADOR",,nHilo+5,65)
   nPuesto ++
Else
   DevOut(str(nPuesto,1,0)+" lugar",,nHilo+5,65)
   nPuesto++
Endif
HB_MutexUnlock(mtxGano)
Return


Ejemplo 2

Si bien nuestros sentidos nos muestran una carrera más pareja, la realidad es que si tenemos una cantidad de cores mayor a la cantidad de hilos de ejecución, pasaría lo mismo que en el ejemplo 1.
Esto es así porque nuestro código es previsible y siempre pone en marcha los hilos en un orden prestablecido.
Cuanto mayor sea la cantidad de hilos, mayor será la diferencia entre la orden de arranque para el primer hilo y la orden de arranque para el último hilo.

Para sincronizar el arranque de uno o varios hilos, tenemos otra utilidad de la entidad Mutex. La pareja suscripción/notificación.
La idea de funcionamiento es que uno o varios hilos se inscriben o suscriben para recibir la orden o notificación de arraque.
Esta orden o notificación puede ser global o individual.
En el ejemplo 3 tenemos el código de la carrera que pondrá a todos los hilos a disposición del sistema operativo para que este los ponga en funcionamiento cuando existan los recursos necesarios para hacerlo.


Static mtxGano
Static nGanador := 0, nPuesto := 1

Function Main()
Local nSec := Seconds() + 0.5, n
Local nCount := 0, mtxSincro := HB_MutexCreate()

cls
mtxGano := HB_MutexCreate()
do while Seconds() < nSec
   nCount++
enddo

For n := 1 To 5
   StartThread(@listos_ya(),n,int(nCount/50),mtxSincro)
Next
ThreadSleep(200) // Simula hacer otras tareas
NotifyAll(mtxSincro)
WaitForThreads()

Procedure listos_ya( nHilo, nCount, mtxSincro )
Local nMax := nCount * 60, nPos := 1, n
Subscribe(mtxSincro)
For n := 0 to nMax
   if n % nCount == 0
      DevOut(Replicate("-",nPos-1),,nHilo+5,1)
      DevOut(chr(2),,nHilo+5,nPos)
      nPos++
   endif
Next
DevOut(Replicate("-",nPos-1),,nHilo+5,1)
DevOut(chr(2),,nHilo+5,nPos)

HB_MutexLock(mtxGano)
If nGanador == 0
   nGanador := nHilo
   DevOut("GANADOR",,nHilo+5,65)
   nPuesto ++
Else
   DevOut(str(nPuesto,1,0)+" lugar",,nHilo+5,65)
   nPuesto++
Endif
HB_MutexUnlock(mtxGano)
Return

Ejemplo 3

Las funciones de notificación y suscripción son las siguientes:

Subscribe( mutex, [timeout], [@event] ) -> info

Inscribe al thread para ser avisado de una notificación.
Si hay notificaciones pendientes, sale inmediatamente, si no, queda bloqueado a la espera de una notificación.
Se puede indicar un tiempo de espera, luego del cual se desbloqueará a pesar de no recibir ninguna notificación.
El thread que notifica, puede enviar una información en el momento de notificar la cual será recibida como valor de retorno de la función de suscripción.
Cuando se especifica un timeout, para saber si hubo o no notificación, se puede especificar un tercer parámetro por referencia, que retornará un valor lógico verdadero si ocurrió la notificación y falso en caso de finalizar por timeout.

SubscribeNow( mutex, [timeout], [@event] ) -> info

El funcionamiento es similar a la función Subscribe(), la única diferencia es que antes de comenzar, borra cualquier notificación pendiente.

Notify( mutex, [info] )

Emite una notificación. Si hay uno o más threads esperando, alguno de ellos es notificado y sale del estado de espera. Los demás threads suscriptos siguen esperando. No hay relación entre el orden de suscripción y el orden de notificación.
Si no hubiera threads en espera, la suscripción queda pendiente en una cola y se van asignando por orden a los threads que se vayan suscribiendo.
Es posible enviar información al thread que espera la suscripción, usando el segundo parámetro de la función.

NotifyAll( mutex, [info] )

Emite una notificación a todos los threads que esten en espera.
Si no hay threads en espera es como si nunca se hubiera ejecutado la función, no se agregan ni quitan notificaciones que esten en cola.
Es posible enviar información al thread que espera la suscripción, usando el segundo parámetro de la función.

Las suscripciones y notificaciones evitan que un proceso tenga que quedar en un loop preguntando cada cierto tiempo por alguna condición.
Si el loop es muy rápido consumirá mucho procesador, si el loop es muy lento se estaría reaccionando tardíamente a un cambio de estado.
Cuando un thread entra en suscripción y no hay notificaciones pendientes, se queda en un estado de espera sin consumir procesador.

Se les ocurren otras posibildades de uso de estas funciones ?
Espero sus comentarios.

Multihilos en xHarbour - Parte III

En la anterior entrega, vimos los problemas que ocurren cuando se incrementan o decrementan variables desde 2 o más hilos y no se protegen con un mutex.

Si ejecutan el último ejemplo de la entrega anterior, pero sin ejecutar la función ThreadSleep(10), casi con seguridad tendrán el siguiente error interno: Premature Pointer Release detected
El error ocurre mucho más frecuentemente con procesadores multicore porque verdaderamente están ejecutando el código en paralelo.

¿A qué se debe? Se debe justamente a que los contadores de referencia del puntero que mantiene al Mutex, se desincronizan porque no se actualizan correctamente, tal como se vió en el primer ejemplo de la última entrega.

¿Cómo solucionar este problema? Desde el 28/6/08 el CVS de xHarbour tiene la corrección aplicada y también en la distribución xHarbour 1.0 de PuertoSur tiene el problema corregido en todos los casos.

xHarbour 1.0 de PuertoSUR

Muchos lo han pedido y ya está disponible la distribución de xHarbour de PuertoSUR de la versión 1.0.
En estos momentos están disponibles para bajar las siguientes compilaciones:

xHarbour 1.0 compatible con FWH 2.4 para Borland 5.5
xHarbour 1.0 compatible con FWH 2.5/2.6 para Borland 5.5
xHarbour 1.0 compatible con FWH 2.7 y sup para Borland 5.5

Esta distribución como anteriores, se provee en forma de instalador que busca la ubicación del compilador de Borland, lo configura y configura los archivos para compilar y enlazar aplicaciones simples.

La distribución incluye junto con las librerías de xHarbour, algunas librerías de contribuciones y algunas librerías extras, además de algunos arreglos posteriores a la liberación:

rddads.lib compatible con Advantage 8.x
mysql3.lib compatible con MySql 3.x
mysql5.lib compatible con MySql 5.x
hbcomm.lib librería de comunicaciones serie con funciones compatibles en parámetros y funcionalidad a las de FW, para paliar la falta de esta funcionalidad en algunas versiones de FWH.