The core protocol of WoopChain
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
woop/aws-experiment-launch/experiment/commander/main.go

211 lines
4.7 KiB

package main
import (
"bufio"
"flag"
"fmt"
"io"
"log"
"net"
"net/http"
"os"
"strings"
"time"
"github.com/simple-rules/harmony-benchmark/aws-experiment-launch/experiment/utils"
"github.com/simple-rules/harmony-benchmark/configr"
)
type commanderSetting struct {
ip string
port string
configURL string
configs [][]string
6 years ago
}
type sessionInfo struct {
id string
uploadFolder string
}
var (
setting commanderSetting
6 years ago
session sessionInfo
)
6 years ago
const (
DistributionFileName = "distribution_config.txt"
)
6 years ago
func readConfigFile() [][]string {
if err := utils.DownloadFile(DistributionFileName, setting.configURL); err != nil {
panic(err)
}
6 years ago
6 years ago
if result, err := configr.ReadConfigFile(DistributionFileName); err != nil {
panic(err)
} else {
return result
6 years ago
}
}
func handleCommand(command string) {
args := strings.Split(command, " ")
if len(args) <= 0 {
return
}
switch cmd := args[0]; cmd {
case "config":
6 years ago
setting.configs = readConfigFile()
if setting.configs != nil {
log.Printf("The loaded config has %v nodes\n", len(setting.configs))
} else {
6 years ago
log.Println("Failed to read config file")
}
case "init":
6 years ago
session.id = time.Now().Format("150405-20060102")
// create upload folder
session.uploadFolder = fmt.Sprintf("upload/%s", session.id)
err := os.MkdirAll(session.uploadFolder, os.ModePerm)
if err != nil {
log.Println("Failed to create upload folder", session.uploadFolder)
return
}
6 years ago
log.Println("New session", session.id)
6 years ago
dictateNodes(fmt.Sprintf("init %v %v %v %v", setting.ip, setting.port, setting.configURL, session.id))
case "ping", "kill", "log", "log2":
6 years ago
dictateNodes(command)
default:
6 years ago
log.Println("Unknown command")
}
}
func config(ip string, port string, configURL string) {
setting.ip = ip
setting.port = port
setting.configURL = configURL
}
6 years ago
func dictateNodes(command string) {
resultChan := make(chan int)
for _, config := range setting.configs {
ip := config[0]
port := "1" + config[1] // the port number of solider is "1" + node port
addr := strings.Join([]string{ip, port}, ":")
6 years ago
go func(resultChan chan int) {
resultChan <- dictateNode(addr, command)
}(resultChan)
}
count := len(setting.configs)
res := 0
6 years ago
for ; count > 0; count-- {
res += <-resultChan
}
6 years ago
log.Printf("Finished %s with %v nodes\n", command, res)
}
func dictateNode(addr string, command string) int {
// creates client
conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
if err != nil {
log.Println(err)
return 0
}
defer conn.Close()
// send command
_, err = conn.Write([]byte(command))
if err != nil {
log.Printf("Failed to send command to %s", addr)
return 0
}
6 years ago
// log.Printf("Send \"%s\" to %s", command, addr)
// read response
buff := make([]byte, 1024)
6 years ago
if n, err := conn.Read(buff); err == nil {
6 years ago
received := string(buff[:n])
// log.Printf("Receive from %s: %s", addr, buff[:n])
if strings.Contains(received, "Failed") {
return 0
} else {
return 1
}
6 years ago
}
6 years ago
return 0
}
func handleUploadRequest(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
// reject non-post requests
jsonResponse(w, http.StatusBadRequest, "Only post request is accepted.")
return
}
reader, err := r.MultipartReader()
if err != nil {
jsonResponse(w, http.StatusBadRequest, err.Error())
return
}
for {
part, err := reader.NextPart()
if err == io.EOF {
break
}
6 years ago
dst, err := os.Create(fmt.Sprintf("%s/%s", session.uploadFolder, part.FileName()))
log.Println(part.FileName())
if err != nil {
jsonResponse(w, http.StatusInternalServerError, err.Error())
return
}
defer dst.Close()
if _, err := io.Copy(dst, part); err != nil {
jsonResponse(w, http.StatusInternalServerError, err.Error())
return
}
}
}
func jsonResponse(w http.ResponseWriter, code int, message string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
fmt.Fprint(w, message)
log.Println(message)
}
func serve() {
http.HandleFunc("/upload", handleUploadRequest)
err := http.ListenAndServe(":"+setting.port, nil)
if err != nil {
log.Fatalf("Failed to setup server! Error: %s", err.Error())
}
6 years ago
log.Printf("Start to host upload endpoint at http://%s:%s/upload\n", setting.ip, setting.port)
}
func main() {
ip := flag.String("ip", "127.0.0.1", "The ip of commander, i.e. this machine")
6 years ago
port := flag.String("port", "8080", "The port which the commander uses to communicate with soldiers")
6 years ago
configURL := flag.String("config_url", "https://s3-us-west-2.amazonaws.com/unique-bucket-bin/distribution_config.txt", "The config URL")
7 years ago
flag.Parse()
config(*ip, *port, *configURL)
go serve()
scanner := bufio.NewScanner(os.Stdin)
for true {
log.Printf("Listening to Your Command:")
if !scanner.Scan() {
break
}
handleCommand(scanner.Text())
}
}