Goで投資の4本値(一定期間内の始値、高値、安値、終値、時間)の値をクライアントに送るTCPサーバを書いています。
送るデータは先頭8バイトはJSONで表された4本値の構造体のバイト数を示していて、その後に実際のJSON文字列が続く形式です。
このTCPサーバのテストのために下記のようなプログラムを書いたのですが、途中でクライアントのほうがなぜかReadをブロックしてしまい(なにか原因が他にサーバーのほうにあるのかもしれませんが...)うまく終了しません。
なぜでしょうか?
下記のテストプログラムのCandleが4本値のstructです。
サーバー側のコードもその下に記載してあります。
環境はx64 Linux, Go1.18です
テストプログラム
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")
}
}
NewCandleServerのコード
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{}}
}