{"id":941,"date":"2024-06-25T11:19:58","date_gmt":"2024-06-25T09:19:58","guid":{"rendered":"https:\/\/blog.adameczek.pl\/?p=941"},"modified":"2024-06-26T13:32:12","modified_gmt":"2024-06-26T11:32:12","slug":"chyba-cos-wpadlo-nowego-czyli-nasluchiwanie-na-nowe-pliki-przez-kontener","status":"publish","type":"post","link":"https:\/\/blog.adameczek.pl\/index.php\/2024\/06\/25\/chyba-cos-wpadlo-nowego-czyli-nasluchiwanie-na-nowe-pliki-przez-kontener\/","title":{"rendered":"Widz\u0119 ciemno\u015b\u0107! Czyli aplikacja w kontenerze nas\u0142uchuje na nowe pliki"},"content":{"rendered":"\n

<\/p>\n\n\n\n

Czasem aplikacja ma reagowa\u0107 na zmiany zachodz\u0105ce w systemie plik\u00f3w aby, np. przetwarza\u0107 nowe pliki, kt\u00f3re w\u0142a\u015bnie kto\u015b do niego doda\u0142. Nadaje si\u0119 do tego \u015bwietnie klasa FileSystemWatcher<\/a> (FSW). Dzia\u0142a w Windows i pod przy odpowiedniej konfiguracji tak\u017ce pod Linuxem. Ale s\u0142odycz si\u0119 ko\u0144czy po skonteneryzowaniu aplikacji. Trafi\u0142 mi si\u0119 w\u0142a\u015bnie taki przypadek. Kontener nie dostaje notyfikacji o zmianach w systemie plik\u00f3w woluminu zamontowanego do kontenera. Zupe\u0142na cisza, albo cytuj\u0105c klasyka \u201eCiemno\u015b\u0107, widz\u0119 ciemno\u015b\u0107!\u201d<\/p>\n\n\n\n

Przyczyn\u0105 tego stanu rzeczy s\u0105 r\u00f3\u017cne mechanizmy generowania i transmisji zdarze\u0144 o zmianie w systemie plik\u00f3w przez r\u00f3\u017cne OS, systemy plik\u00f3w, protoko\u0142y sieciowe. FileSystemWatcher instancjonuje obiekt inotify dla ka\u017cdego \u015bledzonego folderu, kt\u00f3ry nas\u0142uchuje na zdarzenia generowane przez system plik\u00f3w. Je\u015bli jednak nast\u0105pi b\u0142\u0105d sieci, to zdarzenie nie dotrze. Je\u015bli przepe\u0142ni si\u0119 bufor zdarze\u0144, to zdarzenie nie dotrze, Je\u015bli system plik\u00f3w nie potrafi w spos\u00f3b powtarzalny wys\u0142a\u0107 powiadomie\u0144, to one nie b\u0119d\u0105 dociera\u0142y. Tych je\u017celi\u2026 jest sporo i zawsze ko\u0144cz\u0105 si\u0119 tym, \u017ce zdarzenie nie dotrze. FSW sprawdza si\u0119 w zasadzie jedynie dla dysk\u00f3w lokalnych i to najlepiej pod Windows.<\/p>\n\n\n\n

A tymczasem w folderze jest coraz wi\u0119cej plik\u00f3w, a aplikacja nie reaguje. Co robi\u0107? Pooling udzia\u0142u sieciowego? Nic innego nie pozostaje.<\/p>\n\n\n\n

Mo\u017cna do tego zastosowa\u0107 PhysicalFileProvider<\/em>. W odr\u00f3\u017cnieniu od FSW, nie pozwala on na wyb\u00f3r rodzaju zdarze\u0144, kt\u00f3re notyfikuje, ale powinien wystarczy\u0107 przynajmniej jako \u017ar\u00f3d\u0142o informacji, \u017ce \u201eco\u015b si\u0119 zmieni\u0142o\u201d. Po prostu zostanie ustawiony cykliczny pooling folderu. Nie ma innego wyj\u015bcia w Dockerze. Sami b\u0119dziemy musieli dowiedzie\u0107 si\u0119, jak zmiana zasz\u0142a. Jedynie co wiadomo, to \u017ce zasz\u0142a. Ale dobre i to.<\/p>\n\n\n\n

Za to wykonamy porz\u0105dn\u0105 implementacj\u0119 obs\u0142ugi kolejki zdarze\u0144. Przecie\u017c mo\u017ce si\u0119 zdarzy\u0107, \u017ce zdarzenia o nowych plikach b\u0119d\u0105 sp\u0142ywa\u0142y szybciej ni\u017c trwa ich obs\u0142uga. Wykorzystamy generyczn\u0105 BlockingCollection <\/em>do jednoczesnego dodawania i konsumowania zdarze\u0144. To kolejka FIFO, kt\u00f3ra jest bezpieczna dla wielu w\u0105tk\u00f3w (thread-safe). Jeden w\u0105tek mo\u017ce dodawa\u0107 zdarzenia informuj\u0105ce o nowych plikach, a inny obs\u0142ugiwa\u0107 te zdarzenia. Nasz DirectoryWatcher ma nast\u0119puj\u0105cy interfejs:<\/p>\n\n\n\n

public interface IDirectoryWatcher\n{\n    void RegisterCallback(Action callback);\n    Task StartWatching(CancellationToken cancellationToken);\n    void Dispose();\n}<\/code><\/pre><\/div>\n\n\n\n
Po zainstancjonowaniu DirectoryWatchera <\/em>rejestruje si\u0119 callback. Jest on wo\u0142any w momencie obs\u0142ugi zdarzenia pobieranego z kolejki. W przyk\u0142adzie callbackiem jest medoda synchroniczna. Jest ona wo\u0142ana w osobnym w\u0105tku, tym samym kt\u00f3ry obs\u0142uguje kolejk\u0119. Obs\u0142uga callbacka nie nie blokuje dzia\u0142ania pozosta\u0142ego kodu. W razie potrzeby mo\u017cna \u0142atwo zmieni\u0107 kod, aby wywo\u0142anie by\u0142o asynchroniczne.<\/p>\n\n\n\n
Potem wystarczy uruchomi\u0107 DirectoryWatcher<\/em> wywo\u0142uj\u0105c StartWatching<\/em>. Metoda wygl\u0105da tak:<\/p>\n\n\n\n
public async Task StartWatching(CancellationToken ct = default)\n{\n    _fileSystemEventBuffer = new BlockingCollection(EventBufferSize);\n\n    CreateFileWatcher();\n    _fileWatcherCallback = WatchForFileChanges();\n\n    var processBuffer = Task.Run(() =>\n    {\n        foreach (var fileInfo in _fileSystemEventBuffer.GetConsumingEnumerable(ct))\n        {\n            _callback?.Invoke(fileInfo.PhysicalPath!);\n            LogWhenBufferIsEmpty();\n        }\n    }, ct);\n\n    \/\/ Implement resilience to transient IO errors\n    _files = GetFiles();\n    BufferFiles();\n\n    await processBuffer.ConfigureAwait(false);\n}<\/code><\/pre><\/div>\n\n\n\n
Deklaracje zmiennych i sta\u0142ych na razie pomin\u0119. Zostan\u0105 pokazane w dalszej cz\u0119\u015bci. Podobnie metody CreateFileWatcher()<\/em> i WatchForFileChanges()<\/em>. Teraz opisz\u0119 dzia\u0142anie typu BlockingCollection<\/em>. Tworzymy bufor new BlockingCollection(EventBufferSize)<\/code>. W linii 8 uruchamiany jest w\u0105tek, kt\u00f3ry pobiera z kolejki informacje o nowych plikach i wywo\u0142uje callback. Kluczowe tu jest, \u017ce p\u0119tla si\u0119 \u201enie ko\u0144czy\u201d. Po skonsumowaniu wszyskich zdarze\u0144 zostaje \u201eu\u015bpiona\u201d i wznawia dzia\u0142anie kiedy do kolekcji wpadn\u0105 nowe elementy. A mog\u0105 one by\u0107 dodawane w ka\u017cdym momencie w innym w\u0105tku. Mo\u017cemy zatem spokojnie w swoim tempie obs\u0142ugiwa\u0107 zdarzenia nie martwi\u0105c si\u0119 o zablokowanie mo\u017cliwo\u015bci ich dodawania. W linii 18 pobieramy informacje o plikach, bo na starcie programu zak\u0142adamy, \u017ce wszystkie pliki s\u0105 nowe. Definicja metody GetFiles():<\/p>\n\n\n\n
private List GetFiles() => _fileWatcher!.GetDirectoryContents(string.Empty).ToList();<\/code><\/pre><\/div>\n\n\n\n
Nast\u0119pnie wszyskie dane wrzucamy do bufora (linia 19). Kiedy tylko zostanie dodany pierwszy rekord, wznawia prac\u0119 p\u0119tla foreach z linii 10 i wywo\u0142ywany jest callback.<\/p>\n\n\n\n
private void BufferFiles()\n{\n    foreach (var fileInfo in _files!)\n    {\n        if (!_fileSystemEventBuffer!.TryAdd(fileInfo))\n            Console.WriteLine($"Buffer size exceeded ({EventBufferSize}) or buffer is disposed");\n    }\n}<\/code><\/pre><\/div>\n\n\n\n
I do by by\u0142o na tyle, je\u015bli idzie o mechanizm bufora. Wielow\u0105tkowo dodajemy do niego i konsumujemy obiekty. Dla reszty kodu dzia\u0142anie \u201eg\u0142\u00f3wnej p\u0119tli\u201d odbywa si\u0119 \u201ew tle\u201d nie wp\u0142ywaj\u0105c na prac\u0119 innych w\u0105tk\u00f3w. Nie jest to w\u0105tek t\u0142a, ale dzi\u0119ki asynchroniczno\u015bci i bezpiecznemu dla w\u0105tk\u00f3w api BlockingCollection <\/em>mo\u017cemy tak roboczo przyj\u0105\u0107.<\/p>\n\n\n\n
Do om\u00f3wienia zosta\u0142 mechanizm notyfikacji o nowych plikach w folderze. Na pocz\u0105tek metoda CreateFileWatcher()<\/em> wo\u0142ana w linii 5 metody StartWatching()<\/em>. Instancjonujemy PhysicalFileProvider<\/em>, kt\u00f3ry nie b\u0119dzie bra\u0142 pod uwag\u0119 plik\u00f3w ukrytych (tak\u017ce z kropk\u0105) ani systemowych. B\u0119dzie za to cyklicznie si\u0119ga\u0142 do folderu na okoliczno\u015b\u0107 wykrycia zmian w systemie plik\u00f3w. Nic specjalnego.<\/p>\n\n\n\n
private void CreateFileWatcher()\n{\n    _fileWatcher = new PhysicalFileProvider(_directoryToWatch, ExclusionFilters.Sensitive)\n    {\n        UsePollingFileWatcher = true,\n        UseActivePolling = true\n    };\n}<\/code><\/pre><\/div>\n\n\n\n
Za to metoda WatchForFileChanges()<\/em> a w\u0142a\u015bciwie to co ona zapocz\u0105tkowuje jest bardziej interesuj\u0105ce.<\/p>\n\n\n\n
private static ulong _changeLevel;\nprivate IChangeToken? _changeToken;\nprivate IDisposable? _fileWatcherCallback;\n\nprivate IDisposable WatchForFileChanges()\n{\n    _changeToken = _fileWatcher!.Watch("**\/*.*");\n    return _changeToken.RegisterChangeCallback(_ => NotifyFileChange(), default);\n}\n\nprivate void NotifyFileChange()\n{\n    Console.WriteLine("Directory has changed. Callback invoked");\n\n    _fileWatcherCallback = WatchForFileChanges();\n\n    if (0 == Interlocked.CompareExchange(ref _changeLevel, 1, 0))\n        BufferNewFiles();\n    else\n        Interlocked.Exchange(ref _changeLevel, 2);\n}<\/code><\/pre><\/div>\n\n\n\n
W linii 7 wskazujemy jakie pliki nas interesuj\u0105 u\u017cywaj\u0105c filtra. Metoda Watch()<\/em> zwraca IChangeToken<\/em>, kt\u00f3ry b\u0119dzie notyfikowany o zmianach \u015bledzonych plik\u00f3w. Nast\u0119pnie temu tokenowi wskazujemy callback jaki ma wywo\u0142a\u0107 w wyniku notyfikacji (NotifyFileChange<\/em>). W moich pr\u00f3bach po jednorazowej notyfikacji IChangeToken<\/em> przestawa\u0142 by\u0107 u\u017cyteczny. Albo przestawa\u0142 by\u0107 notyfikowany o zmianach, albo rejestacja callbacka wygasa\u0142a. W ka\u017cdym razie w linii 15 ponownie wo\u0142am metod\u0119 WatchForFileChanges()<\/em>, aby notyfikacje zn\u00f3w dzia\u0142a\u0142y. A chcemy, \u017ceby dzia\u0142a\u0142y nieustannie nas\u0142uchuj\u0105c na nowe pliki.<\/p>\n\n\n\n
Teraz robi si\u0119 ciekawiej. Trzeba jako\u015b ogarn\u0105\u0107 r\u00f3\u017cne stany w jakich mo\u017ce si\u0119 znale\u017a\u0107 Watcher<\/em>. Wyr\u00f3\u017cni\u0142em trzy stany:<\/p>\n\n\n\n