Peer to peer (P2P)란 무엇인가?

P2P 네트워크는 클라이언트나 서버란 개념 없이, 오로지 동등한 계층 노드들(peer nodes)이 서로 클라이언트와 서버 역할을 동시에 네트워크 위에서 하게 됩니다.
그래서 P2P 아키텍처에서는 블록체인의 상태를 유지하기 위한 중앙서버가 필요 없습니다. 예를들어 친구에게 비트코인을 보낼 때 비트코인 블록체인은 당신의 수량을 감소시키고 친구의 수량을 증가시켜 현재 상태를 업데이트 할 것입니다.

은행처럼 중앙에서 전체 상태를 유지하는 대신에 비트코인 네트워크상의 모든 노드에 똑같은 전체 거래내역을 복사하여 유지 합니다. 그렇게 하면 네트워크상의 노드 51%가 동의해야만 블록체인을 유지할수 있습니다.

코딩을 시작해 봅시다.

확장성과 유연성을 만들기 위해서는 많은 엔지니어링이 필요합니다. 그래서 개발하기 쉽게 툴을 사용해봅시다.
Go-libp2p라는 라이브러리가 있습니다. 이것은 IPFS에서 만든 팀에서 만들었습니다. 이 라이브러리를 이용하여 개발 해봅시다.

go-libp2p 라이브러리는 2개의 단점을 가집니다.

설치하기가 힙듭니다. gx라는 패키지 관리자를 이용해야하는데 매우 불편합니다.
이 라이브러리는 무거운 라이브러리 입니다. 따라서 코드가 동작할 때 data race(동시에 메모리의 데이터에 접근하는 현상)가 발생할 수 있습니다.

환경설정

환경설정을 위한 좋은 방법은 전체 라이브러리를 다운 받아 라이브러리 안에서 코딩 하는 것입니다. 제공하는 환경 밖에서 개발을 할수 있지만gx가 어떻게 동작 하는지 알아야합니다.
아래 명령어로 설치하세요. 더 자세한 방법은 여기서 확인 가능합니다.
go get -d github.com/libp2p/go-libp2p/...
cd GO설치 경로/src/github.com/libp2p/go-libp2p
make
스크린샷 2018-04-17 오후 9.46.26.png
make deps
스크린샷 2018-04-17 오후 10.06.14.png

gx 패키지 매너저를 통하여 모든 패지지와 종속성을 얻을 수 있습니다. gx좋은 라이브러리 이지만 Go 컨벤션을 많이 깨기 때문에 불편한점이 있습니다.

mkdir ./examples/p2p

examples 폴더안에 p2p폴더를 만들고 p2p폴더안에 main.go 파일을 만드세요.
examples 폴더 구조는 다음과 같습니다.
스크린샷 2018-04-18 오전 12.35.37.png

Imports

package main

import (
        "bufio"
    "context"
    "crypto/rand"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "flag"
    "fmt"
    "io"
    "log"
    mrand "math/rand"
    "os"
    "strconv"
    "strings"
    "sync"
    "time"
    "github.com/davecgh/go-spew/spew"
    golog "github.com/ipfs/go-log"
    libp2p "github.com/libp2p/go-libp2p"
    crypto "github.com/libp2p/go-libp2p-crypto"
    host "github.com/libp2p/go-libp2p-host"
    net "github.com/libp2p/go-libp2p-net"
    peer "github.com/libp2p/go-libp2p-peer"
    pstore "github.com/libp2p/go-libp2p-peerstore"
    ma "github.com/multiformats/go-multiaddr"
    gologging "github.com/whyrusleeping/go-logging"

)

이 튜토리얼을 위해서는 spew 패키지가 필요합니다.

아래 명령어로 설치하세요

go get github.com/davecgh/go-spew/spew

개발에 필요한 전역 변수를 선언 합시다.

type Block struct {
    Index     int   
    Timestamp string
    BPM       int   
    Hash      string
    PrevHash  string
}
var Blockchain []Block
var mutex = &sync.Mutex{}

Block은 트랜잭션의 정보를 가지고 있는 변수 입니다. BPM이 각 블록에 저장됩니다.
Blockchian은 모든 블록을 가지는 변수입니다. Block들을 연결한 정보가 담깁니다.
Mutex는 똑같은 데이터에 동시 접근할 수 없도록 동기화 시켜주는 변수입니다..

함수들을 작성해 봅시다.

func isBlockValid(newBlock, oldBlock Block) bool {
    if oldBlock.Index+1 != newBlock.Index {
        return false
    }

    if oldBlock.Hash != newBlock.PrevHash {
        return false
    }
    
    if calculateHash(newBlock) != newBlock.Hash {
        return false
    }
    
    return true
}

// SHA256 hashing
func calculateHash(block Block) string {
    record := strconv.Itoa(block.Index) + block.Timestamp + strconv.Itoa(block.BPM) + block.PrevHash
    h := sha256.New()
    h.Write([]byte(record))
    hashed := h.Sum(nil)
    return hex.EncodeToString(hashed)
}

// create a new block using previous block's hash
func generateBlock(oldBlock Block, BPM int) Block {

    var newBlock Block
    
    t := time.Now()
    
    newBlock.Index = oldBlock.Index + 1
    newBlock.Timestamp = t.String()
    newBlock.BPM = BPM
    newBlock.PrevHash = oldBlock.Hash
    newBlock.Hash = calculateHash(newBlock)
    
    return newBlock
}

isBlockValid 블록체인에 등록될 블록을 검증하는 함수입니다.
calculateHash SHA256을 이용하여 데이터를 암호화합니다.
generateBlock 필요한 정보를 블록에 담은후 새로운 블록을 생성하는 함수입니다.

###P2P 개발에 필요한 것들

HOST
먼저 호스트를 생성할 수 있는 코드를 작성 해야합니다. 노드가 프로그램을 실행할때 다른 노드들이 연결할수 있는 호스트로써의 역할을 해야합니다. 아래 코드가 있습니다.

func makeBasicHost(listenPort int, secio bool, randseed int64) (host.Host, error) {

    // If the seed is zero, use real cryptographic randomness. Otherwise, use a
    // deterministic randomness source to make generated keys stay the same
    // across multiple runs
    var r io.Reader
    if randseed == 0 {
        r = rand.Reader
    } else {
        r = mrand.New(mrand.NewSource(randseed))
    }
    
    // Generate a key pair for this host. We will use it
    // to obtain a valid host ID.
    priv, _, err := crypto.GenerateKeyPairWithReader(crypto.RSA, 2048, r)
    if err != nil {
        return nil, err
    }
    
    opts := []libp2p.Option{
        libp2p.ListenAddrStrings(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)),
        libp2p.Identity(priv),
    }
    
    if !secio {
        opts = append(opts, libp2p.NoEncryption())
    }
    
    basicHost, err := libp2p.New(context.Background(), opts...)
    if err != nil {
        return nil, err
    }
    
    // Build host multiaddress
    hostAddr, _ := ma.NewMultiaddr(fmt.Sprintf("/ipfs/%s", basicHost.ID().Pretty()))
    
    // Now we can build a full multiaddress to reach this host
    // by encapsulating both addresses:
    addr := basicHost.Addrs()[0]
    fullAddr := addr.Encapsulate(hostAddr)
    log.Printf("I am %s\n", fullAddr)
    if secio {
        log.Printf("Now run \"go run main.go -l %d -d %s -secio\" on a different terminal\n", listenPort+1, fullAddr)
    } else {
        log.Printf("Now run \"go run main.go -l %d -d %s\" on a different terminal\n", listenPort+1, fullAddr)
    }
    
    return basicHost, nil
}

makeBasicHost함수는 3개의 인자를 받은후 Host또는 Error를 반환합니다.

listenPort는 다른 peer가 접근할 수 있도록 지정하는 포트입니다.
Secio는 안전하게 데이터 스트림을 사용할건지(true 또는 false) 정하는 값입니다. 보통 사용하는 것이 좋습니다. secure input/ouput의 약자입니다.
randSeed는 호스트의 임의 주소를 생성할지를 결정하는 부가적인 인자입니다 0이면 임의적인 주소를 생성하고 아니면 고정적인 주소를 사용합니다. 사용하지 않을 것이지만 일단 가지고 있는게 좋을것 같습니다.

함수의 첫번째 if문은 randSeed 사용하는지를 확인하고 호스트의 키를 생성합니다. 공개키와 개인키를 생성하고 opts는 다른 peer가 접속할 수 있도록 주소를 구성합니다.
!secio암호화를 패스하는 부분입니다. 하지만 보안을 위해서 secio를 사용할 예정이기 때문에 이 부분은 적용되지 않습니다. log.Printf를 이용하여 다른 노드에서 접속하는 방법을 콘솔로 출력합니다.

Stream handler
다른 노드에서 현재 호스트로 연결하고 나서 블록을 생성할때 그 블록을 블록체인에 추가해야할지를 결정하는 코드가 필요합니다. 그리고 블록이 추가되고나서 각 peer들에게 알려주는 코드도 필요합니다.
아래 코드가 있습니다.

func handleStream(s net.Stream) {
    

        log.Println("Got a new stream!")
    

        // Create a buffer stream for non blocking read and write.
        rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))
    

        go readData(rw)
        go writeData(rw)
    

        // stream 's' will stay open until you close it (or the other side closes it).
    }

새로운 ReadWrite를 생성하고 Go 루틴을 이용하여 read(다른 노드에서 보내주는 데이터), write(데이터 확인 후 다른 노드에 알려주는 기능)을 분리하여 처리합니다.

###Read

readData함수를 먼저 생성해 봅시다.

블록체인을 언제든지 전송받아 읽을 수 있도록 무한루프안에 로직을 작성합니다. 전송받은 블록체인 JSON 문자열을 ReadString으로 파싱합니다. 만약에 비어있지 않다면 Unmarshal 디코딩합니다.

func readData(rw *bufio.ReadWriter) {
    

        for {
            str, err := rw.ReadString('\n')
            if err != nil {
                log.Fatal(err)
            }
    

            if str == "" {
                return
            }
            if str != "\n" {
    

                chain := make([]Block, 0)
                if err := json.Unmarshal([]byte(str), &chain); err != nil {
                    log.Fatal(err)
                }
    

                mutex.Lock()
                if len(chain) > len(Blockchain) {
                    Blockchain = chain
                    bytes, err := json.MarshalIndent(Blockchain, "", "  ")
                    if err != nil {
    

                        log.Fatal(err)
                    }
                    // Green console color:     \x1b[32m
                    // Reset console color:     \x1b[0m
                    fmt.Printf("\x1b[32m%s\x1b[0m> ", string(bytes))
                }
                mutex.Unlock()
            }
        }
    }

그리고나서 전송 받은 블록의 크기를 체크합니다. 만약에 들어오는 체인의 크기가 블록체인에 저장된 길이 보다 크다면 블록체인을 업데이트합니다.
JSON 형식으로 Marshal 인코딩 한 후 fmt.printf 새로운 블록체인을 보기 쉽게 출력합니다.
다른 노드에서 블록체인을 받고 그 블록체인을 사용하게 된다면 현재 호스트에 접속한 다른 노드에게 업데이트 된 블록체인을 알려줄 수 있는 방법을 찾야합니다.
writeData함수를 통해 알아봅시다.

###Write

func writeData(rw *bufio.ReadWriter) {
    

        go func() {
            for {
                time.Sleep(5 * time.Second)
                mutex.Lock()
                bytes, err := json.Marshal(Blockchain)
                if err != nil {
                    log.Println(err)
                }
                mutex.Unlock()
    

                mutex.Lock()
                rw.WriteString(fmt.Sprintf("%s\n", string(bytes)))
                rw.Flush()
                mutex.Unlock()
    

            }
        }()
    

        stdReader := bufio.NewReader(os.Stdin)
    

        for {
            fmt.Print("> ")
            sendData, err := stdReader.ReadString('\n')
            if err != nil {
                log.Fatal(err)
            }
    

            sendData = strings.Replace(sendData, "\n", "", -1)
            bpm, err := strconv.Atoi(sendData)
            if err != nil {
                log.Fatal(err)
            }
            newBlock := generateBlock(Blockchain[len(Blockchain)-1], bpm)
    

            if isBlockValid(newBlock, Blockchain[len(Blockchain)-1]) {
                mutex.Lock()
                Blockchain = append(Blockchain, newBlock)
                mutex.Unlock()
            }
    

            bytes, err := json.Marshal(Blockchain)
            if err != nil {
                log.Println(err)
            }
    

            spew.Dump(Blockchain)
    

            mutex.Lock()
            rw.WriteString(fmt.Sprintf("%s\n", string(bytes)))
            rw.Flush()
            mutex.Unlock()
        }
    

    }

위 함수의 내용은 다음과 같습니다. 매 5초마다 각 peer에게 업데이트된 블록체인을 알려주는 Go루틴과 다른 노드에서 블록체인을 받고 현재 호스트에 가지고 있는 블록의 크기 보다 크면 받아 들이고 작으면 버리는 함수 이 두가지 기능을 통해 모든 peer는 업데이트된 블록체인을 계속 해서 받을 수 있도록 합니다. Bufio.NewReader를 가지고 새로운 reader를 생성하면 stdin(콘솔 입력)를 통해 BPM을 입력 할 수 있습니다. 무한루프 안에 작성하여 입력되면 새로운 블록이 생성 될수 있도록 합니다. BPM이 블록에 저장되기 위해서는 integer형식이어야 하기 때문에 다른 형식으로 입력되는지 확인합니다. integer 형식이 맞다면 블록을 Masharl 인코딩 하여 콘솔에 출력하여 확인합니다. 그런 다음 rw.WriteString을 사용하여 연결된 peer에 업데이트 된 블록체인을 전송 합니다.

func main() {
        t := time.Now()
        genesisBlock := Block{}
        genesisBlock = Block{0, t.String(), 0, calculateHash(genesisBlock), ""}
    

        Blockchain = append(Blockchain, genesisBlock)
    

        // LibP2P code uses golog to log messages. They log with different
        // string IDs (i.e. "swarm"). We can control the verbosity level for
        // all loggers with:
        golog.SetAllLoggers(gologging.INFO) // Change to DEBUG for extra info
    

        // Parse options from the command line
        listenF := flag.Int("l", 0, "wait for incoming connections")
        target := flag.String("d", "", "target peer to dial")
        secio := flag.Bool("secio", false, "enable secio")
        seed := flag.Int64("seed", 0, "set random seed for id generation")
        flag.Parse()
    

        if *listenF == 0 {
            log.Fatal("Please provide a port to bind on with -l")
        }
    

        // Make a host that listens on the given multiaddress
        ha, err := makeBasicHost(*listenF, *secio, *seed)
        if err != nil {
            log.Fatal(err)
        }
    

        if *target == "" {
            log.Println("listening for connections")
            // Set a stream handler on host A. /p2p/1.0.0 is
            // a user-defined protocol name.
            ha.SetStreamHandler("/p2p/1.0.0", handleStream)
    

            select {} // hang forever
            /**** This is where the listener code ends ****/
        } else {
            ha.SetStreamHandler("/p2p/1.0.0", handleStream)
    

            // The following code extracts target's peer ID from the
            // given multiaddress
            ipfsaddr, err := ma.NewMultiaddr(*target)
            if err != nil {
                log.Fatalln(err)
            }
    

            pid, err := ipfsaddr.ValueForProtocol(ma.P_IPFS)
            if err != nil {
                log.Fatalln(err)
            }
    

            peerid, err := peer.IDB58Decode(pid)
            if err != nil {
                log.Fatalln(err)
            }
    

            // Decapsulate the /ipfs/<peerID> part from the target
            // /ip4/<a.b.c.d>/ipfs/<peer> becomes /ip4/<a.b.c.d>
            targetPeerAddr, _ := ma.NewMultiaddr(
                fmt.Sprintf("/ipfs/%s", peer.IDB58Encode(peerid)))
            targetAddr := ipfsaddr.Decapsulate(targetPeerAddr)
    

            // We have a peer ID and a targetAddr so we add it to the peerstore
            // so LibP2P knows how to contact it
            ha.Peerstore().AddAddr(peerid, targetAddr, pstore.PermanentAddrTTL)
    

            log.Println("opening stream")
            // make a new stream from host B to host A
            // it should be handled on host A by the handler we set above because
            // we use the same /p2p/1.0.0 protocol
            s, err := ha.NewStream(context.Background(), peerid, "/p2p/1.0.0")
            if err != nil {
                log.Fatalln(err)
            }
            // Create a buffered stream so that read and writes are non blocking.
            rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))
    

            // Create a thread to read and write data.
            go writeData(rw)
            go readData(rw)
    

            select {} // hang forever
    

        }
    }