Add a semaphore (#21027)
Этот коммит содержится в:
@@ -682,7 +682,7 @@ func (a *App) UploadFileX(c *request.Context, channelID, name string, input io.R
|
|||||||
|
|
||||||
if *a.Config().FileSettings.ExtractContent {
|
if *a.Config().FileSettings.ExtractContent {
|
||||||
infoCopy := *t.fileinfo
|
infoCopy := *t.fileinfo
|
||||||
a.Srv().Go(func() {
|
a.Srv().GoBuffered(func() {
|
||||||
err := a.ExtractContentFromFileInfo(&infoCopy)
|
err := a.ExtractContentFromFileInfo(&infoCopy)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
mlog.Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
|
mlog.Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
|
||||||
@@ -932,7 +932,7 @@ func (a *App) DoUploadFileExpectModification(c request.CTX, now time.Time, rawTe
|
|||||||
|
|
||||||
if *a.Config().FileSettings.ExtractContent {
|
if *a.Config().FileSettings.ExtractContent {
|
||||||
infoCopy := *info
|
infoCopy := *info
|
||||||
a.Srv().Go(func() {
|
a.Srv().GoBuffered(func() {
|
||||||
err := a.ExtractContentFromFileInfo(&infoCopy)
|
err := a.ExtractContentFromFileInfo(&infoCopy)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
mlog.Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
|
mlog.Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
|
||||||
|
|||||||
@@ -134,6 +134,7 @@ type Server struct {
|
|||||||
|
|
||||||
goroutineCount int32
|
goroutineCount int32
|
||||||
goroutineExitSignal chan struct{}
|
goroutineExitSignal chan struct{}
|
||||||
|
goroutineBuffered chan struct{}
|
||||||
|
|
||||||
EmailService email.ServiceInterface
|
EmailService email.ServiceInterface
|
||||||
|
|
||||||
@@ -208,6 +209,7 @@ func NewServer(options ...Option) (*Server, error) {
|
|||||||
|
|
||||||
s := &Server{
|
s := &Server{
|
||||||
goroutineExitSignal: make(chan struct{}, 1),
|
goroutineExitSignal: make(chan struct{}, 1),
|
||||||
|
goroutineBuffered: make(chan struct{}, runtime.NumCPU()),
|
||||||
RootRouter: rootRouter,
|
RootRouter: rootRouter,
|
||||||
LocalRouter: localRouter,
|
LocalRouter: localRouter,
|
||||||
WebSocketRouter: &WebSocketRouter{
|
WebSocketRouter: &WebSocketRouter{
|
||||||
@@ -1037,6 +1039,26 @@ func (s *Server) Go(f func()) {
|
|||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GoBuffered acts like a semaphore which creates a goroutine, but maintains a record of it
|
||||||
|
// to ensure that execution completes before the server is shutdown.
|
||||||
|
func (s *Server) GoBuffered(f func()) {
|
||||||
|
s.goroutineBuffered <- struct{}{}
|
||||||
|
|
||||||
|
atomic.AddInt32(&s.goroutineCount, 1)
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
f()
|
||||||
|
|
||||||
|
atomic.AddInt32(&s.goroutineCount, -1)
|
||||||
|
select {
|
||||||
|
case s.goroutineExitSignal <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
<-s.goroutineBuffered
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
// WaitForGoroutines blocks until all goroutines created by App.Go exit.
|
// WaitForGoroutines blocks until all goroutines created by App.Go exit.
|
||||||
func (s *Server) WaitForGoroutines() {
|
func (s *Server) WaitForGoroutines() {
|
||||||
for atomic.LoadInt32(&s.goroutineCount) != 0 {
|
for atomic.LoadInt32(&s.goroutineCount) != 0 {
|
||||||
|
|||||||
Ссылка в новой задаче
Block a user