/[unison]/trunk/src/fsmonitor.py
ViewVC logotype

Annotation of /trunk/src/fsmonitor.py

Parent Directory Parent Directory | Revision Log Revision Log


Revision 471 - (hide annotations)
Wed Oct 27 18:23:28 2010 UTC (9 years, 9 months ago) by bcpierce
File MIME type: text/x-python
File size: 23885 byte(s)
* Incorporated new version of fsmonitor.py from Christophe Gohle

1 bcpierce 450 #!/usr/bin/python
2    
3     # a small program to test the possibilities to monitor the file system and
4     # log changes on Windowsm Linux, and OSX
5     #
6     # Originally written by Christoph Gohle (2010)
7     # Modified by Gene Horodecki for Windows
8     # Further modified by Benjamin Pierce
9     # should be distributed under GPL
10    
11     import sys
12     import os
13     import stat
14     from optparse import OptionParser
15     from time import time
16    
17     def mydebug(fmt, *args, **kwds):
18     if not op.debug:
19     return
20    
21     if args:
22     fmt = fmt % args
23    
24     elif kwds:
25     fmt = fmt % kwds
26    
27     print >>sys.stderr, fmt
28    
29     def mymesg(fmt, *args, **kwds):
30     if not op.verbose:
31     return
32    
33     if args:
34     fmt = fmt % args
35    
36     elif kwds:
37     fmt = fmt % kwds
38    
39     print >>sys.stdout, fmt
40    
41     def timer_callback(timer, streamRef):
42     mydebug("CFAbsoluteTimeGetCurrent() => %.3f", CFAbsoluteTimeGetCurrent())
43     mydebug("FSEventStreamFlushAsync(streamRef = %s)", streamRef)
44     FSEventStreamFlushAsync(streamRef)
45    
46     def update_changes(result):
47     mydebug('Update_changes: absresult = %s',result)
48     #print('absresult',result)
49     result = [mangle_filename(path) for path in result]
50     mydebug('Update_changes: mangled = %s',result)
51     #print('magnled', result)
52     result = [relpath(op.root,path) for path in result]
53     #print('relative to root',result)
54     mydebug('Update_changes: relative to root = %s',result)
55    
56     try:
57     f = open(op.absoutfile,'a')
58     for path in result:
59     f.write(path+'\n')
60     f.close()
61     except IOError:
62     mymesg('failed to open log file %s for writing',op.outfile)
63    
64     def update_changes_nomangle(result):
65     # In win32 there are no symlinks, therefore file mangling
66     # is not required
67    
68     mydebug('Changed paths: %s\n',result)
69     try:
70     f = open(op.absoutfile,'a')
71     f.write(result+'\n')
72     f.close()
73     except IOError:
74     mymesg('failed to open log file %s for writing',op.outfile)
75    
76     def mangle_filename(path):
77 bcpierce 471 """because the FSEvents system returns 'real' paths we have to figure out
78 bcpierce 450 if they have been aliased by a symlink and a 'follow' directive in the unison
79     configuration or from the command line.
80     This is done here for path. The return value is the path name using symlinks
81     """
82 bcpierce 471 try:
83     op.symlinks
84     except AttributeError:
85     make_symlinks()
86     #now lets do it
87     result = path
88     for key in op.symlinks:
89     #print path, key
90     if path.startswith(key):
91     result = os.path.join(op.root,os.path.join(op.symlinks[key]+path[len(key):]))
92     #print 'Match!', result
93 bcpierce 450
94 bcpierce 471 return result
95    
96     def make_symlinks():
97     #lets create a dictionary of symlinks that are treated transparently here
98     op.symlinks = {}
99     fl = op.follow
100     try:
101     foll = [f.split(' ',1) for f in fl]
102     except TypeError:
103     foll = []
104     for k,v in foll:
105     if not k=='Path':
106     mymesg('We don\'t support anything but path specifications in follow directives. Especially not %s',k)
107     else:
108     p = v.strip('{}')
109     if not p[-1]=='/':
110     p+='/'
111     op.symlinks[os.path.realpath(os.path.join(op.root,p))]=p
112     mydebug('make_symlinks: symlinks to follow %s',op.symlinks)
113 bcpierce 450
114    
115     def relpath(root,path):
116 bcpierce 471 """returns the path relative to root (which should be absolute)
117     if it is not a path below root or if root is not absolute it returns None
118     """
119 bcpierce 450
120 bcpierce 471 if not os.path.isabs(root):
121     return None
122 bcpierce 450
123 bcpierce 471 abspath = os.path.abspath(path)
124     mydebug('relpath: abspath(%s) = %s', path, abspath)
125 bcpierce 450
126 bcpierce 471 #make sure the root and abspath both end with a '/'
127     if not root[-1]=='/':
128     root += '/'
129     if not abspath[-1]=='/':
130     abspath += '/'
131 bcpierce 450
132 bcpierce 471 mydebug('relpath: root = %s', root)
133 bcpierce 450
134 bcpierce 471 #print root, abspath
135     if not abspath[:len(root)]==root:
136     #print abspath[:len(root)], root
137     return None
138     mydebug('relpath: relpath = %s',abspath[len(root):])
139     return abspath[len(root):]
140 bcpierce 450
141     def my_abspath(path):
142     """expand path including shell variables and homedir
143     to the absolute path
144     """
145     return os.path.abspath(os.path.expanduser(os.path.expandvars(path)))
146 bcpierce 471
147     def update_follow(path):
148     """ tries to find a follow directive that matches path
149     and if path refers to a symbolic link the real path of the symbolic
150     link is returned. """
151     try:
152     op.symlinks
153     except AttributeError:
154     make_symlinks()
155     rpath = relpath(op.root, path)
156     mydebug('update_follow: rpath %s', rpath)
157     result = None
158     foll = None
159     for k in op.symlinks:
160     v = op.symlinks[k]
161     if v==rpath:
162     result = os.path.realpath(os.path.abspath(path))
163     foll = v
164     mydebug('update_follow: link %s, real %s',v,result)
165     break
166     if result:
167     op.symlinks[result] = foll
168    
169     return result, foll
170 bcpierce 450
171     def conf_parser(conffilepath, delimiter = '=', dic = {}):
172     """parse the unison configuration file at conffilename and populate a dictionary
173     with configuration options from there. If dic is a dictionary, these options are added to this
174     one (can be used to recursively call this function for include statements)."""
175     try:
176     conffile = open(conffilepath,'r')
177     except IOError:
178 bcpierce 471 mydebug('could not open configuration file at %s',conffilepath)
179 bcpierce 450 return None
180    
181     res = dic
182    
183     for line in conffile:
184     line = line.strip()
185     if len(line)<1 or line[0]=='#':
186     continue
187     elif line.startswith('include'):
188     dn = os.path.dirname(conffilepath)
189     fn = line.split()[1].strip()
190     conf_parser(os.path.join(dn,fn), dic = res)
191     else:
192     k,v=[s.strip() for s in line.split('=',1)]
193     if res.has_key(k):
194     res[k].append(v)
195     else:
196     res[k]=[v]
197     return res
198    
199     ################################################
200     # Linux specific code here
201     ################################################
202     if sys.platform.startswith('linux'):
203     import pyinotify
204    
205     class HandleEvents(pyinotify.ProcessEvent):
206 bcpierce 471 wm = None
207    
208 bcpierce 450 #def process_IN_CREATE(self, event):
209     # print "Creating:", event.pathname
210    
211     #def process_IN_DELETE(self, event):
212     # print "Removing:", event.pathname
213    
214     #def process_IN_MODIFY(self, event):
215     # print "Modifying:", event.pathname
216    
217     # def process_IN_MOVED_TO(self, event):
218     # print "Moved to:", event.pathname
219    
220     # def process_IN_MOVED_FROM(self, event):
221     # print "Moved from:", event.pathname
222    
223     # def process_IN_ATTRIB(self, event):
224     # print "attributes:", event.pathname
225    
226     def process_default(self, event):
227 bcpierce 471 mydebug('process_default: event %s', event)
228     # code for adding dirs is obsolete since there is the auto_add option
229     # if event.dir:
230     # if event.mask&pyinotify.IN_CREATE:
231     # print 'create:', event.pathname , self.add_watch(event.pathname,rec=True)
232     # elif event.mask&pyinotify.IN_DELETE:
233     # print 'remove', event.pathname, self.remove_watch(event.pathname)
234     # pass
235     # elif event.mask&pyinotify.IN_MOVED_FROM:
236     # print 'move from', event.pathname, self.remove_watch(event.pathname, rec=True)
237     # pass
238     # elif event.mask&pyinotify.IN_MOVED_TO:
239     # print 'move to', event.pathname, self.add_watch(event.pathname,rec=True)
240     # else:
241     # pass
242     #handle creation of links that should be followed
243     if os.path.islink(event.pathname):
244     #special handling for links
245     mydebug('process_default: link %s created/changed. Checking for follows', event.pathname)
246     p, l = update_follow(event.pathname)
247     if p:
248     self.add_watch(p,rec=True,auto_add=True)
249     mydebug('process_default: follow link %s to %s',l,p)
250     #TODO: should handle deletion of links that are followed (delete the respective watches)
251 bcpierce 450 update_changes([event.pathname])
252 bcpierce 471
253     def remove_watch(self, pathname, **kwargs):
254     if self.watches.has_key(pathname):
255     return self.wm.rm_watch(self.watches.pop(pathname),**kwargs)
256     return None
257    
258     def add_watch(self, pathname, **kwargs):
259     neww = self.wm.add_watch(pathname, self.mask, **kwargs)
260     self.watches.update(neww)
261     return neww
262    
263     def init_watches(self, abspaths, follows):
264     self.watches = {}
265     for abspath in abspaths:
266     self.watches.update(self.wm.add_watch(abspath,self.mask,rec=True,auto_add=True))
267     #we have to add watches for follow statements since pyinotify does
268     #not do recursion across symlinks
269     make_symlinks()
270     for link in op.symlinks:
271     mydebug('following symbolic link %s',link)
272     if not self.watches.has_key(link):
273     self.watches.update(self.wm.add_watch(link,self.mask,rec=True,auto_add=True))
274    
275     mydebug('init_watches: added paths %s\n based on paths %s\n and follows %s',self.watches,op.abspaths, op.follow)
276 bcpierce 450
277    
278     def linuxwatcher():
279     p = HandleEvents()
280     wm = pyinotify.WatchManager() # Watch Manager
281 bcpierce 471 p.wm = wm
282     p.mask = pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MODIFY | pyinotify.IN_ATTRIB | pyinotify.IN_MOVED_TO | pyinotify.IN_MOVED_FROM # watched events
283 bcpierce 450
284     notifier = pyinotify.Notifier(wm, p)
285 bcpierce 471 p.init_watches(op.abspaths, op.follow)
286 bcpierce 450 notifier.loop()
287 bcpierce 471
288    
289 bcpierce 450 #################################################
290     # END Linux specific code
291     #################################################
292    
293     #################################################
294     # MacOsX specific code
295     #################################################
296     if sys.platform == 'darwin':
297     from FSEvents import *
298     import objc
299    
300     def filelevel_approx(path):
301     """in order to avoid scanning the entire directory including sub
302     directories by unison, we have to say which files have changed. Because
303     this is a stupid program it only checks modification times within the
304     update interval. in case there are no files modified in this interval,
305     the entire directory is listed.
306     A deleted file can not be found like this. Therefore also deletes will
307     trigger a rescan of the directory (including subdirs)
308    
309     The impact of rescans could be limited if one could make
310     unison work nonrecursively.
311     """
312     result = []
313     #make a list of all files in question (all files in path w/o dirs)
314     try:
315     names = os.listdir(path)
316     except os.error, msg:
317     #path does not exist (anymore?). Add it to the results
318     mydebug("adding nonexisting path %s for sync",path)
319     result.append(path)
320     names = None
321    
322     if names:
323     for nm in names:
324     full_path = os.path.join(path,nm)
325     st = os.lstat(full_path)
326     #see if the dir it was modified recently
327     if st.st_mtime>time()-float(op.latency):
328     result.append(full_path)
329    
330     if result == []:
331     result.append(path)
332    
333     return result
334    
335    
336     def fsevents_callback(streamRef, clientInfo, numEvents, eventPaths, eventMasks, eventIDs):
337     mydebug("fsevents_callback(streamRef = %s, clientInfo = %s, numEvents = %s)", streamRef, clientInfo, numEvents)
338     mydebug("fsevents_callback: FSEventStreamGetLatestEventId(streamRef) => %s", FSEventStreamGetLatestEventId(streamRef))
339     mydebug("fsevents_callback: eventpaths = %s",eventPaths)
340    
341     full_path = clientInfo
342    
343     result = []
344     for i in range(numEvents):
345     path = eventPaths[i]
346     if path[-1] == '/':
347     path = path[:-1]
348    
349     if eventMasks[i] & kFSEventStreamEventFlagMustScanSubDirs:
350     recursive = True
351    
352     elif eventMasks[i] & kFSEventStreamEventFlagUserDropped:
353     mymesg("BAD NEWS! We dropped events.")
354     mymesg("Forcing a full rescan.")
355     recursive = 1
356     path = full_path
357    
358     elif eventMasks[i] & kFSEventStreamEventFlagKernelDropped:
359     mymesg("REALLY BAD NEWS! The kernel dropped events.")
360     mymesg("Forcing a full rescan.")
361     recursive = 1
362     path = full_path
363    
364     else:
365     recursive = False
366    
367     #now we should know what to do: build a file directory list
368     #I assume here, that unison takes a flag for recursive scans
369     if recursive:
370     #we have to check all subdirectories
371     if isinstance(path,list):
372     #we have to check all base paths
373     allpathsrecursive = [p + '\tr']
374     result.extend(path)
375     else:
376     result.append(path+'\tr')
377     else:
378     #just add the path
379     #result.append(path)
380     #try to find out what has changed
381     result.extend(filelevel_approx(path))
382    
383     mydebug('Dirs sent: %s',eventPaths)
384 bcpierce 471 #TODO: handle creation/deletion of links that should be followed
385 bcpierce 450 update_changes(result)
386    
387     try:
388     f = open(op.absstatus,'w')
389     f.write('last_item = %d'%eventIDs[-1])
390     f.close()
391     except IOError:
392     mymesg('failed to open status file %s', op.absstatus)
393    
394     def my_FSEventStreamCreate(paths):
395 bcpierce 471 mydebug('my_FSEventStreamCreate: selected paths are: %s',paths)
396 bcpierce 450
397     if op.sinceWhen == 'now':
398     op.sinceWhen = kFSEventStreamEventIdSinceNow
399 bcpierce 471
400     try:
401     op.symlinks
402     except AttributeError:
403     make_symlinks()
404    
405     for sl in op.symlinks:
406     #check if that path is already there
407     found=False
408     ln = op.symlinks[sl]
409     for path in paths:
410     if relpath(op.root,path)==ln:
411     found = True
412     break
413     if not found:
414     mydebug('my_FSEventStreamCreate: watch followed link %s',ln)
415     paths.append(os.path.join(op.root,ln))
416 bcpierce 450
417     streamRef = FSEventStreamCreate(kCFAllocatorDefault,
418     fsevents_callback,
419     paths, #will this pass properly through? yes it does.
420     paths,
421     int(op.sinceWhen),
422     float(op.latency),
423     int(op.flags))
424     if streamRef is None:
425 bcpierce 471 mymesg("ERROR: FSEVentStreamCreate() => NULL")
426 bcpierce 450 return None
427    
428     if op.verbose:
429     FSEventStreamShow(streamRef)
430    
431     #print ('my_FSE', streamRef)
432    
433     return streamRef
434    
435     def macosxwatcher():
436     #since when? if it is 'now' try to read state
437     if op.sinceWhen == 'now':
438 bcpierce 471 di = conf_parser(op.absstatus)
439     if di and di.has_key('last_item'):
440     #print di['last_item'][-1]
441     op.sinceWhen = di['last_item'][-1]
442 bcpierce 450 #print op.sinceWhen
443    
444     streamRef = my_FSEventStreamCreate(op.abspaths)
445     #print streamRef
446     if streamRef is None:
447     print('failed to get a Stream')
448     exit(1)
449    
450     FSEventStreamScheduleWithRunLoop(streamRef, CFRunLoopGetCurrent(), kCFRunLoopDefaultMode)
451    
452     startedOK = FSEventStreamStart(streamRef)
453     if not startedOK:
454     print("failed to start the FSEventStream")
455     exit(1)
456    
457     if op.flush_seconds >= 0:
458     mydebug("CFAbsoluteTimeGetCurrent() => %.3f", CFAbsoluteTimeGetCurrent())
459    
460     timer = CFRunLoopTimerCreate(None,
461     CFAbsoluteTimeGetCurrent() + float(op.flush_seconds),
462     float(op.flush_seconds),
463     0, 0, timer_callback, streamRef)
464     CFRunLoopAddTimer(CFRunLoopGetCurrent(), timer, kCFRunLoopDefaultMode)
465    
466     try:
467     CFRunLoopRun()
468     except KeyboardInterrupt:
469     mydebug('stop called via Keyboard, cleaning up.')
470     #Stop / Invalidate / Release
471     FSEventStreamStop(streamRef)
472     FSEventStreamInvalidate(streamRef)
473     FSEventStreamRelease(streamRef)
474     mydebug('FSEventStream closed')
475    
476     #################################################
477     # END MacOsX specific code
478     #################################################
479    
480     #################################################
481     # Windows specific code
482     #################################################
483     if sys.platform == 'win32':
484     import win32file
485     import win32con
486     import threading
487    
488     FILE_LIST_DIRECTORY = 0x0001
489    
490     def win32watcherThread(abspath,file_lock):
491     dirHandle = win32file.CreateFile (
492     abspath,
493     FILE_LIST_DIRECTORY,
494     win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
495     None,
496     win32con.OPEN_EXISTING,
497     win32con.FILE_FLAG_BACKUP_SEMANTICS,
498     None
499     )
500     while 1:
501     results = win32file.ReadDirectoryChangesW (
502     dirHandle,
503     1024,
504     True,
505     win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
506     win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
507     win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
508     win32con.FILE_NOTIFY_CHANGE_SIZE |
509     win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
510     win32con.FILE_NOTIFY_CHANGE_SECURITY,
511     None,
512     None
513     )
514     for action, file in results:
515     full_filename = os.path.join (abspath, file)
516     # This will return 'dir updated' for every file update within dir, but
517     # we don't want to send unison on a full dir sync in this situation.
518     if not (os.path.isdir(full_filename) and action == 3):
519     file_lock.acquire()
520     update_changes_nomangle(full_filename)
521     file_lock.release()
522    
523     def win32watcher():
524     file_lock = threading.Lock()
525     threads = [ threading.Thread(target=win32watcherThread,args=(abspath,file_lock,)) for abspath in op.abspaths ]
526     for thread in threads:
527     thread.setDaemon(True)
528     thread.start()
529    
530     try:
531     while 1:
532     pass
533     except KeyboardInterrupt:
534     print "Cleaning up."
535    
536     #################################################
537     # END Windows specific code
538     #################################################
539    
540     if __name__=='__main__':
541     global op
542    
543     usage = """usage: %prog [options] root [path] [path]...
544     This program monitors file system changes on all given (relative to root) paths
545     and dumps paths (relative to root) files to a file. When launched, this file is
546     recreated. While running new events are added. This can be read by UNISON
547     to trigger a sync on these files. If root is a valid unison profile, we attempt
548     to read all the settings from there."""
549    
550     parser = OptionParser(usage=usage)
551     parser.add_option("-w", "--sinceWhen", dest="sinceWhen",
552     help="""starting point for filesystem updates to be captured
553     Defaults to 'now' in the first run
554     or the last caputured change""",default = 'now', metavar="SINCEWHEN")
555     parser.add_option("-l", "--latency", dest="latency",
556     help="set notification LATENCY in seconds. default 5",default = 5, metavar="LATENCY")
557     parser.add_option("-f", "--flags", dest="flags",
558     help="(macosx) set flags (who knows what they mean. defaults to 0",default = 0, metavar="FLAGS")
559     parser.add_option("-s", "--flushseconds", dest="flush_seconds",
560     help="(macosx) TIME interval in second until flush is forced. values < 0 turn it off. ",default = 1, metavar="TIME")
561     parser.add_option("-o", "--outfile", dest="outfile",
562     help="location of the output file. Defaults to UPATH/changes",default = 'changes', metavar="PATH")
563     parser.add_option("-t", "--statefile", dest="statefile",
564     help="(macosx) location of the state file (absolute or relative to UPATH). Defaults to UPATH/state",default = 'state', metavar="PATH")
565     parser.add_option("-u", "--unisonconfig", dest="uconfdir",
566     help='path to the unison config directory. default ~/.unison',
567     default = '~/.unison', metavar = 'UPATH')
568     parser.add_option("-z", "--follow", dest="follow",
569     help="define a FOLLOW directive. This is equivalent to the -follow option in unison \
570     (except that for now only 'Paths' are supported). This option can appear multiple times. \
571     if a unison configuration file is loaded, it takes precedence over this option",
572     action='append',metavar = 'FOLLOW')
573     parser.add_option("-q", "--quiet",
574     action="store_false", dest="verbose", default=True,
575     help="don't print status messages to stdout")
576    
577     parser.add_option("-d", "--debug",
578     action="store_true", dest="debug", default=False,
579     help="print debug messages to stderr")
580    
581    
582     (op, args) = parser.parse_args()
583    
584    
585     if len(args)<1:
586     parser.print_usage()
587     sys.exit()
588    
589     #other paths
590     op.absuconfdir = my_abspath(op.uconfdir)
591     op.absstatus = os.path.join(op.absuconfdir,op.statefile)
592     op.absoutfile = os.path.join(op.absuconfdir,op.outfile)
593    
594    
595     #figure out if the root argument is a valid configuration file name
596     p = args[0]
597     fn = ''
598     if os.path.exists(p) and not os.path.isdir(p):
599     fn = p
600     elif os.path.exists(os.path.join(op.absuconfdir,p)):
601     fn = os.path.join(op.absuconfdir,p)
602     op.unison_conf = conf_parser(fn)
603    
604     #now check for the relevant information
605     root = None
606     paths = None
607     if op.unison_conf and op.unison_conf.has_key('root'):
608     #find the local root
609     root = None
610     paths = None
611     for r in op.unison_conf['root']:
612     if r[0]=='/':
613     root = r
614     if op.unison_conf.has_key('path'):
615     paths = op.unison_conf['path']
616     if op.unison_conf and op.unison_conf.has_key('follow'):
617     op.follow = op.unison_conf['follow']
618     else:
619     #see if follows were defined
620     try:
621     op.follow
622     except AttributeError:
623     op.follow = []
624    
625     if not root:
626     #no root up to here. get it from args
627     root = args[0]
628    
629     if not paths:
630     paths = args[1:]
631    
632     #absolute paths
633     op.root = my_abspath(root)
634     op.abspaths = [os.path.join(root,path) for path in paths]
635     if op.abspaths == []:
636     #no paths specified -> make root the path to observe
637     op.abspaths = [op.root]
638     #print op.root
639     #print op.abspaths
640    
641     mydebug('options: %s',op)
642     mydebug('arguments: %s',args)
643    
644     #cleaning up the change file
645     try:
646     f=open(op.absoutfile,'w')
647     f.close()
648     except IOError:
649     mymesg('failed to open output file. STOP.')
650     exit(1)
651    
652     if sys.platform=='darwin':
653     macosxwatcher()
654     elif sys.platform.startswith('linux'):
655     linuxwatcher()
656     elif sys.platform.startswith('win32'):
657     win32watcher()
658     else:
659     mymesg('unsupported platform %s',sys.platform)
660    
661    

Properties

Name Value
svn:executable *

  ViewVC Help
Powered by ViewVC 1.1.26