Go: Programação Distribuída

De Aulas

TCP Cliente

Efetua uma conexão TCP com um servidor. Aguarda o usuário digitar mensagens e dar ENTER. Para saír, basta digitar EXIT. 

package main

import (
	"bufio"
	"fmt"
	"net"
	"os"
	"strings"
)

func main() {
	arguments := os.Args     // Pega os argumentos da linha de comando
	if len(arguments) == 1 { // Se não tiver argumentos retorna erro
		fmt.Println("Enter with arguments host:port.")
		return
	}

	// Usa os argumentos e se conecta ao servidor host:port
	c, err := net.Dial("tcp", arguments[1])
	if err != nil {
		fmt.Println(err)
		return
	}
	defer c.Close()
	fmt.Print("Digit your message and press key ENTER to send.")
	fmt.Println("To exit, digit EXIT and press key emter.")
	for {
		reader := bufio.NewReader(os.Stdin) // Prepara o buffer de leitura
		fmt.Print("MSG: ")
		text, _ := reader.ReadString('\n') // Le um texto do teclado
		fmt.Fprintf(c, text+"\n")          // Envia o texto pela conexão

		message, _ := bufio.NewReader(c).ReadString('\n') // Aguarda resposta do servidor
		fmt.Print("RCV: " + message)
		// Se a resposta for EXIT, fecha a conexão e o cliente
		if strings.ToUpper(strings.TrimSpace(string(text))) == "EXIT" {
			fmt.Println("TCP client exiting...")
			return
		}
	}
}

TCP Servidor Simples

Inicia o servidor e fica aguardando um cliente. Recebe as mensagens do cliente. Quando o cliente envia EXIT, fecha o servidor e a conexão. 

package main

import (
	"bufio"
	"fmt"
	"net"
	"os"
	"strings"
	"time"
)

func main() {
	arguments := os.Args     // Pega a porta como argumento da linha de comando
	if len(arguments) == 1 { // Se não for passado a porta dá erro
		fmt.Println("Enter with port number in argument")
		return
	}
	l, err := net.Listen("tcp", ":"+arguments[1]) // Inicia a conexão TCP na porta
	if err != nil {
		fmt.Println(err)
		return
	}
	defer l.Close() // Ao final, fecha a conexão.
	fmt.Println("TCP Server initialized at port", arguments[1])

	c, err := l.Accept() // Fica aguardando um cliente se conectar
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println("Client connected...")

	for { // Laço eterno
		// Recebe informações no buffer de leitura
		netData, err := bufio.NewReader(c).ReadString('\n')
		if err != nil {
			fmt.Println(err)
			return
		}
		// Se for EXIT, fecha o sevidor, mas fecha a conexão antes
		if strings.ToUpper(strings.TrimSpace(string(netData))) == "EXIT" {
			fmt.Println("TCP Server terminated...")
			return
		}

		// Mostra a mensagem na tela e envia de volta o horário
		fmt.Print(": ", string(netData))
		t := time.Now()
		myTime := t.Format(time.RFC3339) + "\n"
		c.Write([]byte(myTime)) // Escreve no buffer de escrita para o cliente
	}
}

TCP Servidor com múltiplas conexões

Observe que agora temos um servidor diferente. No laço for do main ficamos aguardando um cliente se conectar. Quando houver uma conexão, essa conexão é transferida para uma função child que é chamada concorrentemente como gorotine e continua o laço, aguardando uma nova conexão.

Cada gorotine, ora chamada child, vai ficar responsável por gerenciar uma conexão específica. Quando o cliente mandar a palavra exit, não fecha o servidor, apenas aquela conexão.

Dessa forma, nosso servidor pode trabalhar com múltiplas conexões simultâneas, respondendo concorrentemente há vários clientes.

Veja que no código, quando um cliente manda uma mensagem, escrevemos qual o endereço da conexão para identificar a mensagem e diferenciá-la das mensagens dos outros clientes. 

package main

import (
	"bufio"
	"fmt"
	"net"
	"os"
	"strings"
	"time"
)

func child(conn net.Conn) {
	addr := conn.RemoteAddr() // Pega o endereço do cliente
	fmt.Println(addr, ": connected...")
	defer fmt.Println(addr, ": disconnected...")
	defer conn.Close() // Ao final, fecha a conexão

	for {
		// Recebe informações no buffer de leitura
		netData, err := bufio.NewReader(conn).ReadString('\n')
		if err != nil {
			fmt.Println(err)
			return
		}
		// Se for EXIT, fecha fecha a conexão
		if strings.ToUpper(strings.TrimSpace(string(netData))) == "EXIT" {
			fmt.Println(addr, ": exit...")
			return
		}
		// Mostra a mensagem na tela e envia de volta o horário
		fmt.Print(addr, " : ", string(netData))
		t := time.Now()
		myTime := t.Format(time.RFC3339) + "\n"
		conn.Write([]byte(myTime)) // Escreve no buffer de escrita para o cliente
	}
}

func main() {
	arguments := os.Args     // Pega a porta como argumento da linha de comando
	if len(arguments) == 1 { // Se não for passado a porta dá erro
		fmt.Println("Enter with port number in argument")
		return
	}
	l, err := net.Listen("tcp", ":"+arguments[1]) // Configura uma porta TCP
	if err != nil {
		fmt.Println(err)
		return
	}
	defer l.Close() // Ao final, fecha a conexão.
	fmt.Println("TCP Server initialized at port", arguments[1])

	for {
		c, err := l.Accept() // Aguarda o próximo cliente se conectar
		if err != nil {
			fmt.Println(err)
			return
		}
		go child(c) // Passa a conexão para uma gorotine gerenciar
	}
}

Daria pra fazer um chat de grupo com isso, não? É um bom exercício.

UDP Cliente

Veja que a conexão com um servidor UPD é diferente, então o cliente UDP também precisa ser específico pra isso. Por isso, não tente usar um cliente TCP para se conectar a um servidor UDP, ou melhor, tente pra ver o que acontece (¬‿¬)

Veja que aqui temos uma troca de mensagens via buffer de bytes. 

package main

import (
	"bufio"
	"fmt"
	"net"
	"os"
	"strings"
)

func main() {
	arguments := os.Args
	if len(arguments) == 1 {
		fmt.Println("Please provide a host:port string")
		return
	}

	s, err := net.ResolveUDPAddr("udp4", arguments[1])
	c, err := net.DialUDP("udp4", nil, s)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println("The UDP server is", c.RemoteAddr().String())
	defer c.Close()

	for {
		reader := bufio.NewReader(os.Stdin)
		fmt.Print(">> ")
		text, _ := reader.ReadString('\n')
		data := []byte(text + "\n")
		_, err = c.Write(data)
		if strings.ToUpper(strings.TrimSpace(string(data))) == "EXIT" {
			fmt.Println("Exiting UDP client!")
			return
		}
		if err != nil {
			fmt.Println(err)
			return
		}
		buffer := make([]byte, 1024)
		n, _, err := c.ReadFromUDP(buffer)
		if err != nil {
			fmt.Println(err)
			return
		}
		fmt.Println("Reply:", string(buffer[0:n]))
	}
}

UDP Servidor

E então, nosso servidor UDP. É muito parecido com o TCP, mas configurado de forma diferente e agora trabalha com mensagens na forma de buffer de bytes. 

package main

import (
	"fmt"
	"math/rand"
	"net"
	"os"
	"strconv"
	"strings"
	"time"
)

func main() {
	arguments := os.Args
	if len(arguments) == 1 {
		fmt.Println("Please provide a port number!")
		return
	}
	s, err := net.ResolveUDPAddr("udp4", ":"+arguments[1])
	if err != nil {
		fmt.Println(err)
		return
	}
	conn, err := net.ListenUDP("udp4", s)
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Close()
	buffer := make([]byte, 1024)
	rand.Seed(time.Now().Unix())

	for {
		/*
			Veja que estamos trabalhando com leitura em buffer, em bytes.
			A função ReadFromUDP retorna também o tamanho da mensagem que
			estamos recebendo no buffer e transformamos em string
		*/
		n, addr, err := conn.ReadFromUDP(buffer)
		fmt.Print(addr, " -> ", string(buffer[0:n-1]))

		// Se a mensagem for exit, fechamos a conexão e o servidor
		if strings.ToUpper(strings.TrimSpace(string(buffer[0:n]))) == "EXIT" {
			fmt.Println("Exiting UDP server!")
			return
		}

		data := []byte(strconv.Itoa(rand.Intn(1000)))
		fmt.Println(addr, "<-", string(data))
		_, err = conn.WriteToUDP(data, addr)
		if err != nil {
			fmt.Println(err)
			return
		}
	}
}

Exemplo de Computação Distribuída

Vamos usar como exemplo um programa que calcula o fatorial de um número, ou seja: 

n! = 1 * 2 * 3 * ... * n

Contudo, vamos dividir o problema em partes e mandar para servidores executar esses pedaços de cálculos para por fim calcular o resultado final no programa principal.

Vamos supor que temos o número 15 e queremos dividir em 4 subprocessos. Também temos 3 servidores disponíveis para executar o cálculo. Veja que um dos servidores vai receber dois pacotes para calcular.

Para isso, dividimos os números em 4 blocos

  • de 1 a 4
  • de 5 a 8
  • de 9 a 12
  • de 13 a 15 (lembrando que não podemos passar do valor que queremos calcular)

No programa, vamos enviando cada um pacote de dados (begin, end) para cada servidor. Quando acabarem os servidores disponíveis, começa novamente do primeiro e vai enviando novos pacotes de dados. Nesse caso, só mandou dois pacotes para o primeiro.

Conforme as gorotines vão recebendo o resultado, vão jogando em um canal buffer.

Quando todas as rotinas terminam, o programa pega todos os resultados e multiplicas eles, gerando o resultado final.

Do lado do servidor, ele apenas multiplica cada número, do número inicial até o final, e envia de volta o resultado.

Veja que estamos trabalhando com mensagens JSON aqui.

Servidor

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"strconv"
)

type Message struct {
	Begin int `json:"begin"`
	End   int `json:"end"`
}

func child(conn net.Conn) {
	defer conn.Close()
	netData, err := bufio.NewReader(conn).ReadString('\n')
	if err != nil {
		fmt.Println(err)
		return
	}
	msg := Message{}
	json.Unmarshal([]byte(netData), &msg)
	out := 1
	for i := msg.Begin; i <= msg.End; i++ {
		out *= i
	}
	fmt.Println(msg.Begin, "->", msg.End, ":", out)
	conn.Write([]byte(strconv.Itoa(out)))
}

func main() {
	arguments := os.Args
	if len(arguments) == 1 {
		fmt.Println("Enter with port number in argument")
		return
	}
	l, err := net.Listen("tcp", ":"+arguments[1])
	if err != nil {
		fmt.Println(err)
		return
	}
	defer l.Close()
	fmt.Println("TCP Server initialized at port", arguments[1])

	for {
		c, err := l.Accept()
		if err != nil {
			fmt.Println(err)
			return
		}
		go child(c)
	}
}

Cliente

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"strconv"
	"sync"
)

type Message struct { // Estrutura da mensagem
	Begin int `json:"begin"`
	End   int `json:"end"`
}

func connect_server(wg *sync.WaitGroup, ch chan int, server string, begin, end int) {
	defer wg.Done()
	c, err := net.Dial("tcp", server) // Conecta ao servidor
	if err != nil {
		fmt.Println(err)
		return
	}
	defer c.Close()
	msg := Message{begin, end}                        // cria uma mensagem na estrutura definida
	msgJson, _ := json.Marshal(msg)                   // transforma em string json
	fmt.Fprintf(c, string(msgJson)+"\n")              // envia ao servidor
	message, _ := bufio.NewReader(c).ReadString('\n') // aguarda a resposta
	result, err := strconv.Atoi(message)              // transforma a resposta para inteiro
	if err == nil {
		fmt.Println(begin, "->", end, ":", result)
		ch <- result // manda o resultado para o programa principal
	}
}

/*
	EXECUÇÃO:
	go run fat_dist_cli.go [value] [rotines] [server:port] ...

	1: Valor para calcular o fatorial
	2: Quantidade de gorotines, ou conexões a servidores
	3 em diante: servidores para se conectar no formato servidor:porta

	EXEMPLO:

	go run fat_dist_cli.go 20 5 localhost:4400 localhost:4402 localhost:4401

	O exemplo acima pede para calcular o fatorial de 20, dividindo em 5
	subrotinas. Em sequência temos 3 servidores para utilizar. Como o número
	de servidores é menor do que a quantidade de subrotinas, dois deles vão receber
	duas requisições de cálculos
*/

func main() {
	args := os.Args                        // pega os argumentos
	value, _ := strconv.Atoi(args[1])      // pega o valor
	rotines, _ := strconv.Atoi(args[2])    // pega a quantidade de rotinas
	size := int(value / rotines)           // calcula o tamanho do bloco a separar
	servers := make([]string, len(args)-3) // cria o vetor de servidores
	for i := 3; i < len(args); i++ {       // coloca os parâmetros como servidores
		servers[i-3] = args[i]
	}

	var wg sync.WaitGroup
	wg.Add(rotines)
	ch := make(chan int, rotines)

	j := 0
	end := 0
	for i := 0; i < rotines; i++ { // cria as subrotinas
		begin := end + 1   // separa o inicio
		end = begin + size // e o fim do bloco de cálculo
		if end > value {   // Não pode passar do valor total
			end = value
		}
		go connect_server(&wg, ch, servers[j], begin, end) // envia para um servidor
		j++
		if j >= len(servers) {
			j = 0
		}
	}
	wg.Wait() // aguarda os servidores terminarem
	close(ch)
	// Pega os resultados e multiplica-os
	result := 1
	for v := range ch {
		result *= v
	}
	// apresenta o resultado final
	fmt.Println("Fatorial de", value, "é", result)
}
Disponível em “https://saulo.arisa.com.br/wiki/index.php?title=Go:_Programação_Distribuída&oldid=9560”