git.sesse.net Git - casparcg/blob - common/polling_filesystem_monitor.cpp

   1 /*
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>
   3 *
   4 * This file is part of CasparCG (www.casparcg.com).
   5 *
   6 * CasparCG is free software: you can redistribute it and/or modify
   7 * it under the terms of the GNU General Public License as published by
   8 * the Free Software Foundation, either version 3 of the License, or
   9 * (at your option) any later version.
  10 *
  11 * CasparCG is distributed in the hope that it will be useful,
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 * GNU General Public License for more details.
  15 *
  16 * You should have received a copy of the GNU General Public License
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
  18 *
  19 * Author: Helge Norberg, helge.norberg@svt.se
  20 */
  21
  22 #include "stdafx.h"
  23
  24 #include "polling_filesystem_monitor.h"
  25
  26 #include <map>
  27 #include <set>
  28 #include <iostream>
  29
  30 #include <boost/thread.hpp>
  31 #include <boost/foreach.hpp>
  32 #include <boost/range/adaptor/map.hpp>
  33 #include <boost/range/algorithm/copy.hpp>
  34 #include <boost/filesystem/fstream.hpp>
  35 #include <boost/filesystem/convenience.hpp>
  36
  37 #include <tbb/atomic.h>
  38 #include <tbb/concurrent_queue.h>
  39
  40 #include "executor.h"
  41
  42 namespace caspar {
  43
  44 class exception_protected_handler
  45 {
  46         filesystem_monitor_handler handler_;
  47 public:
  48         exception_protected_handler(const filesystem_monitor_handler& handler)
  49                 : handler_(handler)
  50         {
  51         }
  52
  53         void operator()(filesystem_event event, const boost::filesystem::wpath& file)
  54         {
  55                 try
  56                 {
  57                         boost::this_thread::interruption_point();
  58                         handler_(event, file);
  59                         boost::this_thread::interruption_point();
  60                 }
  61                 catch (const boost::thread_interrupted&)
  62                 {
  63                         throw;
  64                 }
  65                 catch (...)
  66                 {
  67                         CASPAR_LOG_CURRENT_EXCEPTION();
  68                 }
  69         }
  70 };
  71
  72 class directory_monitor
  73 {
  74         bool report_already_existing_;
  75         boost::filesystem::wpath folder_;
  76         filesystem_event events_mask_;
  77         filesystem_monitor_handler handler_;
  78         initial_files_handler initial_files_handler_;
  79         bool first_scan_;
  80         std::map<boost::filesystem::wpath, std::time_t> files_;
  81         std::map<boost::filesystem::wpath, uintmax_t> being_written_sizes_;
  82 public:
  83         directory_monitor(
  84                         bool report_already_existing,
  85                         const boost::filesystem::wpath& folder,
  86                         filesystem_event events_mask,
  87                         const filesystem_monitor_handler& handler,
  88                         const initial_files_handler& initial_files_handler)
  89                 : report_already_existing_(report_already_existing)
  90                 , folder_(folder)
  91                 , events_mask_(events_mask)
  92                 , handler_(exception_protected_handler(handler))
  93                 , initial_files_handler_(initial_files_handler)
  94                 , first_scan_(true)
  95         {
  96         }
  97
  98         void reemmit_all()
  99         {
 100                 if ((events_mask_ & MODIFIED) == 0)
 101                         return;
 102
 103                 BOOST_FOREACH(auto& file, files_)
 104                         handler_(MODIFIED, file.first);
 105         }
 106
 107         void reemmit(const boost::filesystem::wpath& file)
 108         {
 109                 if ((events_mask_ & MODIFIED) == 0)
 110                         return;
 111
 112                 if (files_.find(file) != files_.end() && boost::filesystem::exists(file))
 113                         handler_(MODIFIED, file);
 114         }
 115
 116         void scan(const boost::function<bool ()>& should_abort)
 117         {
 118                 static const std::time_t NO_LONGER_WRITING_AGE = 3; // Assume std::time_t is expressed in seconds
 119                 using namespace boost::filesystem;
 120
 121                 bool interested_in_removed = (events_mask_ & REMOVED) > 0;
 122                 bool interested_in_created = (events_mask_ & CREATED) > 0;
 123                 bool interested_in_modified = (events_mask_ & MODIFIED) > 0;
 124
 125                 std::set<wpath> removed_files;
 126                 boost::copy(
 127                                 files_ | boost::adaptors::map_keys,
 128                                 std::insert_iterator<decltype(removed_files)>(removed_files, removed_files.end()));
 129
 130                 std::set<wpath> initial_files;
 131
 132                 for (boost::filesystem3::wrecursive_directory_iterator iter(folder_); iter != boost::filesystem3::wrecursive_directory_iterator(); ++iter)
 133                 {
 134                         if (should_abort())
 135                                 return;
 136
 137                         auto& path = iter->path();
 138
 139                         if (is_directory(path))
 140                                 continue;
 141
 142                         auto now = std::time(nullptr);
 143                         std::time_t current_mtime;
 144
 145                         try
 146                         {
 147                                 current_mtime = last_write_time(path);
 148                         }
 149                         catch (...)
 150                         {
 151                                 // Probably removed, will be captured the next round.
 152                                 continue;
 153                         }
 154
 155                         auto time_since_written_to = now - current_mtime;
 156                         bool no_longer_being_written_to = time_since_written_to >= NO_LONGER_WRITING_AGE;
 157                         auto previous_it = files_.find(path);
 158                         bool already_known = previous_it != files_.end();
 159
 160                         if (already_known && no_longer_being_written_to)
 161                         {
 162                                 bool modified = previous_it->second != current_mtime;
 163
 164                                 if (modified && can_read_file(path))
 165                                 {
 166                                         if (interested_in_modified)
 167                                                 handler_(MODIFIED, path);
 168
 169                                         files_[path] = current_mtime;
 170                                         being_written_sizes_.erase(path);
 171                                 }
 172                         }
 173                         else if (no_longer_being_written_to && can_read_file(path))
 174                         {
 175                                 if (interested_in_created && (report_already_existing_ || !first_scan_))
 176                                         handler_(CREATED, path);
 177
 178                                 if (first_scan_)
 179                                         initial_files.insert(path);
 180
 181                                 files_.insert(std::make_pair(path, current_mtime));
 182                                 being_written_sizes_.erase(path);
 183                         }
 184
 185                         removed_files.erase(path);
 186                 }
 187
 188                 BOOST_FOREACH(auto& path, removed_files)
 189                 {
 190                         files_.erase(path);
 191                         being_written_sizes_.erase(path);
 192
 193                         if (interested_in_removed)
 194                                 handler_(REMOVED, path);
 195                 }
 196
 197                 if (first_scan_)
 198                         initial_files_handler_(initial_files);
 199
 200                 first_scan_ = false;
 201         }
 202 private:
 203         bool can_read_file(const boost::filesystem::wpath& file)
 204         {
 205                 boost::filesystem::wifstream stream(file);
 206
 207                 return stream.is_open();
 208         }
 209 };
 210
 211 class polling_filesystem_monitor : public filesystem_monitor
 212 {
 213         directory_monitor root_monitor_;
 214         boost::thread scanning_thread_;
 215         tbb::atomic<bool> running_;
 216         int scan_interval_millis_;
 217         boost::promise<void> initial_scan_completion_;
 218         tbb::concurrent_queue<boost::filesystem::wpath> to_reemmit_;
 219         tbb::atomic<bool> reemmit_all_;
 220 public:
 221         polling_filesystem_monitor(
 222                         const boost::filesystem::wpath& folder_to_watch,
 223                         filesystem_event events_of_interest_mask,
 224                         bool report_already_existing,
 225                         int scan_interval_millis,
 226                         const filesystem_monitor_handler& handler,
 227                         const initial_files_handler& initial_files_handler)
 228                 : root_monitor_(report_already_existing, folder_to_watch, events_of_interest_mask, handler, initial_files_handler)
 229                 , scan_interval_millis_(scan_interval_millis)
 230         {
 231                 running_ = true;
 232                 reemmit_all_ = false;
 233                 scanning_thread_ = boost::thread([this] { scanner(); });
 234         }
 235
 236         virtual ~polling_filesystem_monitor()
 237         {
 238                 running_ = false;
 239                 scanning_thread_.interrupt();
 240                 scanning_thread_.join();
 241         }
 242
 243         virtual boost::unique_future<void> initial_files_processed()
 244         {
 245                 return initial_scan_completion_.get_future();
 246         }
 247
 248         virtual void reemmit_all()
 249         {
 250                 reemmit_all_ = true;
 251         }
 252
 253         virtual void reemmit(const boost::filesystem::wpath& file)
 254         {
 255                 to_reemmit_.push(file);
 256         }
 257 private:
 258         void scanner()
 259         {
 260                 win32_exception::install_handler();
 261
 262                 //detail::SetThreadName(GetCurrentThreadId(), "polling_filesystem_monitor");
 263
 264                 bool running = scan(false);
 265                 initial_scan_completion_.set_value();
 266
 267                 if (running)
 268                         while (scan(true));
 269         }
 270
 271         bool scan(bool sleep)
 272         {
 273                 try
 274                 {
 275                         if (sleep)
 276                                 boost::this_thread::sleep(boost::posix_time::milliseconds(scan_interval_millis_));
 277
 278                         if (reemmit_all_.fetch_and_store(false))
 279                                 root_monitor_.reemmit_all();
 280                         else
 281                         {
 282                                 boost::filesystem::wpath file;
 283
 284                                 while (to_reemmit_.try_pop(file))
 285                                         root_monitor_.reemmit(file);
 286                         }
 287
 288                         root_monitor_.scan([=] { return !running_; });
 289                 }
 290                 catch (const boost::thread_interrupted&)
 291                 {
 292                 }
 293                 catch (...)
 294                 {
 295                         CASPAR_LOG_CURRENT_EXCEPTION();
 296                 }
 297
 298                 return running_;
 299         }
 300 };
 301
 302 struct polling_filesystem_monitor_factory::impl
 303 {
 304         int scan_interval_millis;
 305
 306         impl(int scan_interval_millis)
 307                 : scan_interval_millis(scan_interval_millis)
 308         {
 309         }
 310 };
 311
 312 polling_filesystem_monitor_factory::polling_filesystem_monitor_factory(int scan_interval_millis)
 313         : impl_(new impl(scan_interval_millis))
 314 {
 315 }
 316
 317 polling_filesystem_monitor_factory::~polling_filesystem_monitor_factory()
 318 {
 319 }
 320
 321 filesystem_monitor::ptr polling_filesystem_monitor_factory::create(
 322                 const boost::filesystem::wpath& folder_to_watch,
 323                 filesystem_event events_of_interest_mask,
 324                 bool report_already_existing,
 325                 const filesystem_monitor_handler& handler,
 326                 const initial_files_handler& initial_files_handler)
 327 {
 328         return spl::make_shared<polling_filesystem_monitor>(
 329                         folder_to_watch,
 330                         events_of_interest_mask,
 331                         report_already_existing,
 332                         impl_->scan_interval_millis,
 333                         handler,
 334                         initial_files_handler);
 335 }
 336
 337 }