def watch_backups(machines):
# set nice and ionice for current process
os.setpriority(os.PRIO_PROCESS, 0, 10)
p = psutil.Process(os.getpid())
p.ionice(psutil.IOPRIO_CLASS_IDLE)
i = inotify.adapters.Inotify()
# TODO watch all paths
for name in machines:
d = os.path.join(BU_SYNC_DIR,name)
i.add_watch( d.encode('utf-8') )
print( "Start watching {}".format(d) )
# wait for an inotify event
for event in i.event_gen():
if event is None:
continue
# decode event info
(header, type_names, watch_path, filename) = event
watch_path = watch_path.decode('utf-8')
filename = filename.decode('utf-8')
# only take action of the stamp file was updated
if filename != 'stamp':
continue
if header.mask & (inotify.constants.IN_MOVED_TO|inotify.constants.IN_CLOSE_WRITE):
# only got here if a backup just finished
_LOGGER.info("WD=(%d) MASK=(%d) COOKIE=(%d) LEN=(%d) MASK->NAMES=%s "
"WATCH-PATH=[%s] FILENAME=[%s]",
header.wd, header.mask, header.cookie, header.len, type_names,
watch_path, filename)
try:
# handle the new backup
new_backup(watch_path)
except BackupError as e:
_LOGGER.error(e)
python类IOPRIO_CLASS_IDLE的实例源码
def test_ionice(self):
if LINUX:
from psutil import (IOPRIO_CLASS_NONE, IOPRIO_CLASS_RT,
IOPRIO_CLASS_BE, IOPRIO_CLASS_IDLE)
self.assertEqual(IOPRIO_CLASS_NONE, 0)
self.assertEqual(IOPRIO_CLASS_RT, 1)
self.assertEqual(IOPRIO_CLASS_BE, 2)
self.assertEqual(IOPRIO_CLASS_IDLE, 3)
p = psutil.Process()
try:
p.ionice(2)
ioclass, value = p.ionice()
if enum is not None:
self.assertIsInstance(ioclass, enum.IntEnum)
self.assertEqual(ioclass, 2)
self.assertEqual(value, 4)
#
p.ionice(3)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 3)
self.assertEqual(value, 0)
#
p.ionice(2, 0)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 0)
p.ionice(2, 7)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 7)
#
self.assertRaises(ValueError, p.ionice, 2, 10)
self.assertRaises(ValueError, p.ionice, 2, -1)
self.assertRaises(ValueError, p.ionice, 4)
self.assertRaises(TypeError, p.ionice, 2, "foo")
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_NONE",
p.ionice, psutil.IOPRIO_CLASS_NONE, 1)
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_IDLE",
p.ionice, psutil.IOPRIO_CLASS_IDLE, 1)
self.assertRaisesRegex(
ValueError, "'ioclass' argument must be specified",
p.ionice, value=1)
finally:
p.ionice(IOPRIO_CLASS_NONE)
else:
p = psutil.Process()
original = p.ionice()
self.assertIsInstance(original, int)
try:
value = 0 # very low
if original == value:
value = 1 # low
p.ionice(value)
self.assertEqual(p.ionice(), value)
finally:
p.ionice(original)
#
self.assertRaises(ValueError, p.ionice, 3)
self.assertRaises(TypeError, p.ionice, 2, 1)
def test_ionice(self):
if LINUX:
from psutil import (IOPRIO_CLASS_NONE, IOPRIO_CLASS_RT,
IOPRIO_CLASS_BE, IOPRIO_CLASS_IDLE)
self.assertEqual(IOPRIO_CLASS_NONE, 0)
self.assertEqual(IOPRIO_CLASS_RT, 1)
self.assertEqual(IOPRIO_CLASS_BE, 2)
self.assertEqual(IOPRIO_CLASS_IDLE, 3)
p = psutil.Process()
try:
p.ionice(2)
ioclass, value = p.ionice()
if enum is not None:
self.assertIsInstance(ioclass, enum.IntEnum)
self.assertEqual(ioclass, 2)
self.assertEqual(value, 4)
#
p.ionice(3)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 3)
self.assertEqual(value, 0)
#
p.ionice(2, 0)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 0)
p.ionice(2, 7)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 7)
#
self.assertRaises(ValueError, p.ionice, 2, 10)
self.assertRaises(ValueError, p.ionice, 2, -1)
self.assertRaises(ValueError, p.ionice, 4)
self.assertRaises(TypeError, p.ionice, 2, "foo")
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_NONE",
p.ionice, psutil.IOPRIO_CLASS_NONE, 1)
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_IDLE",
p.ionice, psutil.IOPRIO_CLASS_IDLE, 1)
self.assertRaisesRegex(
ValueError, "'ioclass' argument must be specified",
p.ionice, value=1)
finally:
p.ionice(IOPRIO_CLASS_NONE)
else:
p = psutil.Process()
original = p.ionice()
self.assertIsInstance(original, int)
try:
value = 0 # very low
if original == value:
value = 1 # low
p.ionice(value)
self.assertEqual(p.ionice(), value)
finally:
p.ionice(original)
#
self.assertRaises(ValueError, p.ionice, 3)
self.assertRaises(TypeError, p.ionice, 2, 1)
def test_ionice(self):
if LINUX:
from psutil import (IOPRIO_CLASS_NONE, IOPRIO_CLASS_RT,
IOPRIO_CLASS_BE, IOPRIO_CLASS_IDLE)
self.assertEqual(IOPRIO_CLASS_NONE, 0)
self.assertEqual(IOPRIO_CLASS_RT, 1)
self.assertEqual(IOPRIO_CLASS_BE, 2)
self.assertEqual(IOPRIO_CLASS_IDLE, 3)
p = psutil.Process()
try:
p.ionice(2)
ioclass, value = p.ionice()
if enum is not None:
self.assertIsInstance(ioclass, enum.IntEnum)
self.assertEqual(ioclass, 2)
self.assertEqual(value, 4)
#
p.ionice(3)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 3)
self.assertEqual(value, 0)
#
p.ionice(2, 0)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 0)
p.ionice(2, 7)
ioclass, value = p.ionice()
self.assertEqual(ioclass, 2)
self.assertEqual(value, 7)
#
self.assertRaises(ValueError, p.ionice, 2, 10)
self.assertRaises(ValueError, p.ionice, 2, -1)
self.assertRaises(ValueError, p.ionice, 4)
self.assertRaises(TypeError, p.ionice, 2, "foo")
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_NONE",
p.ionice, psutil.IOPRIO_CLASS_NONE, 1)
self.assertRaisesRegex(
ValueError, "can't specify value with IOPRIO_CLASS_IDLE",
p.ionice, psutil.IOPRIO_CLASS_IDLE, 1)
self.assertRaisesRegex(
ValueError, "'ioclass' argument must be specified",
p.ionice, value=1)
finally:
p.ionice(IOPRIO_CLASS_NONE)
else:
p = psutil.Process()
original = p.ionice()
self.assertIsInstance(original, int)
try:
value = 0 # very low
if original == value:
value = 1 # low
p.ionice(value)
self.assertEqual(p.ionice(), value)
finally:
p.ionice(original)
#
self.assertRaises(ValueError, p.ionice, 3)
self.assertRaises(TypeError, p.ionice, 2, 1)
