github/supysonic
1
0
Fork 0
mirror of https://github.com/spl0k/supysonic.git synced 2024-12-23 01:16:18 +00:00
Code Issues Releases Wiki Activity

Improved queue processing

Browse Source
This commit is contained in:
spl0k 2014-07-27 12:30:45 +02:00
parent 52891cbf4c
commit 132d7b0c8b
1 changed files with 30 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
bin/supysonic-watcher
View File

@ -21,7 +21,7 @@
import time, sys import time, sys
import logging import logging
from threading import Thread, Lock from threading import Thread, Condition, Timer
from logging.handlers import TimedRotatingFileHandler from logging.handlers import TimedRotatingFileHandler
from watchdog.observers import Observer from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler from watchdog.events import PatternMatchingEventHandler
@ -65,47 +65,56 @@ class ScannerProcessingQueue(Thread):
super(ScannerProcessingQueue, self).__init__() super(ScannerProcessingQueue, self).__init__()
self.__logger = logger self.__logger = logger
self.__lock = Lock() self.__cond = Condition()
self.__timer = None
self.__queue = {} self.__queue = {}
self.__running = True self.__running = True
def run(self): def run(self):
while self.__running: while self.__running:
time.sleep(5) time.sleep(0.1)
with self.__cond:
self.__cond.wait()
with self.__lock:
if not self.__queue: if not self.__queue:
continue continue
self.__logger.debug("Instantiating scanner") self.__logger.debug("Instantiating scanner")
scanner = Scanner(db.session) scanner = Scanner(db.session)
self.__lock.acquire() while self.__queue:
while self.__queue: path = sorted(self.__queue.iteritems(), key = lambda i: i[1])[0][0]
path = sorted(self.__queue.iteritems(), key = lambda i: i[1])[0][0] self.__logger.info("Scanning: '%s'", path)
self.__lock.release() scanner.scan_file(path)
self.__logger.info("Scanning: '%s'", path) del self.__queue[path]
scanner.scan_file(path)
self.__lock.acquire() db.session.commit()
del self.__queue[path] db.session.remove()
self.__lock.release() self.__logger.debug("Freeing scanner")
del scanner
db.session.commit()
db.session.remove()
self.__logger.debug("Freeing scanner")
del scanner
def stop(self): def stop(self):
self.__running = False self.__running = False
with self.__cond:
self.__cond.notify()
def put(self, path): def put(self, path):
if not self.__running: if not self.__running:
raise RuntimeError("Trying to put an item in a stopped queue") raise RuntimeError("Trying to put an item in a stopped queue")
with self.__lock: with self.__cond:
self.__queue[path] = time.time() self.__queue[path] = time.time()
if self.__timer:
self.__timer.cancel()
self.__timer = Timer(5, self.__wakeup)
self.__timer.start()
def __wakeup(self):
with self.__cond:
self.__cond.notify()
self.__timer = None
if __name__ == "__main__": if __name__ == "__main__":
if not config.check(): if not config.check():