Analysis.Rmd

---
title: "WORKFLOW_NAME Statistics"
output: pdf_document
---

```{r echo=FALSE}
library(scales)
library(reshape2)
library(ggplot2)
library(plyr)
```

```{r echo=FALSE, cache=TRUE}
get_suffix <- function(x) {
  tail(strsplit(as.character(x),"[.]")[[1]], n=1)
}
```

```{r echo=FALSE, cache=TRUE}
workers <- read.csv('csv_files/worker_nodes.csv', header=TRUE, sep=',')

# Remove unit and convert to numeric values
workers$MIPS <-as.numeric(sub("Mf", "", workers$MIPS))

# Add a column to store the suffix of local SE
workers$SESuffix<-sapply(workers$CloseSE, get_suffix)
```

```{r echo=FALSE, cache=TRUE}
transfers <- read.csv('csv_files/file_transfer.csv', header = TRUE, sep=',')

# Remove entries with 0 file size, if any
transfers = transfers[transfers$FileSize!=0,]

# Simplify the job IDs
transfers$JobId <- transfers$JobId-min(transfers$JobId)

# Convert durations from milliseconds to seconds 
transfers$Time<-transfers$Time/1000

# Add columns to store the suffixes of source and destination SE
transfers$src_suffix<-sapply(transfers$Source, get_suffix)
transfers$dst_suffix<-sapply(transfers$Destination,  get_suffix)
```

```{r echo=FALSE, cache=TRUE}
upload_tests <- transfers[transfers$UpDown == 0,]
uploads <- transfers[transfers$UpDown == 1,]
downloads <- transfers[transfers$UpDown == 2,]
```

```{r echo=FALSE, cache=TRUE}
usedSE <- merge(uploads[names(uploads) %in% c("Source", "Destination","src_suffix", "dst_suffix")],
                workers[names(workers) %in% c("Name", "SiteName")],
                by.x="Source", by.y="Name")
names(usedSE) <- c("Name", "CloseSE", "Country", "SESuffix","SiteName")
```

```{r echo=FALSE, cache=TRUE}
bandwidths_raw <-read.csv('csv_files/se_bandwidth.csv', header=TRUE, sep=',')

# Melt the data frame to have values in a plottable layout
bandwidths <- melt(bandwidths_raw, id.vars="SE", value.name="Bandwidth")

# get rid off of 0 bandwidth, they mean uncomputed values
bandwidths <- bandwidths[bandwidths$Bandwidth > 0,]

# splitting the [AVG/MAX]_[UP/DOWN/ALL] columns to improve layout
bandwidths <- cbind(bandwidths,t(sapply(bandwidths$variable, 
                                        function(x) strsplit(as.character(x),"[_]")[[1]])))
colnames(bandwidths)[c(4,5)] = c("Type", "Direction")

# Discard the "ALL" case
bandwidths <- bandwidths[bandwidths$Direction !="ALL",]
bandwidths$SE <- as.character(bandwidths$SE)
bandwidths$SE <- factor(bandwidths$SE, levels=sort(unique(bandwidths$SE)), ordered=TRUE)
# And thus remove the "variable" column
bandwidths = bandwidths[!names(bandwidths) %in% "variable"]
```

```{r echo=FALSE, cache=TRUE}
timings <- read.csv('csv_files/db_dump.csv', header = TRUE, sep = ' ') 
# Simplify the job IDs
timings$JobId <- timings$JobId-min(timings$JobId)
timings$JobSeq <- as.numeric(row.names(timings))

# Identify the merge job
mergeId <- timings[timings$Command=="merge.sh",]$JobId
mergeSeq <- timings[timings$Command=="merge.sh",]$JobSeq

queuing_step <- timings[names(timings) %in% c("JobSeq", "CreationTime", "DownloadStartTime")]
names(queuing_step)<-c("Start","End", "JobId")
queuing_step$Step<-"Queuing"

download_step <- timings[names(timings) %in% c("JobSeq", "DownloadStartTime")]
names(download_step)<-c("Start","JobId")
download_step$End <- timings$DownloadStartTime+timings$DownloadDuration
download_step$Step <- "Download"

computing_step <- timings[names(timings) %in% c("JobSeq", "ComputeStartTime")]
names(computing_step)=c("Start", "JobId")
computing_step$End <- timings$ComputeStartTime+timings$ComputeDuration
computing_step$Step <-"Computing"

upload_step<-timings[names(timings) %in% c("JobSeq", "UploadStartTime")]
names(upload_step)<-c("Start", "JobId")
upload_step$End <- timings$UploadStartTime+timings$UploadDuration
upload_step$Step <-"Upload"

gantt <- rbind(queuing_step,download_step,computing_step, upload_step)
```

```{r echo=FALSE, cache=TRUE}
binwidth=10
find_bins <- function(origin, s, e, end) {
  c(rep(0,(s-origin)/binwidth), 
    rep(1, (e-s)/binwidth), 
    rep(0,(end-e)/binwidth))
}

df <-gantt[gantt$Step %in% c("Download", "Upload"),]
origin <- floor(min(df$Start)/binwidth)*binwidth
end<- ceiling(max(df$End)/binwidth)*binwidth
bins = ddply(df,.(JobId,Step), 
             function(x) find_bins(origin, 
                                   floor(min(x$Start)/binwidth)*binwidth,
                                   ceiling(max(x$End)/binwidth)*binwidth, end))
bins=bins[,-1]
names(bins) = c("Step", seq(origin, end-binwidth, by=binwidth))

concurrency = melt(ddply(bins,.(Step), colwise(sum)), 
                   id.vars="Step", value.name="Count", variable.name="Start")
concurrency=concurrency[concurrency$Count>0,]
concurrency$Start <- as.integer(as.character(concurrency$Start))
```

## Gantt chart of the workflow execution 

```{r echo=FALSE, fig.height=8.5}
ggplot(gantt)+ geom_segment(aes(x=Start, xend=End, y=JobId, yend=JobId, color=Step), size=1.25) + 
  geom_point(data=timings, aes(x=(DownloadStartTime+TotalDuration), y=JobSeq)) +
  scale_color_discrete(breaks=c("Queuing", "Download", "Computing", "Upload"), name="") +
  scale_y_reverse() + 
  theme(legend.position="top") + xlab("Time (in seconds)") + ylab("Jobs")
```

\newpage

## Characterization of the Worker Nodes used for the execution of the workflow
```{r echo=FALSE, fig.height=4.25}
ggplot(workers, aes(x=SiteName, fill=factor(Core))) + geom_bar() + scale_fill_discrete(name="#Cores") + 
  facet_wrap(~Country, scales="free_x", nrow=1) + theme(axis.text.x = element_text(angle = 50, hjust = 1)) + 
  xlab("Grid Site") + ylab("Number of worker nodes") + theme(legend.position="top")
```

```{r echo=FALSE, fig.height=4.25}
ggplot(workers, aes(x=round(MIPS, -2), fill=factor(SiteName))) + geom_bar(binwidth=100) +
  xlab("BogoMIPS") + ylab("Number of worker nodes")+ scale_fill_discrete(name="Grid Site") +
  facet_wrap(~Country) + theme(axis.text.x = element_text(angle = 50, hjust = 1)) +
  theme(legend.position="top") + guides(fill = guide_legend(nrow = 2)) 
```

\newpage

### Distribution of queuing time by grid site

```{r echo=FALSE, fig.height=4.5}
ggplot(timings, aes(x=Site, y=QueuingDuration))+ geom_boxplot() + 
  theme(axis.text.x = element_text(angle = 50, hjust = 1)) + 
  ylab("Queuing Time (in seconds)") + xlab("") +
  stat_summary(fun.y=mean, geom="line", aes(group=1, color="red")) + guides(color=FALSE)
```

### CloseSE vs. Grid Site by country

```{r echo=FALSE, fig.height=4.5}
ggplot(workers, aes(x=SiteName, y=CloseSE)) + 
  geom_point(aes(color="Declared"), size=2.5) + 
  geom_point(data=usedSE, aes(color="Used"), size=2.5, 
             position=position_jitter(width=0.2,height=0.1)) + 
  scale_color_discrete(breaks=c("Declared", "Used"), labels=c("Declared", "Used"), name="") +
  facet_grid(SESuffix~Country, scales="free") + xlab("") + ylab("Prefered Storage Element") +
  theme(axis.text.x = element_text(angle = 50, hjust = 1)) + 
  theme(legend.position="top")
```

\newpage

## Characterization of the File Transfers done during the execution of the workflow

### Distribution of Upload Test durations (in milliseconds)
```{r echo=FALSE, cache=TRUE}
summary(upload_tests$Time)
```

### By Destination (darkness represents frequency)
```{r echo=FALSE,fig.height=7.5}
ggplot(upload_tests, aes(x=Time, y=Destination)) + geom_point(alpha=0.2, size=3) +
  xlab ("Duration (in seconds)") + facet_grid(dst_suffix~., scales="free")
```

\newpage

### Distribution of GATE download durations (in seconds)
```{r echo=FALSE, cache=TRUE}
summary(downloads[downloads$JobId != mergeId,]$Time)
```

### By Source for each file size
```{r echo=FALSE,fig.height=8}
ggplot(downloads[downloads$JobType == "gate",], aes(x=Source, y=Time, fill=factor(FileSize))) + 
  geom_boxplot() + facet_grid(FileSize~SiteName, scales="free", drop=TRUE) +
  xlab ("") + ylab ("Duration (in seconds)") + 
  theme(axis.text.x = element_text(angle = 50, hjust = 1)) + 
  guides(fill=FALSE)
```

\newpage

### Distribution of GATE upload durations (in seconds)
```{r echo=FALSE}
summary(uploads[uploads$JobId != mergeId,]$Time)
```

```{r echo=FALSE,fig.height=7.5}
ggplot(uploads[uploads$JobId != mergeId,], aes(x=FileSize, y=Time)) + geom_point() + 
  facet_wrap(~Destination, ncol=1, scales = "free_y") + 
  ylab ("Duration (in seconds") + xlab ("File Size (in Bytes)")

```

### Distribution of GATE upload sizes (in Bytes)
```{r echo=FALSE}
summary(uploads[uploads$JobId != mergeId,]$FileSize)
```

\newpage

### Distribution of Merge download durations (in seconds)
```{r echo=FALSE}
# skip the first two downloads (wrapper and input files)
summary(tail(downloads[downloads$JobId == mergeId,], -2)$Time)
```

```{r echo=FALSE,fig.height=7.5}
# skip the first two downloads (wrapper and input files)
ggplot(tail(downloads[downloads$JobId == mergeId,], -2), aes(x=FileSize, y=Time)) + geom_point() + 
  facet_wrap(~Source, ncol=1, scales = "free_y") + 
  ylab ("Duration (in seconds") + xlab ("File Size (in Bytes)")

```

### Distribution of Merge download sizes (in Bytes)
```{r echo=FALSE}
# skip the first two downloads (wrapper and input files)
summary(tail(downloads[downloads$JobId == mergeId,], -2)$FileSize)
```

## Statistics on SE connectivity
```{r echo=FALSE}
downloads$Bandwidth <- (downloads$FileSize/(downloads$Time-0.99))/1000
uploads$Bandwidth <- (uploads$FileSize/(uploads$Time-0.99))/1000
df = downloads[names(downloads) %in% c("Source", "Bandwidth")]
names(df) <- c("SE", "Bandwidth")
df$UpDown <-"Uplink"
df2 = uploads[names(uploads) %in% c("Destination", "Bandwidth")]
names(df2) <- c("SE", "Bandwidth")
df2$UpDown <-"Downlink"
                    
individual_bw = rbind(df,df2)
```
### A view on bandwidth distribution
```{r echo=FALSE, fig.height=8.5}
ggplot(individual_bw, aes(x=SE, y=Bandwidth)) + 
  geom_boxplot() + 
  facet_wrap(~UpDown, scales="free") + scale_y_log10() +
  ylab ("Bandwidth (in KBps)") + xlab ("Storage Element")+ 
  theme(axis.text.x = element_text(angle = 50, hjust = 1))
```

### Distribution of bandwidth values for uploads to SE 

```{r echo=FALSE, fig.height=4.25, message=FALSE}
ggplot(individual_bw[individual_bw$UpDown == "Downlink",], aes(x=Bandwidth)) + 
  geom_histogram() + scale_x_log10()+ facet_wrap(~SE) +
  xlab ("Bandwidth (in KBps)") + ylab ("Count")+ 
  theme(axis.text.x = element_text(angle = 50, hjust = 1))
```

### Distribution of bandwidth values for downloads from SE

```{r echo=FALSE, fig.height=4.25, message=FALSE}
ggplot(individual_bw[individual_bw$UpDown == "Uplink",], aes(x=Bandwidth)) + 
  geom_histogram() + scale_x_log10()+ facet_wrap(~SE) +
  xlab ("Bandwidth (in KBps)") + ylab ("Count")+ 
  theme(axis.text.x = element_text(angle = 50, hjust = 1))
```

\newpage

### Bandwidths as computed during the log extraction

```{r echo=FALSE, fig.height=9}
ggplot(bandwidths, aes(x=SE, y=Bandwidth,  shape=Direction)) + 
  geom_point(size=2.5, color=hue_pal()(2)[2]) + scale_y_log10() + 
  facet_grid(Type~., scales="free") +
  ylab ("Bandwidth (in KBps)") + xlab ("Storage Element")+ 
  scale_shape_manual(breaks=c("UP","DOWN"), labels=c("From", "To"), values=c(25,17))+
  theme(axis.text.x = element_text(angle = 50, hjust = 1))
```

### Number of concurrent transfers (by windows of 10s)

```{r echo=FALSE, fig.height=9}
ggplot(concurrency, aes(x=Start, y=factor(Count), fill=Count)) + 
  geom_bar(stat="identity") + 
  scale_fill_continuous(low="green", high="red", name="") +
  facet_grid(Step~., margins = TRUE) + guides(fill=FALSE) +
  xlab("Time (in seconds)") + ylab("Number of Concurrent Transfers")
```