diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index 8fa4ef8a..5b88d957 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -349,6 +349,7 @@ func (a *serverApp) run(ctx context.Context) error { log.Printf("[WARN] failed to close avatar store, %s", e) } a.notifyService.Close() + a.imageService.Close() log.Print("[INFO] shutdown completed") }() diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go index cd36233b..ef5108d7 100644 --- a/backend/app/rest/api/rest_test.go +++ b/backend/app/rest/api/rest_test.go @@ -234,12 +234,14 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) { Cache: memCache, WebRoot: "/tmp", RemarkURL: "https://demo.remark42.com", - ImageService: &image.FileSystem{ - Location: "/tmp/pics-remark42", - Partitions: 100, - MaxSize: 10000, + ImageService: &image.Service{ + Store: &image.FileSystem{ + Location: "/tmp/pics-remark42", + Partitions: 100, + MaxSize: 10000, + }, + TTL: time.Millisecond * 100, }, - ImageProxy: &proxy.Image{}, ReadOnlyAge: 10, CommentFormatter: store.NewCommentFormatter(&proxy.Image{}), diff --git a/backend/app/store/image/image.go b/backend/app/store/image/image.go index 08097f00..8c5b620f 100644 --- a/backend/app/store/image/image.go +++ b/backend/app/store/image/image.go @@ -10,6 +10,8 @@ import ( "io" "log" "strings" + "sync" + "sync/atomic" "time" "github.com/PuerkitoBio/goquery" @@ -30,6 +32,18 @@ type Service struct { Store TTL time.Duration // for how long file allowed on staging ImageAPI string // image api matching path + + wg sync.WaitGroup + submitCh chan submitReq + once sync.Once + term int32 +} + +const submitQueueSize = 5000 + +type submitReq struct { + ID string + TS time.Time } // Submit multiple ids for delayed commit @@ -38,13 +52,27 @@ func (s *Service) Submit(ids []string) { return } - time.AfterFunc(s.TTL, func() { - for _, id := range ids { - if err := s.Commit(id); err != nil { - log.Printf("[WARN] failed to commit image %s", id) + s.once.Do(func() { + s.submitCh = make(chan submitReq, submitQueueSize) + s.wg.Add(1) + go func() { + defer s.wg.Done() + for req := range s.submitCh { + // wait for TTL expiration with emergency pass on term + for atomic.LoadInt32(&s.term) == 0 && time.Since(req.TS) <= s.TTL { + time.Sleep(time.Millisecond * 10) // small sleep to relive busy wait but keep reactive for term (close) + } + if err := s.Commit(req.ID); err != nil { + log.Printf("[WARN] failed to commit image %s", req.ID) + } } - } + log.Printf("[INFO] image submiter terminated") + }() }) + + for _, id := range ids { + s.submitCh <- submitReq{ID: id, TS: time.Now()} + } } // ExtractPictures gets list of images from the doc html and convert from urls to ids, i.e. user/pic.png @@ -86,3 +114,13 @@ func (s *Service) Cleanup(ctx context.Context) { } } } + +// Close flushes all in-progress submits and enforces waiting commits +func (s *Service) Close() { + log.Printf("[INFO] close image service ") + atomic.AddInt32(&s.term, 1) // enforce non-delayed commits for all ids left in submitCh + if s.submitCh != nil { + close(s.submitCh) + } + s.wg.Wait() +} diff --git a/backend/app/store/image/image_test.go b/backend/app/store/image/image_test.go index ad12dae5..335327fa 100644 --- a/backend/app/store/image/image_test.go +++ b/backend/app/store/image/image_test.go @@ -21,6 +21,19 @@ func TestService_ExtractPictures(t *testing.T) { assert.Equal(t, "user2/pic3.png", ids[1]) } +func TestService_Cleanup(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + store := NewMockStore(ctrl) + store.EXPECT().Cleanup(gomock.Any(), gomock.Any()).Times(10) + + svc := Service{Store: store, TTL: 100 * time.Millisecond} + ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*549) + defer cancel() + svc.Cleanup(ctx) +} + func TestService_Submit(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -34,11 +47,27 @@ func TestService_Submit(t *testing.T) { time.Sleep(time.Millisecond * 500) } -func TestService_SubmitDelay(t *testing.T) { +func TestService_Close(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() store := NewMockStore(ctrl) + store.EXPECT().Commit(gomock.Any()).Times(5) // all 5 should be committed + svc := Service{Store: store, ImageAPI: "/blah/", TTL: time.Millisecond * 500} + svc.Submit([]string{"id1", "id2", "id3"}) + svc.Submit([]string{"id4", "id5"}) + svc.Submit(nil) + svc.Close() +} + +func TestService_SubmitDelay(t *testing.T) { + ctrl := gomock.NewController(t) + defer func() { + ctrl.Finish() + }() + + store := NewMockStore(ctrl) + store.EXPECT().Commit(gomock.Any()).Times(3) // first batch should be committed svc := Service{Store: store, ImageAPI: "/blah/", TTL: time.Millisecond * 100} svc.Submit([]string{"id1", "id2", "id3"}) @@ -46,16 +75,3 @@ func TestService_SubmitDelay(t *testing.T) { svc.Submit([]string{"id4", "id5"}) svc.Submit(nil) } - -func TestService_Cleanup(t *testing.T) { - ctrl := gomock.NewController(t) - defer ctrl.Finish() - - store := NewMockStore(ctrl) - store.EXPECT().Cleanup(gomock.Any(), gomock.Any()).Times(10) - - svc := Service{Store: store, TTL: 100 * time.Millisecond} - ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*550) - defer cancel() - svc.Cleanup(ctx) -}