library(readr)
library(parallel)
library(bench)



### FUNCIONES ####
## leer_trozo: permite cargar una parte de las filas de la base de datos contenida en archivo, donde:

# - archivo: ruta completa o parcial en disco de la base de datos.
# - parte: el numero de porcion que se leera.
# - longitud: numero de filas de cada parte.
# 
leer_trozo = function(archivo,parte,longitud){
   
  trozo = read_csv(archivo, skip = (parte-1)*longitud+1,n_max = longitud,col_names = FALSE,
                   col_types = cols())
  return(trozo)
  
}

## cuenta: Contabiliza cuantos elementos no son NAs
cuenta = function(x) sum(!is.na(x))

## funCuenta: Cuantos elementos diferentes de NA hay en cada columna del trozo leido
funCuentaTiempo = function(archivo,parte,longitud,quitar) {
  
  t1 = Sys.time()
  M = leer_trozo(archivo,parte,longitud)
  M = select(M, -quitar)
  k = apply(M,2,cuenta)
  t2 = Sys.time()
  d = as.numeric(difftime(t2, t1), units="secs")
  d1 = data.frame(t1,t2,d)
  return(d1)
}


#### PARAMETROS ####
k = 10 # total de partes a ser leidas
lon = 100000 # longitud de cada parte
# Posición de las columnas de la base de datos que no son númericas, 
# sino que contienen una etiqueta, y no deben incluirse en los cálculos
nonumer = c(1,2,54,55,61,63,156,158,159,162,168)
# Ruta relativa del archivo que contiene la base de datos
archivo1 = "Desktop/R ZORA Ejemplos/CLASE 1/train_data.csv"

# Se probará con clusters de tamaño desde 1 a max_nodos
max_nodos = 10

tiempos_n = list(max_nodos)

for(num_nodo in 1:max_nodos){
#### CASOS ####
######  Secuencial  #######
ben_sec = bench::mark(
  lapply(1:k, funCuentaTiempo, archivo=archivo1,longitud = lon, quitar = nonumer), 
  iterations = 1
)  


######  Paralelo  #######
cl <- makeCluster(num_nodo)
clusterExport(cl,c('leer_trozo','read_csv','select','cuenta','cols'))

ben_par = bench::mark(
  clusterApply(cl, 1:k, funCuentaTiempo, archivo=archivo1,longitud = lon, quitar = nonumer),
  iterations = 1
)  

stopCluster(cl)


#####  Paralelo LB #######
cl <- makeCluster(num_nodo)
clusterExport(cl,c('leer_trozo','read_csv','select','cuenta','cols'))

ben_parLB = bench::mark(
  
  clusterApplyLB(cl, 1:k, funCuentaTiempo, archivo=archivo1,longitud = lon, quitar = nonumer),
  iterations = 1
)  

stopCluster(cl)


tiempos_n[[num_nodo]] = list( ben_sec$total_time, ben_par$total_time, ben_parLB$total_time)



}


tiempos_nf = rbindlist(tiempos_n)

colnames(tiempos_nf) =  c('Secuencial','Paralelo','Paralelo_LB')


tiempos_nf


######## Graficos #########
maxi = max(tiempos_nf)*1.5 # tomamos el mayor tiempo de todos

# secuencial
plot(1:max_nodos,tiempos_nf$Secuencial,lwd=2,col='tomato1',xlab='Nodos/Operaciones',ylab='Tiempo en segundos',main="Desempeño Secuencial vs Paralelo",
     ylim=c(0,maxi) ,type='l')
points(1:max_nodos,tiempos_nf$Secuencial,pch=16,col='tomato4',cex=1.6)

# paralelo
lines(1:max_nodos,tiempos_nf$Paralelo,lwd=2,col='dodgerblue2')
points(1:max_nodos,tiempos_nf$Paralelo,pch=16,col='dodgerblue4',cex=1.6)

# paralelo LB
lines(1:max_nodos,tiempos_nf$Paralelo_LB,lwd=2,col='seagreen1')
points(1:max_nodos,tiempos_nf$Paralelo_LB,pch=16,col='seagreen3',cex=1.6)

legend(2,maxi,pch=c(16,16,16),col=c('tomato1','dodgerblue2','seagreen1'),cex=0.8,legend=c('Secuencial','Paralelo','Paralelo LB'))


# Aceleracion

adif = tiempos_nf$Secuencial/tiempos_nf$Paralelo
adiflb = tiempos_nf$Secuencial/tiempos_nf$Paralelo_LB

maxa = max(c(adif,adiflb))*1.1
mina = min(c(adif,adiflb))-0.1

plot(1:max_nodos,adif,lwd=2,col='dodgerblue2',xlab='Nodos',ylab='Aceleración',main="Gráfico de Aceleración",
     ylim = c(mina,maxa),type='l')
points(1:max_nodos,adif,pch=16,col='dodgerblue4',cex=1.6)  


lines(1:max_nodos,adiflb,lwd=2,col='seagreen1')
points(1:max_nodos,adiflb,pch=16,col='seagreen3',cex=1.6)  

legend(1,maxa/1.1,pch=c(16,16),col=c('dodgerblue2','seagreen1'),cex=0.8,legend=c('Paralelo','Paralelo LB'))  

# Eficiencia

efic = (adif/(1:max_nodos))*100
eficlb = (adiflb/(1:max_nodos))*100

maxe = max(c(efic,eficlb))+15

plot(1:max_nodos,efic,lwd=2,col='dodgerblue2',xlab='Nodos',ylab='Eficiencia',main="Gráfico de Eficiencia",
     ylim = c(0,maxe),type='l')
points(1:max_nodos,efic,pch=16,col='dodgerblue4',cex=1.6)  


lines(1:max_nodos,eficlb,lwd=2,col='seagreen1')
points(1:max_nodos,eficlb,pch=16,col='seagreen3',cex=1.6)  


legend(1,20,pch=c(16,16),col=c('dodgerblue2','seagreen1'),cex=0.8,legend=c('Paralelo','Paralelo LB'))  



abline(h=50,col="black",lty=3,lwd=3) # linea que marca el 50%
text(1.05,51,'50%',col='black',pos = 3)