From 0b6e138270379d6ff7268141a2951eb99d48d06d Mon Sep 17 00:00:00 2001 From: Pavel Korotkov Date: Wed, 31 Mar 2021 22:08:39 +0300 Subject: [PATCH] [#19] Prepare using connection pool Signed-off-by: Pavel Korotkov --- app.go | 86 +++++++++++++++++-------------------------- neofs/client-plant.go | 20 +++++++++- neofs/credentials.go | 22 +++++------ 3 files changed, 64 insertions(+), 64 deletions(-) diff --git a/app.go b/app.go index 06b0bf7..cd7c9b9 100644 --- a/app.go +++ b/app.go @@ -18,14 +18,13 @@ import ( type ( app struct { - log *zap.Logger - plant neofs.ClientPlant - cfg *viper.Viper - auxiliaryLog logger.Logger - web *fasthttp.Server - jobDone chan struct{} - webDone chan struct{} - enableDefaultTimestamp bool + log *zap.Logger + plant neofs.ClientPlant + cfg *viper.Viper + auxiliaryLog logger.Logger + webServer *fasthttp.Server + jobDone chan struct{} + webDone chan struct{} } App interface { @@ -57,61 +56,55 @@ func WithConfig(c *viper.Viper) Option { func newApp(ctx context.Context, opt ...Option) App { a := &app{ - log: zap.L(), - cfg: viper.GetViper(), - web: new(fasthttp.Server), - jobDone: make(chan struct{}), - webDone: make(chan struct{}), + log: zap.L(), + cfg: viper.GetViper(), + webServer: new(fasthttp.Server), + jobDone: make(chan struct{}), + webDone: make(chan struct{}), } for i := range opt { opt[i](a) } - a.enableDefaultTimestamp = a.cfg.GetBool(cfgUploaderHeaderEnableDefaultTimestamp) a.auxiliaryLog = logger.GRPC(a.log) - if a.cfg.GetBool(cmdVerbose) { grpclog.SetLoggerV2(a.auxiliaryLog) } - // conTimeout := a.cfg.GetDuration(cfgConTimeout) // reqTimeout := a.cfg.GetDuration(cfgReqTimeout) // tckTimeout := a.cfg.GetDuration(cfgRebalance) - // -- setup FastHTTP server -- - a.web.Name = "neofs-http-gate" - a.web.ReadBufferSize = a.cfg.GetInt(cfgWebReadBufferSize) - a.web.WriteBufferSize = a.cfg.GetInt(cfgWebWriteBufferSize) - a.web.ReadTimeout = a.cfg.GetDuration(cfgWebReadTimeout) - a.web.WriteTimeout = a.cfg.GetDuration(cfgWebWriteTimeout) - a.web.DisableHeaderNamesNormalizing = true - a.web.NoDefaultServerHeader = true - a.web.NoDefaultContentType = true - a.web.MaxRequestBodySize = a.cfg.GetInt(cfgWebMaxRequestBodySize) - + a.webServer.Name = "neofs-http-gate" + a.webServer.ReadBufferSize = a.cfg.GetInt(cfgWebReadBufferSize) + a.webServer.WriteBufferSize = a.cfg.GetInt(cfgWebWriteBufferSize) + a.webServer.ReadTimeout = a.cfg.GetDuration(cfgWebReadTimeout) + a.webServer.WriteTimeout = a.cfg.GetDuration(cfgWebWriteTimeout) + a.webServer.DisableHeaderNamesNormalizing = true + a.webServer.NoDefaultServerHeader = true + a.webServer.NoDefaultContentType = true + a.webServer.MaxRequestBodySize = a.cfg.GetInt(cfgWebMaxRequestBodySize) // -- -- -- -- -- -- FIXME -- -- -- -- -- -- // Does not work with StreamRequestBody, // some bugs with readMultipartForm // https://github.com/valyala/fasthttp/issues/968 - a.web.DisablePreParseMultipartForm = true - a.web.StreamRequestBody = a.cfg.GetBool(cfgWebStreamRequestBody) + a.webServer.DisablePreParseMultipartForm = true + a.webServer.StreamRequestBody = a.cfg.GetBool(cfgWebStreamRequestBody) // -- -- -- -- -- -- -- -- -- -- -- -- -- -- - - var cl connectionList + var cl neofs.ConnectionList for i := 0; ; i++ { address := a.cfg.GetString(cfgPeers + "." + strconv.Itoa(i) + ".address") weight := a.cfg.GetFloat64(cfgPeers + "." + strconv.Itoa(i) + ".weight") if address == "" { break } - cl = append(cl, connection{address: address, weight: weight}) - a.log.Info("add connection peer", zap.String("address", address), zap.Float64("weight", weight)) + cl.Add(address, weight) + a.log.Info("add connection", zap.String("address", address), zap.Float64("weight", weight)) } sort.Sort(sort.Reverse(cl)) - cred, err := neofs.NewCredentials(a.cfg.GetString(cmdNeoFSKey)) + creds, err := neofs.NewCredentials(a.cfg.GetString(cmdNeoFSKey)) if err != nil { - a.log.Fatal("could not get credentials", zap.Error(err)) + a.log.Fatal("could not get neofs credentials", zap.Error(err)) } - a.plant, err = neofs.NewClientPlant(ctx, cl[0].address, cred) + a.plant, err = neofs.NewClientPlant(ctx, cl, creds) if err != nil { a.log.Fatal("failed to create neofs client") } @@ -120,7 +113,6 @@ func newApp(ctx context.Context, opt ...Option) App { func (a *app) Wait() { a.log.Info("starting application") - select { case <-a.jobDone: // wait for job is stopped <-a.webDone @@ -136,10 +128,11 @@ func (a *app) Worker(ctx context.Context) { func (a *app) Serve(ctx context.Context) { go func() { <-ctx.Done() - a.log.Info("shutting down web server", zap.Error(a.web.Shutdown())) + a.log.Info("shutting down web server", zap.Error(a.webServer.Shutdown())) close(a.webDone) }() - uploader := uploader.New(a.log, a.plant, a.enableDefaultTimestamp) + edts := a.cfg.GetBool(cfgUploaderHeaderEnableDefaultTimestamp) + uploader := uploader.New(a.log, a.plant, edts) downloader, err := downloader.New(ctx, a.log, a.plant) if err != nil { a.log.Fatal("failed to create downloader", zap.Error(err)) @@ -167,19 +160,8 @@ func (a *app) Serve(ctx context.Context) { } bind := a.cfg.GetString(cfgListenAddress) a.log.Info("running web server", zap.String("address", bind)) - a.web.Handler = r.Handler - if err := a.web.ListenAndServe(bind); err != nil { + a.webServer.Handler = r.Handler + if err := a.webServer.ListenAndServe(bind); err != nil { a.log.Fatal("could not start server", zap.Error(err)) } } - -type connection struct { - address string - weight float64 -} - -type connectionList []connection - -func (p connectionList) Len() int { return len(p) } -func (p connectionList) Less(i, j int) bool { return p[i].weight < p[j].weight } -func (p connectionList) Swap(i, j int) { p[i], p[j] = p[j], p[i] } diff --git a/neofs/client-plant.go b/neofs/client-plant.go index cf3aaa9..babcaff 100644 --- a/neofs/client-plant.go +++ b/neofs/client-plant.go @@ -102,9 +102,27 @@ func (cc *neofsClient) OwnerID() *owner.ID { return cc.ownerID } -func NewClientPlant(ctx context.Context, address string, creds Credentials) (ClientPlant, error) { +type Connection struct { + address string + weight float64 +} + +type ConnectionList []Connection + +func (p ConnectionList) Len() int { return len(p) } +func (p ConnectionList) Less(i, j int) bool { return p[i].weight < p[j].weight } +func (p ConnectionList) Swap(i, j int) { p[i], p[j] = p[j], p[i] } + +func (cl *ConnectionList) Add(address string, weight float64) ConnectionList { + *cl = append(*cl, Connection{address, weight}) + return *cl +} + +func NewClientPlant(ctx context.Context, connectionList ConnectionList, creds Credentials) (ClientPlant, error) { toctx, c := context.WithTimeout(ctx, nodeConnectionTimeout) defer c() + // TODO: Use connection pool here. + address := connectionList[0].address conn, err := grpc.DialContext(toctx, address, grpc.WithInsecure(), grpc.WithBlock()) if err != nil { if err == context.DeadlineExceeded { diff --git a/neofs/credentials.go b/neofs/credentials.go index 48faf64..20b8c9b 100644 --- a/neofs/credentials.go +++ b/neofs/credentials.go @@ -16,10 +16,10 @@ type ( PrivateKey() *ecdsa.PrivateKey } - cred struct { - key *ecdsa.PrivateKey - owner *owner.ID - wif string + credentials struct { + key *ecdsa.PrivateKey + ownerID *owner.ID + wif string } ) @@ -34,26 +34,26 @@ func NewCredentials(secret string) (Credentials, error) { } // PrivateKey returns ecdsa.PrivateKey. -func (c *cred) PrivateKey() *ecdsa.PrivateKey { +func (c *credentials) PrivateKey() *ecdsa.PrivateKey { return c.key } // PublicKey returns ecdsa.PublicKey. -func (c *cred) PublicKey() *ecdsa.PublicKey { +func (c *credentials) PublicKey() *ecdsa.PublicKey { return &c.key.PublicKey } // Owner returns owner.ID. -func (c *cred) Owner() *owner.ID { - return c.owner +func (c *credentials) Owner() *owner.ID { + return c.ownerID } // WIF returns string representation of WIF. -func (c *cred) WIF() string { +func (c *credentials) WIF() string { return c.wif } -func setFromPrivateKey(key *ecdsa.PrivateKey) (*cred, error) { +func setFromPrivateKey(key *ecdsa.PrivateKey) (*credentials, error) { wallet, err := owner.NEO3WalletFromPublicKey(&key.PublicKey) if err != nil { return nil, err @@ -63,5 +63,5 @@ func setFromPrivateKey(key *ecdsa.PrivateKey) (*cred, error) { if err != nil { return nil, err } - return &cred{key: key, owner: ownerID, wif: wif}, nil + return &credentials{key: key, ownerID: ownerID, wif: wif}, nil }