

Goで投資の4本値(一定期間内の始値、高値、安値、終値、時間)の値をクライアントに送るTCPサーバを書いています。  
送るデータは先頭8バイトはJSONで表された4本値の構造体のバイト数を示していて、その後に実際のJSON文字列が続く形式です。

このTCPサーバのテストのために下記のようなプログラムを書いたのですが、途中でクライアントのほうがなぜかReadをブロックしてしまい(なにか原因が他にサーバーのほうにあるのかもしれませんが...)うまく終了しません。  
なぜでしょうか?  
下記のテストプログラムのCandleが4本値のstructです。  
サーバー側のコードもその下に記載してあります。

環境はx64 Linux, Go1.18です

##### テストプログラム

[code]

    
    import (
        "fmt"
        "log"
        "net"
        "strconv"
        "strings"
        "sync"
        "time"
    )
    
    type Candle struct {
        Count    int       `json:"Count,omitempty"`
        Open     float64   `json:"Open"`
        Close    float64   `json:"Close"`
        Low      float64   `json:"Low"`
        High     float64   `json:"High"`
        Datetime time.Time `json:"Datetime"`
    }
    
    func main() {
        data_buf := make([]byte, 1024*1024)
        server_addr := "localhost"
        port := 33333
        port_str := "33333"
        JST := time.FixedZone("JST", 9*60*60)
        cs := make([]Candle, 0, 200)
        newline_sep := strings.Split(data_csv, "\n") //data_csvはCandleの値のcsv文字列
        for _, v := range newline_sep {
            var c Candle
            comma_sep := strings.Split(v, ",")
            date_str := comma_sep[0] + " " + comma_sep[1]
            date, err := time.ParseInLocation("2006/01/02 15:04", date_str, JST)
            if err != nil {
                log.Fatal(err)
            }
            c.Datetime = date
            open, err := strconv.ParseFloat(comma_sep[2], 64)
            if err != nil {
                log.Fatal(err)
            }
            high, err := strconv.ParseFloat(comma_sep[3], 64)
            if err != nil {
                log.Fatal(err)
            }
            low, err := strconv.ParseFloat(comma_sep[4], 64)
            if err != nil {
                log.Fatal(err)
            }
            clo, err := strconv.ParseFloat(comma_sep[5], 64)
            if err != nil {
                log.Fatal(err)
            }
            c.Open = open
            c.Close = clo
            c.High = high
            c.Low = low
            cs = append(cs, c)
        } // ここまででサーバに送るCandleの配列を作成
        num_candle := len(cs)
        fmt.Println("num_candle ", num_candle)
        candle_chan := make(chan Candle)
    
        go func(server_chan chan Candle) { //サーバーを別のgo routineで起動。サーバはserver_chanからCandleを受け取ってクライアントに流す
            s := NewCandleServer(server_chan)
            s.Run(server_addr, port)
        }(candle_chan)
        wg := &sync.WaitGroup;{}
        wg.Add(1)
        go func() {
            time.Sleep(1500 * time.Millisecond)
            for _, v := range cs {
                candle_chan  len(data_buf) {
                data_buf = make([]byte, size)
            }
            fmt.Println("read hang before")
            _, err = conn.Read(data_buf)
            fmt.Println("read hang")
            if err != nil {
                log.Fatal("payload read error:", err)
            }
            i += 1
        }
        wg.Wait()
        if i != num_candle {
            _ = fmt.Errorf("something Error")
        }
    }
    
    
    
[/code]

##### NewCandleServerのコード

[code]

    import (
        "encoding/json"
        "fmt"
        "investment/calc"
        "investment/kabucom3"
        "log"
        "net"
        "strconv"
        "sync"
    )
    
    type Servable interface {
        kabucom3.PositionsSuccess | calc.Candle
    }
    
    func ByteLength(b []byte) ([]byte, int) {
        buff := make([]byte, 8)
        b_len := len(b)
        b_len64 := uint64(b_len)
        for i := 0; i < 8; i++ {
            buff[i] = byte((b_len64 >> ((7 - i) * 8))) & byte(0xff)
        }
        buff = append(buff, b...)
        return buff, b_len
    }
    
    type CandleServer[T Servable] struct {
        Conns *sync.Map
        C     chan T
    }
    
    func (t *CandleServer[T]) Run(addr string, port int) {
        defer func() {
            t.Conns.Range(func(k, v any) bool {
                c := k.(*net.Conn)
                (*c).Close()
                return true
            })
        }()
        go func() {
            var c int = 0
            for {
                select {
                case v := <-t.C:
                    c += 1
                    fmt.Println("candle_server", "case", c)
                    b, err := json.Marshal(v)
                    if err != nil {
                        fmt.Println("hang ", err)
                    }
                    result_buff, _ := ByteLength(b)
                    t.Conns.Range(func(k, v any) bool {
                        c := k.(*net.Conn)
                        _, err := (*c).Write(result_buff)
                        if err != nil {
                            (*c).Close()
                            t.Conns.Delete(k)
                        }
                        return true
                    })
                    fmt.Println("candle_server", "after range")
                }
            }
        }()
        ln, err := net.Listen("tcp", addr+":"+strconv.Itoa(port))
        if err != nil {
            log.Fatal("Listen Error", err)
        }
        for {
            conn, err := ln.Accept()
            if err != nil {
                log.Fatal("Accept Error", err)
            }
            t.Conns.Store(&conn, struct{}{})
        }
    
    }
    
    func NewCandleServer[T Servable](c chan T) *CandleServer[T] {
        return &CandleServer[T]{C: c, Conns: &sync.Map{}}
    }
    
[/code]


CC BY-SA 4.0
2022-05-30T07:07:29.063
89132
2022-06-08T02:45:50.760
2022-05-30T09:44:57.537
5246
5246
question
go tcp
Goでなぜか実行がブロックされてしまいプログラムが終了しない


io.ReaderのRead(buff)メソッドはbuffサイズ分の受信が完了するまでブロックします。  
送信側でそれより短い送信内容だとReadが完了することがありません。

JSONのデコード結果を受け取るのにストリーミングデコーダーが標準で用意されています。

[code]

    var result map[string]interface{}
    if err := json.NewDecorder(conn).Decode(&result); err != nil {
      エラーハンドリング
    }
    
[/code]

この方法ならJSONとして解釈できるまでストリームをReadします。


CC BY-SA 4.0
2022-06-08T02:45:50.760
89277
2022-06-08T02:45:50.760
11051
89132
answer