shkong/MiraiGo
1
0
Fork 0
mirror of https://github.com/Mrs4s/MiraiGo.git synced 2025-05-04 19:17:38 +08:00
Code Issues Projects Releases Wiki Activity

switch errgroup.

Browse Source
This commit is contained in:
wdvxdr 2021-07-31 22:30:19 +08:00
parent 624a7e4101
commit feb482e0e3
No known key found for this signature in database
GPG Key ID: 703F8C071DE7A1B6
3 changed files with 12 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
client/highway.go
View File

@ -15,6 +15,7 @@ import (
"time" "time"
"github.com/pkg/errors" "github.com/pkg/errors"
"golang.org/x/sync/errgroup"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
"github.com/Mrs4s/MiraiGo/binary" "github.com/Mrs4s/MiraiGo/binary"
@ -232,7 +233,6 @@ func (c *QQClient) highwayUploadFileMultiThreadingByBDH(path string, cmdId int32
rspExt []byte rspExt []byte
BlockId = ^uint32(0) // -1 BlockId = ^uint32(0) // -1
uploadedCount uint32 uploadedCount uint32
lastErr error
cond = sync.NewCond(&sync.Mutex{}) cond = sync.NewCond(&sync.Mutex{})
) )
// Init Blocks // Init Blocks
@ -253,6 +253,8 @@ func (c *QQClient) highwayUploadFileMultiThreadingByBDH(path string, cmdId int32
}) })
} }
doUpload := func() error { doUpload := func() error {
defer cond.Signal()
conn, err := net.DialTimeout("tcp", c.srvSsoAddrs[0], time.Second*20) conn, err := net.DialTimeout("tcp", c.srvSsoAddrs[0], time.Second*20)
if err != nil { if err != nil {
return errors.Wrap(err, "connect error") return errors.Wrap(err, "connect error")
@ -279,13 +281,10 @@ func (c *QQClient) highwayUploadFileMultiThreadingByBDH(path string, cmdId int32
block := blocks[nextId] block := blocks[nextId]
if block.Id == len(blocks)-1 { if block.Id == len(blocks)-1 {
cond.L.Lock() cond.L.Lock()
for atomic.LoadUint32(&uploadedCount) != uint32(len(blocks)-1) && lastErr == nil { for atomic.LoadUint32(&uploadedCount) != uint32(len(blocks))-1 {
cond.Wait() cond.Wait()
} }
cond.L.Unlock() cond.L.Unlock()
if lastErr != nil {
break
}
} }
buffer = buffer[:blockSize] buffer = buffer[:blockSize]
_, _ = chunk.Seek(block.BeginOffset, io.SeekStart) _, _ = chunk.Seek(block.BeginOffset, io.SeekStart)
@ -347,19 +346,13 @@ func (c *QQClient) highwayUploadFileMultiThreadingByBDH(path string, cmdId int32
} }
return nil return nil
} }
wg := sync.WaitGroup{}
wg.Add(threadCount) group := errgroup.Group{}
for i := 0; i < threadCount; i++ { for i := 0; i < threadCount; i++ {
go func() { group.Go(doUpload)
defer wg.Done()
defer cond.Signal()
if err := doUpload(); err != nil {
lastErr = err
} }
}() err = group.Wait()
} return rspExt, err
wg.Wait()
return rspExt, lastErr
} }
func (c *QQClient) highwaySendHeartbreak(conn net.Conn) error { func (c *QQClient) highwaySendHeartbreak(conn net.Conn) error {

1
go.mod
View File

@ -6,5 +6,6 @@ require (
github.com/json-iterator/go v1.1.10 github.com/json-iterator/go v1.1.10
github.com/modern-go/reflect2 v1.0.1 github.com/modern-go/reflect2 v1.0.1
github.com/pkg/errors v0.9.1 github.com/pkg/errors v0.9.1
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
google.golang.org/protobuf v1.25.0 google.golang.org/protobuf v1.25.0
) )

2
go.sum
View File

@ -52,6 +52,8 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=