@@ -1,9 +1,13 @@ +# -*- coding: utf-8 -*- from __future__ import with_statement import os import logging from threading import Semaphore +import threading from subprocess import * +import Jobs +import time #Creating a backup repo: # mkdir /tmp/hs-test/ @@ -21,6 +25,16 @@ from subprocess import * # cd /tmp/hs-test # git --git-dir=/tmp/hs-test/.home-sync/git --work-tree=/tmp/hs-test pull localhost:/home/trever/.home-sync/git +class GitCallback: + def FilesAdded(self,files): + pass + + def FilesCommitted(self,files): + pass + +class GitProcessException(Jobs.JobException): + pass + class GitInterface: FILES_MODIFIED = 1 FILES_CACHED = 2 @@ -29,25 +43,62 @@ class GitInterface: FILES_IGNORED = 16 FILES_STAGE = 32 - def __init__(self,gitDir,workTree='',bare=False): + def __init__(self,gitDir,workTree='', bare=False, callback=GitCallback()): self.dir = gitDir self.work = workTree self.lock = Semaphore() self.bare = bare self.log = logging.getLogger('GitInterface') - + self.callback = callback + def exists(self): return os.path.exists(self.dir) - def open(self,*args): + def writeHead(self): + with self.lock: + git = self.open("update-ref","HEAD", self.head) + ret = git.wait() + + def readHead(self): + with self.lock: + git = self.open("rev-parse", "HEAD",stdout=PIPE) + try: + git.wait() + except GitProcessException: + pass + self.head = git.stdout.readline().strip() + git.stdout.close() + if self.head == "HEAD": #Returns the revision name for a bad name + self.head = "" + + def open(self, *args, **kwargs): + if ("stdout" in kwargs): + stdout = kwargs["stdout"] + else: + stdout = None + if ("stdin" in kwargs): + stdin = kwargs["stdin"] + else: + stdin = None args = map(str,args) if self.bare: args = ["git","--git-dir=%s"%(self.dir),"--bare"]+args else: args = ["git","--git-dir=%s"%(self.dir),"--work-tree=%s"%(self.work)] + args - self.log.debug("Running %s"%(args)) - #return Popen(args, stdout=PIPE) - return Popen(args) + self.log.debug("Running %s in %s"%(args, repr(threading.currentThread()))) + time.sleep(0.0001) + proc = Popen(args, stdout=stdout, stdin=stdin) + time.sleep(0.0001) + def waitForGit(message="", fatal=True): + ret = proc.popen_wait() + if (ret != 0 and fatal): + raise GitProcessException(ret, message) + return ret + + proc.popen_wait = proc.wait + proc.wait = waitForGit + self.log.debug("Process started as PID %i", proc.pid) + return proc def init(self): with self.lock: @@ -65,10 +116,11 @@ class GitInterface: git.wait() def setConfig(self,name,value): - git = self.open("config",name,str(value)) + with self.lock: + git = self.open("config",name,str(value)) + git.wait() def files(self,listType=FILES_MODIFIED): - self.log.debug("Listing...") flags = "" if (listType & GitInterface.FILES_MODIFIED): flags+="m" @@ -84,7 +136,7 @@ class GitInterface: flags+="s" if (flags==""): flags="m" - git = self.open("ls-files","-"+flags) + git = self.open("ls-files","-"+flags, stdout=PIPE) list = [] while (True): file = git.stdout.readline().strip() #FIXME: Just trim the one newline @@ -95,63 +147,57 @@ class GitInterface: git.wait() return list - def add(self,*files): + def addToIndex(self,*files): with self.lock: - git = self.open("add","-v","--",*files) - #while(True): - #file = git.stdout.readline().strip().split()[1:] - #if (file == []): - #break - #callback(''.join(file).strip("'")) + git = self.open("update-index","--add","-z","--stdin", stdin=PIPE, stdout=PIPE) + #i = 0 + for f in files: + #i+=1 + git.stdin.write(f+"\0") + #self.log.debug("Wrote %i/%i files: %s", i, len(files), f) + git.stdin.close() + self.callback.FilesAdded(files) git.wait() + git.stdout.close() + self.log.debug("Done") - def update(self,path): + def commitIndex(self, message): + self.log.debug("Commiting %s...",message) + self.readHead() with self.lock: - git = self.open("add","-u","--",path) + git = self.open("write-tree", stdout=PIPE) + tree = git.stdout.readline().strip() git.wait() - - def commitAll(self, message): - self.log.debug("Committing all changes: %s",message) - with self.lock: - git = self.open("commit","-a","-m",message) - #while (True): - #line=git.stdout.readline().strip().split()[5:] - #if line == []: - #break - #callback(''.join(line)) + git.stdout.close() + if self.head == "": + git = self.open("commit-tree", tree, stdin=PIPE, stdout=PIPE) + else: + git = self.open("commit-tree",tree,"-p",self.head, stdin=PIPE, stdout=PIPE) + git.stdin.write(message) + git.stdin.close() git.wait() - self.log.debug("Commited.") - - def commit(self, message, *files): - self.log.debug("Commiting %s...",message) - with self.lock: - git = self.open("commit","-m",message,"--",*files) - #while (True): - #line=git.stdout.readline().strip().split()[5:] - #if line == []: - #break - #callback(''.join(line.split()[5:])) - self.log.debug("Commited.") + self.head = git.stdout.read().strip() + git.stdout.close() + self.writeHead() + self.log.debug("Commited.") - def compact(self): - self.lock.acquire() - self.log.debug("Compressing git repo") - git = self.open("gc","--aggressive") - git.wait() - self.lock.release() + #def compact(self): + # self.lock.acquire() + # self.log.debug("Compressing git repo") + # git = self.open("gc","--aggressive") + # git.wait() + # self.lock.release() def revisionList(self, path): - self.lock.acquire() - self.log.debug("Requesting revision history of %s",path) - git = self.open("rev-list","--timestamp","HEAD",path) - ret = [] - while(True): - commit = git.stdout.readline().strip().split() - if (commit == []): - break - commit[0] = int(commit[0]) - #ret.append(commit) - ret.append([commit[1],commit[0]]) - git.wait() - self.lock.release() - return ret \ No newline at end of file + with self.lock: + self.log.debug("Requesting revision history of %s",path) + git = self.open("rev-list","--timestamp","HEAD",path, stdout=PIPE) + ret = [] + while(True): + commit = git.stdout.readline().strip().split() + if (commit == []): + break + commit[0] = int(commit[0]) + ret.append([commit[1],commit[0]]) + git.wait() + return ret \ No newline at end of file HomeSync/GitInterface.py @@ -1,12 +1,23 @@ +# -*- coding: utf-8 -*- import threading import logging +import time + +class JobException(Exception): + def __init__(self, status=-1, message=""): + self.status = status + self.message = message + self.job = None + + def __str__(self): + return "%s exited with non-zero status of %s (%s)" % (self.job, self.status, self.message) class JobQueue(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.log = logging.getLogger('JobQueue') self.queue = [] - self.queueReady = threading.Condition() + self.queueReady = threading.Condition(threading.Lock()) self.running = threading.Event() self.running.set() self.active = None @@ -16,8 +27,8 @@ class JobQueue(threading.Thread): self.queue.insert(0,job) self.queue = sorted(self.queue) - self.log.debug("Queued job %s"%(job)) - self.log.debug("Arguments: %s %s",job.args, job.kwargs) + #self.log.debug("Queued job %s", repr(job)) + #self.log.debug("Arguments: %s %s", job.args, job.kwargs) self.queueReady.notify() self.queueReady.release() self.NewJob(job.id) @@ -31,15 +42,18 @@ class JobQueue(threading.Thread): self.running.wait() self.queueReady.acquire() while self.haveJobs() == False: + self.log.debug("Waiting for a new job") self.queueReady.wait() self.queue = sorted(self.queue) - self.active= self.queue.pop() + self.active = self.queue.pop() self.queueReady.release() - self.log.info("Starting job %s",self.active) - self.log.debug("Arguments: %s %s",self.active.args, self.active.kwargs) + self.log.info("Starting job %s",repr(self.active)) self.active.start() self.active.join() - self.log.info("Job complete: %s", self.active) + self.log.info("Job complete: %s", repr(self.active)) + if (self.active.exception): + self.log.error("Exception thrown in job. Passing up.") + raise self.active.exception self.JobComplete(self.active.id) del self.active self.active = None @@ -65,10 +79,11 @@ class JobQueue(threading.Thread): class Job(threading.Thread): id = 0 - def __init__(self,call, args=[], kwargs={}): + def __init__(self, call, args=[], kwargs={}): Job.id += 1 self.prio = 0 self.id = Job.id + self.exception = None threading.Thread.__init__(self) self.call=call @@ -94,19 +109,31 @@ class Job(threading.Thread): Job.id-=1 def __str__(self): - return "#%i (%i) %s"%(self.id, self.prio, self.name) + return self.name + + def __repr__(self): + if (self.exception == None): + status = "OK" + else: + status = "EXCEPTION" + return "<ID:%i Priority:%i Status:%s Name:%s>"%(self.id, self.prio, status, self.name) def run(self): - self.call(*self.args,**self.kwargs) + try: + self.call(*self.args,**self.kwargs) + except JobException, e: + e.job = self + self.exception = e + raise e class JobGroup(Job, JobQueue): def __init__(self): Job.__init__(self, self.runJobs) JobQueue.__init__(self) - self.name='' + self.name="JobGroup" - def __str__(self): - return Job.__str__(self)+str(map(str,self.queue)) + def __repr__(self): + return Job.__repr__(self)+str(map(str,self.queue)) def runJobs(self): while self.haveJobs(): HomeSync/Jobs.py @@ -1,8 +1,10 @@ -from GitInterface import GitInterface +# -*- coding: utf-8 -*- +from GitInterface import GitInterface, GitCallback from Jobs import JobQueue, Job, JobGroup import logging import os from fnmatch import fnmatch +#from git import * DEFAULT_IGNORES=[ "/.home-sync", @@ -21,9 +23,12 @@ DEFAULT_IGNORES=[ ] class SyncArchive: - def __init__(self, path, backup=False): + def __init__(self, path, backup=False, gitHandler=None): self.backup = backup self.path = path + if gitHandler == None: + gitHandler=GitCallback() + self.handler = gitHandler if backup: bare = True @@ -49,7 +54,8 @@ class SyncArchive: file = file.lstrip(self.path) for pattern in self.ignoreList(): if fnmatch(file, "*/"+pattern): - return true + return True + return False def create(self): if self.backup: @@ -81,10 +87,10 @@ class SyncArchive: def addFiles(self, *files): self.commits+=1 j = JobGroup() - j.enqueue( Job(self.git.add, args=files) ) - j.enqueue( Job(self.git.commit, args=("%s added"%(files),)+files) ) + j.enqueue( Job(self.git.addToIndex, args=files) ) + j.enqueue( Job(self.git.commitIndex, args=("%i files added"%(len(files)),)) ) if self.commits>100: - j.enqueue( Job(self.git.compact) ) #Make sure we compact after we commit and add + #j.enqueue( Job(self.git.compact) ) #Make sure we compact after we commit and add self.commits=0 self.jobs.enqueue(j) return j HomeSync/SyncArchive.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- import os, sys import logging import threading @@ -7,7 +8,9 @@ from AvahiBrowser import AvahiBrowser from DeviceWatcher import DeviceWatcher from SyncArchive import SyncArchive from Jobs import Job +from GitInterface import GitInterface, GitCallback import dbus.service +import pyinotify class Crawler: def __init__(self,archive): @@ -16,125 +19,148 @@ class Crawler: def start(self): self.dive(self.archive.path) + self.archive.jobs.join() + + def dive(self,*paths): + for path in paths: + files = os.listdir(path) + fileList = [] + dirList = [] + for f in files: + f = os.path.join(path,f) + if os.path.isfile(f) and not self.archive.isIgnored(f): + fileList.append(f) + if os.path.isdir(f): + dirList.append(f) + if fileList != []: + self.archive.addFiles(*fileList).setPriority(-1) + if dirList != []: + j = Job(self.dive,args=dirList ) + j.setPriority(-2) + self.archive.jobs.enqueue(j) + +class Watcher(pyinotify.ProcessEvent): + def __init__(self, archive): + pyinotify.ProcessEvent.__init__(self) + self.archive = archive + self.wm = pyinotify.WatchManager() + self.notifier = pyinotify.ThreadedNotifier(self.wm, self) + + def start(self): + self.notifier.start() + self.wm.add_watch(self.archive.path, pyinotify.EventCodes.IN_CREATE | pyinotify.EventCodes.IN_MODIFY, rec=True) - def dive(self,path): - files = os.listdir(path) - list = [] + def process_IN_CREATE(self, event): + print "Created:", os.path.join(event.path, event.name) + self.archive.addFiles(os.path.join(event.path, event.name)) + + def process_IN_MODIFY(self, event): + print "Modified:", os.path.join(event.path, event.name) + self.archive.addFiles(os.path.join(event.path, event.name)) + +class DaemonCallback(GitCallback): + def __init__(self, daemon): + self.d = daemon + + def FilesAdded(self,files): for f in files: - if os.path.isfile(os.path.join(path,f)) and not self.archive.isIgnored(os.path.join(path,f)): - list.append(os.path.join(path,f)) - if os.path.isdir(os.path.join(path,f)): - self.archive.jobs.enqueue( Job(self.dive,args=(os.path.join(path, f),)) ).setPriority(-2) - if list != []: - self.archive.addFiles(*list).setPriority(-1) + self.d.FileAdded(f) + + def FilesCommitted(self,files): + for f in files: + self.d.FileCommited(f) -class SyncDaemon(dbus.service.Object): +class SyncDaemon:#(dbus.service.Object): def __init__(self): self.log = logging.getLogger('SyncDaemon') self.log.debug("Created SyncDaemon") self.log.debug("Attaching to D-BUS") self.name = dbus.service.BusName("net.wm161.HomeSync",bus=dbus.SessionBus()) - dbus.service.Object.__init__(self, object_path="/Server", conn=dbus.SessionBus(), bus_name=self.name) + #dbus.service.Object.__init__(self, object_path="/Server", conn=dbus.SessionBus(), bus_name=self.name) home = os.environ["HOME"] os.chdir(home) - self.archive = SyncArchive(home, backup=False) + self.archiveHandler = DaemonCallback(self) + self.archive = SyncArchive(home, backup=False, gitHandler = self.archiveHandler) self.archive.create() - os.nice(5) - #if (self.git.exists() == False): - #self.log.info("Missing Git repo. Running git-init.") - #self.jobs.enqueue(self.git.init) - #self.jobs.enqueue(self.git.setConfig,"core.excludesfile",home+"/.home-sync/ignore") - #self.jobs.enqueue(self.git.setConfig,"core.compression",-1) - - #self.log.debug("Watching %s"%(home)) - #self.watcher = SyncWatcher() - #self.watcher.add(home) - #QtCore.QObject.connect(self.watcher,QtCore.SIGNAL("directoryChanged(QString)"),self.dirUpdate) + #os.nice(5) + self.Ready() - self.log.info("Starting index crawler") + self.log.info("Starting indexing crawler") self.crawler = Crawler(self.archive) self.crawler.start() + self.watcher = Watcher(self.archive) + self.watcher.start() - self.avahi = AvahiBrowser() - self.deviceWatcher = DeviceWatcher() + #self.avahi = AvahiBrowser() + #self.deviceWatcher = DeviceWatcher() + self.log.info("Ready for action.") def dirUpdate(self,path): path = unicode(path).replace(os.environ["HOME"]+"/",'./',1) self.DirectoryChanged(path) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') def PushToServer(self, remote): pass - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', out_signature='as') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', out_signature='as') def DiscoveredHosts(self): pass - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', out_signature='as', in_signature='i') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', out_signature='as', in_signature='i') def ListFiles(self,type): self.log.debug("Listing files of type %i"%(type)) return self.git.files(type) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') def AddFile(self,file): self.log.debug("Adding %s to repository",file) self.jobs.enqueue(self.git.add,file,self.FileAdded) self.jobs.enqueue(self.git.commit, "%s added"%(file), callback=self.FileCommitted) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server') def ArchiveNow(self): self.log.info("Requested an immediate archive.") self.jobs.enqueue(self.git.commitAll,"On Demand Archive", callback=self.FileCommitted) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') def CommitFile(self, file): self.log.info("Committing %s",file) self.jobs.enqueue(self.git.commit,"Requested commit for %s"%(file), path=file, callback=self.FileCommitted) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') def CreateBackup(self, path): self.log.info("Creating backup on %s",path) backup = GitInterface(path,bare=True) backup.init() - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s') def Backup(self, path): self.log.info("Pushing backup to %s",path) #backup = GitInterface(path,bare=True) self.jobs.enqueue(self.git.push,path) - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server') + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server') def ExitDaemon(self): self.log.info("Exiting") sys.exit() - @dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s', out_signature="a(si)") + #@dbus.service.method(dbus_interface='net.wm161.HomeSync.Server', in_signature='s', out_signature="a(si)") def FileRevisions(self,path): self.log.debug("Requesting revisions of %s"%(path)) return self.git.revisionList(path) - @dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server') + #@dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server') def Ready(self): self.log.debug("Ready.") - @dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') + #@dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') def FileAdded(self,file): self.log.debug("File added: %s",file) - @dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') + #@dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') def FileCommitted(self, file): self.log.debug("File committed: %s",file) - - @dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') - def FileChanged(self,path): - #TODO: Detect batch updates - self.log.debug("Detected change in file %s"%(path)) - self.jobs.enqueue(self.git.commit,"%s modification"%(path), path=path, callback=self.FileCommitted) - - @dbus.service.signal(dbus_interface='net.wm161.HomeSync.Server', signature='s') - def DirectoryChanged(self,path): - self.log.debug("Detected change in directory %s"%(path)) - self.jobs.enqueue(self.git.commit,"%s modification"%(file), path=path, callback=self.FileCommitted) HomeSync/SyncDaemon.py @@ -1,30 +1,19 @@ #!/usr/bin/env python +# -*- coding: utf-8 -*- import sys import logging from HomeSync.SyncDaemon import SyncDaemon -try: - from dbus.mainloop.glib import DBusGMainLoop - import gobject - glibLoop = True -except: - try: - from dbus.mainloop.qt import DBusQtMainLoop - from PyQt4 import QtCore - glibLoop = False - except: - print "No D-BUS main loops found." - sys.exit() +from dbus.mainloop.glib import DBusGMainLoop +import gobject + if __name__ == "__main__": logging.basicConfig(level=logging.DEBUG) - if glibLoop: - DBusGMainLoop(set_as_default=True) - server = SyncDaemon() - loop = gobject.MainLoop() - loop.run() - else: - DBusQtMainLoop(set_as_default=True) - server = SyncDaemon() - loop = QtCore.QEventLoop() - sys.exit(loop._exec()) + #DBusGMainLoop(set_as_default=True) + server = SyncDaemon() + #loop = gobject.MainLoop() + #try: + #loop.run() + #except KeyboardInterrupt: + #server.ExitDaemon() \ No newline at end of file sync-daemon.py 855eebcdb0ae841a985d1211a01652349932981e Trever Fischer wm161@wm161.net http://github.com/workman161/homesync/commit/9d2b919e135baad95330a5e529b6032dd2af6c4f 9d2b919e135baad95330a5e529b6032dd2af6c4f 2009-04-02T11:51:48-07:00 2009-04-02T11:51:48-07:00 Added exception handling to jobs Implemented more git commands to GitInterface, renamed a few to make more sense Made the background indexing crawler into its own thread Added inotify support to track modifications in realtime Temporarily removed DBus to workaround a global interpreter lock bug with python-dbus 74d50710cbb667e5c4f69a2e57bf3a20e71ac827 Trever Fischer wm161@wm161.net